diff --git a/ai/.env.example b/ai/.env.example index 73c30d3b..4ce2bed8 100644 --- a/ai/.env.example +++ b/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 diff --git a/ai/CLAUDE.md b/ai/CLAUDE.md index 214efcbb..b4344cf4 100644 --- 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: 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` diff --git a/ai/src/ai_server/config/settings.py b/ai/src/ai_server/config/settings.py index 7f51c3a7..6a6c2655 100644 --- a/ai/src/ai_server/config/settings.py +++ b/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 diff --git a/ai/src/ai_server/voice/tts/factory.py b/ai/src/ai_server/voice/tts/factory.py index db79d52a..2e5d1c2d 100644 --- a/ai/src/ai_server/voice/tts/factory.py +++ b/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 누락") diff --git a/ai/src/ai_server/voice/tts/gateway.py b/ai/src/ai_server/voice/tts/gateway.py new file mode 100644 index 00000000..206f03c3 --- /dev/null +++ b/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", + ) diff --git a/ai/tests/test_gateway_tts.py b/ai/tests/test_gateway_tts.py new file mode 100644 index 00000000..c8889f43 --- /dev/null +++ b/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") diff --git a/ai/tests/test_tts_factory.py b/ai/tests/test_tts_factory.py index 9ced23b3..a84a6255 100644 --- a/ai/tests/test_tts_factory.py +++ b/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) diff --git a/docs/environment.md b/docs/environment.md index 114adc67..c47932f8 100644 --- a/docs/environment.md +++ b/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 타임아웃(초)