Feature/interview voice streaming

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

@@ -5,18 +5,39 @@
 import structlog
 from aio_pika.abc import AbstractIncomingMessage
 
+from ai_server.chain.sentence_split import next_sentences
 from ai_server.messaging.idempotency import LruIdempotencyStore
 from ai_server.messaging.publisher import CallbackPublisher
+from ai_server.messaging.session_notify import SessionRealtimeNotifier
 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
+from ai_server.voice.tts.audio_join import join_audio
+from ai_server.voice.tts.base import TtsError, TtsProvider, TtsResult
 
 log = structlog.get_logger(__name__)
 
 
+def _split_all_sentences(text: str) -> list[str]:
+    """완성된 텍스트를 문장 단위로 모두 쪼갠다(스트리밍과 달리 마지막 문장도 포함)."""
+    sents, consumed = next_sentences(text, 0)
+    tail = text[consumed:].strip()
+    if tail:
+        sents.append(tail)
+    cleaned = [s for s in sents if s.strip()]
+    if cleaned:
+        return cleaned
+    stripped = text.strip()
+    return [stripped] if stripped else []
+
+
 class TtsConsumer:
-    """generate.tts consumer — 질문 텍스트를 TTS 합성 → S3 PUT → callback.tts 발행."""
+    """generate.tts consumer — 질문 텍스트를 TTS 합성 → S3 PUT → callback.tts 발행.
+
+    session_notifier 가 주입되면 문장 단위 세그먼트로 합성·발행해(SESSION_MESSAGE_AUDIO)
+    첫 문장부터 즉시 재생되게 하고, 합성된 세그먼트를 이어붙여 전체 파일도 저장한다(다시 듣기용).
+    notifier 가 없으면 기존처럼 전체 텍스트를 한 번에 합성한다.
+    """
 
     def __init__(
         self,
@@ -28,6 +49,7 @@ def __init__(
         callback_routing_key: str,
         voice: str,
         key_template: str,
+        session_notifier: SessionRealtimeNotifier | None = None,
     ) -> None:
         self._tts = tts
         self._storage = storage
@@ -36,6 +58,7 @@ def __init__(
         self._callback_routing_key = callback_routing_key
         self._voice = voice
         self._key_template = key_template
+        self._notifier = session_notifier
 
     async def handle(self, message: AbstractIncomingMessage) -> None:
         async with message.process(requeue=False):
@@ -61,44 +84,16 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 message_id=envelope.message_id,
                 session_id=req.session_id,
                 interview_message_id=req.message_id,
+                streaming=self._notifier is not None,
                 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
+            if self._notifier is not None:
+                result = await self._synthesize_segmented(envelope, req)
+            else:
+                result = await self._synthesize_once(envelope, req)
+            if result is None:
+                return  # 실패 — 콜백은 내부에서 이미 발행
 
             key = self._build_key(req.session_id, req.message_id, result.content_type)
             try:
@@ -135,6 +130,122 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 key=key,
             )
 
+    async def _synthesize_once(
+        self, envelope: Envelope[GenerateTtsRequest], req: GenerateTtsRequest
+    ) -> TtsResult | None:
+        """전체 텍스트를 한 번에 합성(레거시/notifier 미주입 경로). 실패 시 콜백 발행 후 None."""
+        try:
+            return 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 None
+        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 None
+
+    async def _synthesize_segmented(
+        self, envelope: Envelope[GenerateTtsRequest], req: GenerateTtsRequest
+    ) -> TtsResult | None:
+        """문장 단위로 순차 합성 → 세그먼트 PUT + SESSION_MESSAGE_AUDIO 즉시 발행.
+
+        seq 는 '성공한' 세그먼트만 0..n 연속 부여해 프론트 큐가 빈 seq 에서 멈추지 않게 한다.
+        합성된 세그먼트를 이어붙여 전체 파일(다시 듣기용)을 반환한다. 전부 실패하면 콜백 발행 후 None.
+        """
+        sentences = _split_all_sentences(req.text)
+        if not sentences:
+            await self._publish(
+                envelope,
+                TtsCallbackPayload(
+                    session_id=req.session_id,
+                    message_id=req.message_id,
+                    status="FAILED",
+                    error_code="TTS_EMPTY_TEXT",
+                ),
+            )
+            return None
+
+        parts: list[TtsResult] = []
+        seq = 0
+        for sentence in sentences:
+            try:
+                res = await self._tts.synthesize(sentence, voice=self._voice)
+            except Exception as exc:  # noqa: BLE001 — 한 문장 실패가 전체를 막지 않음
+                log.warning(
+                    "tts.segment.failed",
+                    session_id=req.session_id,
+                    interview_message_id=req.message_id,
+                    seq=seq,
+                    error=str(exc),
+                )
+                continue
+            ext = self._EXT_BY_CONTENT_TYPE.get(
+                res.content_type.split(";")[0].strip(), "mp3"
+            )
+            seg_key = (
+                f"interview/tts/{req.session_id}/{req.message_id}/seg-{seq}.{ext}"
+            )
+            try:
+                await self._storage.put_bytes(
+                    seg_key, res.audio_bytes, content_type=res.content_type
+                )
+            except Exception as exc:  # noqa: BLE001
+                log.warning(
+                    "tts.segment.storage_failed",
+                    key=seg_key,
+                    error=str(exc),
+                )
+                continue
+            await self._notifier.emit_audio(
+                session_id=req.session_id,
+                message_id=req.message_id,
+                seq=seq,
+                ext=ext,
+                duration_sec=res.duration_sec,
+                trace_id=envelope.trace_id,
+            )
+            parts.append(res)
+            seq += 1
+
+        if not parts:
+            await self._publish(
+                envelope,
+                TtsCallbackPayload(
+                    session_id=req.session_id,
+                    message_id=req.message_id,
+                    status="FAILED",
+                    error_code="TTS_FAILED",
+                ),
+            )
+            return None
+        return join_audio(parts)
+
     _EXT_BY_CONTENT_TYPE = {
         "audio/wav": "wav",
         "audio/mpeg": "mp3",

ai/src/ai_server/messaging/runner.py

@@ -236,7 +236,8 @@ def __init__(self, settings: Settings) -> None:
             core_client=core_client,
         )
 
-        # 질문 TTS (Part A) — tts 인스턴스는 위에서 이미 생성됨
+        # 질문 TTS (Part A) — tts 인스턴스는 위에서 이미 생성됨.
+        # notifier 주입 시 문장 단위 세그먼트로 발행해 첫 문장부터 즉시 재생되게 한다.
         self._tts_consumer = TtsConsumer(
             tts=tts,
             storage=storage,
@@ -245,6 +246,7 @@ def __init__(self, settings: Settings) -> None:
             callback_routing_key=settings.ai_callback_routing_tts,
             voice=settings.openai_tts_voice,
             key_template=settings.tts_audio_key_template,
+            session_notifier=session_notifier,
         )
 
         self._consumers: list[tuple[AbstractRobustQueue, str]] = []

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

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import io
+import wave
+
+from ai_server.voice.tts.base import TtsResult
+
+
+def join_audio(parts: list[TtsResult]) -> TtsResult:
+    """여러 세그먼트 오디오를 하나로 합친다(재합성 없음).
+
+    - WAV(Gemini/게이트웨이): PCM 프레임을 이어붙여 단일 WAV 로 — 헤더 중복 없는 정상 파일.
+    - 그 외(mp3 등): 컨테이너 바이트 단순 연결(best-effort, 대부분 플레이어가 재생).
+    """
+    if not parts:
+        raise ValueError("join_audio: empty parts")
+    if len(parts) == 1:
+        return parts[0]
+
+    total_dur = _sum_dur(parts)
+    if all(_ctype(p) == "audio/wav" for p in parts):
+        return TtsResult(
+            audio_bytes=_concat_wav([p.audio_bytes for p in parts]),
+            duration_sec=total_dur,
+            content_type="audio/wav",
+        )
+    return TtsResult(
+        audio_bytes=b"".join(p.audio_bytes for p in parts),
+        duration_sec=total_dur,
+        content_type=parts[0].content_type,
+    )
+
+
+def _ctype(part: TtsResult) -> str:
+    return part.content_type.split(";")[0].strip()
+
+
+def _sum_dur(parts: list[TtsResult]) -> float | None:
+    if any(p.duration_sec is None for p in parts):
+        return None
+    return round(sum(p.duration_sec or 0.0 for p in parts), 2)
+
+
+def _concat_wav(wavs: list[bytes]) -> bytes:
+    frames: list[bytes] = []
+    params = None
+    for w in wavs:
+        with wave.open(io.BytesIO(w), "rb") as reader:
+            if params is None:
+                params = reader.getparams()
+            frames.append(reader.readframes(reader.getnframes()))
+    assert params is not None
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as out:
+        out.setnchannels(params.nchannels)
+        out.setsampwidth(params.sampwidth)
+        out.setframerate(params.framerate)
+        out.writeframes(b"".join(frames))
+    return buf.getvalue()