Building an AI VTuber Agent, Part 5: TTS

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
Building an AI VTuber Agent, Part 5: TTS (current post)
Building an AI VTuber Agent, Part 6: Live2D
Building an AI VTuber Agent, Part 7: Streaming Integration
Building an AI VTuber Agent, Part 8: Production

Introduction

In Part 4, Haru got ears. The STT pipeline was complete — Silero VAD for voice activity detection, faster-whisper for transcription, and an interrupt controller to manage conversation flow.

Now we need to go the other direction. Haru needs to speak.

TTS (Text-to-Speech) looks simple on the surface. Text in, audio out, done — right? In practice, that assumption runs straight into three walls.

First, the voice sounds terrible. Free TTS solutions like Edge TTS produce voices that have zero character to them. The intonation is mechanical, there's no emotion, and above all it doesn't sound like "Haru."

Second, the latency is too high. If you wait for the LLM to finish generating a 500-token response before handing it off to TTS, the user sits in silence for 3–4 seconds. The conversational flow completely breaks down.

Third, there's no emotion. We put a lot of work into designing Haru's emotional states back in Part 2. But whether Haru is happy or nervous, TTS delivers everything in the same flat voice. You've built a character whose voice has no character — a contradiction.

Part 5 solves all three.

TTS Pipeline: Overall Architecture

Before writing any code, let's map out the data flow.

LLM streaming token stream
      ↓
Sentence splitter (SentenceSplitter)
      ↓ sentence-level chunks
TTS synthesis (XTTS-v2 / GPT-SoVITS)
      ↓ PCM audio
[RVC conversion — optional]
      ↓
Audio chunk queue (asyncio.Queue)
      ↓
WebSocket → browser AudioContext playback

There is one core design principle: don't wait for the full LLM response. The moment the first sentence is complete, TTS synthesis starts. While the second sentence is being synthesized, the first sentence's audio is already playing. This is pipelining.

Bad approach:
  LLM [--generate all 500 tokens--] → TTS [--synthesize everything--] → playback starts
  Latency: 3–5 seconds

Good approach:
  LLM [--first sentence--] → TTS [synthesize] → playback starts
                [--second sentence--] → TTS [synthesize] → queued
                              [--third sentence--] → TTS [synthesize] → queued
  Latency: 0.5–1.5 seconds (to first utterance)

The time-to-first-utterance drops from the full LLM generation time to first sentence generation time + TTS synthesis time. That's the whole point of this architecture.

Choosing a TTS Backend — Three Options

This series considers three TTS backend options.

Model	Approach	Korean	Cloning	Speed	Quality
XTTS-v2	Zero-shot cloning	✅ Good	3-sec sample	⚡⚡ Moderate	✅✅ Good
GPT-SoVITS	Few-shot training	✅✅ Very good	Tens of seconds+	⚡⚡ Moderate	✅✅✅ Best
RVC	Voice conversion (post-processing)	Depends on base TTS	Conversion-based	⚡⚡⚡ Fast	✅✅ Good

The strategy adopted in this series is a layered setup.

Base layer: XTTS-v2
  → Works immediately with just a 3-second reference audio sample
  → Zero-shot; no separate training required

High-quality layer: GPT-SoVITS (optional)
  → For when you have enough character-specific voice data
  → More natural, more character-appropriate intonation

Conversion layer: RVC (optional)
  → Converts the output of either model to a different voice
  → Useful when you want to reproduce a specific voice actor's voice

XTTS-v2 — Zero-Shot Voice Cloning

XTTS (Cross-Lingual Text-to-Speech) v2 is a model developed by the Coqui TTS team that reproduces a speaker's voice from as little as 3–6 seconds of audio. It supports 17 languages, including Korean. Because you pass the audio sample as a reference at inference time with no additional training, it's called "zero-shot."

Model Architecture

XTTS-v2 operates in three stages.

Input text + reference audio
      ↓
1. GPT-based language model (text → codebook tokens)
      ↓  +  speaker embedding (extracted from reference audio)
2. Codebook tokens → mel spectrogram
      ↓
3. HiFi-GAN vocoder → PCM audio

A speaker embedding is extracted from the reference audio and used to condition the GPT decoding stage. This embedding captures the speaker's voice characteristics.

Installation and Implementation

uv add TTS soundfile resampy

# src/vtuber/tts/xtts.py
import asyncio
import io
import numpy as np
import soundfile as sf
from TTS.api import TTS


class XTTSv2Backend:
    """Zero-shot voice cloning TTS backed by XTTS-v2"""

    SAMPLE_RATE = 24000  # XTTS-v2 output sample rate

    def __init__(
        self,
        device: str = "cpu",       # "cpu" | "cuda"
        speaker_wav: str = "",      # path to reference audio file
        language: str = "ko",
    ):
        self._tts = TTS(
            model_name="tts_models/multilingual/multi-dataset/xtts_v2",
            progress_bar=False,
        ).to(device)
        self._speaker_wav = speaker_wav
        self._language    = language

    def synthesize(self, text: str, speed: float = 1.0) -> np.ndarray:
        """Convert text to an audio array (synchronous)"""
        wav = self._tts.tts(
            text=text,
            speaker_wav=self._speaker_wav,
            language=self._language,
            speed=speed,
        )
        return np.array(wav, dtype=np.float32)

    async def synthesize_async(
        self, text: str, speed: float = 1.0
    ) -> np.ndarray:
        """Async wrapper — runs CPU inference in a thread pool"""
        return await asyncio.to_thread(self.synthesize, text, speed)

    def to_wav_bytes(self, audio: np.ndarray) -> bytes:
        buf = io.BytesIO()
        sf.write(buf, audio, self.SAMPLE_RATE, format="WAV", subtype="PCM_16")
        return buf.getvalue()

    async def health_check(self) -> bool:
        return self._tts is not None

asyncio.to_thread() wraps the synchronous inference call in an async context. Skip this and the entire event loop blocks during TTS inference. This is the same pattern applied for the same reason in the STT post.

Preparing Reference Audio

XTTS-v2 quality depends heavily on the reference audio. What makes a good reference:

✅ Length: 3–12 seconds (too short = insufficient characteristics; too long = introduces noise)
✅ Quality: Clean audio with no background noise
✅ Content: Sentences with varied intonation (natural conversation preferred over monotone reading)
✅ Format: WAV at 22050 Hz or higher
❌ Avoid: Background music, echo, compression artifacts

# scripts/prepare_reference.py
import soundfile as sf
import numpy as np


def normalize_reference_audio(
    input_path: str,
    output_path: str,
    target_sample_rate: int = 22050,
    target_duration_s: float = 6.0,
) -> None:
    """Preprocess reference audio into a format optimized for XTTS-v2"""
    audio, sr = sf.read(input_path)

    # Convert to mono
    if audio.ndim > 1:
        audio = audio.mean(axis=1)

    # Resample
    if sr != target_sample_rate:
        import resampy
        audio = resampy.resample(audio, sr, target_sample_rate)
        sr = target_sample_rate

    # Trim to target length
    target_samples = int(target_duration_s * sr)
    if len(audio) > target_samples:
        audio = audio[:target_samples]
    elif len(audio) < sr * 2:
        print(f"⚠️ Reference audio too short: {len(audio)/sr:.1f}s")

    # Peak normalization
    peak = np.abs(audio).max()
    if peak > 0:
        audio = audio / peak * 0.95

    sf.write(output_path, audio, sr, subtype="PCM_16")
    print(f"✅ Saved: {output_path} ({len(audio)/sr:.1f}s, {sr}Hz)")

Providing multiple reference files averages the speaker embeddings, producing more stable results. XTTS-v2 accepts a list for speaker_wav.

# Providing multiple references
self._tts.tts(
    text=text,
    speaker_wav=["ref1.wav", "ref2.wav", "ref3.wav"],
    language="ko",
)

GPT-SoVITS — Few-Shot Character Voice

If XTTS-v2 is "zero-shot, ready to use out of the box," GPT-SoVITS is the model that produces "far more natural results through training."

GitHub: https://github.com/RVC-Boss/GPT-SoVITS
⭐ 38,000+ | Python | Started January 2024

GPT-SoVITS uses a two-stage architecture:

1. GPT model (text → semantic tokens)
   - Input: text + semantic tokens from reference audio
   - Output: semantic token sequence for the target utterance

2. SoVITS model (semantic tokens → audio)
   - Input: semantic tokens + speaker embedding
   - Output: high-quality audio waveform

The biggest difference from XTTS-v2 is that fine-tuning is supported. Collect voice data for your character, fine-tune GPT-SoVITS on it, and you get intonation and emotional expression that zero-shot simply cannot match.

Training Data Requirements

Minimum: 30+ seconds of clean voice audio
Recommended: 5–30 minutes
Optimal: 1+ hour (e.g., publicly available VA recordings)

For a VTuber character, you can extract clips from past streams or record audio directly. Since the voice must be isolated without background noise, it's standard practice to first run the audio through a source-separation model like demucs.

API-Based Integration

GPT-SoVITS ships with its own API server. We run it as a separate process and call it over HTTP from our FastAPI server.

# GPT-SoVITS API server (separate process)
cd /path/to/GPT-SoVITS
python api_v2.py --port 9880

# src/vtuber/tts/gptsovits.py
import asyncio
import io
import httpx
import numpy as np
import soundfile as sf


class GPTSoVITSBackend:
    """TTS backend that communicates with the GPT-SoVITS API server"""

    SAMPLE_RATE = 32000  # GPT-SoVITS default output

    def __init__(
        self,
        base_url: str = "http://localhost:9880",
        ref_audio_path: str = "",
        ref_text: str = "",        # Transcript of reference audio (improves accuracy)
        top_k: int = 15,
        top_p: float = 1.0,
        temperature: float = 1.0,
        speed: float = 1.0,
    ):
        self._base_url   = base_url
        self._ref_audio  = ref_audio_path
        self._ref_text   = ref_text
        self._defaults   = {
            "top_k": top_k, "top_p": top_p,
            "temperature": temperature, "speed": speed,
        }
        self._client = httpx.AsyncClient(timeout=30.0)

    async def synthesize_async(
        self,
        text: str,
        speed: float | None = None,
        temperature: float | None = None,
    ) -> np.ndarray:
        params = {
            **self._defaults,
            "text":           text,
            "text_lang":      "ko",
            "ref_audio_path": self._ref_audio,
            "prompt_text":    self._ref_text,
            "prompt_lang":    "ko",
        }
        if speed is not None:
            params["speed"] = speed
        if temperature is not None:
            params["temperature"] = temperature

        resp = await self._client.get(f"{self._base_url}/tts", params=params)
        resp.raise_for_status()

        audio, sr = sf.read(io.BytesIO(resp.content))
        if audio.ndim > 1:
            audio = audio.mean(axis=1)
        return audio.astype(np.float32)

    async def health_check(self) -> bool:
        try:
            resp = await self._client.get(f"{self._base_url}/")
            return resp.status_code == 200
        except Exception:
            return False

XTTS vs GPT-SoVITS: Practical Comparison

Criterion	XTTS-v2	GPT-SoVITS (fine-tuned)
Korean intonation naturalness	✅ Good	✅✅ Excellent
Character voice reproduction	△ Limited by 3-second sample	✅✅ High fidelity
Setup time	Immediate	Several hours of training
Emotional range	△ Limited	✅ Natural
GPU VRAM requirement	~4 GB	~8 GB
CPU inference	✅ (slow)	△ Very slow

Practical takeaway: Use XTTS-v2 for rapid prototyping; use GPT-SoVITS fine-tuning when you need a polished, character-accurate voice. The implementation in this series uses an abstraction layer that supports both backends.

TTS Backend Abstraction

Just as we made the LLM backend swappable, we apply the same abstraction to TTS.

# src/vtuber/tts/base.py
from abc import ABC, abstractmethod
import numpy as np


class TTSBackend(ABC):
    """Interface that all TTS backends must implement"""

    SAMPLE_RATE: int = 24000  # Override in each implementation

    @abstractmethod
    async def synthesize_async(
        self,
        text: str,
        speed: float = 1.0,
    ) -> np.ndarray: ...

    @abstractmethod
    async def health_check(self) -> bool: ...

RVC — A Post-Processing Layer for Voice Conversion

RVC (Retrieval-based Voice Conversion) converts TTS output into a different voice. It doesn't generate speech directly — it transforms it. Train an RVC model on recordings of the target voice, and it will convert any TTS output to sound like that voice.

RVC occupies a special place in the VTuber world because it can swap the vocal characteristics (timbre, harmonic structure) while preserving the intonation from the TTS output. For reproducing a specific VA or character voice, it can outperform both XTTS and GPT-SoVITS.

RVC Pipeline

XTTS-v2 / GPT-SoVITS output
      ↓ PCM audio
RVC (f0 extraction + speaker characteristic conversion)
      ↓
Audio converted to target speaker's voice

uv add rvc-python

# src/vtuber/tts/rvc.py
import asyncio
import numpy as np
from dataclasses import dataclass


@dataclass
class RVCConfig:
    model_path:    str
    index_path:    str   = ""
    f0_method:     str   = "rmvpe"  # rmvpe | crepe | pm
    f0_up_key:     int   = 0        # pitch shift in semitones
    index_rate:    float = 0.75     # index retrieval ratio (0–1)
    filter_radius: int   = 3
    rms_mix_rate:  float = 0.25
    protect:       float = 0.33     # unvoiced sound protection (recommend < 0.5)


class RVCBackend:
    """Voice conversion post-processing via RVC"""

    def __init__(self, config: RVCConfig):
        self._config      = config
        self._sample_rate = 40000   # RVC default output
        self._vc          = None

    def _lazy_load(self) -> None:
        """Load model on first use — reduces startup time"""
        if self._vc is not None:
            return
        from rvc import Config, load_hubert, get_vc
        cfg             = Config()
        hubert          = load_hubert(cfg)
        self._vc, self._net_g, self._sample_rate, *_ = get_vc(
            cfg, self._config.model_path, self._config.index_path
        )
        self._hubert = hubert
        self._cfg    = cfg

    def convert(self, audio: np.ndarray, input_sr: int) -> np.ndarray:
        """Convert audio with RVC (synchronous)"""
        self._lazy_load()
        from rvc import rvc_infer

        result, *_ = rvc_infer(
            index_file=self._config.index_path,
            index_rate=self._config.index_rate,
            f0up_key=self._config.f0_up_key,
            f0method=self._config.f0_method,
            filter_radius=self._config.filter_radius,
            rms_mix_rate=self._config.rms_mix_rate,
            protect=self._config.protect,
            model_file=self._config.model_path,
            audio_input=(audio, input_sr),
            vc=self._vc,
            net_g=self._net_g,
            hubert_model=self._hubert,
            cfg=self._cfg,
        )
        return result.astype(np.float32)

    async def convert_async(
        self, audio: np.ndarray, input_sr: int
    ) -> np.ndarray:
        return await asyncio.to_thread(self.convert, audio, input_sr)

With RVC, the f0 (fundamental frequency) extraction method is the primary determinant of output quality. rmvpe is currently the most accurate and stable option. crepe is faster on GPU but more involved to install, and pm is lightweight but less accurate.

Emotion Tag → TTS Parameter Mapping

In part 2, we designed the LLM responses to carry [EMOTION:happy] tags. This is where those tags actually get used.

We adjust TTS speed and RVC pitch (f0_up_key) based on the detected emotion. Full-blown emotional TTS is far more complex, but these two parameters alone produce a noticeable difference.

# src/vtuber/tts/emotion_mapper.py
from dataclasses import dataclass
from ..character.state import EmotionState


@dataclass
class TTSParams:
    speed:     float = 1.0   # 1.0 baseline; higher = faster
    pitch_key: int   = 0     # semitones (RVC f0_up_key)
    volume:    float = 1.0   # volume multiplier


# Emotion → TTS parameter mapping table
EMOTION_TTS_MAP: dict[EmotionState, TTSParams] = {
    EmotionState.CALM:      TTSParams(speed=1.00, pitch_key=0,  volume=1.00),
    EmotionState.HAPPY:     TTSParams(speed=1.08, pitch_key=1,  volume=1.05),
    EmotionState.CURIOUS:   TTSParams(speed=1.05, pitch_key=0,  volume=1.00),
    EmotionState.EXCITED:   TTSParams(speed=1.15, pitch_key=2,  volume=1.10),
    EmotionState.NERVOUS:   TTSParams(speed=1.12, pitch_key=1,  volume=0.95),
    EmotionState.PLAYFUL:   TTSParams(speed=1.10, pitch_key=1,  volume=1.05),
    EmotionState.TIRED:     TTSParams(speed=0.92, pitch_key=-1, volume=0.90),
    EmotionState.CONCERNED: TTSParams(speed=0.95, pitch_key=-1, volume=0.95),
}


def get_tts_params(emotion: EmotionState) -> TTSParams:
    return EMOTION_TTS_MAP.get(emotion, TTSParams())

These numbers are not ground truth. The optimal values vary by character and by the underlying TTS model. Treat the values above as a starting point and tune them by ear.

Sentence Splitter — The Core of Streaming

Assembling LLM streaming tokens into "complete sentences ready for TTS" is the heart of this pipeline. Cut too early and the speech sounds choppy; cut too late and latency climbs.

# src/vtuber/tts/splitter.py
import re


# 문장 종료 패턴 — 한국어 어미 + 문장부호
SENTENCE_END = re.compile(
    r'(?<=[.!?。！？])\s+'        # 문장부호 + 공백
    r'|(?<=[.!?。！？])$'         # 문장부호로 끝나는 경우
    r'|(?<=다[.。]\s)'             # "~다. " 패턴
    r'|(?<=요[.。]\s)'             # "~요. " 패턴
    r'|(?<=죠[.。]\s)'             # "~죠. " 패턴
)

MIN_CHUNK_LEN = 10  # TTS로 넘기기에 충분한 최소 길이


class SentenceSplitter:
    """LLM 스트리밍 토큰을 문장 단위 청크로 분리"""

    def __init__(self, min_len: int = MIN_CHUNK_LEN):
        self._buffer  = ""
        self._min_len = min_len

    def feed(self, token: str) -> list[str]:
        """토큰을 추가하고 완성된 문장이 있으면 반환"""
        self._buffer += token
        chunks        = []

        while True:
            match = SENTENCE_END.search(self._buffer)
            if not match:
                break

            sentence     = self._buffer[: match.start()].strip()
            self._buffer = self._buffer[match.end():]

            if len(sentence) >= self._min_len:
                chunks.append(sentence)

        return chunks

    def flush(self) -> str | None:
        """스트리밍 종료 시 버퍼에 남은 텍스트 반환"""
        remaining    = self._buffer.strip()
        self._buffer = ""
        return remaining if remaining else None

MIN_CHUNK_LEN = 10 is the minimum length that still makes sense to send to TTS — short enough to handle phrases like "안녕하세요". Anything shorter is usually an interjection or a stray word, and it's better to merge it with the next chunk.

TTSOrchestrator — Full Pipeline Integration

This orchestrator wires every component together, handling the full path from LLM streaming → sentence splitting → TTS synthesis → audio queue.

# src/vtuber/tts/orchestrator.py
import asyncio
import heapq
import io
import numpy as np
import soundfile as sf
from dataclasses import dataclass, field

from .base import TTSBackend
from .splitter import SentenceSplitter
from .emotion_mapper import get_tts_params, TTSParams
from .rvc import RVCBackend
from ..character.state import EmotionState
from ..stt.interrupt import InterruptController


@dataclass
class AudioChunk:
    audio:       np.ndarray
    text:        str
    index:       int
    sample_rate: int


class OrderedAudioQueue:
    """인덱스 기반으로 순서를 보장하는 오디오 큐

    병렬 합성 시 두 번째 문장이 첫 번째보다 먼저 완성될 수 있음 —
    PriorityQueue로 순서를 보장해서 재생 순서를 지킨다.
    """

    def __init__(self):
        self._heap:       list[tuple[int, AudioChunk]] = []
        self._next_index: int          = 0
        self._lock:       asyncio.Lock  = asyncio.Lock()
        self._event:      asyncio.Event = asyncio.Event()
        self._done:       bool          = False

    async def put(self, chunk: AudioChunk) -> None:
        async with self._lock:
            heapq.heappush(self._heap, (chunk.index, chunk))
            self._event.set()

    async def close(self) -> None:
        async with self._lock:
            self._done = True
            self._event.set()

    async def __aiter__(self):
        while True:
            await self._event.wait()
            async with self._lock:
                if self._heap and self._heap[0][0] == self._next_index:
                    _, chunk = heapq.heappop(self._heap)
                    self._next_index += 1
                    if not self._heap:
                        self._event.clear()
                    yield chunk
                elif self._done and not self._heap:
                    return
                else:
                    self._event.clear()


class TTSOrchestrator:
    """LLM 스트리밍 → TTS 합성 → 오디오 큐 통합 관리"""

    def __init__(
        self,
        tts:       TTSBackend,
        interrupt: InterruptController,
        rvc:       RVCBackend | None = None,
    ):
        self._tts       = tts
        self._interrupt = interrupt
        self._rvc       = rvc

    async def process_llm_stream(
        self,
        token_stream,                          # AsyncIterator[str]
        emotion: EmotionState = EmotionState.CALM,
        queue:   OrderedAudioQueue | None = None,
    ) -> None:
        """LLM 토큰 스트림을 받아 TTS 합성하고 큐에 적재"""
        if queue is None:
            queue = OrderedAudioQueue()

        splitter   = SentenceSplitter()
        tts_params = get_tts_params(emotion)
        tasks: list[asyncio.Task] = []
        chunk_idx = 0

        async for token in token_stream:
            if self._interrupt.is_cancelled:
                for t in tasks:
                    t.cancel()
                break

            sentences = splitter.feed(token)
            for sentence in sentences:
                idx  = chunk_idx
                chunk_idx += 1
                task = asyncio.create_task(
                    self._synth_and_enqueue(sentence, idx, tts_params, queue)
                )
                tasks.append(task)

        # 스트림 종료 후 버퍼 플러시
        if not self._interrupt.is_cancelled:
            remaining = splitter.flush()
            if remaining:
                await self._synth_and_enqueue(
                    remaining, chunk_idx, tts_params, queue
                )

        # 모든 합성 태스크 완료 대기
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)

        await queue.close()

    async def _synth_and_enqueue(
        self,
        text:      str,
        index:     int,
        params:    TTSParams,
        queue:     OrderedAudioQueue,
    ) -> None:
        """단일 문장을 합성하고 큐에 적재"""
        if self._interrupt.is_cancelled:
            return

        audio = await self._tts.synthesize_async(text, speed=params.speed)

        # RVC 후처리 (선택)
        if self._rvc is not None and not self._interrupt.is_cancelled:
            audio = await self._rvc.convert_async(
                audio, self._tts.SAMPLE_RATE
            )
            sample_rate = self._rvc._sample_rate
        else:
            sample_rate = self._tts.SAMPLE_RATE

        # 볼륨 조절
        audio = audio * params.volume

        if not self._interrupt.is_cancelled:
            await queue.put(
                AudioChunk(
                    audio=audio, text=text,
                    index=index, sample_rate=sample_rate,
                )
            )

    @staticmethod
    def to_wav_bytes(audio: np.ndarray, sample_rate: int) -> bytes:
        buf = io.BytesIO()
        sf.write(buf, audio, sample_rate, format="WAV", subtype="PCM_16")
        return buf.getvalue()

The two key properties of this design are ordering guarantees and interrupt safety.

Multiple sentences are synthesized in parallel via asyncio.create_task(), but OrderedAudioQueue enforces playback order using index-based priority.
When an interrupt is triggered, both in-flight synthesis and queue insertion stop immediately.
The queue.close() signal lets the consumer detect end-of-stream.

Audio Fade Processing — Eliminating Chunk Seams

Concatenating raw TTS output can introduce pop noise at chunk boundaries due to mismatched audio levels. Short fade-in/out envelopes fix this.

# src/vtuber/tts/audio_utils.py
import numpy as np

FADE_MS = 10  # 페이드 길이 (밀리초)


def apply_fade(audio: np.ndarray, sample_rate: int) -> np.ndarray:
    """청크 시작/끝에 짧은 페이드 적용"""
    fade_samples = int(sample_rate * FADE_MS / 1000)
    if len(audio) < fade_samples * 2:
        return audio

    result               = audio.copy()
    result[:fade_samples]  *= np.linspace(0, 1, fade_samples)
    result[-fade_samples:] *= np.linspace(1, 0, fade_samples)
    return result

Streaming Audio over WebSocket

This is the WebSocket endpoint that sends synthesized audio to the browser. It extends the /voice/ws endpoint built in part 4, delivering TTS output over the same channel.

# src/vtuber/api/voice.py (TTS 연동 추가)
import asyncio
import base64
import json
import uuid

import numpy as np
from fastapi import APIRouter, WebSocket, WebSocketDisconnect

from ..character.state import CharacterState, EmotionState
from ..llm.base import ChatMessage, LLMBackend
from ..memory.manager import MemoryManager
from ..character.prompt import PromptBuilder
from ..stt.pipeline import STTPipeline
from ..stt.interrupt import InterruptController
from ..tts.orchestrator import TTSOrchestrator, OrderedAudioQueue
from ..tts.audio_utils import apply_fade
from ..tts.emotion_mapper import get_tts_params

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


async def stream_tts_to_client(
    ws:           WebSocket,
    orchestrator: TTSOrchestrator,
    llm_stream,
    emotion:      EmotionState,
) -> None:
    """LLM streaming → TTS → WebSocket delivery"""
    queue = OrderedAudioQueue()

    # Start TTS synthesis as a background task
    synth_task = asyncio.create_task(
        orchestrator.process_llm_stream(llm_stream, emotion, queue)
    )

    await ws.send_text(json.dumps({"type": "tts_start"}, ensure_ascii=False))

    async for chunk in queue:
        if orchestrator._interrupt.is_cancelled:
            break

        audio     = apply_fade(chunk.audio, chunk.sample_rate)
        wav_bytes = orchestrator.to_wav_bytes(audio, chunk.sample_rate)

        await ws.send_text(json.dumps({
            "type":        "audio_chunk",
            "index":       chunk.index,
            "text":        chunk.text,
            "audio_b64":   base64.b64encode(wav_bytes).decode(),
            "sample_rate": chunk.sample_rate,
        }, ensure_ascii=False))

    await synth_task
    await ws.send_text(json.dumps({"type": "tts_end"}, ensure_ascii=False))


@router.websocket("/ws")
async def voice_websocket(
    websocket:      WebSocket,
    stt:            STTPipeline,
    llm:            LLMBackend,
    orchestrator:   TTSOrchestrator,
    prompt_builder: PromptBuilder,
    memory_manager: MemoryManager,
):
    await websocket.accept()
    session_id = str(uuid.uuid4())
    state      = CharacterState()
    history:   list[ChatMessage] = []

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

            if "bytes" in data:
                event = await stt.process_chunk(data["bytes"])
                if event is None:
                    continue

                if event.type == "interrupt":
                    orchestrator._interrupt.trigger()
                    await websocket.send_text(
                        json.dumps({"type": "interrupted"}, ensure_ascii=False)
                    )

                elif event.type == "transcript" and event.result:
                    text    = event.result.text
                    user_id = f"haru_{session_id}"

                    # Memory retrieval (part 3)
                    memory_ctx = await memory_manager.retrieve_context(user_id, text)

                    # System prompt construction (part 2)
                    system_prompt = prompt_builder.build_system_prompt(state)
                    mem_section   = memory_ctx.to_prompt_section()
                    if mem_section:
                        system_prompt += "\n\n" + mem_section

                    messages = [
                        ChatMessage(role="system", content=system_prompt),
                        *history[-10:],
                        ChatMessage(role="user", content=text),
                    ]

                    # Reset interrupt, then start LLM + TTS streaming
                    orchestrator._interrupt.reset()
                    stt.set_ai_speaking(True)

                    await websocket.send_text(
                        json.dumps({"type": "ai_speaking_start"}, ensure_ascii=False)
                    )

                    llm_stream = llm.chat_stream(messages)
                    await stream_tts_to_client(
                        websocket, orchestrator, llm_stream, state.emotion
                    )

                    stt.set_ai_speaking(False)
                    await websocket.send_text(
                        json.dumps({"type": "ai_speaking_end"}, ensure_ascii=False)
                    )

            elif "text" in data:
                msg = json.loads(data["text"])
                # Frontend can directly override the emotion state
                if msg.get("type") == "set_emotion":
                    try:
                        state.emotion = EmotionState(msg["emotion"])
                    except ValueError:
                        pass

    except WebSocketDisconnect:
        pass

Audio is Base64-encoded and sent as JSON text frames. Mixing binary and text frames complicates demultiplexing on the browser side, so everything goes out as text. Sentence-level chunks are typically 10–50 KB, so the encoding overhead stays within a practical range.

Browser Audio Playback

// src/frontend/audio/player.ts
export class AudioPlayer {
  private context:   AudioContext;
  private nextStart: number = 0;  // Scheduled start time for the next chunk

  constructor() {
    this.context = new AudioContext();
  }

  async playChunk(base64wav: string): Promise<void> {
    const bytes   = Uint8Array.from(atob(base64wav), c => c.charCodeAt(0));
    const decoded = await this.context.decodeAudioData(bytes.buffer);

    const source = this.context.createBufferSource();
    source.buffer = decoded;
    source.connect(this.context.destination);

    const now     = this.context.currentTime;
    const startAt = Math.max(now, this.nextStart);
    source.start(startAt);
    this.nextStart = startAt + decoded.duration;
  }

  stop(): void {
    // Immediately stop all playing audio
    this.context.close();
    this.context   = new AudioContext();
    this.nextStart = 0;
  }
}

nextStart is the key. Each chunk's playback start time is scheduled to begin exactly when the previous chunk ends. Because this relies on AudioContext.currentTime-based scheduling rather than JavaScript's setTimeout, timing is far more precise. Chunks stitch together seamlessly without any buffering gaps.

Measured Latency and Optimizations

Actual measurements of end-to-end performance:

Environment: CPU only (Apple M2 Pro), XTTS-v2, faster-whisper small

Stage	Time	Notes
VAD + end-of-speech detection	~600ms	Based on SILENCE_TIMEOUT_MS
faster-whisper STT	~400ms	3-second utterance
Qdrant memory retrieval	~50ms	Local mode
LLM first token	~300ms	Mistral 7B, Ollama
First sentence complete	~800ms	10–20 tokens
XTTS-v2 first sentence synthesis	~1,200ms	CPU, ~15-character sentence
End-to-end (first utterance)	~2.5–3.5s

On CPU, XTTS-v2 synthesis is the bottleneck. On GPU (RTX 3080), synthesis drops to 200–400ms, bringing end-to-end latency down to around 1.5 seconds.

Optimization 1: Model Warm-up

# src/vtuber/main.py lifespan에 추가
async def lifespan(app: FastAPI):
    dummy = np.zeros(24000, dtype=np.float32)  # 1초 침묵
    await tts_backend.synthesize_async("안녕하세요, 하루예요.")
    print("✅ XTTS-v2 워밍업 완료")
    yield

Optimization 2: Short-response Fallback

# Short sentences go to the faster Edge TTS (no voice customization)
FAST_TTS_THRESHOLD = 15  # character count

async def select_backend(text: str) -> TTSBackend:
    if len(text) < FAST_TTS_THRESHOLD:
        return edge_tts_backend   # instant, but no custom voice
    return xtts_backend           # slower, but character voice

Optimization 3: Pre-synthesize the Next Chunk While Streaming

The architecture already covers this optimization: using asyncio.create_task(), synthesis starts as soon as each sentence is segmented. While the LLM generates the second sentence, the first is being synthesized; while the first plays back, the second is being synthesized.

Troubleshooting — Common Issues

Issue 1: XTTS-v2 Korean Prosody Sounds Unnatural

Try providing multiple reference audio clips. The speaker embeddings are averaged together, producing more stable results.

self._tts.tts(
    text=text,
    speaker_wav=["ref1.wav", "ref2.wav", "ref3.wav"],
    language="ko",
)

Also, Korean prosody reproduction is significantly better when the reference audio comes from natural conversational speech rather than flat, reading-style recordings.

Issue 2: Sentences Get Cut Off Mid-way

Korean sentence-final endings can inadvertently trigger the SENTENCE_END pattern. Increase MIN_CHUNK_LEN, or tighten the pattern by requiring whitespace after the punctuation.

# Stricter termination condition: punctuation must be followed by whitespace
SENTENCE_END = re.compile(r'(?<=[.!?。！？])\s+')

Issue 3: Pop Noise Between Audio Chunks

Apply apply_fade() to every chunk. A 10ms fade-in/fade-out eliminates most pop artifacts.

Issue 4: Voice Sounds Robotic After RVC Conversion

Set the f0 extraction method to rmvpe and confirm that protect=0.33 is the default. Lowering index_rate to 0.5 or below often improves naturalness.

RVCConfig(
    model_path="haru_rvc.pth",
    f0_method="rmvpe",
    index_rate=0.4,    # lowered from 0.75 → 0.4
    protect=0.33,
)

Updated Project Structure

These are the modules added going from Part 4 to Part 5.

vtuber-assistant/
├── src/
│   └── vtuber/
│       ├── tts/                    ← Added in Part 5
│       │   ├── base.py             # TTSBackend abstraction
│       │   ├── xtts.py             # XTTS-v2 implementation
│       │   ├── gptsovits.py        # GPT-SoVITS API client
│       │   ├── rvc.py              # RVC post-processing
│       │   ├── splitter.py         # SentenceSplitter
│       │   ├── emotion_mapper.py   # Emotion → TTS parameters
│       │   ├── audio_utils.py      # Fade processing utilities
│       │   └── orchestrator.py     # TTSOrchestrator (unified)
│       ├── stt/                    ← Part 4
│       ├── character/              ← Part 2
│       ├── llm/                    ← Part 2
│       ├── memory/                 ← Part 3
│       └── api/
│           └── voice.py            ← TTS integration update
├── assets/
│   └── haru/
│       ├── reference.wav           ← XTTS-v2 reference audio
│       └── rvc_model.pth           ← RVC model (optional)

Additional Dependencies

# Add to pyproject.toml dependencies
"TTS>=0.22.0",
"soundfile>=0.12.1",
"resampy>=0.4.3",
# For RVC (optional)
"rvc-python>=0.2.4",

Wrap-Up

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

XTTS-v2 backend: Zero-shot voice cloning, ready to use with a 3-second reference clip
GPT-SoVITS backend: API server integration for fine-tuned character voices
RVC post-processing: An optional layer that converts TTS output into a target voice
TTSBackend abstraction: An interface that lets you swap backends without changing any other code
SentenceSplitter: Splits LLM streaming tokens into sentence units to minimize latency
Emotion → TTS parameters: Dynamically adjusts speed and pitch_key based on emotional state
OrderedAudioQueue: A priority queue that handles parallel synthesis while preserving playback order
TTSOrchestrator: Integrates the full pipeline — interrupt-safe and fully async
Browser AudioContext playback: Precise scheduling with seamless transitions between chunks

Haru can now speak. Her vocal tone shifts with her emotions, she starts talking before the LLM has finished generating a response, and she stops immediately when interrupted.

One thing is still missing: Haru's body. Audio without anything on screen isn't a VTuber. Part 6 covers rendering a Live2D avatar in the browser, animating her mouth while she speaks, and building an expression system that reacts to her emotional state.

Coming Up Next

Part 6: VTuber Animation — A Living Avatar

While the response audio played through the speakers, Haru's mouth moved and her eyes blinked on screen.

Topics covered:

Live2D Cubism Web SDK: Fundamentals of rendering a Live2D model in the browser
Lip Sync: Analyzing the TTS audio waveform to drive mouth-shape parameters in real time
Emotion → Expression mapping: Automatically converting LLM emotion tags into Live2D Expression parameters
Automatic motion system: Idle animations, eye blinking, and breathing effects
Real-time WebSocket sync: Precise synchronization between audio playback timing and animation