feat(ai): TTS를 Mindlogic 게이트웨이(충남대 키)로 라우팅 가능하게

ai/.env.example

@@ -65,8 +65,9 @@ WHISPER_MODEL=whisper-1
 WHISPER_LANGUAGE=ko
 WHISPER_TIMEOUT_SEC=60
 
-# TTS (질문 음성화). auto=GEMINI_API_KEY > OPENAI_API_KEY 순으로 선택, 둘 다 없으면 mock.
-# Deepgram/OpenAI TTS 는 한국어 미지원 → 기본은 Gemini TTS(GEMINI_API_KEY 재사용, raw PCM→WAV).
+# TTS (질문 음성화). auto=LLM_API_KEY(gateway) > GEMINI_API_KEY > OPENAI_API_KEY 순, 없으면 mock.
+# gateway: Mindlogic 게이트웨이(/audio/speech)가 Gemini TTS 로 라우팅 → 직접 GEMINI_API_KEY 429 부하 분산.
+# Deepgram/OpenAI TTS 는 한국어 미지원. gateway/gemini 모두 GEMINI_TTS_MODEL/VOICE 공유, raw PCM→WAV.
 TTS_PROVIDER=auto
 OPENAI_TTS_MODEL=gpt-4o-mini-tts
 OPENAI_TTS_VOICE=alloy

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: 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 이 내부 호스트라 브라우저 직접 접근 불가).
+- **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` 추상화 + `GatewayTtsProvider`(Mindlogic 게이트웨이 `/audio/speech`, LLM_API_KEY, raw PCM→WAV)/`GeminiTtsProvider`(직접 GEMINI_API_KEY)/`OpenAiTtsProvider`(`gpt-4o-mini-tts`, mp3)/`MockTtsProvider`, `build_tts_provider` factory(`TTS_PROVIDER=auto`면 **LLM_API_KEY(gateway) > GEMINI_API_KEY > OPENAI_API_KEY** 순 — 게이트웨이 우선으로 직접 키 429 부하 분산). gateway/gemini 는 `gemini_tts_model`/`gemini_tts_voice` 공유. `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`

ai/src/ai_server/config/settings.py

@@ -65,9 +65,10 @@ class Settings(BaseSettings):
     deepgram_live_endpointing_ms: int = 800  # 무음 800ms → utterance end
     voice_stream_internal_path: str = "/internal/voice/stream"
 
-    # TTS (질문 음성화). "auto" 면 gemini > openai 순으로 키 보유 시 선택, 둘 다 없으면 mock.
-    # Deepgram/OpenAI TTS 는 한국어 미지원이라 기본은 Gemini TTS(GEMINI_API_KEY 재사용).
-    tts_provider: Literal["auto", "mock", "openai", "gemini"] = "auto"
+    # TTS (질문 음성화). "auto" 면 gateway(충남대 키) > gemini(직접 키) > openai, 없으면 mock.
+    # gateway 는 Mindlogic 게이트웨이(/audio/speech)가 Gemini TTS 로 라우팅 — 직접 키 429 부하 분산.
+    # Deepgram/OpenAI TTS 는 한국어 미지원. gateway/gemini 모두 gemini_tts_model/voice 를 공유한다.
+    tts_provider: Literal["auto", "mock", "openai", "gemini", "gateway"] = "auto"
     openai_tts_model: str = "gpt-4o-mini-tts"
     openai_tts_voice: str = "alloy"
     openai_tts_timeout_sec: float = 30.0

ai/src/ai_server/voice/tts/factory.py

@@ -4,6 +4,7 @@
 
 from ai_server.config.settings import Settings
 from ai_server.voice.tts.base import TtsProvider
+from ai_server.voice.tts.gateway import GatewayTtsProvider
 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
@@ -12,17 +13,34 @@
 
 
 def build_tts_provider(settings: Settings) -> TtsProvider:
-    """TTS 공급자 선택. auto → gemini(한국어) > openai 키 보유 순, 둘 다 없으면 mock."""
+    """TTS 공급자 선택. auto → gateway(충남대 키) > gemini(직접 키) > openai, 없으면 mock.
+
+    gateway 를 우선해 직접 GEMINI_API_KEY 의 429 부하를 던다(게이트웨이가 Gemini TTS 로 라우팅).
+    """
     provider = (settings.tts_provider or "auto").lower()
 
     if provider == "auto":
-        if settings.gemini_api_key:
+        if settings.llm_api_key:
+            provider = "gateway"
+        elif settings.gemini_api_key:
             provider = "gemini"
         elif settings.openai_api_key:
             provider = "openai"
         else:
             provider = "mock"
 
+    if provider == "gateway":
+        if not settings.llm_api_key:
+            log.warn("tts.fallback_to_mock", reason="LLM_API_KEY 누락")
+            return MockTtsProvider()
+        return GatewayTtsProvider(
+            api_key=settings.llm_api_key,
+            base_url=settings.llm_base_url,
+            model=settings.gemini_tts_model,
+            voice=settings.gemini_tts_voice,
+            timeout_sec=settings.gemini_tts_timeout_sec,
+        )
+
     if provider == "gemini":
         if not settings.gemini_api_key:
             log.warn("tts.fallback_to_mock", reason="GEMINI_API_KEY 누락")

ai/src/ai_server/voice/tts/gateway.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import httpx
+import structlog
+
+from ai_server.voice.tts.base import TtsError, TtsProvider, TtsResult
+from ai_server.voice.tts.gemini import _parse_rate, pcm_to_wav
+
+log = structlog.get_logger(__name__)
+
+
+class GatewayTtsProvider(TtsProvider):
+    """충남대 Mindlogic 게이트웨이 TTS (OpenAI 호환 /audio/speech).
+
+    게이트웨이가 Gemini TTS 로 라우팅하며 raw PCM(L16, mono)을 바디로 직접 반환한다
+    (네이티브 Gemini 의 base64 JSON 과 다름). WAV 로 감싸 audio/wav 로 저장한다.
+    GEMINI_API_KEY(직접) 대신 LLM_API_KEY(게이트웨이)를 써서 직접 키의 429 부하를 던다.
+    """
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        base_url: str,
+        model: str,
+        voice: str,
+        timeout_sec: float,
+        client: httpx.AsyncClient | None = None,
+    ) -> None:
+        self._api_key = api_key
+        self._base_url = base_url.rstrip("/")
+        self.model_name = model
+        self._voice = voice
+        self._timeout = timeout_sec
+        self._client = client
+
+    async def synthesize(self, text: str, *, voice: str) -> TtsResult:
+        # consumer 가 넘기는 voice 는 OpenAI 용("alloy") 이므로 Gemini 전용 voice 를 쓴다.
+        url = f"{self._base_url}/audio/speech"
+        body = {"model": self.model_name, "input": text, "voice": self._voice}
+        headers = {"Authorization": f"Bearer {self._api_key}"}
+        try:
+            if self._client is not None:
+                resp = await self._client.post(url, headers=headers, json=body)
+            else:
+                async with httpx.AsyncClient(timeout=self._timeout) as client:
+                    resp = await client.post(url, headers=headers, 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"gateway tts status {resp.status_code}: {resp.text[:200]}",
+            )
+        audio = resp.content
+        if not audio:
+            raise TtsError("TTS_EMPTY_AUDIO", "gateway tts empty audio")
+
+        content_type = resp.headers.get("content-type", "").lower()
+        # 게이트웨이는 raw PCM(L16/24kHz) 을 반환 → WAV 로 감싼다.
+        # 혹시 이미 컨테이너(wav/mpeg)면 그대로 전달.
+        if (
+            "pcm" in content_type
+            or "l16" in content_type
+            or "octet-stream" in content_type
+        ):
+            rate = _parse_rate(content_type)
+            wav = pcm_to_wav(audio, sample_rate=rate)
+            duration = round(len(audio) / float(rate * 2), 2)  # mono 16-bit
+            return TtsResult(
+                audio_bytes=wav, duration_sec=duration, content_type="audio/wav"
+            )
+        return TtsResult(
+            audio_bytes=audio,
+            duration_sec=None,
+            content_type=content_type.split(";")[0] or "audio/mpeg",
+        )

ai/tests/test_gateway_tts.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import wave
+import io
+
+import httpx
+import pytest
+
+from ai_server.voice.tts.base import TtsError
+from ai_server.voice.tts.gateway import GatewayTtsProvider
+
+
+def _client(*, status: int, content: bytes, content_type: str) -> httpx.AsyncClient:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            status, content=content, headers={"content-type": content_type}
+        )
+
+    return httpx.AsyncClient(transport=httpx.MockTransport(handler))
+
+
+@pytest.mark.asyncio
+async def test_gateway_tts_wraps_raw_pcm_into_wav():
+    pcm = b"\x00\x01" * 24000  # 1초 분량(24kHz mono 16-bit)
+    async with _client(status=200, content=pcm, content_type="audio/pcm") as client:
+        provider = GatewayTtsProvider(
+            api_key="k",
+            base_url="https://gw/v1/gateway",
+            model="gemini-2.5-flash-preview-tts",
+            voice="Kore",
+            timeout_sec=5.0,
+            client=client,
+        )
+        result = await provider.synthesize("질문입니다", voice="alloy")
+
+    assert result.content_type == "audio/wav"
+    # 유효한 WAV 컨테이너인지 확인
+    with wave.open(io.BytesIO(result.audio_bytes), "rb") as wav:
+        assert wav.getframerate() == 24000
+        assert wav.getnchannels() == 1
+    assert result.duration_sec == pytest.approx(1.0, abs=0.05)
+
+
+@pytest.mark.asyncio
+async def test_gateway_tts_raises_on_error_status():
+    async with _client(
+        status=400, content=b'{"detail":"bad"}', content_type="application/json"
+    ) as client:
+        provider = GatewayTtsProvider(
+            api_key="k",
+            base_url="https://gw/v1/gateway",
+            model="gemini-2.5-flash-preview-tts",
+            voice="Kore",
+            timeout_sec=5.0,
+            client=client,
+        )
+        with pytest.raises(TtsError):
+            await provider.synthesize("질문", voice="alloy")

ai/tests/test_tts_factory.py

@@ -1,5 +1,6 @@
 from ai_server.config.settings import Settings
 from ai_server.voice.tts.factory import build_tts_provider
+from ai_server.voice.tts.gateway import GatewayTtsProvider
 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
@@ -12,11 +13,19 @@ def _settings(**over):
         s3_access_key="x",
         s3_secret_key="x",
         s3_bucket_name="b",
+        # 게이트웨이가 auto 최우선이므로, 명시 안 한 테스트는 비활성으로 고정(.env 누수 방지).
+        llm_api_key="",
     )
     base.update(over)
     return Settings(**base)
 
 
+def test_auto_prefers_gateway_when_llm_key_present():
+    # 충남대 게이트웨이 키가 있으면 직접 Gemini 키보다 우선(429 부하 분산).
+    s = _settings(tts_provider="auto", llm_api_key="gw-key", gemini_api_key="g-test")
+    assert isinstance(build_tts_provider(s), GatewayTtsProvider)
+
+
 def test_auto_falls_back_to_mock_without_key():
     s = _settings(tts_provider="auto", openai_api_key="", gemini_api_key="")
     assert isinstance(build_tts_provider(s), MockTtsProvider)

docs/environment.md

@@ -148,7 +148,7 @@ ANALYZED_WEB_RESUME_MD_KEY_TEMPLATE=analyzed/web-resume/{resume_id}/summary.md
 # ===== STT/TTS (Phase 2) =====
 STT_PROVIDER=whisper-api          # whisper-api | whisper-self-hosted
 WHISPER_MODEL=whisper-1           # OpenAI Whisper API 모델
-TTS_PROVIDER=auto                 # auto | mock | openai (auto=OPENAI_API_KEY 보유 시 openai)
+TTS_PROVIDER=auto                 # auto | mock | openai | gemini | gateway (auto=LLM_API_KEY(gateway) > GEMINI_API_KEY > OPENAI_API_KEY)
 OPENAI_TTS_MODEL=gpt-4o-mini-tts  # OpenAI TTS 모델 (질문 음성화)
 OPENAI_TTS_VOICE=alloy            # TTS 보이스
 OPENAI_TTS_TIMEOUT_SEC=30         # TTS HTTP 타임아웃(초)