Open-Source Real-Time Conversational Voice Chatbot Building Guide: Barge-In Architecture and Implementation

• Introduction
• Comparing and Choosing the Open-Source Stack
  • VAD (Voice Activity Detection) Comparison
  • Wake Word Options
  • STT (Speech-to-Text) Comparison
  • Comparing Local LLM Serving
  • TTS (Text-to-Speech) Comparison
  • Comparing Real-Time Audio Transport
• Architecture Design
  • The Full Pipeline Architecture
  • Core Design Principles
• State Machine Design
  • State Transition Rules
  • The Order of Events When Barge-In Happens
• A Minimal Working Python Example
  • Installing the Dependencies
  • The Core Code
  • Key Points in the Code
• Error Handling and Queue Design
  • Retry Strategy
  • Queue Management Patterns
  • Error Recovery Matrix
• A Practical Configuration Guide
  • Standalone Local vs Server-Client
    • Standalone Local (All-in-One)
    • Server-Client
    • Hybrid (Recommended)
  • Optimization Tips for Environments Without a GPU
  • Tips for Improving Korean Quality
• Latency Benchmark Targets
  • Latency Targets per Stage
  • Achievable Latency by Environment
• Implementation Checklist
  • Preparing the Environment
  • Downloading the Models
  • Verifying the Pipeline
  • Production Deployment
• Troubleshooting Guide
• Advanced Topic: Going to Production with LiveKit Agents
• Additional Open-Source Frameworks
• Conclusion
• References

Introduction

Anyone who has used a voice AI assistant has felt it at least once: the AI is answering at length, and the frustration of having to hear it out before you can say anything. In conversation between people you can cut in while the other person is speaking (barge-in), and they naturally stop and listen to you. Implementing barge-in in a real-time voice chatbot is not a matter of "can't we just stop the TTS?" You have to control the whole pipeline: cancel the echo in a full-duplex environment where microphone input and speaker output run at the same time, detect user speech with VAD (Voice Activity Detection), cancel the LLM generation and TTS synthesis already in flight, and handle the new user input.

This article designs a real-time voice chatbot architecture you can build 100% from open source, and covers how to implement the core features, barge-in included, in Python. The goal is a system that runs on a single local GPU (or CPU only) with no commercial API.

Comparing and Choosing the Open-Source Stack

VAD (Voice Activity Detection) Comparison

VAD, which decides whether the user is speaking or silent, is the core trigger for barge-in. In a real-time system the decision has to be complete within 30ms.

Recommendation: for an open-source build, Silero VAD is the best on accuracy and licensing. webrtcvad shows too many false negatives (dropping half an utterance), and Picovoice Cobra is accurate but commercially licensed.

Wake Word Options

If you want to wake the chatbot with a specific phrase instead of leaving the microphone on all the time, you need wake word detection.

Recommendation: for a fully open-source build, openWakeWord + Silero VAD. If commercial components are acceptable, Porcupine leads on accuracy.

STT (Speech-to-Text) Comparison

Recommendation: on a GPU, faster-whisper (large-v3 or medium) is best on both accuracy and speed. In a lightweight CPU-only environment, Vosk has the edge thanks to native real-time streaming support.

Comparing Local LLM Serving

Recommendation: for prototypes and single-user setups, Ollama is easiest to install and run. With 5+ concurrent connections, vLLM wins on throughput and latency predictability. Both offer an OpenAI-compatible API, so you can swap them with no code change.

TTS (Text-to-Speech) Comparison

Recommendation: for a barge-in system, Piper is the best choice. It reaches 50x real-time speed on CPU alone, so TTS latency is negligible, it supports streaming synthesis, and it is easy to stop instantly. If voice quality matters more, use XTTS v2 - but since Coqui AI shut down (2025.12) you will need to use the community fork (coqui-tts).

Comparing Real-Time Audio Transport

Recommendation: for a standalone local run, WebSocket is enough. If you need a browser client or network quality varies, LiveKit gives you echo cancellation, NAT traversal, and SFU scaling all at once. LiveKit's Agents framework has the STT/LLM/TTS pipeline and barge-in turn detection built in, which helps in production.

Architecture Design

The Full Pipeline Architecture

┌─────────────────────────────────────────────────────────────┐
│                        Client Device                         │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐               │
│  │   Mic    │───▶│  Audio   │───▶│ WebSocket│──── network ──┤
│  │  Input   │    │  Capture │    │  Client  │               │
│  └──────────┘    └──────────┘    └──────────┘               │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐               │
│  │ Speaker  │◀───│  Audio   │◀───│ WebSocket│◀── network ──┤
│  │  Output  │    │ Playback │    │  Client  │               │
│  └──────────┘    └──────────┘    └──────────┘               │
└─────────────────────────────────────────────────────────────┘
                           │ ▲
                     audio │ │ audio
                     bytes │ │ bytes
                           ▼ │
┌─────────────────────────────────────────────────────────────┐
│                     Voice Pipeline Server                    │
│                                                             │
│  ┌───────────┐   ┌───────────┐   ┌───────────┐             │
│  │ Silero    │──▶│ faster-   │──▶│  Ollama   │             │
│  │ VAD       │   │ whisper   │   │  (LLM)    │             │
│  │           │   │ (STT)     │   │           │             │
│  └───────────┘   └───────────┘   └─────┬─────┘             │
│       │                                 │                   │
│       │  barge-in                token   │                   │
│       │  signal                 stream  │                   │
│       ▼                                 ▼                   │
│  ┌───────────┐               ┌───────────┐                  │
│  │  State    │◀─────────────▶│  Piper    │                  │
│  │  Machine  │   cancel/     │  (TTS)    │                  │
│  │           │   resume      │           │                  │
│  └───────────┘               └───────────┘                  │
│                                    │                        │
│                              audio │ chunks                 │
│                                    ▼                        │
│                             ┌───────────┐                   │
│                             │  Audio    │                   │
│                             │  Output   │                   │
│                             │  Queue    │                   │
│                             └───────────┘                   │
└─────────────────────────────────────────────────────────────┘

Core Design Principles

1. Asynchronous pipeline: every component runs on asyncio and is connected by asyncio.Queue
2. Cancellable at any moment: LLM streaming and TTS synthesis are managed as an asyncio.Task so a .cancel() call stops them immediately
3. Full-Duplex: microphone input and speaker output run at the same time, and echo cancellation (AEC) filters out the system's own output
4. State-driven control: a state machine decides the flow of the whole pipeline

State Machine Design

Implementing barge-in reliably requires clear state transitions. We design it with the following 5 states.

                    ┌──────────────────────────────────┐
                    │                                  │
                    ▼                                  │
              ┌──────────┐                             │
              │          │  voice_detected             │
         ┌───▶│   IDLE   │─────────────────┐           │
         │    │          │                 │           │
         │    └──────────┘                 ▼           │
         │                          ┌──────────┐       │
         │                          │LISTENING │       │
         │                          │          │       │
         │                          └────┬─────┘       │
         │                               │             │
         │                    silence_   │             │
         │                    detected   │             │
         │                               ▼             │
         │                          ┌──────────┐       │
         │         timeout/         │THINKING  │       │
         │         error            │          │       │
         │           ┌──────────────┴────┬─────┘       │
         │           │                   │             │
         │           │        first_     │             │
         │           │        audio_     │             │
         │           │        chunk      │             │
         │           ▼                   ▼             │
         │    ┌──────────┐         ┌──────────┐        │
         │    │  ERROR   │         │ SPEAKING │        │
         │    │          │         │          │───┐    │
         │    └──────────┘         └────┬─────┘   │    │
         │                              │         │    │
         │                   voice_     │   barge_│    │
         │                   end        │   in    │    │
         │                              │         │    │
         │                              ▼         ▼    │
         │                                  ┌──────────┐
         └──────────────────────────────────│INTERRUPT │
                                            │ED        │
                                            └──────────┘

State Transition Rules

from enum import Enum, auto

class State(Enum):
    IDLE = auto()        # waiting (for a wake word or detected speech)
    LISTENING = auto()   # recording the user's speech
    THINKING = auto()    # STT → LLM processing
    SPEAKING = auto()    # playing back TTS audio
    INTERRUPTED = auto() # barge-in occurred, clean up then move to LISTENING

TRANSITIONS = {
    State.IDLE:        {Event.VOICE_DETECTED: State.LISTENING},
    State.LISTENING:   {Event.SILENCE_DETECTED: State.THINKING,
                        Event.TIMEOUT: State.IDLE},
    State.THINKING:    {Event.FIRST_AUDIO_CHUNK: State.SPEAKING,
                        Event.ERROR: State.IDLE,
                        Event.BARGE_IN: State.INTERRUPTED},
    State.SPEAKING:    {Event.PLAYBACK_DONE: State.IDLE,
                        Event.BARGE_IN: State.INTERRUPTED},
    State.INTERRUPTED: {Event.CLEANUP_DONE: State.LISTENING},
}

The Order of Events When Barge-In Happens

1. VAD detects user speech during SPEAKING/THINKING
2. The state moves to INTERRUPTED
3. Immediately: clear the TTS audio output queue and stop speaker playback
4. Immediately: cancel the LLM streaming Task (task.cancel())
5. Immediately: cancel the TTS synthesis Task
6. Preserve the partial response in history (keeping context)
7. Emit the CLEANUP_DONE event → move to LISTENING
8. Start recording the new user utterance

A Minimal Working Python Example

Below is minimal working code implementing the full pipeline - microphone input → VAD → STT → LLM → TTS → speaker output - with barge-in support.

Installing the Dependencies

pip install silero-vad faster-whisper openai-whisper piper-tts \
            sounddevice numpy httpx asyncio
# Install Ollama (macOS/Linux)
curl -fsSL https://ollama.com/install.sh | sh
ollama pull gemma3:4b  # or whichever model you want

The Core Code

import asyncio
import numpy as np
import sounddevice as sd
import httpx
import io
import wave
from enum import Enum, auto
from collections import deque
from dataclasses import dataclass, field
from typing import Optional

# ──────────────────────────────────────────────
# 1. State machine definition
# ──────────────────────────────────────────────
class State(Enum):
    IDLE = auto()
    LISTENING = auto()
    THINKING = auto()
    SPEAKING = auto()
    INTERRUPTED = auto()

class Event(Enum):
    VOICE_DETECTED = auto()
    SILENCE_DETECTED = auto()
    FIRST_AUDIO_CHUNK = auto()
    PLAYBACK_DONE = auto()
    BARGE_IN = auto()
    CLEANUP_DONE = auto()
    TIMEOUT = auto()
    ERROR = auto()

TRANSITIONS = {
    State.IDLE:        {Event.VOICE_DETECTED: State.LISTENING},
    State.LISTENING:   {Event.SILENCE_DETECTED: State.THINKING,
                        Event.TIMEOUT: State.IDLE},
    State.THINKING:    {Event.FIRST_AUDIO_CHUNK: State.SPEAKING,
                        Event.ERROR: State.IDLE,
                        Event.BARGE_IN: State.INTERRUPTED},
    State.SPEAKING:    {Event.PLAYBACK_DONE: State.IDLE,
                        Event.BARGE_IN: State.INTERRUPTED},
    State.INTERRUPTED: {Event.CLEANUP_DONE: State.LISTENING},
}

@dataclass
class PipelineContext:
    state: State = State.IDLE
    audio_buffer: bytearray = field(default_factory=bytearray)
    conversation: list = field(default_factory=list)
    llm_task: Optional[asyncio.Task] = None
    tts_task: Optional[asyncio.Task] = None
    playback_queue: asyncio.Queue = field(default_factory=asyncio.Queue)
    interrupt_event: asyncio.Event = field(default_factory=asyncio.Event)

    def transition(self, event: Event) -> bool:
        allowed = TRANSITIONS.get(self.state, {})
        if event in allowed:
            old = self.state
            self.state = allowed[event]
            print(f"[FSM] {old.name} --{event.name}--> {self.state.name}")
            return True
        print(f"[FSM] {self.state.name}: {event.name} ignored")
        return False

# ──────────────────────────────────────────────
# 2. VAD module (Silero VAD)
# ──────────────────────────────────────────────
import torch

class VoiceActivityDetector:
    def __init__(self, threshold: float = 0.5):
        self.model, self.utils = torch.hub.load(
            'snakers4/silero-vad', 'silero_vad', onnx=True
        )
        self.threshold = threshold
        self.sample_rate = 16000
        self._silence_frames = 0
        self.silence_limit = 30  # 30 frames × 30ms = 900ms

    def process_chunk(self, audio_chunk: np.ndarray) -> str:
        """Takes a 30ms audio chunk and returns 'speech'/'silence'/'end'."""
        tensor = torch.from_numpy(audio_chunk).float()
        prob = self.model(tensor, self.sample_rate).item()

        if prob >= self.threshold:
            self._silence_frames = 0
            return "speech"
        else:
            self._silence_frames += 1
            if self._silence_frames >= self.silence_limit:
                self._silence_frames = 0
                return "end"
            return "silence"

    def reset(self):
        self.model.reset_states()
        self._silence_frames = 0

# ──────────────────────────────────────────────
# 3. STT module (faster-whisper)
# ──────────────────────────────────────────────
from faster_whisper import WhisperModel

class SpeechToText:
    def __init__(self, model_size: str = "medium", device: str = "cuda"):
        self.model = WhisperModel(
            model_size, device=device, compute_type="float16"
        )

    def transcribe(self, audio: np.ndarray) -> str:
        segments, _ = self.model.transcribe(
            audio, language="ko", beam_size=5,
            vad_filter=True
        )
        return " ".join(seg.text for seg in segments).strip()

# ──────────────────────────────────────────────
# 4. LLM module (Ollama OpenAI-compatible API)
# ──────────────────────────────────────────────
async def stream_llm_response(
    messages: list[dict],
    model: str = "gemma3:4b",
    base_url: str = "http://localhost:11434/v1",
) -> asyncio.AsyncGenerator:
    """Streaming request to Ollama, yielding token by token."""
    async with httpx.AsyncClient(timeout=30.0) as client:
        async with client.stream(
            "POST",
            f"{base_url}/chat/completions",
            json={
                "model": model,
                "messages": messages,
                "stream": True,
                "max_tokens": 512,
            },
        ) as resp:
            async for line in resp.aiter_lines():
                if not line.startswith("data: "):
                    continue
                data = line[6:]
                if data == "[DONE]":
                    break
                import json
                chunk = json.loads(data)
                delta = chunk["choices"][0].get("delta", {})
                if content := delta.get("content"):
                    yield content

# ──────────────────────────────────────────────
# 5. TTS module (Piper)
# ──────────────────────────────────────────────
import subprocess

class TextToSpeech:
    def __init__(self, model_path: str, config_path: str):
        self.model_path = model_path
        self.config_path = config_path

    async def synthesize(self, text: str) -> bytes:
        """Converts text into WAV bytes."""
        proc = await asyncio.create_subprocess_exec(
            "piper",
            "--model", self.model_path,
            "--config", self.config_path,
            "--output-raw",
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, _ = await proc.communicate(text.encode("utf-8"))
        return stdout

    async def synthesize_streaming(
        self, text_chunks: asyncio.Queue, audio_queue: asyncio.Queue
    ):
        """Synthesizes sentence by sentence and pushes into the audio queue."""
        buffer = ""
        sentence_delimiters = {".", "!", "?", "。", "!", "?", "\n"}

        while True:
            chunk = await text_chunks.get()
            if chunk is None:  # end signal
                if buffer.strip():
                    audio = await self.synthesize(buffer)
                    await audio_queue.put(audio)
                await audio_queue.put(None)  # playback end signal
                break

            buffer += chunk
            # Synthesize as soon as a sentence is complete
            for delim in sentence_delimiters:
                if delim in buffer:
                    parts = buffer.split(delim, 1)
                    sentence = parts[0] + delim
                    buffer = parts[1] if len(parts) > 1 else ""
                    if sentence.strip():
                        audio = await self.synthesize(sentence.strip())
                        await audio_queue.put(audio)
                    break

# ──────────────────────────────────────────────
# 6. Audio playback (barge-in aware)
# ──────────────────────────────────────────────
async def play_audio(ctx: PipelineContext, sample_rate: int = 22050):
    """Pulls chunks from the audio queue and plays them. Stops instantly on interrupt."""
    while True:
        audio_bytes = await ctx.playback_queue.get()
        if audio_bytes is None:
            ctx.transition(Event.PLAYBACK_DONE)
            break

        audio_array = np.frombuffer(audio_bytes, dtype=np.int16)
        chunk_size = sample_rate // 10  # play in 100ms units

        for i in range(0, len(audio_array), chunk_size):
            if ctx.interrupt_event.is_set():
                # barge-in! stop playback immediately
                # drain the queue
                while not ctx.playback_queue.empty():
                    try:
                        ctx.playback_queue.get_nowait()
                    except asyncio.QueueEmpty:
                        break
                return

            segment = audio_array[i:i + chunk_size]
            sd.play(segment, samplerate=sample_rate, blocking=True)

# ──────────────────────────────────────────────
# 7. Main pipeline (with barge-in support)
# ──────────────────────────────────────────────
async def run_pipeline():
    ctx = PipelineContext()
    vad = VoiceActivityDetector(threshold=0.5)
    stt = SpeechToText(model_size="medium", device="cuda")
    tts = TextToSpeech(
        model_path="ko_KR-kss-medium.onnx",
        config_path="ko_KR-kss-medium.onnx.json",
    )

    SAMPLE_RATE = 16000
    CHUNK_MS = 30
    CHUNK_SAMPLES = int(SAMPLE_RATE * CHUNK_MS / 1000)

    print("[Pipeline] 시작. 말씀하세요...")

    while True:
        ctx.interrupt_event.clear()

        # ── IDLE → LISTENING ──
        if ctx.state == State.IDLE:
            vad.reset()
            ctx.audio_buffer = bytearray()

            # Read a 30ms chunk from the microphone
            audio_chunk = sd.rec(
                CHUNK_SAMPLES, samplerate=SAMPLE_RATE,
                channels=1, dtype="float32"
            )
            sd.wait()
            audio_np = audio_chunk.flatten()

            result = vad.process_chunk(audio_np)
            if result == "speech":
                ctx.transition(Event.VOICE_DETECTED)
                ctx.audio_buffer.extend(
                    (audio_np * 32767).astype(np.int16).tobytes()
                )

        # ── LISTENING ──
        elif ctx.state == State.LISTENING:
            audio_chunk = sd.rec(
                CHUNK_SAMPLES, samplerate=SAMPLE_RATE,
                channels=1, dtype="float32"
            )
            sd.wait()
            audio_np = audio_chunk.flatten()
            ctx.audio_buffer.extend(
                (audio_np * 32767).astype(np.int16).tobytes()
            )

            result = vad.process_chunk(audio_np)
            if result == "end":
                ctx.transition(Event.SILENCE_DETECTED)

        # ── THINKING ──
        elif ctx.state == State.THINKING:
            # STT conversion
            audio_data = np.frombuffer(
                ctx.audio_buffer, dtype=np.int16
            ).astype(np.float32) / 32767.0
            user_text = stt.transcribe(audio_data)
            print(f"[User] {user_text}")

            if not user_text.strip():
                ctx.transition(Event.ERROR)
                continue

            ctx.conversation.append({"role": "user", "content": user_text})

            # LLM streaming + TTS at the same time
            text_queue: asyncio.Queue = asyncio.Queue()
            full_response = []

            async def llm_to_tts():
                try:
                    async for token in stream_llm_response(ctx.conversation):
                        if ctx.interrupt_event.is_set():
                            break
                        full_response.append(token)
                        await text_queue.put(token)
                    await text_queue.put(None)
                except asyncio.CancelledError:
                    await text_queue.put(None)

            async def tts_worker():
                await tts.synthesize_streaming(text_queue, ctx.playback_queue)

            async def vad_monitor():
                """Detects barge-in during THINKING/SPEAKING."""
                while ctx.state in (State.THINKING, State.SPEAKING):
                    chunk = sd.rec(
                        CHUNK_SAMPLES, samplerate=SAMPLE_RATE,
                        channels=1, dtype="float32"
                    )
                    sd.wait()
                    r = vad.process_chunk(chunk.flatten())
                    if r == "speech" and ctx.state == State.SPEAKING:
                        print("[Barge-In] User speech detected! Stopping the response")
                        ctx.interrupt_event.set()
                        ctx.transition(Event.BARGE_IN)
                        return
                    await asyncio.sleep(0.01)

            # Run the tasks
            ctx.llm_task = asyncio.create_task(llm_to_tts())
            ctx.tts_task = asyncio.create_task(tts_worker())
            playback_task = asyncio.create_task(play_audio(ctx))
            monitor_task = asyncio.create_task(vad_monitor())

            ctx.transition(Event.FIRST_AUDIO_CHUNK)

            # Wait for completion or interrupt
            done, pending = await asyncio.wait(
                [ctx.llm_task, ctx.tts_task, playback_task, monitor_task],
                return_when=asyncio.FIRST_COMPLETED,
            )

            # Clean up when barge-in happens
            if ctx.interrupt_event.is_set():
                for task in pending:
                    task.cancel()
                    try:
                        await task
                    except asyncio.CancelledError:
                        pass

                # Preserve the partial response
                partial = "".join(full_response)
                if partial:
                    ctx.conversation.append(
                        {"role": "assistant", "content": partial + " [중단됨]"}
                    )
                ctx.transition(Event.CLEANUP_DONE)
                ctx.audio_buffer = bytearray()
                vad.reset()
            else:
                # Normal completion
                for task in pending:
                    await task
                response_text = "".join(full_response)
                ctx.conversation.append(
                    {"role": "assistant", "content": response_text}
                )
                print(f"[Assistant] {response_text}")

        # ── INTERRUPTED → LISTENING ──
        elif ctx.state == State.INTERRUPTED:
            ctx.transition(Event.CLEANUP_DONE)

        await asyncio.sleep(0.001)

if __name__ == "__main__":
    asyncio.run(run_pipeline())

Key Points in the Code

1. interrupt_event: propagates the barge-in signal as an asyncio.Event. Every worker checks this event
2. asyncio.wait(return_when=FIRST_COMPLETED): when the VAD monitor detects barge-in, the remaining tasks are cancelled immediately
3. Preserving the partial response: an interrupted response is also kept in the conversation history, tagged [중단됨] (interrupted), so context is not lost
4. Sentence-level TTS: while the LLM streams tokens, TTS synthesis starts the moment a sentence delimiter (., ?, ! and so on) appears

Error Handling and Queue Design

Retry Strategy

import asyncio
from dataclasses import dataclass

@dataclass
class RetryConfig:
    max_retries: int = 3
    base_delay: float = 0.5
    max_delay: float = 5.0

async def with_retry(coro_fn, config: RetryConfig = RetryConfig()):
    """Exponential backoff retry wrapper."""
    for attempt in range(config.max_retries):
        try:
            return await coro_fn()
        except Exception as e:
            if attempt == config.max_retries - 1:
                raise
            delay = min(
                config.base_delay * (2 ** attempt),
                config.max_delay
            )
            print(f"[Retry] {attempt+1}/{config.max_retries}: {e}, "
                  f"retrying after {delay}s")
            await asyncio.sleep(delay)

Queue Management Patterns

class AudioPipelineQueues:
    """Pipeline queue management. Each stage is connected by an async queue."""

    def __init__(self, maxsize: int = 100):
        self.vad_to_stt: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
        self.stt_to_llm: asyncio.Queue = asyncio.Queue(maxsize=10)
        self.llm_to_tts: asyncio.Queue = asyncio.Queue(maxsize=50)
        self.tts_to_speaker: asyncio.Queue = asyncio.Queue(maxsize=20)

    async def flush_all(self):
        """Drains every queue on barge-in."""
        for q in [self.vad_to_stt, self.stt_to_llm,
                   self.llm_to_tts, self.tts_to_speaker]:
            while not q.empty():
                try:
                    q.get_nowait()
                except asyncio.QueueEmpty:
                    break

    async def flush_downstream(self):
        """Drains only the queues after STT (new input is preserved on barge-in)."""
        for q in [self.llm_to_tts, self.tts_to_speaker]:
            while not q.empty():
                try:
                    q.get_nowait()
                except asyncio.QueueEmpty:
                    break

Error Recovery Matrix

A Practical Configuration Guide

Standalone Local vs Server-Client

Standalone Local (All-in-One)

┌──────────────────────────────────┐
│          Single Machine          │
│  Mic → VAD → STT → LLM → TTS   │
│         → Speaker               │
│                                  │
│  GPU: RTX 3060+ (12GB VRAM)     │
│  or CPU-only (Vosk + Piper)     │
└──────────────────────────────────┘

Advantages: 0ms network latency, perfect privacy, no internet required Drawbacks: hardware constraints, limits on model size

Server-Client

┌──────────┐     WebSocket/     ┌────────────────┐
│  Client  │◀────WebRTC────────▶│   GPU Server    │
│ (Raspberry Pi│    audio stream   │ STT + LLM + TTS │
│  / browser)  │                  │                │
└──────────┘                    └────────────────┘

Advantages: a lightweight client, the ability to use powerful models, multiple clients Drawbacks: added network latency (20~100ms), server cost

Hybrid (Recommended)

┌──────────────────┐          ┌────────────────┐
│     Client       │  text    │   GPU Server   │
│  Mic → VAD → STT │─────────▶│  LLM (vLLM)   │
│  (Vosk, local)   │  text    │               │
│  Speaker ← TTS   │◀─────────│               │
│  (Piper, local)  │          └────────────────┘
└──────────────────┘

VAD/STT/TTS run locally (minimizing latency) and only the LLM runs on the server. Only text crosses the network, so the bandwidth burden is minimal.

Optimization Tips for Environments Without a GPU

You can run a real-time voice chatbot even without a GPU (on a Raspberry Pi, an older laptop, and so on).

Extra optimization tips:

1. Use ONNX Runtime: both Silero VAD and Piper support ONNX models. Optimize CPU inference with onnxruntime
2. INT8 quantization: faster-whisper's compute_type="int8" improves CPU performance by 2~3x
3. Adjust the audio chunk size: raising it from 30ms to 60ms halves the number of VAD calls (an accuracy trade-off)
4. LLM quantization: q4_0 or q4_K_M quantization cuts RAM usage by 60%
5. Avoid batching: on CPU, prefer streaming. Fix the batch size at 1

Tips for Improving Korean Quality

Korean speech recognition and synthesis need extra optimization compared with English.

STT Korean optimization:

• With faster-whisper, state language="ko" explicitly (a 3~5% WER improvement over auto-detection)
• Setting initial_prompt="음성 인식 결과입니다." supplies a Korean context hint
• beam_size=5 or higher is recommended (Korean has many homophones, so beam search helps a lot)
• Enable the VAD filter (vad_filter=True) to strip silent stretches

TTS Korean optimization:

• Piper Korean model: ko_KR-kss-medium (based on the KSS dataset)
• XTTS v2 supports Korean zero-shot, but fine-tuning improves naturalness substantially
• Preprocess mixed numeric/Latin text: apply a conversion rule such as "3시 30분" → "세시 삼십분"
• When synthesizing sentence by sentence, split on periods (Korean intonation changes little after a comma, so the sentence is the right unit)

LLM Korean optimization:

• State it in the system prompt: "한국어로 간결하게 답변하세요. 2~3문장 이내로." (a Korean instruction to answer concisely and briefly)
• Ollama model choice: gemma3:4b (good Korean quality) or EXAONE-3.5-2.4B (LG, Korean-specialized)
• Capping the response length (max_tokens: 256) raises the chance the response finishes before a barge-in

Latency Benchmark Targets

The latency targets at which a real-time voice conversation feels natural to a user are as follows.

Latency Targets per Stage

Achievable Latency by Environment

Implementation Checklist

Before starting the project, run through the checklist below.

Preparing the Environment

• Python 3.10+ installed
• CUDA 12.x + cuDNN (when using a GPU)
• portaudio system library installed (brew install portaudio / apt install portaudio19-dev)
• Microphone and speaker confirmed working
• Ollama installed and the model downloaded (ollama pull gemma3:4b)

Downloading the Models

• Silero VAD ONNX model (downloads automatically)
• faster-whisper model (medium or large-v3)
• Piper Korean model (ko_KR-kss-medium)
• (Optional) openWakeWord model

Verifying the Pipeline

• VAD alone: microphone → VAD → console output
• STT alone: WAV file → faster-whisper → text
• LLM alone: check the Ollama streaming response with curl
• TTS alone: text → Piper → play the WAV file
• Integration test: confirm the whole pipeline works
• barge-in test: confirm it stops instantly when you speak during a response

Production Deployment

• Error logging configured (structured logs)
• Metrics collected (per-stage latency)
• Memory leak test (long-running)
• Concurrent connection test (for the server-client setup)

Troubleshooting Guide

Advanced Topic: Going to Production with LiveKit Agents

To go beyond a local prototype into production, consider the LiveKit Agents framework. It builds the STT/LLM/TTS pipeline and turn detection into a WebRTC-based SFU server, and supports browser, mobile, and IoT clients alike.

# A barge-in voice chatbot on LiveKit Agents (brief example)
from livekit.agents import AutoSubscribe, JobContext, WorkerOptions, cli
from livekit.agents.voice import Agent, AgentSession
from livekit.plugins import silero, openai, deepgram

async def entrypoint(ctx: JobContext):
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)

    agent = Agent(
        vad=silero.VAD.load(),
        stt=deepgram.STT(language="ko"),
        llm=openai.LLM(
            base_url="http://localhost:11434/v1",  # Ollama
            model="gemma3:4b",
        ),
        tts=openai.TTS(),  # or a custom Piper TTS plugin
        # barge-in is enabled by default - when VAD detects user speech
        # it stops the LLM/TTS automatically and starts a new turn
    )
    session = AgentSession()
    await session.start(agent=agent, room=ctx.room)

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

LiveKit Agents handles barge-in with a turn detection transformer model, giving more refined interruption judgements than plain VAD (distinguishing a cough or background noise).

Additional Open-Source Frameworks

Conclusion

Building a real-time barge-in voice chatbot from open source alone is entirely feasible. Three things matter:

1. An asynchronous pipeline: connect every component with asyncio and design it so Task.cancel() can stop it instantly
2. State machine driven control: define clear transition rules across 5 states (IDLE/LISTENING/THINKING/SPEAKING/INTERRUPTED)
3. Managing the latency budget: target under 800ms in total, allocating and measuring a budget for each stage

The most practical starting point is the Silero VAD + faster-whisper + Ollama + Piper combination. With a GPU you can respond within 800ms, and on CPU alone the Vosk + Piper combination with a remote LLM attached reaches responses in the 2-second range.

In a Korean setting, stating language="ko" for STT, steering the LLM towards concise answers, and preprocessing numbers and Latin script for TTS are what determine quality. Use the code in this article as a base, swapping and improving each module one at a time, and build your own voice AI assistant.

References

1. Silero VAD — GitHub — MIT-licensed VAD, ONNX/PyTorch support
2. faster-whisper — GitHub — fast Whisper inference on CTranslate2
3. whisper.cpp — GitHub — a ggml-based Whisper C++ implementation, Metal/CUDA/Vulkan
4. Vosk — GitHub — lightweight offline STT with native streaming support
5. Piper TTS — GitHub — lightweight VITS-based TTS with multilingual support
6. Ollama — official site — one-command local LLM serving
7. LiveKit Agents — GitHub — a real-time voice AI agent framework
8. Pipecat — GitHub — an open-source multimodal conversational AI framework
9. openWakeWord — GitHub — open-source wake word detection
10. RealtimeSTT — GitHub — real-time STT with VAD and wake word
11. vLLM — GitHub — a high-performance LLM serving engine
12. Coqui TTS (community fork) — PyPI — XTTS v2 voice cloning TTS