feat(backend): generate.tts 큐·라우팅키 설정 및 declare

Author: csh1668

ai/.env.example

@@ -56,9 +56,15 @@ DEEPGRAM_BASE_URL=https://api.deepgram.com/v1
 DEEPGRAM_MODEL=whisper-large
 DEEPGRAM_LANGUAGE=ko
 DEEPGRAM_TIMEOUT_SEC=60
-# OpenAI Whisper (Mindlogic 미지원, 직접 호출).
+# OpenAI Whisper (Mindlogic 미지원, 직접 호출). OPENAI_API_KEY/OPENAI_BASE_URL 은 TTS 와 공용.
 OPENAI_API_KEY=
 OPENAI_BASE_URL=https://api.openai.com/v1
 WHISPER_MODEL=whisper-1
 WHISPER_LANGUAGE=ko
 WHISPER_TIMEOUT_SEC=60
+
+# TTS (질문 음성화). auto=OPENAI_API_KEY 보유 시 openai, 없으면 mock. (OPENAI_API_KEY 재사용)
+TTS_PROVIDER=auto
+OPENAI_TTS_MODEL=gpt-4o-mini-tts
+OPENAI_TTS_VOICE=alloy
+OPENAI_TTS_TIMEOUT_SEC=30

ai/CLAUDE.md

@@ -95,6 +95,7 @@ ai/
 | `ai.analyze.web` | `analyze.web` | 본 구현 (URL → trafilatura) |
 | `ai.generate.questions` | `generate.questions` | 본 구현 (Pro 모델, 질문 풀 생성, US-18) |
 | `ai.generate.followup` | `generate.followup` | 본 구현 (Flash 모델, 답변 평가+꼬리질문, US-19) |
+| `ai.generate.tts` | `generate.tts` | 본 구현 (질문 음성화, OpenAI TTS → S3 → `callback.tts`) |
 
 콜백 발행: `ai.callback.{type}` 익스체인지.
 상세 envelope/스키마/재시도: [`/docs/messaging.md`](../docs/messaging.md).
@@ -188,8 +189,8 @@ chain = prompt | llm | PydanticOutputParser(pydantic_object=...)
   - 비용: $0.006 / 분 (1시간 면접 ≈ ₩500 / USD ≈ $0.36)
   - 셀프호스팅 옵션: `whisper.cpp` 또는 `faster-whisper` (GPU 권장, 비용 ↓ but 운영 부담 ↑)
   - 브라우저 내장 SpeechRecognition API는 정확도 부족으로 채택 안 함
-- TTS 제공자 미정 → 도입 시 본 섹션 갱신
-- 추상화 계층 두기: `voice/stt/base.py` (interface), `voice/stt/whisper_api.py`, `voice/tts/{provider}.py`
+- **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` 발행.
+- 추상화 계층 두기: `voice/stt/base.py` (interface), `voice/stt/whisper_api.py`, `voice/tts/base.py` + `voice/tts/{provider}.py`
 - 분석:
   - WPM = words / minutes
   - 간투어: 한국어 정규식 `r"\b(음+|어+|그+)\b"` 카운트
@@ -324,6 +325,8 @@ docker run --env-file .env -p 8000:8000 stackup-ai
 - **스토리지 추상화** (`storage/`): `S3Storage`(기본) / `LocalFilesystemStorage`. `STORAGE_BACKEND` 토글.
 - **LLM 호출 로깅 본 구현** (`observability/llm_logging_callback.py`, US-30):
   LangChain `AsyncCallbackHandler` 가 토큰/latency 측정 → Core `/api/internal/ai-logs` POST.
-- 음성 모듈은 Phase 2
+- **질문 TTS consumer 본 구현** (`messaging/consumers/tts_consumer.py`, `voice/tts/`):
+  `generate.tts` 수신 → OpenAI TTS 합성(`OpenAiTtsProvider`, mock fallback) → S3 PUT(`interview/tts/{sessionId}/{messageId}.mp3`) → `callback.tts` 발행.
+- 음성 분석(STT/WPM/filler) 모듈은 Phase 2
 
 각 도입 시 본 문서 갱신.

ai/src/ai_server/config/settings.py

@@ -26,12 +26,14 @@ class Settings(BaseSettings):
     ai_queue_followup: str = "ai.generate.followup"
     ai_queue_feedback: str = "ai.generate.feedback"
     ai_queue_voice: str = "ai.analyze.voice"
+    ai_queue_tts: str = "ai.generate.tts"
     ai_queue_prefetch: int = 10
     ai_callback_exchange: str = "stackup.ai-to-core"
     ai_callback_routing_analysis: str = "callback.analysis"
     ai_callback_routing_questions: str = "callback.questions"
     ai_callback_routing_feedback: str = "callback.feedback"
     ai_callback_routing_voice: str = "callback.voice"
+    ai_callback_routing_tts: str = "callback.tts"
     # AI -> RealTime 직접 발행 (분석 단계 진행 상황). Core 를 거치지 않는 휘발성 알림.
     ai_realtime_exchange: str = "stackup.realtime"
     ai_realtime_routing_user: str = "realtime.user.notify"
@@ -52,6 +54,13 @@ class Settings(BaseSettings):
     deepgram_model: str = "whisper-large"   # 한국어 정확도 우선; 저비용 우선 시 nova-2.
     deepgram_language: str = "ko"
     deepgram_timeout_sec: float = 60.0
+
+    # TTS (질문 음성화). "auto" 면 openai 키 보유 시 openai, 없으면 mock.
+    tts_provider: Literal["auto", "mock", "openai"] = "auto"
+    openai_tts_model: str = "gpt-4o-mini-tts"
+    openai_tts_voice: str = "alloy"
+    openai_tts_timeout_sec: float = 30.0
+    tts_audio_key_template: str = "interview/tts/{session_id}/{message_id}.mp3"
     # 음성 분석
     voice_filler_pattern: str = r"(?:음+|어+|그+|아+)"
     ai_publisher_name: str = "ai-server"

ai/src/ai_server/messaging/consumers/tts_consumer.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import structlog
+from aio_pika.abc import AbstractIncomingMessage
+
+from ai_server.messaging.idempotency import LruIdempotencyStore
+from ai_server.messaging.publisher import CallbackPublisher
+from ai_server.model.envelope import Envelope
+from ai_server.model.messages.tts import GenerateTtsRequest, TtsCallbackPayload
+from ai_server.storage.base import ObjectStorage
+from ai_server.voice.tts.base import TtsError, TtsProvider
+
+log = structlog.get_logger(__name__)
+
+
+class TtsConsumer:
+    """generate.tts consumer — 질문 텍스트를 TTS 합성 → S3 PUT → callback.tts 발행."""
+
+    def __init__(
+        self,
+        *,
+        tts: TtsProvider,
+        storage: ObjectStorage,
+        publisher: CallbackPublisher,
+        idempotency: LruIdempotencyStore,
+        callback_routing_key: str,
+        voice: str,
+        key_template: str,
+    ) -> None:
+        self._tts = tts
+        self._storage = storage
+        self._publisher = publisher
+        self._idempotency = idempotency
+        self._callback_routing_key = callback_routing_key
+        self._voice = voice
+        self._key_template = key_template
+
+    async def handle(self, message: AbstractIncomingMessage) -> None:
+        async with message.process(requeue=False):
+            try:
+                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)
+                raise
+
+            if self._idempotency.is_seen_then_mark(envelope.message_id):
+                log.info("tts.idempotent.skip", message_id=envelope.message_id)
+                return
+
+            req = envelope.payload
+            log.info(
+                "tts.synthesize.start",
+                message_id=envelope.message_id,
+                session_id=req.session_id,
+                interview_message_id=req.message_id,
+                trace_id=envelope.trace_id,
+            )
+
+            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,
+                ))
+                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",
+                ))
+                return
+
+            key = self._key_template.format(session_id=req.session_id, message_id=req.message_id)
+            try:
+                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",
+                ))
+                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)
+
+    async def _publish(self, envelope, payload: TtsCallbackPayload) -> None:
+        await self._publisher.publish(
+            routing_key=self._callback_routing_key,
+            message_type="callback.tts",
+            payload=payload,
+            trace_id=envelope.trace_id,
+            correlation_id=envelope.message_id,
+            context=envelope.context,
+        )

ai/src/ai_server/messaging/runner.py

@@ -33,8 +33,10 @@
 from ai_server.messaging.consumers.feedback_consumer import FeedbackConsumer
 from ai_server.messaging.consumers.followup_consumer import FollowupConsumer
 from ai_server.messaging.consumers.questions_consumer import QuestionsConsumer
+from ai_server.messaging.consumers.tts_consumer import TtsConsumer
 from ai_server.messaging.consumers.voice_consumer import VoiceConsumer
 from ai_server.voice.stt.factory import build_stt_provider
+from ai_server.voice.tts.factory import build_tts_provider
 from ai_server.messaging.consumers.repository_consumer import RepositoryConsumer
 from ai_server.messaging.consumers.resume_consumer import ResumeConsumer
 from ai_server.messaging.consumers.web_consumer import WebResumeConsumer
@@ -210,6 +212,18 @@ def __init__(self, settings: Settings) -> None:
             core_client=core_client,
         )
 
+        # 질문 TTS (Part A)
+        tts = build_tts_provider(settings)
+        self._tts_consumer = TtsConsumer(
+            tts=tts,
+            storage=storage,
+            publisher=self._publisher,
+            idempotency=self._idempotency,
+            callback_routing_key=settings.ai_callback_routing_tts,
+            voice=settings.openai_tts_voice,
+            key_template=settings.tts_audio_key_template,
+        )
+
         self._consumers: list[tuple[AbstractRobustQueue, str]] = []
 
     async def start(self) -> None:
@@ -254,6 +268,11 @@ async def start(self) -> None:
             queue_name=self._settings.ai_queue_voice,
             handler=self._voice_consumer.handle,
         )
+        await self._start_consumer(
+            channel,
+            queue_name=self._settings.ai_queue_tts,
+            handler=self._tts_consumer.handle,
+        )
 
     async def _start_consumer(self, channel, *, queue_name, handler) -> None:
         queue = await channel.declare_queue(

ai/src/ai_server/model/messages/tts.py

@@ -0,0 +1,28 @@
+from typing import Literal
+
+from pydantic import BaseModel
+
+from ai_server.model._config import camel_config
+
+
+class GenerateTtsRequest(BaseModel):
+    """Core 가 질문 메시지 commit 후 발행. AI 가 text 를 TTS 합성."""
+    model_config = camel_config()
+
+    session_id: int
+    message_id: int                      # interview_messages.id (INTERVIEWER 질문)
+    text: str
+    mode: Literal["PERSONALITY", "TECHNICAL", "INTEGRATED"]
+    job_category: Literal["FRONTEND", "BACKEND", "INFRA", "DBA"]
+
+
+class TtsCallbackPayload(BaseModel):
+    """AI → Core. 합성 결과(S3 키) 또는 실패. Core TtsCallbackPayload 와 호환."""
+    model_config = camel_config()
+
+    session_id: int
+    message_id: int
+    status: Literal["SUCCEEDED", "FAILED"]
+    audio_key: str | None = None         # camelCase 직렬화 → audioKey
+    duration_sec: float | None = None
+    error_code: str | None = None

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

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+
+class TtsError(Exception):
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+
+
+@dataclass(frozen=True)
+class TtsResult:
+    audio_bytes: bytes
+    duration_sec: float | None
+    content_type: str = "audio/mpeg"
+
+
+class TtsProvider(ABC):
+    model_name: str = "tts"
+
+    @abstractmethod
+    async def synthesize(self, text: str, *, voice: str) -> TtsResult: ...

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

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import structlog
+
+from ai_server.config.settings import Settings
+from ai_server.voice.tts.base import TtsProvider
+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."""
+    provider = (settings.tts_provider or "auto").lower()
+
+    if provider == "auto":
+        provider = "openai" if settings.openai_api_key else "mock"
+
+    if provider == "openai":
+        if not settings.openai_api_key:
+            log.warn("tts.fallback_to_mock", reason="OPENAI_API_KEY 누락")
+            return MockTtsProvider()
+        return OpenAiTtsProvider(
+            api_key=settings.openai_api_key,
+            base_url=settings.openai_base_url,
+            model=settings.openai_tts_model,
+            timeout_sec=settings.openai_tts_timeout_sec,
+        )
+
+    if provider == "mock":
+        return MockTtsProvider()
+
+    log.warn("tts.unknown_provider_fallback_mock", provider=provider)
+    return MockTtsProvider()

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

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from ai_server.voice.tts.base import TtsProvider, TtsResult
+
+
+class MockTtsProvider(TtsProvider):
+    model_name = "mock-tts"
+
+    async def synthesize(self, text: str, *, voice: str) -> TtsResult:
+        # 텍스트 길이에 비례한 가짜 오디오 바이트 + 추정 길이(분당 ~300자 가정).
+        payload = ("MOCKMP3:" + text).encode("utf-8")
+        duration = max(0.5, round(len(text) / 5.0, 2))
+        return TtsResult(audio_bytes=payload, duration_sec=duration, content_type="audio/mpeg")