--- title: "I Built My Own Voice Agent in Python to Understand How It Works" author: "Samod Alex" published: "2026-11-02T12:15:00.023Z" updated: "2026-11-02T08:16:01.977Z" url: "https://zyvop.com/i-built-my-own-voice-agent-in-python-to-understand-how-it-works-uf1d9" canonical: "https://zyvop.com/i-built-my-own-voice-agent-in-python-to-understand-how-it-works-uf1d9" tags: ["text-to-speech","Python","Speech-to-Text","Claude","Voice Agents"] --- # I Built My Own Voice Agent in Python to Understand How It Works Build a voice agent from scratch in Python: it listens, thinks with Claude, talks back, and stops when you interrupt. Includes tests that run offline. ![Cover image]() 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: ```mermaid 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: ```css 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: ```bash 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: ```mermaid 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: ```mermaid 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. ```mermaid 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: ```mermaid 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 ``` ```bash 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: ```sql 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: ```bash 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.