I am using livekit-test. How do I integrate custom knowledgebase to test the voice ai agent ?

Thanks for reaching out, Jay. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

I am using Python and ChromaDB

Thanks — I’m responding to your follow-up about using Python with ChromaDB to integrate a custom knowledge base into your livekit-test workflow for a voice AI agent. Below is a focused, practical guide you can adapt. Summary of approach • Run a RAG lookup against ChromaDB during the agent lifecycle (e.g., in

on_user_turn_completed

). • Inject the retrieved snippet into the chat context so the assistant reply (and resulting TTS) is grounded in KB content. • During tests, mock the ChromaDB lookup for deterministic, reproducible assertions. • Optionally use an LLM judge to validate correctness. Example components (Python) 1) Minimal ChromaDB RAG lookup (concept) • Implement a function that queries Chroma, ranks results, and returns a short, relevant snippet.

from chromadb import Client
from chromadb.config import Settings

# Initialize client (adjust settings/embedding method as needed)
client = Client(Settings(chroma_db_impl="chromadb.db.sqlite", persist_directory="./chroma_store"))
collection = client.get_or_create_collection("my_kb")

async def chroma_rag_lookup(query: str, k: int = 3) -> str:
    # If you use synchronous Chroma calls, you can keep this non-async.
    results = collection.query(query_texts=[query], n_results=k)
    # results["documents"] typically contains nested lists: choose top hits
    docs = results.get("documents", [[]])[0]
    # Concatenate or select the most relevant snippet(s)
    if not docs:
        return "No relevant information found."
    # Return a short summary/concatenation
    return "\n\n".join(docs[:k])

2) Injecting KB results into the agent lifecycle • Use

on_user_turn_completed

to fetch Chroma results and add them to the turn context so the next assistant generation uses them.

async def on_user_turn_completed(self, turn_ctx: ChatContext, new_message: ChatMessage) -> None:
    query = new_message.text_content()
    rag_content = await chroma_rag_lookup(query)   # await only if your lookup is async
    if rag_content:
        turn_ctx.add_message(
            role="assistant",
            content=f"Relevant knowledgebase content:\n{rag_content}"
        )

Notes: • Keep injected snippets short and clearly labeled so the model knows to use them as grounding. • If you summarize results (preferred), consider running a concise summarization step before injection to reduce token usage. 3) Mocking Chroma in tests (deterministic) • During livekit-test runs, mock the

chroma_rag_lookup

function so tests don’t depend on the DB and return predictable content.

from livekit.agents import mock_tools

def fake_chroma_rag(query: str) -> str:
    if "refund" in query.lower():
        return "Our refund policy: returns accepted within 30 days after purchase."
    return "No relevant information found."

with mock_tools(MyAssistantClass, {"chroma_rag_lookup": fake_chroma_rag}):
    result = await session.run(user_input="What is your refund policy?")
    await result.expect.next_event().is_message(role="assistant").judge(
        llm, intent="Explains refund policy correctly based on KB content."
    )

4) Voice / TTS considerations • Confirm your TTS pipeline consumes the assistant message created after your RAG injection. • If you synthesize speech immediately after generation, ensure the final assistant message contains the KB-grounded content before TTS runs. 5) Multi-turn and evaluation • Test multi-turn flows where later questions reference earlier KB content to validate context persistence. • Use an LLM-as-judge to assert whether the assistant response correctly uses KB content and answers intent. 6) Practical tips • Store and return short snippets (1–3 paragraphs) to avoid excessive token use. • Consider caching frequent queries to speed tests. • If your embeddings/Chroma setup is synchronous, use synchronous calls in the agent lifecycle or wrap them with an executor to avoid blocking the event loop. If you’d like, I can: • Provide a complete runnable example that wires ChromaDB, the agent lifecycle hook, and a minimal livekit-test session in Python. • Show how to summarize retrieved docs before injection (to save tokens). • Help with the exact mocking pattern for your test runner (pytest / asyncio specifics). Which would you like next? A full runnable example or the summarization + token-saving pattern?