Voice AI agent open source stack: build a low-latency pipeli

You have tried the 10-minute tutorials and ended up with a robot that takes 3 seconds to respond, cannot be interrupted, and fails silently when the microphone gain drifts. Target end-to-end latency for a perceived instant response is 300-800ms, but most DIY builds hit 2-5 seconds. This guide shows you how to build a production-ready voice AI agent open source stack with streaming speech-to-text, interruptible responses, and graceful degradation to local models when APIs fail.

What is the best open source stack for a voice AI agent?

A voice AI agent open source stack combines wake word detection (Porcupine or openWakeWord), streaming speech recognition (faster-whisper), local LLM inference (Ollama or vLLM), and text-to-speech (Piper or Coqui). Frameworks like Pipecat and LiveKit Agents orchestrate these components with real-time streaming, interruption handling, and latency targeting. As of 2024, these modular stacks outperform monolithic solutions because you can optimize or replace individual stages [github.com/pipecat-ai].

Core components: VAD, STT, LLM, TTS, orchestration

A voice agent pipeline has five stages. Voice Activity Detection (VAD) determines when audio contains speech versus silence. Speech-to-Text (STT) converts audio to text. The Large Language Model (LLM) generates a response. Text-to-Speech (TTS) synthesizes audio. The orchestration layer manages state, handles interrupts, and coordinates timing.

When open source beats SaaS: compliance, latency, cost modeling

Open source wins when you need data sovereignty, predictable costs, or latency control. Voice SaaS APIs charge $0.01-0.06 per minute. Self-hosting becomes cost-effective above 500 minutes per day. Local inference also eliminates round-trip latency to cloud APIs, which is critical for real-time conversation.

How do you architect a low-latency voice AI pipeline?

Modern voice agents use streaming architectures with WebRTC or WebSocket connections, processing audio chunks in parallel rather than waiting for turn completion. A typical latency budget breaks down as 50ms for VAD, 100-300ms for STT, 200-500ms for LLM time-to-first-token, and 50-150ms for TTS. You must measure each stage to identify your bottleneck.

Streaming vs. turn-based architectures

Turn-based architectures wait for the user to stop speaking before processing. This adds latency because you cannot start STT until the utterance ends. Streaming architectures process audio in real-time chunks. The LLM can begin generating responses before the user finishes speaking, dramatically reducing perceived latency.

The 5-stage pipeline: capture to TTS

flowchart LR
    A[Capture] --> B[VAD]
    B --> C[STT]
    C --> D[LLM]
    D --> E[TTS]
    E -->|Audio Output| A

Audio capture pulls frames from the microphone at 16kHz or 48kHz. VAD filters out silence. STT streams partial transcripts. The LLM yields tokens as they generate. TTS streams audio back. Each stage runs in a separate thread or process.

Managing state and context across interruptions

When a user interrupts, you must truncate the LLM context. If the LLM was generating “The weather in San Francisco is” and the user says “stop”, the next turn should not include that partial output. Use a sliding window context manager that discards unfinished assistant responses.

Which open source tools form the best stack in 2024?

The dominant production stack combines Porcupine or openWakeWord for wake word, Whisper (faster-whisper) for STT, local LLMs via llama.cpp or Ollama for privacy, and Piper or Coqui TTS for synthesis [github.com/SYSTRAN]. Pipecat or LiveKit Agents provide the orchestration layer. Each tool has trade-offs between speed, accuracy, and resource usage.

Wake word: Porcupine vs. openWakeWord vs. custom

Porcupine (Picovoice) offers patented, reliable wake word detection with paid licensing for commercial use. openWakeWord is free and supports training custom wake words from 10-20 samples. Choose Porcupine for production reliability. Choose openWakeWord when you need customization or want to avoid licensing fees [github.com/dscripka].

STT: Whisper variants benchmarked

faster-whisper uses CTranslate2 to achieve a 4x speedup over original Whisper with identical accuracy. whisper.cpp runs efficiently on CPU. Distil-Whisper is 49% smaller and faster but slightly less accurate on rare words. For real-time agents, use faster-whisper on GPU or whisper.cpp on CPU. Use full Whisper for archival transcription [github.com/SYSTRAN].

Variant	Latency (RTF)	Accuracy	Hardware
Whisper large-v3	0.5-1.0	Highest	GPU
faster-whisper large-v3	0.1-0.2	Highest	GPU
Distil-Whisper	0.05-0.1	High	GPU/CPU
whisper.cpp tiny.en	0.3-0.5	Medium	CPU

RTF = Real-Time Factor. Lower is better. RTF 0.1 means 1 second of audio processes in 0.1 seconds.

LLM inference: Ollama, vLLM, and llama.cpp trade-offs

Ollama simplifies local LLM deployment with a single command. vLLM optimizes throughput for batched requests. llama.cpp offers the most control over quantization and memory. For voice agents, time-to-first-token matters more than throughput. Use Ollama for ease of use. Use llama.cpp when you need precise memory control [github.com/ollama].

TTS: Piper for speed, Coqui for quality, XTTS for voice cloning

Piper TTS synthesizes speech in 1ms on CPU but sounds robotic. Coqui TTS produces natural speech in 200-500ms but requires GPU. XTTS v2 supports voice cloning from a reference sample. Use Piper for low-latency utility responses. Use Coqui or XTTS for personas requiring natural intonation [github.com/rhasspy].

How do you handle interruptions and barge-in?

True conversational agents must detect when the user speaks during TTS playback, immediately halt audio output, and process the new utterance. This requires duplex audio streams, VAD on the input channel during output, and state management to reset the LLM context appropriately.

Implementing duplex audio with pyaudio or sounddevice

You must play audio while simultaneously recording. Use pyaudio or sounddevice with callback modes. Run input and output streams in separate threads. Queue audio chunks for output while continuously checking VAD on input.

import sounddevice as sd
import numpy as np
from queue import Queue

# Configuration
SAMPLE_RATE = 16000
CHANNELS = 1
DTYPE = np.int16

# Audio queues
input_queue = Queue()
output_queue = Queue()

def audio_callback(indata, outdata, frames, time, status):
    # Put input audio in queue for VAD processing
    input_queue.put(indata.copy())
    
    # Get output audio if available
    if not output_queue.empty():
        outdata[:] = output_queue.get()
    else:
        outdata.fill(0)

# Start duplex stream
stream = sd.RawStream(
    samplerate=SAMPLE_RATE,
    channels=CHANNELS,
    dtype=DTYPE,
    callback=audio_callback,
    blocksize=480  # 30ms at 16kHz
)

VAD-based barge-in detection patterns

Run VAD continuously on the input stream, even during TTS playback. When VAD detects speech above a threshold, stop TTS output immediately. Use Silero VAD for accuracy or WebRTC VAD for speed. Tune the threshold to your acoustic environment to avoid false triggers from ambient noise.

Context truncation strategies when interrupted mid-generation

When interrupted, the LLM may have generated partial output. Do not include this in the conversation history. Truncate the assistant message at the interruption point. Mark the turn as interrupted so the LLM understands context continuity.

Prerequisites

Install the required packages with pinned versions.

pip install faster-whisper==1.0.3
pip install pyaudio==0.2.14
pip install sounddevice==0.5.0
pip install numpy==1.26.4
pip install openai==1.51.0
pip install pvporcupine==3.0.2
pip install openwakeword==0.6.0
pip install piper-tts==1.2.0
pip install silero-vad==4.0

Hardware requirements:

• Microphone configured at 16kHz or 48kHz
• GPU recommended for faster-whisper (CUDA 11.8+)
• 8GB+ RAM for local LLM (16GB+ for larger models)
• Ollama or llama.cpp server running locally

Last verified: 2026-10-12

[VERIFY: Is Pipecat still actively maintained and at what version as of 2026-10-12?]

Step-by-step implementation

Build a minimal voice agent that listens for a wake word, transcribes speech, generates a response, and speaks it back.

1. Setting up the audio capture

Configure audio capture at 16kHz, which most speech models expect. Resample if your microphone only supports 48kHz.

import sounddevice as sd
import numpy as np
from collections import deque

class AudioCapture:
    def __init__(self, sample_rate=16000, channels=1):
        self.sample_rate = sample_rate
        self.channels = channels
        self.buffer = deque(maxlen=sample_rate * 10)  # 10 seconds buffer
        self.stream = None
    
    def start(self):
        self.stream = sd.InputStream(
            samplerate=self.sample_rate,
            channels=self.channels,
            dtype=np.float32,
            callback=self._callback
        )
        self.stream.start()
    
    def _callback(self, indata, frames, time, status):
        self.buffer.extend(indata[:, 0])
    
    def get_audio(self, duration_seconds):
        samples = int(self.sample_rate * duration_seconds)
        if len(self.buffer) < samples:
            return None
        return np.array(list(self.buffer)[-samples:])

2. Adding wake word detection

Use openWakeWord for a free, customizable wake word detector. Load the doorbell model or train your own.

import openwakeword
import numpy as np

class WakeWordDetector:
    def __init__(self, model_name="hey_jarvis"):
        self.model = openwakeword.Model()
        self.threshold = 0.5
    
    def detect(self, audio_chunk: np.ndarray) -> bool:
        # audio_chunk should be 16kHz float32 numpy array
        prediction = self.model.predict(audio_chunk)
        # Check if any model scored above threshold
        for model_name, score in prediction.items():
            if score > self.threshold:
                return True
        return False

# Usage
detector = WakeWordDetector()
if detector.detect(audio_chunk):
    print("Wake word detected!")

3. Streaming speech-to-text with faster-whisper

faster-whisper streams partial transcripts. Configure it to use GPU if available.

from faster_whisper import WhisperModel

class StreamingSTT:
    def __init__(self, model_size="base", device="cuda"):
        self.model = WhisperModel(model_size, device=device, compute_type="float16")
    
    def transcribe(self, audio: np.ndarray):
        # audio must be float32, 16kHz
        segments, info = self.model.transcribe(audio, beam_size=5)
        full_text = ""
        for segment in segments:
            full_text += segment.text
            yield segment.text  # Streaming partial results
        return full_text

# Usage
stt = StreamingSTT(model_size="base.en")
for partial in stt.transcribe(audio_chunk):
    print(f"Partial: {partial}")

4. LLM integration with Ollama

Connect to a local Ollama server. Use streaming to get time-to-first-token under 200ms.

import openai
import os

class LLMClient:
    def __init__(self, base_url="http://localhost:11434/v1", model="llama3.2"):
        self.client = openai.OpenAI(
            base_url=base_url,
            api_key=os.getenv("OLLAMA_API_KEY", "ollama")
        )
        self.model = model
        self.messages = []
    
    def add_message(self, role, content):
        self.messages.append({"role": role, "content": content})
    
    def generate(self, user_input: str):
        self.add_message("user", user_input)
        response = self.client.chat.completions.create(
            model=self.model,
            messages=self.messages,
            stream=True
        )
        full_response = ""
        for chunk in response:
            if chunk.choices[0].delta.content:
                text = chunk.choices[0].delta.content
                full_response += text
                yield text
        self.add_message("assistant", full_response)

5. Text-to-speech with Piper

Piper TTS runs fast on CPU. It outputs WAV audio you can play directly.

import subprocess
import tempfile
import sounddevice as sd
import soundfile as sf

class PiperTTS:
    def __init__(self, model_path="en_US-lessac-medium.onnx"):
        self.model_path = model_path
    
    def speak(self, text: str):
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=True) as f:
            subprocess.run(
                ["piper", "--model", self.model_path, "--output_file", f.name],
                input=text.encode(),
                check=True
            )
            data, samplerate = sf.read(f.name)
            sd.play(data, samplerate)
            sd.wait()

# Usage
tts = PiperTTS()
tts.speak("Hello, I am your voice assistant.")

6. Building the main orchestration loop

Combine all components into a single loop that manages state and handles interruptions.

import time
import threading

class VoiceAgent:
    def __init__(self):
        self.capture = AudioCapture()
        self.wake_word = WakeWordDetector()
        self.stt = StreamingSTT()
        self.llm = LLMClient()
        self.tts = PiperTTS()
        self.running = False
        self.speaking = False
    
    def start(self):
        self.capture.start()
        self.running = True
        self._run_loop()
    
    def _run_loop(self):
        print("Listening for wake word...")
        while self.running:
            # Get recent audio
            audio = self.capture.get_audio(duration_seconds=1.5)
            if audio is None:
                time.sleep(0.1)
                continue
            
            # Check for wake word
            if not self.speaking and self.wake_word.detect(audio):
                print("Wake word detected! Listening...")
                self._handle_turn()
    
    def _handle_turn(self):
        # Record user speech
        audio = self.capture.get_audio(duration_seconds=5)
        
        # Transcribe
        user_text = ""
        for partial in self.stt.transcribe(audio):
            user_text = partial
        print(f"User: {user_text}")
        
        # Generate response
        print("Generating response...")
        response_text = ""
        for chunk in self.llm.generate(user_text):
            response_text += chunk
        print(f"Assistant: {response_text}")
        
        # Speak response
        self.speaking = True
        self.tts.speak(response_text)
        self.speaking = False

# Run the agent
if __name__ == "__main__":
    agent = VoiceAgent()
    agent.start()

Trade-offs and when to choose another approach

This open source stack optimizes for flexibility and cost at scale. It is not the right choice for every situation.

Factor	Open Source Stack	SaaS APIs
Setup complexity	High (days)	Low (hours)
Latency control	Full control	Limited
Data privacy	On-premise	Cloud processing
Cost at low volume	Higher (GPU time)	Lower
Cost at high volume	Lower (fixed hardware)	Higher (per-minute)
Voice quality	Moderate to high	Highest

Choose open source when:

• You process over 500 minutes per day
• Data residency requires on-premise processing
• You need low-latency customization

Choose SaaS when:

• You need instant setup
• Quality matters more than cost
• You have sporadic usage patterns

Common errors and fixes

Real error messages and proven solutions.

1. Audio buffer underrun causes robotic speech

Error: ALSA lib pcm.c:8545:(snd_pcm_recover) underrun occurred

Cause: The audio callback is not providing data fast enough. The output queue empties and you get pops or clicks.

Fix: Reduce block size and pre-buffer audio. Use a ring buffer to maintain at least 100ms of audio ahead.

# Pre-fill queue before playing
for _ in range(3):
    output_queue.put(silence_chunk)

2. Sample rate mismatch causes pitch distortion

Error: Audio sounds too fast or slow, or has clicking artifacts.

Cause: Your microphone captures at 48kHz but Whisper expects 16kHz. Resampling is incorrect or missing.

Fix: Explicitly resample using a high-quality resampler.

import librosa

audio_16k = librosa.resample(audio_48k, orig_sr=48000, target_sr=16000)

3. VAD false positives from keyboard typing

Error: Agent activates randomly without speech.

Cause: Silero VAD is sensitive to percussive sounds like keyboard clicks.

Fix: Increase the VAD threshold or add a frequency filter. Use a bandpass filter to reject low-frequency thumps.

from scipy.signal import butter, lfilter

def bandpass_filter(audio, lowcut=300, highcut=3400, fs=16000, order=5):
    nyq = 0.5 * fs
    low = lowcut / nyq
    high = highcut / nyq
    b, a = butter(order, [low, high], btype='band')
    return lfilter(b, a, audio)

filtered_audio = bandpass_filter(raw_audio)

4. LLM streaming stalls without timeout

Error: Agent hangs silently after user speaks.

Cause: The LLM API call hangs. No timeout handler exists to detect and recover.

Fix: Wrap the generation in a timeout using asyncio.wait_for or a thread timeout.

import asyncio

async def generate_with_timeout(llm_client, prompt, timeout=10.0):
    try:
        return await asyncio.wait_for(