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. 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. 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 WebRTC VAD 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.
# 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 — 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. 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
- Raw WebSocket Voice Agent with AssemblyAI's Voice Agent API
Shittu Olumide 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 Twitter.