开源实时对话式语音聊天机器人构建指南：Barge-In(应答中断)支持架构与实现

┌─────────────────────────────────────────────────────────────┐
│                        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    │                   │
│                             └───────────┘                   │
└─────────────────────────────────────────────────────────────┘

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

from enum import Enum, auto

class State(Enum):
    IDLE = auto()        # 待机中(等待唤醒词或语音检测)
    LISTENING = auto()   # 正在录制用户发话
    THINKING = auto()    # STT → LLM 处理中
    SPEAKING = auto()    # 正在播放 TTS 语音
    INTERRUPTED = auto() # 发生 barge-in,清理后转为 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},
}

pip install silero-vad faster-whisper openai-whisper piper-tts \
            sounddevice numpy httpx asyncio
# 安装 Ollama (macOS/Linux)
curl -fsSL https://ollama.com/install.sh | sh
ollama pull gemma3:4b  # 或替换为你想用的模型

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. 状态机定义
# ──────────────────────────────────────────────
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 模块(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:
        """接收 30ms 音频分块,返回 '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 模块(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 模块(Ollama OpenAI 兼容 API)
# ──────────────────────────────────────────────
async def stream_llm_response(
    messages: list[dict],
    model: str = "gemma3:4b",
    base_url: str = "http://localhost:11434/v1",
) -> asyncio.AsyncGenerator:
    """向 Ollama 发起流式请求,按 token 逐个 yield。"""
    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 模块(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:
        """将文本转换为 WAV 字节。"""
        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
    ):
        """按句子合成 TTS 后放入音频队列。"""
        buffer = ""
        sentence_delimiters = {".", "!", "?", "。", "!", "?", "\n"}

        while True:
            chunk = await text_chunks.get()
            if chunk is None:  # 结束信号
                if buffer.strip():
                    audio = await self.synthesize(buffer)
                    await audio_queue.put(audio)
                await audio_queue.put(None)  # 播放结束信号
                break

            buffer += chunk
            # 句子完成时立即合成
            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. 音频播放(应对 barge-in)
# ──────────────────────────────────────────────
async def play_audio(ctx: PipelineContext, sample_rate: int = 22050):
    """从音频队列取出分块进行播放。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  # 按 100ms 为单位播放

        for i in range(0, len(audio_array), chunk_size):
            if ctx.interrupt_event.is_set():
                # barge-in!立即中断播放
                # 清空队列
                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. 主流水线(支持 Barge-In)
# ──────────────────────────────────────────────
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()

            # 从麦克风读取 30ms 分块
            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 转换
            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 流式生成 + TTS 同时执行
            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():
                """在 THINKING/SPEAKING 状态中检测 barge-in。"""
                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] 检测到用户发话!中断响应")
                        ctx.interrupt_event.set()
                        ctx.transition(Event.BARGE_IN)
                        return
                    await asyncio.sleep(0.01)

            # 执行任务
            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)

            # 等待完成或中断
            done, pending = await asyncio.wait(
                [ctx.llm_task, ctx.tts_task, playback_task, monitor_task],
                return_when=asyncio.FIRST_COMPLETED,
            )

            # 发生 barge-in 时进行清理
            if ctx.interrupt_event.is_set():
                for task in pending:
                    task.cancel()
                    try:
                        await task
                    except asyncio.CancelledError:
                        pass

                # 保留部分响应
                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:
                # 正常完成
                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())

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()):
    """指数退避重试包装器。"""
    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"{delay}s 后重试")
            await asyncio.sleep(delay)

class AudioPipelineQueues:
    """管理流水线队列。各阶段之间通过异步队列连接。"""

    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):
        """在 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):
        """只清空 STT 之后的队列(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

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

┌──────────┐     WebSocket/     ┌────────────────┐
│  Client  │◀────WebRTC────────▶│   GPU Server    │
│ (树莓派   │    audio stream   │ STT + LLM + TTS │
│  / 浏览器) │                  │                │
└──────────┘                    └────────────────┘

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

# 基于 LiveKit Agents 的 barge-in 语音聊天机器人(简化示例)
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(),  # 或自定义的 Piper TTS 插件
        # barge-in 默认启用 — 当 VAD 检测到用户发话时
        # 会自动中断 LLM/TTS 并开始新的一轮
    )
    session = AgentSession()
    await session.start(agent=agent, room=ctx.room)

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