I Built My Own Voice Agent in Python to Understand How It Works

A developer tutorial demonstrates building a working voice agent in roughly 330 lines of Python plus tests, without any voice framework, using faster-whisper for local speech-to-text, Anthropic's Claude as the streaming LLM, and OpenAI's text-to-speech for output. The agent slices microphone input into 20 ms frames (320 samples at 16 kHz), detects speech by comparing each frame's RMS volume against a running background-noise estimate, and supports barge-in by clearing the playback buffer and cancelling the Claude reply when the user interrupts. The post argues the loop is made of ordinary parts — ears, transcription, brain, and mouth — and keeps all hardware access in a single audio_io.py file so the rest can be tested with fake audio.

Voice agents feel like magic until you take one apart. You talk, it pauses for a beat, it answers in a natural voice, and if you cut it off mid-sentence it stops and listens. Behind that is a small loop of ordinary parts, and none of them is exotic. In this post we build one from scratch in about 330 lines of Python, plus tests, with no voice framework. By the end you'll have an agent that listens through your microphone, thinks with Claude, talks back through your speakers, and stops talking the moment you interrupt it. The shape of a voice agent Every voice agent is the same four parts in a loop: 1. Ears: turn a stream of microphone audio into "someone is talking" and "they're done". 2. Transcription STT : turn what they said into text. 3. Brain LLM : decide what to say, ideally while streaming the answer. 4. Mouth TTS : turn the answer into audio and play it, and stop instantly if the person interrupts. Here is how the pieces connect: flowchart TD subgraph hear "Listening" mic "Microphone, 20 ms frames" -- vad "Turn detector" vad -- |"turn finished"| stt "Speech-to-text" end subgraph speak "Speaking" llm "Claude, streaming" -- split "Sentence splitter" split -- tts "Text-to-speech" tts -- buf "Playback buffer" buf -- spk "Speakers" end stt -- llm vad -.- |"barge-in: clear the buffer"| buf vad -.- |"barge-in: cancel the reply"| llm And the files we'll write, one per idea: | File | Job | |---|---| | vad.py | Detect speech and decide when a turn starts and ends | | providers.py | Whisper STT , Claude LLM , OpenAI TTS , each behind one small method | | sentences.py | Cut a streaming reply into speakable sentences | | playback.py | The buffer between the agent and the speakers | | audio io.py | The only file that touches real audio hardware | | agent.py | The orchestration, including interruptions | | main.py | Wiring | Keeping the hardware in one small file matters later: everything else can be tested with fake audio. Setup You need Python 3.10 or newer, a microphone, headphones I'll explain why below , and two API keys: one for Anthropic the brain and one for OpenAI the voice . Speech-to-text runs locally, so it needs no key. Create a folder with this layout. Every file is shown in full later in the post, in the order we build it: voice-agent/ requirements.txt vad.py sentences.py playback.py audio io.py providers.py agent.py main.py test voice agent.py Start with requirements.txt : anthropic httpx numpy sounddevice faster-whisper pytest The versions are left unpinned on purpose. Once everything works on your machine, run pip freeze and keep the exact versions you tested with. Then install and set your keys: python -m venv .venv && source .venv/bin/activate pip install -r requirements.txt export ANTHROPIC API KEY=your-key export OPENAI API KEY=your-key On Linux you may also need PortAudio sudo apt install libportaudio2 . On macOS and Windows the sounddevice package bundles it. The first run downloads the Whisper model. Steps 1 and 2: Hearing, and knowing when someone is done The microphone gives us a continuous stream of numbers. We slice it into 20 ms frames 320 samples at 16 kHz and ask a simple question of each frame: is this speech? A production system would use a trained model such as Silero VAD or WebRTC VAD. For learning, a much simpler idea is enough: speech is louder than the room. We compute the volume of each frame its RMS and compare it to a running estimate of the background noise. If a frame is three times louder than the noise, it counts as speech. The noise estimate needs one trick. Quiet frames update it quickly, but loud frames only nudge it. That way a fan switching on is eventually absorbed into the background, while a few seconds of speech barely moves the baseline. A single loud frame proves nothing, so the TurnDetector sits on top and runs a small state machine: stateDiagram-v2 direction LR Idle: Idle, keeps the last 300 ms of audio Speaking: Speaking, collects every frame -- Idle Idle -- Speaking: 120 ms of speech, emit start Speaking -- Idle: 700 ms of silence, emit end with the audio Three numbers control it: - start ms=120 : speech must last 120 ms before we believe it. A door slam or a cough is shorter than that. - end ms=700 : we call the turn finished after 700 ms of silence. This is the most important dial in the whole system. Shorter feels snappy but cuts people off when they pause to think. Longer is polite but adds dead air to every single reply. Every voice agent picks a spot on this dial. - preroll ms=300 : we only know someone is talking 120 ms after they started, by which point the first word is already gone. So we keep a rolling 300 ms of recent audio and prepend it to the utterance. python vad.py from collections import deque import numpy as np SAMPLE RATE = 16 000 FRAME MS = 20 FRAME SAMPLES = SAMPLE RATE FRAME MS // 1000 320 samples per frame FRAME BYTES = FRAME SAMPLES 2 16-bit audio, 2 bytes per sample class EnergyVAD: """Says 'speech' when a frame is clearly louder than the room's background noise.""" def init self, ratio=3.0, floor=150.0 : self.ratio = ratio how many times louder than the noise counts as speech self.floor = floor never trigger below this volume RMS, int16 units self.noise = 100.0 running estimate of the background level self.boost = 1.0 raise it to make triggering harder we use this while talking def is speech self, frame: bytes - bool: x = np.frombuffer frame, dtype=np.int16 .astype np.float32 rms = float np.sqrt np.mean x x if x.size else 0.0 speech = rms max self.floor, self.noise self.ratio self.boost Quiet frames update the noise estimate quickly. Loud frames only nudge it, so a fan that switches on is eventually treated as background instead of speech. rate = 0.0002 if speech else 0.05 self.noise += rate rms - self.noise return speech class TurnDetector: """Turns a stream of 20 ms frames into 'start' and 'end' with the audio events.""" def init self, vad, start ms=120, end ms=700, preroll ms=300, max s=20 : self.vad = vad self.start frames = start ms // FRAME MS speech needed before we believe it self.end frames = end ms // FRAME MS silence needed before we call it done self.max frames = max s 1000 // FRAME MS self.preroll = deque maxlen=preroll ms // FRAME MS self. reset def reset self : self.speaking = False self.run = 0 consecutive speech frames seen while idle self.silence = 0 consecutive silent frames seen while speaking self.frames = def feed self, frame: bytes : speech = self.vad.is speech frame if not self.speaking: self.preroll.append frame self.run = self.run + 1 if speech else 0 if self.run = self.start frames: self.speaking = True self.frames = list self.preroll keep the lead-in so the first word is not clipped return "start", None return None self.frames.append frame self.silence = 0 if speech else self.silence + 1 if self.silence = self.end frames or len self.frames = self.max frames: audio = b"".join self.frames self.preroll.clear self. reset return "end", audio return None Step 3: Three services behind three tiny interfaces Each external service gets one class with one method. That keeps the agent logic free of vendor details and makes every piece swappable. Speech-to-text is faster-whisper , a fast local implementation of OpenAI's Whisper. It receives one finished utterance and returns text. That is the simplest design, but transcription only starts once you stop talking. More on that at the end. The brain is Claude, streamed. We use a small, fast model because in a voice conversation latency matters more than depth. The system prompt is doing real work: everything the model writes will be read aloud, so it must avoid markdown, lists and emoji, and keep answers short. If you want smarter answers, swap in a larger model and accept a longer pause. Text-to-speech is one HTTP request per sentence that returns raw 24 kHz, 16-bit mono audio. Raw PCM means no decoding step, so the bytes can go straight to the speakers. python providers.py import os import anthropic import httpx import numpy as np from faster whisper import WhisperModel SYSTEM = "You are a voice assistant in a live spoken conversation. " "Answer in one to three short sentences of plain spoken English. " "No markdown, lists, emoji or abbreviations, because everything you write is read aloud. " "If the user interrupted you, just respond to what they said last." class WhisperSTT: """Local speech-to-text. Takes one finished utterance, returns its text.""" def init self, model="base.en" : self.model = WhisperModel model, device="cpu", compute type="int8" def transcribe self, pcm: bytes - str: audio = np.frombuffer pcm, dtype=np.int16 .astype np.float32 / 32768.0 segments, = self.model.transcribe audio, language="en", beam size=1, condition on previous text=False return " ".join s.text.strip for s in segments .strip class ClaudeLLM: """Streams the reply as it is written, so we can start speaking before it is finished.""" def init self, model="claude-haiku-5-5" : self.client = anthropic.AsyncAnthropic reads ANTHROPIC API KEY self.model = model async def stream self, history : async with self.client.messages.stream model=self.model, max tokens=300, system=SYSTEM, messages=history as stream: async for text in stream.text stream: yield text class OpenAITTS: """Text in, raw 24 kHz 16-bit mono audio out.""" def init self, voice="alloy", model="gpt-4o-mini-tts" : self.voice, self.model = voice, model self.http = httpx.AsyncClient timeout=30, headers={"Authorization": f"Bearer {os.environ 'OPENAI API KEY' }"} async def synth self, text: str - bytes: r = await self.http.post "https://api.openai.com/v1/audio/speech", json={"model": self.model, "voice": self.voice, "input": text, "response format": "pcm"}, r.raise for status return r.content Step 4: Speak sentence by sentence Claude streams its reply a few words at a time. We could wait for the whole reply before speaking, but that adds seconds. We could speak every word as it arrives, but TTS needs whole sentences to get the rhythm right. The compromise is to cut the stream into sentences and send each one to TTS the moment it is complete. The payoff is overlap. While one sentence is playing, the next is already being synthesized, so the listener hears a continuous voice even though the reply is still being written: sequenceDiagram participant C as Claude participant A as Agent participant T as Text-to-speech participant S as Speakers C-- A: sentence 1 complete A- T: synthesize sentence 1 T-- A: audio 1 A- S: queue audio 1, playback starts C-- A: sentence 2 complete A- T: synthesize sentence 2 Note over S: sentence 1 is still playing T-- A: audio 2 A- S: queue audio 2 behind audio 1 The splitter has two small details. A sentence ends at . , or ? followed by whitespace, so 3.5 stays in one piece. And it won't emit anything shorter than 15 characters on its own, because a lone "Hi." makes for an awkward TTS request and sounds clipped, so it waits and joins the next sentence. python sentences.py import re A sentence ends at . or ? plus any closing quote or bracket followed by whitespace. Requiring the whitespace keeps "3.5" and "example.com" in one piece. SENTENCE END = re.compile r""" . ? "' \ \s""" class SentenceSplitter: """Turns a stream of text deltas into sentences that are ready to be spoken.""" def init self, min chars=15 : self.buf = "" self.min chars = min chars "Hi." alone is a poor TTS request, so it waits for a friend def feed self, delta: str - list str : self.buf += delta out, pos = , 0 while m := SENTENCE END.search self.buf, pos : if len self.buf : m.end .strip < self.min chars: pos = m.end too short, keep accumulating continue out.append self.buf : m.end .strip self.buf, pos = self.buf m.end : , 0 return out def flush self - list str : rest, self.buf = self.buf.strip , "" return rest if rest else Step 5: The buffer between the agent and the speakers Here is the part that makes interruption possible. The sound card works on its own thread and demands audio on a fixed schedule: every few milliseconds it asks for the next chunk. The agent works on the asyncio thread and produces audio in bursts, one sentence at a time. The PlaybackBuffer sits between them. php flowchart LR agent "Agent, asyncio thread" -- |"push a sentence"| buf "PlaybackBuffer" buf -- |"read, every few ms"| card "Sound card, audio thread" agent -.- |"clear on barge-in"| buf It has three jobs: - read n is called by the sound card and always returns exactly n bytes, padded with silence when there is nothing to play. - clear is the barge-in button. It throws away everything that has not been played yet, so the speakers go quiet almost immediately, apart from whatever the sound card has already queued. - take spoken returns the sentences whose playback actually started, so the agent can remember what it really said rather than what it meant to say. python playback.py import threading from collections import deque class PlaybackBuffer: """Audio the agent wants to say. The sound card drains it, the agent fills it.""" def init self : self. lock = threading.Lock self. chunks = deque each item: sentence, pcm bytes, read position self. spoken = sentences whose playback has started def push self, sentence: str, pcm: bytes : pcm = pcm : len pcm // 2 2 int16 samples must be whole with self. lock: self. chunks.append sentence, pcm, 0 def read self, nbytes: int - bytes: """Called from the audio thread. Always returns exactly nbytes, padded with silence.""" out = bytearray with self. lock: while len out < nbytes and self. chunks: chunk = self. chunks 0 sentence, pcm, pos = chunk if pos == 0: self. spoken.append sentence take = pcm pos : pos + nbytes - len out out += take chunk 2 += len take if chunk 2 = len pcm : self. chunks.popleft out += b"\x00" nbytes - len out return bytes out def clear self : """Barge-in: throw away everything that has not been played yet.""" with self. lock: self. chunks.clear @property def busy self - bool: with self. lock: return bool self. chunks def take spoken self - list str : with self. lock: spoken, self. spoken = self. spoken, return spoken Both audio callbacks run on separate threads. The microphone callback hands frames to asyncio with call soon threadsafe , and the speaker callback just pulls from the buffer. python audio io.py import asyncio import sounddevice as sd from vad import FRAME SAMPLES, SAMPLE RATE TTS RATE = 24 000 async def mic frames : """Yields 20 ms chunks of microphone audio as raw int16 bytes.""" loop = asyncio.get running loop q: asyncio.Queue bytes = asyncio.Queue def on audio indata, frames, time info, status : runs on the audio thread loop.call soon threadsafe q.put nowait, bytes indata with sd.RawInputStream samplerate=SAMPLE RATE, channels=1, dtype="int16", blocksize=FRAME SAMPLES, callback=on audio : while True: yield await q.get def speaker buffer : """An output stream that plays whatever is in the PlaybackBuffer, and silence otherwise.""" def fill outdata, frames, time info, status : runs on the audio thread outdata : = buffer.read frames 2 return sd.RawOutputStream samplerate=TTS RATE, channels=1, dtype="int16", blocksize=1024, callback=fill Step 6: The agent, and what happens when you interrupt it This is where everything comes together. Read run first: it feeds each microphone frame to the turn detector and reacts to two events. - end means the person finished a turn. We start a new turn task: transcribe, stream Claude's reply, split into sentences, synthesize, and push the audio into the buffer. - start while we are talking means they have begun speaking over us. We call interrupt , which clears the buffer first silence is the priority and then cancels the reply task. Here is one interruption from start to finish: sequenceDiagram autonumber participant D as TurnDetector participant A as Agent participant C as Claude participant B as PlaybackBuffer participant S as Speakers D- A: end of turn, with audio A- C: transcript, stream a reply C-- A: sentences stream back A- B: push audio for each sentence B- S: sentence 1 starts playing D- A: start, you are talking over it A- B: clear A-xC: cancel, which closes the stream Note over A: history keeps sentence 1 only D- A: end of turn, with the new audio A- C: new transcript, with that history Four details are worth slowing down for. The turn task stays alive while audio plays. The loop at the end of turn waits for the buffer to drain. Without it, the task would finish as soon as the last sentence was synthesized, which is typically long before it finishes being spoken, and there would be nothing left to cancel when the user interrupts. Cancellation reaches all the way to Claude. Cancelling the task unwinds the aclosing block around the stream, which closes the connection, so you stop generating words nobody will hear. The history records what was said, not what was planned. In the finally block we append only the sentences that started playing. If Claude wrote three sentences and you cut it off during the first, the history contains one. Otherwise the model would believe you heard an explanation you never did, and its next answer would build on it. If the interruption lands before any audio plays, no assistant message is stored, and add user merges your next utterance into the previous user message so the roles keep alternating. Our own voice can interrupt us. While the agent talks, its voice leaks from the speakers into the microphone, and the detector can mistake it for you. BARGE IN BOOST raises the bar to 2.5 times the usual threshold while we're speaking. That is a patch, not a fix, which is why the setup says to wear headphones. The real fix is acoustic echo cancellation, which browsers and phone stacks do for you. python agent.py import asyncio import time from contextlib import aclosing from playback import PlaybackBuffer from sentences import SentenceSplitter from vad import EnergyVAD, TurnDetector BARGE IN BOOST = 2.5 while we talk, the mic must hear someone 2.5x louder to count it as an interruption class VoiceAgent: def init self, stt, llm, tts, buffer: PlaybackBuffer : self.stt, self.llm, self.tts, self.buffer = stt, llm, tts, buffer self.detector = TurnDetector EnergyVAD self.history: list dict = self.task: asyncio.Task | None = None the reply currently being generated or spoken async def run self, frames : async for frame in frames: Our own voice leaks into the mic, so be harder to impress while we are talking. self.detector.vad.boost = BARGE IN BOOST if self.buffer.busy else 1.0 event = self.detector.feed frame if event is None: continue kind, audio = event if kind == "start" and self.buffer.busy: await self.interrupt they started talking over us: stop now elif kind == "end": await self.interrupt drop any reply that is still in flight self.task = asyncio.create task self.turn audio, time.monotonic async def interrupt self : self.buffer.clear silence first, the rest can follow if self.task and not self.task.done : self.task.cancel await asyncio.gather self.task, return exceptions=True async def turn self, audio: bytes, t0: float : marks = {} try: text = await asyncio.to thread self.stt.transcribe, audio marks "stt" = time.monotonic - t0 if len text < 2: noise, or Whisper imagining words in silence return self.add user text print f"you: {text}" splitter = SentenceSplitter async with aclosing self.llm.stream list self.history as deltas: async for delta in deltas: marks.setdefault "first token", time.monotonic - t0 for sentence in splitter.feed delta : await self.say sentence, marks, t0 for sentence in splitter.flush : await self.say sentence, marks, t0 while self.buffer.busy: stay alive until the audio is out, so we can be cut off await asyncio.sleep 0.02 finally: Whether we finished or were interrupted, remember only what was actually said aloud. spoken = self.buffer.take spoken if spoken: self.history.append {"role": "assistant", "content": " ".join spoken } async def say self, sentence: str, marks: dict, t0: float : pcm = await self.tts.synth sentence if "first audio" not in marks: marks "first audio" = time.monotonic - t0 print f"time: stt {marks 'stt' :.2f}s, first token {marks 'first token' :.2f}s, " f"first audio {marks 'first audio' :.2f}s counted from the end-of-turn decision " self.buffer.push sentence, pcm print f"agent: {sentence}" def add user self, text: str : if self.history and self.history -1 "role" == "user": the last reply was never heard: merge self.history -1 "content" += " " + text else: self.history.append {"role": "user", "content": text} Step 7: Run it The entry point only wires the parts together: python main.py import asyncio from agent import VoiceAgent from audio io import mic frames, speaker from playback import PlaybackBuffer from providers import ClaudeLLM, OpenAITTS, WhisperSTT async def main : buffer = PlaybackBuffer agent = VoiceAgent WhisperSTT , ClaudeLLM , OpenAITTS , buffer with speaker buffer : print "Listening. Wear headphones. Ctrl-C to quit." await agent.run mic frames if name == " main ": try: asyncio.run main except KeyboardInterrupt: pass python main.py Say something, and you'll see your transcript, the agent's sentences as they are queued for speech, and a timing line like this for every reply: you: what is a socket time: stt 0.xxs, first token 0.xxs, first audio 0.xxs counted from the end-of-turn decision agent: A socket is one end of a network connection. The x digits depend on your machine and your providers, so I won't quote numbers. Two things to keep in mind when reading them: - The clock starts when the turn detector decides you're done, so add the 700 ms endpointing wait to get what you actually experience. - The first audio mark is when audio enters the buffer, a few milliseconds before you hear it. A common target for a conversation that feels natural is roughly one second from the end of your speech to the start of the reply. Use the three timings to see which stage is eating your budget. Testing a voice agent without a microphone It sounds like voice code can only be tested by talking to it. But the hard parts of a voice agent are timing and state, not audio: when does a turn start, what counts as an interruption, what does the history contain afterwards. All of that can be tested with synthetic audio and fakes for each service. Here is the complete test file. It imports only vad.py , sentences.py , playback.py and agent.py , never providers.py or audio io.py , so it runs on a machine with no PortAudio, no API keys and no Whisper model. python test voice agent.py import asyncio import numpy as np from agent import VoiceAgent from playback import PlaybackBuffer from sentences import SentenceSplitter from vad import FRAME BYTES, FRAME SAMPLES, SAMPLE RATE, EnergyVAD, TurnDetector ---------- helpers: synthetic audio ---------- def to frames x : x = x.astype np.int16 return x i : i + FRAME SAMPLES .tobytes for i in range 0, len x , FRAME SAMPLES def noise n frames, amp=30, seed=0 : rng = np.random.default rng seed return to frames rng.normal 0, amp, n frames FRAME SAMPLES def tone n frames, amp=3000, hz=217.3 : odd frequency, so no two frames are identical t = np.arange n frames FRAME SAMPLES / SAMPLE RATE return to frames amp np.sin 2 np.pi hz t def events frames, detector=None : detector = detector or TurnDetector EnergyVAD out = for i, f in enumerate frames : e = detector.feed f if e: out.append i, e return out ---------- turn detection ---------- def test detects start and end of a turn : speech = tone 50 got = events noise 25 + speech + noise 60, seed=1 assert kind for , kind, in got == "start", "end" start i, , , end i, , audio = got assert 25 + 5 <= start i <= 25 + 7 about 120 ms after the first loud frame assert 75 + 33 <= end i <= 75 + 36 about 700 ms after the last loud frame The lead-in is kept, so the first word is not clipped: the first speech frame is in there. assert audio.find speech 0 == 9 FRAME BYTES 9 frames of lead-in, then the first speech frame def test short blip is ignored : assert events noise 25 + tone 3 + noise 50, seed=1 == def test steady loud room stops counting as speech : frames = noise 1500, amp=400 a fan, 30 seconds of it got = events frames assert got it fools the detector at first... assert all i < 1000 for i, kind, in got if kind == "start" ...but it adapts ---------- sentence splitting ---------- TEXT = "Sure thing. A socket is one end of a network connection. It has an address and a port Does that help?" EXPECTED = "Sure thing. A socket is one end of a network connection.", "It has an address and a port ", "Does that help?", def split all chunks : s, out = SentenceSplitter , for c in chunks: out += s.feed c return out + s.flush def test sentences do not depend on how the stream is chunked : assert split all TEXT == EXPECTED assert split all list TEXT == EXPECTED one character at a time assert split all TEXT i : i + 7 for i in range 0, len TEXT , 7 == EXPECTED def test decimals are not sentence ends : assert split all "It is about 3.5 miles away. Really." == "It is about 3.5 miles away.", "Really." ---------- playback buffer ---------- def test playback buffer reads across chunks and pads with silence : b = PlaybackBuffer b.push "a", b"\x01\x00" 5 10 bytes b.push "b", b"\x02\x00" 3 6 bytes assert b.read 8 == b"\x01\x00" 4 assert b.take spoken == "a" assert b.read 8 == b"\x01\x00" + b"\x02\x00" 3 assert b.take spoken == "b" assert not b.busy assert b.read 4 == b"\x00" 4 def test clear keeps only sentences that started playing : b = PlaybackBuffer b.push "a", b"\x01\x00" 5 b.push "b", b"\x02\x00" 5 b.read 2 b.clear assert not b.busy assert b.read 4 == b"\x00" 4 assert b.take spoken == "a" ---------- the whole agent, no microphone, no network ---------- class FakeSTT: def init self, lines : self.lines = list lines def transcribe self, pcm : return self.lines.pop 0 class FakeLLM: def init self, replies : self.replies, self.seen = list replies , async def stream self, history : self.seen.append dict m for m in history for sentence in self.replies.pop 0 : for word in sentence.split : yield word + " " await asyncio.sleep 0 class FakeTTS: async def synth self, text : await asyncio.sleep 0 return b"\x01\x00" 24 000 exactly one second of audio per sentence def drive script, stt lines, replies : """Runs the agent against a scripted microphone. The 'sound card' plays 20 ms per frame.""" buffer = PlaybackBuffer llm = FakeLLM replies agent = VoiceAgent FakeSTT stt lines , llm, FakeTTS , buffer async def pace frames : for f in frames: buffer.read 960 20 ms of audio at 24 kHz await asyncio.sleep 0.002 let the agent's tasks make progress yield f async def mic : async for f in script buffer, pace : yield f async def go : await agent.run mic while buffer.busy: let the last reply finish playing buffer.read 960 await asyncio.sleep 0.001 if agent.task: await agent.task asyncio.run go return agent, llm S1 = "A socket is one end of a network connection." S2 = "You can think of it as a phone jack." S3 = "Programs plug into it to talk." S4 = "A port picks which program gets the data." def test barge in stops the agent and history keeps only what was said : async def script buffer, pace : async for f in pace noise 25 : yield f async for f in pace tone 40 : yield f "what is a socket" while not buffer. spoken: wait for the agent to start talking async for f in pace noise 1 : yield f async for f in pace noise 15 : yield f it talks for a moment async for f in pace tone 40 : yield f "no wait, what is a port" async for f in pace noise 300, seed=2 : yield f agent, llm = drive script, "what is a socket", "no wait, what is a port" , S1, S2, S3 , S4 assert len llm.seen == 2 assert llm.seen 0 == {"role": "user", "content": "what is a socket"} assert llm.seen 1 == the model is told only what the user actually heard {"role": "user", "content": "what is a socket"}, {"role": "assistant", "content": S1}, {"role": "user", "content": "no wait, what is a port"}, assert agent.history -1 == {"role": "assistant", "content": S4} def test uninterrupted reply is spoken in full : async def script buffer, pace : async for f in pace noise 25 : yield f async for f in pace tone 40 : yield f async for f in pace noise 400, seed=2 : yield f agent, = drive script, "what is a socket" , S1, S2 assert agent.history == {"role": "user", "content": "what is a socket"}, {"role": "assistant", "content": f"{S1} {S2}"}, def test empty transcript is ignored : async def script buffer, pace : async for f in pace noise 25 : yield f async for f in pace tone 40 : yield f async for f in pace noise 100, seed=2 : yield f agent, llm = drive script, "" , assert llm.seen == and agent.history == Run it from the project folder: pytest The file has two halves worth reading closely. Turn detection. Frames of low noise and a sine tone stand in for speech. We check that turns start and end where they should, that the lead-in audio is kept, that a short blip is ignored, and that a steady loud room stops fooling the detector. The sentence splitter and the playback buffer get plain input and output checks. The whole agent. Speech-to-text, Claude and text-to-speech are replaced by fakes that return scripted text and one second of audio per sentence. A fake sound card reads 20 ms from the playback buffer for every microphone frame, so time is counted in frames instead of seconds and the scenario is deterministic. In the barge-in test, the user asks a question, the agent starts speaking, and 15 frames later the user interrupts. We then check what the model is told on the next turn: the question, only the first sentence of the interrupted answer, and the interruption. Two lessons from writing these: - Break your code on purpose. Remove the barge-in branch, record the whole planned reply, drop the pre-roll audio, and confirm a test fails each time. This exposed a weak pre-roll test: a tone at a round frequency repeats every few frames, so the first speech frame looked identical to a later one and the test passed even without the lead-in. An odd frequency such as 217.3 Hz fixed it. - Know what the tests do not cover. These tests cover the logic: turn detection, sentence splitting, the buffer, and the whole barge-in flow. They cannot cover your microphone, your speakers, or the three real services. Those you exercise by running the agent. From empty folder to running agent 1. Create the folder and requirements.txt from the Setup section, then install the dependencies. 2. Copy the seven source files from Steps 1 to 7, plus test voice agent.py , into the folder. 3. Run pytest . All ten tests should pass. 4. Set the two API keys, put on headphones, and run python main.py . The first start downloads the Whisper model, so give it a moment. 5. Say something, then interrupt the answer halfway through. What this version gets wrong It is a learning tool, and it has real limits: - The detector is naive. Volume is a weak proxy for speech. Music, a loud keyboard or a second voice will trigger it. A trained VAD fixes most of that. - Transcription waits for you to finish. Streaming STT, which transcribes while you speak, takes the whole transcription time off the critical path and is usually the biggest latency win available. - TTS is one request per sentence. Streaming the audio bytes as they are generated would cut the delay before the first word. - Whisper imagines words in near-silence. It sometimes returns a phantom "Thank you." for background noise. The len text < 2 check only catches the empty case, so a real system filters more aggressively. - Errors are not handled. An exception inside the turn task is easy to miss, because nobody awaits the task. Wrap turn in a try/except that logs and recovers, and retry failed API calls. - The history grows forever. Cap or summarize it for long conversations. - There are no tools. A real agent would look things up and take actions. That means handling tool-use blocks in the Claude stream, and deciding what the agent says while a tool runs. When to use a framework instead Building it by hand shows where the time goes and why each design choice exists, but for production you would normally not keep it. LiveKit Agents and Pipecat handle transport, echo cancellation, trained turn-taking models and telephony. Speech-to-speech models skip the three-stage pipeline entirely, trading control over each stage for lower latency and more natural speech. Having built the loop yourself, you'll know what those tools do for you, and which knob to reach for when something feels off: the pause, the cut-off, or the interruption that didn't land.