Building Voice-Controlled AI Agents

AssemblyAI's 2026 guide to building voice-controlled AI agents identifies orchestration—latency, turn-taking, tool calls, and interruption handling—as the core engineering challenge, not the prompt or model. The article breaks the pipeline into streaming speech recognition, turn detection, streaming generation, interruption handling, and tool calling under voice constraints, noting that human conversation has a natural 200–300ms gap between speakers and that response delays beyond 500ms feel noticeably slow, with current speech-to-speech systems clustering in the 0.8 to 3 second time-to-first-token range across leading providers.

Building Voice-Controlled AI Agents Building a voice-controlled AI agents isn't hard, 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. 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 https://www.assemblyai.com/blog/build-a-voice-agent-5-minutes-voice-agent-api . 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 turn. Once the speaker actually stops at t=700ms, the clock runs uninterrupted and correctly fires end of turn at t=1300ms. The second run is where the ceiling earns its place: with the semantic signal saying the utterance still sounds unfinished, the minimum threshold is ignored entirely and the turn ends only once silence hits the hard ceiling. That's the entire value of separating min silence ms and max silence ms as two distinct, tunable numbers rather than a single fixed timeout. Streaming the Response Into Text-to-Speech This section makes the streaming architecture from the first section concrete at the handoff point that matters most. Once a turn is detected, the LLM should stream tokens as they're generated rather than waiting for the full response, and TTS should begin synthesizing audio from the first complete sentence rather than waiting for the entire reply. The unit that actually gets handed from the LLM stream to the TTS engine isn't a token and isn't the full response; it's a complete sentence, detected the instant its boundary appears in the accumulating buffer. sentence chunker.py Prerequisites: Python 3.10+, standard library only Run: python sentence chunker.py import asyncio import re SENTENCE END PATTERN = re.compile r' ?<= . ? \s+' async def mock llm token stream text: str : """ Stands in for a real streaming LLM call. Yields one token word at a time, simulating tokens arriving incrementally rather than the full response appearing all at once. """ for word in text.split " " : yield word + " " await asyncio.sleep 0 async def stream sentences token stream - list str : """ The handoff unit between LLM streaming and TTS synthesis: complete sentences, not raw tokens. The instant a sentence boundary appears in the accumulated buffer, that sentence is yielded so TTS can start speaking it while the LLM is still generating what comes after it. """ buffer = "" sentences = async for token in token stream: buffer += token match = SENTENCE END PATTERN.search buffer while match: sentence = buffer :match.start + 1 .strip sentences.append sentence print f" sentence ready for TTS '{sentence}'" buffer = buffer match.end : match = SENTENCE END PATTERN.search buffer Whatever remains once the stream ends is the final fragment -- still needs to be flushed to TTS even without terminal punctuation. if buffer.strip : sentences.append buffer.strip print f" final fragment flushed '{buffer.strip }'" return sentences async def main : text = "Let me check that for you. Your order shipped yesterday and " "should arrive Thursday. Is there anything else I can help with" sentences = await stream sentences mock llm token stream text print f"\nTotal sentences yielded: {len sentences }" asyncio.run main How to run : python sentence chunker.py , no dependencies required. Three sentences come out, and the first one, "Let me check that for you," is ready for TTS to start speaking well before the LLM has finished composing the third. That early handoff is the entire reason streaming TTS feels responsive: the user hears the agent start talking within a few hundred milliseconds of the LLM beginning to generate, instead of waiting for the complete response to finish first. Handling Interruption Without Breaking State Barge-in is widely treated as the single hardest part of voice agent engineering, and it's worth spending the most care on here too. Barge-in requires four things to happen together: stopping TTS playback, canceling in-flight TTS generation, canceling LLM generation, and resetting stream state https://inworld.ai/resources/best-speech-to-speech-apis . Miss any one of these, and the agent either talks over the user or, more confusingly, finishes its old thought out loud after being interrupted, which feels broken in a way that's hard to diagnose from the outside if you don't already know to look at all four steps individually. The piece that determines whether barge-in is reliable rather than just present is false-positive prevention. False-barge-in fires when the voice activity detector VAD mistakes background noise, a cough, or a side conversation for a genuine interruption, and the agent cuts itself off mid-sentence for no reason the user can perceive. Prevention combines three signals: an energy threshold, typically -45 to -35 decibels relative to full scale dBFS , a voice classifier such as Silero VAD or that distinguishes actual speech from noise, and a minimum-duration guard requiring 200 to 300ms of sustained voice before the barge-in actually fires. A single loud cough should never stop the agent mid-sentence; that's specifically what the duration guard exists to prevent. WebRTC VAD https://github.com/wiseman/py-webrtcvad bargein detector.py Prerequisites: Python 3.10+, standard library only Run: python bargein detector.py from dataclasses import dataclass @dataclass class AudioChunk: energy dbfs: float signal energy in dBFS voice confidence: float 0.0-1.0, output of a voice classifier like Silero VAD timestamp ms: int class BargeInDetector: """ Combines three signals to decide whether the user is genuinely interrupting the agent, or whether background noise, a cough, or a side conversation is being mistaken for real speech. Missing any one of these three checks is what causes false-barge-in. """ def init self, energy threshold dbfs: float = -40.0, within the -45 to -35 production range voice confidence threshold: float = 0.6, min duration ms: int = 250, within the 200-300ms production range : self.energy threshold = energy threshold dbfs self.voice threshold = voice confidence threshold self.min duration ms = min duration ms self. candidate start ms: int | None = None def process chunk self, chunk: AudioChunk - bool: """ Returns True the instant a real barge-in should fire -- i.e. all three conditions have held continuously for at least min duration ms. """ passes energy = chunk.energy dbfs self.energy threshold passes voice = chunk.voice confidence self.voice threshold if not passes energy and passes voice : Signal dropped below threshold -- reset the candidate window so a brief loud noise can't accumulate duration across separate bursts. self. candidate start ms = None return False if self. candidate start ms is None: self. candidate start ms = chunk.timestamp ms sustained duration = chunk.timestamp ms - self. candidate start ms return sustained duration = self.min duration ms if name == " main ": A genuine interruption: strong, sustained voice signal for 300ms detector 1 = BargeInDetector real interruption = AudioChunk -30, 0.9, t for t in range 0, 350, 50 fires 1 = detector 1.process chunk c for c in real interruption print f"Real interruption sustained 300ms : fired={any fires 1 }" A single short cough: high energy but drops immediately, never sustains detector 2 = BargeInDetector cough = AudioChunk -28, 0.8, 0 , AudioChunk -50, 0.1, 50 , AudioChunk -50, 0.1, 100 , fires 2 = detector 2.process chunk c for c in cough print f"Single cough <100ms : fired={any fires 2 }" Loud background noise: passes the energy threshold but fails voice classification detector 3 = BargeInDetector background noise = AudioChunk -32, 0.25, t for t in range 0, 400, 50 fires 3 = detector 3.process chunk c for c in background noise print f"Loud non-voice background noise: fired={any fires 3 }" How to run : python bargein detector.py , no dependencies required. The detector fires on the genuine sustained interruption and correctly stays silent on both the brief cough and the loud-but-not-voice-like background noise. That third case is the one worth dwelling on: noise that's loud enough to pass the energy threshold alone would trigger a false barge-in constantly in a noisy room, which is exactly why the voice classifier check exists as a second, independent gate rather than relying on volume alone. One production-reported failure mode is worth naming plainly before moving on: barge-in teardown becomes genuinely dangerous when downstream automation has already triggered before the interruption fires https://www.sigmamind.ai/blog/create-voice-ai-agent-step-by-step-architecture-66b0b — a booking pipeline call, or a database write that's already in flight. Canceling LLM token generation mid-stream is safe; the tokens just stop. Canceling a payment that's already left your system is a different problem entirely, and it's the reason the next section's tool-result buffering exists. Tool Calling Mid-Conversation Tool calling in a voice context has a problem that simply doesn't exist in a text-based chat interface: the gap between a tool call firing and its result arriving is audible. Dead air during a phone call makes users assume the call dropped, which prompts them to start talking and interrupt the tool call that's still in progress. In a text chat, a three-second pause while a function executes is invisible. On a phone call, it's the difference between feeling responsive and feeling broken. The documented fix has a name: the preamble technique, instructing the model to narrate what it's doing before and during a tool call, saying something like "Let me check that for you" or "One moment while I pull that up," which keeps the conversation audibly alive while the function actually executes. It's a prompting pattern, not a code pattern, but it solves a problem that's specific to voice and worth naming here because it pairs directly with the second hard problem this section covers. That second problem is what happens to a tool result if the user interrupts before it ever arrives. The documented production pattern is to accumulate tool results as they come in, and only actually send them once the current turn finishes cleanly, discarding any pending results entirely if the turn was interrupted instead https://www.assemblyai.com/blog/raw-websocket-voice-agent-voice-agent-api . Sending a stale tool result into a conversation that's already moved on past it creates exactly the kind of state confusion the previous section's barge-in handling exists to prevent in the first place. tool result buffer.py Prerequisites: Python 3.10+, standard library only Run: python tool result buffer.py from dataclasses import dataclass from enum import Enum class TurnOutcome Enum : CLEAN COMPLETION = "clean completion" INTERRUPTED = "interrupted" @dataclass class PendingToolResult: call id: str result: dict class ToolResultBuffer: """ Implements the documented production pattern: accumulate tool results as they arrive mid-turn, but only send them once the current turn finishes cleanly. If the turn was interrupted instead, discard everything pending -- sending a stale tool result into a conversation that already moved on is worse than not responding to the tool call at all. """ def init self : self. pending: list PendingToolResult = def accumulate self, call id: str, result: dict - None: self. pending.append PendingToolResult call id, result def resolve turn self, outcome: TurnOutcome - list PendingToolResult : """ Called when the current conversational turn ends. Flushes every pending result downstream on a clean completion, or discards all of them on an interruption -- there's no partial-credit path here. """ pending = list self. pending self. pending.clear if outcome == TurnOutcome.CLEAN COMPLETION: return pending return interrupted -- discard everything, send nothing if name == " main ": Tool call resolves, turn completes cleanly -- result is sent buffer 1 = ToolResultBuffer buffer 1.accumulate "call abc123", {"temp c": 22, "condition": "sunny"} flushed 1 = buffer 1.resolve turn TurnOutcome.CLEAN COMPLETION print f"Clean completion: {len flushed 1 } result s sent - {flushed 1}" Tool call resolves, but user interrupts before the turn completes -- the result must be discarded, not sent into a conversation that moved on buffer 2 = ToolResultBuffer buffer 2.accumulate "call def456", {"confirmation code": "CONF7821"} flushed 2 = buffer 2.resolve turn TurnOutcome.INTERRUPTED print f"Interrupted turn: {len flushed 2 } result s sent correctly discarded " Multiple parallel tool calls in one turn, resolved together buffer 3 = ToolResultBuffer buffer 3.accumulate "call weather", {"temp c": 18} buffer 3.accumulate "call calendar", {"next slot": "2026-06-22T14:00"} flushed 3 = buffer 3.resolve turn TurnOutcome.CLEAN COMPLETION print f"Parallel tool calls, clean completion: {len flushed 3 } result s sent" How to run : python tool result buffer.py , no dependencies required. The interrupted scenario sends zero results, even though the tool call itself completed successfully and produced a perfectly valid confirmation code. That's deliberate: the user has already moved the conversation somewhere else by the time that result would arrive, and injecting it anyway would be answering a question that's no longer the one being asked. The third scenario confirms the pattern holds for parallel tool calls too — both results flush together once the turn that contained them resolves cleanly, which matters because modern voice models support parallel tool calling, meaning multiple tools can fire simultaneously within a single turn. How the Components Actually Connect Putting the pieces back together: audio comes in, streaming STT emits partial transcripts as the words arrive, turn detection watches the silence pattern in that same audio stream and decides when the user has actually finished, the LLM streams a response while TTS begins speaking the first complete sentence well before the rest has been generated, barge-in can interrupt at any point downstream of the user starting to talk again, and a tool call, when the model needs to look something up or take an action, inserts a preamble-and-buffer detour into the middle of that flow rather than just leaving dead air. Vendors increasingly bundle this entire chain into a single WebSocket endpoint: AssemblyAI's Voice Agent API, OpenAI's Realtime API , and similar offerings handle STT, LLM orchestration, TTS, turn detection, and barge-in server-side over one connection, which is why most teams building voice agents in 2026 reasonably integrate against one of these rather than hand-rolling all five components covered in this article. That's a sound default. But understanding what each component does specifically, not just that "the voice agent" handles it, is what turns "the agent feels broken" from a mystery into a debuggable problem: a too-eager barge-in threshold, a missing preamble during a slow tool call, a transcript error on an order number that a slightly different VAD tuning would have caught. Conclusion A voice agent is not a chatbot with a microphone taped to one end and a speaker to the other. It's five components solving five problems that don't exist at all in text-based conversation: streaming transcription instead of a blocking call, turn detection as its own tunable policy rather than a fixed timeout, sentence-level handoff from the LLM to TTS instead of waiting for the full response, barge-in detection built from three combined signals rather than a single noise threshold, and tool-call result buffering that accounts for the user moving on before the result arrives. The latency budget underneath all of it is unforgiving — 500ms is roughly the line between feeling natural and feeling noticeably slow — and every one of these components either protects that budget or quietly breaks it. Most teams will reasonably build on a bundled realtime API rather than implementing all five from first principles. But knowing precisely what each piece is responsible for is what makes it possible to actually fix a voice agent that feels wrong, instead of just restarting it and hoping. Resources : Build a Voice Agent in 5 Minutes with AssemblyAI's Voice Agent API https://www.assemblyai.com/blog/build-a-voice-agent-5-minutes-voice-agent-api Raw WebSocket Voice Agent with AssemblyAI's Voice Agent API https://www.assemblyai.com/blog/raw-websocket-voice-agent-voice-agent-api is a software engineer and technical writer passionate about leveraging cutting-edge technologies to craft compelling narratives, with a keen eye for detail and a knack for simplifying complex concepts. You can also find Shittu on Shittu Olumide https://www.linkedin.com/in/olumide-shittu/