Live kit agents notes

🎯 High-Level Overview

The LiveKit Agents framework is designed to build conversational AI agents that can:

• Hear (STT - Speech-to-Text)
• Understand (LLM - Language Model)
• Speak (TTS - Text-to-Speech)
• React (Function tools, handoffs, interruptions)

---

🏗️ Core Architecture Components

1. AgentSession (The Brain)

Location: livekit/agents/voice/agent_session.py

The AgentSession is the main runtime orchestrator that glues everything together.

What it does:

• Manages the entire conversation lifecycle
• Connects audio/video I/O with STT, TTS, LLM
• Handles turn detection and interruptions
• Maintains chat history
• Manages agent switching (multi-agent scenarios)

Key properties:

session = AgentSession(
    stt=STT_ENGINE,           # Speech recognition
    llm=LLM_ENGINE,           # Intelligence
    tts=TTS_ENGINE,           # Voice synthesis
    vad=VAD_ENGINE,           # Voice activity detection
    tools=[...],              # Function tools
    turn_handling={...},      # Interrupt & endpointing config
)

---

2. Agent (The Personality)

Location: livekit/agents/voice/agent.py

An Agent is a persona with instructions and behaviors.

What it does:

• Defines instructions/system prompt
• Holds tools specific to this agent
• Implements lifecycle hooks (on_enter, on_user_turn_completed)
• Can be swapped during a session (multi-agent)

Simple example:

class MyAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a helpful assistant",
            tools=[my_tool_func],
        )
    
    async def on_enter(self):
        # Called when agent starts
        self.session.generate_reply(
            instructions="Say hello"
        )

---

3. AgentActivity (The Worker)

Location: livekit/agents/voice/agent_activity.py

The AgentActivity is the active runtime for an Agent - it does the actual work.

What it does:

• Processes audio streams
• Runs STT, LLM, TTS pipelines
• Handles function tool execution
• Manages interruptions and turn-taking

Internal tasks:

• _audio_recognition: Processes incoming user audio
• _realtime_reply_task: Handles realtime API responses
• _pipeline_reply_task: Handles standard STT→LLM→TTS pipeline

---

🔄 Simple Request-Response Flow

Here’s how a basic conversation flows through the system:

┌──────────────────────────────────────────────────────────────���──┐
│                       USER SPEAKS                               │
└────────────────────────────┬────────────────────────────────────┘
                             ↓
┌─────────────────────────────────────────────────────────────────┐
│  VAD (Voice Activity Detector) - Silero                         │
│  → "Is the user speaking?"                                      │
│  → YES: Mark as "speaking" state, start recording              │
└────────────────────────────┬────────────────────────────────────┘
                             ↓
┌─────────────────────────────────────────────────────────────────┐
│  STT (Speech-to-Text) - Deepgram                               │
│  → "What did they say?"                                        │
│  → Returns: "What's the weather?"                              │
└────────────────────────────┬────────────────────────────────────┘
                             ↓
┌─────────────────────────────────────────────────────────────────┐
│  Turn Endpointing                                               │
│  → Waits for silence (0.5-3 seconds configurable)              │
│  → User turn is "complete"                                     │
└────────────────────────────┬────────────────────────────────────┘
                             ↓
┌─────────────────────────────────────────────────────────────────┐
│  LLM (Language Model) - OpenAI GPT-4                           │
│  → Chat history + "What's the weather?"                        │
│  → Thinks... → "The weather is sunny"                          │
│  → May call tools if needed                                    │
└────────────────────────────┬────────────────────────────────────┘
                             ↓
┌─────────────────────────────────────────────────────────────────┐
│  TTS (Text-to-Speech) - Cartesia                               │
│  → "The weather is sunny"                                      │
│  → Returns: Audio frames                                       │
└────────────────────────────┬────────────────────────────────────┘
                             ↓
┌─────────────────────────────────────────────────────────────────┐
│               AGENT SPEAKS TO USER                              │
│  Audio frames sent to speakers/network                         │
└─────────────────────────────────────────────────────────────────┘

---

📊 Key Component Interactions

Component Matrix

Component	Role	Provided By	Example
STT	Converts audio → text	Plugin	Deepgram, OpenAI
LLM	Processes text → response	Plugin	GPT-4, Claude
TTS	Converts text → audio	Plugin	Cartesia, OpenAI
VAD	Detects voice activity	Plugin	Silero, built-in
Interruption Detection	Detects user cutoffs	Built-in	Adaptive transformer
Tools	Functions agent can call	User-defined	@function_tool
Chat Context	Conversation history	Built-in	Maintained automatically

---

🔌 Inference Model

LiveKit Inference provides a unified API for model access:

# Before (plugin-specific):
from livekit.plugins import openai, deepgram
stt = deepgram.STT(model="nova-3")
llm = openai.LLM(model="gpt-4-mini")
 
# After (unified):
from livekit.agents import inference
stt = inference.STT("deepgram/nova-3")
llm = inference.LLM("openai/gpt-4-mini")

Benefits:

• Switch providers easily
• Use LiveKit Cloud for unified billing
• Consistent error handling

---

🎯 Turn Detection & Interruptions

This is crucial for natural conversations.

Turn Handling Options:

session = AgentSession(
    turn_handling={
        "endpointing": {
            "mode": "vad",              # VAD or STT-based
            "min_delay": 0.5,           # Min silence before turn ends
            "max_delay": 3.0,           # Max wait time
        },
        "interruption": {
            "enabled": True,
            "mode": "adaptive",         # VAD or adaptive transformer
            "min_duration": 0.2,        # Min speech to interrupt
            "resume_after_interrupt": True,  # Resume if false interrupt
        },
    }
)

Interruption Flow:

1. User starts speaking while agent is speaking
2. Interruption detector triggers
3. Agent’s audio playback pauses
4. Agent shifts to “listening” state
5. Process user input
6. Generate reply

---

🛠️ Function Tools Architecture

Simple Tool Definition:

from livekit.agents import function_tool, RunContext
 
@function_tool
async def lookup_weather(context: RunContext, location: str):
    """Get weather for a location."""
    return {"weather": "sunny", "temp": 70}
 
# Use it
agent = Agent(
    instructions="You can look up weather",
    tools=[lookup_weather],
)

Tool Execution Flow:

┌──────────────────────────┐
│  LLM Response            │
│  "I'll check weather"    │
└────────────┬─────────────┘
             ↓
┌──────────────────────────┐
│  Function Call Detection │
│  tool_name: "lookup_weather"
│  args: {"location": "NYC"}
└────────────┬─────────────┘
             ↓
┌──────────────────────────┐
│  Execute Tool            │
│  lookup_weather("NYC")   │
│  → Returns weather data  │
└────────────┬─────────────┘
             ↓
┌──────────────────────────┐
│  Add to Chat Context     │
│  FunctionCallOutput      │
└────────────┬─────────────┘
             ↓
┌──────────────────────────┐
│  Generate Reply          │
│  "It's sunny in NYC!"    │
└──────────────────────────┘

Max tool steps: Configurable (default: 3 consecutive calls)

---

🔄 Multi-Agent Handoff

Switch between agents dynamically:

class IntroAgent(Agent):
    @function_tool
    async def gather_info(self, context: RunContext, name: str):
        # Return new agent and transition message
        story_agent = StoryAgent(name)
        return story_agent, "Now let me tell you a story!"
 
# Agent switching happens automatically
# Session maintains chat history across handoffs

---

📝 Session Lifecycle

Startup:

session = AgentSession(stt=..., llm=..., tts=...)
agent = MyAgent()
 
# Start session
await session.start(agent=agent, room=ctx.room)

During Session:

# Generate response
session.generate_reply(
    user_input="Hello",
    instructions="Be friendly"
)
 
# Agent can speak directly
session.say("Let me help you!")
 
# Interrupt current speech
session.interrupt()
 
# Commit user turn (flush STT buffer)
await session.commit_user_turn()

Shutdown:

await session.aclose()

---

🎛️ Configuration: Simple vs Complex

Simple (Works out-of-box):

session = AgentSession(
    stt="deepgram/nova-3",
    llm="openai/gpt-4-mini",
    tts="cartesia/sonic-3",
)

Advanced (Full control):

session = AgentSession(
    stt=deepgram.STT(model="nova-3", language="en"),
    llm=openai.LLM(model="gpt-4"),
    tts=cartesia.TTS(model="sonic-3", voice="id-123"),
    vad=silero.VAD.load(),
    turn_handling={
        "endpointing": {"mode": "vad", "min_delay": 0.3, "max_delay": 2.5},
        "interruption": {"enabled": True, "mode": "adaptive"},
    },
    max_tool_steps=5,
    min_consecutive_speech_delay=0.1,
    aec_warmup_duration=2.0,  # Echo cancellation warmup
)

---

📊 Agent States

The agent cycles through states:

initializing → listening → speaking → thinking → listening → ...
     ↑                                                      ↓
     └──────────────────────────────────────────────────────┘

User States:

• listening: User is quiet
• speaking: User is talking
• away: No activity for timeout duration

---

🎪 Real-time vs Pipeline Modes

Pipeline Mode (Standard):

Audio → STT → LLM → TTS → Audio

• More control, better for complex logic
• Slightly higher latency
• Default choice

Realtime Mode:

Audio ⇄ Realtime API (STT + LLM + TTS in one)

• Lower latency (150ms range)
• Streaming input/output
• Used by OpenAI Realtime API

from livekit.plugins import openai
 
session = AgentSession(
    llm=openai.realtime.RealtimeModel(voice="echo")
)

---

📚 Example: Complete Flow

from livekit.agents import AgentSession, Agent, JobContext
from livekit.plugins import openai, silero
 
class CustomerServiceAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You're a helpful customer service rep. Keep responses concise.",
            tools=[lookup_order, process_refund],
        )
    
    async def on_enter(self):
        self.session.generate_reply(
            instructions="Greet the user professionally"
        )
    
    async def on_user_turn_completed(self):
        # Called after user finishes speaking
        pass
 
@server.rtc_session()
async def entrypoint(ctx: JobContext):
    session = AgentSession(
        stt="deepgram/nova-3",
        llm="openai/gpt-4-mini",
        tts="cartesia/sonic-3",
        vad=silero.VAD.load(),
    )
    
    agent = CustomerServiceAgent()
    await session.start(agent=agent, room=ctx.room)
    # Session runs until shutdown

---

🔍 Events & Monitoring

@session.on("user_state_changed")
def on_user_state(event):
    print(f"User: {event.old_state} → {event.new_state}")
 
@session.on("agent_state_changed")
def on_agent_state(event):
    print(f"Agent: {event.old_state} → {event.new_state}")
 
@session.on("user_input_transcribed")
def on_transcript(event):
    print(f"User said: {event.transcript}")
 
@session.on("session_usage_updated")
def on_usage(event):
    print(f"Used: {event.usage}")  # Token/billing info

---

🎁 Key Takeaways

Concept	Purpose
AgentSession	Orchestrates everything - the runtime
Agent	Defines personality, instructions, tools
AgentActivity	Active execution of an Agent
STT/LLM/TTS	The modular AI pipeline
Tools	Give agents abilities to take action
Turn Handling	Natural conversation flow (interruptions, endpointing)
Chat Context	Persistent conversation history
AgentServer	Manages job scheduling across agents

---

📖 Architecture Diagram

┌─────────────────────────────────────────────────────────────┐
│                      AgentServer                             │
│         (Job scheduling & dispatch)                         │
└────────────────────┬────────────────────────────────────────┘
                     ↓
┌─────────────────────────────────────────────────────────────┐
│                    AgentSession                              │
│  ┌──────────────────────────────────────────────────────┐  │
│  │  Input                 Core            Output         │  │
│  │  ────────            ──────            ──────         │  │
│  │  Audio    ────→   AgentActivity  ────→  Audio        │  │
│  │  Video    ────→   [Agent Loop]   ────→  Video        │  │
│  │  Text     ────→   Turn Handling  ────→  Text         │  │
│  │                    Chat History                       │  │
│  │                                                       │  │
│  │  Components:                                         │  │
│  │  • STT (Speech→Text)                                │  │
│  │  • LLM (Text→Response)                              │  │
│  │  • TTS (Response→Speech)                            │  │
│  │  • VAD (Voice Detection)                            │  │
│  │  • Tools (Function Execution)                       │  │
│  │  • Turn Detector (Interruption/Endpointing)        │  │
│  └──────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                     ↑
         ┌───────────┴───────────┐
         ↓                       ↓
    ┌─────────┐            ┌─────────┐
    │  Agent  │            │  Agent  │
    │    1    │            │    2    │
    │ (Can    │            │(Swapped │
    │  switch)│            │ in)     │
    └─────────┘            └─────────┘

---

This framework makes building voice AI agents straightforward by handling complex real-time audio streaming, turn-taking logic, and LLM orchestration transparently! 🚀

LiveKit Agents: Deep Dive - AgentSession, AgentActivity & STT→LLM→TTS Pipeline

This is an in-depth architectural guide covering how the framework processes audio through the complete pipeline.

📊 Complete Architecture Overview

┌─────────────────────────────────────────────────────────────────────────┐
│                              AgentSession                                │
│  ┌───────────────────────────────────────────────────────────────────┐  │
│  │                        AgentActivity                               │  │
│  │  ┌──────────────────────────────────────────────────────────────┐ │  │
│  │  │                   Audio Recognition                           │ │  │
│  │  │  (STT → VAD → Turn Detection → Chat Context)                │ │  │
│  │  └──────────────────┬───────────────────────────────────────────┘ │  │
│  │                     │                                               │  │
│  │                     ↓                                               │  │
│  │  ┌──────────────────────────────────────────────────────────────┐ │  │
│  │  │         LLM Generation (Agent.llm_node)                      │ │  │
│  │  │  Chat Context → LLM Stream → Text Chunks → Tools             │ │  │
│  │  └──────────────────┬───────────────────────────────────────────┘ │  │
│  │                     │                                               │  │
│  │       ┌─────────────┴─────────────┐                                │  │
│  │       ↓                           ↓                                │  │
│  │  ┌─────────────┐          ┌──────────────┐                        │  │
│  │  │ TTS Node    │          │ Transcription│                        │  │
│  │  │ (Audio)     │          │ Node (Text)  │                        │  │
│  │  └──────┬──────┘          └──────┬───────┘                        │  │
│  │         │                        │                                │  │
│  │  ┌──────↓────────────────────────↓──────┐                        │  │
│  │  │   Audio Output       Text Output      │                        │  │
│  │  │   (Playback)         (Transcription)  │                        │  │
│  │  └───────────────────────────────────────┘                        │  │
│  └──────────────────────────────────────────────────────────────────┘ │  │
│                                                                          │  │
│  Input → STT → LLM → TTS → Output                                      │  │
│  (Audio)  (Text) (Response) (Audio)                                    │  │
└─────────────────────────────────────────────────────────────────────────┘

---

🔧 Core Classes & Their Responsibilities

1. AgentSession (The Orchestrator)

File: livekit/agents/voice/agent_session.py

Purpose: Manages the entire lifecycle and state of a conversation

class AgentSession:
    """The top-level runtime that:
    1. Connects audio/video I/O
    2. Manages STT, LLM, TTS instances
    3. Handles turn detection & interruptions
    4. Maintains chat history
    5. Manages agent switching
    """
    
    # Key properties
    stt: STT | None              # Speech recognition engine
    llm: LLM | RealtimeModel     # Language model
    tts: TTS | None              # Text-to-speech engine
    vad: VAD | None              # Voice activity detector
    
    # Key methods
    async def start(agent, room)           # Start session
    def generate_reply(user_input)         # Generate response
    def interrupt(force=False)             # Interrupt current speech
    async def aclose()                     # Cleanup

State Management:

• _user_state: “listening” | “speaking” | “away”
• _agent_state: “initializing” | “listening” | “speaking” | “thinking”
• _activity: Current AgentActivity (the active agent’s worker)
• _chat_ctx: Global conversation history

---

2. AgentActivity (The Worker)

File: livekit/agents/voice/agent_activity.py

Purpose: Executes the actual audio processing for one agent

class AgentActivity:
    """Handles:
    1. Audio input processing (push_audio)
    2. STT streaming
    3. LLM generation (with tool execution)
    4. TTS synthesis
    5. Audio/text output forwarding
    """
    
    # Key properties
    agent: Agent                 # The agent being executed
    session: AgentSession        # Parent session
    
    # Resolves to session if agent doesn't define
    @property
    def stt(self) -> STT:
        return self.agent.stt or self.session.stt
    
    @property
    def llm(self) -> LLM | RealtimeModel:
        return self.agent.llm or self.session.llm
    
    @property
    def tts(self) -> TTS:
        return self.agent.tts or self.session.tts
    
    # Key methods
    def push_audio(frame)                  # Push audio to be transcribed
    async def drain()                      # Wait for current turn to finish
    def interrupt(force=False)             # Interrupt agent speaking
    def say(text)                          # Agent says something directly
    def generate_reply(user_input)         # Generate response to user

Internal Tasks:

• _audio_recognition: Receives and transcribes user audio (STT)
• _pipeline_reply_task: Orchestrates LLM → TTS pipeline
• _realtime_reply_task: For Realtime API models
• Speech scheduling and interruption handling

---

🎯 End-to-End Flow: STT → LLM → TTS

Complete Request-Response Cycle

┌─────────────────────────────────────────────────────────────────────────┐
│                          USER SPEAKS (AUDIO)                             │
└────────────────────────────────┬────────────────────────────────────────┘
                                 ↓
┌─────────────────────────────────────────────────────────────────────────┐
│ STEP 1: AgentSession.input.audio receives frames                        │
│   → _forward_audio_task routes frames to AgentActivity.push_audio()     │
└────────────────────────────────┬────────────────────────────────────────┘
                                 ↓
┌─────────────────────────────────────────────────────────────────────────┐
│ STEP 2: AgentActivity._audio_recognition processes audio                │
│   ┌─────────────────────────────────────────────────────────────────┐  │
│   │ A. STT Pipeline (_AudioRecognition):                            │  │
│   │    • Receives audio frames from AgentActivity                   │  │
│   │    • Calls Agent.stt_node(audio) → STT.stream()               │  │
│   │    • STT receives audio and returns SpeechEvent(text=...)      │  │
│   │    ↓                                                             │  │
│   │ B. Emit "user_input_transcribed" event                         │  │
│   │    ↓                                                             │  │
│   │ C. Update user_state → "speaking"                              │  │
│   │    ↓                                                             │  │
│   │ D. Turn Detection & Endpointing:                               │  │
│   │    • VAD detects silence (min 0.5s → max 3s)                  │  │
│   │    • Emits turn_complete signal                                │  │
│   └────────────────────────────┬────────────────────────────────────┘  │
└────────────────────────────────┬────────────────────────────────────────┘
                                 ↓
┌─────────────────────────────────────────────────────────────────────────┐
│ STEP 3: on_user_turn_completed triggered                                │
│   → Calls _on_user_turn_completed() → generate_reply()                  │
└────────────────────────────────┬────────────────────────────────────────┘
                                 ↓
┌─────────────────────────────────────────────────────────────────────────┐
│ STEP 4: LLM Generation (_pipeline_reply_task)                           │
│   ┌─────────────────────────────────────────────────────────────────┐  │
│   │ A. Create ChatContext with user message + history              │  │
│   │    ↓                                                             │  │
│   │ B. Call Agent.llm_node(chat_ctx, tools) → LLM.chat()          │  │
│   │    ↓                                                             │  │
│   │ C. LLM streams chunks:                                         │  │
│   │    • ChatChunk(delta=ChoiceDelta(content="Hello"))            │  │
│   │    • ChatChunk(delta=ChoiceDelta(tool_calls=[...]))           │  │
│   │    • ChatChunk(usage=CompletionUsage(...))                    │  │
│   │    ↓                                                             │  │
│   │ D. Collect text in buffer (or execute tools)                  │  │
│   │    ↓                                                             │  │
│   │ E. Emit llm_node_ttft metric (Time-to-first-token)           │  │
│   └────────────────────────────┬────────────────────────────────────┘  │
└────────────────────────────────┬────────────────────────────────────────┘
                                 ↓
┌─────────────────────────────────────────────────────────────────────────┐
│ STEP 5: TTS Synthesis (_tts_task_impl or preemptive TTS)                │
│   ┌─────────────────────────────────────────────────────────────────┐  │
│   │ A. Text from LLM goes to Agent.tts_node()                      │  │
│   │    ↓                                                             │  │
│   │ B. Call TTS.stream() (for streaming) or .synthesize()          │  │
│   │    ↓                                                             │  │
│   │ C. TTS sends back audio frames (SynthesizedAudio)              │  │
│   │    ↓                                                             │  │
│   │ D. Emit tts_node_ttfb metric (Time-to-first-byte of audio)    │  │
│   └────────────────────────────┬──────────��─────────────────────────┘  │
└────────────────────────────────┬────────────────────────────────────────┘
                                 ↓
┌─────────────────────────────────────────────────────────────────────────┐
│ STEP 6: Output Forwarding                                               │
│   ┌─────────────────────────────────────────────────────────────────┐  │
│   │ A. Audio Output:                                               │  │
│   │    • perform_audio_forwarding() pushes frames to output.audio  │  │
│   │    • Plays to user via RTC track or custom handler            │  │
│   │    ↓                                                             │  │
│   │ B. Text Output:                                                │  │
│   │    • perform_text_forwarding() to output.transcription        │  │
│   │    • Sends transcript to client                                │  │
│   │    ↓                                                             │  │
│   │ C. Update agent_state → "speaking"                            │  │
│   │    ↓                                                             │  │
│   │ D. Add to chat context (ChatMessage with role="assistant")    │  │
│   └────────────────────────────┬────────────────────────────────────┘  │
└────────────────────────────────┬────────────────────────────────────────┘
                                 ↓
┌─────────────────────────────────────────────────────────────────────────┐
│ STEP 7: Wait for playout to finish                                      │
│   • Emit e2e_latency metric                                             │
│   • Update agent_state → "listening"                                    │
│   • Return to waiting for next user input                               │
└─────────────────────────────────────────────────────────────────────────┘

---

💻 Deep Dive: Key Classes

STT (Speech-to-Text)

File: livekit/agents/stt/stt.py

@dataclass
class STTCapabilities:
    streaming: bool                              # Stream audio or batch?
    interim_results: bool                        # Send partial results?
    diarization: bool = False                    # Multi-speaker detection
    aligned_transcript: Literal["word", "chunk", False] = False  # Word-level timestamps
    offline_recognize: bool = True               # Supports batch mode
 
@dataclass
class SpeechEvent:
    type: SpeechEventType  # START_OF_SPEECH | INTERIM_TRANSCRIPT | FINAL_TRANSCRIPT
    request_id: str
    alternatives: list[SpeechData]  # Multiple recognition hypotheses
    speech_start_time: float | None  # Server-reported onset time
 
class STT(ABC, EventEmitter):
    """Abstract base for all STT implementations"""
    
    @property
    def model(self) -> str:
        """Override to return model name"""
        return "unknown"
    
    @property
    def provider(self) -> str:
        """Override to return provider name"""
        return "unknown"
    
    @property
    def capabilities(self) -> STTCapabilities:
        return self._capabilities
    
    @abstractmethod
    async def _recognize_impl(
        self,
        buffer: AudioBuffer,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions,
    ) -> SpeechEvent:
        """Batch recognition - process entire audio buffer at once"""
        ...
    
    async def recognize(self, buffer, *, language=NOT_GIVEN, conn_options=...):
        """Public API with retry logic"""
        for i in range(conn_options.max_retry + 1):
            try:
                event = await self._recognize_impl(buffer, ...)
                self.emit("metrics_collected", STTMetrics(...))
                return event
            except APIError as e:
                if i == conn_options.max_retry:
                    raise
                await asyncio.sleep(retry_interval)
    
    def stream(self, *, language=NOT_GIVEN, conn_options=...) -> RecognizeStream:
        """Return a streaming recognition stream"""
        raise NotImplementedError(...)

RecognizeStream (for streaming STT):

class RecognizeStream(ABC):
    """Async generator-like interface for streaming STT"""
    
    def push_frame(self, frame: rtc.AudioFrame):
        """Push audio frame to be recognized"""
        self._input_ch.send_nowait(frame)
    
    def flush(self):
        """Mark end of segment"""
        self._input_ch.send_nowait(self._FlushSentinel())
    
    def end_input(self):
        """Mark end of entire input"""
        self.flush()
        self._input_ch.close()
    
    async def __anext__(self) -> SpeechEvent:
        """Async iteration - get next recognition event"""
        return await self._event_aiter.__anext__()

---

LLM (Large Language Model)

File: livekit/agents/llm/llm.py

class CompletionUsage(BaseModel):
    completion_tokens: int       # Output tokens
    prompt_tokens: int          # Input tokens  
    prompt_cached_tokens: int = 0
    total_tokens: int
    service_tier: str | None = None
 
class FunctionToolCall(BaseModel):
    type: Literal["function"] = "function"
    name: str                   # Function name to call
    arguments: str              # JSON string of args
    call_id: str                # Unique ID for this call
 
class ChatChunk(BaseModel):
    id: str
    delta: ChoiceDelta | None   # Incremental change
    usage: CompletionUsage | None
 
class ChoiceDelta(BaseModel):
    role: ChatRole | None = None    # system | user | assistant
    content: str | None = None      # Text content
    tool_calls: list[FunctionToolCall] = []
 
class LLM(ABC, EventEmitter):
    """Abstract base for all LLM implementations"""
    
    @property
    def model(self) -> str:
        return "unknown"
    
    @property
    def provider(self) -> str:
        return "unknown"
    
    @abstractmethod
    def chat(
        self,
        *,
        chat_ctx: ChatContext,
        tools: list[Tool] | None = None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> LLMStream:
        """Generate a chat completion stream"""
        ...
 
class LLMStream(ABC):
    """Async generator for LLM streaming"""
    
    async def __anext__(self) -> ChatChunk:
        """Get next chunk of generation"""
        ...
    
    async def collect(self) -> CollectedResponse:
        """Collect entire stream into single response"""
        text = ""
        tool_calls = []
        async for chunk in self:
            if chunk.delta:
                if chunk.delta.content:
                    text += chunk.delta.content
                if chunk.delta.tool_calls:
                    tool_calls.extend(chunk.delta.tool_calls)
        return CollectedResponse(text=text, tool_calls=tool_calls)

---

TTS (Text-to-Speech)

File: livekit/agents/tts/tts.py

@dataclass
class TTSCapabilities:
    streaming: bool              # WebSocket streaming or batch?
    aligned_transcript: bool = False  # Word-level timestamps in response?
 
@dataclass
class SynthesizedAudio:
    frame: rtc.AudioFrame        # Audio data
    request_id: str              # Provider request ID
    segment_id: str = ""         # Which segment (streaming only)
    is_final: bool = False       # Last frame of segment?
    delta_text: str = ""         # What text was synthesized
 
class TTS(ABC, EventEmitter):
    """Abstract base for all TTS implementations"""
    
    def __init__(
        self,
        *,
        capabilities: TTSCapabilities,
        sample_rate: int,
        num_channels: int,
    ):
        self._capabilities = capabilities
        self._sample_rate = sample_rate
        self._num_channels = num_channels
    
    @property
    def model(self) -> str:
        return "unknown"
    
    @property
    def provider(self) -> str:
        return "unknown"
    
    @abstractmethod
    def synthesize(
        self,
        text: str,
        *,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> ChunkedStream:
        """Synthesize single text chunk - returns async generator"""
        ...
    
    def stream(self, *, conn_options=...) -> SynthesizeStream:
        """Return a streaming synthesis stream"""
        raise NotImplementedError(...)

ChunkedStream (for non-streaming TTS):

class ChunkedStream(ABC):
    """For TTS providers that accept a text chunk and return audio"""
    
    async def __anext__(self) -> SynthesizedAudio:
        """Get next audio frame"""
        ...
    
    async def collect(self) -> rtc.AudioFrame:
        """Collect all frames into single audio"""
        frames = []
        async for ev in self:
            frames.append(ev.frame)
        return rtc.combine_audio_frames(frames)

SynthesizeStream (for streaming TTS):

class SynthesizeStream(ABC):
    """For TTS providers that accept streaming text and output audio"""
    
    def push_text(self, token: str):
        """Push text to be synthesized"""
        self._input_ch.send_nowait(token)
    
    def flush(self):
        """Mark end of current segment"""
        self._input_ch.send_nowait(self._FlushSentinel())
    
    def end_input(self):
        """Mark end of all input"""
        self.flush()
        self._input_ch.close()
    
    async def __anext__(self) -> SynthesizedAudio:
        """Get next audio frame"""
        ...

---

🛠️ Creating Custom STT, LLM, TTS

Custom STT Implementation

from livekit.agents import stt
from livekit import rtc
from typing import AsyncGenerator
 
class MyCustomSTT(stt.STT):
    def __init__(self):
        super().__init__(
            capabilities=stt.STTCapabilities(
                streaming=True,              # Supports real-time audio
                interim_results=True,        # Sends partial transcripts
                aligned_transcript="word",   # Word-level timestamps
            )
        )
        self._client = MySTTClient()  # Your API client
    
    @property
    def model(self) -> str:
        return "my-model-v1"
    
    @property
    def provider(self) -> str:
        return "my-company"
    
    # For batch recognition (non-streaming)
    async def _recognize_impl(
        self,
        buffer: AudioBuffer,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions,
    ) -> stt.SpeechEvent:
        """Process entire audio buffer at once"""
        audio_data = rtc.combine_audio_frames(buffer).to_wav_bytes()
        
        # Call your STT API
        response = await self._client.recognize(
            audio=audio_data,
            language=language or "en",
            timeout=conn_options.timeout
        )
        
        # Convert to SpeechEvent
        return stt.SpeechEvent(
            type=stt.SpeechEventType.FINAL_TRANSCRIPT,
            request_id=response.request_id,
            alternatives=[
                stt.SpeechData(
                    text=response.transcript,
                    language=LanguageCode("en"),
                    confidence=response.confidence,
                    words=[
                        stt.TimedString(
                            text=word.text,
                            start_time=word.start,
                            end_time=word.end
                        )
                        for word in response.words
                    ]
                )
            ]
        )
    
    # For streaming recognition
    def stream(
        self,
        *,
        language: NotGivenOr[str] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> stt.RecognizeStream:
        return MySTTStream(
            stt=self,
            language=language,
            conn_options=conn_options,
            client=self._client
        )
 
class MySTTStream(stt.RecognizeStream):
    def __init__(self, stt, language, conn_options, client):
        super().__init__(stt=stt, conn_options=conn_options)
        self._language = language
        self._client = client
        self._ws = None
    
    async def _run(self) -> None:
        """Main loop - connect to server, push/receive data"""
        # Connect WebSocket
        self._ws = await self._client.connect_stream(
            language=self._language or "en"
        )
        
        try:
            # Send audio as it comes in
            async def _send_audio():
                async for item in self._input_ch:
                    if isinstance(item, self._FlushSentinel):
                        await self._ws.send_message({"type": "flush"})
                    else:
                        # item is rtc.AudioFrame
                        await self._ws.send_audio(item.data)
            
            send_task = asyncio.create_task(_send_audio())
            
            # Receive transcripts
            async for transcript in self._ws.receive_transcripts():
                event = stt.SpeechEvent(
                    type=stt.SpeechEventType.FINAL_TRANSCRIPT 
                        if transcript.is_final 
                        else stt.SpeechEventType.INTERIM_TRANSCRIPT,
                    request_id=transcript.id,
                    alternatives=[
                        stt.SpeechData(
                            text=transcript.text,
                            language=LanguageCode("en"),
                            confidence=transcript.confidence,
                        )
                    ]
                )
                self._event_ch.send_nowait(event)
        
        finally:
            await send_task
            await self._ws.close()

Custom LLM Implementation

from livekit.agents import llm
 
class MyCustomLLM(llm.LLM):
    def __init__(self, api_key: str, model: str = "my-model"):
        super().__init__()
        self._api_key = api_key
        self._model = model
        self._client = MyLLMClient(api_key=api_key)
    
    @property
    def model(self) -> str:
        return self._model
    
    @property
    def provider(self) -> str:
        return "my-company"
    
    def chat(
        self,
        *,
        chat_ctx: llm.ChatContext,
        tools: list[llm.Tool] | None = None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: NotGivenOr[llm.ToolChoice] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> llm.LLMStream:
        return MyLLMStream(
            llm=self,
            chat_ctx=chat_ctx,
            tools=tools or [],
            conn_options=conn_options,
            client=self._client,
            tool_choice=tool_choice,
        )
 
class MyLLMStream(llm.LLMStream):
    def __init__(self, llm, chat_ctx, tools, conn_options, client, tool_choice):
        super().__init__(llm=llm, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options)
        self._client = client
        self._tool_choice = tool_choice
    
    async def _run(self) -> None:
        """Stream LLM response"""
        # Build messages from chat context
        messages = []
        for msg in self._chat_ctx.messages:
            messages.append({
                "role": msg.role,
                "content": msg.text_content or ""
            })
        
        # Convert tools to provider format
        tools_payload = []
        for tool in self._tools:
            tools_payload.append({
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.raw_schema
            })
        
        # Stream from API
        async for chunk in self._client.stream_completion(
            model=self._llm.model,
            messages=messages,
            tools=tools_payload if tools_payload else None,
            tool_choice=self._tool_choice,
            timeout=self._conn_options.timeout
        ):
            # Parse chunk
            if chunk.type == "text_delta":
                chat_chunk = llm.ChatChunk(
                    id=chunk.id,
                    delta=llm.ChoiceDelta(content=chunk.text)
                )
            elif chunk.type == "tool_call_delta":
                chat_chunk = llm.ChatChunk(
                    id=chunk.id,
                    delta=llm.ChoiceDelta(
                        tool_calls=[
                            llm.FunctionToolCall(
                                name=chunk.tool_name,
                                arguments=chunk.tool_args,
                                call_id=chunk.call_id
                            )
                        ]
                    )
                )
            elif chunk.type == "usage":
                chat_chunk = llm.ChatChunk(
                    id=chunk.id,
                    usage=llm.CompletionUsage(
                        completion_tokens=chunk.completion_tokens,
                        prompt_tokens=chunk.prompt_tokens,
                        total_tokens=chunk.total_tokens
                    )
                )
            
            self._event_ch.send_nowait(chat_chunk)

Custom TTS Implementation

from livekit.agents import tts
from livekit import rtc
 
class MyCustomTTS(tts.TTS):
    def __init__(self, api_key: str, voice: str = "default"):
        super().__init__(
            capabilities=tts.TTSCapabilities(
                streaming=True,           # Supports streaming
                aligned_transcript=True   # Can provide word timestamps
            ),
            sample_rate=24000,
            num_channels=1
        )
        self._api_key = api_key
        self._voice = voice
        self._client = MyTTSClient(api_key=api_key)
    
    @property
    def model(self) -> str:
        return "my-tts-v1"
    
    @property
    def provider(self) -> str:
        return "my-company"
    
    def synthesize(
        self,
        text: str,
        *,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
    ) -> tts.ChunkedStream:
        # For non-streaming TTS, wrap stream implementation
        return self._synthesize_with_stream(text, conn_options=conn_options)
    
    def stream(
        self,
        *,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
    ) -> tts.SynthesizeStream:
        return MyTTSStream(
            tts=self,
            conn_options=conn_options,
            client=self._client,
            voice=self._voice
        )
 
class MyTTSStream(tts.SynthesizeStream):
    def __init__(self, tts, conn_options, client, voice):
        super().__init__(tts=tts, conn_options=conn_options)
        self._client = client
        self._voice = voice
        self._ws = None
    
    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        """Stream TTS generation"""
        # Initialize output with audio properties
        output_emitter.initialize(
            request_id=shortuuid(),
            sample_rate=self._tts.sample_rate,
            num_channels=self._tts.num_channels,
            mime_type="audio/pcm",
            stream=True
        )
        
        # Connect to TTS API WebSocket
        self._ws = await self._client.connect_stream(
            voice=self._voice,
            sample_rate=self._tts.sample_rate
        )
        
        try:
            # Forward input text to server
            async def _send_text():
                async for item in self._input_ch:
                    if isinstance(item, self._FlushSentinel):
                        await self._ws.send({"type": "flush"})
                    else:
                        await self._ws.send({"type": "text", "content": item})
            
            send_task = asyncio.create_task(_send_text())
            
            # Receive audio frames from server
            segment_id = shortuuid()
            output_emitter.start_segment(segment_id=segment_id)
            
            async for audio_chunk in self._ws.receive_audio():
                # Decode/parse audio from provider
                if audio_chunk.format == "pcm":
                    output_emitter.push(audio_chunk.data)
                
                # Handle segment boundaries
                if audio_chunk.is_segment_end:
                    output_emitter.end_segment()
                    segment_id = shortuuid()
                    output_emitter.start_segment(segment_id=segment_id)
            
        finally:
            await send_task
            output_emitter.end_input()
            await self._ws.close()

---

🔄 AgentSession → AgentActivity Data Flow

Session Start: await session.start(agent, room)

# Inside AgentSession.start():
async def start(self, agent, room=None):
    self._agent = agent
    
    # Create AgentActivity to execute the agent
    self._activity = AgentActivity(agent, self)
    
    # Start the activity
    await self._activity.start()
    
    # Start forwarding audio/video
    self._forward_audio_atask = asyncio.create_task(
        self._forward_audio_task()
    )
 
async def _forward_audio_task(self):
    """Routes incoming audio to the activity"""
    async for frame in self.input.audio:
        if self._activity:
            self._activity.push_audio(frame)

User Speaks: Audio enters the pipeline

# Audio arrives from RTC room/input
# ↓
# _forward_audio_task receives it
# ↓
# AgentActivity.push_audio(frame)
#   ↓
#   _audio_recognition.push_audio(frame)
#     ↓
#     STT stream receives frame
#       ↓
#       STT API returns SpeechEvent
#         ↓
#         _on_user_turn_completed() if turn ended
#           ↓
#           generate_reply() called

---

📊 Configuration: Simple vs Advanced

Simple Setup:

session = AgentSession(
    stt="deepgram/nova-3:en",
    llm="openai/gpt-4-mini",
    tts="cartesia/sonic-3"
)

Advanced Setup with Custom Implementation:

session = AgentSession(
    stt=MyCustomSTT(api_key="..."),
    llm=MyCustomLLM(api_key="..."),
    tts=MyCustomTTS(api_key="..."),
    vad=silero.VAD.load(),
    turn_handling={
        "endpointing": {
            "mode": "vad",
            "min_delay": 0.3,
            "max_delay": 2.5
        },
        "interruption": {
            "enabled": True,
            "mode": "adaptive",
            "min_duration": 0.2
        }
    },
    max_tool_steps=5
)

---

🎯 Key Takeaways

Component	Purpose	Key Method
AgentSession	Orchestrates entire conversation	async def start()
AgentActivity	Executes one agent’s logic	def push_audio()
STT	Converts audio → text	def stream() or async def recognize()
LLM	Generates text response	def chat() returns LLMStream
TTS	Converts text → audio	def stream() or def synthesize()
ChatContext	Maintains history	Added via session
AudioRecognition	Manages STT pipeline	Internal to AgentActivity
SpeechHandle	Tracks individual responses	Manages interruptions, metrics

Flow Summary:

Audio (STT) → Text (LLM) → Response Text (TTS) → Audio (Output)
     ↑                                                    ↓
     └────────────────────────────────────────────────────┘
         (Turn Detection, Interruptions, Tools)

This architecture provides composability, streaming support, and real-time responsiveness! 🚀