**관련 시리즈** - [AI VTuber 에이전트 구축기 1편: 개요](https://hrletsgo.me/documents/vtuber-ai-assistant-series-1-introduction) - [AI VTuber 에이전트 구축기 2편: LLM과 페르소나](https://hrletsgo.me/documents/vtuber-ai-assistant-series-2-llm-persona) - [AI VTuber 에이전트 구축기 3편: 메모리](https://hrletsgo.me/documents/vtuber-ai-assistant-series-3-memory-system) - [AI VTuber 에이전트 구축기 4편: STT](https://hrletsgo.me/documents/vtuber-ai-assistant-series-4-stt) - AI VTuber 에이전트 구축기 5편: TTS **(현재글)** - [AI VTuber 에이전트 구축기 6편: Live2D](https://hrletsgo.me/documents/vtuber-ai-assistant-series-6-live2d) - [AI VTuber 에이전트 구축기 7편: 방송 연동](https://hrletsgo.me/documents/vtuber-ai-assistant-series-7-streaming) - [AI VTuber 에이전트 구축기 8편: 운영](https://hrletsgo.me/documents/vtuber-ai-assistant-series-8-production) --- ## 들어가며 4편에서 하루는 귀를 얻었다. Silero VAD로 음성 구간을 감지하고, faster-whisper로 텍스트를 뽑아내고, 인터럽트 컨트롤러로 대화 흐름을 제어하는 STT 파이프라인이 완성됐다. 이제 반대 방향을 뚫어야 한다. **하루가 말을 하는** 방법. TTS(Text-to-Speech)는 겉보기엔 단순하다. 텍스트 넣으면 오디오 나오면 끝 아닌가? 실제로는 이렇게 생각했다가 세 가지 현실의 벽에 부딪힌다. **첫째, 목소리가 형편없다.** Edge TTS 같은 무료 TTS로 만든 목소리로는 캐릭터감이 전혀 안 산다. 억양이 기계적이고, 감정이 없고, 무엇보다 "하루"의 목소리가 아니다. **둘째, 레이턴시가 너무 길다.** LLM이 500토큰짜리 응답을 다 생성하고 나서 TTS에 넘기면, 사용자는 3~4초를 침묵 속에 기다려야 한다. 그 사이에 대화의 흐름이 완전히 끊긴다. **셋째, 감정이 없다.** 2편에서 하루의 감정 상태를 열심히 설계했다. 그런데 happy든 nervous든 TTS는 같은 목소리로 말한다. 캐릭터를 만들어놓고 목소리에는 캐릭터가 없는 모순이다. 5편에서 이 세 가지를 전부 해결한다. --- ## TTS 파이프라인 전체 구조 코드를 쓰기 전에 데이터 흐름부터 그린다. ``` LLM 스트리밍 토큰 스트림 ↓ 문장 분리기 (SentenceSplitter) ↓ 문장 단위 청크 TTS 합성 (XTTS-v2 / GPT-SoVITS) ↓ PCM 오디오 [RVC 변환 — 선택] ↓ 오디오 청크 큐 (asyncio.Queue) ↓ WebSocket → 브라우저 AudioContext 재생 ``` 핵심 설계 원칙이 하나 있다. **LLM 응답 전체를 기다리지 않는다.** 첫 문장이 완성되는 순간 TTS 합성을 시작하고, 두 번째 문장을 합성하는 동안 첫 번째 문장의 오디오를 재생한다. 파이프라이닝이다. ``` 나쁜 방식: LLM [--500토큰 전체 생성--] → TTS [--전체 합성--] → 재생 시작 레이턴시: 3~5초 좋은 방식: LLM [--첫 문장--] → TTS [합성] → 재생 시작 [--두 번째 문장--] → TTS [합성] → 큐 대기 [--세 번째 문장--] → TTS [합성] → 큐 대기 레이턴시: 0.5~1.5초 (첫 발화까지) ``` "첫 발화까지의 시간"을 LLM 전체 생성 시간이 아니라 **첫 문장 생성 시간 + TTS 합성 시간**으로 줄인다. 이것이 이 구조의 핵심이다. --- ## TTS 백엔드 선택 — 세 가지 선택지 이 시리즈에서 TTS 백엔드 옵션은 세 가지다. | 모델 | 방식 | 한국어 | 클로닝 | 속도 | 품질 | |------|------|--------|--------|------|------| | **XTTS-v2** | Zero-shot 클로닝 | ✅ 우수 | 3초 샘플 | ⚡⚡ 보통 | ✅✅ 우수 | | **GPT-SoVITS** | Few-shot 학습 | ✅✅ 매우 우수 | 수십 초 이상 | ⚡⚡ 보통 | ✅✅✅ 최고 | | **RVC** | 음성 변환 (후처리) | 원본 TTS 의존 | 변환 방식 | ⚡⚡⚡ 빠름 | ✅✅ 우수 | 이 시리즈에서 채택하는 전략은 **계층형 구성**이다. ``` 기본 레이어: XTTS-v2 → 3초짜리 기준 음성 샘플만 있으면 즉시 사용 가능 → zero-shot이라 별도 학습 불필요 고품질 레이어: GPT-SoVITS (선택) → 캐릭터 전용 목소리 데이터를 충분히 모은 경우 → 더 자연스럽고 캐릭터다운 억양 변환 레이어: RVC (선택) → 두 모델의 출력을 다른 목소리로 변환 → 특정 성우 목소리를 재현하고 싶을 때 ``` --- ## XTTS-v2 — Zero-shot 음성 클로닝 XTTS(Cross-Lingual Text-to-Speech) v2는 Coqui TTS 팀이 개발한 모델로, **3~6초짜리 음성 샘플만으로** 해당 화자의 목소리를 재현한다. 17개 언어를 지원하고 한국어도 포함된다. 별도 학습 없이 inference 시점에 음성 샘플을 레퍼런스로 넣는 방식이라 "zero-shot"이다. ### 모델 구조 XTTS-v2는 세 단계로 동작한다. ``` 입력 텍스트 + 레퍼런스 오디오 ↓ 1. GPT 기반 언어 모델 (텍스트 → 코드북 토큰) ↓ + 스피커 임베딩 (레퍼런스 오디오에서 추출) 2. 코드북 토큰 → 멜 스펙트로그램 ↓ 3. HiFi-GAN 보코더 → PCM 오디오 ``` 레퍼런스 오디오에서 스피커 임베딩을 추출해서 GPT 디코딩 단계에 컨디셔닝하는 방식이다. 이 임베딩이 "이 사람의 목소리 특성"을 담는다. ### 설치와 구현 ```bash uv add TTS soundfile resampy ``` ```python # src/vtuber/tts/xtts.py import asyncio import io import numpy as np import soundfile as sf from TTS.api import TTS class XTTSv2Backend: """XTTS-v2 기반 zero-shot 음성 클로닝 TTS""" SAMPLE_RATE = 24000 # XTTS-v2 출력 샘플레이트 def __init__( self, device: str = "cpu", # "cpu" | "cuda" speaker_wav: str = "", # 레퍼런스 음성 파일 경로 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: """텍스트를 오디오 배열로 변환 (동기)""" 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: """비동기 래퍼 — CPU 추론을 스레드 풀에서 실행""" 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()`로 동기 추론을 비동기로 감싼다. 이걸 빠뜨리면 TTS 추론 중에 이벤트 루프 전체가 멈춘다. STT 편에서도 같은 이유로 적용한 패턴이다. ### 레퍼런스 음성 준비 XTTS-v2 품질은 레퍼런스 음성에 크게 의존한다. 좋은 레퍼런스 음성의 조건: ``` ✅ 길이: 3~12초 (짧으면 특성이 부족, 너무 길면 오히려 노이즈) ✅ 품질: 배경 소음 없는 클리어한 음성 ✅ 내용: 다양한 억양이 포함된 문장 (단조로운 낭독보다 자연스러운 대화 선호) ✅ 포맷: 22050Hz 이상 WAV ❌ 피해야 할 것: 배경 음악, 에코, 압축 아티팩트 ``` ```python # 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: """레퍼런스 음성을 XTTS-v2에 최적화된 형태로 전처리""" audio, sr = sf.read(input_path) # 모노 변환 if audio.ndim > 1: audio = audio.mean(axis=1) # 리샘플링 if sr != target_sample_rate: import resampy audio = resampy.resample(audio, sr, target_sample_rate) sr = target_sample_rate # 목표 길이로 자르기 target_samples = int(target_duration_s * sr) if len(audio) > target_samples: audio = audio[:target_samples] elif len(audio) < sr * 2: print(f"⚠️ 레퍼런스 음성이 너무 짧음: {len(audio)/sr:.1f}초") # 피크 정규화 peak = np.abs(audio).max() if peak > 0: audio = audio / peak * 0.95 sf.write(output_path, audio, sr, subtype="PCM_16") print(f"✅ 저장: {output_path} ({len(audio)/sr:.1f}초, {sr}Hz)") ``` 레퍼런스를 여러 개 제공하면 스피커 임베딩이 평균화되어 더 안정적인 결과가 나온다. XTTS-v2는 `speaker_wav`에 리스트를 받을 수 있다. ```python # 레퍼런스 복수 제공 self._tts.tts( text=text, speaker_wav=["ref1.wav", "ref2.wav", "ref3.wav"], language="ko", ) ``` --- ## GPT-SoVITS — Few-shot 캐릭터 목소리 XTTS-v2가 "즉시 쓸 수 있는 zero-shot"이라면, GPT-SoVITS는 "학습을 통해 훨씬 자연스러운 결과"를 내는 모델이다. > **GitHub**: https://github.com/RVC-Boss/GPT-SoVITS > ⭐ 38,000+ | Python | 2024년 1월 시작 GPT-SoVITS는 두 단계 구조다. ``` 1. GPT 모델 (텍스트 → 시맨틱 토큰) - 입력: 텍스트 + 레퍼런스 오디오의 시맨틱 토큰 - 출력: 타겟 발화의 시맨틱 토큰 시퀀스 2. SoVITS 모델 (시맨틱 토큰 → 오디오) - 입력: 시맨틱 토큰 + 스피커 임베딩 - 출력: 고품질 오디오 파형 ``` XTTS-v2와의 가장 큰 차이는 **fine-tuning이 가능하다**는 점이다. 캐릭터의 목소리 데이터를 모아서 GPT-SoVITS를 fine-tune하면, zero-shot으로는 불가능한 수준의 자연스러운 억양과 감정 표현이 가능해진다. ### 필요한 학습 데이터 ``` 최소: 30초 이상의 클리어한 음성 데이터 권장: 5~30분 최적: 1시간 이상 (공개 성우 영상 등 활용) ``` VTuber 캐릭터라면 과거 방송 클립에서 추출하거나 직접 녹음한 데이터를 사용한다. **배경 소음 없이 목소리만 분리**되어야 하기 때문에, `demucs` 같은 음원 분리 모델을 먼저 거치는 것이 일반적이다. ### API 방식 통합 GPT-SoVITS는 자체 API 서버를 제공한다. 우리 FastAPI 서버에서 HTTP로 호출하는 방식으로 분리 실행한다. ```bash # GPT-SoVITS API 서버 (별도 프로세스) cd /path/to/GPT-SoVITS python api_v2.py --port 9880 ``` ```python # src/vtuber/tts/gptsovits.py import asyncio import io import httpx import numpy as np import soundfile as sf class GPTSoVITSBackend: """GPT-SoVITS API 서버와 통신하는 TTS 백엔드""" SAMPLE_RATE = 32000 # GPT-SoVITS 기본 출력 def __init__( self, base_url: str = "http://localhost:9880", ref_audio_path: str = "", ref_text: str = "", # 레퍼런스 오디오의 텍스트 (정확도 향상) 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 실전 비교 | 항목 | XTTS-v2 | GPT-SoVITS (fine-tuned) | |------|---------|------------------------| | 한국어 억양 자연스러움 | ✅ 양호 | ✅✅ 우수 | | 캐릭터 목소리 재현도 | △ 3초 샘플 한계 | ✅✅ 높음 | | 준비 시간 | 즉시 사용 | 학습 수 시간 | | 감정 표현 폭 | △ 제한적 | ✅ 자연스러움 | | GPU VRAM 요구 | ~4GB | ~8GB | | CPU 실행 | ✅ (느림) | △ 매우 느림 | **실용적 결론**: 빠르게 프로토타입을 만들 때는 XTTS-v2, 캐릭터 완성도를 높이고 싶을 때는 GPT-SoVITS fine-tuning이다. 이 시리즈의 구현은 두 백엔드를 모두 지원하는 추상화를 쓴다. --- ## TTS 백엔드 추상화 LLM 백엔드를 교체 가능하게 만든 것처럼, TTS 백엔드도 동일하게 추상화한다. ```python # src/vtuber/tts/base.py from abc import ABC, abstractmethod import numpy as np class TTSBackend(ABC): """모든 TTS 백엔드가 구현해야 하는 인터페이스""" SAMPLE_RATE: int = 24000 # 각 구현에서 오버라이드 @abstractmethod async def synthesize_async( self, text: str, speed: float = 1.0, ) -> np.ndarray: ... @abstractmethod async def health_check(self) -> bool: ... ``` --- ## RVC — 목소리를 바꾸는 후처리 레이어 RVC(Retrieval-based Voice Conversion)는 TTS 출력을 다른 목소리로 변환하는 기술이다. 직접 생성이 아니라 **변환**이다. 원하는 목소리의 학습 데이터를 모아서 RVC 모델을 학습시키면, 어떤 TTS 출력이든 그 목소리로 바꿔준다. VTuber 세계에서 RVC가 특별한 위치를 차지하는 이유는, TTS의 억양은 유지하면서 목소리 특성(음색, 배음 구조)만 교체하는 게 가능하기 때문이다. 특정 성우나 캐릭터의 목소리를 재현하는 데 XTTS나 GPT-SoVITS보다 더 나은 경우가 있다. ### RVC 파이프라인 ``` XTTS-v2 / GPT-SoVITS 출력 ↓ PCM 오디오 RVC (f0 추출 + 스피커 특성 변환) ↓ 목표 화자의 목소리로 변환된 오디오 ``` ```bash uv add rvc-python ``` ```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 # 반음 단위 피치 변환 index_rate: float = 0.75 # 인덱스 검색 비율 (0~1) filter_radius: int = 3 rms_mix_rate: float = 0.25 protect: float = 0.33 # 무성음 보호 (0.5 미만 권장) class RVCBackend: """RVC 기반 음성 변환 후처리""" def __init__(self, config: RVCConfig): self._config = config self._sample_rate = 40000 # RVC 기본 출력 self._vc = None def _lazy_load(self) -> None: """모델을 첫 사용 시 로드 — 시작 시간 단축""" 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: """오디오를 RVC로 변환 (동기)""" 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) ``` RVC는 f0(기본 주파수) 추출 방식이 품질을 좌우한다. `rmvpe`가 현재 가장 정확하고 안정적인 방식이다. `crepe`는 GPU에서 더 빠르지만 설치가 복잡하고, `pm`은 가볍지만 정확도가 떨어진다. --- ## 감정 태그 → TTS 파라미터 매핑 2편에서 LLM 응답에 `[EMOTION:happy]` 태그를 붙이도록 설계했다. 이것이 TTS에서 실제로 쓰이는 순간이다. 감정에 따라 TTS의 속도(speed)와 RVC의 피치(f0_up_key)를 조절한다. 완전한 감정 TTS를 구현하는 건 훨씬 복잡하지만, 이 두 파라미터만으로도 체감 차이가 상당하다. ```python # src/vtuber/tts/emotion_mapper.py from dataclasses import dataclass from ..character.state import EmotionState @dataclass class TTSParams: speed: float = 1.0 # 1.0 기준, 높을수록 빠름 pitch_key: int = 0 # 반음 단위 (RVC f0_up_key) volume: float = 1.0 # 볼륨 배율 # 감정 → TTS 파라미터 매핑 테이블 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()) ``` 숫자들은 절대적 정답이 아니다. 캐릭터마다, 기반 TTS 모델마다 최적값이 다르다. 위 값들은 출발점이고, 실제로 들어보면서 조정해야 한다. --- ## 문장 분리기 — 스트리밍의 핵심 LLM 스트리밍 토큰을 받아서 "TTS에 넘길 수 있는 완전한 문장"으로 조립하는 것이 이 파이프라인의 핵심이다. 너무 일찍 자르면 어색하게 끊기고, 너무 늦게 자르면 레이턴시가 늘어난다. ```python # 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`은 "안녕하세요" 같은 짧은 문장도 TTS로 보내기 위한 최솟값이다. 이보다 짧은 건 대부분 감탄사나 단절된 단어라 다음 청크와 합치는 게 낫다. --- ## TTSOrchestrator — 전체 파이프라인 통합 모든 컴포넌트를 연결해서, LLM 스트리밍 → 문장 분리 → TTS 합성 → 오디오 큐까지 통합하는 오케스트레이터다. ```python # 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() ``` 이 설계의 핵심은 **순서 보장과 인터럽트 안전성**이다. - `asyncio.create_task()`로 여러 문장을 병렬 합성하지만, `OrderedAudioQueue`가 인덱스 기반으로 순서를 보장한다 - 인터럽트가 트리거되면 진행 중인 합성과 큐 적재가 모두 멈춘다 - `queue.close()` 신호로 소비자가 스트리밍 종료를 감지한다 --- ## 오디오 페이드 처리 — 청크 간 이음새 제거 TTS 출력을 그대로 붙이면 각 청크 사이의 오디오 레벨 불일치로 팝 노이즈가 발생할 수 있다. 짧은 페이드인/아웃으로 해결한다. ```python # 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 ``` --- ## WebSocket으로 오디오 스트리밍 합성된 오디오를 브라우저로 전송하는 WebSocket 엔드포인트다. 4편에서 만든 `/voice/ws`를 확장해서 TTS 출력도 같은 채널로 내려보낸다. ```python # 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 스트리밍 → TTS → WebSocket 전송""" queue = OrderedAudioQueue() # TTS 합성을 백그라운드 태스크로 시작 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}" # 메모리 검색 (3편) memory_ctx = await memory_manager.retrieve_context(user_id, text) # 시스템 프롬프트 구성 (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), ] # 인터럽트 초기화 후 LLM + TTS 스트리밍 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"]) # 프론트엔드에서 직접 감정 오버라이드 가능 if msg.get("type") == "set_emotion": try: state.emotion = EmotionState(msg["emotion"]) except ValueError: pass except WebSocketDisconnect: pass ``` 오디오를 Base64로 인코딩해서 JSON 텍스트 프레임으로 보낸다. 이진 프레임과 텍스트 프레임을 섞으면 브라우저 측에서 구분이 복잡해지기 때문에 텍스트로 통일한다. 문장 단위 청크는 보통 10~50KB 수준이라 오버헤드는 실용 범위 내다. ### 브라우저 오디오 재생 ```typescript // src/frontend/audio/player.ts export class AudioPlayer { private context: AudioContext; private nextStart: number = 0; // 다음 청크의 재생 예약 시간 constructor() { this.context = new AudioContext(); } async playChunk(base64wav: string): Promise { 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 { // 재생 중인 오디오 즉시 중단 this.context.close(); this.context = new AudioContext(); this.nextStart = 0; } } ``` `nextStart`가 핵심이다. 각 청크의 재생 시작 시간을 이전 청크의 종료 시간 이후로 예약한다. `AudioContext.currentTime` 기반 스케줄링이라 JavaScript의 `setTimeout`보다 훨씬 정밀하다. 버퍼링 없이 청크들이 이음새 없이 연결된다. --- ## 레이턴시 실측과 최적화 실제로 얼마나 빠른지 측정한 결과다. ``` 환경: CPU only (Apple M2 Pro), XTTS-v2, faster-whisper small ``` | 단계 | 소요 시간 | 비고 | |------|---------|------| | VAD + 발화 종료 감지 | ~600ms | SILENCE_TIMEOUT_MS 기준 | | faster-whisper STT | ~400ms | 3초 발화 기준 | | Qdrant 메모리 검색 | ~50ms | 로컬 모드 | | LLM 첫 토큰 | ~300ms | Mistral 7B, Ollama | | 첫 문장 완성 | ~800ms | 10~20토큰 기준 | | XTTS-v2 첫 문장 합성 | ~1,200ms | CPU, ~15자 문장 | | **end-to-end (첫 발화)** | **~2.5~3.5초** | | CPU에서는 XTTS-v2 합성이 병목이다. GPU(RTX 3080 기준)에서는 합성이 200~400ms로 줄어들어 end-to-end가 1.5초 내외까지 떨어진다. ### 최적화 1: 모델 워밍업 ```python # 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 ``` ### 최적화 2: 짧은 반응어 폴백 ```python # 짧은 문장은 빠른 Edge TTS (커스텀 불가)로 처리 FAST_TTS_THRESHOLD = 15 # 자 수 기준 async def select_backend(text: str) -> TTSBackend: if len(text) < FAST_TTS_THRESHOLD: return edge_tts_backend # 즉각, 커스텀 불가 return xtts_backend # 느리지만 캐릭터 목소리 ``` ### 최적화 3: 스트리밍 동안 다음 청크 미리 합성 `asyncio.create_task()`로 문장이 분리되자마자 합성을 시작하는 구조가 이미 이 최적화를 포함하고 있다. LLM이 두 번째 문장을 생성하는 동안 첫 번째 문장이 합성되고, 첫 번째가 재생되는 동안 두 번째가 합성된다. --- ## 트러블슈팅 — 자주 만나는 문제들 ### 문제 1: XTTS-v2 한국어 억양이 어색하다 레퍼런스 음성을 여러 개 제공해보자. 스피커 임베딩이 평균화되어 더 안정적인 결과가 나온다. ```python self._tts.tts( text=text, speaker_wav=["ref1.wav", "ref2.wav", "ref3.wav"], language="ko", ) ``` 또한 레퍼런스 음성이 단조로운 낭독체보다 자연스러운 대화체에서 추출됐을 때 한국어 억양 재현이 훨씬 나아진다. ### 문제 2: 문장이 중간에 잘린다 한국어 어미 구조 때문에 SENTENCE_END 패턴이 의도치 않게 발동하는 경우가 있다. `MIN_CHUNK_LEN`을 늘리거나, 패턴에 공백 조건을 더 엄격하게 추가한다. ```python # 더 엄격한 종료 조건: 문장부호 + 공백 필수 SENTENCE_END = re.compile(r'(?<=[.!?。！？])\s+') ``` ### 문제 3: 오디오 청크 사이에 팝 노이즈가 들린다 `apply_fade()`를 모든 청크에 적용한다. 10ms 페이드인/아웃으로 대부분의 팝 노이즈가 제거된다. ### 문제 4: RVC 변환 후 목소리가 로봇처럼 들린다 f0 추출 방식을 `rmvpe`로 설정하고, `protect=0.33`이 기본값인지 확인한다. `index_rate`를 0.5 이하로 낮추면 자연스러움이 올라가는 경우가 있다. ```python RVCConfig( model_path="haru_rvc.pth", f0_method="rmvpe", index_rate=0.4, # 0.75 → 0.4로 낮춤 protect=0.33, ) ``` --- ## 업데이트된 프로젝트 구조 4편에서 5편으로 오면서 추가된 모듈들이다. ``` vtuber-assistant/ ├── src/ │ └── vtuber/ │ ├── tts/ ← 5편에서 추가 │ │ ├── base.py # TTSBackend 추상화 │ │ ├── xtts.py # XTTS-v2 구현 │ │ ├── gptsovits.py # GPT-SoVITS API 클라이언트 │ │ ├── rvc.py # RVC 후처리 │ │ ├── splitter.py # SentenceSplitter │ │ ├── emotion_mapper.py # 감정 → TTS 파라미터 │ │ ├── audio_utils.py # 페이드 처리 유틸 │ │ └── orchestrator.py # TTSOrchestrator (통합) │ ├── stt/ ← 4편 │ ├── character/ ← 2편 │ ├── llm/ ← 2편 │ ├── memory/ ← 3편 │ └── api/ │ └── voice.py ← TTS 연동 업데이트 ├── assets/ │ └── haru/ │ ├── reference.wav ← XTTS-v2 레퍼런스 음성 │ └── rvc_model.pth ← RVC 모델 (선택) ``` ### 추가 의존성 ```toml # pyproject.toml dependencies에 추가 "TTS>=0.22.0", "soundfile>=0.12.1", "resampy>=0.4.3", # RVC 사용 시 (선택) "rvc-python>=0.2.4", ``` --- ## 마무리 이번 편에서 구현한 것을 정리하면: 1. **XTTS-v2 백엔드**: zero-shot 음성 클로닝, 3초 레퍼런스로 즉시 사용 2. **GPT-SoVITS 백엔드**: API 서버 방식 통합, fine-tuning된 캐릭터 목소리 3. **RVC 후처리**: TTS 출력을 원하는 목소리로 변환하는 선택적 레이어 4. **TTSBackend 추상화**: 백엔드 교체가 코드 변경 없이 가능한 인터페이스 5. **SentenceSplitter**: LLM 스트리밍 토큰을 문장 단위로 분리, 레이턴시 최소화 6. **감정 → TTS 파라미터**: speed, pitch_key를 감정 상태에 따라 동적 조정 7. **OrderedAudioQueue**: 병렬 합성 + 순서 보장을 동시에 해결하는 우선순위 큐 8. **TTSOrchestrator**: 전체 파이프라인 통합, 인터럽트 안전, 비동기 처리 9. **브라우저 AudioContext 재생**: 청크 간 이음새 없는 정밀 스케줄링 이제 하루는 말할 수 있다. 감정에 따라 목소리 톤이 달라지고, LLM이 응답을 다 만들기 전에 말을 시작하고, 인터럽트되면 즉시 멈춘다. 한 가지가 아직 없다. 하루의 **몸**이다. 목소리는 나오는데 화면에 아무것도 없으면 VTuber가 아니다. 6편에서는 Live2D 아바타를 브라우저에 렌더링하고, 하루가 말할 때 입이 움직이고, 감정에 따라 표정이 바뀌는 애니메이션 시스템을 구축한다. --- ## 다음 편 예고 **6편: VTuber 애니메이션 — 살아있는 아바타** > _응답 오디오가 스피커에서 흘러나오는 동안, 화면 속 하루의 입이 움직이고 눈이 깜빡였다._ 다룰 내용: - **Live2D Cubism Web SDK**: 브라우저에서 Live2D 모델 렌더링 기초 - **립싱크 (Lip Sync)**: TTS 오디오 파형을 분석해서 입 모양 파라미터를 실시간으로 제어 - **감정 → 표정 매핑**: LLM 감정 태그 → Live2D Expression 파라미터 자동 변환 - **자동 모션 시스템**: 아이들 애니메이션, 눈 깜빡임, 호흡 효과 - **실시간 WebSocket 동기화**: 오디오 재생 타이밍과 애니메이션의 정밀한 동기화