Guide for building production-ready LiveKit voice AI agents with multi-agent workflows and intelligent handoffs.
Build production-ready voice AI agents using LiveKit Agents framework with support for multi-agent workflows, intelligent handoffs, and specialized agent capabilities.
LiveKit Agents enables building real-time multimodal AI agents with voice capabilities. This skill helps you create sophisticated voice systems where multiple specialized agents can seamlessly hand off conversations based on context, user needs, or business logic.
┌─────────────────────────────────────────────────┐
│ AgentSession (Orchestrator) │
│ ├─ Shared VAD, STT, TTS, LLM services │
│ ├─ Shared UserData context │
│ └─ Agent lifecycle management │
└─────────────────────────────────────────────────┘
│
┌─────────────┼─────────────┐
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Agent A │ │ Agent B │ │ Agent C │
│ ├─Instructions │ ├─Instructions │ ├─Instructions
│ ├─Tools │ ├─Tools │ ├─Tools
│ └─Handoff │ └─Handoff │ └─Handoff
└─────────┘ └─────────┘ └─────────┘
Load core documentation:
https://docs.livekit.io/agents/https://docs.livekit.io/agents/build/https://docs.livekit.io/agents/build/workflows/https://docs.livekit.io/agents/build/testing/Study example implementations:
https://github.com/livekit-examples/agent-starter-pythonhttps://github.com/livekit-examples/multi-agent-pythonhttps://github.com/livekit/agents/tree/main/examples/voice_agentsLoad reference documentation:
Determine your agent workflow:
Customer Support Pattern:
Greeting Agent → Triage Agent → Technical Support → Escalation Agent
Sales Pipeline Pattern:
Intro Agent → Qualification Agent → Demo Agent → Account Executive Handoff
Service Workflow Pattern:
Reception Agent → Information Gathering → Specialist Agent → Confirmation Agent
Plan your agents:
Create a dataclass to store information that persists across agents:
from dataclasses import dataclass, field
@dataclass
class ConversationData:
"""Shared context across all agents"""
user_name: str = ""
user_email: str = ""
issue_category: str = ""
collected_details: list[str] = field(default_factory=list)
escalation_needed: bool = False
# Add fields relevant to your use case
Use the provided template as a starting point:
your-agent-project/
├── src/
│ ├── agent.py # Main entry point
│ ├── agents/
│ │ ├── __init__.py
│ │ ├── intro_agent.py # Initial agent
│ │ ├── specialist_agent.py
│ │ └── escalation_agent.py
│ ├── models/
│ │ └── shared_data.py # UserData dataclass
│ └── tools/
│ └── custom_tools.py # Business-specific tools
├── tests/
│ └── test_agent.py # pytest tests
├── pyproject.toml # Dependencies with uv
├── .env.example # Environment variables template
├── Dockerfile # Container definition
└── README.md
Use the quick start script or copy template files:
./templates/ directoryInstall uv package manager:
curl -LsSf https://astral.sh/uv/install.sh | sh
Create project with dependencies:
# Initialize project
uv init your-agent-project
cd your-agent-project
# Add dependencies
uv add "livekit-agents>=1.3.3"
uv add "livekit-plugins-openai" # For OpenAI LLM & TTS
uv add "livekit-plugins-deepgram" # For Deepgram STT
uv add "livekit-plugins-silero" # For Silero VAD
uv add "python-dotenv" # For environment variables
# Add testing dependencies
uv add --dev "pytest"
uv add --dev "pytest-asyncio"
Set up environment variables:
# Copy from template
cp .env.example .env
# Edit with your credentials
# LIVEKIT_URL=wss://your-livekit-server.com
# LIVEKIT_API_KEY=your-api-key
# LIVEKIT_API_SECRET=your-api-secret
# OPENAI_API_KEY=your-openai-key
# DEEPGRAM_API_KEY=your-deepgram-key
Create main entry point (src/agent.py):
Load the complete template: 🚀 Main Entry Point Template
Key patterns:
prewarm() to load static resources (VAD models) before sessions startAgentSession[YourDataClass] with shared services@server.rtc_session() decorator for the main handlerExample structure:
from livekit import rtc
from livekit.agents import (
Agent,
AgentSession,
JobContext,
JobProcess,
WorkerOptions,
cli,
)
from livekit.plugins import openai, deepgram, silero
import logging
from dotenv import load_dotenv
from agents.intro_agent import IntroAgent
from models.shared_data import ConversationData
load_dotenv()
logger = logging.getLogger("voice-agent")
def prewarm(proc: JobProcess):
"""Load static resources before sessions start"""
# Load VAD model once and reuse across sessions
proc.userdata["vad"] = silero.VAD.load()
async def entrypoint(ctx: JobContext):
"""Main agent entry point"""
logger.info("Starting voice agent session")
# Get prewarmed VAD
vad = ctx.proc.userdata["vad"]
# Initialize session with shared services
session = AgentSession[ConversationData](
vad=vad,
stt=deepgram.STT(model="nova-2-general"),
llm=openai.LLM(model="gpt-4o-mini"),
tts=openai.TTS(voice="alloy"),
userdata=ConversationData(),
)
# Connect to room
await ctx.connect()
# Start with intro agent
intro_agent = IntroAgent()
# Run session (handles all handoffs automatically)
await session.start(agent=intro_agent, room=ctx.room)
if __name__ == "__main__":
cli.run_app(
WorkerOptions(
entrypoint_fnc=entrypoint,
prewarm_fnc=prewarm,
)
)
Agent structure:
Each agent should:
Agent base class__init__Load templates:
Example agent with handoff:
from livekit.agents import Agent, RunContext
from livekit.agents.llm import function_tool
from typing import Annotated
from models.shared_data import ConversationData
from agents.specialist_agent import SpecialistAgent
class IntroAgent(Agent):
"""Initial agent that greets users and routes to specialists"""
def __init__(self):
super().__init__(
instructions="""You are a friendly voice assistant that helps customers.
Your role:
1. Greet the user warmly
2. Ask for their name and what they need help with
3. Gather basic information about their request
4. Transfer to a specialist agent when you have enough information
Be conversational, friendly, and efficient. Once you understand their
need and have their name, immediately transfer to the specialist."""
)
@function_tool
async def transfer_to_specialist(
self,
context: RunContext[ConversationData],
user_name: Annotated[str, "The user's name"],
issue_category: Annotated[str, "Category: technical, billing, or general"],
issue_description: Annotated[str, "Brief description of the user's issue"],
):
"""Transfer the conversation to a specialist agent.
Call this when you have gathered the user's name and understand
their issue well enough to categorize it.
"""
# Store data in shared context
context.userdata.user_name = user_name
context.userdata.issue_category = issue_category
context.userdata.collected_details.append(issue_description)
# Create and return specialist agent
specialist = SpecialistAgent(
category=issue_category,
chat_ctx=self.chat_ctx, # Preserve conversation history
)
return specialist, f"Let me connect you with our {issue_category} specialist."
Key handoff patterns:
context.userdata with collected informationchat_ctx=self.chat_ctx to maintain conversation(new_agent, transition_message)Add business-specific tools to your agents using @function_tool:
from livekit.agents.llm import function_tool
from livekit.agents import RunContext
from typing import Annotated
@function_tool
async def lookup_order_status(
context: RunContext,
order_id: Annotated[str, "The order ID to look up"],
) -> str:
"""Look up the status of an order by order ID.
Returns the current status, shipping info, and estimated delivery.
"""
# Your API call here
try:
# result = await your_api.get_order(order_id)
return f"Order {order_id} is currently being processed..."
except Exception as e:
raise ToolError(f"Could not find order {order_id}. Please verify the order ID.")
@function_tool
async def schedule_callback(
context: RunContext,
phone_number: Annotated[str, "Customer's phone number"],
preferred_time: Annotated[str, "Preferred callback time"],
) -> str:
"""Schedule a callback for the customer."""
# Your scheduling logic here
return f"Callback scheduled for {preferred_time}"
Best practices for tools:
Annotated to add parameter descriptionsToolErrorOverride services per agent:
Different agents can use different models:
from livekit.plugins import openai, elevenlabs
class EscalationAgent(Agent):
def __init__(self):
super().__init__(
instructions="You help escalate issues to human operators...",
# Use a different TTS for this agent
tts=elevenlabs.TTS(
voice="professional_voice_id",
),
# Use a more capable LLM
llm=openai.LLM(model="gpt-4o"),
)
Available plugins:
LLM Providers:
livekit-plugins-openai: GPT-4o, GPT-4o-minilivekit-plugins-anthropic: Claude Sonnet, Opuslivekit-plugins-groq: Fast Llama inferenceSTT Providers:
livekit-plugins-deepgram: Nova-2 modelslivekit-plugins-assemblyai: Universal streaminglivekit-plugins-google: Google Speech-to-TextTTS Providers:
livekit-plugins-openai: Natural voiceslivekit-plugins-elevenlabs: High-quality voiceslivekit-plugins-cartesia: Low-latency Sonic modelsVAD:
livekit-plugins-silero: Multilingual voice detectionLiveKit provides a testing framework with pytest integration.
Load testing guide: 🧪 Complete Testing Guide
Example test structure:
import pytest
from livekit.agents import AgentSession
from livekit.plugins import openai
from agents.intro_agent import IntroAgent
from models.shared_data import ConversationData
@pytest.mark.asyncio
async def test_intro_agent_greeting():
"""Test that intro agent greets user properly"""
async with AgentSession(
llm=openai.LLM(model="gpt-4o-mini"),
userdata=ConversationData(),
) as sess:
agent = IntroAgent()
await sess.start(agent)
result = await sess.run(user_input="Hello")
# Assert greeting behavior
result.expect.next_event().is_message(role="assistant")
result.expect.contains_message("help")
@pytest.mark.asyncio
async def test_handoff_to_specialist():
"""Test that agent hands off correctly with context"""
async with AgentSession(
llm=openai.LLM(model="gpt-4o-mini"),
userdata=ConversationData(),
) as sess:
agent = IntroAgent()
await sess.start(agent)
result = await sess.run(
user_input="Hi, I'm John and I need help with my billing"
)
# Expect function call for handoff
result.expect.next_event().is_function_call(name="transfer_to_specialist")
# Verify userdata was updated
assert sess.userdata.user_name == "John"
assert "billing" in sess.userdata.issue_category.lower()
@pytest.mark.asyncio
async def test_tool_usage():
"""Test that agent correctly uses custom tools"""
async with AgentSession(
llm=openai.LLM(model="gpt-4o-mini"),
userdata=ConversationData(),
) as sess:
agent = SpecialistAgent(category="technical")
await sess.start(agent)
result = await sess.run(
user_input="What's the status of order #12345?"
)
# Expect tool call
result.expect.next_event().is_function_call(name="lookup_order_status")
result.expect.next_event().is_function_call_output()
Testing areas:
Run tests:
# Run all tests
uv run pytest
# Run with verbose output
uv run pytest -v
# Run specific test
uv run pytest tests/test_agent.py::test_handoff_to_specialist
Before deployment, verify:
Code Quality:
Functionality:
Performance:
Testing:
Load Dockerfile template: 🐳 Dockerfile Template
Example Dockerfile:
FROM python:3.11-slim
WORKDIR /app
# Install uv
RUN pip install uv
# Copy project files
COPY pyproject.toml uv.lock ./
COPY src/ ./src/
# Install dependencies
RUN uv sync --frozen
# Run agent
CMD ["uv", "run", "python", "src/agent.py", "start"]
Build and run:
# Build image
docker build -t your-voice-agent .
# Run container
docker run -d \
--env-file .env \
--name voice-agent \
your-voice-agent
Production environment variables:
# LiveKit Connection
LIVEKIT_URL=wss://your-production-server.com
LIVEKIT_API_KEY=your-production-key
LIVEKIT_API_SECRET=your-production-secret
# AI Services
OPENAI_API_KEY=sk-...
DEEPGRAM_API_KEY=...
# Agent Configuration
LOG_LEVEL=INFO
NUM_IDLE_PROCESSES=3 # Number of warmed processes to keep ready
Add logging:
import logging
logger = logging.getLogger("voice-agent")
logger.setLevel(logging.INFO)
# In your agents
logger.info(f"Starting session with user: {context.userdata.user_name}")
logger.info(f"Handoff from {self.__class__.__name__} to SpecialistAgent")
logger.error(f"Tool execution failed: {error}")
Track metrics:
from livekit.agents import metrics
# Create usage collector
collector = metrics.UsageCollector()
# In entrypoint
session = AgentSession[ConversationData](
# ... other params
usage_collector=collector,
)
# Log usage on completion
@ctx.on("agent_completed")
async def log_metrics():
logger.info(f"Session usage: {collector.get_summary()}")
Monitor:
Worker Options:
cli.run_app(
WorkerOptions(
entrypoint_fnc=entrypoint,
prewarm_fnc=prewarm,
num_idle_processes=3, # Processes to keep warm
)
)
Production settings:
num_idle_processes=0 (no warming)num_idle_processes=3+ (keep processes ready)Kubernetes deployment:
# Entry flow
GreetingAgent → TriageAgent → SupportAgent → EscalationAgent
↓
(Resolves issue or escalates)
Use when:
IntroAgent → QualificationAgent → DemoAgent → HandoffAgent
↓
(Disqualified → FollowUpAgent)
Use when:
WelcomeAgent → DataCollectionAgent → VerificationAgent → ConfirmationAgent
Use when:
RouterAgent ─┬→ TechnicalAgent
├→ BillingAgent
├→ SalesAgent
└→ GeneralAgent
Use when:
✅ DO:
❌ DON'T:
Good handoff triggers:
Poor handoff triggers:
Always preserve:
chat_ctx)Consider resetting:
Effective tools:
Tool organization:
Symptoms: Agent doesn't call transfer function
Solutions:
@function_tool)Symptoms: New agent doesn't know previous information
Solutions:
context.userdata is updated before handoffchat_ctx=self.chat_ctx to preserve historySymptoms: Audio cutting out, robotic voice
Solutions:
Symptoms: Agent doesn't use available tools
Solutions:
Symptoms: Slow responses, delays
Solutions:
prewarm()ctx.connect()The templates and patterns in this skill support various use cases:
Flow: Welcome → Menu Navigation → Order Taking → Payment → Confirmation
Implementation: Use Linear Pipeline pattern from Multi-Agent Patterns with the OrderData model from shared_data.py.
Flow: Greeting → Triage → Troubleshooting → Resolution/Escalation
Implementation: Use Escalation Hierarchy pattern with the SupportTicket model. See the provided templates for intro, specialist, and escalation agents.
Flow: Reception → Availability Check → Booking → Confirmation
Implementation: Use Linear Pipeline pattern. Customize ConversationData to track appointment details, availability, and booking confirmation.
Note: The templates in ./templates/ provide a complete working implementation. Adapt the agents and data models to your specific use case.
Load these resources as needed:
https://docs.livekit.io/agents/https://docs.livekit.io/agents/build/https://docs.livekit.io/agents/build/workflows/https://docs.livekit.io/agents/build/tools/https://docs.livekit.io/agents/build/testing/https://github.com/livekit-examples/agent-starter-pythonhttps://github.com/livekit-examples/multi-agent-pythonhttps://github.com/livekit/agents/tree/main/examples/voice_agentsFor a fast start with a working example:
./scripts/quickstart.sh my-agent-projectThis creates a complete project with:
https://cloud.livekit.iohttps://github.com/livekit-examples for more patternshttps://docs.livekit.io/reference/python/For issues or questions:
https://github.com/livekit/agents/issues