Voice & realtime agents

1 · What a voice agent is essential

A voice agent is an agent you talk to out loud and that talks back — a phone support line, a hands-free assistant, a drive-thru order taker. A realtime agent is one where the interaction is live and continuous: the user speaks, the agent responds within a fraction of a second, and either side can interrupt the other. That live, low-latency, interruptible quality is what makes it feel like a conversation instead of a walkie-talkie.

Under the hood a voice turn is a loop. Sound goes in through a microphone, becomes text or tokens, an agent decides what to say, that reply becomes sound, and it comes out a speaker. The whole loop has to close fast enough that the human doesn't feel a gap.

The dashed idea not drawn as a box: barge-in. While the speaker is playing the agent's reply, the mic is still listening. If the human starts talking, the agent must stop its own playback and listen — the loop jumps from Speaker back to Mic. Without that, talking over the agent does nothing and it feels robotic.

2 · Two architectures: pipeline vs speech-to-speech essential

There are two ways to build the loop. The classic pipeline chains three separate models: STT (speech-to-text) transcribes audio to words, an LLM reasons over the text, and TTS (text-to-speech) speaks the answer. Each stage is swappable and debuggable, but each adds latency, and tone/emotion is lost when audio becomes plain text.

Speech-to-speech (a native realtime/voice model) takes audio in and emits audio out directly — no text bottleneck in the middle. It's lower latency and keeps prosody (laughter, hesitation, emphasis), but it's harder to inspect, log, and guardrail because you don't automatically get a transcript.

3 · Streaming transport: why not request/response intermediate

A normal API call is request → wait → full response. That's fatal for voice: you'd record the whole sentence, upload it, wait for the whole reply, then play it — seconds of dead air. Realtime voice uses a persistent bidirectional connection (a WebSocket or WebRTC) so audio flows continuously in both directions. You send mic audio in small chunks (e.g. every 20 ms) as you capture it, and you start playing the reply as the first audio frames arrive — before the agent has finished its whole sentence. That overlap is what hides latency.

The runnable model below streams a 'reply' out in chunks and shows that the user hears the first audio long before the last chunk is produced — the core reason streaming feels instant. It is pure stdlib and prints a timeline.

streaming.pydef stream_reply(chunks, per_chunk_ms):
    """Yield (elapsed_ms, chunk) as if streaming audio out frame by frame."""
    elapsed = 0
    for chunk in chunks:
        elapsed += per_chunk_ms
        yield elapsed, chunk

reply = ["Sure,", " I can", " help", " with", " that."]
first_audio_at = None
last_audio_at = 0
for elapsed, chunk in stream_reply(reply, per_chunk_ms=120):
    if first_audio_at is None:
        first_audio_at = elapsed
    last_audio_at = elapsed

# In a NON-streaming design the user hears nothing until the whole reply is done.
print("time to FIRST audio (streaming): ", first_audio_at, "ms")
print("time to LAST audio (whole reply):", last_audio_at, "ms")
print("non-streaming would wait:        ", last_audio_at, "ms before ANY sound")

time to FIRST audio (streaming):  120 ms
time to LAST audio (whole reply): 600 ms
non-streaming would wait:         600 ms before ANY sound

4 · Turn detection: whose turn is it? intermediate

In text chat, pressing Enter marks end-of-turn. In voice there is no Enter — the agent must infer when the human has finished speaking so it can respond, and not cut them off mid-sentence. This is turn detection (a.k.a. endpointing), usually built on VAD (voice activity detection): detect speech vs silence, and when silence lasts longer than a threshold, treat the turn as over.

Model this as a tiny state machine. States: LISTENING (user talking), MAYBE_DONE (a silence gap started), and AGENT_TURN (silence held long enough — respond). This runs offline over a fake frame stream.

turns.pySILENCE_HANGOVER_MS = 500   # how long silence must hold before the turn ends

def detect_turns(frames, frame_ms=100):
    """frames: list of 'speech'/'silence'. Emit an event when a turn ends."""
    state = "LISTENING"
    silence_ms = 0
    events = []
    for i, f in enumerate(frames):
        t = i * frame_ms
        if f == "speech":
            silence_ms = 0
            state = "LISTENING"
        else:  # silence
            silence_ms += frame_ms
            if state == "LISTENING":
                state = "MAYBE_DONE"
            if state == "MAYBE_DONE" and silence_ms >= SILENCE_HANGOVER_MS:
                state = "AGENT_TURN"
                events.append(("turn_end", t))
                state = "LISTENING"     # reset for the next turn
                silence_ms = 0
    return events

# user says something, pauses briefly (not enough), then a real long pause
stream = (["speech"] * 6 + ["silence"] * 3      # 300 ms pause: too short
          + ["speech"] * 4 + ["silence"] * 6)   # 600 ms pause: turn ends
print(detect_turns(stream))

[('turn_end', 1700)]

5 · Barge-in: letting the human interrupt advanced

Barge-in is the ability to interrupt the agent while it is speaking — the single biggest thing that separates a natural voice agent from an IVR phone tree. When the user starts talking over the agent's playback, you must (1) immediately stop TTS playback, (2) discard the rest of the queued reply, and (3) switch to listening. The hard part is echo: the mic hears the agent's own voice from the speaker, so you need echo cancellation or you'll 'barge in' on yourself.

bargein.pyclass VoiceSession:
    def __init__(self):
        self.state = "IDLE"
        self.play_queue = []
        self.log = []

    def agent_speaks(self, chunks):
        self.state = "SPEAKING"
        self.play_queue = list(chunks)
        self.log.append(f"agent speaking, {len(self.play_queue)} chunks queued")

    def on_user_speech(self, is_agents_own_echo):
        # Echo cancellation: ignore the agent's own voice coming back in.
        if is_agents_own_echo:
            self.log.append("ignored echo of own voice")
            return
        if self.state == "SPEAKING":
            dropped = len(self.play_queue)
            self.play_queue.clear()          # stop + flush playback
            self.state = "LISTENING"
            self.log.append(f"BARGE-IN: stopped playback, dropped {dropped} chunks")

s = VoiceSession()
s.agent_speaks(["chunk1", "chunk2", "chunk3", "chunk4"])
s.on_user_speech(is_agents_own_echo=True)    # speaker bleed -> ignored
s.on_user_speech(is_agents_own_echo=False)   # real interruption -> barge-in
print("final state:", s.state)
for line in s.log:
    print("-", line)

final state: LISTENING
- agent speaking, 4 chunks queued
- ignored echo of own voice
- BARGE-IN: stopped playback, dropped 4 chunks

6 · The latency budget: why ~800 ms matters advanced

Humans notice conversational lag. In natural conversation, gaps between turns are roughly 200 ms; once the agent's response gap climbs past about 800 ms it starts to feel sluggish and people begin to talk over it. So end-to-end latency — from the user finishing their sentence to hearing the first word back — is a hard budget you must divide across every stage: capture + network up, STT finalization, LLM time-to-first-token, TTS time-to-first-audio, network down + playback.

The tool below adds up a per-stage budget and tells you whether you fit under a target, and how much headroom is left. Change the numbers to see where the time goes.

latency.pydef latency_budget(stages, target_ms=800):
    """stages: dict of stage -> milliseconds. Report total vs target."""
    total = sum(stages.values())
    headroom = target_ms - total
    worst = max(stages, key=stages.get)
    return {
        "total_ms": total,
        "target_ms": target_ms,
        "within_budget": total <= target_ms,
        "headroom_ms": headroom,
        "biggest_stage": worst,
    }

stages = {
    "capture+upload": 60,
    "stt_finalize":   150,
    "llm_first_token": 350,
    "tts_first_audio": 120,
    "network_down+play": 80,
}
result = latency_budget(stages, target_ms=800)
for k, v in result.items():
    print(f"{k}: {v}")

total_ms: 760
target_ms: 800
within_budget: True
headroom_ms: 40
biggest_stage: llm_first_token

7 · Production concerns: VAD, echo, fallback, cost professional

A demo that works at your desk breaks on a real phone line. The production checklist:

Production voice hardening

1. VAD / endpointing tuned per language and noise floor — noisy environments false-trigger on background sound.
2. Echo cancellation (AEC) so the mic doesn't hear the speaker; without it, barge-in fires on the agent's own voice.
3. Fallback path: if speech-to-speech or a stage times out, degrade gracefully (retry, switch to the text pipeline, or 'Sorry, I didn't catch that').
4. Interruptibility everywhere: any long agent action (a tool call, a lookup) must remain interruptible, or barge-in dies mid-task.
5. Cost & concurrency: realtime audio is billed by the second and holds an open connection per call — capacity planning is about concurrent sessions, not requests.

Cost for realtime voice is dominated by connection-time, not per-request tokens. Model the monthly bill from concurrent-call assumptions:

cost.pydef voice_cost(calls_per_day, avg_call_min, cost_per_min, days=30):
    """Realtime voice is billed by audio minute (in + out)."""
    minutes = calls_per_day * avg_call_min * days
    monthly = minutes * cost_per_min
    return {
        "billed_minutes_per_month": minutes,
        "monthly_cost_usd": round(monthly, 2),
        "cost_per_call_usd": round(avg_call_min * cost_per_min, 4),
    }

print(voice_cost(calls_per_day=2000, avg_call_min=3.0, cost_per_min=0.08))

{'billed_minutes_per_month': 180000.0, 'monthly_cost_usd': 14400.0, 'cost_per_call_usd': 0.24}

8 · Tech-lead — owning a voice-agent system tech-lead

A lead owns the end-to-end contract: a latency SLO (e.g. p95 first-audio < 800 ms) measured in production, an architecture decision (pipeline vs speech-to-speech) justified by that SLO and by observability needs, a transcription path for logging/guardrails even in speech-to-speech, and a fallback + concurrency plan so a bad network or a traffic spike degrades gracefully instead of dropping calls.

The real SDK usage looks like the sketch below. It is illustrative only — it needs the vendor SDK, a network connection, and credentials, so it is NOT part of the runnable labs. Every concept in it (streaming, turn events, barge-in) you already modeled offline above.

sdk_sketch.py# ILLUSTRATIVE — requires a realtime voice SDK, credentials, and a live socket.
# Do NOT expect this to run with `python file.py`; the labs above model the ideas offline.
import realtime_voice_sdk as rv   # placeholder for the actual vendor SDK

async def run_call(mic_stream, speaker):
    session = await rv.connect(model="voice-realtime", voice="warm")
    async for event in session.stream(mic_stream):     # bidirectional websocket
        if event.type == "partial_transcript":
            pass                                        # show live captions
        elif event.type == "turn_end":                 # VAD says user finished
            await session.respond()
        elif event.type == "audio_out":
            speaker.play(event.audio)                   # stream out as it arrives
        elif event.type == "user_started_speaking":
            await session.cancel_response()             # BARGE-IN: stop + flush

stages = {"asr": 150, "llm_first_token": 400, "tts_first_audio": 120}  # ms
total = sum(stages.values())
BUDGET = 800
print("total:", total, "ms")                 # 670 ms
print("within budget" if total <= BUDGET else "TOO SLOW",
      f"({BUDGET - total} ms headroom)")      # within budget (130 ms headroom)

archs = {
    "pipeline (ASR+LLM+TTS)": {
        "latency": "higher (3 hops)", "transcripts": "native (text between hops)",
        "prosody": "lost/limited", "control": "high -- swap any stage"},
    "speech-to-speech": {
        "latency": "lower (1 model)", "transcripts": "must be derived",
        "prosody": "preserved", "control": "low -- opaque"},
}
for name, a in archs.items():
    print(name); [print(f"  {k}: {v}") for k, v in a.items()]

# regulated line needs auditable transcripts + stage control:
print("\nRecommendation: PIPELINE -- transcripts are first-class and each "
      "stage is swappable/loggable, which a compliance audit requires.")

def rms(frame):
    return (sum(x * x for x in frame) / len(frame)) ** 0.5

def detect_turns(frames, threshold=0.2, hangover=3):
    speaking = False; silence = 0; turns = []; start = None
    for i, f in enumerate(frames):
        loud = rms(f) > threshold
        if loud:
            silence = 0
            if not speaking: speaking = True; start = i
        elif speaking:
            silence += 1
            if silence >= hangover:            # end turn after enough silence
                turns.append((start, i - hangover)); speaking = False
    if speaking: turns.append((start, len(frames) - 1))
    return turns

loud, quiet = [0.5, 0.5], [0.0, 0.0]
frames = [quiet, loud, loud, quiet, quiet, quiet, loud]
print(detect_turns(frames))    # [(1, 2), (6, 6)]

class VoiceAgent:
    def __init__(self): self.state = "idle"
    def on_event(self, event):
        if event == "user_speaks":
            if self.state == "speaking":
                self.state = "listening"; return "BARGE-IN: stop TTS, listen"
            self.state = "listening"; return "listening"
        if event == "user_stops":
            self.state = "thinking"; return "run LLM"
        if event == "llm_done":
            self.state = "speaking"; return "start TTS"
        if event == "tts_done":
            self.state = "idle"; return "idle"
        return self.state

a = VoiceAgent()
for e in ["user_speaks", "user_stops", "llm_done", "user_speaks", "tts_done"]:
    print(e, "->", a.on_event(e))
# the 2nd user_speaks arrives while speaking -> BARGE-IN

class VoiceService:
    def __init__(self): self.fallbacks = 0
    def respond(self, transcript, llm):
        try:
            reply = llm(transcript, timeout=1.5)   # must beat the latency budget
            if not reply:
                raise ValueError("empty reply")
            return reply
        except Exception as e:
            self.fallbacks += 1
            return "Sorry, could you repeat that?"  # holding phrase, never silence

def flaky_llm(text, timeout):
    if "trigger" in text:
        raise TimeoutError("model slow")
    return f"You said: {text}"

svc = VoiceService()
print(svc.respond("hello", flaky_llm))    # You said: hello
print(svc.respond("trigger", flaky_llm))  # holding phrase
print("fallbacks:", svc.fallbacks)        # 1

from anthropic import Anthropic

client = Anthropic()

def stream_reply(user_text, on_token):
    with client.messages.stream(
        model="claude-sonnet-4-5",
        max_tokens=200,
        messages=[{"role": "user", "content": user_text}],
    ) as stream:
        for text in stream.text_stream:      # tokens as they arrive
            on_token(text)                    # feed straight into TTS
        return stream.get_final_message()

# on_token would push each chunk to the TTS engine so audio starts
# after the FIRST sentence, not the whole reply.

Knowledge check check yourself