오픈소스 실시간 대화형 음성 챗봇 구축 가이드: 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에 스트리밍 요청, 토큰 단위로 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))