For the past two months now, I have been making Voice Agent Runtime, a distributed, self-hosted voice agent runtime built across a Go orchestrator and a Python inference engine, connected over a custom gRPC bidirectional stream. The runtime is designed so that agent behavior, such as persona, system prompt, domain; is entirely driven by configuration, not by code. There various ways you can make this using third party sources such as ElevenLabs, Twilio, Vapi, etc; so why i made this, tbh, I wanted something good for my project and since nearly everyone are making wrappers i thought to pull this off as a piece of development and moreover, i will gain knowledge as well. More I will write a separate blog on that…
Coming to the topic, true-duplex, so it is a mechanism that both the participating parties are allowed to exchange data bidirectionally, without blocking either of them. So exactly, if we send the packets, in a naive way, the channel gets locked (basic OS concept of read/write) so that data wont get corrupted on the way by the other parties. So this mechanism are made using Mutexes, in which it allows the system to just hold the resources if the other party wants to modify(write) to the same channel. This is called as half-duplex. So if we are making a system where both the parties can communicate, think of it like a phone call, neither of the party need to wait for the other one, whenever they need they can talk without data corruption.
The half-duplex can lead to Race-Condition and this is something we definitely dont want in our any app, and specfically in real-time application its a sin.
This was the picture in my project looked like:
def handle_session(stream):
for event in stream:
if event.type == "audio":
audio_buffer.append(event.audio)
if event.type == "end_of_utterance":
# STEP 1: Wait for ALL audio to arrive
full_audio = b''.join(audio_buffer)
# STEP 2: Transcribe (BLOCKS until done)
text = stt.transcribe(full_audio)
# STEP 3: LLM (BLOCKS until done)
response = llm.generate(text)
# STEP 4: TTS (BLOCKS until done)
audio_response = tts.synthesize(response)
# STEP 5: Send back (BLOCKS until sent)
yield Event(audio=audio_response)
audio_buffer.clear()
The problem:
TIME ──────────────────────────────────────────────────────────────>
CLIENT: [SPEAKING............][WAITING........................][SPEAKING]
│ │ │
▼ ▼ ▼
GO: [Receiving audio][Wait for Python][Send response][Wait for audio]
│
▼
PYTHON: [STT][LLM][TTS]
│ │ │
▼ ▼ ▼
BLOCK BLOCK BLOCK
Total latency: STT time + LLM time + TTS time + network
User experience: "Why is it pausing?"
Since the power of concurrency i was having using threads(python) and goroutines(Go) but these are just names for “an independent little worker inside a program that runs alongside other workers, instead of waiting in line behind them.
To fix this I need to implement something like:
import threading
import queue
# THE MAILBOX - thread-safe queue between workers
audio_queue = queue.Queue()
def handle_session(stream):
"""
This is the gRPC handler. It's the MAIN THREAD.
Its ONLY job is to be the write_pump (Worker 3).
"""
# Start Worker 1 (the listener) as a separate thread
listener_thread = threading.Thread(
target=read_pump,
args=(stream,),
daemon=True # Dies when main thread exits
)
listener_thread.start()
# THIS loop IS Worker 3 (the relay)
# It runs in the main thread because gRPC needs it here
while True:
# BLOCKS until audio chunk appears in queue
audio_chunk = audio_queue.get()
# Immediately send to Go
yield Event(audio=audio_chunk)
def read_pump(stream):
"""
WORKER 1: The Listener
Runs in its own thread, never stops.
"""
audio_buffer = []
for event in stream: # This BLOCKS waiting for Go
if event.type == "audio":
audio_buffer.append(event.audio)
elif event.type == "end_of_utterance":
# Got a complete utterance!
full_audio = b''.join(audio_buffer)
audio_buffer.clear()
# DON'T process here! Spawn a new worker!
worker = threading.Thread(
target=_run_utterance,
args=(full_audio,),
daemon=True
)
worker.start()
## Crucial Design Pattern
# Immediately go back to listening!
# Don't wait for the worker to finish!
def _run_utterance(audio):
"""
WORKER 2: The Thinker (temporary) does, ai part, then dies,
spawns again for the next utterance.
Created fresh for each utterance.
"""
# STT
text = stt.transcribe(audio)
# LLM - but we want STREAMING responses, not blocking!
for text_chunk in llm.generate_stream(text):
# TTS - also streaming!
for audio_chunk in tts.synthesize_stream(text_chunk):
# Put in the mailbox for Worker 3
audio_queue.put(audio_chunk)
# This thread dies here. Mission complete. Ohhooo!!
So now the visualization looks something like this:
TIME ──────────────────────────────────────────────────────────────────>
CLIENT: [SPEAKING utterance 1][SPEAKING utterance 2][SPEAKING...]
│ │
▼ ▼
GO: [Sending audio 1] [Sending audio 2]
│ │
▼ │
read_pump:[Receiving 1][end!] [Receiving 2][end!]
│ │ │ │
│ └─spawn thread──> │ └─spawn thread──> ...
▼ ▼
Thread 1: [STT][LLM──stream──>][TTS──stream──>]
│ │
▼ ▼
QUEUE: [chunk][chunk] [chunk][chunk]...
│ │
▼ ▼
write_pump: [Send][Send] [Send][Send]...
│ │
▼ ▼
GO: [Receiving audio response][Receiving more]...
│
▼
CLIENT: [HEARING response] [CAN STILL SPEAK!]
KEY INSIGHT: While Thread 1 is processing utterance 1,
read_pump is ALREADY receiving utterance 2!
Client can speak at ANY TIME.
Problem Solved! Not yet.
Remember I have written utterance in the above code. I didnt have thought that this small utterance and to figure this out; i.e. how should I know that i need to fire the worker 2; I need to spend one month dedicating to this.. and it was painful. But we live in the era of AI, so why need to worry, indeed this is the crucial part where i need to think humanely.
So, now the problem shifted - how to know when and what is the utterance we got and its proper indexed chunk. The problem of finding utterance and the correct time and with correct boundaries was so much in complex that I dedicated a complete one phase on it; which deserves its own blog.
TLDR; checkout HERE.
Now coming back to the architecture, i dont have much clue on how various big firms make these system decision and how they evolve and make it. I started writing a naive code logic where there was one massive function doing everything, then moved to the three-worker concept I showed you above. But here is the truth about system design that I learned the hard way: the proof-of-concept code is never the production code.
While the 3-worker architecture fixed the duplex problem, passing six different arguments into a spawned thread like
threading.Thread(
target=_run_utterance,
args=(
session_id,
audio,
prompt,
queue,
context,
event,
),
)
the above code gets messy incredibly fast. If you ever need to add a new feature, like tracking latency metrics or handling cancellations, you have to update every single thread spawn point.
From the very begining, I started this project as a software development gem, how exactly big firms make sota softwares; they dont rely upon API keys; they make, fail, understand the very concept of the work.
They Write Code that Evolves.
I treated my project the same way. Once the raw duplex streaming worked (Milestone 1 & 2; link), I spent the next few weeks refactoring the internals for clean state management before moving on to the VAD(voice-activity-detection; new blog soon).
If you look at my current main.py today, it looks vastly different from the snippet above. It evolved into something much more robust, primarily using Python's @dataclass to hold session state.
Instead of passing a loose collection of variables through threads, I created a SessionContext:
@dataclass
class SessionContext:
session_id: str
outbound_queue: queue.Queue
grpc_context: grpc.ServicerContext
utterance_done_event: threading.Event = field(default_factory=threading.Event)
system_prompt: str | None = None
Now, every worker thread just takes one thing: ctx. Need to check if the user disconnected? ctx.grpc_context.is_active(). Need to trigger the next sentence? ctx.outbound_queue.put(). It encapsulates the entire life of a single phone call into one clean object.
Secondly, the monolithic read_pump I wrote earlier got broken down into distinct, single-responsibility handlers:
def _read_pump(self, request_iterator, ctx: SessionContext) -> None:
vad = VADDetector(VAD_MODEL_PATH)
preprocessor = AudioPreprocessor(source_sr=SOURCE_SAMPLE_RATE)
ctx.utterance_done_event.set()
try:
for event in request_iterator:
ctx.session_id = event.session_id
if event.HasField("control"):
self._handle_control_event(ctx, vad, event.control)
elif event.HasField("audio"):
self._handle_audio_event(ctx, vad, preprocessor, event.audio)
# ...
Notice in the snippet their is silently sitting an event, called utterance_done_event, this alone was the life saver before integrating VAD to the system…
As you can see, how _read_pump is no longer doing the AI logic. It's just a router. If it sees an audio chunk, it hands it to _handle_audio_event. If it sees a control signal, it hands it to _handle_control_event. This modular design was an absolute lifesaver when it came to integrating the VAD (which, as I mentioned, deserves its own deep dive). Because I had separated the routing from the execution, plugging the VAD into _handle_audio_event meant I didn't have to touch my gRPC streaming logic or my thread management at all.
But there was one final, massive hurdle during this evolution: The "Two Bosses" Problem:
During my transition from Milestone 3 (Go sending manual boundaries) to Milestone 4 (Python VAD detecting boundaries autonomously), both systems were live at the same time. What if the VAD detected the end of a sentence at the exact millisecond Go decided to send a manual
END_OF_UTTERANCEsignal?
If I wasn’t careful, my system would dispatch two inference threads for the exact same audio, resulting in the LLM talking over itself.
To solve this, I implemented a gate using the utterance_done_event inside a new method called _dispatch_utterance:
def _dispatch_utterance(self, ctx: SessionContext, vad: VADDetector) -> None:
frames = vad.get_utterance_frames()
if len(frames) == 0:
return
utterance_bytes = frames_to_wav(frames, AudioPreprocessor.TARGET_SR)
# THE GATE: If a thread is already running, wait for it to finish.
if not ctx.utterance_done_event.is_set():
ctx.utterance_done_event.wait()
ctx.utterance_done_event.clear() # Lock the gate
# ... spawn the inference thread ...
Now, it doesn't matter who yells "I'm done!" first; Go or the VAD. The _dispatch_utterance method acts as an idempotency key. If the VAD fires, it locks the gate. If Go tries to fire 50ms later, the gate is locked, and the signal is safely ignored. No double-processing. No garbled audio.
The Final Picture
What started as a naive, blocking script evolved into a highly concurrent, state-managed, dual-trigger system.
Today, my _read_pump listens continuously. The VAD slices the incoming 44.1kHz audio down to 16kHz, feeds it frame-by-frame into a 4-state debounce machine, and silently drops the breath pauses. When the user actually stops talking, the VAD yields a command. The dispatcher checks the gate, grabs the accumulated audio buffer, and spins up a daemon thread.
Meanwhile, the main thread (the write_pump) is entirely free to just sit in a while True loop, pulling synthesized TTS audio chunks out of the queue and yielding them back to Go over gRPC the exact millisecond they are generated.
The client speaks, the system listens, thinks, and speaks back. True duplex. No walkie-talkies. No blocking. Just a clean, continuous stream of bytes.
So yeah, thanks for reading, you are cool, wierd and non-chalant who read my silly blog on a silly project, though it consumed most of my fat(i lost 5 kg lol), and time. I am still making still learning and most importantly still evolving. Each day just a 10% improvement and here I am, a naive API caller to a full architectural driven, Software developer making decision at the system level.
Thanks again. And if you are a Recruiter who has read this, read my resume here, and HIRE ME PLESAE🥺.