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:
Ears: turn a stream of microphone audio into "someone is talking" and "they're done".
Transcription (STT): turn what they said into text.
Brain (LLM): decide what to say, ideally while streaming the answer.
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"| llmAnd the files we'll write, one per idea:
File | Job |
|---|---|
| Detect speech and decide when a turn starts and ends |
| Whisper (STT), Claude (LLM), OpenAI (TTS), each behind one small method |
| Cut a streaming reply into speakable sentences |
| The buffer between the agent and the speakers |
| The only file that touches real audio hardware |
| The orchestration, including interruptions |
| 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.pyStart with requirements.txt:
anthropic
httpx
numpy
sounddevice
faster-whisper
pytestThe 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-keyOn 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 audioThree 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.
# 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 NoneStep 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.
# 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.contentStep 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 1The 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.
# 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.
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"| bufIt has three jobs:
read(n)is called by the sound card and always returns exactlynbytes, 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.
# 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 spokenBoth 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.
# 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.
endmeans the person finished a turn. We start a newturntask: transcribe, stream Claude's reply, split into sentences, synthesize, and push the audio into the buffer.startwhile we are talking means they have begun speaking over us. We callinterrupt(), 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 historyFour 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.
# 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:
# 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:
passpython main.pySay 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.
# 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:
pytestThe 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
Create the folder and
requirements.txtfrom the Setup section, then install the dependencies.Copy the seven source files from Steps 1 to 7, plus
test_voice_agent.py, into the folder.Run
pytest. All ten tests should pass.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.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) < 2check 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
turnin 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.
Comments(0)
Join the discussion by logging into your account.
No comments yet. Be the first to comment!