diff --git a/ai/src/ai_server/messaging/consumers/tts_consumer.py b/ai/src/ai_server/messaging/consumers/tts_consumer.py index 9fa3b41b..e0a3f2db 100644 --- a/ai/src/ai_server/messaging/consumers/tts_consumer.py +++ b/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", diff --git a/ai/src/ai_server/messaging/runner.py b/ai/src/ai_server/messaging/runner.py index 9217323d..372a54bd 100644 --- a/ai/src/ai_server/messaging/runner.py +++ b/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]] = [] diff --git a/ai/src/ai_server/voice/tts/audio_join.py b/ai/src/ai_server/voice/tts/audio_join.py new file mode 100644 index 00000000..a9a5bd53 --- /dev/null +++ b/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() diff --git a/ai/tests/test_tts_consumer.py b/ai/tests/test_tts_consumer.py index 20d8972d..b8c0cd8a 100644 --- a/ai/tests/test_tts_consumer.py +++ b/ai/tests/test_tts_consumer.py @@ -1,4 +1,6 @@ +import io import json +import wave from datetime import datetime, timezone import pytest @@ -17,6 +19,43 @@ async def synthesize(self, text: str, *, voice: str) -> TtsResult: ) +def _wav_bytes(n_frames: int) -> bytes: + buf = io.BytesIO() + with wave.open(buf, "wb") as w: + w.setnchannels(1) + w.setsampwidth(2) + w.setframerate(24000) + w.writeframes(b"\x00\x01" * n_frames) + return buf.getvalue() + + +class _PerSentenceWavTts(TtsProvider): + """문장마다 길이가 다른 정상 WAV 를 돌려준다(합본 검증용).""" + + model_name = "fake-wav-seg" + + def __init__(self): + self.calls: list[str] = [] + + async def synthesize(self, text: str, *, voice: str) -> TtsResult: + self.calls.append(text) + return TtsResult( + audio_bytes=_wav_bytes(100 * len(self.calls)), + duration_sec=0.5, + content_type="audio/wav", + ) + + +class _RecordingNotifier: + def __init__(self): + self.audio: list[dict] = [] + + async def emit_audio(self, *, session_id, message_id, seq, ext, duration_sec, trace_id): + self.audio.append( + {"session_id": session_id, "message_id": message_id, "seq": seq, "ext": ext} + ) + + class _FakeStorage: def __init__(self): self.puts = [] @@ -56,7 +95,7 @@ async def __aexit__(self_, *a): return _Ctx() -def _envelope_body(): +def _envelope_body(text: str = "질문 본문"): return json.dumps( { "messageId": "m-1", @@ -68,7 +107,7 @@ def _envelope_body(): "payload": { "sessionId": 7, "messageId": 42, - "text": "질문 본문", + "text": text, "mode": "TECHNICAL", "jobCategory": "BACKEND", }, @@ -127,3 +166,86 @@ async def test_tts_consumer_uses_wav_extension_for_wav_audio(): assert key == "interview/tts/7/42.wav" assert ctype == "audio/wav" assert publisher.published[0][2].audio_key == "interview/tts/7/42.wav" + + +@pytest.mark.asyncio +async def test_tts_consumer_streams_segments_then_assembles_full_file(): + # notifier 주입 시: 문장마다 세그먼트 PUT + emit_audio, 마지막에 전체 파일 PUT + callback. + storage = _FakeStorage() + publisher = _FakePublisher() + notifier = _RecordingNotifier() + tts = _PerSentenceWavTts() + consumer = TtsConsumer( + tts=tts, + storage=storage, + publisher=publisher, + idempotency=_FakeIdem(), + callback_routing_key="callback.tts", + voice="alloy", + key_template="interview/tts/{session_id}/{message_id}.mp3", + session_notifier=notifier, + ) + + await consumer.handle(_FakeMsg(_envelope_body("첫 문장입니다. 둘째 문장입니다."))) + + # 문장 2개 → 세그먼트 2개 합성 + assert tts.calls == ["첫 문장입니다.", "둘째 문장입니다."] + # seq 0,1 로 라이브 푸시 + assert [a["seq"] for a in notifier.audio] == [0, 1] + assert all(a["ext"] == "wav" for a in notifier.audio) + # 세그먼트 키 + 전체 파일 키가 모두 PUT 됨 + seg_keys = [k for (k, _d, _c) in storage.puts if "/seg-" in k] + assert seg_keys == [ + "interview/tts/7/42/seg-0.wav", + "interview/tts/7/42/seg-1.wav", + ] + full_key, full_data, full_ctype = storage.puts[-1] + assert full_key == "interview/tts/7/42.wav" + assert full_ctype == "audio/wav" + # 합본은 세그먼트 PCM 프레임 합(100+200) 을 담은 정상 WAV + with wave.open(io.BytesIO(full_data), "rb") as r: + assert r.getnframes() == 300 + # callback.tts SUCCEEDED + 전체 키 + rk, mtype, payload = publisher.published[0] + assert rk == "callback.tts" + assert payload.status == "SUCCEEDED" + assert payload.audio_key == "interview/tts/7/42.wav" + + +@pytest.mark.asyncio +async def test_tts_consumer_segment_seq_stays_contiguous_when_one_fails(): + # 중간 문장 합성이 실패해도 seq 는 0,1 로 연속 유지(프론트 큐 정지 방지). + storage = _FakeStorage() + publisher = _FakePublisher() + notifier = _RecordingNotifier() + + class _FlakyTts(TtsProvider): + model_name = "flaky" + + def __init__(self): + self.n = 0 + + async def synthesize(self, text, *, voice): + self.n += 1 + if self.n == 2: + raise RuntimeError("boom") + return TtsResult( + audio_bytes=_wav_bytes(50), duration_sec=0.3, content_type="audio/wav" + ) + + consumer = TtsConsumer( + tts=_FlakyTts(), + storage=storage, + publisher=publisher, + idempotency=_FakeIdem(), + callback_routing_key="callback.tts", + voice="alloy", + key_template="interview/tts/{session_id}/{message_id}.mp3", + session_notifier=notifier, + ) + + await consumer.handle(_FakeMsg(_envelope_body("하나. 둘. 셋."))) + + # 3문장 중 2번째 실패 → 성공 2개에 seq 0,1 연속 부여 + assert [a["seq"] for a in notifier.audio] == [0, 1] + assert publisher.published[0][2].status == "SUCCEEDED" diff --git a/frontend/src/features/interview/model/useLiveInterview.ts b/frontend/src/features/interview/model/useLiveInterview.ts index 7184e5e6..670a049a 100644 --- a/frontend/src/features/interview/model/useLiveInterview.ts +++ b/frontend/src/features/interview/model/useLiveInterview.ts @@ -14,11 +14,12 @@ import { pendingAnswers, toOptimisticMessage } from './optimistic' import type { OptimisticAnswer } from './optimistic' import { applyDelta, isStreamingMessage, FOLLOWUP_GENERATING_TEXT } from './streamingBuffer' import type { DeltaPayload } from './streamingBuffer' +import type { DeliveryMode } from './useDeliveryMode' export type ThreadItem = Message & { key: string; streaming?: boolean } export type ConnectionStatus = 'connecting' | 'open' | 'closed' -export function useLiveInterview(sessionId: number) { +export function useLiveInterview(sessionId: number, deliveryMode: DeliveryMode = 'text') { const navigate = useNavigate() const queryClient = useQueryClient() const sessionQuery = useSession(sessionId) @@ -29,6 +30,11 @@ export function useLiveInterview(sessionId: number) { const [optimistic, setOptimistic] = useState([]) const [connection, setConnection] = useState('connecting') const [deltaBuffer, setDeltaBuffer] = useState>({}) + // 라이브 세그먼트 오디오가 지금 재생 중인 메시지(아바타·질문 카드의 '말하는 중' 표시용). + const [speakingAudio, setSpeakingAudio] = useState<{ msgId: number | null; playing: boolean }>({ + msgId: null, + playing: false, + }) // ── 라이브 세그먼트 TTS ────────────────────────────────────────────────── const audioElRef = useRef(null) @@ -36,6 +42,9 @@ export function useLiveInterview(sessionId: number) { const queueRef = useRef | null>(null) const currentAudioMsgId = useRef(null) const segmentedIds = useRef>(new Set()) + // 소켓 onEvent 클로저에서 최신 모드를 읽기 위한 ref. 텍스트 모드는 음성 자동재생 안 함. + const deliveryModeRef = useRef(deliveryMode) + deliveryModeRef.current = deliveryMode if (queueRef.current === null) { queueRef.current = createSegmentQueue(async (url) => { if (typeof Audio === 'undefined') return @@ -68,15 +77,25 @@ export function useLiveInterview(sessionId: number) { ...pending.map((o) => ({ ...toOptimisticMessage(o), key: `opt-${o.tempId}` })), ] - // 세그먼트 오디오 엘리먼트 ended 리스너 — 마운트 1회. + // 세그먼트 오디오 엘리먼트 재생 리스너 — 마운트 1회. + // play/pause/ended 로 '말하는 중' 상태를 갱신해 아바타·질문 카드가 동기화되게 한다. useEffect(() => { if (typeof Audio === 'undefined') return if (!audioElRef.current) audioElRef.current = new Audio() const el = audioElRef.current - const onEnded = () => queueRef.current?.onEnded() + const onEnded = () => { + setSpeakingAudio((s) => ({ ...s, playing: false })) + queueRef.current?.onEnded() + } + const onPlay = () => setSpeakingAudio({ msgId: currentAudioMsgId.current, playing: true }) + const onPause = () => setSpeakingAudio((s) => ({ ...s, playing: false })) el.addEventListener('ended', onEnded) + el.addEventListener('play', onPlay) + el.addEventListener('pause', onPause) return () => { el.removeEventListener('ended', onEnded) + el.removeEventListener('play', onPlay) + el.removeEventListener('pause', onPause) if (lastUrlRef.current) URL.revokeObjectURL(lastUrlRef.current) } }, []) @@ -123,6 +142,9 @@ export function useLiveInterview(sessionId: number) { setDeltaBuffer((b) => applyDelta(b, payload)) } } else if (action.kind === 'queue-audio') { + // 텍스트 모드에서는 음성을 자동재생하지 않는다(읽는 도중 끼어들기 방지). + // 수동 재생은 StageQuestion 의 전체 파일(callback.tts) 경로로 제공된다. + if (deliveryModeRef.current !== 'voice') return const p = (frame.data as { data?: { messageId: number; seq: number; ext: string } } | undefined)?.data if (p && typeof p.messageId === 'number') { segmentedIds.current.add(p.messageId) @@ -167,6 +189,12 @@ export function useLiveInterview(sessionId: number) { const wasSegmented = useCallback((id: number) => segmentedIds.current.has(id), []) + // 해당 메시지의 라이브 세그먼트 오디오가 지금 재생 중인지. + const isSpeaking = useCallback( + (id: number) => speakingAudio.msgId === id && speakingAudio.playing, + [speakingAudio], + ) + // 첫 질문이 실제 content 를 갖고 도착했는지. 면접 스테이지 진입 전에 이걸 기다려 // 사용자가 스테이지에 들어서면 바로 질문을 볼 수 있게 한다(빈 대기 화면 회피). const firstQuestionReady = items.some((m) => { @@ -189,6 +217,7 @@ export function useLiveInterview(sessionId: number) { isLoading: sessionQuery.isLoading, questionStreaming, wasSegmented, + isSpeaking, firstQuestionReady, } } diff --git a/frontend/src/features/interview/ui/live/InterviewStage.tsx b/frontend/src/features/interview/ui/live/InterviewStage.tsx index 7bc4ed19..2b525f9a 100644 --- a/frontend/src/features/interview/ui/live/InterviewStage.tsx +++ b/frontend/src/features/interview/ui/live/InterviewStage.tsx @@ -4,7 +4,7 @@ import { Button } from '@/shared/ui/Button' import { isQuestion, isTranscribing, sessionProgress } from '@/domain/session' import type { Session } from '@/domain/session' import type { ConnectionStatus, ThreadItem } from '../../model/useLiveInterview' -import { useDeliveryMode } from '../../model/useDeliveryMode' +import type { DeliveryMode } from '../../model/useDeliveryMode' import { ConnectionBanner } from './ConnectionBanner' import { AnswerComposer } from './AnswerComposer' import { StageQuestion } from './StageQuestion' @@ -56,6 +56,9 @@ export function InterviewStage({ voiceUploading, onEnd, wasSegmented, + isSpeaking, + deliveryMode, + onDeliveryModeChange, }: { session: Session connection: ConnectionStatus @@ -67,13 +70,16 @@ export function InterviewStage({ voiceUploading: boolean onEnd: () => void wasSegmented: (id: number) => boolean + isSpeaking: (id: number) => boolean + deliveryMode: DeliveryMode + onDeliveryModeChange: (mode: DeliveryMode) => void }) { const [transcriptOpen, setTranscriptOpen] = useState(false) - const [deliveryMode, setDeliveryMode] = useDeliveryMode() const progress = sessionProgress(session) const currentQuestion = [...items].reverse().find(isQuestion) const lastItem = items[items.length - 1] const transcribing = Boolean(lastItem && isTranscribing(lastItem)) + const speaking = currentQuestion ? isSpeaking(currentQuestion.id ?? -1) : false return (
@@ -105,7 +111,7 @@ export function InterviewStage({

- + {connLabel[connection]}