Building Voice-Controlled AI Agents

--> Building Voice-Controlled AI Agents - KDnuggets

-->

Join Newsletter

Introduction

Most people picture building a voice agent as stitching three things together: speech-to-text (STT), a large language model (LLM), and text-to-speech (TTS). Wire them up, and you're done. That picture is correct as far as it goes, and it describes the simplest architecture, where each stage waits for the previous one to fully complete before starting. It's also not the production-standard pattern in 2026, because it's far too slow for anything that needs to feel like a real conversation.

The actual hard part isn't the prompt, and it isn't even the model. It's orchestration: latency, turn-taking, tool calls, and interruption handling, layered on top of that basic STT-LLM-TTS chain. This is the actual engineering challenge precisely: voice is a turn-taking problem, not a transcription problem; semantic end-of-turn detection, barge-in cancellation, streaming, and time-to-first-token are the levers that separate a voice agent that feels natural from one that feels like a phone tree with a chatbot bolted onto it.

This article breaks the pipeline into its real components — streaming speech recognition, turn detection, streaming generation, interruption handling, and tool calling under voice constraints — and shows what each one is responsible for, where it actually breaks, and includes a tested code excerpt that makes the responsibility concrete. None of the code here needs a live microphone or a paid API key to run; each component is demonstrated in isolation, the way you'd actually reason about it before deciding what your system needs.

Why the Sequential Pattern Doesn't Work

Start with the architecture choice underneath everything else, because it determines whether the rest of this article's concerns even apply to your system.

In the sequential pattern, the user speaks, STT transcribes the full utterance, the LLM generates the full response, TTS synthesizes the full audio, and only then does the user hear anything. It's the simplest pattern to build and reason about. It's also the slowest, because every stage sits idle waiting for the one before it to fully finish, and those delays stack on top of each other.

The streaming pattern is the production standard instead: each stage streams its output to the next incrementally. STT streams partial transcripts to the LLM, the LLM streams tokens to TTS, and TTS synthesizes and plays audio from the first complete sentence while the LLM is still generating everything after it. This is genuinely harder to build; it demands careful handling of interruptions, buffering, and partial state, which is exactly what the rest of this article walks through, but it's the only pattern that hits a usable latency budget.

That budget isn't a vague aspiration. Human conversation has a natural 200 to 300ms gap between speakers. Response delays beyond 500ms feel noticeably slow, and delays beyond 3 seconds cause most users to disengage or assume the system is broken. Current speech-to-speech systems cluster in the 0.8 to 3 second time-to-first-token range across leading providers, which means the architecture decision alone is what determines whether your agent lands in the "feels natural" zone or the "caller hangs up" zone, before a single word of the actual response has been considered.

Streaming Speech-to-Text

The first component's job in a voice agent is not "transcribe this audio file." It's continuously processing an incoming audio stream and emitting transcripts as the user is still speaking, then signaling once it's confident they've finished. Production STT for voice agents runs over a persistent WebSocket connection. Audio goes out in small chunks, roughly 50ms at a time, and streaming transcript events come back — not a single blocking call that returns text once at the very end.

This distinction matters because of how the transcript actually changes mid-stream. A real streaming STT engine emits partial events that update as more audio arrives and the model revises its best guess, followed by one final event once it's confident the words have settled. Accuracy on entities — order numbers, phone numbers, and proper nouns — matters disproportionately here, because a single misheard digit breaks a downstream function lookup entirely, in a way that a human listener would have caught by simply asking the caller to confirm.

streaming_stt.py

Prerequisites: Python 3.10+, standard library only

Run: python streaming_stt.py

import asyncio from dataclasses import dataclass from enum import Enum

class TranscriptEventType(Enum): PARTIAL = "transcript.user.delta" # live, still-changing transcript FINAL = "transcript.user" # confirmed, won't change again

@dataclass class TranscriptEvent: event_type: TranscriptEventType text: str confidence: float = 1.0

class MockStreamingSTT: """ Stands in for a real STT WebSocket connection. Real implementations send audio chunks and receive these same two event types back -- partial deltas while the user is mid-utterance, then one final event once the model is confident the words are settled. """ def init(self, simulated_utterance: str): words = simulated_utterance.split() self._partial_stages = [" ".join(words[:i]) for i in range(1, len(words) + 1)]

async def stream_events(self): for stage in self._partial_stages[:-1]: yield TranscriptEvent(TranscriptEventType.PARTIAL, stage, confidence=0.7) await asyncio.sleep(0) # yield control, simulating real async I/O yield TranscriptEvent(TranscriptEventType.FINAL, self._partial_stages[-1], confidence=0.97)

async def consume_transcript_stream(stt: MockStreamingSTT): """ The pattern every voice agent client implements: render partial transcripts live for responsiveness, but only act on the FINAL event downstream -- partials can and do change before that. """ final_transcript = None partial_count = 0

async for event in stt.stream_events(): if event.event_type == TranscriptEventType.PARTIAL: partial_count += 1 print(f" [partial] '{event.text}' (confidence={event.confidence})") elif event.event_type == TranscriptEventType.FINAL: final_transcript = event.text print(f" [FINAL] '{event.text}' (confidence={event.confidence})")

return final_transcript, partial_count

async def main(): stt = MockStreamingSTT("My order number is A B 3 7 9 2") final_text, n_partials = await consume_transcript_stream(stt) print(f"\nFinal transcript used downstream: '{final_text}'") print(f"Partial events received before final: {n_partials}")

asyncio.run(main())

How to run: python streaming_stt.py, no dependencies required.

The downstream code only ever acts on the single FINAL event, even though nine partial transcripts streamed in before it as the simulated utterance built up word by word. That separation — render partials for live feedback, act only on the confirmed final — is what every real streaming STT client implements, whether it's AssemblyAI's Voice Agent API or any other production endpoint.

Turn Detection: Deciding When the User Is Actually Done

This component is easy to skip mentally because it feels like it should just be part of the STT step. It isn't, and treating it as a separate concern is what makes it tunable. Turn detection is the system's specific method for deciding when the caller has finished speaking and the agent should respond, and it consumes the audio stream's silence pattern, not the transcript's text content, which is why it's a distinct piece of logic from STT.

Get this wrong in either direction, and the conversation breaks differently. Too eager, and the agent interrupts a speaker who paused mid-thought to think. Too slow, and every single exchange carries an awkward dead-air gap that makes the whole system feel sluggish even when the LLM itself responds instantly. Production systems control this with two numbers: a minimum silence duration before declaring end-of-turn, commonly around 600ms, which ends the turn only when the transcript side also suggests the utterance sounds finished, and a maximum silence ceiling that forces a response even on an ambiguous pause, often around 1500ms. Deliberate-speech contexts like eldercare or healthcare warrant raising that ceiling toward 2500ms; fast-paced conversational contexts warrant dropping the minimum toward 300ms. This is a tunable policy decision specific to your use case, not a fixed constant baked into the architecture.

turn_detection.py

Prerequisites: Python 3.10+, standard library only

Run: python turn_detection.py

from dataclasses import dataclass from enum import Enum

class TurnState(Enum): LISTENING = "listening" SILENCE_PENDING = "silence_pending" # silence detected, not yet long enough to decide END_OF_TURN = "end_of_turn"

@dataclass class AudioFrame: is_speech: bool timestamp_ms: int

class TurnDetector: """ Standalone turn-detection state machine -- consumes a stream of (is_speech, timestamp) frames and decides when the user has finished speaking. Deliberately separate from STT: STT produces transcripts; turn detection decides WHEN to stop listening and let the agent respond, using the silence pattern in the audio stream itself. """ def init(self, min_silence_ms: int = 600, max_silence_ms: int = 1500): self.min_silence_ms = min_silence_ms self.max_silence_ms = max_silence_ms self._silence_start: int | None = None self.state = TurnState.LISTENING

def process_frame(self, frame: AudioFrame, utterance_looks_complete: bool = True) -> TurnState: """ utterance_looks_complete carries the semantic signal from the transcript side -- whether what the user has said so far sounds like a finished thought. The minimum threshold ends the turn only when that signal agrees; the maximum threshold ends it regardless. """ if frame.is_speech:

Any speech resets the silence clock entirely

self._silence_start = None self.state = TurnState.LISTENING return self.state

if self._silence_start is None: self._silence_start = frame.timestamp_ms

silence_duration = frame.timestamp_ms - self._silence_start

if silence_duration >= self.max_silence_ms: self.state = TurnState.END_OF_TURN # hard ceiling -- force a response elif silence_duration >= self.min_silence_ms and utterance_looks_complete: self.state = TurnState.END_OF_TURN # confident enough silence has settled else: self.state = TurnState.SILENCE_PENDING

return self.state

if name == "main": print("Complete-sounding utterance -- the minimum threshold applies:") detector = TurnDetector(min_silence_ms=600, max_silence_ms=1500) frames = [ AudioFrame(True, 0), AudioFrame(True, 100), AudioFrame(True, 200), AudioFrame(False, 300), AudioFrame(False, 400), # brief pause -- a thinking pause AudioFrame(True, 500), AudioFrame(True, 600), # speaker resumes AudioFrame(False, 700), AudioFrame(False, 900), AudioFrame(False, 1100), AudioFrame(False, 1300), # silence clock reaches 600ms here ]

for f in frames: state = detector.process_frame(f) print(f" t={f.timestamp_ms:>5}ms speech={f.is_speech!s:>5} -> {state.value}")

print("\nUtterance that still sounds unfinished -- the ceiling applies:") trailing_detector = TurnDetector(min_silence_ms=600, max_silence_ms=1500) trailing_frames = [AudioFrame(True, 0)] + [AudioFrame(False, t) for t in range(100, 1800, 400)]

for f in trailing_frames: state = trailing_detector.process_frame(f, utterance_looks_complete=False) print(f" t={f.timestamp_ms:>5}ms speech={f.is_speech!s:>5} -> {state.value}")

How to run: python turn_detection.py, no dependencies required.

The pause between t=300ms and t=500ms never escalates past silence_pending, because the speaker resumes before the silence clock crosses the minimum threshold — exactly the kind of mid-sentence thinking pause that shouldn't end the tu

[truncated for AI cost control]