Building an AI VTuber Agent, Part 4: Speech-to-Text

Related Series

Building an AI VTuber Agent, Part 1: Overview
Building an AI VTuber Agent, Part 2: LLM and Persona
Building an AI VTuber Agent, Part 3: Memory
Building an AI VTuber Agent, Part 4: STT (current post)
Building an AI VTuber Agent, Part 5: TTS
Building an AI VTuber Agent, Part 6: Live2D
Building an AI VTuber Agent, Part 7: Streaming Integration
Building an AI VTuber Agent, Part 8: Operations

Introduction

In Part 3, Haru got real memory — episode memories stored in a Qdrant vector DB, semantic memories extracted by the LLM, and a natural forgetting curve modeled after Ebbinghaus.

But up to this point, talking to Haru still requires a keyboard. In a real VTuber context, typing feels out of place. One of the things that makes Neuro-sama compelling is how she responds immediately when a viewer speaks to her. In Part 4, we give Haru a pair of ears.

The STT pipeline looks simple on the surface, but getting it to actually work well involves more complexity than you'd expect.

Without VAD, there's no way to know when to run STT — leaving the mic open continuously means silence gets pushed through the STT pipeline too
Streaming STT is fundamentally a buffer management problem — when to cut chunks, when to commit a final result
Interrupt handling is a system-wide concern — not just the STT layer; LLM inference, TTS playback, and audio buffers all have to cooperate
Echo cancellation is essential in headphone-free setups — if Haru's TTS output leaks back into the mic, you get an infinite loop

Let's work through each of these.

The Audio Pipeline Architecture

Before writing any code, map out the data flow.

Browser mic (MediaStream API)
      ↓ PCM 16kHz, mono
WebSocket audio stream (chunked)
      ↓
VAD (Silero VAD)
      ├── Silence detected → hold buffer
      └── Speech detected → accumulate buffer
            ↓ End-of-utterance detected
      faster-whisper inference
            ↓ text + language + confidence
      LLM processing (Part 3 character server)
            ↓ streaming response + emotion metadata
      TTS pipeline (Part 5)
            ↓ audio playing...

      [Interrupt signal: VAD detects new speech]
            ↓
      TTS stops immediately → LLM cancelled → new input processed

In this flow, VAD controls nearly everything. VAD detecting speech starts the pipeline; VAD detecting silence triggers STT; VAD detecting new speech while the AI is talking fires an interrupt.

VAD — The Ear That Listens

Voice Activity Detection (VAD) distinguishes segments of an audio stream that contain human speech from those that don't. Simple in concept, but it has to stay accurate even with background noise, keyboard sounds, and the AI's own TTS output leaking into the mic.

VAD Options Compared

Library	Approach	Accuracy	Latency	Setup
`webrtcvad`	Energy-based	△ Moderate	⚡⚡⚡ Very fast	Easy
`silero-vad`	Neural network (ONNX)	✅✅ Very high	⚡⚡ Fast	Moderate
`pyannote-audio`	Deep learning	✅✅ Best-in-class	⚡ Slow	Complex

Choice: Silero VAD. It's robust to background noise, highly accurate, and fast enough on CPU via the ONNX runtime. webrtcvad is energy-based and therefore vulnerable to echo and noise. pyannote-audio is overkill — it's built for speaker diarization.

uv add silero-vad torch torchaudio

Basic Silero VAD Implementation

# src/vtuber/stt/vad.py
import torch
import numpy as np
from dataclasses import dataclass

# Load Silero VAD model once at module level
_model, _utils = torch.hub.load(
    repo_or_dir="snakers4/silero-vad",
    model="silero_vad",
    force_reload=False,
    onnx=True,          # Use ONNX runtime (CPU-optimized)
)
(get_speech_timestamps, _, read_audio, *_) = _utils


@dataclass
class VADChunk:
    audio: np.ndarray   # PCM float32, 16kHz
    is_speech: bool
    speech_prob: float  # 0.0 ~ 1.0


class SileroVAD:
    """Speech detector backed by Silero VAD"""

    SAMPLE_RATE = 16000
    CHUNK_SIZE  = 512   # 32ms @ 16kHz (Silero recommended value)
    THRESHOLD   = 0.5   # Speech classification threshold

    def __init__(self, threshold: float = THRESHOLD):
        self._model     = _model
        self._threshold = threshold
        self._reset_state()

    def _reset_state(self) -> None:
        self._h = torch.zeros((2, 1, 64))
        self._c = torch.zeros((2, 1, 64))

    def process_chunk(self, pcm_bytes: bytes) -> VADChunk:
        """Accept 16-bit PCM bytes and classify as speech or non-speech"""
        audio  = np.frombuffer(pcm_bytes, dtype=np.int16).astype(np.float32) / 32768.0
        tensor = torch.from_numpy(audio).unsqueeze(0)  # (1, 512)

        with torch.no_grad():
            speech_prob, self._h, self._c = self._model(
                tensor, self.SAMPLE_RATE, self._h, self._c
            )

        prob = speech_prob.item()
        return VADChunk(audio=audio, is_speech=prob >= self._threshold, speech_prob=prob)

    def reset(self) -> None:
        """Reset VAD state for a new utterance"""
        self._reset_state()

Silero VAD maintains internal state across frames (self._h, self._c). Keeping this state alive as chunks stream through is what gives it high accuracy. When a session ends, call reset() to clear the state.

Utterance Boundary Detector

This is a state machine that sits on top of VAD output and determines when an utterance starts and ends. Naively treating "speech → silence" as an utterance boundary causes splits on brief pauses within a sentence. A silence timeout approach solves this.

# src/vtuber/stt/boundary.py
from dataclasses import dataclass, field
from enum import Enum
import time


class SpeechState(Enum):
    IDLE     = "idle"     # Silence
    SPEAKING = "speaking" # Utterance in progress
    ENDING   = "ending"   # Waiting to confirm end (short pause)


@dataclass
class SpeechEvent:
    type:         str        # "start" | "end" | "cancel"
    audio_frames: list       # Accumulated audio chunks
    duration_ms:  float = 0  # Utterance duration


class SpeechBoundaryDetector:
    """Detects utterance start/end events from a VAD chunk stream"""

    SILENCE_TIMEOUT_MS = 600    # 600ms of silence → end of utterance
    MIN_SPEECH_MS      = 200    # Utterances shorter than 200ms are treated as noise
    MAX_SPEECH_MS      = 15000  # Utterances longer than 15s are force-ended

    def __init__(self):
        self._state:         SpeechState = SpeechState.IDLE
        self._buffer:        list        = []
        self._speech_start:  float       = 0.0
        self._silence_start: float       = 0.0

    def feed(self, chunk: VADChunk) -> SpeechEvent | None:
        """Feed a VAD chunk; returns an event or None"""
        now = time.monotonic() * 1000  # ms

        if self._state == SpeechState.IDLE:
            if chunk.is_speech:
                self._state        = SpeechState.SPEAKING
                self._speech_start = now
                self._buffer       = [chunk.audio]
                return SpeechEvent(type="start", audio_frames=[])

        elif self._state == SpeechState.SPEAKING:
            self._buffer.append(chunk.audio)
            duration = now - self._speech_start

            if not chunk.is_speech:
                self._state         = SpeechState.ENDING
                self._silence_start = now
            elif duration > self.MAX_SPEECH_MS:
                return self._finalize(now)

        elif self._state == SpeechState.ENDING:
            self._buffer.append(chunk.audio)
            silence_ms = now - self._silence_start

            if chunk.is_speech:
                # Speech resumed after a short pause → continue utterance
                self._state = SpeechState.SPEAKING
            elif silence_ms >= self.SILENCE_TIMEOUT_MS:
                return self._finalize(now)

        return None

    def _finalize(self, now: float) -> SpeechEvent:
        duration = now - self._speech_start
        frames   = list(self._buffer)
        self._state  = SpeechState.IDLE
        self._buffer = []

        if duration < self.MIN_SPEECH_MS:
            return SpeechEvent(type="cancel", audio_frames=[], duration_ms=duration)

        return SpeechEvent(type="end", audio_frames=frames, duration_ms=duration)

SILENCE_TIMEOUT_MS = 600 is empirically determined. Too short and utterances get cut mid-sentence; too long and response latency increases. Korean tends to have longer sentence-final endings, so 700–800ms may feel more natural. Tune it to your environment.

faster-whisper — Real-Time STT Engine

Why faster-whisper

OpenAI's original Whisper is accurate but slow to infer. faster-whisper reimplements the same Whisper architecture on top of CTranslate2, delivering up to 4× the throughput at lower memory usage. It supports INT8 quantization out of the box, making it practical even without a GPU.

uv add faster-whisper

Model size	Parameters	VRAM (FP16)	Korean accuracy	CPU inference*
tiny	39M	~1GB	△ Low	~0.3s
base	74M	~1GB	△ Moderate	~0.5s
small	244M	~2GB	✅ Good	~1.5s
medium	769M	~5GB	✅✅ Great	~4s
large-v3	1550M	~10GB	✅✅ Best	~10s

*CPU baseline, 5-second utterance / Apple M2 Pro reference

This series uses: small (CPU) / large-v3 (GPU), switchable via an environment variable.

STT Engine Implementation

# src/vtuber/stt/engine.py
import asyncio
import numpy as np
from dataclasses import dataclass
from faster_whisper import WhisperModel


@dataclass
class TranscribeResult:
    text:        str
    language:    str
    confidence:  float   # 0.0 ~ 1.0 (approximate, based on avg_logprob)
    duration_ms: float


class FasterWhisperSTT:
    """faster-whisper-based STT engine"""

    SAMPLE_RATE = 16000

    def __init__(
        self,
        model_size:   str        = "small",
        device:       str        = "cpu",       # "cpu" | "cuda" | "auto"
        compute_type: str        = "int8",      # "int8" | "float16" | "float32"
        language:     str | None = "ko",        # None for auto-detect
    ):
        self._model    = WhisperModel(model_size, device=device, compute_type=compute_type)
        self._language = language

    def transcribe(self, audio_frames: list[np.ndarray]) -> TranscribeResult:
        """Transcribe a list of VAD-extracted audio frames to text"""
        audio       = np.concatenate(audio_frames, axis=0)
        duration_ms = len(audio) / self.SAMPLE_RATE * 1000

        segments, info = self._model.transcribe(
            audio,
            language=self._language,
            beam_size=5,
            vad_filter=False,     # already passed through VAD, so disable
            word_timestamps=False,
        )

        texts, log_probs = [], []
        for seg in segments:
            texts.append(seg.text.strip())
            if seg.avg_logprob is not None:
                log_probs.append(seg.avg_logprob)

        full_text   = " ".join(texts)
        avg_logprob = sum(log_probs) / len(log_probs) if log_probs else -1.0
        confidence  = min(1.0, max(0.0, float(np.exp(avg_logprob))))

        return TranscribeResult(
            text=full_text,
            language=info.language,
            confidence=confidence,
            duration_ms=duration_ms,
        )

    async def transcribe_async(
        self, audio_frames: list[np.ndarray]
    ) -> TranscribeResult:
        """Async wrapper — runs CPU-intensive inference in a thread pool"""
        return await asyncio.to_thread(self.transcribe, audio_frames)

asyncio.to_thread() is the critical piece here. faster-whisper's inference is synchronous, so it must run in a separate thread to avoid blocking FastAPI's event loop. Miss this, and every other WebSocket connection freezes for the duration of an STT inference call.

WebSocket Audio Stream — Connecting the Browser Microphone

This is the channel that sends browser microphone input to the server. The reason to use WebSocket instead of HTTP is that we need bidirectional, real-time communication — the server must be able to push STT results, interrupt signals, and LLM responses to the client at any time.

Browser Audio Capture (Frontend)

// src/frontend/audio/capture.ts
const SAMPLE_RATE = 16000;
const CHUNK_MS    = 32;  // 32ms = 512 samples for Silero VAD

export class AudioCapture {
  private ws:        WebSocket;
  private context:   AudioContext | null        = null;
  private processor: ScriptProcessorNode | null = null;

  constructor(wsUrl: string) {
    this.ws = new WebSocket(wsUrl);
    this.ws.binaryType = "arraybuffer";
  }

  async start(): Promise<void> {
    const stream = await navigator.mediaDevices.getUserMedia({
      audio: {
        sampleRate:       SAMPLE_RATE,
        channelCount:     1,
        echoCancellation: true,   // enable browser-level echo cancellation
        noiseSuppression: true,   // enable background noise suppression
        autoGainControl:  true,   // enable automatic gain control
      },
    });

    this.context = new AudioContext({ sampleRate: SAMPLE_RATE });
    const source = this.context.createMediaStreamSource(stream);

    // Extract PCM via ScriptProcessorNode
    const chunkSize = Math.round(SAMPLE_RATE * CHUNK_MS / 1000);  // 512
    this.processor  = this.context.createScriptProcessor(chunkSize, 1, 1);

    this.processor.onaudioprocess = (event) => {
      if (this.ws.readyState !== WebSocket.OPEN) return;

      const float32 = event.inputBuffer.getChannelData(0);
      // Convert Float32 → Int16 (server expects int16 PCM)
      const int16 = new Int16Array(float32.length);
      for (let i = 0; i < float32.length; i++) {
        int16[i] = Math.max(-32768, Math.min(32767, float32[i] * 32768));
      }
      this.ws.send(int16.buffer);
    };

    source.connect(this.processor);
    this.processor.connect(this.context.destination);
  }

  stop(): void {
    this.processor?.disconnect();
    this.context?.close();
  }
}

echoCancellation: true matters. It enables WebRTC-stack echo cancellation at the browser level. That said, it isn't sufficient on its own, so server-side echo cancellation is applied as well — covered later.

Server-Side WebSocket Handler

# src/vtuber/api/voice.py
import asyncio, json, numpy as np
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from ..stt.vad      import SileroVAD
from ..stt.boundary import SpeechBoundaryDetector
from ..stt.engine   import FasterWhisperSTT
from ..stt.interrupt import InterruptController

router = APIRouter(prefix="/voice", tags=["voice"])


class VoiceSession:
    """Per-connection voice processing state for a single WebSocket"""

    def __init__(self, ws: WebSocket, stt: FasterWhisperSTT):
        self.ws             = ws
        self.stt            = stt
        self.vad            = SileroVAD()
        self.boundary       = SpeechBoundaryDetector()
        self.interrupt      = InterruptController()
        self.is_ai_speaking = False  # True while AI TTS is playing

    async def send_json(self, data: dict) -> None:
        await self.ws.send_text(json.dumps(data, ensure_ascii=False))

    async def process_chunk(self, pcm_bytes: bytes) -> None:
        # Echo cancellation: raise the VAD threshold while the AI is speaking
        self.vad._threshold = 0.8 if self.is_ai_speaking else 0.5

        chunk = self.vad.process_chunk(pcm_bytes)
        event = self.boundary.feed(chunk)

        if event is None:
            return

        if event.type == "start":
            await self.send_json({"type": "speech_start"})
            # Human starts speaking while AI is talking → interrupt
            if self.is_ai_speaking:
                await self.interrupt.trigger()
                await self.send_json({"type": "interrupt"})

        elif event.type == "end":
            await self.send_json({"type": "speech_end"})
            result = await self.stt.transcribe_async(event.audio_frames)
            if result.text.strip():
                await self.send_json({
                    "type":        "transcript",
                    "text":        result.text,
                    "language":    result.language,
                    "confidence":  round(result.confidence, 3),
                    "duration_ms": round(event.duration_ms),
                })

        elif event.type == "cancel":
            pass  # utterance too short — ignore


@router.websocket("/ws")
async def voice_websocket(websocket: WebSocket, stt: FasterWhisperSTT):
    await websocket.accept()
    session = VoiceSession(websocket, stt)

    try:
        while True:
            data = await websocket.receive()

            if "bytes" in data:
                await session.process_chunk(data["bytes"])

            elif "text" in data:
                msg = json.loads(data["text"])
                if msg.get("type") == "ai_speaking_start":
                    session.is_ai_speaking = True
                elif msg.get("type") == "ai_speaking_end":
                    session.is_ai_speaking = False

    except WebSocketDisconnect:
        pass

Interrupt Mechanism — System-Wide Coordination

Interrupts aren't just an STT-layer concern. When a person starts speaking while the AI is talking, three things must happen in sequence:

Stop audio playback — halt TTS output immediately
Cancel LLM streaming — abort the in-progress LLM inference
Process new input — pass the STT result to the LLM

There must be no race conditions between these three steps. Propagating signals via asyncio.Event is the cleanest approach.

InterruptController

# src/vtuber/stt/interrupt.py
import asyncio
from dataclasses import dataclass, field


@dataclass
class InterruptController:
    """인터럽트 신호를 시스템 전체에 전파하는 컨트롤러"""

    _cancel_token:    asyncio.Event = field(default_factory=asyncio.Event)
    _interrupt_event: asyncio.Event = field(default_factory=asyncio.Event)

    async def trigger(self) -> None:
        """인터럽트 발생 — 진행 중인 모든 스트리밍 취소"""
        self._cancel_token.set()
        self._interrupt_event.set()

    def reset(self) -> None:
        """새 발화 처리 전 상태 초기화"""
        self._cancel_token.clear()
        self._interrupt_event.clear()

    @property
    def is_cancelled(self) -> bool:
        return self._cancel_token.is_set()

    async def wait_for_interrupt(self) -> None:
        await self._interrupt_event.wait()

Handling Interrupts in LLM Streaming

We update the chat_stream endpoint from Part 2. The key is checking is_cancelled on every iteration of the streaming loop.

# src/vtuber/api/chat.py (인터럽트 처리 추가)

async def generate(interrupt: InterruptController):
    full_response = []
    try:
        async for chunk in llm.chat_stream(messages):
            # 인터럽트 체크 — 취소 요청이 있으면 즉시 스트림 종료
            if interrupt.is_cancelled:
                yield f"data: {json.dumps({'type': 'interrupted'}, ensure_ascii=False)}\n\n"
                return

            full_response.append(chunk.content)
            if chunk.content:
                yield f"data: {json.dumps({'type': 'token', 'content': chunk.content}, ensure_ascii=False)}\n\n"

            if chunk.is_final:
                # ... (3편과 동일한 메모리 저장 처리)
                break

    except asyncio.CancelledError:
        yield f"data: {json.dumps({'type': 'interrupted'}, ensure_ascii=False)}\n\n"
    finally:
        interrupt.reset()

Client-Side Interrupt Flow

// src/frontend/audio/interrupt.ts
export class VoiceClient {
  private ws:           WebSocket;
  private currentAudio: HTMLAudioElement | null = null;

  onInterrupt(): void {
    // 1. 재생 중인 오디오 즉시 중단
    if (this.currentAudio) {
      this.currentAudio.pause();
      this.currentAudio.currentTime = 0;
      this.currentAudio = null;
    }

    // 2. 서버에 AI 재생 종료 알림
    this.ws.send(JSON.stringify({ type: "ai_speaking_end" }));

    // 3. 진행 중인 LLM 스트리밍 중단 요청
    this.ws.send(JSON.stringify({ type: "cancel_llm" }));

    console.log("[Interrupt] AI stopped. Listening for new input.");
  }
}

Stopping audio first matters. Audio must be silenced before canceling the LLM — that way the user experiences immediate silence on screen even while the server is still processing the cancellation.

Echo Cancellation — Working Without Headphones

When AI audio plays through speakers without headphones, the microphone picks it up. VAD mistakes this for a real human voice, and STT transcribes what the AI just said — creating an infinite loop. Echo cancellation breaks that loop. We use three layers of defense.

Layer 1: Browser Built-In Echo Cancellation

We already enabled this earlier with echoCancellation: true. In most environments, this alone is sufficient. The browser's WebRTC stack computes the delay between speaker output and microphone input and filters out the echo.

Layer 2: Server-Side Playback State Gating

A fallback for environments where browser echo cancellation doesn't work (some Linux setups, older browsers, Bluetooth speakers). While the AI is playing TTS, we raise the VAD threshold (0.5 → 0.8) to reduce the chance that echo gets classified as speech.

# 레이어 2: VAD 임계값 동적 조정 (VoiceSession.process_chunk 내부)
self.vad._threshold = 0.8 if self.is_ai_speaking else 0.5

Layer 3: Text-Level Echo Removal (Last Resort)

If the STT result has very high text similarity to what the AI just said, we discard it.

# src/vtuber/stt/echo_filter.py
from difflib import SequenceMatcher


class EchoFilter:
    """STT 결과가 AI 발화의 에코인지 검사"""

    SIMILARITY_THRESHOLD = 0.75

    def __init__(self):
        self._recent_ai_texts: list[str] = []

    def register_ai_speech(self, text: str) -> None:
        """AI가 방금 말한 텍스트를 등록 (TTS 재생 직전에 호출)"""
        self._recent_ai_texts.append(text)
        self._recent_ai_texts = self._recent_ai_texts[-3:]

    def is_echo(self, transcript: str) -> bool:
        """STT 결과가 AI 발화의 에코인지 판단"""
        for ai_text in self._recent_ai_texts:
            ratio = SequenceMatcher(None, transcript, ai_text).ratio()
            if ratio >= self.SIMILARITY_THRESHOLD:
                return True
        return False

Language Detection and Multilingual Handling

faster-whisper auto-detects the spoken language by default. Pinning to Korean (language="ko") is faster, but if users frequently mix in English or Japanese, setting language=None for auto-detection is the better choice.

Common cases in a Korean VTuber environment:

"요즘 LLM inference optimization 공부하는데..."  ← 영어 혼용
"저 오늘 야키토리 먹었어요"                       ← 일본어 단어 포함
"CUDA OOM 오류 어떻게 고쳐요?"                   ← 영문 기술 용어

faster-whisper's auto-detection handles this kind of code-switching reasonably well. That said, auto-detection adds an extra 100–200ms because it analyzes the first 30 tokens before transcribing.

# 환경 변수로 언어 전략 선택
import os
WHISPER_LANGUAGE = os.getenv("WHISPER_LANGUAGE", "ko")  # "ko" | "auto"

stt = FasterWhisperSTT(
    model_size="small",
    language=None if WHISPER_LANGUAGE == "auto" else WHISPER_LANGUAGE,
)

Assembling the Full Pipeline

We wire VAD → speech boundary → STT → echo filter together into a single STTPipeline.

# src/vtuber/stt/pipeline.py
import asyncio
from dataclasses import dataclass
from .vad         import SileroVAD
from .boundary    import SpeechBoundaryDetector
from .engine      import FasterWhisperSTT, TranscribeResult
from .echo_filter import EchoFilter
from .interrupt   import InterruptController


@dataclass
class STTEvent:
    type:   str                       # "speech_start" | "transcript" | "interrupt"
    result: TranscribeResult | None = None


class STTPipeline:
    """VAD → 경계 감지 → STT → 에코 필터 통합 파이프라인"""

    def __init__(
        self,
        stt:       FasterWhisperSTT,
        interrupt: InterruptController,
    ):
        self._stt            = stt
        self._vad            = SileroVAD()
        self._boundary       = SpeechBoundaryDetector()
        self._echo           = EchoFilter()
        self._interrupt      = interrupt
        self._is_ai_speaking = False

    def set_ai_speaking(self, speaking: bool) -> None:
        self._is_ai_speaking = speaking

    def register_ai_speech(self, text: str) -> None:
        self._echo.register_ai_speech(text)

    async def process_chunk(self, pcm_bytes: bytes) -> STTEvent | None:
        """오디오 청크 처리 → 이벤트 반환 또는 None"""
        self._vad._threshold = 0.8 if self._is_ai_speaking else 0.5

        chunk = self._vad.process_chunk(pcm_bytes)
        event = self._boundary.feed(chunk)

        if event is None:
            return None

        if event.type == "start":
            if self._is_ai_speaking:
                await self._interrupt.trigger()
                return STTEvent(type="interrupt")
            return STTEvent(type="speech_start")

        elif event.type == "end":
            result = await self._stt.transcribe_async(event.audio_frames)
            if not result.text.strip():
                return None
            if self._echo.is_echo(result.text):
                return None  # 에코 필터링
            return STTEvent(type="transcript", result=result)

        return None

Updated Project Structure

vtuber-assistant/
├── src/
│   └── vtuber/
│       ├── stt/               ← 4편에서 추가
│       │   ├── vad.py         # SileroVAD
│       │   ├── boundary.py    # SpeechBoundaryDetector
│       │   ├── engine.py      # FasterWhisperSTT
│       │   ├── echo_filter.py # EchoFilter
│       │   ├── interrupt.py   # InterruptController
│       │   └── pipeline.py    # STTPipeline (통합)
│       ├── api/
│       │   ├── chat.py        ← 인터럽트 처리 업데이트
│       │   └── voice.py       ← 4편에서 추가 (WebSocket)
│       ├── character/         ← 2편
│       ├── llm/               ← 2편
│       └── memory/            ← 3편

Additional Dependencies

# pyproject.toml dependencies에 추가
"faster-whisper>=1.0.0",
"silero-vad>=5.1.0",
"torch>=2.0.0",

Latency Benchmarks and Optimization

Comparing the STT latency estimates from Part 1 against actual measurements, CPU performance comes in slower than expected.

Estimated (Part 1):    ~200ms  (short utterances)
Measured (small, CPU): 350~800ms (varies significantly with utterance length)
Measured (small, GPU): 80~200ms

Three optimizations are applied.

Optimization 1: Model Warmup

At server startup, run a dummy inference to load the model into memory. This eliminates model-loading latency on the first real request.

# src/vtuber/main.py lifespan에 추가
async def lifespan(app: FastAPI):
    dummy_audio = [np.zeros(16000, dtype=np.float32)]  # 1초 침묵
    stt_engine.transcribe(dummy_audio)
    print("✅ Whisper 모델 워밍업 완료")
    yield

Optimization 2: Use the tiny Model for Short Utterances

Short responses like "uh-huh," "right," or "yeah" are recognized well enough by the tiny model. The model is selected dynamically based on utterance length.

class AdaptiveSTT:
    """발화 길이에 따라 모델을 동적으로 선택하는 STT"""

    SHORT_THRESHOLD_S = 2.0  # 2초 미만이면 tiny 사용

    def __init__(self):
        self._tiny  = FasterWhisperSTT(model_size="tiny")
        self._small = FasterWhisperSTT(model_size="small")

    async def transcribe_async(
        self, audio_frames: list[np.ndarray]
    ) -> TranscribeResult:
        audio      = np.concatenate(audio_frames)
        duration_s = len(audio) / FasterWhisperSTT.SAMPLE_RATE
        model      = self._tiny if duration_s < self.SHORT_THRESHOLD_S else self._small
        return await asyncio.to_thread(model.transcribe, audio_frames)

Optimization 3: Prefetch Memory as Soon as Speech Starts

Starting an async memory retrieval the moment a speech-start event arrives is worth considering. There's no partial text available, so practical gains are limited — but overlapping STT completion with memory retrieval is a valid direction.

Verifying End-to-End Behavior

Start the server and confirm the WebSocket connection is working.

# Silero VAD 모델 다운로드 (최초 1회)
python -c "import torch; torch.hub.load('snakers4/silero-vad', 'silero_vad', onnx=True)"

# faster-whisper 모델 다운로드 (최초 1회)
python -c "from faster_whisper import WhisperModel; WhisperModel('small', device='cpu', compute_type='int8')"

# 서버 실행
uv run uvicorn src.vtuber.main:app --reload --port 8000

# WebSocket 테스트 (wscat)
wscat -c ws://localhost:8000/voice/ws

Expected event flow when saying "안녕하세요" into the microphone:

← {"type": "speech_start"}
← {"type": "speech_end"}
← {"type": "transcript", "text": "안녕하세요.", "language": "ko", "confidence": 0.921, "duration_ms": 1200}

Interrupt test:

→ {"type": "ai_speaking_start"}    (TTS 재생 시작 알림)
  [AI가 말하는 도중 사람이 말을 시작]
← {"type": "interrupt"}             (서버: 인터럽트 트리거)
← {"type": "speech_start"}
  [발화 종료 후]
← {"type": "transcript", "text": "잠깐만요, 그건 좀 달라요.", ...}

Wrap-Up

Here's a summary of everything implemented in this part:

Silero VAD: Neural-network-based voice activity detection, with dynamic threshold scaling while the AI is speaking
SpeechBoundaryDetector: A state machine that tolerates brief silences between words before declaring end-of-utterance
FasterWhisperSTT: INT8 quantization, CPU/GPU switching, async thread wrapper
WebSocket pipeline: Browser PCM → server VAD → STT → client events
InterruptController: asyncio.Event-based cancellation of both LLM streaming and TTS playback simultaneously
Three-layer echo cancellation: Browser WebRTC + VAD threshold gating + text similarity filter

Haru can now listen. Voice conversations work without any text input, and if you interrupt while she's speaking, she stops immediately and starts listening again.

Part 5 covers the other direction — how Haru speaks. We'll build a character voice with XTTS-v2 and GPT-SoVITS, then pipeline LLM streaming and TTS in chunk-sized units to minimize time-to-first-audio.

Coming Up Next

Part 5: TTS — Building a Character Voice

The response text arrived. 0.2 seconds later, Haru's voice came through the speakers.

Topics covered:

XTTS-v2: Zero-shot voice cloning from a 3-second audio sample, multilingual support
GPT-SoVITS: Few-shot training for a more natural character voice
LLM → TTS pipeline: Minimizing latency by processing sentence-level chunks
Expressive speech: Mapping LLM emotion tags to TTS speed and pitch parameters
RVC (Retrieval-based Voice Conversion): Converting TTS output to a target voice