Livekit Notes

LiveKit is an open-source WebRTC-based media server that implements a Selective Forwarding Unit (SFU) architecture. It’s written in Go and enables:

• Real-time video/audio conferencing
• Multi-user scalable media routing
• AI agent integration with voice, LLM, and TTS
• Advanced features like simulcast, selective subscription, and speaker detection

🏗️ Core Architecture

1. SFU (Selective Forwarding Unit) Model

Participant A ──→ LiveKit Server (SFU) ──→ Participant B
                        │
                        ├─→ Stream Allocator (bandwidth mgmt)
                        ├─→ Forwarder (RTP packet forwarding)
                        └─→ Buffer (packet buffering)

Key Principle: The server receives media from publishers, selectively forwards to subscribers based on bandwidth, layer requirements, and subscription permissions.

2. Directory Structure

pkg/
├── rtc/              # Real-time communication core
│   ├── room.go              # Room management
│   ├── participant.go        # Participant logic
│   ├── subscribedtrack.go    # Subscriber track handling
│   └── uptrackmanager.go     # Published track management
├── sfu/              # Media forwarding engine
│   ├── forwarder.go          # RTP packet forwarding logic
│   ├── downtrack.go          # Subscriber-side media track
│   ├── buffer/               # Packet buffering
│   └── streamallocator/      # Bandwidth allocation
├── service/          # Service layer
│   ├── roommanager.go        # Room lifecycle management
│   └── rtcservice.go         # RTC service HTTP interface
└── routing/          # Inter-node communication
    └── roommanager.go        # RoomManager client

cmd/server/          # Server entry point
└── main.go           # CLI and initialization

🔄 Participant Lifecycle & Connection Flow

Phase 1: Join Request

Client (Browser/Agent)
    ↓ (HTTP POST with JWT token)
    ↓
RTCService.serve() [pkg/service/rtcservice.go]
    ↓ (validates token & room)
    ↓
RoomManager.StartSession() [pkg/service/roommanager.go]
    ↓ (creates or gets room)
    ↓
Room.Join() [pkg/rtc/room.go]
    ↓ (adds participant to room)
    ↓
ParticipantImpl created

Key Interfaces

// LocalParticipant - represents a participant in the room
type LocalParticipant interface {
    ID() livekit.ParticipantID
    Identity() livekit.ParticipantIdentity
    Kind() livekit.ParticipantType
    
    // Publishing tracks
    GetPublishedTracks() []MediaTrack
    
    // Subscribing to tracks
    SubscribeToTrack(trackID TrackID) error
    
    // Signaling
    SendJoinResponse(*JoinResponse) error
    SendParticipantUpdate([]*ParticipantInfo) error
    SendSpeakerUpdate([]*SpeakerInfo, force bool) error
}
 
// Room - manages all participants and tracks
type Room struct {
    participants map[ParticipantIdentity]LocalParticipant
    trackManager *RoomTrackManager
    // ...media management
}

---

📡 Media Flow: Publish & Subscribe

Publishing (Uplink)

Publisher sends media
    ↓
ParticipantImpl.onMediaTrack() [pkg/rtc/participant.go]
    ↓ (WebRTC track received from publisher)
    ↓
mediaTrackReceived() creates MediaTrack
    ↓
MediaTrack added to UpTrackManager
    ↓
Room notified via listener: OnTrackPublished()
    ↓
Broadcast to all subscribers

Code Reference:

// pkg/rtc/participant.go - Line 2243
func (p *ParticipantImpl) onMediaTrack(rtcTrack *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
    track := sfu.NewTrackRemoteFromSdp(rtcTrack, codec)
    publishedTrack, _, _, _ := p.mediaTrackReceived(track, rtpReceiver)
    // Track is now available to subscribers
}

Subscribing (Downlink)

Subscriber joins / requests track
    ↓
Room.subscribeToExistingTracks() [pkg/rtc/room.go]
    ↓ (finds published tracks to subscribe to)
    ↓
Create DownTrack [pkg/sfu/downtrack.go]
    ↓ (represents the media stream to this subscriber)
    ↓
Add DownTrack to publisher's spreader
    ↓
Forwarder begins packet translation
    ↓
Packets forwarded to subscriber

DownTrack Flow:

// pkg/sfu/downtrack.go - Line 414
type DownTrack struct {
    id          TrackID                    // Track identifier
    kind        webrtc.RTPCodecType       // Audio or Video
    ssrc        uint32                    // Synchronization source
    receiver    TrackReceiver             // Reference to published track
    // ...
}
 
// Subscriber receives media through DownTrack
func (d *DownTrack) Kind() webrtc.RTPCodecType { return d.kind }
func (d *DownTrack) SSRC() uint32 { return d.ssrc }

---

🎯 Media Forwarding: The Forwarder

The Forwarder is the heart of the SFU - it handles RTP packet translation, layer selection, and bandwidth management.

Forwarder Responsibilities

// pkg/sfu/forwarder.go - Line 173
type Forwarder struct {
    mime                    mime.MimeType        // Codec type
    kind                    webrtc.RTPCodecType  // Audio/Video
    
    // Video layer selection
    vls                     videolayerselector.VideoLayerSelector
    
    // Bandwidth tracking
    rtpStats                *rtpstats.RTPStatsSender
    
    // Translation parameters
    rtpMunger               *RTPMunger           // Modifies RTP headers
    codecMunger             codecmunger.CodecMunger // Codec-specific handling
}

Forwarder Operations

Operation	Purpose	Key Method
DetermineCodec()	Identify codec and set up layer selection	DetermineCodec()
GetTranslationParams()	Calculate how to forward an RTP packet	GetTranslationParams()
AllocateOptimal()	Select video layer based on bandwidth	AllocateOptimal()
Pause()	Pause forwarding (muted or low bandwidth)	Pause()
Resync()	Re-synchronize forwarding stream	Resync()

Example: Video Layer Allocation

// pkg/sfu/forwarder.go - Line 968
func (f *Forwarder) AllocateOptimal(
    availableLayers []int32,  // Layers available from publisher
    brs Bitrates,             // Bandwidth information
    allowOvershoot bool,
    hold bool,
) VideoAllocation {
    // Determines which spatial/temporal layer to forward
    // Returns:
    // - TargetLayer: which layer to send
    // - BandwidthRequested: bandwidth needed
    // - PauseReason: why forwarding is paused (if any)
}

Bandwidth Management Pause Reasons

// pkg/sfu/forwarder.go - Line 63
const (
    VideoPauseReasonNone       = 0  // Normal operation
    VideoPauseReasonMuted      = 1  // Subscriber muted track
    VideoPauseReasonPubMuted   = 2  // Publisher muted track
    VideoPauseReasonFeedDry    = 3  // No data from publisher
    VideoPauseReasonBandwidth  = 4  // Insufficient bandwidth
)

---

🤖 Agent Integration Architecture

LiveKit has built-in support for AI Agents. Here’s how agents integrate:

Agent Types

// From README: LiveKit Agents SDK available at github.com/livekit/agents
// - Python agents (github.com/livekit/agents)
// - Node.js agents (github.com/livekit/agents-js)

Agent Dispatch & Job Management

// pkg/rtc/room.go - Line 68
type AgentStore interface {
    StoreAgentDispatch(ctx context.Context, dispatch *livekit.AgentDispatch) error
    DeleteAgentDispatch(ctx context.Context, dispatch *livekit.AgentDispatch) error
    ListAgentDispatches(ctx context.Context, roomName livekit.RoomName) ([]*livekit.AgentDispatch, error)
    
    StoreAgentJob(ctx context.Context, job *livekit.Job) error
    DeleteAgentJob(ctx context.Context, job *livekit.Job) error
}
 
// In Room struct - Line 84
type Room struct {
    agentClient      agent.Client
    agentStore       AgentStore
    agentDispatches  map[string]*agentDispatch
    agentParticpants map[livekit.ParticipantIdentity]*agentJob
    // ...
}

Agent Join Flow

Room.Join() triggered (either participant or agent)
    ↓
launchTargetAgents() called [pkg/rtc/room.go]
    ↓ (finds which agents should launch)
    ↓
AgentDispatch consulted (metadata about agent config)
    ↓
Agent job created
    ↓
Agent joins as special participant type
    ↓
Agent receives:
    ├─ AUDIO from participants
    ├─ VIDEO from participants (if enabled)
    └─ DATA packets for signaling

---

🎤 Voice, LLM, TTS Agent Flow (Detailed)

This is the typical flow for an AI voice assistant agent:

1. Agent Initialization

Agent SDK (Python/Node.js)
    ↓ (imports livekit.agents)
    ↓
Connect to LiveKit room with JWT token
    ↓
Agent.ParticipantType = AGENT (special flag)
    ↓
Room.Join() recognizes agent participant
    ↓
Agent receives live audio stream

2. Voice Input Pipeline (Speech Recognition)

Real-time Audio from Participants
    ↓
Audio Track received by Agent
    ↓
Agent processes audio chunks:
    ├─→ VAD (Voice Activity Detection) - determine if speech present
    ├─→ Audio buffering (typically 20ms chunks)
    └─→ Send to Speech-to-Text (STT/Transcription) service
            (Google Cloud Speech, Azure, Deepgram, etc.)
    ↓
Transcribed text output

Related Code:

// pkg/rtc/participant.go - Line 2243
// Audio tracks are received and made available to all subscribers
// An agent subscribes to all audio and receives the frames in real-time

3. LLM Processing

Transcribed Text (from STT)
    ↓
Agent sends to LLM (OpenAI, Anthropic, etc.)
    ├─ Context: conversation history
    ├─ System prompt: agent behavior/personality
    ├─ Temperature: randomness
    └─ Max tokens: response length
    ↓
LLM generates response text
    ↓ (streaming or batched response)

4. Voice Output Pipeline (Text-to-Speech)

LLM Response Text
    ↓
Send to TTS service (ElevenLabs, Google Cloud TTS, etc.)
    ↓
TTS generates audio waveform:
    ├─ Voice selection (male, female, tone)
    ├─ Speaking rate
    ├─ Pitch, emotion parameters
    └─ Output format: PCM, MP3, WAV
    ↓
Audio frames generated

5. Audio Publishing Back to Room

TTS Audio Frames
    ↓
Agent creates AudioTrack (as publisher)
    ↓
Frames written to track:
    for chunk in audio_frames:
        track.write_sample(chunk)
    ↓
Track encoded and sent via WebRTC
    ↓
Room.mediaTrackReceived() processes agent audio
    ↓
Forwarder forwards agent audio to all participants
    ↓
Participants hear agent speaking

Complete Agent Communication Loop

┌─────────────────────────────────────────────────┐
│  Participant speaks → Audio published to room   │
├─────────────────────────────────────────────────┤
│                                                 │
│  Agent receives audio frame                     │
│    ↓                                            │
│  STT processes: "Hello, how are you?"          │
│    ↓                                            │
│  LLM generates: "I'm doing well, thanks!"      │
│    ↓                                            │
│  TTS creates audio of response                  │
│    ↓                                            │
│  Agent publishes audio track back               │
│                                                 │
│  Participant receives agent audio               │
│    ↓                                            │
│  Hears: "I'm doing well, thanks!" (spoken)     │
│                                                 │
└─────────────────────────────────────────────────┘
        ← REPEAT (real-time conversation) →

---

📊 Key Data Structures

Room

type Room struct {
    protoRoom  *livekit.Room              // Room metadata
    participants map[ParticipantIdentity]LocalParticipant
    trackManager *RoomTrackManager        // Track registry
    agentDispatches map[string]*agentDispatch
}

Participant

type ParticipantImpl struct {
    identity    ParticipantIdentity
    id          ParticipantID
    kind        ParticipantType            // AGENT, UNSET_PARTICIPANT_TYPE, etc.
    state       ParticipantInfo_State      // ACTIVE, DISCONNECTED, etc.
    
    // Publishers (tracks this participant sends)
    upTrackManager *UpTrackManager
    
    // Subscribers (tracks this participant receives)
    subscriptionPermission *SubscriptionPermission
}

MediaTrack (Published)

type MediaTrack interface {
    ID() TrackID
    Kind() webrtc.RTPCodecType           // Audio or Video
    Source() TrackSource                 // CAMERA, MICROPHONE, SCREEN_SHARE
    
    AddSubscriber(sub LocalParticipant, downTrack *DownTrack)
    RemoveSubscriber(subID ParticipantID)
}

DownTrack (Subscriber View)

type DownTrack struct {
    id          TrackID
    kind        webrtc.RTPCodecType
    ssrc        uint32                   // Sender's synchronization source
    receiver    TrackReceiver            // The published track
    
    // Subscribers who are receiving this stream
    forwarders  map[ParticipantID]*Forwarder
}

---

🔐 Room Lifecycle

1. CREATE ROOM
   RoomManager.CreateRoom() → Room created in memory + stored

2. PARTICIPANT JOINS
   RoomManager.StartSession() → Participant added to room
   
3. MEDIA FLOWS
   - Participants publish audio/video
   - Tracks broadcasted to subscribers
   - Forwarders handle packet forwarding
   
4. AGENT JOINS (if configured)
   - Agent launches automatically or on-demand
   - Subscribes to participant media
   - Publishes responses
   
5. PARTICIPANT LEAVES
   Room.RemoveParticipant() → Participant removed, tracks cleaned

6. ROOM CLOSES
   - Last participant leaves
   - Room.CloseIfEmpty() checks timeout
   - Room deleted, resources freed

---

🌐 Distributed Architecture

LiveKit supports multiple nodes for scalability:

┌─────────────────┐     ┌─────────────────┐
│   RTC Node 1    │─────│   RTC Node 2    │
├─────────────────┤     ├─────────────────┤
│ Room A          │     │ Room B          │
│ Participant 1   │     │ Participant 3   │
│ Participant 2   │     │ Participant 4   │
└─────────────────┘     └─────────────────┘
        │                       │
        └───────────┬───────────┘
                    │
            ┌───────────────┐
            │ Message Bus   │
            │ (psRPC/Redis) │
            └───────────────┘

Routing Layer (pkg/routing/) manages:

• Room allocation to nodes
• Inter-node communication
• Participant migration between nodes

---

🔧 Configuration & Server Startup

Starting LiveKit Server

# Development mode (insecure, for testing)
livekit-server --dev
 
# Production with config file
livekit-server --config config.yaml
 
# Key configuration in config-sample.yaml:
# - RTC ports (UDP/TCP for media)
# - TURN server settings
# - API keys and authentication
# - Room timeout settings
# - Agent configuration

Server Entry Point

// cmd/server/main.go - Line 102
func main() {
    // 1. Parse CLI flags
    // 2. Load configuration
    // 3. Initialize logging
    // 4. Create RoomManager
    // 5. Start HTTP server
    // 6. Accept WebRTC connections
}

---

📈 Performance & Bandwidth Management

Stream Allocator

// pkg/sfu/streamallocator/streamallocator.go
// Manages bandwidth allocation across all tracks
// - Monitors bandwidth per subscriber
// - Selects optimal layers for each track
// - Handles congestion/recovery

Forward Stats

// pkg/sfu/forwardstats.go
// Tracks:
// - Latency (arrival → forwarding)
// - Packet loss
// - Bitrate

---

🧪 Testing & Debugging

Test Participant Creation

// pkg/rtc/room_test.go
newRoomWithParticipants(t, testRoomOpts{num: 3})
// Creates test room with mock participants

Logging

// Throughout codebase:
logger.Debugw("participant joined", "participantID", participant.ID())
logger.Errorw("could not join room", err)
 
// Run with: --dev flag for debug logging

---

🔗 Key External Dependencies

• Pion WebRTC - WebRTC implementation
• Protocol Buffers - Message serialization (github.com/livekit/protocol)
• psRPC - Inter-node communication
• Redis (optional) - Distributed state
• Prometheus - Metrics

---

📚 Key Files Reference

File	Purpose	Key Functions
cmd/server/main.go	Server entry point	main(), startServer()
pkg/service/roommanager.go	Room lifecycle	StartSession(), CreateRoom()
pkg/rtc/room.go	Room logic	Join(), RemoveParticipant()
pkg/rtc/participant.go	Participant state	onMediaTrack(), SubscribeToTrack()
pkg/sfu/forwarder.go	Media forwarding	GetTranslationParams(), AllocateOptimal()
pkg/sfu/downtrack.go	Subscriber track	NewDownTrack(), WriteRTP()
pkg/sfu/streamallocator/	Bandwidth mgmt	AddTrack(), RemoveTrack()

---

Complete Guide: LiveKit Voice Agents Architecture & Communication Flow

Part 1: How LiveKit Enables Twilio-Like Server Connections

Overview: Server-Controlled Call Handling

LiveKit provides server-based call handling similar to Twilio through:

1. Telephony Integration (SIP/PSTN)
2. Server-Side Agent Dispatch
3. Job Context & Room Management

Architecture Diagram

┌─────────────────────────────────────────────────────────────┐
│                    Client/Phone User                         │
│                                                              │
│  (Browser WebRTC Client OR Phone via SIP/PSTN)             │
└──────────────────────┬──────────────────────────────────────┘
                       │ WebRTC Audio/Video or SIP Call
                       ▼
┌─────────────────────────────────────────────────────────────┐
│              LiveKit Server (Selective Forwarding Unit)      │
│                                                              │
│  ┌────────────────────────────────────────────────────────┐│
│  │ Room Management (WebRTC SFU)                           ││
│  │ ├─ Manages participant connections                      ││
│  │ ├─ Routes media between participants                    ││
│  │ └─ Handles connection negotiation                       ││
│  └────────────────────────────────────────────────────────┘│
└─────────────────────────────────┬──────────────────────────┘
                                  │
                                  ▼
            ┌─────────────────────────────────────┐
            │   Agent Dispatcher / Job Queue      │
            │                                     │
            │ ┌────────────────────────────────┐ │
            │ │ Create dispatch request        │ │
            │ │ (includes room name & metadata)│ │
            │ └────────────────────────────────┘ │
            └────────────┬────────────────────────┘
                         │
                         ▼
            ┌─────────────────────────────────────┐
            │   LiveKit Agent Server              │
            │   (Runs your Python/Node.js code)  │
            │                                     │
            │  ┌───────────────────────────────┐ │
            │  │ JobContext                    │ │
            │  │ ├─ Room reference             │ │
            │  │ ├─ Connect to LiveKit server  │ │
            │  │ └─ Metadata (user data)       │ │
            │  └───────────────────────────────┘ │
            │                                     │
            │  ┌───────────────────────────────┐ │
            │  │ AgentSession                  │ │
            │  │ ├─ STT (Speech-to-Text)       │ │
            │  │ ├��� LLM (AI Brain)             │ │
            │  │ ├─ TTS (Text-to-Speech)       │ │
            │  │ └─ VAD (Voice Detection)      │ │
            │  └───────────────────────────────┘ │
            │                                     │
            │  ┌───────────────────────────────┐ │
            │  │ Agent (Your Code)             │ │
            │  │ ├─ System instructions        │ │
            │  │ ├─ Tool definitions           │ │
            │  │ └─ Custom logic               │ │
            │  └───────────────────────────────┘ │
            └─────────────────────────────────────┘

---

Part 2: Server Connection Flow (Like Twilio)

Step 1: Dispatch Request Created

# Client-side or backend server creates a dispatch
from livekit import api
 
lkapi = api.LiveKitAPI()
 
# Create a dispatch - this tells LiveKit to launch an agent
dispatch = await lkapi.agent_dispatch.create_dispatch(
    api.CreateAgentDispatchRequest(
        agent_name="my-phone-agent",
        room="customer-123",
        metadata=json.dumps({
            "user_request": "Check my checking balance",
            "phone_number": "+1-555-0123",
            "customer_id": "cust_456"
        })
    )
)

What happens:

• A dispatch is created in LiveKit’s database
• The AgentServer polls for new dispatch requests
• When found, it launches a new agent job with the given parameters

Step 2: Agent Server Receives Job

# Your agent server listening for jobs
from livekit import agents
 
server = agents.AgentServer()
 
@server.rtc_session(agent_name="my-phone-agent")
async def entrypoint(ctx: agents.JobContext) -> None:
    # ctx now contains:
    # - ctx.room: The LiveKit room to connect to
    # - ctx.job: The job details
    # - ctx.metadata: The user metadata from dispatch
    
    await ctx.connect()  # Connect to LiveKit
    
    # Extract metadata
    user_data = json.loads(ctx.job.metadata)
    user_request = user_data["user_request"]
    phone_number = user_data["phone_number"]
    
    print(f"Agent assigned to handle: {user_request}")
    print(f"Customer phone: {phone_number}")

Step 3: Agent Joins Room with Caller

Timeline:
┌─────────────────────────────────────────────────────────────┐
│ T0: Caller dials number / Client connects to room           │
├─────────────────────────────────────────────────────────────┤
│ T1: Dispatcher creates dispatch with caller's room name     │
├─────────────────────────────────────────────────────────────┤
│ T2: Agent server receives dispatch, spawns agent job        │
├─────────────────────────────────────────────────────────────┤
│ T3: Agent calls ctx.connect() → joins room                 │
├─────────────────────────────────────────────────────────────┤
│ T4: Agent and caller are now in same room                  │
│    WebRTC session established bidirectionally              │
├─────────────────────────────────────────────────────────────┤
│ T5: Agent receives audio from caller                        │
│    → STT (transcribes audio to text)                        │
│    → LLM (generates response)                               │
│    → TTS (converts response to audio)                       │
│    → Audio sent back to caller                              │
└─────────────────────────────────────────────────────────────┘

Real Example: Bank IVR System (Like Twilio)

# From examples/telephony/bank-ivr/dial_bank_agent.py
 
async def call_ivr_system(phone_number: str, user_request: str) -> None:
    """Create a dispatch and add a SIP participant to call the phone number"""
    lkapi = api.LiveKitAPI()
 
    logger.info(f"Creating dispatch for agent in room {ROOM_NAME}")
    
    # Step 1: Create dispatch with user request metadata
    dispatch = await lkapi.agent_dispatch.create_dispatch(
        api.CreateAgentDispatchRequest(
            agent_name=AGENT_NAME,  # e.g., "bank-ivr-navigator"
            room=ROOM_NAME,          # e.g., "bank-call-room-123"
            metadata=user_request    # e.g., "Check my checking balance"
        )
    )
 
    # Step 2: Create SIP call to phone number
    # This connects the phone caller to the LiveKit room
    await lkapi.sip.create_sip_participant(
        api.CreateSIPParticipantRequest(
            sip_trunk_id=OUTBOUND_TRUNK_ID,
            sip_call_to=phone_number,
            room=ROOM_NAME,
            participant_identity="caller",
            participant_name="Customer"
        )
    )

---

Part 3: STT → LLM → TTS Communication Flow

Real-Time Pipeline Architecture

┌──────────────────────────────────────────────────────────────────────┐
│                          AgentSession                                 │
│  (The runtime that manages all components)                            │
├──────────────────────────────────────────────────────────────────────┤
│                                                                        │
│  ┌──────────────────┐        ┌──────────────────┐                    │
│  │  AUDIO INPUT     │        │   AGENT BRAIN    │                    │
│  │  from room       │        │    (Agent code)  │                    │
│  │                  │        │                  │                    │
│  │  (AudioFrames)   │        │  ┌────────────┐  │                    │
│  └────────┬─────────┘        │  │Instructions│  │                    │
│           │                  │  │ & Tools    │  │                    │
│           │                  │  └────────────┘  │                    │
│           │                  └────────┬─────────┘                    │
│           │                           │                               │
│           ▼                           │                               │
│  ┌──────────────────┐        ┌───────▼──────────┐                   │
│  │  VAD             │        │  LLMNode /       │                   │
│  │ (Voice Activity  │        │  RealtimeModel   │                   │
│  │  Detector)       │        │                  │                   │
│  │                  │        │  Generates text  │                   │
│  │  Detects speech  │        │  responses       │                   │
│  │  start/end       │        │                  │                   │
│  └────────┬─────────┘        └────────┬─────────┘                   │
│           │                           │                               │
│           ▼                           │                               │
│  ┌──────────────────┐        ┌───────▼──────────┐                   │
│  │  STT             │        │  Tool            │                   │
│  │ (Speech-to-Text) │        │  Execution       │                   │
│  │                  │        │                  │                   │
│  │  Receives audio  │        │  Function calls  │                   │
│  │  frames          │        │  (if LLM chose)  │                   │
│  │  ↓               │        │                  │                   │
│  │  Sends to        │        │  Returns result  │                   │
│  │  speech engine   │        │  to LLM          │                   │
│  │  (Deepgram,      │        │                  │                   │
│  │  Google, etc.)   │        │                  │                   │
│  │  ↓               │        │                  │                   │
│  │  Gets back       │        │                  │                   │
│  │  transcribed     │        │                  │                   │
│  │  text            │        │                  │                   │
│  └────────┬─────────┘        └────────┬─────────┘                   │
│           │                           │                               │
│           ▼                           │                               │
│  ┌──────────────────────────────────┐│                               │
│  │  Chat Context / Conversation     ││                               │
│  │  History                         ││                               │
│  │                                  ││                               │
│  │  [{                              ││                               │
│  │    "role": "user",               ││                               │
│  │    "text": "What's the weather?" ││                               │
│  │  },                              ││                               │
│  │  {                               ││                               │
│  │    "role": "assistant",          ││                               │
│  │    "text": "It's sunny..."       ││                               │
│  │  }]                              ││                               │
│  └──────────────────────────────────┘│                               │
│                                       ▼                               │
│                            ┌──────────────────┐                      │
│                            │  TTS             │                      │
│                            │ (Text-to-Speech) │                      │
│                            │                  │                      │
│                            │  Takes text from │                      │
│                            │  LLM response    │                      │
│                            │  ↓               │                      │
│                            │  Sends to        │                      │
│                            │  speech engine   │                      │
│                            │  (Cartesia,      │                      │
│                            │  ElevenLabs,etc.)│                      │
│                            │  ↓               │                      │
│                            │  Gets back       │                      │
│                            │  audio frames    │                      │
│                            │                  │                      │
│                            └────────┬─────────┘                      │
│                                     │                                │
│                                     ▼                                │
│                            ┌──────────────────┐                      │
│                            │  AUDIO OUTPUT    │                      │
│                            │  to room         │                      │
│                            │                  │                      │
│                            │  (AudioFrames)   │                      │
│                            │  Published to    │                      │
│                            │  all subscribers │                      │
│                            └──────────────────┘                      │
│                                                                       │
└──────────────────────────────────────────────────────────────────────┘

---

Part 4: Detailed STT → LLM Communication

How STT and LLM Actually Communicate

They DON’T communicate directly! The communication happens through the AgentActivity and Chat Context.

# livekit-agents/livekit/agents/voice/agent_activity.py
 
class AgentActivity(RecognitionHooks):
    @property
    def stt(self) -> stt.STT | None:
        # Get STT from agent or session
        return self._agent.stt if is_given(self._agent.stt) else self._session.stt
 
    @property
    def llm(self) -> llm.LLM | llm.RealtimeModel | None:
        # Get LLM from agent or session
        return self._agent.llm if is_given(self._agent.llm) else self._session.llm
 
    @property
    def tts(self) -> tts.TTS | None:
        # Get TTS from agent or session
        return self._agent.tts if is_given(self._agent.tts) else self._session.tts

Communication Flow Step-by-Step

From Code: livekit-agents/livekit/agents/voice/audio_recognition.py

class _STTPipeline:
    """Transferable STT pipeline that survives agent handoff."""
 
    def __init__(self, stt_node: io.STTNode) -> None:
        self._stt_node = stt_node
        self._audio_ch = aio.Chan[rtc.AudioFrame]()        # Audio input channel
        self._event_ch = aio.Chan[stt.SpeechEvent]()       # Transcription output channel
        self._pump_task = asyncio.create_task(self._stt_pump())

Flow Diagram with Server Processing

Step 1: Audio Stream from Client
┌──────────────────────────────────────────────────┐
│ Caller speaks: "What's the weather?"             │
│ Audio sent via WebRTC to LiveKit Server          │
└──────────┬───────────────────────────────────────┘
           │
           ▼ (20ms audio chunks)
┌──────────────────────────────────────────────────┐
│ Agent Process (on agent server)                  │
│                                                   │
│ Step 1: Audio Frame Buffered                     │
│ ┌────────────────────────────────────────────┐   │
│ │ input.audio.recv()                         │   │
│ │ → Stream of RTCFrame objects               │   │
│ └─────────────┬──────────────────────────────┘   │
│               │                                   │
│ Step 2: Send to VAD & STT                        │
│               ▼                                   │
│ ┌────────────────────────────────────────────┐   │
│ │ VAD (Voice Activity Detection)              │   │
│ │ ┌──────────────────────────────────────┐   │   │
│ │ │ Silero VAD checks: is this speech?   │   │   │
│ │ │ ├─ No voice? → wait                  │   │   │
│ │ │ ├─ Voice detected? → send to STT    │   │   │
│ │ │ └─ Silence after speech? → finalize  │   │   │
│ │ └──────────────┬───────────────────────┘   │   │
│ └────────────────┼──────────────────────────┘   │
│                  │                               │
│ Step 3: STT Processes                           │
│                  ▼                               │
│ ┌────────────────────────────────────────────┐   │
│ │ STT.stream()                               │   │
│ │ ┌──────────────────────────────────────┐   │   │
│ │ │ For each audio frame:                │   │   │
│ │ │ 1. stream.push_frame(frame)          │   │   │
│ │ │    → Sends to external service       │   │   │
│ │ │       (Deepgram, Google, etc.)       │   │   │
│ │ │                                       │   │   │
│ │ │ 2. async for event in stream:        │   │   │
│ │ │    await_transcription_response      │   │   │
│ │ │    → Receives: SpeechEvent           │   │   │
│ │ │       {                              │   │   │
│ │ │         "type": "final",             │   │   │
│ │ │         "text": "What's weather?",   │   │   │
│ │ │         "confidence": 0.98           │   │   │
│ │ │       }                              │   │   │
│ │└──────────────────────────────────────┘   │   │
│ └─────────────┬────────────────────────────┘   │
│               │                                 │
│ Step 4: Text Enters Chat Context                │
│               ▼                                 │
│ ┌──────────────────��─────────────────────────┐   │
│ │ chat_context.add_message(                  │   │
│ │     role="user",                           │   │
│ │     text="What's the weather?"             │   │
│ │ )                                          │   │
│ │                                            │   │
│ │ Chat Context Now Contains:                 │   │
│ │ ┌──────────────────────────────────────┐   │   │
│ │ │ [                                    │   │   │
│ │ │   {                                  │   │   │
│ │ │     "role": "system",                │   │   │
│ │ │     "content": "You are helpful..." │   │   │
│ │ │   },                                 │   │   │
│ │ │   {                                  │   │   │
│ │ │     "role": "user",                  │   │   │
│ │ │     "content": "What's weather?"     │   │   │
│ │ │   }                                  │   │   │
│ │ │ ]                                    │   │   │
│ │ └──────────────────────────────────────┘   │   │
│ └─────────────┬────────────────────────────┘   │
└───────────────┼────────────────────────────────┘
                │
                ▼
        [NETWORK CALL]
    
┌──────────────────────────────────────────────────┐
│ External STT Service (Deepgram, Google, etc.)   │
│                                                   │
│ receives: audio stream                           │
│ returns: {"transcript": "What's the weather?"}  │
└──────────────────────────────────────────────────┘

        [RECEIVES TRANSCRIPT]
                │
                ▼

Step 5: LLM Processes Chat Context
┌──────────────────────────────────────────────────┐
│ Agent Process (continued)                        │
│                                                   │
│ LLM.chat(chat_context)                          │
│ ┌────────────────────────────────────────────┐   │
│ │ Sends to LLM service (OpenAI, etc.):       │   │
│ │ {                                           │   │
│ │   "model": "gpt-4-mini",                    │   │
│ │   "messages": [                             │   │
│ │     {"role": "system", "content": "..."},   │   │
│ │     {"role": "user", "content": "What's..."}│  │
│ │   ],                                        │   │
│ │   "tools": [...]  // if agent has tools    │   │
│ │ }                                           │   │
│ └─────────────┬────────────────────────────┘   │
│               │                                 │
│  Receives:   │  Streaming response chunks      │
│               ▼                                 │
│ ┌────────────────────────────────────────────┐   │
│ │ stream.choice.delta.content                │   │
│ │                                             │   │
│ │ Accumulates: "It's sunny and..."           │   │
│ │              "It's sunny and 72 degrees..."│   │
│ │              "It's sunny and 72 degrees F."│   │
│ └─────────────┬────────────────────────────┘   │
│               │                                 │
│ Step 6: Text Sent to TTS                        │
│               ▼                                 │
│ ┌────────────────────────────────────────────┐   │
│ │ TTS.stream()                                │   │
│ │ ┌──────────────────────────────────────┐   │   │
│ │ │ For each LLM text chunk:             │   │   │
│ │ │ 1. stream.push_text("It's sunny...")  │   │   │
│ │ │    → Sends to TTS service            │   │   │
│ │ │       (Cartesia, ElevenLabs, etc.)   │   │   │
│ │ │                                       │   │   │
│ │ │ 2. async for audio_chunk in stream:  │   │   │
│ │ │    → Receives: AudioFrame            │   │   │
│ │ │       encoded at 16kHz PCM           │   │   │
│ │ │                                       │   │   │
│ │ │ 3. output.audio.send(frame)          │   │   │
│ │ │    → Publishes to room               │   │   │
│ │ └──────────────────────────────────────┘   │   │
│ └─────────────┬────────────────────────────┘   │
└───────────────┼────────────────────────────────┘
                │
                ▼
        [NETWORK CALL]
    
┌──────────────────────────────────────────────────┐
│ External TTS Service                             │
│                                                   │
│ receives: "It's sunny and 72 degrees F"         │
│ returns: audio stream (16kHz PCM)               │
└──────────────────────────────────────────────────┘

        [RECEIVES AUDIO]
                │
                ▼

Step 7: Audio Sent Back to Caller
┌──────────────────────────────────────────────────┐
│ LiveKit Server (Forwarding)                      │
│                                                   │
│ Agent audio track published to room:             │
│ ├─ Room forwards to caller                       │
│ └─ Caller hears agent response                   │
│                                                   │
│ Caller: "Hears: It's sunny and 72 degrees F"     │
└──────────────────────────────────────────────────┘

---

Part 5: Where Communication REALLY Happens (No Direct Connect!)

Key Insight: Everything Goes Through AgentActivity and ChatContext

The STT and LLM never talk directly. Instead:

# File: livekit-agents/livekit/agents/voice/agent_activity.py - Line 2449
 
async def _pipeline_reply_task_impl(
    self,
    *,
    speech_handle: SpeechHandle,
    chat_ctx: llm.ChatContext,           # ← This is the bridge!
    tools: list[llm.Tool | llm.Toolset],
    model_settings: ModelSettings,
    new_message: llm.ChatMessage | None = None,
    instructions: str | Instructions | None = None,
):
    """
    This is WHERE the magic happens:
    
    1. STT produces: SpeechEvent (transcribed text)
    2. Event triggers: on_final_transcript()
    3. Which calls: _add_user_message_to_chat_ctx()
    4. Which updates: chat_ctx with new text
    5. Then: llm_node gets called with updated chat_ctx
    6. LLM reads: chat_ctx.messages (including new user text)
    7. LLM generates: response text
    8. Response goes to: tts_node
    9. TTS converts: text → audio → published to room
    """

The Actual Message Flow

# Simplified pseudo-code showing communication flow
 
async def conversation_loop():
    """Main loop inside AgentActivity"""
    
    # Initialize chat context
    chat_ctx = ChatContext(
        system_prompt="You are a helpful assistant"
    )
    
    while agent_running:
        # ╔════════════════════════════════════════════════════════════╗
        # ║ PHASE 1: LISTEN & TRANSCRIBE (STT)                        ║
        # ╚════════════════════════════════════════════════════════════╝
        
        audio_frames = await room_input.audio.recv()  # Get audio from caller
        
        # Process through STT
        stt_stream = stt.stream()  # Create STT stream
        for frame in audio_frames:
            stt_stream.push_frame(frame)
        
        # Get transcription event
        async for stt_event in stt_stream:
            if stt_event.final:  # ← Final transcription received
                user_text = stt_event.text
                
                # ╔════════════════════════════════════════════════════════════╗
                # ║ BRIDGE: ADD TO CHAT CONTEXT                               ║
                # ╚════════════════════════════════════════════════════════════╝
                
                chat_ctx.add_message(
                    role="user",
                    content=user_text
                )
                
                print(f"STT Output → ChatContext Updated: '{user_text}'")
        
        # ╔════════════════════════════════════════════════════════════╗
        # ║ PHASE 2: GENERATE RESPONSE (LLM)                          ║
        # ╚════════════════════════════════════════════════════════════╝
        
        # LLM reads from chat_ctx (which now has user's text!)
        llm_stream = llm.chat(chat_ctx=chat_ctx)  # ← Gets updated context
        
        response_text = ""
        async for chunk in llm_stream:
            response_text += chunk.text
            
            # ╔════════════════════════════════════════════════════════════╗
            # ║ PHASE 3: CONVERT TO SPEECH (TTS) - Streaming             ║
            # ╚════════════════════════════════════════════════════════════╝
            
            # Send chunk to TTS immediately (don't wait for full LLM response)
            tts_stream = tts.stream()
            await tts_stream.push_text(chunk.text)
            
            # Receive audio and send to caller
            async for audio_frame in tts_stream:
                await room_output.audio.send(audio_frame)
        
        # ╔════════════════════════════════════════════════════════════╗
        # ║ UPDATE CHAT CONTEXT WITH ASSISTANT RESPONSE               ║
        # ╚════════════════════════════════════════════════════════════╝
        
        chat_ctx.add_message(
            role="assistant",
            content=response_text
        )
        
        print(f"LLM Output → ChatContext Updated: '{response_text}'")
        print(f"ChatContext now has conversation history")
        print(f"Ready for next user input...")

---

Part 6: Complete Real-World Voice Agent Example

# examples/voice_agents/basic_agent.py - Simplified
 
from livekit.agents import (
    Agent, AgentServer, AgentSession, JobContext, cli
)
from livekit.agents import inference
 
class MyVoiceAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a helpful weather assistant.",
            # Agent can have its own components or use session's
        )
    
    async def on_enter(self):
        """Called when agent joins the room"""
        # Proactively greet the user
        await self.session.generate_reply(
            instructions="Greet the user and ask how you can help"
        )
 
 
server = AgentServer()
 
 
@server.rtc_session()
async def entrypoint(ctx: JobContext):
    """
    This function runs on the agent server.
    
    Flow:
    1. Dispatcher creates job with room name & metadata
    2. Agent server receives job
    3. This entrypoint() function executes
    4. Agent joins room via ctx.connect()
    5. Agent and user are now in same WebRTC session
    6. Media flows through agent's pipeline
    """
    
    # Create agent session with all components
    session = AgentSession(
        # EARS: Speech-to-Text (receives user's voice)
        stt=inference.STT("deepgram/nova-3", language="multi"),
        
        # BRAIN: Large Language Model (generates responses)
        llm=inference.LLM("openai/gpt-4.1-mini"),
        
        # VOICE: Text-to-Speech (sends agent's voice back)
        tts=inference.TTS("cartesia/sonic-3"),
    )
    
    # Create agent instance
    agent = MyVoiceAgent()
    
    # START THE MAGIC ✨
    # This method:
    # - Connects to LiveKit room
    # - Sets up audio input/output
    # - Starts the STT → LLM → TTS pipeline
    # - Handles turn detection, interruptions, etc.
    await session.start(agent=agent, room=ctx.room)
 
 
if __name__ == "__main__":
    cli.run_app(server)

---

Part 7: Two Processing Modes

Mode 1: Standard Pipeline (STT + LLM + TTS)

┌─────────────────────────────────────────────┐
│ Audio Input (user speech)                   │
└────────────┬────────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────┐
│ STT: "What's the weather?" (text)           │
│ Service: Deepgram, Google, Whisper          │
└────────────┬───────────���────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────┐
│ LLM: Generate response (text)               │
│ Model: GPT-4, Claude, Gemini                │
│ "It's sunny and 72 degrees"                 │
└────────────┬────────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────┐
│ TTS: Generate audio output                  │
│ Service: Cartesia, ElevenLabs, Google       │
└────────────┬────────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────┐
│ Audio Output (agent voice to user)          │
└─────────────────────────────────────────────┘

Latency: ~500-2000ms (multiple external API calls)

Mode 2: Realtime LLM (Audio → LLM → Audio)

# For even lower latency, use Realtime APIs
 
from livekit.plugins import openai
 
session = AgentSession(
    llm=openai.realtime.RealtimeModel(
        voice="coral"  # OpenAI's realtime API
    )
)

┌─────────────────────────────────────────────┐
│ Audio Input (user speech)                   │
└────────────┬────────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────┐
│ Realtime LLM (handles everything!)          │
│ - Understands audio directly                │
│ - Generates response                        │
│ - Creates output audio                      │
│                                             │
│ Service: OpenAI Realtime, Google Gemini    │
└────────────┬────────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────┐
│ Audio Output (agent voice to user)          │
└─────────────────────────────────────────────┘

Latency: ~200-500ms (single WebSocket connection)

---

Part 8: Telephony Integration (PSTN/SIP Calls)

How Phone Calls Work

# From examples/telephony/bank-ivr/dial_bank_agent.py
 
async def call_phone_number(phone_number: str) -> None:
    """Make an outbound phone call like Twilio"""
    
    lkapi = api.LiveKitAPI()
    
    # Create dispatch (tells LiveKit to launch agent)
    dispatch = await lkapi.agent_dispatch.create_dispatch(
        api.CreateAgentDispatchRequest(
            agent_name="ivr-navigator",
            room="call-room-456",
            metadata=json.dumps({
                "target_number": phone_number,
                "user_request": "Check account balance"
            })
        )
    )
    
    # Create SIP call (connects phone to room)
    await lkapi.sip.create_sip_participant(
        api.CreateSIPParticipantRequest(
            sip_trunk_id="ST_abc123",  # Your SIP trunk
            sip_call_to=phone_number,   # The number to call
            room="call-room-456",        # Same room as agent
            participant_identity="remote-caller"
        )
    )
    
    # Now:
    # 1. Agent joins room via JobContext
    # 2. Phone caller joins room via SIP
    # 3. Both are connected in same WebRTC session
    # 4. Agent hears caller and responds via voice

Architecture with Telephony

┌──────────────┐
│ Phone Number │
│ +1-555-1234  │
└──────┬───────┘
       │
       ▼ (SIP Trunk)
┌──────────────────────────────────────────┐
│ LiveKit Server                           │
│ ┌────────────────────────────────────┐   │
│ │ Room: "call-room-456"              │   │
│ │                                    │   │
│ │ ├─ SIP Participant (phone caller)  │   │
│ │ ├─ Agent Participant (AI agent)    │   │
│ │                                    │   │
│ │ Media flows between them via SFU   │   │
│ └────────────────────────────────────┘   │
└──────┬──────────────────────────────────┘
       │
       ▼ (Agent Server)
┌──────────────────────────────────────────┐
│ Agent Process                            │
│ (running MyVoiceAgent)                   │
│                                          │
│ jobctx.connect()                         │
│   → Joins room as "agent" participant    │
│                                          │
│ Receives audio from SIP caller           │
│ Processes via STT → LLM → TTS           │
│ Sends audio back to SIP caller           │
└──────────────────────────────────────────┘

---

Summary: Key Differences from Twilio

Feature	Twilio	LiveKit
Architecture	SaaS cloud-only	Open-source + SaaS options
Call Routing	HTTP webhooks	Agent Dispatch + Job Queue
AI Components	Built-in basic voice	Bring your own STT/LLM/TTS
Customization	Limited	Unlimited (open-source)
Deployment	Cloud only	Self-hosted or cloud
Real-time AI	Limited	Full real-time voice agents
Telephony	First-class support	Via SIP trunks
Server Control	Webhook-based	Code-based (JobContext)

---

Complete Flow Summary

┌─────────────────────────────────────────────────────────────┐
│ 1. USER INITIATES CONTACT                                   │
│    ├─ Phone call via PSTN/SIP                              │
│    ├─ Browser connects to WebRTC room                      │
│    └─ Mobile app joins room                                │
├─────────────────────────────────────────────────────────────┤
│ 2. DISPATCHER CREATES AGENT JOB                             │
│    ├─ Create dispatch with room name                       │
│    ├─ Include metadata (user request, context)             │
│    └─ Agent server polls for new jobs                      │
├─────────────────────────────────────────────────────────────┤
│ 3. AGENT SPAWNED                                            │
│    ├─ JobContext.connect() called                          │
│    ├─ Agent joins LiveKit room                             │
│    └─ Agent and user in same room                          │
├─────────────────────────────────────────────────────────────┤
│ 4. AUDIO PIPELINE STARTS                                    │
│    ├─ User speaks → Audio frames                           │
│    ├─ Agent receives → STT processes                       │
│    ├─ Text enters → Chat context                           │
│    ├─ LLM reads → Generates response                       │
│    ├─ Response → TTS generates audio                       │
│    └─ Audio → Published to room → User hears               │
├─────────────────────────────────────────────────────────────┤
│ 5. CONVERSATION CONTINUES                                   │
│    ├─ Chat context keeps conversation history              │
│    ├─ Tools can be called for business logic                │
│    ├─ Agent can transfer/handoff to human                  │
│    └─ Loop repeats for each user turn                      │
├─────────────────────────────────────────────────────────────┤
│ 6. SESSION ENDS                                             │
│    ├─ User hangs up or agent ends call                     │
│    ├─ Room cleaned up                                       │
│    └─ Agent process terminates                             │
└─────────────────────────────────────────────────────────────┘

This is the complete LiveKit voice agent architecture - a fully open-source alternative to Twilio with much more control, flexibility, and real-time AI capabilities! 🎤