diff --git a/ai/.env.example b/ai/.env.example index 400b1c85..4fd31f57 100644 --- 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 index a67e1291..e3b7a7cf 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: 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 index 01d8eaa2..9adf36d9 100644 --- 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 index 62a8341b..9fa3b41b 100644 --- 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 index a06adb5f..db79d52a 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.gemini import GeminiTtsProvider from ai_server.voice.tts.mock import MockTtsProvider from ai_server.voice.tts.openai_tts import OpenAiTtsProvider @@ -11,11 +12,28 @@ 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 new file mode 100644 index 00000000..29754e40 --- /dev/null +++ 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 index 2c0f5cb8..20d8972d 100644 --- a/ai/tests/test_tts_consumer.py +++ b/ai/tests/test_tts_consumer.py @@ -4,9 +4,19 @@ import pytest from ai_server.messaging.consumers.tts_consumer import TtsConsumer +from ai_server.voice.tts.base import TtsProvider, TtsResult from ai_server.voice.tts.mock import MockTtsProvider +class _WavTts(TtsProvider): + model_name = "fake-wav" + + async def synthesize(self, text: str, *, voice: str) -> TtsResult: + return TtsResult( + audio_bytes=b"RIFF....WAVE", duration_sec=1.2, content_type="audio/wav" + ) + + class _FakeStorage: def __init__(self): self.puts = [] @@ -19,7 +29,9 @@ class _FakePublisher: def __init__(self): self.published = [] - async def publish(self, *, routing_key, message_type, payload, trace_id, correlation_id, context): + async def publish( + self, *, routing_key, message_type, payload, trace_id, correlation_id, context + ): self.published.append((routing_key, message_type, payload)) @@ -35,25 +47,34 @@ def __init__(self, body: bytes): def process(self, requeue=False): class _Ctx: - async def __aenter__(self_): return self_ - async def __aexit__(self_, *a): return False + async def __aenter__(self_): + return self_ + + async def __aexit__(self_, *a): + return False + return _Ctx() def _envelope_body(): - return json.dumps({ - "messageId": "m-1", - "messageType": "generate.tts", - "version": "v1", - "traceId": "t-1", - "publishedAt": datetime.now(timezone.utc).isoformat(), - "publisher": "core-server", - "payload": { - "sessionId": 7, "messageId": 42, "text": "질문 본문", - "mode": "TECHNICAL", "jobCategory": "BACKEND", - }, - "context": {"userId": 1, "sessionId": 7}, - }).encode("utf-8") + return json.dumps( + { + "messageId": "m-1", + "messageType": "generate.tts", + "version": "v1", + "traceId": "t-1", + "publishedAt": datetime.now(timezone.utc).isoformat(), + "publisher": "core-server", + "payload": { + "sessionId": 7, + "messageId": 42, + "text": "질문 본문", + "mode": "TECHNICAL", + "jobCategory": "BACKEND", + }, + "context": {"userId": 1, "sessionId": 7}, + } + ).encode("utf-8") @pytest.mark.asyncio @@ -83,3 +104,26 @@ async def test_tts_consumer_synthesizes_and_publishes(): assert payload.status == "SUCCEEDED" assert payload.audio_key == "interview/tts/7/42.mp3" assert payload.message_id == 42 + + +@pytest.mark.asyncio +async def test_tts_consumer_uses_wav_extension_for_wav_audio(): + # WAV(Gemini) 결과면 템플릿의 .mp3 확장자를 .wav 로 교체해 저장해야 한다. + storage = _FakeStorage() + publisher = _FakePublisher() + consumer = TtsConsumer( + tts=_WavTts(), + storage=storage, + publisher=publisher, + idempotency=_FakeIdem(), + callback_routing_key="callback.tts", + voice="alloy", + key_template="interview/tts/{session_id}/{message_id}.mp3", + ) + + await consumer.handle(_FakeMsg(_envelope_body())) + + key, _data, ctype = storage.puts[0] + assert key == "interview/tts/7/42.wav" + assert ctype == "audio/wav" + assert publisher.published[0][2].audio_key == "interview/tts/7/42.wav" diff --git a/ai/tests/test_tts_factory.py b/ai/tests/test_tts_factory.py index 38591c14..9ced23b3 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.gemini import GeminiTtsProvider from ai_server.voice.tts.mock import MockTtsProvider from ai_server.voice.tts.openai_tts import OpenAiTtsProvider @@ -17,15 +18,33 @@ def _settings(**over): def test_auto_falls_back_to_mock_without_key(): - s = _settings(tts_provider="auto", openai_api_key="") + s = _settings(tts_provider="auto", openai_api_key="", gemini_api_key="") assert isinstance(build_tts_provider(s), MockTtsProvider) def test_auto_picks_openai_with_key(): - s = _settings(tts_provider="auto", openai_api_key="sk-test") + s = _settings(tts_provider="auto", openai_api_key="sk-test", gemini_api_key="") assert isinstance(build_tts_provider(s), OpenAiTtsProvider) def test_explicit_mock(): s = _settings(tts_provider="mock", openai_api_key="sk-test") assert isinstance(build_tts_provider(s), MockTtsProvider) + + +def test_auto_prefers_gemini_when_key_present(): + # Deepgram/OpenAI TTS 는 한국어 미지원 → gemini 키가 있으면 gemini 우선. + s = _settings( + tts_provider="auto", gemini_api_key="g-test", openai_api_key="sk-test" + ) + assert isinstance(build_tts_provider(s), GeminiTtsProvider) + + +def test_explicit_gemini(): + s = _settings(tts_provider="gemini", gemini_api_key="g-test") + assert isinstance(build_tts_provider(s), GeminiTtsProvider) + + +def test_gemini_without_key_falls_back_to_mock(): + s = _settings(tts_provider="gemini", gemini_api_key="") + assert isinstance(build_tts_provider(s), MockTtsProvider)