Generalist Voice Agent

pip install "livekit-agents>=1.0.0" \
  livekit-plugins-openai livekit-plugins-deepgram \
  livekit-plugins-silero livekit-plugins-cartesia \
  moss python-dotenv

docs/
  customer_support/   ← refunds, account help, shipping policies
  product_faq/        ← features, pricing, integrations
  hr_policies/        ← leave, benefits, onboarding

python create_index.py --index-name customer_support --docs-dir ./docs/customer_support
python create_index.py --index-name product_faq      --docs-dir ./docs/product_faq

from __future__ import annotations
import json, logging, os
from typing import Annotated
from dotenv import load_dotenv
from livekit.agents import Agent, AgentSession, JobContext, WorkerOptions, cli, function_tool
from livekit.plugins import cartesia, deepgram, openai, silero
from moss import MossClient, QueryOptions

load_dotenv()

VOICE_RULES = """
- Speak in short, natural sentences — this is a phone call, not a chat UI.
- Never read out bullet points, markdown, URLs, or document IDs.
- Never mention searching or databases.
- If you cannot find the answer, say so and offer to help with something else.
"""

class MossVoiceAgent(Agent):
    def __init__(self, moss_client: MossClient, index_name: str, instructions: str):
        super().__init__(instructions=instructions + VOICE_RULES)
        self._moss = moss_client
        self._index_name = index_name

    @function_tool
    async def moss_search(
        self,
        query: Annotated[str, "Concise query capturing what the caller wants to know."],
    ) -> str:
        """Retrieve relevant information to answer the caller's question.
        Call this whenever you need factual context before responding.
        """
        result = await self._moss.query(
            self._index_name, query, QueryOptions(top_k=5, alpha=0.5)
        )
        if not result.docs:
            return "No relevant information found."
        return "\n\n---\n\n".join(doc.text for doc in result.docs)

async def entrypoint(ctx: JobContext) -> None:
    await ctx.connect()

    # Resolve persona from room metadata: {"persona": "customer_support"}
    metadata = json.loads(ctx.room.metadata or "{}")
    persona_id = metadata.get("persona", "customer_support")

    with open("personas.json", encoding="utf-8") as f:
        personas = json.load(f)
    persona = personas.get(persona_id, next(iter(personas.values())))
    moss_client = MossClient(os.environ["MOSS_PROJECT_ID"], os.environ["MOSS_PROJECT_KEY"])
    await moss_client.load_index(persona["index_name"])  # pull into local memory

    agent = MossVoiceAgent(moss_client, persona["index_name"], persona["instructions"])

    session = AgentSession(
        stt=deepgram.STT(model="nova-2"),
        llm=openai.LLM(model="gpt-4.1-mini"),
        tts=cartesia.TTS(),
        vad=silero.VAD.load(),
    )
    await session.start(agent=agent, room=ctx.room)

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

# Development (auto-reloads on file changes)
python agent.py dev

# Production
python agent.py start

{"persona": "product_faq"}