Merge pull request #74 from Team-StackUp/feature/gemini-tts

i3months · web-flow · commit def5feea437b · 2026-06-03T13:16:23.000+09:00
feat(ai): 질문 TTS를 Gemini TTS로 — 한국어 음성 합성 (기존 GEMINI_API_KEY 재사용)
diff --git a/ai/.env.example b/ai/.env.example
@@ -63,8 +63,13 @@ WHISPER_MODEL=whisper-1
 WHISPER_LANGUAGE=ko
 WHISPER_TIMEOUT_SEC=60
 
-# TTS (질문 음성화). auto=OPENAI_API_KEY 보유 시 openai, 없으면 mock. (OPENAI_API_KEY 재사용)
+# TTS (질문 음성화). auto=GEMINI_API_KEY > OPENAI_API_KEY 순으로 선택, 둘 다 없으면 mock.
+# Deepgram/OpenAI TTS 는 한국어 미지원 → 기본은 Gemini TTS(GEMINI_API_KEY 재사용, raw PCM→WAV).
 TTS_PROVIDER=auto
 OPENAI_TTS_MODEL=gpt-4o-mini-tts
 OPENAI_TTS_VOICE=alloy
 OPENAI_TTS_TIMEOUT_SEC=30
+GEMINI_TTS_MODEL=gemini-2.5-flash-preview-tts
+GEMINI_TTS_VOICE=Kore
+GEMINI_TTS_BASE_URL=https://generativelanguage.googleapis.com/v1beta
+GEMINI_TTS_TIMEOUT_SEC=30
diff --git a/ai/CLAUDE.md b/ai/CLAUDE.md
@@ -190,7 +190,7 @@ chain = prompt | llm | PydanticOutputParser(pydantic_object=...)
   - 비용: $0.006 / 분 (1시간 면접 ≈ ₩500 / USD ≈ $0.36)
   - 셀프호스팅 옵션: `whisper.cpp` 또는 `faster-whisper` (GPU 권장, 비용 ↓ but 운영 부담 ↑)
   - 브라우저 내장 SpeechRecognition API는 정확도 부족으로 채택 안 함
-- **TTS: OpenAI TTS 채택** (`voice/tts/`) — 질문(INTERVIEWER) 메시지 음성화. `TtsProvider` 추상화 + `OpenAiTtsProvider`(`gpt-4o-mini-tts`, mp3)/`MockTtsProvider`, `build_tts_provider` factory(`TTS_PROVIDER=auto`면 OPENAI_API_KEY 보유 시 openai). `generate.tts` consumer 가 합성 → S3 PUT → `callback.tts` 발행.
+- **TTS: Gemini TTS 기본(한국어)** (`voice/tts/`) — 질문(INTERVIEWER) 메시지 음성화. Deepgram/OpenAI TTS 는 한국어 미지원이라 `GeminiTtsProvider`(`gemini-2.5-flash-preview-tts`, voice=Kore)가 기본. Gemini 는 raw PCM(L16/24kHz)을 반환하므로 WAV 로 감싸 `audio/wav` 로 저장. `TtsProvider` 추상화 + `GeminiTtsProvider`/`OpenAiTtsProvider`(`gpt-4o-mini-tts`, mp3)/`MockTtsProvider`, `build_tts_provider` factory(`TTS_PROVIDER=auto`면 GEMINI_API_KEY > OPENAI_API_KEY 순). `generate.tts` consumer 가 합성 → S3 PUT(확장자는 content_type 기준) → `callback.tts` 발행. 재생은 Core 오디오 프록시(`GET /api/sessions/{sid}/messages/{mid}/audio`) 경유(MinIO presigned URL 이 내부 호스트라 브라우저 직접 접근 불가).
 - **스트리밍 STT (실시간 음성 답변, RT3): Deepgram Live** (`voice/stt/deepgram_live.py`) — `websockets`로 Deepgram WS(`wss://api.deepgram.com/v1/listen`, nova-2)에 연결, interim/final 자막을 실시간 반환. `voice/stt/live.py`(`LiveSttProvider`/`LiveSttSession` 추상) + `voice/stt/mock_live.py`(키 없을 때 fallback) + `voice/stt/live_factory.py`(`LIVE_STT_PROVIDER=auto`면 DEEPGRAM_API_KEY 보유 시 deepgram_live).
   - FastAPI WS 엔드포인트 `/internal/voice/stream`(`api/voice_stream.py`): RealTime이 프록시한 오디오를 받아 부분/최종 자막을 다운 프레임(`transcript.partial`/`transcript.final`)으로 보내고, 발화 종료(`stop` 또는 UtteranceEnd) 시 메트릭 계산 후 `callback.voice` 발행 → 기존 followup 파이프라인 재사용.
 - 추상화 계층 두기: `voice/stt/base.py` (interface), `voice/stt/whisper_api.py`, `voice/tts/base.py` + `voice/tts/{provider}.py`
diff --git a/ai/src/ai_server/config/settings.py b/ai/src/ai_server/config/settings.py
@@ -66,11 +66,17 @@ class Settings(BaseSettings):
     deepgram_live_endpointing_ms: int = 800  # 무음 800ms → utterance end
     voice_stream_internal_path: str = "/internal/voice/stream"
 
-    # TTS (질문 음성화). "auto" 면 openai 키 보유 시 openai, 없으면 mock.
-    tts_provider: Literal["auto", "mock", "openai"] = "auto"
+    # TTS (질문 음성화). "auto" 면 gemini > openai 순으로 키 보유 시 선택, 둘 다 없으면 mock.
+    # Deepgram/OpenAI TTS 는 한국어 미지원이라 기본은 Gemini TTS(GEMINI_API_KEY 재사용).
+    tts_provider: Literal["auto", "mock", "openai", "gemini"] = "auto"
     openai_tts_model: str = "gpt-4o-mini-tts"
     openai_tts_voice: str = "alloy"
     openai_tts_timeout_sec: float = 30.0
+    # Gemini TTS — 한국어 지원. raw PCM(L16) 반환 → 저장 시 WAV 로 감싼다.
+    gemini_tts_model: str = "gemini-2.5-flash-preview-tts"
+    gemini_tts_voice: str = "Kore"
+    gemini_tts_base_url: str = "https://generativelanguage.googleapis.com/v1beta"
+    gemini_tts_timeout_sec: float = 30.0
     tts_audio_key_template: str = "interview/tts/{session_id}/{message_id}.mp3"
     # 음성 분석
     voice_filler_pattern: str = r"(?:음+|어+|그+|아+)"
diff --git a/ai/src/ai_server/messaging/consumers/tts_consumer.py b/ai/src/ai_server/messaging/consumers/tts_consumer.py
@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import re
+
 import structlog
 from aio_pika.abc import AbstractIncomingMessage
 
@@ -38,9 +40,15 @@ def __init__(
     async def handle(self, message: AbstractIncomingMessage) -> None:
         async with message.process(requeue=False):
             try:
-                envelope = Envelope[GenerateTtsRequest].model_validate_json(message.body)
+                envelope = Envelope[GenerateTtsRequest].model_validate_json(
+                    message.body
+                )
             except Exception as exc:
-                log.error("tts.parse.failed", error=str(exc), delivery_tag=message.delivery_tag)
+                log.error(
+                    "tts.parse.failed",
+                    error=str(exc),
+                    delivery_tag=message.delivery_tag,
+                )
                 raise
 
             if self._idempotency.is_seen_then_mark(envelope.message_id):
@@ -59,38 +67,87 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
             try:
                 result = await self._tts.synthesize(req.text, voice=self._voice)
             except TtsError as exc:
-                log.error("tts.synthesize.failed", code=exc.code, error=exc.message,
-                          session_id=req.session_id)
-                await self._publish(envelope, TtsCallbackPayload(
-                    session_id=req.session_id, message_id=req.message_id,
-                    status="FAILED", error_code=exc.code,
-                ))
+                log.error(
+                    "tts.synthesize.failed",
+                    code=exc.code,
+                    error=exc.message,
+                    session_id=req.session_id,
+                )
+                await self._publish(
+                    envelope,
+                    TtsCallbackPayload(
+                        session_id=req.session_id,
+                        message_id=req.message_id,
+                        status="FAILED",
+                        error_code=exc.code,
+                    ),
+                )
                 return
             except Exception as exc:
-                log.error("tts.synthesize.unexpected", error=str(exc), session_id=req.session_id)
-                await self._publish(envelope, TtsCallbackPayload(
-                    session_id=req.session_id, message_id=req.message_id,
-                    status="FAILED", error_code="TTS_FAILED",
-                ))
+                log.error(
+                    "tts.synthesize.unexpected",
+                    error=str(exc),
+                    session_id=req.session_id,
+                )
+                await self._publish(
+                    envelope,
+                    TtsCallbackPayload(
+                        session_id=req.session_id,
+                        message_id=req.message_id,
+                        status="FAILED",
+                        error_code="TTS_FAILED",
+                    ),
+                )
                 return
 
-            key = self._key_template.format(session_id=req.session_id, message_id=req.message_id)
+            key = self._build_key(req.session_id, req.message_id, result.content_type)
             try:
-                await self._storage.put_bytes(key, result.audio_bytes, content_type=result.content_type)
+                await self._storage.put_bytes(
+                    key, result.audio_bytes, content_type=result.content_type
+                )
             except Exception as exc:
                 log.error("tts.storage.failed", error=str(exc), key=key)
-                await self._publish(envelope, TtsCallbackPayload(
-                    session_id=req.session_id, message_id=req.message_id,
-                    status="FAILED", error_code="TTS_STORAGE_FAILED",
-                ))
+                await self._publish(
+                    envelope,
+                    TtsCallbackPayload(
+                        session_id=req.session_id,
+                        message_id=req.message_id,
+                        status="FAILED",
+                        error_code="TTS_STORAGE_FAILED",
+                    ),
+                )
                 return
 
-            await self._publish(envelope, TtsCallbackPayload(
-                session_id=req.session_id, message_id=req.message_id,
-                status="SUCCEEDED", audio_key=key, duration_sec=result.duration_sec,
-            ))
-            log.info("tts.synthesize.done", session_id=req.session_id,
-                     interview_message_id=req.message_id, key=key)
+            await self._publish(
+                envelope,
+                TtsCallbackPayload(
+                    session_id=req.session_id,
+                    message_id=req.message_id,
+                    status="SUCCEEDED",
+                    audio_key=key,
+                    duration_sec=result.duration_sec,
+                ),
+            )
+            log.info(
+                "tts.synthesize.done",
+                session_id=req.session_id,
+                interview_message_id=req.message_id,
+                key=key,
+            )
+
+    _EXT_BY_CONTENT_TYPE = {
+        "audio/wav": "wav",
+        "audio/mpeg": "mp3",
+        "audio/ogg": "ogg",
+        "audio/mp4": "m4a",
+    }
+
+    def _build_key(self, session_id: int, message_id: int, content_type: str) -> str:
+        # 템플릿의 확장자를 실제 오디오 포맷(content_type)에 맞춰 교체한다.
+        # (Gemini=WAV, OpenAI=mp3 등 공급자마다 다름 → Core 프록시가 확장자로 content-type 판단)
+        base = self._key_template.format(session_id=session_id, message_id=message_id)
+        ext = self._EXT_BY_CONTENT_TYPE.get(content_type.split(";")[0].strip(), "mp3")
+        return re.sub(r"\.[^./]+$", "", base) + "." + ext
 
     async def _publish(self, envelope, payload: TtsCallbackPayload) -> None:
         await self._publisher.publish(
diff --git a/ai/src/ai_server/voice/tts/factory.py b/ai/src/ai_server/voice/tts/factory.py
@@ -4,18 +4,36 @@
 
 from ai_server.config.settings import Settings
 from ai_server.voice.tts.base import TtsProvider
+from ai_server.voice.tts.gemini import GeminiTtsProvider
 from ai_server.voice.tts.mock import MockTtsProvider
 from ai_server.voice.tts.openai_tts import OpenAiTtsProvider
 
 log = structlog.get_logger(__name__)
 
 
 def build_tts_provider(settings: Settings) -> TtsProvider:
-    """TTS 공급자 선택. auto → openai 키 보유 시 openai, 없으면 mock."""
+    """TTS 공급자 선택. auto → gemini(한국어) > openai 키 보유 순, 둘 다 없으면 mock."""
     provider = (settings.tts_provider or "auto").lower()
 
     if provider == "auto":
-        provider = "openai" if settings.openai_api_key else "mock"
+        if settings.gemini_api_key:
+            provider = "gemini"
+        elif settings.openai_api_key:
+            provider = "openai"
+        else:
+            provider = "mock"
+
+    if provider == "gemini":
+        if not settings.gemini_api_key:
+            log.warn("tts.fallback_to_mock", reason="GEMINI_API_KEY 누락")
+            return MockTtsProvider()
+        return GeminiTtsProvider(
+            api_key=settings.gemini_api_key,
+            base_url=settings.gemini_tts_base_url,
+            model=settings.gemini_tts_model,
+            voice=settings.gemini_tts_voice,
+            timeout_sec=settings.gemini_tts_timeout_sec,
+        )
 
     if provider == "openai":
         if not settings.openai_api_key:
diff --git a/ai/src/ai_server/voice/tts/gemini.py b/ai/src/ai_server/voice/tts/gemini.py
@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import base64
+import io
+import wave
+
+import httpx
+import structlog
+
+from ai_server.voice.tts.base import TtsError, TtsProvider, TtsResult
+
+log = structlog.get_logger(__name__)
+
+
+def _parse_rate(mime: str, default: int = 24000) -> int:
+    # 예: "audio/L16;codec=pcm;rate=24000"
+    for part in mime.split(";"):
+        part = part.strip()
+        if part.startswith("rate="):
+            try:
+                return int(part[len("rate=") :])
+            except ValueError:
+                return default
+    return default
+
+
+def pcm_to_wav(
+    pcm: bytes, *, sample_rate: int, channels: int = 1, sample_width: int = 2
+) -> bytes:
+    """raw PCM(L16) 을 WAV 컨테이너로 감싼다(브라우저 재생 가능)."""
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as wav:
+        wav.setnchannels(channels)
+        wav.setsampwidth(sample_width)
+        wav.setframerate(sample_rate)
+        wav.writeframes(pcm)
+    return buf.getvalue()
+
+
+class GeminiTtsProvider(TtsProvider):
+    """Gemini TTS (generateContent, responseModalities=AUDIO). 한국어 지원.
+
+    Gemini 는 raw PCM(L16, mono)을 base64 로 반환하므로 WAV 로 감싸 audio/wav 로 저장한다.
+    Deepgram/OpenAI TTS 가 한국어 미지원이라 기존 GEMINI_API_KEY 를 재사용한다.
+    """
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        base_url: str,
+        model: str,
+        voice: str,
+        timeout_sec: float,
+    ) -> None:
+        self._api_key = api_key
+        self._base_url = base_url.rstrip("/")
+        self.model_name = model
+        self._voice = voice
+        self._timeout = timeout_sec
+
+    async def synthesize(self, text: str, *, voice: str) -> TtsResult:
+        # consumer 가 넘기는 voice 는 OpenAI 용("alloy") 이므로 Gemini 전용 voice 를 쓴다.
+        voice_name = self._voice
+        url = f"{self._base_url}/models/{self.model_name}:generateContent"
+        body = {
+            "contents": [{"parts": [{"text": text}]}],
+            "generationConfig": {
+                "responseModalities": ["AUDIO"],
+                "speechConfig": {
+                    "voiceConfig": {"prebuiltVoiceConfig": {"voiceName": voice_name}}
+                },
+            },
+        }
+        try:
+            async with httpx.AsyncClient(timeout=self._timeout) as client:
+                resp = await client.post(url, params={"key": self._api_key}, json=body)
+        except httpx.HTTPError as exc:
+            raise TtsError("TTS_HTTP_ERROR", str(exc)) from exc
+        if resp.status_code != 200:
+            raise TtsError(
+                "TTS_API_ERROR",
+                f"gemini tts status {resp.status_code}: {resp.text[:200]}",
+            )
+        try:
+            data = resp.json()
+            part = data["candidates"][0]["content"]["parts"][0]
+            inline = part.get("inlineData") or part.get("inline_data")
+            pcm = base64.b64decode(inline["data"])
+            mime = inline.get("mimeType") or inline.get("mime_type") or ""
+        except (KeyError, IndexError, TypeError, ValueError) as exc:
+            raise TtsError("TTS_EMPTY_AUDIO", f"gemini tts no audio: {exc}") from exc
+        if not pcm:
+            raise TtsError("TTS_EMPTY_AUDIO", "empty pcm")
+        rate = _parse_rate(mime)
+        wav = pcm_to_wav(pcm, sample_rate=rate)
+        duration = round(len(pcm) / float(rate * 2), 2)  # mono 16-bit
+        return TtsResult(
+            audio_bytes=wav, duration_sec=duration, content_type="audio/wav"
+        )
diff --git a/ai/tests/test_tts_consumer.py b/ai/tests/test_tts_consumer.py
diff --git a/ai/tests/test_tts_factory.py b/ai/tests/test_tts_factory.py
