From f458162e9fd07f66532b4e44b68e2ff91e761046 Mon Sep 17 00:00:00 2001 From: SeoHyeon Date: Tue, 2 Jun 2026 18:05:45 +0900 Subject: [PATCH] =?UTF-8?q?feat(backend,ai,realtime):=20=EC=9D=8C=EC=84=B1?= =?UTF-8?q?=20=EC=8A=A4=ED=8A=B8=EB=A6=BC=20WS=20=EA=B4=80=EB=A0=A8=20?= =?UTF-8?q?=EC=9E=91=EC=97=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ai/CLAUDE.md | 7 +- ai/pyproject.toml | 1 + ai/src/ai_server/api/voice_stream.py | 136 +++++++++++++++ ai/src/ai_server/config/settings.py | 8 + ai/src/ai_server/main.py | 6 + ai/src/ai_server/messaging/runner.py | 4 + ai/src/ai_server/voice/stt/deepgram_live.py | 157 ++++++++++++++++++ ai/src/ai_server/voice/stt/live.py | 45 +++++ ai/src/ai_server/voice/stt/live_factory.py | 30 ++++ ai/src/ai_server/voice/stt/mock_live.py | 67 ++++++++ ai/tests/test_mock_live.py | 25 +++ ai/tests/test_voice_stream_endpoint.py | 46 +++++ ai/uv.lock | 2 + .../application/VoiceAnswerUploadService.java | 42 +++-- .../application/VoiceStreamService.java | 23 +++ .../dto/VoiceStreamBeginResult.java | 4 + .../presentation/VoiceAnswerController.java | 24 +++ .../dto/VoiceStreamBeginResponse.java | 9 + .../application/VoiceStreamServiceTest.java | 39 +++++ docs/data-flow.md | 20 +++ docs/environment.md | 9 + docs/event-stream.md | 11 ++ realtime/.env.example | 2 + realtime/CLAUDE.md | 6 +- realtime/cmd/realtime/main.go | 3 +- realtime/internal/config/config.go | 1 + realtime/internal/transport/router.go | 15 +- realtime/internal/transport/ws_audio.go | 84 ++++++++++ realtime/internal/transport/ws_audio_test.go | 53 ++++++ 29 files changed, 861 insertions(+), 18 deletions(-) create mode 100644 ai/src/ai_server/api/voice_stream.py create mode 100644 ai/src/ai_server/voice/stt/deepgram_live.py create mode 100644 ai/src/ai_server/voice/stt/live.py create mode 100644 ai/src/ai_server/voice/stt/live_factory.py create mode 100644 ai/src/ai_server/voice/stt/mock_live.py create mode 100644 ai/tests/test_mock_live.py create mode 100644 ai/tests/test_voice_stream_endpoint.py create mode 100644 backend/src/main/java/com/stackup/stackup/session/application/VoiceStreamService.java create mode 100644 backend/src/main/java/com/stackup/stackup/session/application/dto/VoiceStreamBeginResult.java create mode 100644 backend/src/main/java/com/stackup/stackup/session/presentation/dto/VoiceStreamBeginResponse.java create mode 100644 backend/src/test/java/com/stackup/stackup/session/application/VoiceStreamServiceTest.java create mode 100644 realtime/internal/transport/ws_audio.go create mode 100644 realtime/internal/transport/ws_audio_test.go diff --git a/ai/CLAUDE.md b/ai/CLAUDE.md index 8fc4a3a4..a67e1291 100644 --- a/ai/CLAUDE.md +++ b/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) 각 도입 시 본 문서 갱신. diff --git a/ai/pyproject.toml b/ai/pyproject.toml index 543cfeba..d6ad403a 100644 --- a/ai/pyproject.toml +++ b/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] diff --git a/ai/src/ai_server/api/voice_stream.py b/ai/src/ai_server/api/voice_stream.py new file mode 100644 index 00000000..6dd1e49e --- /dev/null +++ b/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 diff --git a/ai/src/ai_server/config/settings.py b/ai/src/ai_server/config/settings.py index cd0482b5..4601b3f1 100644 --- a/ai/src/ai_server/config/settings.py +++ b/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" diff --git a/ai/src/ai_server/main.py b/ai/src/ai_server/main.py index 9046b7ed..ef464ed9 100644 --- a/ai/src/ai_server/main.py +++ b/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 diff --git a/ai/src/ai_server/messaging/runner.py b/ai/src/ai_server/messaging/runner.py index 242279e0..36cd7186 100644 --- a/ai/src/ai_server/messaging/runner.py +++ b/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() diff --git a/ai/src/ai_server/voice/stt/deepgram_live.py b/ai/src/ai_server/voice/stt/deepgram_live.py new file mode 100644 index 00000000..707e0c3f --- /dev/null +++ b/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 diff --git a/ai/src/ai_server/voice/stt/live.py b/ai/src/ai_server/voice/stt/live.py new file mode 100644 index 00000000..d77c202a --- /dev/null +++ b/ai/src/ai_server/voice/stt/live.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import AsyncIterator +from dataclasses import dataclass + +from ai_server.voice.stt.base import TranscriptionResult + + +@dataclass(frozen=True) +class LiveTranscriptEvent: + text: str + is_final: bool # 발화(utterance) 확정 조각인지 + speech_final: bool # 발화 끝(턴 종료 신호) + + +class LiveSttSession(ABC): + """단일 스트리밍 STT 세션. push()로 오디오 청크 투입, events()로 부분/최종 수신.""" + + @abstractmethod + async def start(self) -> None: ... + + @abstractmethod + async def push(self, chunk: bytes) -> None: ... + + @abstractmethod + async def finish(self) -> None: + """오디오 입력 종료 신호(클라 stop). 남은 최종 이벤트 flush 유도.""" + + @abstractmethod + def events(self) -> AsyncIterator[LiveTranscriptEvent]: ... + + @abstractmethod + async def result(self) -> TranscriptionResult: + """세션 종료 후 누적 최종 transcript + segment(메트릭 계산용).""" + + @abstractmethod + async def close(self) -> None: ... + + +class LiveSttProvider(ABC): + model_name: str = "live-stt" + + @abstractmethod + def open_session(self, *, content_type: str, language: str | None) -> LiveSttSession: ... diff --git a/ai/src/ai_server/voice/stt/live_factory.py b/ai/src/ai_server/voice/stt/live_factory.py new file mode 100644 index 00000000..7153d8c8 --- /dev/null +++ b/ai/src/ai_server/voice/stt/live_factory.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import structlog + +from ai_server.config.settings import Settings +from ai_server.voice.stt.live import LiveSttProvider +from ai_server.voice.stt.mock_live import MockLiveSttProvider + +log = structlog.get_logger(__name__) + + +def build_live_stt_provider(settings: Settings) -> LiveSttProvider: + provider = (settings.live_stt_provider or "auto").lower() + if provider == "auto": + provider = "deepgram_live" if settings.deepgram_api_key else "mock" + + if provider == "deepgram_live": + if not settings.deepgram_api_key: + log.warn("live_stt.fallback_to_mock", reason="DEEPGRAM_API_KEY 누락") + return MockLiveSttProvider() + from ai_server.voice.stt.deepgram_live import DeepgramLiveSttProvider + + return DeepgramLiveSttProvider( + api_key=settings.deepgram_api_key, + url=settings.deepgram_live_url, + model=settings.deepgram_live_model, + language=settings.deepgram_live_language, + endpointing_ms=settings.deepgram_live_endpointing_ms, + ) + return MockLiveSttProvider() diff --git a/ai/src/ai_server/voice/stt/mock_live.py b/ai/src/ai_server/voice/stt/mock_live.py new file mode 100644 index 00000000..d6550952 --- /dev/null +++ b/ai/src/ai_server/voice/stt/mock_live.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator + +from ai_server.voice.stt.base import TranscriptionResult, TranscriptionSegment +from ai_server.voice.stt.live import LiveSttProvider, LiveSttSession, LiveTranscriptEvent + + +class _MockLiveSession(LiveSttSession): + def __init__(self, script: list[str]) -> None: + self._script = script or ["테스트 transcript"] + self._queue: asyncio.Queue[LiveTranscriptEvent | None] = asyncio.Queue() + self._pushes = 0 + self._final_text = self._script[-1] + + async def start(self) -> None: + return None + + async def push(self, chunk: bytes) -> None: + # 청크마다 부분 transcript 한 조각 방출 (스크립트 순서대로). + idx = min(self._pushes, len(self._script) - 1) + await self._queue.put( + LiveTranscriptEvent(text=self._script[idx], is_final=False, speech_final=False) + ) + self._pushes += 1 + + async def finish(self) -> None: + await self._queue.put( + LiveTranscriptEvent(text=self._final_text, is_final=True, speech_final=True) + ) + await self._queue.put(None) # 종료 센티넬 + + 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: + return TranscriptionResult( + text=self._final_text, + language="ko", + duration_sec=float(max(1, self._pushes)), + segments=[ + TranscriptionSegment( + start_sec=0.0, + end_sec=float(max(1, self._pushes)), + text=self._final_text, + avg_logprob=-0.1, + ) + ], + ) + + async def close(self) -> None: + return None + + +class MockLiveSttProvider(LiveSttProvider): + model_name = "mock-live" + + def __init__(self, script: list[str] | None = None) -> None: + self._script = script or ["부분", "부분 transcript", "부분 transcript 완성"] + + def open_session(self, *, content_type: str, language: str | None) -> LiveSttSession: + return _MockLiveSession(self._script) diff --git a/ai/tests/test_mock_live.py b/ai/tests/test_mock_live.py new file mode 100644 index 00000000..303c3481 --- /dev/null +++ b/ai/tests/test_mock_live.py @@ -0,0 +1,25 @@ +import pytest + +from ai_server.voice.stt.mock_live import MockLiveSttProvider + + +@pytest.mark.asyncio +async def test_mock_live_emits_partials_then_final(): + provider = MockLiveSttProvider(script=["안녕", "안녕하세요", "안녕하세요 반갑습니다"]) + session = provider.open_session(content_type="audio/webm", language="ko") + await session.start() + # 오디오 청크 3개 투입 + for i in range(3): + await session.push(b"audiochunk") + await session.finish() + + events = [] + async for ev in session.events(): + events.append(ev) + + assert any(not e.is_final for e in events), "부분 이벤트 존재" + assert events[-1].speech_final is True + result = await session.result() + assert result.text == "안녕하세요 반갑습니다" + assert result.segments, "메트릭용 segment 존재" + await session.close() diff --git a/ai/tests/test_voice_stream_endpoint.py b/ai/tests/test_voice_stream_endpoint.py new file mode 100644 index 00000000..28f76a43 --- /dev/null +++ b/ai/tests/test_voice_stream_endpoint.py @@ -0,0 +1,46 @@ +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from ai_server.api.voice_stream import router +from ai_server.voice.stt.mock_live import MockLiveSttProvider + + +class _FakePublisher: + def __init__(self): + self.published = [] + + async def publish(self, **kwargs): + self.published.append(kwargs) + + +class _Settings: + core_internal_api_key = "" # 빈 값이면 인증 skip + voice_filler_pattern = r"(음+|어+|그+)" + ai_callback_routing_voice = "callback.voice" + + +def _app(publisher): + app = FastAPI() + app.include_router(router) + app.state.live_stt_provider = MockLiveSttProvider(script=["안녕", "안녕하세요"]) + app.state.callback_publisher = publisher + app.state.settings = _Settings() + return app + + +def test_stream_publishes_callback_voice_on_stop(): + publisher = _FakePublisher() + client = TestClient(_app(publisher)) + with client.websocket_connect("/internal/voice/stream?sessionId=7&messageId=42") as ws: + ws.send_bytes(b"chunk1") + ws.send_bytes(b"chunk2") + # 부분 자막 수신 + first = ws.receive_json() + assert first["type"] in ("transcript.partial", "transcript.final") + ws.send_text('{"type":"stop"}') + # 연결 종료 후 callback.voice 발행되어 있어야 함 + assert publisher.published, "callback.voice 발행됨" + pub = publisher.published[0] + assert pub["message_type"] == "callback.voice" + assert pub["payload"].interview_message_id == 42 + assert pub["payload"].session_id == 7 diff --git a/ai/uv.lock b/ai/uv.lock index bc87a385..6eba5261 100644 --- a/ai/uv.lock +++ b/ai/uv.lock @@ -1967,6 +1967,7 @@ dependencies = [ { name = "structlog" }, { name = "trafilatura" }, { name = "uvicorn", extra = ["standard"] }, + { name = "websockets" }, ] [package.optional-dependencies] @@ -2002,6 +2003,7 @@ requires-dist = [ { name = "structlog", specifier = ">=25.5.0" }, { name = "trafilatura", specifier = ">=2.0.0" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.42.0" }, + { name = "websockets", specifier = ">=13" }, ] provides-extras = ["dev"] diff --git a/backend/src/main/java/com/stackup/stackup/session/application/VoiceAnswerUploadService.java b/backend/src/main/java/com/stackup/stackup/session/application/VoiceAnswerUploadService.java index 8cc83298..5287ba5e 100644 --- a/backend/src/main/java/com/stackup/stackup/session/application/VoiceAnswerUploadService.java +++ b/backend/src/main/java/com/stackup/stackup/session/application/VoiceAnswerUploadService.java @@ -41,19 +41,24 @@ public class VoiceAnswerUploadService { private final RabbitMessagePublisher publisher; private final RabbitMqProperties properties; + // 스트리밍/배치 음성 답변이 공유하는 placeholder 생성 결과. + public record VoicePlaceholder(InterviewSession session, InterviewMessage placeholder, + InterviewMessage parentQuestion) {} + + // 세션 조회 → idempotency → 상태/직전메시지 검증 → placeholder save. + // 배치 업로드(submit)와 스트리밍 시작(VoiceStreamService) 양쪽에서 재사용한다. @Transactional - public MessageResult submit(Long userId, Long sessionId, VoiceAnswerUploadCommand cmd) { - validate(cmd); + public VoicePlaceholder createVoicePlaceholder(Long userId, Long sessionId, String idempotencyKey) { InterviewSession session = sessionRepository.findByIdAndUser_IdAndDeletedFalse(sessionId, userId) .orElseThrow(() -> new DomainException(ApiErrorCode.SESSION_NOT_FOUND)); - if (cmd.idempotencyKey() != null && !cmd.idempotencyKey().isBlank()) { - var existing = messageRepository.findBySession_IdAndIdempotencyKey(sessionId, cmd.idempotencyKey()); + if (idempotencyKey != null && !idempotencyKey.isBlank()) { + var existing = messageRepository.findBySession_IdAndIdempotencyKey(sessionId, idempotencyKey); if (existing.isPresent()) { - return MessageResult.of(existing.get()); + InterviewMessage m = existing.get(); + return new VoicePlaceholder(session, m, m.getParentMessage()); } } - if (session.getStatus() != SessionStatus.IN_PROGRESS) { throw new DomainException(ApiErrorCode.SESSION_INVALID_STATE); } @@ -64,13 +69,22 @@ public MessageResult submit(Long userId, Long sessionId, VoiceAnswerUploadComman throw new DomainException(ApiErrorCode.SESSION_INVALID_STATE); } int nextSeq = latest.getSequenceNumber() + 1; - - // 메시지 ID 없이도 키를 만들어야 하므로 먼저 INSERT 후 key 갱신은 ID 의존. 단순화: 임시키로 PUT 후 INSERT. - // 더 안전한 패턴: messageRepository.save 먼저 (id 채번) → key 결정 → S3 PUT → audio_file_path 갱신. InterviewMessage placeholder = messageRepository.save( InterviewMessage.voiceInterviewee(session, nextSeq, latest, - cmd.idempotencyKey() != null && !cmd.idempotencyKey().isBlank() ? cmd.idempotencyKey() : null) + idempotencyKey != null && !idempotencyKey.isBlank() ? idempotencyKey : null) ); + return new VoicePlaceholder(session, placeholder, latest); + } + + @Transactional + public MessageResult submit(Long userId, Long sessionId, VoiceAnswerUploadCommand cmd) { + validate(cmd); + VoicePlaceholder vp = createVoicePlaceholder(userId, sessionId, cmd.idempotencyKey()); + // idempotency 재호출이면 이미 완료된 메시지일 수 있음 — 기존 동작 유지: 그대로 반환. + InterviewMessage placeholder = vp.placeholder(); + if (placeholder.getAudioFilePath() != null) { + return MessageResult.of(placeholder); // 이미 업로드됨(중복) + } String key = buildKey(sessionId, placeholder.getId(), cmd.contentType()); storage.put(key, cmd.content(), cmd.size(), cmd.contentType()); placeholder.attachAudio(key); @@ -78,12 +92,12 @@ public MessageResult submit(Long userId, Long sessionId, VoiceAnswerUploadComman AnalyzeVoicePayload payload = new AnalyzeVoicePayload( sessionId, placeholder.getId(), - latest.getId(), + vp.parentQuestion().getId(), key, cmd.contentType(), - latest.getContent(), - session.getMode().name(), - session.getJobCategory().name() + vp.parentQuestion().getContent(), + vp.session().getMode().name(), + vp.session().getJobCategory().name() ); publisher.publishToAi( properties.routingKeys().analyzeVoice(), diff --git a/backend/src/main/java/com/stackup/stackup/session/application/VoiceStreamService.java b/backend/src/main/java/com/stackup/stackup/session/application/VoiceStreamService.java new file mode 100644 index 00000000..d0e25a6d --- /dev/null +++ b/backend/src/main/java/com/stackup/stackup/session/application/VoiceStreamService.java @@ -0,0 +1,23 @@ +package com.stackup.stackup.session.application; + +import com.stackup.stackup.session.application.VoiceAnswerUploadService.VoicePlaceholder; +import com.stackup.stackup.session.application.dto.VoiceStreamBeginResult; +import com.stackup.stackup.session.domain.InterviewMessage; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; + +// 스트리밍 음성 답변 시작: placeholder 메시지만 생성(S3/analyze.voice 없음). +// 실제 오디오는 RealTime WS → AI WS 로 흐르고, 종료 시 AI 가 callback.voice 발행. +@Service +@RequiredArgsConstructor +public class VoiceStreamService { + + private final VoiceAnswerUploadService uploadService; + + public VoiceStreamBeginResult begin(Long userId, Long sessionId, String idempotencyKey) { + VoicePlaceholder vp = uploadService.createVoicePlaceholder(userId, sessionId, idempotencyKey); + InterviewMessage placeholder = vp.placeholder(); + Long parentId = vp.parentQuestion() != null ? vp.parentQuestion().getId() : null; + return new VoiceStreamBeginResult(placeholder.getId(), parentId, placeholder.getSequenceNumber()); + } +} diff --git a/backend/src/main/java/com/stackup/stackup/session/application/dto/VoiceStreamBeginResult.java b/backend/src/main/java/com/stackup/stackup/session/application/dto/VoiceStreamBeginResult.java new file mode 100644 index 00000000..e0f4cc56 --- /dev/null +++ b/backend/src/main/java/com/stackup/stackup/session/application/dto/VoiceStreamBeginResult.java @@ -0,0 +1,4 @@ +package com.stackup.stackup.session.application.dto; + +public record VoiceStreamBeginResult(Long messageId, Long parentMessageId, Integer sequenceNumber) { +} diff --git a/backend/src/main/java/com/stackup/stackup/session/presentation/VoiceAnswerController.java b/backend/src/main/java/com/stackup/stackup/session/presentation/VoiceAnswerController.java index 13c8ccd1..75d7a2b6 100644 --- a/backend/src/main/java/com/stackup/stackup/session/presentation/VoiceAnswerController.java +++ b/backend/src/main/java/com/stackup/stackup/session/presentation/VoiceAnswerController.java @@ -2,9 +2,11 @@ import com.stackup.stackup.common.security.UserPrincipal; import com.stackup.stackup.session.application.VoiceAnswerUploadService; +import com.stackup.stackup.session.application.VoiceStreamService; import com.stackup.stackup.session.application.dto.MessageResult; import com.stackup.stackup.session.application.dto.VoiceAnswerUploadCommand; import com.stackup.stackup.session.presentation.dto.MessageResponse; +import com.stackup.stackup.session.presentation.dto.VoiceStreamBeginResponse; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.responses.ApiResponse; import io.swagger.v3.oas.annotations.responses.ApiResponses; @@ -29,6 +31,7 @@ public class VoiceAnswerController { private final VoiceAnswerUploadService uploadService; + private final VoiceStreamService streamService; @Operation( operationId = "submitVoiceAnswer", @@ -61,4 +64,25 @@ public MessageResponse submit( MessageResult result = uploadService.submit(principal.userId(), sessionId, cmd); return MessageResponse.from(result); } + + @Operation( + operationId = "beginVoiceStream", + summary = "실시간 음성 답변 시작 (placeholder 생성)", + description = "WS 오디오 스트림 전에 호출. placeholder 메시지를 만들고 messageId 를 반환. " + + "이 messageId 를 WS 쿼리로 넘긴다." + ) + @ApiResponses({ + @ApiResponse(responseCode = "201", description = "placeholder 생성"), + @ApiResponse(responseCode = "404", description = "세션 없음"), + @ApiResponse(responseCode = "422", description = "세션 상태/직전 메시지 조건 불만족") + }) + @PostMapping("/stream-begin") + @ResponseStatus(HttpStatus.CREATED) + public VoiceStreamBeginResponse beginStream( + @AuthenticationPrincipal UserPrincipal principal, + @PathVariable Long sessionId, + @RequestHeader(value = "Idempotency-Key", required = false) String idempotencyKey + ) { + return VoiceStreamBeginResponse.from(streamService.begin(principal.userId(), sessionId, idempotencyKey)); + } } diff --git a/backend/src/main/java/com/stackup/stackup/session/presentation/dto/VoiceStreamBeginResponse.java b/backend/src/main/java/com/stackup/stackup/session/presentation/dto/VoiceStreamBeginResponse.java new file mode 100644 index 00000000..fa56b49c --- /dev/null +++ b/backend/src/main/java/com/stackup/stackup/session/presentation/dto/VoiceStreamBeginResponse.java @@ -0,0 +1,9 @@ +package com.stackup.stackup.session.presentation.dto; + +import com.stackup.stackup.session.application.dto.VoiceStreamBeginResult; + +public record VoiceStreamBeginResponse(Long messageId, Long parentMessageId, Integer sequenceNumber) { + public static VoiceStreamBeginResponse from(VoiceStreamBeginResult r) { + return new VoiceStreamBeginResponse(r.messageId(), r.parentMessageId(), r.sequenceNumber()); + } +} diff --git a/backend/src/test/java/com/stackup/stackup/session/application/VoiceStreamServiceTest.java b/backend/src/test/java/com/stackup/stackup/session/application/VoiceStreamServiceTest.java new file mode 100644 index 00000000..94dd06d7 --- /dev/null +++ b/backend/src/test/java/com/stackup/stackup/session/application/VoiceStreamServiceTest.java @@ -0,0 +1,39 @@ +package com.stackup.stackup.session.application; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.when; + +import com.stackup.stackup.session.application.VoiceAnswerUploadService.VoicePlaceholder; +import com.stackup.stackup.session.application.dto.VoiceStreamBeginResult; +import com.stackup.stackup.session.domain.InterviewMessage; +import com.stackup.stackup.session.domain.InterviewSession; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class VoiceStreamServiceTest { + + @Mock VoiceAnswerUploadService uploadService; + @InjectMocks VoiceStreamService service; + + @Test + void begin_createsPlaceholder_returnsIds() { + InterviewSession session = org.mockito.Mockito.mock(InterviewSession.class); + InterviewMessage placeholder = org.mockito.Mockito.mock(InterviewMessage.class); + InterviewMessage parent = org.mockito.Mockito.mock(InterviewMessage.class); + when(placeholder.getId()).thenReturn(50L); + when(placeholder.getSequenceNumber()).thenReturn(4); + when(parent.getId()).thenReturn(49L); + when(uploadService.createVoicePlaceholder(1L, 7L, "idem-1")) + .thenReturn(new VoicePlaceholder(session, placeholder, parent)); + + VoiceStreamBeginResult result = service.begin(1L, 7L, "idem-1"); + + assertThat(result.messageId()).isEqualTo(50L); + assertThat(result.parentMessageId()).isEqualTo(49L); + assertThat(result.sequenceNumber()).isEqualTo(4); + } +} diff --git a/docs/data-flow.md b/docs/data-flow.md index 9a606293..d080c7f8 100644 --- a/docs/data-flow.md +++ b/docs/data-flow.md @@ -83,6 +83,26 @@ → [Core] 영속 후 RealtimeNotifyPublisher.publishToSession → [RealTime] WS/SSE → 다음 질문 push ``` +#### 3.2-1 음성 답변 — 실시간 스트리밍 분기 (RT3, Phase 2) + +배치 업로드(`POST .../messages/voice` multipart → `analyze.voice`) 대신, 마이크 오디오를 WS로 흘려 실시간 자막 + 즉시 다음 질문을 받는 저지연 경로. + +``` +[Frontend] POST /api/sessions/{id}/messages/voice/stream-begin + → [Core] placeholder voiceInterviewee("(transcribing)") INSERT → { messageId, parentMessageId } 반환 +[Frontend] WS /realtime/sessions/{id}/audio?access_token=&messageId=M + → [RealTime] WSAudioHandler: 브라우저 WS ↔ AI WS(/internal/voice/stream) 순수 오디오 프록시 (PG·MQ 미접근) + → [Frontend] 바이너리 오디오 프레임 업 ↕ AI가 transcript.partial/final JSON 다운 +[AI] DeepgramLive(또는 MockLive)로 부분 자막 실시간 반환 + → [Frontend] {"type":"stop"} OR Deepgram UtteranceEnd → 발화 종료 +[AI] 최종 transcript + 메트릭(WPM/filler/silence) 계산 + → RabbitMQ publish: stackup.ai-to-core / callback.voice { sessionId, interviewMessageId, transcript, metrics } +[Core] VoiceCallbackService(무변경) → placeholder completeWithTranscript + message_voice_analyses INSERT + → AnswerSubmittedEvent → generate.followup → 기존 followup 파이프라인(다음 질문 + Phase1 TTS) 재사용 +``` + +> `callback.voice` 이후는 배치 경로와 100% 동일(`VoiceCallbackService` 재사용). 신규는 "오디오가 AI에 닿는 경로(브라우저→RealTime WS→AI WS)" + Deepgram Live STT 뿐. + ### 3.3 세션 종료 ``` diff --git a/docs/environment.md b/docs/environment.md index c4f990d7..c1c861de 100644 --- a/docs/environment.md +++ b/docs/environment.md @@ -150,6 +150,14 @@ TTS_PROVIDER=auto # auto | mock | openai (auto=OPENAI_API_KEY OPENAI_TTS_MODEL=gpt-4o-mini-tts # OpenAI TTS 모델 (질문 음성화) OPENAI_TTS_VOICE=alloy # TTS 보이스 OPENAI_TTS_TIMEOUT_SEC=30 # TTS HTTP 타임아웃(초) + +# ===== 실시간 스트리밍 STT (RT3 음성 답변, Phase 2) ===== +LIVE_STT_PROVIDER=auto # auto | mock | deepgram_live (auto=DEEPGRAM_API_KEY 보유 시 deepgram_live, 없으면 mock) +DEEPGRAM_API_KEY= # Deepgram streaming API 키 (live STT) +DEEPGRAM_LIVE_URL=wss://api.deepgram.com/v1/listen # Deepgram Live WS 엔드포인트 +DEEPGRAM_LIVE_MODEL=nova-2 # 스트리밍 모델 (저지연, 한국어 지원) +DEEPGRAM_LIVE_LANGUAGE=ko # 인식 언어 +DEEPGRAM_LIVE_ENDPOINTING_MS=800 # 무음 N ms → utterance end ``` --- @@ -181,6 +189,7 @@ VITE_SENTRY_DSN= # 옵션 | `REALTIME_SSE_PING_INTERVAL` | `30s` | SSE heartbeat 주기 | | `REALTIME_SSE_SLOW_CONSUMER_TIMEOUT` | `5s` | 구독자 send timeout | | `REALTIME_SSE_BUFFER_SIZE` | `16` | 구독자별 채널 버퍼 | +| `REALTIME_AI_WS_URL` | `ws://localhost:8000/internal/voice/stream` | AI 음성 스트림 WS base URL (RT3 오디오 프록시 업스트림). compose 내부는 `ws://ai:8000/internal/voice/stream` | 루트 `.env.example`은 `REALTIME_PORT`, `REALTIME_LOG_LEVEL` 만 노출 (compose에서 매핑). diff --git a/docs/event-stream.md b/docs/event-stream.md index 6158190b..8c255e7d 100644 --- a/docs/event-stream.md +++ b/docs/event-stream.md @@ -13,6 +13,7 @@ GET /realtime/stream/me # user 채널 SSE (userId는 토큰에 GET /realtime/stream/sessions/{sessionId} # session 채널 SSE (feedback.ready 등 비-라이브) GET /realtime/stream/documents/{documentId} WS /realtime/sessions/{sessionId} # RT1 라이브 텍스트 면접 (양방향) +WS /realtime/sessions/{sessionId}/audio # RT3 실시간 음성 답변 스트림 (오디오 업/자막 다운) ``` - 제공 주체: **RealTime Server (Go)**. Core는 직접 SSE를 서빙하지 않고 `stackup.realtime` exchange로 발행만 한다 (RealTime이 consume → fan-out). @@ -25,6 +26,16 @@ WS /realtime/sessions/{sessionId} # RT1 라이브 텍스트 면접 (양 - 권한 (리소스 스코프): 토큰은 `resourceType`(`USER`/`SESSION`)·`resourceId` claim을 담는다. 소유권은 **발급 시점**에 Core가 검증하고(USER=`POST /api/auth/stream-token` 본인, SESSION=`POST /api/sessions/{id}/stream-token` 소유권 체크 후 발급), RealTime은 path 리소스와 토큰 리소스의 일치만 확인한다(불일치 → 403). 이로써 RealTime은 PG 무접근으로 소유권을 판정한다. `documents/{id}` 채널 스코프는 MVP deferred(인증 토큰만). - 연결 유지: 약 30초마다 `: ping ` heartbeat 코멘트 송신 (`REALTIME_SSE_PING_INTERVAL`). +### WS 실시간 음성 스트림 (RT3) +- 경로 `WS /realtime/sessions/{id}/audio?access_token=&messageId=`. 인증·리소스 스코프는 RT1과 동일(`SESSION` 토큰 ↔ path id 일치). `messageId`는 사전에 Core `POST /api/sessions/{id}/messages/voice/stream-begin`이 만든 placeholder 메시지 id. +- RealTime은 **순수 오디오 양방향 프록시**다(오디오 내용을 해석하지 않음). 브라우저 WS ↔ AI WS(`/internal/voice/stream`)를 프레임 타입 보존하며 양방향 복사한다. +- **업(클라→서버, 바이너리)**: 마이크 오디오 프레임(예: `audio/webm` opus). 프레임을 그대로 AI로 전달한다. +- **다운(서버→클라, JSON)** — AI가 발행, RealTime이 그대로 전달: + - `{ "type": "transcript.partial", "text": "안녕하세" }` — 부분(interim) 자막 + - `{ "type": "transcript.final", "text": "안녕하세요", "messageId": 50 }` — 발화(utterance) 확정 조각 + - `{ "type": "error", "code": "..." }` — 처리 오류 +- **제어(클라→서버, 텍스트)**: `{ "type": "stop" }` — 발화 종료. AI가 남은 최종 transcript flush 후 메트릭(WPM/filler/silence)을 계산해 `callback.voice`를 발행하고 연결을 닫는다. 이후 기존 followup 파이프라인(다음 질문 생성)으로 이어진다. + --- ## 2. 이벤트 포맷 diff --git a/realtime/.env.example b/realtime/.env.example index f39c2a3f..401c953a 100644 --- a/realtime/.env.example +++ b/realtime/.env.example @@ -13,3 +13,5 @@ REALTIME_JWT_SECRET=local-development-jwt-secret-must-be-replaced REALTIME_CORE_BASE_URL=http://localhost:38010 REALTIME_INTERNAL_API_KEY=local-development-internal-api-key REALTIME_WS_WRITE_TIMEOUT=10s +# AI 음성 스트림 WS 프록시 대상 (compose 내부에서는 ws://ai:8000/internal/voice/stream) +REALTIME_AI_WS_URL=ws://ai:8000/internal/voice/stream diff --git a/realtime/CLAUDE.md b/realtime/CLAUDE.md index 73ee1037..acd9ad26 100644 --- a/realtime/CLAUDE.md +++ b/realtime/CLAUDE.md @@ -81,6 +81,7 @@ config는 cmd, internal/* 모두에서 import 가능 - ❌ 비즈니스 로직 (질문 생성, 분석) — AI 또는 Core - ❌ RabbitMQ에 publish — Core를 통해서만 (`architecture.md §4.1`) - ❌ PostgreSQL 접근 — WS 답변도 Core 내부 REST(`POST /api/internal/sessions/{id}/messages`)로 프록시. RealTime은 PG·MQ publish 모두 미접근 +- ❌ AI WS 오디오 프록시(RT3) — 오디오 전용 순수 바이트 파이프, 비즈니스 로직 없음. 음성 인식·메트릭·콜백 발행은 모두 AI 책임. RealTime은 브라우저↔AI WS 양방향 복사만 --- @@ -93,10 +94,11 @@ config는 cmd, internal/* 모두에서 import 가능 | RT2 | GET | `/realtime/stream/documents/{id}` | document 채널 SSE | 활성 | | RT2 | GET | `/realtime/stream/sessions/{id}` | session 채널 SSE (feedback.ready 등 비-라이브) | 활성 | | RT1 | WS | `/realtime/sessions/{id}` | 라이브 텍스트 면접 (서버→클라 push + 클라→서버 답변) | 활성 | -| RT3 | WS | `/realtime/sessions/{id}/audio` | 음성 스트림 | **미구현** (US-Voice-01) | +| RT3 | WS | `/realtime/sessions/{id}/audio` | 실시간 음성 답변 스트림 (오디오 업 ↔ 자막 다운, AI WS 프록시) | 활성 | > 모든 `/realtime/stream/*` (SSE) 및 `/realtime/sessions/{id}` (WS) 는 인증 필요 — `?access_token=` 쿼리로 Core 발급 토큰 검증 (EventSource/WS 헤더 한계 우회). `internal/auth` 미들웨어가 처리. > WS는 SSE(`/realtime/stream/*`)와 **다른 path**라 Upgrade 헤더 분기가 필요 없다. WS 핸들러(`coder/websocket`)는 session 채널 fan-out을 그대로 구독(서버→클라)하고, 수신한 답변(`{type:"answer",content,idempotencyKey?}`)을 `internal/core` 클라이언트로 Core 내부 REST(`POST /api/internal/sessions/{id}/messages`)에 프록시한다. +> RT3(`/realtime/sessions/{id}/audio`)는 `WSAudioHandler`(`transport/ws_audio.go`)가 브라우저 WS를 AI WS(`REALTIME_AI_WS_URL` + `?sessionId=&messageId=`, `X-Internal-API-Key` 헤더)로 **양방향 프록시**한다. `messageId`는 사전에 Core `POST /api/sessions/{id}/messages/voice/stream-begin`이 만든 placeholder. 오디오 내용은 해석하지 않고 프레임 타입을 보존하며 복사만(`copyWS`). --- @@ -154,6 +156,7 @@ Heartbeat (proxy keepalive): | `REALTIME_JWT_SECRET` | `local-development-jwt-secret-must-be-replaced` | **Core `JWT_SECRET`과 동일값 필수** (Stream 토큰 검증 키. 키=`SHA-256(secret)`, HS256). 불일치 시 SSE/WS 인증 401 | | `REALTIME_CORE_BASE_URL` | `http://localhost:38010` | Core 내부 REST base URL (WS 답변 프록시). compose 내부는 `http://backend:38010` | | `REALTIME_INTERNAL_API_KEY` | `local-development-internal-api-key` | **Core `CORE_INTERNAL_API_KEY`와 동일값 필수** (AI 서버도 공유). `X-Internal-API-Key` | +| `REALTIME_AI_WS_URL` | `ws://localhost:8000/internal/voice/stream` | AI 음성 스트림 WS base URL (RT3 오디오 프록시 업스트림). compose 내부는 `ws://ai:8000/internal/voice/stream` | | `REALTIME_WS_WRITE_TIMEOUT` | `10s` | WS write 타임아웃 | --- @@ -218,6 +221,7 @@ docker build -t stackup-realtime ./realtime - Stream 토큰 인증 활성 — `?access_token=` 검증 (`internal/auth`, Core와 동일 HS256 규약) - AMQP `q.realtime.session.notify` consumer 활성 — `messageType` 기반 채널 라우팅 → fan-out - WebSocket(RT1 라이브 면접) 활성 — `/realtime/sessions/{id}` 서버→클라 push + 클라→서버 답변 프록시(Core 내부 REST) +- WebSocket(RT3 실시간 음성) 활성 — `/realtime/sessions/{id}/audio` 브라우저↔AI WS 오디오 프록시(`WSAudioHandler`, `REALTIME_AI_WS_URL`). 오디오 전용 순수 파이프, STT·메트릭·`callback.voice`는 AI 책임 - 리소스 소유권 검증 미구현 — 현재 토큰 진위(userId)만 검증. 후속 플랜에서 리소스 스코프 토큰 또는 Core 조회로 강화 - DLQ 활성 — handler 실패 메시지는 `dlq.q.realtime.session.notify` 로 격리 - Prometheus 노출 미구현 diff --git a/realtime/cmd/realtime/main.go b/realtime/cmd/realtime/main.go index 45a252d8..1eecbd0a 100644 --- a/realtime/cmd/realtime/main.go +++ b/realtime/cmd/realtime/main.go @@ -41,7 +41,8 @@ func main() { verifier := auth.NewStreamTokenVerifier(cfg.JWTSecret) coreClient := core.NewClient(cfg.CoreBaseURL, cfg.InternalApiKey, cfg.WSWriteTimeout) wsHandler := transport.NewWSHandler(registry, coreClient, cfg.WSWriteTimeout) - router := transport.NewRouter(sseHandler, wsHandler, verifier) + audioHandler := transport.NewWSAudioHandler(cfg.AIWSURL, cfg.InternalApiKey, cfg.WSWriteTimeout) + router := transport.NewRouter(sseHandler, wsHandler, audioHandler, verifier) srv := &http.Server{ Addr: cfg.ListenAddr, diff --git a/realtime/internal/config/config.go b/realtime/internal/config/config.go index 643f77e6..fa15ac8c 100644 --- a/realtime/internal/config/config.go +++ b/realtime/internal/config/config.go @@ -14,6 +14,7 @@ type Config struct { JWTSecret string `env:"REALTIME_JWT_SECRET" envDefault:"local-development-jwt-secret-must-be-replaced"` CoreBaseURL string `env:"REALTIME_CORE_BASE_URL" envDefault:"http://localhost:38010"` InternalApiKey string `env:"REALTIME_INTERNAL_API_KEY" envDefault:"local-development-internal-api-key"` + AIWSURL string `env:"REALTIME_AI_WS_URL" envDefault:"ws://localhost:8000/internal/voice/stream"` WSWriteTimeout time.Duration `env:"REALTIME_WS_WRITE_TIMEOUT" envDefault:"10s"` SSEPingInterval time.Duration `env:"REALTIME_SSE_PING_INTERVAL" envDefault:"30s"` SSESlowConsumerTimeout time.Duration `env:"REALTIME_SSE_SLOW_CONSUMER_TIMEOUT" envDefault:"5s"` diff --git a/realtime/internal/transport/router.go b/realtime/internal/transport/router.go index 7b1d005c..cd036ed0 100644 --- a/realtime/internal/transport/router.go +++ b/realtime/internal/transport/router.go @@ -12,7 +12,7 @@ import ( "github.com/go-chi/chi/v5/middleware" ) -func NewRouter(sse *SSEHandler, ws *WSHandler, verifier *auth.StreamTokenVerifier) http.Handler { +func NewRouter(sse *SSEHandler, ws *WSHandler, audio *WSAudioHandler, verifier *auth.StreamTokenVerifier) http.Handler { r := chi.NewRouter() r.Use(middleware.Recoverer) r.Use(trace.Middleware) @@ -58,6 +58,19 @@ func NewRouter(sse *SSEHandler, ws *WSHandler, verifier *auth.StreamTokenVerifie } ws.ServeWS(w, req) }) + pr.Get("/realtime/sessions/{id}/audio", func(w http.ResponseWriter, req *http.Request) { + id, err := strconv.ParseInt(chi.URLParam(req, "id"), 10, 64) + if err != nil || id <= 0 { + http.Error(w, "invalid id", http.StatusBadRequest) + return + } + c, _ := auth.ClaimsFromContext(req.Context()) + if c.ResourceType != "SESSION" || c.ResourceID != id { + http.Error(w, "token resource mismatch", http.StatusForbidden) + return + } + audio.ServeAudioWS(w, req) + }) }) return r diff --git a/realtime/internal/transport/ws_audio.go b/realtime/internal/transport/ws_audio.go new file mode 100644 index 00000000..902336b4 --- /dev/null +++ b/realtime/internal/transport/ws_audio.go @@ -0,0 +1,84 @@ +package transport + +import ( + "context" + "fmt" + "log/slog" + "net/http" + "strconv" + "time" + + "github.com/Team-StackUp/stackup/realtime/internal/auth" + "github.com/Team-StackUp/stackup/realtime/internal/trace" + "github.com/coder/websocket" + "github.com/go-chi/chi/v5" +) + +// WSAudioHandler 는 브라우저 WS(오디오 업/자막 다운)를 AI WS 로 양방향 프록시한다. +// RealTime 은 오디오 내용을 해석하지 않는다 (순수 바이트 파이프 + 종료 처리). +type WSAudioHandler struct { + AIBaseURL string + InternalKey string + WriteTimeout time.Duration +} + +func NewWSAudioHandler(aiWSURL, internalKey string, writeTimeout time.Duration) *WSAudioHandler { + return &WSAudioHandler{AIBaseURL: aiWSURL, InternalKey: internalKey, WriteTimeout: writeTimeout} +} + +func (h *WSAudioHandler) ServeAudioWS(w http.ResponseWriter, r *http.Request) { + sid, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil || sid <= 0 { + http.Error(w, "invalid session id", http.StatusBadRequest) + return + } + mid, err := strconv.ParseInt(r.URL.Query().Get("messageId"), 10, 64) + if err != nil || mid <= 0 { + http.Error(w, "invalid messageId", http.StatusBadRequest) + return + } + userID := auth.UserIDFromContext(r.Context()) + traceID := trace.FromContext(r.Context()) + + client, err := websocket.Accept(w, r, nil) + if err != nil { + slog.Warn("ws_audio.accept.failed", "err", err) + return + } + defer client.CloseNow() + + ctx := r.Context() + aiURL := fmt.Sprintf("%s?sessionId=%d&messageId=%d", h.AIBaseURL, sid, mid) + upstream, _, err := websocket.Dial(ctx, aiURL, &websocket.DialOptions{ + HTTPHeader: http.Header{"X-Internal-API-Key": {h.InternalKey}}, + }) + if err != nil { + slog.Warn("ws_audio.dial_ai.failed", "err", err, "session_id", sid) + _ = client.Close(websocket.StatusInternalError, "ai upstream unavailable") + return + } + defer upstream.CloseNow() + slog.Info("ws_audio.proxy.start", "session_id", sid, "message_id", mid, "user_id", userID, "trace_id", traceID) + + errc := make(chan error, 2) + // 브라우저 → AI (오디오 업) + go func() { errc <- copyWS(ctx, client, upstream) }() + // AI → 브라우저 (자막 다운) + go func() { errc <- copyWS(ctx, upstream, client) }() + + <-errc // 한쪽이 끝나면 종료 + slog.Info("ws_audio.proxy.end", "session_id", sid, "message_id", mid) +} + +// copyWS 는 src 에서 받은 프레임을 dst 로 그대로 전달한다 (타입 보존). +func copyWS(ctx context.Context, src, dst *websocket.Conn) error { + for { + typ, data, err := src.Read(ctx) + if err != nil { + return err + } + if err := dst.Write(ctx, typ, data); err != nil { + return err + } + } +} diff --git a/realtime/internal/transport/ws_audio_test.go b/realtime/internal/transport/ws_audio_test.go new file mode 100644 index 00000000..bd8d71a2 --- /dev/null +++ b/realtime/internal/transport/ws_audio_test.go @@ -0,0 +1,53 @@ +package transport + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/coder/websocket" +) + +// TestCopyWS_ForwardsFrames 는 coder/websocket 왕복으로 프레임이 그대로 전달되는지 검증한다. +// fake 서버는 한 프레임을 받아 그대로 echo 하고, copyWS 가 쓰는 Read/Write 의 기본 흐름을 +// 통해 바이너리 페이로드가 손실 없이 돌아오는지 확인한다. +func TestCopyWS_ForwardsFrames(t *testing.T) { + srvA := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c, err := websocket.Accept(w, r, nil) + if err != nil { + return + } + defer c.CloseNow() + ctx := r.Context() + // 한 프레임 받고 echo + _, data, err := c.Read(ctx) + if err != nil { + return + } + _ = c.Write(ctx, websocket.MessageBinary, data) + <-ctx.Done() + })) + defer srvA.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + url := "ws" + srvA.URL[len("http"):] + conn, _, err := websocket.Dial(ctx, url, nil) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.CloseNow() + + if err := conn.Write(ctx, websocket.MessageBinary, []byte("hi")); err != nil { + t.Fatalf("write: %v", err) + } + _, data, err := conn.Read(ctx) + if err != nil { + t.Fatalf("read: %v", err) + } + if string(data) != "hi" { + t.Fatalf("want hi got %s", data) + } +}