feat(backend,ai,realtime): 음성 스트림 WS 관련 작업

ai/CLAUDE.md

@@ -17,6 +17,7 @@
 | 스키마 | Pydantic 2.x + pydantic-settings |
 | HTTP | httpx |
 | MQ | aio-pika (async AMQP) |
+| WebSocket | FastAPI WS (서버) + `websockets` (Deepgram Live 클라이언트) |
 | 객체 스토리지 | boto3 (S3 호환) |
 | LLM | LangChain 1.x (core + community) |
 | 로깅 | structlog |
@@ -190,6 +191,8 @@ chain = prompt | llm | PydanticOutputParser(pydantic_object=...)
   - 셀프호스팅 옵션: `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` 발행.
+- **스트리밍 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`
 - 분석:
   - WPM = words / minutes
@@ -327,6 +330,8 @@ docker run --env-file .env -p 8000:8000 stackup-ai
   LangChain `AsyncCallbackHandler` 가 토큰/latency 측정 → Core `/api/internal/ai-logs` POST.
 - **질문 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
+- **실시간 스트리밍 음성 답변 본 구현** (RT3, `api/voice_stream.py`, `voice/stt/{live,mock_live,deepgram_live,live_factory}.py`):
+  FastAPI WS `/internal/voice/stream` 수신(RealTime 프록시 경유) → Deepgram Live(`deepgram_live.py`, mock fallback)로 부분/최종 자막 다운 → 발화 종료 시 메트릭 계산 후 `callback.voice` 발행. `VoiceCallbackService`/followup 무변경 재사용. 신규 의존성 `websockets`.
+- 배치 음성 분석(STT/WPM/filler) 모듈은 `voice/stt/whisper_api.py`(+ Deepgram) + `voice/analysis/metrics.py`로 본 구현(`analyze.voice` consumer)
 
 각 도입 시 본 문서 갱신.

ai/pyproject.toml

@@ -24,6 +24,7 @@ dependencies = [
     "langchain-community>=0.4.1",
     "langchain-openai>=0.3.0",
     "structlog>=25.5.0",
+    "websockets>=13",
 ]
 
 [project.optional-dependencies]

ai/src/ai_server/api/voice_stream.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import asyncio
+
+import structlog
+from fastapi import APIRouter, Query, WebSocket, WebSocketDisconnect
+
+from ai_server.config.settings import Settings
+from ai_server.messaging.publisher import CallbackPublisher
+from ai_server.model.envelope import MessageContext
+from ai_server.model.messages.voice import VoiceCallbackPayload
+from ai_server.voice.analysis.metrics import analyze
+from ai_server.voice.stt.live import LiveSttProvider
+
+log = structlog.get_logger(__name__)
+
+router = APIRouter()
+
+
+def _provider(ws: WebSocket) -> LiveSttProvider:
+    return ws.app.state.live_stt_provider
+
+
+def _publisher(ws: WebSocket) -> CallbackPublisher:
+    return ws.app.state.callback_publisher
+
+
+def _settings(ws: WebSocket) -> Settings:
+    return ws.app.state.settings
+
+
+@router.websocket("/internal/voice/stream")
+async def voice_stream(
+    ws: WebSocket,
+    session_id: int = Query(..., alias="sessionId"),
+    message_id: int = Query(..., alias="messageId"),
+    content_type: str = Query("audio/webm", alias="contentType"),
+    api_key: str | None = Query(None, alias="apiKey"),
+) -> None:
+    settings = _settings(ws)
+    # 내부 인증: RealTime 이 전달한 키 검증(쿼리 또는 헤더).
+    expected = settings.core_internal_api_key
+    provided = api_key or ws.headers.get("x-internal-api-key")
+    if expected and provided != expected:
+        await ws.close(code=4401)
+        return
+
+    await ws.accept()
+    provider = _provider(ws)
+    publisher = _publisher(ws)
+    session = provider.open_session(content_type=content_type, language=None)
+    await session.start()
+
+    async def pump_transcripts() -> None:
+        async for ev in session.events():
+            payload = {
+                "type": "transcript.final" if ev.is_final else "transcript.partial",
+                "text": ev.text,
+            }
+            if ev.is_final:
+                payload["messageId"] = message_id
+            try:
+                await ws.send_json(payload)
+            except Exception:  # noqa: BLE001
+                return
+
+    pump = asyncio.create_task(pump_transcripts())
+    stopped = False
+    try:
+        while True:
+            msg = await ws.receive()
+            if msg["type"] == "websocket.disconnect":
+                break
+            if msg.get("bytes") is not None:
+                await session.push(msg["bytes"])
+            elif msg.get("text") is not None and '"stop"' in msg["text"]:
+                stopped = True
+                break
+    except WebSocketDisconnect:
+        pass
+    finally:
+        await session.finish()
+        # 남은 최종 이벤트 flush. Deepgram 등 upstream 이 CloseStream 에 무응답이면
+        # pump(events 생성기 소진)이 영구 대기할 수 있으므로 timeout 으로 상한.
+        try:
+            await asyncio.wait_for(pump, timeout=10.0)
+        except asyncio.TimeoutError:
+            pump.cancel()
+            log.warn("voice_stream.pump.timeout", session_id=session_id, message_id=message_id)
+        result = await session.result()
+        await session.close()
+
+        if result.text.strip():
+            metrics = analyze(result, filler_pattern=settings.voice_filler_pattern)
+            cb = VoiceCallbackPayload(
+                session_id=session_id,
+                interview_message_id=message_id,
+                transcript=result.text,
+                speaking_rate_wpm=metrics.speaking_rate_wpm,
+                silence_duration_sec=metrics.silence_duration_sec,
+                filler_word_counts=metrics.filler_word_counts,
+                pronunciation_accuracy=metrics.pronunciation_accuracy,
+                error_code=None,
+            )
+        else:
+            cb = VoiceCallbackPayload(
+                session_id=session_id,
+                interview_message_id=message_id,
+                transcript=None,
+                filler_word_counts={},
+                error_code="TRANSCRIPTION_EMPTY",
+            )
+        try:
+            await publisher.publish(
+                routing_key=settings.ai_callback_routing_voice,
+                message_type="callback.voice",
+                payload=cb,
+                trace_id=f"voice-stream-{session_id}-{message_id}",
+                correlation_id=f"voice-stream-{message_id}",
+                context=MessageContext(session_id=session_id),
+            )
+            log.info(
+                "voice_stream.callback.published",
+                session_id=session_id,
+                interview_message_id=message_id,
+                stopped=stopped,
+                chars=len(result.text or ""),
+            )
+        except Exception as exc:  # noqa: BLE001
+            log.error(
+                "voice_stream.callback.failed", error=str(exc), session_id=session_id
+            )
+        try:
+            await ws.close()
+        except Exception:  # noqa: BLE001
+            pass

ai/src/ai_server/config/settings.py

@@ -55,6 +55,14 @@ class Settings(BaseSettings):
     deepgram_language: str = "ko"
     deepgram_timeout_sec: float = 60.0
 
+    # 스트리밍 STT (실시간 음성 답변). "auto" 면 DEEPGRAM_API_KEY 보유 시 deepgram_live, 없으면 mock.
+    live_stt_provider: Literal["auto", "mock", "deepgram_live"] = "auto"
+    deepgram_live_url: str = "wss://api.deepgram.com/v1/listen"
+    deepgram_live_model: str = "nova-2"          # 스트리밍은 nova-2(저지연). 한국어 지원.
+    deepgram_live_language: str = "ko"
+    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"
     openai_tts_model: str = "gpt-4o-mini-tts"

ai/src/ai_server/main.py

@@ -5,8 +5,10 @@
 from fastapi import FastAPI
 
 from ai_server.api.health import router as health_router
+from ai_server.api.voice_stream import router as voice_stream_router
 from ai_server.config.settings import Settings, get_settings
 from ai_server.messaging.runner import MessagingRuntime
+from ai_server.voice.stt.live_factory import build_live_stt_provider
 
 log = structlog.get_logger(__name__)
 
@@ -18,6 +20,9 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         app.state.messaging = runtime
         try:
             await runtime.start()
+            app.state.settings = settings
+            app.state.live_stt_provider = build_live_stt_provider(settings)
+            app.state.callback_publisher = runtime.publisher
             yield
         finally:
             await runtime.stop()
@@ -36,6 +41,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     )
 
     app.include_router(health_router)
+    app.include_router(voice_stream_router)
 
     return app
 

ai/src/ai_server/messaging/runner.py

@@ -226,6 +226,10 @@ def __init__(self, settings: Settings) -> None:
 
         self._consumers: list[tuple[AbstractRobustQueue, str]] = []
 
+    @property
+    def publisher(self) -> CallbackPublisher:
+        return self._publisher
+
     async def start(self) -> None:
         await self._connection.open()
         await self._publisher.open()

ai/src/ai_server/voice/stt/deepgram_live.py

@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import math
+from collections.abc import AsyncIterator
+from typing import Any
+
+import structlog
+import websockets
+
+from ai_server.voice.stt.base import TranscriptionResult, TranscriptionSegment
+from ai_server.voice.stt.live import LiveSttProvider, LiveSttSession, LiveTranscriptEvent
+
+log = structlog.get_logger(__name__)
+
+
+class _DeepgramLiveSession(LiveSttSession):
+    def __init__(self, *, api_key: str, url: str, model: str, language: str,
+                 endpointing_ms: int, content_type: str) -> None:
+        self._api_key = api_key
+        self._url = url
+        self._model = model
+        self._language = language
+        self._endpointing_ms = endpointing_ms
+        self._content_type = content_type
+        self._ws: Any | None = None
+        self._queue: asyncio.Queue[LiveTranscriptEvent | None] = asyncio.Queue()
+        self._recv_task: asyncio.Task[None] | None = None
+        self._finals: list[str] = []
+        self._segments: list[TranscriptionSegment] = []
+
+    def _query(self) -> str:
+        # encoding/sample_rate 는 컨테이너(webm/opus)면 Deepgram 이 자동 디코드.
+        params = {
+            "model": self._model,
+            "language": self._language,
+            "smart_format": "true",
+            "interim_results": "true",
+            # endpointing: N ms 무음 후 speech_final=true 로 턴 종료 신호 (Deepgram 공식).
+            "endpointing": str(self._endpointing_ms),
+            # utterance_end_ms: 별도 UtteranceEnd 메시지 backstop (interim_results 필요, 최소 1000).
+            "utterance_end_ms": str(max(1000, self._endpointing_ms)),
+            "vad_events": "true",
+        }
+        return self._url + "?" + "&".join(f"{k}={v}" for k, v in params.items())
+
+    async def start(self) -> None:
+        self._ws = await websockets.connect(
+            self._query(),
+            additional_headers={"Authorization": f"Token {self._api_key}"},
+        )
+        self._recv_task = asyncio.create_task(self._recv_loop())
+
+    async def _recv_loop(self) -> None:
+        assert self._ws is not None
+        try:
+            async for raw in self._ws:
+                msg = json.loads(raw)
+                mtype = msg.get("type")
+                if mtype == "Results":
+                    alt = (((msg.get("channel") or {}).get("alternatives") or [{}])[0])
+                    text = str(alt.get("transcript") or "")
+                    is_final = bool(msg.get("is_final"))
+                    speech_final = bool(msg.get("speech_final"))
+                    if text:
+                        await self._queue.put(
+                            LiveTranscriptEvent(
+                                text=text, is_final=is_final, speech_final=speech_final
+                            )
+                        )
+                    if is_final and text:
+                        self._finals.append(text)
+                        start = float(msg.get("start", 0.0))
+                        dur = float(msg.get("duration", 0.0))
+                        conf = alt.get("confidence")
+                        self._segments.append(
+                            TranscriptionSegment(
+                                start_sec=start,
+                                end_sec=start + dur,
+                                text=text,
+                                avg_logprob=_conf_to_logprob(conf),
+                            )
+                        )
+                elif mtype == "UtteranceEnd":
+                    await self._queue.put(
+                        LiveTranscriptEvent(text="", is_final=True, speech_final=True)
+                    )
+        except Exception as exc:  # noqa: BLE001
+            log.warn("deepgram_live.recv.closed", error=str(exc))
+        finally:
+            await self._queue.put(None)
+
+    async def push(self, chunk: bytes) -> None:
+        if self._ws is not None:
+            await self._ws.send(chunk)
+
+    async def finish(self) -> None:
+        if self._ws is not None:
+            await self._ws.send(json.dumps({"type": "CloseStream"}))
+
+    async def events(self) -> AsyncIterator[LiveTranscriptEvent]:
+        while True:
+            ev = await self._queue.get()
+            if ev is None:
+                return
+            yield ev
+
+    async def result(self) -> TranscriptionResult:
+        text = " ".join(self._finals).strip()
+        dur = self._segments[-1].end_sec if self._segments else None
+        return TranscriptionResult(
+            text=text,
+            language=self._language,
+            duration_sec=dur,
+            segments=list(self._segments),
+        )
+
+    async def close(self) -> None:
+        if self._recv_task is not None:
+            self._recv_task.cancel()
+        if self._ws is not None:
+            await self._ws.close()
+
+
+class DeepgramLiveSttProvider(LiveSttProvider):
+    model_name = "deepgram-live"
+
+    def __init__(self, *, api_key: str, url: str, model: str, language: str,
+                 endpointing_ms: int) -> None:
+        if not api_key:
+            raise ValueError("Deepgram API key 누락")
+        self._api_key = api_key
+        self._url = url
+        self._model = model
+        self._language = language
+        self._endpointing_ms = endpointing_ms
+
+    def open_session(self, *, content_type: str, language: str | None) -> LiveSttSession:
+        return _DeepgramLiveSession(
+            api_key=self._api_key,
+            url=self._url,
+            model=self._model,
+            language=language or self._language,
+            endpointing_ms=self._endpointing_ms,
+            content_type=content_type,
+        )
+
+
+def _conf_to_logprob(confidence: Any) -> float | None:
+    if confidence is None:
+        return None
+    try:
+        c = float(confidence)
+    except (TypeError, ValueError):
+        return None
+    return math.log(min(c, 1.0)) if c > 0 else None