Building a Voice Agent

01 Overview

A voice agent is an AI system that listens to human speech, understands intent, reasons over context, and responds with natural-sounding speech — all in real time. Modern voice agents combine automatic speech recognition (ASR/STT), large language models (LLMs), and neural text-to-speech (TTS) into a low-latency pipeline.

02 System Architecture

03 Voice Pipeline (Step by Step)

# Pseudocode: Streaming voice pipeline
async def voice_pipeline(audio_stream):
    # Stage 1: VAD → filter silence
    speech_chunks = vad.filter(audio_stream)

    # Stage 2: Streaming STT → interim + final transcripts
    async for transcript in stt.transcribe_stream(speech_chunks):
        if transcript.is_final:
            # Stage 3: Stream LLM response token by token
            sentence_buffer = ""
            async for token in llm.stream(transcript.text, context):
                sentence_buffer += token

                # Stage 4: Send complete sentences to TTS
                if ends_with_punctuation(sentence_buffer):
                    async for audio in tts.synthesize_stream(sentence_buffer):
                        yield audio  # → speaker
                    sentence_buffer = ""

04 Latency Budget

Voice agents are latency-critical. Humans perceive pauses >600ms as unnatural. The target is <1 second from end of user speech to beginning of agent speech.

05 Speech-to-Text (STT / ASR)

Automatic Speech Recognition converts audio waveforms into text. For voice agents, streaming STT is essential — results must arrive incrementally as the user speaks.

06 STT Engines Compared

# Deepgram Streaming STT (WebSocket)
import asyncio
from deepgram import DeepgramClient, LiveOptions, LiveTranscriptionEvents

dg = DeepgramClient(api_key="YOUR_KEY")
connection = dg.listen.asyncwebsocket.v("1")

async def on_message(self, result, **kwargs):
    transcript = result.channel.alternatives[0].transcript
    is_final = result.is_final
    if is_final and transcript:
        print(f"Final: {transcript}")
        # → Send to LLM

connection.on(LiveTranscriptionEvents.Transcript, on_message)

options = LiveOptions(
    model="nova-2",
    language="en",
    encoding="linear16",
    sample_rate=16000,
    interim_results=True,
    endpointing=300,  # ms of silence before final
    smart_format=True,
    vad_events=True,
)
await connection.start(options)

# Send audio chunks (20ms frames)
async for chunk in mic_stream:
    connection.send(chunk)

06A Why Deepgram — Deep Dive

Deepgram is the recommended STT engine for production voice agents. Here's a detailed analysis of why it outperforms alternatives for real-time conversational AI.

06B Deepgram Features for Voice Agents

endpointing=300   # 300ms silence = end of utterance
endpointing=500   # 500ms for cautious endpointing
endpointing=false  # Disable (you handle it)

utterance_end_ms=1000  # Gap between utterances
interim_results=true   # Get partial transcripts
vad_events=true        # Speech start/stop events

keywords=[
  "Acme:2",         # Boost "Acme" by 2x
  "SKU:1.5",        # Product codes
  "onboarding:1.5"  # Domain terms
]

06C Deepgram Implementation

# Complete Deepgram Streaming STT for Voice Agent
import asyncio, json
from deepgram import (
    DeepgramClient,
    DeepgramClientOptions,
    LiveTranscriptionEvents,
    LiveOptions,
)

class DeepgramSTTEngine:
    """Production-ready Deepgram STT wrapper for voice agents."""

    def __init__(self, api_key: str, on_transcript, on_speech_started=None):
        self.client = DeepgramClient(api_key, DeepgramClientOptions(
            options={"keepalive": "true"}  # Persistent connection
        ))
        self.on_transcript = on_transcript
        self.on_speech_started = on_speech_started
        self.connection = None

    async def connect(self):
        self.connection = self.client.listen.asyncwebsocket.v("1")

        # Register event handlers
        self.connection.on(LiveTranscriptionEvents.Transcript, self._on_message)
        self.connection.on(LiveTranscriptionEvents.SpeechStarted, self._on_speech_started)
        self.connection.on(LiveTranscriptionEvents.UtteranceEnd, self._on_utterance_end)
        self.connection.on(LiveTranscriptionEvents.Error, self._on_error)

        options = LiveOptions(
            model="nova-2",            # Best for conversational speech
            language="en",
            encoding="linear16",       # 16-bit PCM
            sample_rate=16000,        # 16kHz mono
            channels=1,
            interim_results=True,     # Get partial transcripts for UI
            endpointing=300,          # 300ms silence = final
            utterance_end_ms=1000,    # Utterance boundary detection
            smart_format=True,        # Auto-format numbers, dates
            punctuate=True,           # Add punctuation
            vad_events=True,          # Speech start/stop events
            filler_words=False,       # Remove "um", "uh"
        )

        if not await self.connection.start(options):
            raise ConnectionError("Failed to connect to Deepgram")
        print("✓ Deepgram STT connected")

    async def send_audio(self, audio_bytes: bytes):
        """Send raw audio chunk (20ms frame = 640 bytes at 16kHz/16bit)."""
        if self.connection:
            self.connection.send(audio_bytes)

    async def _on_message(self, _self, result, **kwargs):
        transcript = result.channel.alternatives[0].transcript
        if not transcript:
            return

        if result.is_final:
            # Final transcript → send to LLM
            confidence = result.channel.alternatives[0].confidence
            await self.on_transcript(transcript, is_final=True, confidence=confidence)
        else:
            # Interim → update UI only
            await self.on_transcript(transcript, is_final=False)

    async def _on_speech_started(self, _self, speech_started, **kwargs):
        # User started speaking → interrupt agent if needed
        if self.on_speech_started:
            await self.on_speech_started()

    async def _on_utterance_end(self, _self, utterance_end, **kwargs):
        # Clean boundary between utterances
        pass

    async def _on_error(self, _self, error, **kwargs):
        print(f"Deepgram error: {error}")

    async def close(self):
        if self.connection:
            await self.connection.finish()

07 Streaming Recognition

08 Voice Activity Detection (VAD)

VAD distinguishes human speech from silence, noise, and background audio. It's the gatekeeper that decides when to start and stop STT processing.

# Silero VAD example
import torch

model, utils = torch.hub.load(
    repo_or_dir='snakers4/silero-vad',
    model='silero_vad',
    trust_repo=True
)
(get_speech_timestamps, _, read_audio, _, _) = utils

# Real-time frame-by-frame
def process_frame(audio_chunk_tensor):
    speech_prob = model(audio_chunk_tensor, 16000).item()
    is_speech = speech_prob > 0.5
    return is_speech

09 Natural Language Understanding (NLU) & Intent Detection

NLU processes the transcribed text to extract meaning — intents, entities, sentiment, and dialog acts. This is a critical question for voice agent design: how do you detect what the user wants?

The Three Approaches to Intent Detection

09A Intent Detection — Full Comparison

09B Intent Detection Approaches (Detailed)

Approach 1: LLM Function Calling as Intent Detection

The most common modern approach. Define intents as "functions" — the LLM decides which function to call based on the user's speech. This effectively combines NLU + action routing in one step.

# LLM Function Calling = Intent Detection + Entity Extraction
# Define your intents as tools/functions

tools = [
    {
        "type": "function",
        "function": {
            "name": "check_order_status",       # ← This IS the intent
            "description": "User wants to check the status of an existing order",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {                   # ← This IS the entity
                        "type": "string",
                        "description": "Order ID or number"
                    }
                }
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "transfer_to_human",
            "description": "User wants to speak with a human agent",
            "parameters": {
                "type": "object",
                "properties": {
                    "department": {
                        "type": "string",
                        "enum": ["billing", "support", "sales"]
                    },
                    "reason": {"type": "string"}
                }
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "make_payment",
            "description": "User wants to make a payment on their account",
            "parameters": {
                "type": "object",
                "properties": {
                    "amount": {"type": "number"},
                    "account_id": {"type": "string"}
                }
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "general_question",
            "description": "User has a general question not covered by specific functions",
            "parameters": {
                "type": "object",
                "properties": {
                    "question": {"type": "string"}
                }
            }
        }
    }
]

# User says: "I want to check on order number 4567"
# LLM returns: tool_call(name="check_order_status", args={"order_id": "4567"})
#                          ↑ intent                        ↑ entity

Approach 2: Traditional NLU (Rasa / Dialogflow)

# Rasa NLU Pipeline (nlu.yml)
# Train a dedicated ML model for intent classification

nlu:
- intent: check_order
  examples: |
    - where is my order
    - check order status
    - what's the status of order [4567](order_id)
    - track my package
    - I want to know where my delivery is
    - can you look up order [AB-1234](order_id)

- intent: make_payment
  examples: |
    - I'd like to pay my bill
    - make a payment of [$50](amount)
    - pay [100 dollars](amount) on my account
    - how do I pay

- intent: transfer_to_human
  examples: |
    - let me talk to a real person
    - transfer me to an agent
    - I want to speak to someone
    - get me a human

# Result: {"intent": "check_order", "confidence": 0.94,
#          "entities": [{"entity": "order_id", "value": "4567"}]}

Approach 3: Fast Embedding Classifier (SetFit / Sentence-BERT)

# Ultra-fast intent detection using sentence embeddings
# Only needs 8-16 examples per intent to train

from setfit import SetFitModel, SetFitTrainer

# Train with minimal examples
train_data = [
    ("check my order", "check_order"),
    ("where is my package", "check_order"),
    ("track delivery", "check_order"),
    ("order status", "check_order"),
    ("pay my bill", "make_payment"),
    ("make a payment", "make_payment"),
    ("talk to a human", "transfer"),
    ("speak to agent", "transfer"),
    # ... 8-16 examples per intent
]

model = SetFitModel.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")
trainer = SetFitTrainer(model=model, train_dataset=train_data)
trainer.train()

# Inference: <5ms!
intent = model.predict("I need to check on order 4567")
# → "check_order"

Approach 4: Hybrid Router (Recommended for Production Voice Agents)

# Hybrid: Fast classifier + LLM fallback
# This is the production-recommended approach for voice agents

class HybridIntentRouter:
    def __init__(self):
        self.fast_classifier = SetFitModel.from_pretrained("./intent-model")
        self.confidence_threshold = 0.85
        self.llm = OpenAIClient()

    async def detect_intent(self, transcript: str) -> dict:
        # Step 1: Try fast classifier (~5ms)
        predictions = self.fast_classifier.predict_proba([transcript])
        top_intent = predictions.argmax()
        confidence = predictions.max()

        if confidence >= self.confidence_threshold:
            # High confidence → use fast result (saves 200-400ms!)
            return {
                "intent": top_intent,
                "confidence": confidence,
                "method": "fast_classifier",
                "latency_ms": 5,
            }

        # Step 2: Low confidence → fall back to LLM (200-400ms)
        llm_result = await self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": "Classify the user's intent. Return JSON: {intent, entities, confidence}"
            }, {
                "role": "user",
                "content": transcript
            }],
            response_format={"type": "json_object"},
        )
        return {**json.loads(llm_result.choices[0].message.content), "method": "llm_fallback"}

09C LangChain / LangGraph Role in Voice Agents

Since LangChain and LangGraph are often confused with NLU, here's exactly what role they play in a voice agent pipeline.

# LangGraph voice agent with intent routing
# Note: Intent detection happens INSIDE the LLM call, not from LangGraph

from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal

class VoiceState(TypedDict):
    transcript: str
    intent: str
    entities: dict
    response: str
    conversation_history: list

# Node 1: Detect intent (uses LLM — LangGraph doesn't do this itself)
async def detect_intent(state: VoiceState) -> VoiceState:
    # Option A: Fast classifier
    result = fast_classifier.predict(state["transcript"])
    # Option B: LLM function calling
    # result = await llm.classify(state["transcript"])
    state["intent"] = result.intent
    state["entities"] = result.entities
    return state

# Router: LangGraph routes based on detected intent
def route_intent(state: VoiceState) -> Literal["order", "payment", "transfer", "general"]:
    intent_map = {
        "check_order": "order",
        "make_payment": "payment",
        "transfer_to_human": "transfer",
    }
    return intent_map.get(state["intent"], "general")

# Build graph
graph = StateGraph(VoiceState)
graph.add_node("detect_intent", detect_intent)
graph.add_node("order", handle_order_check)
graph.add_node("payment", handle_payment)
graph.add_node("transfer", handle_transfer)
graph.add_node("general", handle_general_query)
graph.add_node("respond", generate_voice_response)

graph.set_entry_point("detect_intent")
graph.add_conditional_edges("detect_intent", route_intent)
for node in ["order", "payment", "transfer", "general"]:
    graph.add_edge(node, "respond")
graph.add_edge("respond", END)

voice_agent = graph.compile()

09D Intent Detection Decision Guide

10 Dialog Management

Controls the flow of conversation — tracking state, managing turns, handling context switches, and deciding what action to take next.

11 LLM Integration

The LLM is the reasoning brain of the voice agent. It processes the user's transcript, conversation history, and system instructions to generate responses.

# Voice-optimized LLM prompt
SYSTEM_PROMPT = """You are a helpful voice assistant for Acme Corp customer support.

VOICE-SPECIFIC RULES:
- Keep responses SHORT (1-3 sentences). Voice != chat.
- Use conversational language, contractions, natural phrasing.
- NEVER use markdown, bullet points, URLs, or special formatting.
- Spell out numbers: "twenty three" not "23".
- For lists, say "first... second... third..." not "1. 2. 3."
- If unsure, ask ONE clarifying question at a time.
- Acknowledge the user before answering: "Sure!", "Great question.", etc.

FUNCTION CALLING:
- Use check_order_status(order_id) for order inquiries.
- Use transfer_to_human(department) if user explicitly asks for a person.
- Use schedule_callback(phone, time) for callback requests.

CONTEXT:
- Customer: {customer_name}
- Account tier: {tier}
- Previous interactions: {history_summary}
"""

# Streaming LLM call
async def stream_llm_response(transcript, context):
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT.format(**context)},
            *conversation_history,
            {"role": "user", "content": transcript}
        ],
        stream=True,
        temperature=0.7,
        max_tokens=150,  # Keep voice responses short
    )

    sentence_buffer = ""
    async for chunk in response:
        token = chunk.choices[0].delta.content or ""
        sentence_buffer += token

        # Yield complete sentences for TTS
        if any(sentence_buffer.rstrip().endswith(p) for p in (".", "!", "?", ",")):
            yield sentence_buffer.strip()
            sentence_buffer = ""

    if sentence_buffer.strip():
        yield sentence_buffer.strip()

12 RAG for Voice Agents

Retrieval-Augmented Generation connects your voice agent to enterprise knowledge bases, FAQs, product docs, and customer data — so it gives accurate, grounded answers.

13 Text-to-Speech (TTS)

TTS converts the LLM's text response into natural-sounding audio. Modern neural TTS produces near-human quality. For voice agents, streaming TTS is critical — audio begins playing before the full text is synthesized.

14 TTS Engines Compared

# ElevenLabs Streaming TTS
from elevenlabs import ElevenLabs

client = ElevenLabs(api_key="YOUR_KEY")

def stream_tts(text: str):
    audio_stream = client.text_to_speech.convert_as_stream(
        voice_id="pNInz6obpgDQGcFmaJgB",  # "Adam"
        text=text,
        model_id="eleven_turbo_v2_5",
        output_format="pcm_16000",  # Raw PCM for low latency
    )
    for audio_chunk in audio_stream:
        yield audio_chunk  # Send to speaker/WebSocket

# Cartesia Streaming TTS (ultra-low latency)
from cartesia import Cartesia

cartesia = Cartesia(api_key="YOUR_KEY")

async def stream_cartesia(text: str):
    output = await cartesia.tts.sse(
        model_id="sonic-english",
        transcript=text,
        voice_id="a0e99841-438c-4a64-b679-ae501e7d6091",
        output_format={"container": "raw", "encoding": "pcm_s16le", "sample_rate": 16000},
        stream=True,
    )
    async for chunk in output:
        yield chunk["audio"]

14A Why ElevenLabs — Deep Dive

ElevenLabs delivers the most natural-sounding AI voices on the market. For enterprise voice agents where brand perception and user trust depend on voice quality, ElevenLabs is the premium choice.

14B ElevenLabs Implementation

# Complete ElevenLabs Streaming TTS for Voice Agents
import asyncio
from elevenlabs import ElevenLabs
from elevenlabs.core import ApiError

class ElevenLabsTTSEngine:
    """Production ElevenLabs TTS with streaming and voice management."""

    def __init__(self, api_key: str, voice_id: str = "pNInz6obpgDQGcFmaJgB"):
        self.client = ElevenLabs(api_key=api_key)
        self.voice_id = voice_id

    def stream_audio(self, text: str, model: str = "eleven_turbo_v2_5"):
        """Stream audio chunks for a text sentence.

        Models:
        - eleven_turbo_v2_5: Fastest (~100ms), good quality — USE FOR VOICE AGENTS
        - eleven_multilingual_v2: Best quality (~200ms), all 29 languages
        - eleven_monolingual_v1: English only, legacy
        """
        audio_stream = self.client.text_to_speech.convert_as_stream(
            voice_id=self.voice_id,
            text=text,
            model_id=model,
            output_format="pcm_16000",      # Raw PCM for lowest latency
            voice_settings={
                "stability": 0.5,            # 0=expressive, 1=stable
                "similarity_boost": 0.75,    # Closeness to original voice
                "style": 0.0,               # 0=neutral, 1=exaggerated
                "use_speaker_boost": True,   # Enhance clarity
            },
            optimize_streaming_latency=3,  # 0-4, higher = faster but lower quality
        )

        for audio_chunk in audio_stream:
            yield audio_chunk

    async def synthesize_for_twilio(self, text: str):
        """Generate audio in mulaw format for Twilio Media Streams."""
        audio_stream = self.client.text_to_speech.convert_as_stream(
            voice_id=self.voice_id,
            text=text,
            model_id="eleven_turbo_v2_5",
            output_format="ulaw_8000",  # Native Twilio format!
        )
        for chunk in audio_stream:
            yield chunk

    def get_voices(self):
        """List available voices."""
        return self.client.voices.get_all()

    def clone_voice(self, name: str, audio_files: list):
        """Clone a voice from audio samples."""
        return self.client.clone(
            name=name,
            files=audio_files,
            description="Custom brand voice for voice agent"
        )

14C Why Cartesia — Deep Dive

Cartesia (Sonic model) delivers the lowest TTS latency in the market, making it the ideal choice when response speed is the primary concern.

14D Cartesia Implementation

# Complete Cartesia Sonic Streaming TTS
import asyncio
from cartesia import Cartesia

class CartesiaTTSEngine:
    """Production Cartesia TTS with WebSocket streaming."""

    def __init__(self, api_key: str, voice_id: str):
        self.client = Cartesia(api_key=api_key)
        self.voice_id = voice_id
        self.ws = None

    async def connect_websocket(self):
        """Establish persistent WebSocket for lowest latency."""
        self.ws = self.client.tts.websocket()
        print("✓ Cartesia WebSocket connected")

    async def stream_audio(self, text: str, context_id: str = "default"):
        """Stream audio via persistent WebSocket connection.

        context_id: Use same ID for sentences in one turn
        to maintain prosody continuity across chunks.
        """
        output = self.ws.send(
            model_id="sonic-english",
            transcript=text,
            voice={
                "mode": "id",
                "id": self.voice_id,
                # Emotion mixing example:
                # "mode": "embedding",
                # "embedding": blend(professional_emb, warm_emb, 0.7)
            },
            output_format={
                "container": "raw",
                "encoding": "pcm_s16le",
                "sample_rate": 16000,
            },
            context_id=context_id,  # Prosody continuity
            stream=True,
        )

        for chunk in output:
            # chunk contains: audio bytes + optional word timestamps
            yield chunk["audio"]

    async def stream_for_twilio(self, text: str):
        """Generate mulaw audio for Twilio telephony."""
        output = self.ws.send(
            model_id="sonic-english",
            transcript=text,
            voice={"mode": "id", "id": self.voice_id},
            output_format={
                "container": "raw",
                "encoding": "pcm_mulaw",   # Native Twilio format
                "sample_rate": 8000,       # Telephony standard
            },
            stream=True,
        )
        for chunk in output:
            yield chunk["audio"]

    async def close(self):
        if self.ws:
            self.ws.close()

14E Choosing ElevenLabs vs Cartesia

15 Voice Cloning

Create a custom brand voice from audio samples. Requires as little as 30 seconds of clean audio with some providers.

16 SSML & Prosody Control

Speech Synthesis Markup Language (SSML) gives fine-grained control over how TTS engines pronounce text.

<!-- SSML Example -->
<speak>
  <prosody rate="medium" pitch="+5%">
    Welcome to Acme support!
  </prosody>
  <break time="300ms"/>
  Your order
  <say-as interpret-as="characters">AB</say-as>
  <say-as interpret-as="cardinal">1234</say-as>
  is on its way.
  <emphasis level="strong">Is there anything else I can help with?</emphasis>
</speak>

17 WebSocket Streaming

WebSockets provide full-duplex, low-latency communication for real-time audio streaming between client and server.

# FastAPI WebSocket voice agent server
import asyncio
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/voice")
async def voice_endpoint(ws: WebSocket):
    await ws.accept()

    stt = StreamingSTT()
    llm = LLMClient()
    tts = StreamingTTS()

    try:
        while True:
            # Receive audio from client
            audio_data = await ws.receive_bytes()

            # Feed to streaming STT
            transcript = await stt.process(audio_data)

            if transcript and transcript.is_final:
                # Stream LLM → TTS → audio back to client
                async for sentence in llm.stream(transcript.text):
                    async for audio_chunk in tts.synthesize(sentence):
                        await ws.send_bytes(audio_chunk)
    except Exception:
        await ws.close()

18 Why Twilio — Deep Dive

Twilio is the recommended telephony platform for connecting voice agents to the phone network. It provides the bridge between PSTN/SIP phone calls and your WebSocket-based voice agent pipeline.

18A Twilio Voice Agent Architecture

18B Twilio Media Streams Protocol

Twilio Media Streams is the API that connects phone calls to your voice agent via WebSocket. Understanding its message protocol is essential.

18C Twilio Complete Implementation

# ============================================
# Complete Twilio Voice Agent (FastAPI)
# Integrates: Twilio + Deepgram + LLM + TTS
# ============================================

import asyncio, json, base64
from fastapi import FastAPI, WebSocket, Request
from fastapi.responses import HTMLResponse
from twilio.twiml.voice_response import VoiceResponse, Connect

app = FastAPI()

# ─── 1. WEBHOOK: Twilio calls this when a call arrives ───
@app.post("/twilio-webhook")
async def twilio_webhook(request: Request):
    """Twilio hits this endpoint when someone calls your number.
    Returns TwiML that tells Twilio to open a Media Stream."""
    response = VoiceResponse()

    # Optional: play greeting before connecting to AI
    response.say("Connecting you to our AI assistant.", voice="Polly.Joanna")

    # Connect call audio to your WebSocket
    connect = Connect()
    stream = connect.stream(
        url=f"wss://your-server.com/twilio-stream",
        status_callback="https://your-server.com/stream-status",
        status_callback_method="POST",
    )
    # Pass custom parameters to your WebSocket handler
    stream.parameter(name="caller_number", value=str(request.form.get("From", "")))
    stream.parameter(name="call_sid", value=str(request.form.get("CallSid", "")))

    response.append(connect)
    return HTMLResponse(content=str(response), media_type="application/xml")


# ─── 2. WEBSOCKET: Receives real-time audio from Twilio ───
@app.websocket("/twilio-stream")
async def twilio_media_stream(ws: WebSocket):
    await ws.accept()

    # State for this call
    stream_sid = None
    call_sid = None
    caller_number = None
    is_agent_speaking = False

    # Initialize pipeline components
    deepgram_stt = DeepgramSTTEngine(
        api_key=DG_API_KEY,
        on_transcript=lambda t, **kw: handle_transcript(t, ws, stream_sid, **kw),
        on_speech_started=lambda: handle_barge_in(ws, stream_sid),
    )
    await deepgram_stt.connect()

    try:
        async for message in ws.iter_text():
            data = json.loads(message)
            event = data["event"]

            if event == "connected":
                print("✓ Twilio WebSocket connected")

            elif event == "start":
                stream_sid = data["start"]["streamSid"]
                call_sid = data["start"]["callSid"]
                custom = data["start"].get("customParameters", {})
                caller_number = custom.get("caller_number")
                print(f"📞 Call started: {call_sid} from {caller_number}")

                # Send initial greeting via TTS
                await send_tts_to_twilio(
                    "Hi! I'm your AI assistant. How can I help you today?",
                    ws, stream_sid
                )

            elif event == "media":
                # Decode base64 mulaw audio from Twilio
                audio_bytes = base64.b64decode(data["media"]["payload"])
                # Forward to Deepgram STT (accepts mulaw natively)
                await deepgram_stt.send_audio(audio_bytes)

            elif event == "dtmf":
                digit = data["dtmf"]["digit"]
                print(f"📱 DTMF: {digit}")

            elif event == "mark":
                # Audio playback reached a marker
                marker_name = data["mark"]["name"]
                if marker_name == "end_of_response":
                    is_agent_speaking = False

            elif event == "stop":
                print(f"📞 Call ended: {call_sid}")
                break

    except Exception as e:
        print(f"Error: {e}")
    finally:
        await deepgram_stt.close()


# ─── 3. HELPER: Send TTS audio back to Twilio ───
async def send_tts_to_twilio(text: str, ws: WebSocket, stream_sid: str):
    """Generate TTS audio and stream it back to the Twilio caller."""
    tts = ElevenLabsTTSEngine(api_key=ELEVEN_API_KEY, voice_id=VOICE_ID)
    # OR: tts = CartesiaTTSEngine(api_key=CARTESIA_KEY, voice_id=VOICE_ID)

    async for audio_chunk in tts.synthesize_for_twilio(text):
        payload = base64.b64encode(audio_chunk).decode("utf-8")
        await ws.send_json({
            "event": "media",
            "streamSid": stream_sid,
            "media": {"payload": payload}
        })

    # Add marker to know when playback finishes
    await ws.send_json({
        "event": "mark",
        "streamSid": stream_sid,
        "mark": {"name": "end_of_response"}
    })


# ─── 4. HELPER: Handle barge-in (user interrupts agent) ───
async def handle_barge_in(ws: WebSocket, stream_sid: str):
    """User started speaking while agent is talking. Clear audio."""
    await ws.send_json({
        "event": "clear",
        "streamSid": stream_sid,
    })
    print("⚡ Barge-in: cleared Twilio audio queue")

18D Twilio Advanced Features

from twilio.rest import Client

client = Client(TWILIO_SID, TWILIO_TOKEN)

# Transfer call to human agent queue
client.calls(call_sid).update(
    twiml='<Response><Dial><Queue>support</Queue></Dial></Response>'
)

# Enable recording via TwiML
response = VoiceResponse()
response.record(
    recording_status_callback="/recording-done",
    transcribe=True,
    max_length=3600,  # 1 hour max
)

call = client.calls.create(
    to="+1234567890",
    from_="+1987654321",  # Your Twilio #
    url="https://your-server.com/twilio-webhook",
    status_callback="https://your-server.com/call-status",
)

# In WebSocket handler:
elif event == "dtmf":
    digit = data["dtmf"]["digit"]
    if digit == "0":
        await transfer_to_human()
    elif digit == "*":
        await repeat_last_message()

19 WebRTC Integration

WebRTC provides peer-to-peer audio/video with built-in echo cancellation, noise suppression, and adaptive bitrate. Ideal for browser-based voice agents.

Frameworks with WebRTC: LiveKit Daily.co Pipecat

20 Interruption Handling (Barge-in)

Users will interrupt the agent mid-sentence. The agent must detect this, stop speaking immediately, and process the new input.

# Interruption handling logic
class InterruptionHandler:
    def __init__(self):
        self.is_agent_speaking = False
        self.playback_task = None
        self.audio_buffer = asyncio.Queue()

    async def on_user_speech_detected(self):
        """Called when VAD detects user speech during agent output."""
        if self.is_agent_speaking:
            # 1. Cancel current TTS playback
            if self.playback_task:
                self.playback_task.cancel()

            # 2. Flush audio buffer
            while not self.audio_buffer.empty():
                self.audio_buffer.get_nowait()

            # 3. Send clear message to client
            await self.send_clear_audio()

            self.is_agent_speaking = False
            print("⚡ Barge-in detected — agent stopped")

21 Voice AI Frameworks

Frameworks that provide pre-built pipelines for voice agent development, handling the complex orchestration of STT, LLM, TTS, and transport.

22 LiveKit Agents

Open-source framework for building real-time voice (and video) AI agents. Production-ready with WebRTC transport, plugin system, and turn detection.

# LiveKit Voice Agent
from livekit.agents import AutoSubscribe, JobContext, WorkerOptions, cli
from livekit.agents.voice_assistant import VoiceAssistant
from livekit.plugins import deepgram, openai, silero, cartesia

async def entrypoint(ctx: JobContext):
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)

    assistant = VoiceAssistant(
        vad=silero.VAD.load(),
        stt=deepgram.STT(model="nova-2"),
        llm=openai.LLM(model="gpt-4o"),
        tts=cartesia.TTS(voice="79a125e8-cd45-4c13-8a67-188112f4dd22"),

        # Interruption config
        interrupt_min_words=2,
        allow_interruptions=True,

        # Turn detection
        min_endpointing_delay=0.5,
    )

    assistant.start(ctx.room)
    await assistant.say("Hi! How can I help you today?")

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

23 Pipecat

Open-source framework (by Daily.co) for building voice and multimodal AI agents. Uses a pipeline architecture with composable processors.

# Pipecat Voice Pipeline
from pipecat.pipeline import Pipeline
from pipecat.transports.services.daily import DailyTransport
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.services.openai import OpenAILLMService
from pipecat.services.cartesia import CartesiaTTSService

transport = DailyTransport(room_url, token, "Voice Agent")
stt = DeepgramSTTService(api_key=DG_KEY)
llm = OpenAILLMService(model="gpt-4o", api_key=OAI_KEY)
tts = CartesiaTTSService(api_key=CART_KEY, voice_id="...")

pipeline = Pipeline([
    transport.input(),   # Audio from user (WebRTC)
    stt,                 # Speech → Text
    llm,                 # Text → Response text
    tts,                 # Response text → Audio
    transport.output(),  # Audio to user (WebRTC)
])

24 Vocode

Open-source library for building voice agents with telephony support (Twilio, Vonage). Good for phone-based agents.

Key features: Twilio integration, agent actions (transfer, end call), conversation management, endpointing configuration.

25 Managed Platforms (Vapi / Retell / Bland)

26 Function Calling in Voice Agents

Voice agents need to perform real actions — check databases, place orders, transfer calls. Function calling (tool use) lets the LLM trigger backend operations.

# Function calling with voice agent
tools = [
    {
        "type": "function",
        "function": {
            "name": "check_order_status",
            "description": "Check current status of a customer order",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"}
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "transfer_call",
            "description": "Transfer to human agent in specified department",
            "parameters": {
                "type": "object",
                "properties": {
                    "department": {"type": "string", "enum": ["billing", "support", "sales"]}
                }
            }
        }
    }
]

# During voice pipeline: when LLM returns tool_call
async def handle_tool_call(tool_call):
    # Say a filler while executing
    await tts.say("Let me check that for you...")

    result = await execute_function(tool_call.name, tool_call.arguments)

    # Feed result back to LLM for verbal response
    return result

27 Multimodal (GPT-4o Realtime API)

OpenAI's Realtime API provides speech-to-speech without separate STT/TTS — the model directly processes audio input and generates audio output.

# OpenAI Realtime API (WebSocket)
import websockets, json, base64

url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
headers = {"Authorization": f"Bearer {API_KEY}", "OpenAI-Beta": "realtime=v1"}

async with websockets.connect(url, extra_headers=headers) as ws:
    # Configure session
    await ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "modalities": ["text", "audio"],
            "voice": "alloy",
            "turn_detection": {"type": "server_vad", "threshold": 0.5},
            "tools": tools,
        }
    }))

    # Send audio frames directly
    await ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": base64.b64encode(audio_bytes).decode()
    }))

28 Emotion Detection & Sentiment

Detect user frustration, confusion, or satisfaction from voice cues (tone, pitch, pace) and text sentiment to adapt agent behavior.

29 Multilingual Support

30 Context & Memory

Voice conversations require persistent context across turns and sessions.

31 Deployment & Scaling

32 Monitoring & Analytics

Tools: Langfuse OpenTelemetry Grafana Datadog

32A Production Metrics — Numbers You Need for Interviews

When deployed in production, you need concrete metrics to prove your system works. Below are the actual KPIs a production voice agent should hit, how to measure them, and what to say in interviews.

32B Cost Per Call Analysis

Understanding your unit economics per call is critical for production planning and interviews. Here's the full breakdown:

32C Observability Implementation

Concrete code and configuration for production monitoring.

from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.metrics import MeterProvider
import time

tracer = trace.get_tracer("voice-agent")
meter  = metrics.get_meter("voice-agent")

# ── Define Metrics ────────────────────────────────────
call_counter      = meter.create_counter("voice.calls.total")
active_calls      = meter.create_up_down_counter("voice.calls.active")
turn_latency      = meter.create_histogram("voice.turn.latency_ms")
stt_latency       = meter.create_histogram("voice.stt.latency_ms")
llm_ttft          = meter.create_histogram("voice.llm.ttft_ms")
tts_ttfb          = meter.create_histogram("voice.tts.ttfb_ms")
barge_in_counter  = meter.create_counter("voice.barge_in.total")
error_counter     = meter.create_counter("voice.errors.total")
task_completion   = meter.create_counter("voice.task.completed")
escalation_count  = meter.create_counter("voice.escalations.total")
cost_per_call     = meter.create_histogram("voice.cost.per_call_usd")

# ── Trace a Full Conversational Turn ──────────────────
async def handle_turn(session, audio_chunk):
    with tracer.start_as_current_span("voice.turn") as span:
        span.set_attribute("call.id", session.call_id)
        turn_start = time.perf_counter()

        # STT
        with tracer.start_as_current_span("voice.stt"):
            t0 = time.perf_counter()
            transcript = await session.stt.transcribe(audio_chunk)
            stt_latency.record((time.perf_counter() - t0) * 1000)

        # LLM
        with tracer.start_as_current_span("voice.llm"):
            t0 = time.perf_counter()
            response_stream = session.llm.stream(transcript)
            first_token = await response_stream.__anext__()
            llm_ttft.record((time.perf_counter() - t0) * 1000)

        # TTS
        with tracer.start_as_current_span("voice.tts"):
            t0 = time.perf_counter()
            audio_out = await session.tts.synthesize_stream(first_token)
            tts_ttfb.record((time.perf_counter() - t0) * 1000)

        turn_latency.record((time.perf_counter() - turn_start) * 1000)

32D Load Testing & Benchmarks

Results from production load testing — use these numbers to answer interview questions about scale.

32E Business Impact Metrics

Translate technical metrics into business value — essential for stakeholder conversations and interviews.

32F Interview Cheat Sheet — Key Numbers

Quick-reference numbers to cite confidently in interviews when asked about your voice agent deployment.

33 Testing Strategies

34 Security & Privacy

35 Compliance

36 Glossary

37 Quick Reference — Recommended Stack