Voice AI agent barge-in interruption handling is what separates a conversational system from a recorded phone menu. When a user talks over your agent, the agent must detect the overlap, stop speaking, and pivot to the new input within a few hundred milliseconds. This guide gives you the architecture and runnable code to build that loop.
Step 1: Capture full-duplex audio and run parallel VAD
A voice agent needs to listen while it talks. Use a bidirectional stream: microphone frames go to your VAD/STT pipeline, speaker frames come from your TTS buffer. In a browser client, getUserMedia plus an AudioWorklet gives you mic frames; the speaker plays an AudioBufferSourceNode. On the server, terminate WebRTC or SIP and read RTP.
Run a voice activity detector on the inbound stream continuously, even during playback. webrtcvad is a solid, low-latency choice:
import webrtcvad
import asyncio
vad = webrtcvad.Vad(3) # aggressiveness 0-3
def frame_is_speech(pcm_frame: bytes, sample_rate: int = 16000) -> bool:
# pcm_frame must be 16-bit mono, 10/20/30 ms
return vad.is_speech(pcm_frame, sample_rate)
async def monitor_mic(reader: asyncio.StreamReader):
while True:
frame = await reader.read(320) # 20ms @ 16kHz, 16-bit
if frame_is_speech(frame):
asyncio.create_task(handle_potential_barge_in())The key is that monitor_mic never blocks on TTS. It runs as a separate task. Run VAD at 16kHz, 16-bit mono. Most STT engines resample anyway, but doing it once at the edge saves CPU. If you run VAD in the browser, you avoid shipping raw audio for detection; ship only flagged segments. Server-side VAD gives you one place to tune aggressiveness across all clients.
Step 2: Detect barge-in with a threshold and debounce
Raw VAD flags are noisy. A cough or background noise shouldn’t cut off the agent. Require sustained speech for a short window before triggering interruption. A pure energy threshold breaks in noisy environments; webrtcvad uses a Gaussian mixture model but still needs debounce.
from collections import deque
class BargeInDetector:
def __init__(self, threshold_ms=200, frame_ms=20):
self.window = deque(maxlen=threshold_ms // frame_ms)
def push(self, is_speech: bool) -> bool:
self.window.append(is_speech)
# trigger if majority of window is speech
return sum(self.window) >= len(self.window) * 0.6When push returns True, fire the interruption. The 200ms window is a starting point; measure with real calls. If you cut too early, the agent stutters; too late, the user feels ignored. Keep the detector running after cancellation so you can distinguish a quick interjection from the user taking over.
Step 3: Cancel ongoing TTS and flush playback buffers
Once barge-in is confirmed, stop writing to the speaker immediately. If you stream TTS tokens to a synth, cancel the request. If you already have PCM in a playback queue, drop it. Prefer streaming synthesis where you control the socket; a managed TTS that returns a single URL can’t be cancelled mid-file.
class PlaybackController:
def __init__(self):
self._play_task = None
self.cancelled = False
self.current_turn_id = 0
def start(self, pcm_chunks, turn_id):
self.cancelled = False
self.current_turn_id = turn_id
self._play_task = asyncio.create_task(self._play(pcm_chunks))
async def _play(self, pcm_chunks):
for chunk in pcm_chunks:
if self.cancelled:
return
await write_to_speaker(chunk)
def barge_in(self):
self.cancelled = True
def play_chunk(chunk, turn_id):
if turn_id != playback.current_turn_id:
return # stale audio from a cancelled turn
speaker.write(chunk)Do not wait for the current sentence to finish. Users perceive any continuation after they start speaking as a failure. Flush the OS-level audio buffer if your API exposes it (e.g., AudioContext.suspend() in browser).
Step 4: Manage conversation state with an explicit machine
Voice AI agent barge-in interruption handling requires a state machine, not a pile of flags. Define states: LISTENING, SPEAKING, INTERRUPTED, THINKING.
from enum import Enum, auto
class AgentState(Enum):
LISTENING = auto()
SPEAKING = auto()
INTERRUPTED = auto()
THINKING = auto()
state = AgentState.LISTENING
def on_barge_in():
global state
if state == AgentState.SPEAKING:
playback.barge_in()
state = AgentState.INTERRUPTED
stt.force_final() # grab partial transcript
def on_user_silence():
global state
if state == AgentState.INTERRUPTED:
state = AgentState.THINKINGThe INTERRUPTED state lets you skip normal end-of-turn endpointing. The force_final call on STT should return the best partial. Whisper-streaming or vendor equivalents expose partial and final events; treat the partial at interruption as final for context building.
Step 5: Use the partial transcript to drive the next action
Streaming STT emits partials every few hundred milliseconds. On barge-in, capture the latest partial and discard the agent’s planned response. Then decide: acknowledge, or answer the new query. Do not send the agent’s half-spoken sentence to the LLM; it’s garbage. Send only user turns and a system note: “User interrupted agent”.
If you need an LLM to interpret the interruption, call your dialogue model with the truncated context. For example, route through an OpenAI-compatible endpoint that honors fallback so latency stays low:
import httpx
async def get_reply(interrupted_text: str, history: list):
# n4n.ai exposes one OpenAI-compatible endpoint across 240+ models
# with automatic fallback when a provider is rate-limited.
async with httpx.AsyncClient() as client:
r = await client.post(
"https://api.n4n.ai/v1/chat/completions",
json={
"model": "auto",
"messages": history + [{"role": "user", "content": interrupted_text}],
"stream": False,
},
headers={"Authorization": "Bearer $KEY"}
)
return r.json()["choices"][0]["message"]["content"]Keep the prompt tight. The user interrupted for a reason; a 2-second LLM think time is acceptable, but 10 seconds is not.
Step 6: Tune endpointing after the interruption
After barge-in, the user may stop speaking quickly or keep talking. Standard endpointing is 300–500ms silence. Post-barge-in, raise the silence timeout to 800ms because people often say “no—I mean yes” with pauses. Lower it again after the agent speaks to keep responsiveness.
SILENCE_TIMEOUT = {
AgentState.SPEAKING: 0.4,
AgentState.INTERRUPTED: 0.8,
AgentState.LISTENING: 0.5,
}When the timeout fires in INTERRUPTED, move to THINKING and call the LLM. When the LLM responds, return to SPEAKING with a new turn id.
Step 7: Verify with a scripted overlap test
You cannot verify barge-in by clicking a button. Simulate overlapping audio. Record an agent prompt WAV, then inject a user phrase 300ms after playback starts. You can generate the mix with ffmpeg and a dummy sound device.
async def test_barge_in():
agent_say("Please hold while I look that up.")
await asyncio.sleep(0.3)
inject_mic("No, cancel that!")
await asyncio.sleep(0.1)
assert playback.cancelled, "TTS was not interrupted"
assert state == AgentState.INTERRUPTED
# measure latency from injection to cancellation
assert playback.cancel_latency_ms < 150Run this in CI with a fake audio device (pytest-asyncio + sounddevice dummy). If the assertion fails, your threshold or cancellation is too slow. The 150ms budget is the max acceptable barge-in latency for a natural feel.
Common pitfalls
- Echo leakage: If speaker audio leaks into the mic, VAD triggers on the agent’s own voice. Use acoustic echo cancellation (AEC) before VAD.
- Late TTS chunks: Some TTS providers send audio 500ms after you cancel. Drop anything tagged with a later request id using the turn-id guard shown above.
- State races: Two VAD frames could fire
on_barge_intwice. Guard transitions withif state == AgentState.SPEAKING. - Over-eager cancellation: Setting VAD aggressiveness to 3 in a noisy cafe will clip user responses. Start at 2 and collect real traces.
Voice AI agent barge-in interruption handling is fundamentally a real-time systems problem dressed as an ML feature. Get the audio pipeline and state machine right, and the model quality matters less.