From 939cc585c02a49320e50ccfdf436f505cdcc162f Mon Sep 17 00:00:00 2001 From: SeoHyeon Date: Thu, 4 Jun 2026 21:14:45 +0900 Subject: [PATCH 1/8] =?UTF-8?q?feat(ai):=20=EA=BC=AC=EB=A6=AC=EC=A7=88?= =?UTF-8?q?=EB=AC=B8=20=ED=86=A0=ED=81=B0=20=EC=8A=A4=ED=8A=B8=EB=A6=AC?= =?UTF-8?q?=EB=B0=8D=20(intent-first=20+=20astream=20+=20SESSION=5FMESSAGE?= =?UTF-8?q?=5FDELTA)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit followup 출력을 // 구분자 포맷으로 바꾸고, astream 으로 질문 토큰만 SessionRealtimeNotifier 를 통해 stackup.realtime 로 직접 발행(Core 우회). DONT_KNOW 면 델타 미발행. 종료 후 parse_followup_result 로 검증해 기존 callback.questions(FOLLOWUP, followupMessageId) 유지. runner 에 스트리밍 생성기+세션 notifier 와이어링. --- .../chain/followup_generation_chain.py | 156 +++++++++++++++- .../chain/prompts/followup_generation.py | 11 +- .../messaging/consumers/followup_consumer.py | 58 ++++-- ai/src/ai_server/messaging/runner.py | 11 ++ ai/src/ai_server/messaging/session_notify.py | 56 ++++++ ai/src/ai_server/model/messages/followup.py | 2 + ai/src/ai_server/model/messages/realtime.py | 16 ++ ai/tests/test_followup_consumer.py | 173 +++++++++++++----- ai/tests/test_followup_parse.py | 31 ++++ ai/tests/test_followup_streaming.py | 69 +++++++ ai/tests/test_session_notify.py | 34 ++++ 11 files changed, 552 insertions(+), 65 deletions(-) create mode 100644 ai/src/ai_server/messaging/session_notify.py create mode 100644 ai/tests/test_followup_parse.py create mode 100644 ai/tests/test_followup_streaming.py create mode 100644 ai/tests/test_session_notify.py diff --git a/ai/src/ai_server/chain/followup_generation_chain.py b/ai/src/ai_server/chain/followup_generation_chain.py index 54ffc8d5..ae191101 100644 --- a/ai/src/ai_server/chain/followup_generation_chain.py +++ b/ai/src/ai_server/chain/followup_generation_chain.py @@ -1,6 +1,8 @@ from __future__ import annotations -from typing import Literal, Protocol +import re +from collections.abc import Awaitable, Callable +from typing import Any, Literal, Protocol from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate @@ -16,11 +18,123 @@ class FollowupResult(BaseModel): followup_question: str = Field(..., description="한국어 꼬리질문 1개") - answer_evaluation: AnswerEvaluation + answer_evaluation: AnswerEvaluation | None = None # 답변 의도. NORMAL=정상답변, DONT_KNOW=모름/포기, CLARIFICATION=질문 재설명 요청. answer_intent: Literal["NORMAL", "DONT_KNOW", "CLARIFICATION"] = "NORMAL" +_INTENT_RE = re.compile(r"\s*(.*?)\s*", re.DOTALL) +_QUESTION_RE = re.compile(r"\s*(.*?)\s*", re.DOTALL) +_META_RE = re.compile(r"\s*(\{.*?\})\s*", re.DOTALL) +_VALID_INTENTS = {"NORMAL", "DONT_KNOW", "CLARIFICATION"} + + +def extract_question_span(text: str) -> str: + """완성 또는 진행 중 텍스트에서 안쪽만 추출(닫는 태그 없어도 가능).""" + closed = _QUESTION_RE.search(text) + if closed: + return closed.group(1).strip() + open_idx = text.find("") + if open_idx == -1: + return "" + tail = text[open_idx + len("") :] + tail = tail.split("", 1)[0] + # 닫는/메타 태그가 스트림 청크 경계로 잘려 들어와도(예: "질문?" not in tail[lt:]: + tail = tail[:lt] + return tail.strip() + + +def parse_followup_result(text: str) -> FollowupResult: + """종료된 누적 텍스트를 FollowupResult 로. 태그 누락 시 전체를 질문으로 폴백.""" + intent_m = _INTENT_RE.search(text) + intent = intent_m.group(1).strip().upper() if intent_m else "NORMAL" + if intent not in _VALID_INTENTS: + intent = "NORMAL" + + question = extract_question_span(text) + if not question: + question = text.strip() + + evaluation = None + meta_m = _META_RE.search(text) + if meta_m: + try: + evaluation = AnswerEvaluation.model_validate_json(meta_m.group(1)) + except Exception: + evaluation = None + + return FollowupResult( + followup_question=question, + answer_evaluation=evaluation, + answer_intent=intent, + ) + + +class StreamingFollowupGenerator: + """단일 LLM 콜을 astream 으로 흘리며 토큰만 콜백으로 내보낸다. + + intent 가 DONT_KNOW 면 질문 델타를 보내지 않는다(Core 가 폐기하므로). + 종료 후 누적 텍스트를 parse_followup_result 로 검증해 반환. + """ + + def __init__(self, *, prompt: ChatPromptTemplate, llm: Any) -> None: + self._prompt = prompt + self._llm = llm + + async def stream( + self, + *, + on_question_token: Callable[[str], Awaitable[None] | None], + job_category: str, + mode: str, + previous_question: str, + answer_text: str, + context: str, + parent_category: str, + expected_signal: str, + history: str, + ) -> FollowupResult: + all_vars = { + "job_category": job_category, + "mode": mode, + "previous_question": previous_question, + "answer_text": answer_text, + "context": context, + "parent_category": parent_category, + "expected_signal": expected_signal, + "history": history, + } + # 프롬프트가 선언한 변수만 전달(템플릿마다 변수 집합이 다를 수 있음). + wanted = set(self._prompt.input_variables) + prompt_vars = {k: v for k, v in all_vars.items() if k in wanted} + messages = self._prompt.format_messages(**prompt_vars) + + acc = "" + emitted_q = "" + async for chunk in self._llm.astream(messages): + piece = getattr(chunk, "content", "") or "" + if not piece: + continue + acc += piece + intent_known = _INTENT_RE.search(acc) + if not intent_known: + continue + if intent_known.group(1).strip().upper() == "DONT_KNOW": + continue + q_now = extract_question_span(acc) + if len(q_now) > len(emitted_q): + delta = q_now[len(emitted_q) :] + emitted_q = q_now + res = on_question_token(delta) + if res is not None and hasattr(res, "__await__"): + await res + + return parse_followup_result(acc) + + class FollowupGenerator(Protocol): async def generate( self, @@ -103,3 +217,41 @@ def build_followup_generation_chain( callbacks=callbacks, ) return prompt | llm | parser + + +def build_streaming_followup_generator( + settings: Settings, core_client: CoreClient | None = None +) -> StreamingFollowupGenerator: + """스트리밍 꼬리질문 생성기 빌더. + + format_instructions 없는 프롬프트 + Flash LLM 을 조합해 + StreamingFollowupGenerator 를 반환한다. + """ + from langchain_openai import ChatOpenAI + + prompt = ChatPromptTemplate.from_messages( + [ + ("system", SYSTEM_PROMPT), + ("human", HUMAN_PROMPT), + ] + ) + + callbacks = [] + if core_client is not None: + callbacks.append( + CoreAiLogCallback( + core_client=core_client, + request_type="generate.followup.stream", + default_model=settings.llm_flash_model, + ) + ) + + llm = ChatOpenAI( + model=settings.llm_flash_model, + temperature=settings.llm_flash_temperature, + api_key=settings.llm_api_key or None, + base_url=settings.llm_base_url, + max_tokens=settings.llm_flash_max_tokens, + callbacks=callbacks, + ) + return StreamingFollowupGenerator(prompt=prompt, llm=llm) diff --git a/ai/src/ai_server/chain/prompts/followup_generation.py b/ai/src/ai_server/chain/prompts/followup_generation.py index aeefdaf6..70921d21 100644 --- a/ai/src/ai_server/chain/prompts/followup_generation.py +++ b/ai/src/ai_server/chain/prompts/followup_generation.py @@ -26,7 +26,13 @@ "이때 followup_question 에는 새 꼬리질문이 아니라 **직전 질문을 더 쉽고 구체적으로 다시 설명한 문장**을 담으세요.\n" " - NORMAL: 그 외 정상 답변. followup_question 은 평소처럼 가장 약한 축을 파는 꼬리질문.\n" " - DONT_KNOW/CLARIFICATION 이면 채점(specificity/logic/correctness)은 보수적으로(낮게/null) 둡니다.\n" - "- 응답은 반드시 지정된 JSON 스키마를 따릅니다." + "- 출력은 정확히 다음 3개 블록을 이 순서로만 작성합니다(설명·코드펜스 금지):\n" + " NORMAL 또는 DONT_KNOW 또는 CLARIFICATION\n" + " 지원자에게 보여줄 한국어 질문 1개(또는 CLARIFICATION 시 재설명 문장)\n" + " {{\"specificity\": <0~5>, \"logic\": <0~5>, " + "\"structure\": \"FULL_STAR|PARTIAL_STAR|NONE\", \"correctness\": <0~5 또는 null>}}\n" + "- 를 가장 먼저 확정해 출력합니다. DONT_KNOW 여도 블록은 비우지 말고 한 문장 넣되, " + "표시는 시스템이 결정합니다." ) HUMAN_PROMPT = ( @@ -41,6 +47,5 @@ "직전 질문:\n{previous_question}\n\n" "직전 질문의 기대 신호(평가 기준):\n{expected_signal}\n\n" "지원자 답변:\n{answer_text}\n\n" - "검색 문서 컨텍스트:\n---\n{context}\n---\n\n" - "{format_instructions}" + "검색 문서 컨텍스트:\n---\n{context}\n---\n" ) diff --git a/ai/src/ai_server/messaging/consumers/followup_consumer.py b/ai/src/ai_server/messaging/consumers/followup_consumer.py index 280980aa..65734e5c 100644 --- a/ai/src/ai_server/messaging/consumers/followup_consumer.py +++ b/ai/src/ai_server/messaging/consumers/followup_consumer.py @@ -3,10 +3,14 @@ import structlog from aio_pika.abc import AbstractIncomingMessage -from ai_server.chain.followup_generation_chain import FollowupGenerator +from ai_server.chain.followup_generation_chain import ( + FollowupGenerator, + StreamingFollowupGenerator, +) from ai_server.core.client import CoreClient 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.followup import ( FollowupCallbackPayload, @@ -31,6 +35,8 @@ def __init__( rag_top_k: int = 5, reranker: Reranker | None = None, candidate_k: int = 20, + streaming_generator: StreamingFollowupGenerator | None = None, + session_notifier: SessionRealtimeNotifier | None = None, ) -> None: self._generator = generator self._publisher = publisher @@ -41,6 +47,8 @@ def __init__( self._rag_top_k = rag_top_k self._reranker = reranker or NoopReranker() self._candidate_k = max(candidate_k, rag_top_k) + self._streaming = streaming_generator + self._notifier = session_notifier async def handle(self, message: AbstractIncomingMessage) -> None: async with message.process(requeue=False): @@ -73,16 +81,43 @@ async def handle(self, message: AbstractIncomingMessage) -> None: trace_id=envelope.trace_id, ) - result = await self._generator.generate( - job_category=req.job_category, - mode=req.mode, - previous_question=req.previous_question, - answer_text=req.answer_text, - context=await self._build_rag_context(req), - parent_category=req.parent_category or "UNKNOWN", - expected_signal=req.parent_expected_signal or "(none)", - history=_format_history(req.history), - ) + rag_context = await self._build_rag_context(req) + + if self._streaming is not None and self._notifier is not None: + seq = {"n": 0} + + async def _on_token(delta: str) -> None: + await self._notifier.emit_delta( + session_id=req.session_id, + message_id=req.followup_message_id, + seq=seq["n"], + text=delta, + trace_id=envelope.trace_id, + ) + seq["n"] += 1 + + result = await self._streaming.stream( + on_question_token=_on_token, + job_category=req.job_category, + mode=req.mode, + previous_question=req.previous_question, + answer_text=req.answer_text, + context=rag_context, + parent_category=req.parent_category or "UNKNOWN", + expected_signal=req.parent_expected_signal or "(none)", + history=_format_history(req.history), + ) + else: + result = await self._generator.generate( + job_category=req.job_category, + mode=req.mode, + previous_question=req.previous_question, + answer_text=req.answer_text, + context=rag_context, + parent_category=req.parent_category or "UNKNOWN", + expected_signal=req.parent_expected_signal or "(none)", + history=_format_history(req.history), + ) payload = FollowupCallbackPayload( session_id=req.session_id, @@ -92,6 +127,7 @@ async def handle(self, message: AbstractIncomingMessage) -> None: followup_question=result.followup_question, answer_evaluation=result.answer_evaluation, answer_intent=result.answer_intent, + followup_message_id=req.followup_message_id, ) await self._publisher.publish( diff --git a/ai/src/ai_server/messaging/runner.py b/ai/src/ai_server/messaging/runner.py index 02448a61..79c5b71c 100644 --- a/ai/src/ai_server/messaging/runner.py +++ b/ai/src/ai_server/messaging/runner.py @@ -17,6 +17,7 @@ from ai_server.chain.followup_generation_chain import ( LlmFollowupGenerator, build_followup_generation_chain, + build_streaming_followup_generator, ) from ai_server.chain.feedback_generation_chain import ( LlmFeedbackGenerator, @@ -45,6 +46,7 @@ from ai_server.messaging.idempotency import LruIdempotencyStore from ai_server.messaging.progress import AnalysisProgressNotifier from ai_server.messaging.publisher import CallbackPublisher +from ai_server.messaging.session_notify import SessionRealtimeNotifier from ai_server.storage.factory import build_storage log = structlog.get_logger(__name__) @@ -188,6 +190,13 @@ def __init__(self, settings: Settings) -> None: followup_generator = LlmFollowupGenerator( build_followup_generation_chain(settings, core_client=core_client) ) + streaming_followup_generator = build_streaming_followup_generator( + settings, core_client=core_client + ) + session_notifier = SessionRealtimeNotifier( + publisher=self._realtime_publisher, + routing_key="realtime.session.notify", + ) self._followup_consumer = FollowupConsumer( generator=followup_generator, publisher=self._publisher, @@ -197,6 +206,8 @@ def __init__(self, settings: Settings) -> None: embedder=embedder, reranker=reranker, candidate_k=settings.rerank_candidate_k, + streaming_generator=streaming_followup_generator, + session_notifier=session_notifier, ) # 종합 피드백 생성 (US-24) diff --git a/ai/src/ai_server/messaging/session_notify.py b/ai/src/ai_server/messaging/session_notify.py new file mode 100644 index 00000000..b0b35f46 --- /dev/null +++ b/ai/src/ai_server/messaging/session_notify.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import structlog + +from ai_server.messaging.publisher import CallbackPublisher +from ai_server.model.envelope import MessageContext +from ai_server.model.messages.realtime import ( + SessionMessageDeltaData, + SessionNotifyPayload, +) + +log = structlog.get_logger(__name__) + +SESSION_MESSAGE_DELTA_EVENT = "SESSION_MESSAGE_DELTA" + + +class SessionRealtimeNotifier: + """AI -> RealTime 세션 채널 직접 발행. 꼬리질문 토큰 델타(휘발성). + + 발행 실패가 생성 파이프라인을 막아선 안 된다 — 예외를 삼키고 경고만. + """ + + def __init__(self, *, publisher: CallbackPublisher, routing_key: str) -> None: + self._publisher = publisher + self._routing_key = routing_key + + async def emit_delta( + self, + *, + session_id: int, + message_id: int, + seq: int, + text: str, + trace_id: str, + ) -> None: + payload = SessionNotifyPayload( + event_type=SESSION_MESSAGE_DELTA_EVENT, + data=SessionMessageDeltaData(message_id=message_id, seq=seq, text=text), + ) + try: + await self._publisher.publish( + routing_key=self._routing_key, + message_type=self._routing_key, + payload=payload, + trace_id=trace_id, + correlation_id=f"delta-{message_id}-{seq}", + context=MessageContext(session_id=session_id), + ) + except Exception: + log.warning( + "session.delta.publish_failed", + session_id=session_id, + message_id=message_id, + seq=seq, + trace_id=trace_id, + ) diff --git a/ai/src/ai_server/model/messages/followup.py b/ai/src/ai_server/model/messages/followup.py index 4ef93fdb..1463c890 100644 --- a/ai/src/ai_server/model/messages/followup.py +++ b/ai/src/ai_server/model/messages/followup.py @@ -24,6 +24,7 @@ class GenerateFollowupRequest(BaseModel): session_id: int parent_message_id: int # 직전 질문 메시지 ID answer_message_id: int # 답변 메시지 ID + followup_message_id: int # Core 가 선INSERT 한 placeholder 질문 메시지 id previous_question: str answer_text: str mode: InterviewMode @@ -58,3 +59,4 @@ class FollowupCallbackPayload(BaseModel): answer_evaluation: AnswerEvaluation | None = None # 답변 의도: NORMAL | DONT_KNOW | CLARIFICATION. Core 가 흐름 분기에 사용. answer_intent: str = "NORMAL" + followup_message_id: int # placeholder UPDATE 대상 diff --git a/ai/src/ai_server/model/messages/realtime.py b/ai/src/ai_server/model/messages/realtime.py index 2abf28ae..09d133ab 100644 --- a/ai/src/ai_server/model/messages/realtime.py +++ b/ai/src/ai_server/model/messages/realtime.py @@ -21,3 +21,19 @@ class RealtimeNotifyPayload(BaseModel): event_type: str data: AnalysisProgressData + + +# AI -> RealTime 세션 채널 직접 발행. 꼬리질문 토큰 델타(휘발성). Core 우회. +class SessionMessageDeltaData(BaseModel): + model_config = camel_config() + + message_id: int + seq: int + text: str # 이번 델타에서 추가된 조각(누적 아님) + + +class SessionNotifyPayload(BaseModel): + model_config = camel_config() + + event_type: str + data: SessionMessageDeltaData diff --git a/ai/tests/test_followup_consumer.py b/ai/tests/test_followup_consumer.py index a4f232db..11dfdb7a 100644 --- a/ai/tests/test_followup_consumer.py +++ b/ai/tests/test_followup_consumer.py @@ -44,6 +44,7 @@ def _envelope() -> bytes: "sessionId": 99, "parentMessageId": 501, "answerMessageId": 502, + "followupMessageId": 503, "previousQuestion": "결제 outbox 어떻게 구현?", "answerText": "RabbitMQ로 보냈습니다.", "mode": "TECHNICAL", @@ -55,17 +56,36 @@ def _envelope() -> bytes: return json.dumps(env).encode() +def _make_streaming_generator(result: FollowupResult): + """streaming_generator 페이크: stream() 호출 시 on_question_token 한 번 호출 후 result 반환.""" + streaming = MagicMock() + + async def _stream(*, on_question_token, **kwargs): + await on_question_token("토큰") + return result + + streaming.stream = _stream + return streaming + + +def _make_session_notifier(): + notifier = MagicMock() + notifier.emit_delta = AsyncMock() + return notifier + + @pytest.mark.asyncio async def test_consumer_generates_followup_and_publishes_callback(): - generator = MagicMock() - generator.generate = AsyncMock( - return_value=FollowupResult( - followup_question="구체적으로 outbox 테이블 스키마와 polling 주기는?", - answer_evaluation=AnswerEvaluation( - specificity=2.0, logic=3.0, structure="PARTIAL_STAR" - ), - ) + followup_result = FollowupResult( + followup_question="구체적으로 outbox 테이블 스키마와 polling 주기는?", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=3.0, structure="PARTIAL_STAR" + ), ) + generator = MagicMock() + generator.generate = AsyncMock(return_value=followup_result) + streaming_generator = _make_streaming_generator(followup_result) + session_notifier = _make_session_notifier() publisher = MagicMock() publisher.publish = AsyncMock() @@ -74,10 +94,14 @@ async def test_consumer_generates_followup_and_publishes_callback(): publisher=publisher, idempotency=LruIdempotencyStore(max_size=10), callback_routing_key="callback.questions", + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) await consumer.handle(_StubMessage(_envelope())) - generator.generate.assert_awaited_once() + # streaming path이 사용되므로 generator.generate 는 호출되지 않음 + generator.generate.assert_not_awaited() + session_notifier.emit_delta.assert_awaited_once() publisher.publish.assert_awaited_once() payload: FollowupCallbackPayload = publisher.publish.await_args.kwargs["payload"] assert payload.session_id == 99 @@ -85,20 +109,20 @@ async def test_consumer_generates_followup_and_publishes_callback(): assert payload.parent_message_id == 501 assert payload.followup_question.startswith("구체적으로") assert payload.answer_evaluation.structure == "PARTIAL_STAR" + assert payload.followup_message_id == 503 assert publisher.publish.await_args.kwargs["message_type"] == "callback.questions" @pytest.mark.asyncio async def test_consumer_injects_followup_rag_context_when_available(): - generator = MagicMock() - generator.generate = AsyncMock( - return_value=FollowupResult( - followup_question="outbox 저장과 발행의 원자성은 어떻게 보장했나요?", - answer_evaluation=AnswerEvaluation( - specificity=2.0, logic=3.0, structure="PARTIAL_STAR" - ), - ) + followup_result = FollowupResult( + followup_question="outbox 저장과 발행의 원자성은 어떻게 보장했나요?", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=3.0, structure="PARTIAL_STAR" + ), ) + generator = MagicMock() + generator.generate = AsyncMock(return_value=followup_result) publisher = MagicMock() publisher.publish = AsyncMock() core = MagicMock() @@ -115,6 +139,18 @@ async def test_consumer_injects_followup_rag_context_when_available(): embedder = MagicMock() embedder.embed = AsyncMock(return_value=[[0.1, 0.2, 0.3]]) + # Track kwargs passed to stream + received_kwargs: dict = {} + + async def _stream(*, on_question_token, **kwargs): + received_kwargs.update(kwargs) + await on_question_token("토큰") + return followup_result + + streaming_generator = MagicMock() + streaming_generator.stream = _stream + session_notifier = _make_session_notifier() + consumer = FollowupConsumer( generator=generator, publisher=publisher, @@ -123,6 +159,8 @@ async def test_consumer_injects_followup_rag_context_when_available(): core_client=core, embedder=embedder, rag_top_k=3, + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) await consumer.handle(_StubMessage(_envelope())) @@ -132,21 +170,21 @@ async def test_consumer_injects_followup_rag_context_when_available(): assert call.kwargs["document_ids"] == [7] assert call.kwargs["top_k"] == 20 # candidate_k (리랭크 후보 수) assert call.kwargs["query_text"] # 하이브리드 검색: 쿼리 텍스트 동봉 - context = generator.generate.await_args.kwargs["context"] - assert "Outbox rows are inserted in the same transaction" in context + assert ( + "Outbox rows are inserted in the same transaction" in received_kwargs["context"] + ) @pytest.mark.asyncio async def test_consumer_falls_back_when_followup_rag_fails(): - generator = MagicMock() - generator.generate = AsyncMock( - return_value=FollowupResult( - followup_question="실패 재처리는 어떻게 했나요?", - answer_evaluation=AnswerEvaluation( - specificity=2.0, logic=3.0, structure="PARTIAL_STAR" - ), - ) + followup_result = FollowupResult( + followup_question="실패 재처리는 어떻게 했나요?", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=3.0, structure="PARTIAL_STAR" + ), ) + generator = MagicMock() + generator.generate = AsyncMock(return_value=followup_result) publisher = MagicMock() publisher.publish = AsyncMock() core = MagicMock() @@ -154,6 +192,17 @@ async def test_consumer_falls_back_when_followup_rag_fails(): embedder = MagicMock() embedder.embed = AsyncMock(return_value=[[0.1, 0.2, 0.3]]) + received_kwargs: dict = {} + + async def _stream(*, on_question_token, **kwargs): + received_kwargs.update(kwargs) + await on_question_token("토큰") + return followup_result + + streaming_generator = MagicMock() + streaming_generator.stream = _stream + session_notifier = _make_session_notifier() + consumer = FollowupConsumer( generator=generator, publisher=publisher, @@ -161,10 +210,12 @@ async def test_consumer_falls_back_when_followup_rag_fails(): callback_routing_key="callback.questions", core_client=core, embedder=embedder, + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) await consumer.handle(_StubMessage(_envelope())) - assert generator.generate.await_args.kwargs["context"] == "(none)" + assert received_kwargs["context"] == "(none)" @pytest.mark.asyncio @@ -175,12 +226,18 @@ async def test_consumer_idempotent_skip(): publisher.publish = AsyncMock() idempotency = LruIdempotencyStore(max_size=10) idempotency.is_seen_then_mark("m-1") + streaming_generator = _make_streaming_generator( + FollowupResult(followup_question="Q") + ) + session_notifier = _make_session_notifier() consumer = FollowupConsumer( generator=generator, publisher=publisher, idempotency=idempotency, callback_routing_key="callback.questions", + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) await consumer.handle(_StubMessage(_envelope())) generator.generate.assert_not_awaited() @@ -212,22 +269,35 @@ def test_answer_evaluation_correctness_defaults_none_and_parses(): @pytest.mark.asyncio async def test_consumer_passes_parent_category_and_history_to_generator(): - generator = MagicMock() - generator.generate = AsyncMock( - return_value=FollowupResult( - followup_question="새 각도 질문", - answer_evaluation=AnswerEvaluation( - specificity=2.0, logic=2.0, structure="NONE" - ), - ) + followup_result = FollowupResult( + followup_question="새 각도 질문", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=2.0, structure="NONE" + ), ) + generator = MagicMock() + generator.generate = AsyncMock(return_value=followup_result) publisher = MagicMock() publisher.publish = AsyncMock() + + received_kwargs: dict = {} + + async def _stream(*, on_question_token, **kwargs): + received_kwargs.update(kwargs) + await on_question_token("토큰") + return followup_result + + streaming_generator = MagicMock() + streaming_generator.stream = _stream + session_notifier = _make_session_notifier() + consumer = FollowupConsumer( generator=generator, publisher=publisher, idempotency=LruIdempotencyStore(max_size=10), callback_routing_key="callback.questions", + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) env = { "messageId": "m-9", @@ -240,6 +310,7 @@ async def test_consumer_passes_parent_category_and_history_to_generator(): "sessionId": 99, "parentMessageId": 501, "answerMessageId": 502, + "followupMessageId": 503, "previousQuestion": "Q?", "answerText": "A.", "mode": "TECHNICAL", @@ -255,32 +326,36 @@ async def test_consumer_passes_parent_category_and_history_to_generator(): } await consumer.handle(_StubMessage(json.dumps(env).encode())) - kwargs = generator.generate.await_args.kwargs - assert kwargs["parent_category"] == "PROJECT_DEEP_DIVE" - assert kwargs["expected_signal"] == "동시성 제어를 DB 레벨까지 설명하는지" - assert "이전 질문" in kwargs["history"] - assert "면접관:" in kwargs["history"] + assert received_kwargs["parent_category"] == "PROJECT_DEEP_DIVE" + assert received_kwargs["expected_signal"] == "동시성 제어를 DB 레벨까지 설명하는지" + assert "이전 질문" in received_kwargs["history"] + assert "면접관:" in received_kwargs["history"] @pytest.mark.asyncio async def test_callback_includes_answer_message_id(): - generator = MagicMock() - generator.generate = AsyncMock( - return_value=FollowupResult( - followup_question="Q", - answer_evaluation=AnswerEvaluation( - specificity=2.0, logic=2.0, structure="NONE" - ), - ) + followup_result = FollowupResult( + followup_question="Q", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=2.0, structure="NONE" + ), ) + generator = MagicMock() + generator.generate = AsyncMock(return_value=followup_result) publisher = MagicMock() publisher.publish = AsyncMock() + streaming_generator = _make_streaming_generator(followup_result) + session_notifier = _make_session_notifier() + consumer = FollowupConsumer( generator=generator, publisher=publisher, idempotency=LruIdempotencyStore(max_size=10), callback_routing_key="callback.questions", + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) await consumer.handle(_StubMessage(_envelope())) payload = publisher.publish.await_args.kwargs["payload"] assert payload.answer_message_id == 502 # _envelope 의 answerMessageId + assert payload.followup_message_id == 503 # _envelope 의 followupMessageId diff --git a/ai/tests/test_followup_parse.py b/ai/tests/test_followup_parse.py new file mode 100644 index 00000000..e67b82e2 --- /dev/null +++ b/ai/tests/test_followup_parse.py @@ -0,0 +1,31 @@ +from ai_server.chain.followup_generation_chain import parse_followup_result + + +def test_parses_intent_question_meta(): + text = ( + "NORMAL" + "그 설계에서 트랜잭션 경계를 어디에 뒀나요?" + '{"specificity": 2, "logic": 3, "structure": "PARTIAL_STAR", "correctness": null}' + ) + r = parse_followup_result(text) + assert r.answer_intent == "NORMAL" + assert r.followup_question == "그 설계에서 트랜잭션 경계를 어디에 뒀나요?" + assert r.answer_evaluation.specificity == 2 + assert r.answer_evaluation.correctness is None + + +def test_fallback_when_no_tags(): + r = parse_followup_result("이건 그냥 질문 한 줄") + assert r.followup_question == "이건 그냥 질문 한 줄" + assert r.answer_intent == "NORMAL" + assert r.answer_evaluation is None + + +def test_extract_question_only_helper(): + from ai_server.chain.followup_generation_chain import extract_question_span + + assert extract_question_span("NORMAL안녕") == "안녕" + assert ( + extract_question_span("NORMAL안녕{}") + == "안녕" + ) diff --git a/ai/tests/test_followup_streaming.py b/ai/tests/test_followup_streaming.py new file mode 100644 index 00000000..ee5b977e --- /dev/null +++ b/ai/tests/test_followup_streaming.py @@ -0,0 +1,69 @@ +import pytest +from langchain_core.prompts import ChatPromptTemplate +from ai_server.chain.followup_generation_chain import StreamingFollowupGenerator + + +class _FakeStreamLLM: + """astream 시 토큰 조각을 순차로 내는 가짜 모델.""" + + def __init__(self, pieces): + self._pieces = pieces + + async def astream(self, _input, **_kw): + from langchain_core.messages import AIMessageChunk + + for p in self._pieces: + yield AIMessageChunk(content=p) + + +@pytest.mark.asyncio +async def test_stream_emits_question_tokens_and_final_result(): + pieces = [ + "NORMAL그 ", + "설계에서 ", + "트레이드오프는?", + "", + '{"specificity":2,"logic":3,"structure":"NONE","correctness":null}', + ] + prompt = ChatPromptTemplate.from_messages([("human", "{answer_text}")]) + gen = StreamingFollowupGenerator(prompt=prompt, llm=_FakeStreamLLM(pieces)) + + seen = [] + result = await gen.stream( + on_question_token=lambda piece: seen.append(piece), + job_category="BACKEND", + mode="TECHNICAL", + previous_question="q", + answer_text="a", + context="(none)", + parent_category="X", + expected_signal="(none)", + history="(none)", + ) + joined = "".join(seen) + assert "그 설계에서 트레이드오프는?" in joined + assert "" not in joined and "" not in joined # 질문만 흘렀다 + assert result.answer_intent == "NORMAL" + assert result.followup_question == "그 설계에서 트레이드오프는?" + + +@pytest.mark.asyncio +async def test_stream_dont_know_emits_no_question_tokens(): + pieces = ["DONT_KNOW모르면 이렇게{}"] + prompt = ChatPromptTemplate.from_messages([("human", "{answer_text}")]) + gen = StreamingFollowupGenerator(prompt=prompt, llm=_FakeStreamLLM(pieces)) + seen = [] + result = await gen.stream( + on_question_token=lambda piece: seen.append(piece), + job_category="BACKEND", + mode="TECHNICAL", + previous_question="q", + answer_text="a", + context="(none)", + parent_category="X", + expected_signal="(none)", + history="(none)", + ) + assert seen == [] # DONT_KNOW 는 질문 델타 0개 + assert result.answer_intent == "DONT_KNOW" diff --git a/ai/tests/test_session_notify.py b/ai/tests/test_session_notify.py new file mode 100644 index 00000000..363a2e02 --- /dev/null +++ b/ai/tests/test_session_notify.py @@ -0,0 +1,34 @@ +import pytest +from ai_server.messaging.session_notify import SessionRealtimeNotifier + + +class _FakePublisher: + def __init__(self): + self.calls = [] + + async def publish(self, **kwargs): + self.calls.append(kwargs) + + +@pytest.mark.asyncio +async def test_emit_delta_publishes_session_event(): + pub = _FakePublisher() + n = SessionRealtimeNotifier(publisher=pub, routing_key="realtime.session.notify") + await n.emit_delta(session_id=7, message_id=503, seq=0, text="안녕", trace_id="t1") + assert len(pub.calls) == 1 + call = pub.calls[0] + assert call["routing_key"] == "realtime.session.notify" + assert call["message_type"] == "realtime.session.notify" + assert call["context"].session_id == 7 + assert call["payload"].event_type == "SESSION_MESSAGE_DELTA" + assert call["payload"].data.text == "안녕" + + +@pytest.mark.asyncio +async def test_emit_delta_swallows_publish_errors(): + class Boom: + async def publish(self, **kwargs): + raise RuntimeError("down") + + n = SessionRealtimeNotifier(publisher=Boom(), routing_key="realtime.session.notify") + await n.emit_delta(session_id=1, message_id=2, seq=0, text="x", trace_id="t") From 7bc0033662dd8a02a1ad3b8df08d311fa9c0ba4a Mon Sep 17 00:00:00 2001 From: SeoHyeon Date: Thu, 4 Jun 2026 21:14:55 +0900 Subject: [PATCH 2/8] =?UTF-8?q?feat(core):=20=EA=BC=AC=EB=A6=AC=EC=A7=88?= =?UTF-8?q?=EB=AC=B8=20placeholder=20=EC=8A=A4=ED=8A=B8=EB=A6=AC=EB=B0=8D?= =?UTF-8?q?=20lifecycle?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 답변 직후 INTERVIEWER placeholder((생성 중), CREATED) 선INSERT + SESSION_MESSAGE 발행 + generate.followup 에 followupMessageId 동봉. 콜백 분기: NORMAL→completeFollowup UPDATE+카운트, CLARIFICATION→UPDATE(카운트 X), DONT_KNOW→placeholder DELETE 후 advanceToNextGeneral. placeholder 없으면 레거시 INSERT 폴백. --- .../application/QuestionsCallbackService.java | 55 +++++---- .../application/SessionFollowupRequester.java | 13 +++ .../dto/GenerateFollowupPayload.java | 1 + .../dto/QuestionsCallbackPayload.java | 3 +- .../session/domain/InterviewMessage.java | 19 ++++ .../QuestionsCallbackServiceTest.java | 104 +++++++++++++++++- .../session/domain/InterviewMessageTest.java | 60 ++++++++++ 7 files changed, 227 insertions(+), 28 deletions(-) create mode 100644 backend/src/test/java/com/stackup/stackup/session/domain/InterviewMessageTest.java diff --git a/backend/src/main/java/com/stackup/stackup/session/application/QuestionsCallbackService.java b/backend/src/main/java/com/stackup/stackup/session/application/QuestionsCallbackService.java index f0885a58..7525344c 100644 --- a/backend/src/main/java/com/stackup/stackup/session/application/QuestionsCallbackService.java +++ b/backend/src/main/java/com/stackup/stackup/session/application/QuestionsCallbackService.java @@ -163,9 +163,17 @@ private void applyFollowup(InterviewSession session, QuestionsCallbackPayload pa : messageRepository.findById(payload.parentMessageId()).orElse(null); String intent = payload.answerIntent() == null ? "NORMAL" : payload.answerIntent(); - // 모르겠음 → 이 주제 더 안 파고 다음 일반질문으로(없으면 종료). 답변평가는 기록(낮은 점수 의미 있음). + InterviewMessage placeholder = payload.followupMessageId() == null + ? null + : messageRepository.findById(payload.followupMessageId()).orElse(null); + + // 모르겠음 → 이 주제 그만, 다음 일반질문. placeholder 는 삭제(seq 연속성 유지). if ("DONT_KNOW".equalsIgnoreCase(intent)) { recordAnswerEvaluation(payload); + if (placeholder != null) { + messageRepository.delete(placeholder); + messageRepository.flush(); + } log.info("callback.questions DONT_KNOW — advancing to next general. sessionId={}", session.getId()); advanceToNextGeneral(session.getId()); return; @@ -176,31 +184,32 @@ private void applyFollowup(InterviewSession session, QuestionsCallbackPayload pa return; } - // 부연 요청 → 같은 질문 재설명. 질문 수(k)·깊이(m) 미반영, 답변평가 기록 안 함. - if ("CLARIFICATION".equalsIgnoreCase(intent)) { - int seq = (int) messageRepository.countBySession_Id(session.getId()) + 1; - InterviewMessage clar = messageRepository.save( - InterviewMessage.clarification(session, seq, payload.followupQuestion(), parent)); - publishQuestionEvents(session, clar, "CLARIFICATION"); - log.info("callback.questions CLARIFICATION — re-explained question. sessionId={}, msg={}", - session.getId(), clar.getId()); - return; + boolean clarification = "CLARIFICATION".equalsIgnoreCase(intent); + if (!clarification) { + recordAnswerEvaluation(payload); } - // 정상 답변 → 꼬리질문 삽입. - recordAnswerEvaluation(payload); - int nextSeq = (int) messageRepository.countBySession_Id(session.getId()) + 1; - InterviewMessage message = messageRepository.save( - InterviewMessage.followup(session, nextSeq, payload.followupQuestion(), parent) - ); - session.incrementQuestionCount(); - publishQuestionEvents(session, message, "FOLLOWUP_READY"); - - // maxQuestions(k) 도달 시 자동 종료. - maybeAutoEnd(session); + InterviewMessage message; + if (placeholder != null) { + placeholder.completeFollowup(payload.followupQuestion(), clarification); + message = messageRepository.save(placeholder); + } else { + // 레거시 폴백: placeholder 없이 도착한 콜백(롤아웃 호환). + int nextSeq = (int) messageRepository.countBySession_Id(session.getId()) + 1; + message = messageRepository.save(clarification + ? InterviewMessage.clarification(session, nextSeq, payload.followupQuestion(), parent) + : InterviewMessage.followup(session, nextSeq, payload.followupQuestion(), parent)); + } - log.info("callback.questions FOLLOWUP processed. sessionId={}, msg={}, totalQ={}", - session.getId(), message.getId(), session.getTotalQuestionCount()); + if (!clarification) { + session.incrementQuestionCount(); + } + publishQuestionEvents(session, message, clarification ? "CLARIFICATION" : "FOLLOWUP_READY"); + if (!clarification) { + maybeAutoEnd(session); + } + log.info("callback.questions FOLLOWUP processed. sessionId={}, msg={}, clarification={}", + session.getId(), message.getId(), clarification); } public record SessionStateNotice(Long sessionId, String status, String reason) { diff --git a/backend/src/main/java/com/stackup/stackup/session/application/SessionFollowupRequester.java b/backend/src/main/java/com/stackup/stackup/session/application/SessionFollowupRequester.java index cc567232..577beb43 100644 --- a/backend/src/main/java/com/stackup/stackup/session/application/SessionFollowupRequester.java +++ b/backend/src/main/java/com/stackup/stackup/session/application/SessionFollowupRequester.java @@ -3,6 +3,8 @@ import com.stackup.stackup.common.config.properties.RabbitMqProperties; import com.stackup.stackup.common.messaging.MessageContext; import com.stackup.stackup.common.messaging.RabbitMessagePublisher; +import com.stackup.stackup.common.messaging.RealtimeNotifyEvent; +import com.stackup.stackup.common.sse.SseEventType; import com.stackup.stackup.session.application.dto.GenerateFollowupPayload; import com.stackup.stackup.session.application.dto.GenerateFollowupPayload.HistoryItem; import com.stackup.stackup.session.application.event.AnswerSubmittedEvent; @@ -35,6 +37,7 @@ public class SessionFollowupRequester { private final InterviewMessageRepository messageRepository; private final SessionContextRepository contextRepository; private final QuestionsCallbackService questionsCallbackService; + private final org.springframework.context.ApplicationEventPublisher events; @Transactional(propagation = Propagation.REQUIRES_NEW) @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) @@ -74,6 +77,15 @@ public void onAnswerSubmitted(AnswerSubmittedEvent event) { .map(m -> new HistoryItem(m.getRole().name(), m.getContent())) .toList(); + // 스트리밍 표시용 placeholder 선INSERT (seq 예약). 콜백에서 content 채움. + int placeholderSeq = (int) messageRepository.countBySession_Id(session.getId()) + 1; + InterviewMessage placeholder = messageRepository.save( + InterviewMessage.followupPlaceholder(session, placeholderSeq, parent)); + // placeholder 를 프론트 메시지 목록에 즉시 노출 → 토큰 델타가 채울 버블 확보. + // REQUIRES_NEW 트랜잭션 commit 후 RealtimeNotifyEventListener 가 발행하므로 조회 보장. + events.publishEvent(RealtimeNotifyEvent.session( + session.getId(), SseEventType.SESSION_MESSAGE, placeholder.getId())); + GenerateFollowupPayload payload = new GenerateFollowupPayload( session.getId(), parent.getId(), @@ -85,6 +97,7 @@ public void onAnswerSubmitted(AnswerSubmittedEvent event) { contextDocumentIds, parent.getCategory(), parent.getExpectedSignal(), + placeholder.getId(), history ); publisher.publishToAi( diff --git a/backend/src/main/java/com/stackup/stackup/session/application/dto/GenerateFollowupPayload.java b/backend/src/main/java/com/stackup/stackup/session/application/dto/GenerateFollowupPayload.java index 0bc78376..72f67825 100644 --- a/backend/src/main/java/com/stackup/stackup/session/application/dto/GenerateFollowupPayload.java +++ b/backend/src/main/java/com/stackup/stackup/session/application/dto/GenerateFollowupPayload.java @@ -15,6 +15,7 @@ public record GenerateFollowupPayload( List contextDocumentIds, // RAG(자료 근거/correctness) 용 String parentCategory, // 직전 질문 카테고리 (루브릭 선택) String parentExpectedSignal, // 직전 질문이 기대하는 핵심(평가 관점) — 충족도 채점용 + Long followupMessageId, // 스트리밍 표시용 placeholder 메시지 ID (AI 콜백에서 채움) List history // 최근 대화 (중복 회피) ) { public record HistoryItem(String role, String content) { diff --git a/backend/src/main/java/com/stackup/stackup/session/application/dto/QuestionsCallbackPayload.java b/backend/src/main/java/com/stackup/stackup/session/application/dto/QuestionsCallbackPayload.java index 28e57fab..626c66a4 100644 --- a/backend/src/main/java/com/stackup/stackup/session/application/dto/QuestionsCallbackPayload.java +++ b/backend/src/main/java/com/stackup/stackup/session/application/dto/QuestionsCallbackPayload.java @@ -12,7 +12,8 @@ public record QuestionsCallbackPayload( Long answerMessageId, // FOLLOWUP 시 평가 대상 답변 메시지 String followupQuestion, // FOLLOWUP 시 사용 AnswerEvaluation answerEvaluation, // FOLLOWUP 시 옵션 - String answerIntent // NORMAL | DONT_KNOW | CLARIFICATION (FOLLOWUP) + String answerIntent, // NORMAL | DONT_KNOW | CLARIFICATION (FOLLOWUP) + Long followupMessageId // 스트리밍 placeholder 메시지 ID (AI→Core 콜백, null 가능) ) { public boolean isPool() { return "POOL".equals(kind); diff --git a/backend/src/main/java/com/stackup/stackup/session/domain/InterviewMessage.java b/backend/src/main/java/com/stackup/stackup/session/domain/InterviewMessage.java index 37db87b5..4831454a 100644 --- a/backend/src/main/java/com/stackup/stackup/session/domain/InterviewMessage.java +++ b/backend/src/main/java/com/stackup/stackup/session/domain/InterviewMessage.java @@ -36,6 +36,7 @@ public class InterviewMessage extends BaseTimeEntity { public static final String VOICE_TRANSCRIPTION_PENDING_TEXT = "(transcribing)"; public static final String VOICE_TRANSCRIPTION_FAILED_TEXT = "음성 인식에 실패했습니다. 텍스트로 다시 답변해 주세요."; + public static final String FOLLOWUP_GENERATING_TEXT = "(생성 중)"; @Id @GeneratedValue(strategy = GenerationType.IDENTITY) @@ -151,6 +152,14 @@ public static InterviewMessage followup(InterviewSession session, int seq, Strin return m; } + // 스트리밍 꼬리질문 placeholder. content 는 콜백에서 채운다. CHECK 제약(blank 금지) 위해 sentinel. + // TTS 는 생성 시점이 아니라 completeFollowup 에서 요청한다. + public static InterviewMessage followupPlaceholder(InterviewSession session, int seq, + InterviewMessage parent) { + return new InterviewMessage(session, seq, MessageRole.INTERVIEWER, + FOLLOWUP_GENERATING_TEXT, parent, MessageStatus.CREATED, null); + } + // 부연: 직전 질문을 다시 설명해 같은 차례를 유지. 질문 수/깊이에 카운트 안 함. public static InterviewMessage clarification(InterviewSession session, int seq, String content, InterviewMessage parent) { @@ -211,6 +220,16 @@ public void failTts() { this.ttsStatus = TtsStatus.FAILED; } + // 스트리밍 종료 후 placeholder 를 실제 질문으로 확정. clarification=true 면 질문 수/깊이 미반영. + public void completeFollowup(String content, boolean clarification) { + if (content != null && !content.isBlank()) { + this.content = content; + } + this.clarification = clarification; + markTtsPending(); + this.status = MessageStatus.COMPLETED; + } + public void completeWithTranscript(String transcript) { if (transcript != null && !transcript.isBlank()) { this.content = transcript; diff --git a/backend/src/test/java/com/stackup/stackup/session/application/QuestionsCallbackServiceTest.java b/backend/src/test/java/com/stackup/stackup/session/application/QuestionsCallbackServiceTest.java index c6572204..a32695f6 100644 --- a/backend/src/test/java/com/stackup/stackup/session/application/QuestionsCallbackServiceTest.java +++ b/backend/src/test/java/com/stackup/stackup/session/application/QuestionsCallbackServiceTest.java @@ -139,7 +139,7 @@ void apply_followupPersistsAnswerEvaluationOntoAnswerMessage() { QuestionsCallbackPayload payload = new QuestionsCallbackPayload( 11L, "FOLLOWUP", null, 500L, 600L, "다음 질문?", - new QuestionsCallbackPayload.AnswerEvaluation(2.0, 3.0, "PARTIAL_STAR", 1.5), "NORMAL" + new QuestionsCallbackPayload.AnswerEvaluation(2.0, 3.0, "PARTIAL_STAR", 1.5), "NORMAL", null ); QuestionsCallbackEnvelope env = new QuestionsCallbackEnvelope( "m-eval", "callback.questions", "1", "t", null, "ai", payload, null); @@ -165,7 +165,7 @@ void apply_clarificationReExplainsWithoutCountingQuestion() { InterviewSession session = sessionFixture(11L, SessionStatus.IN_PROGRESS); QuestionsCallbackPayload payload = new QuestionsCallbackPayload( 11L, "FOLLOWUP", null, 500L, 600L, "쉽게 다시 설명: 트랜잭션이란…", - null, "CLARIFICATION" + null, "CLARIFICATION", null ); QuestionsCallbackEnvelope env = new QuestionsCallbackEnvelope( "m-clar", "callback.questions", "1", "t", null, "ai", payload, null); @@ -185,16 +185,112 @@ void apply_clarificationReExplainsWithoutCountingQuestion() { assertThat(session.getTotalQuestionCount()).isEqualTo(0); } + // ── 새 placeholder 경로 테스트 ───────────────────────────────────────────── + + @Test + void apply_followupNormal_updatesPlaceholderInPlaceAndCounts() { + InterviewSession session = sessionFixture(20L, SessionStatus.IN_PROGRESS); + // placeholder: followupPlaceholder 의 sentinel content + InterviewMessage placeholder = InterviewMessage.followupPlaceholder( + session, 3, parentMessageFixture(session)); + ReflectionTestUtils.setField(placeholder, "id", 301L); + + QuestionsCallbackPayload payload = new QuestionsCallbackPayload( + 20L, "FOLLOWUP", null, 200L, null, "꼬리질문 내용?", + null, "NORMAL", 301L + ); + QuestionsCallbackEnvelope env = new QuestionsCallbackEnvelope( + "m-ph-normal", "callback.questions", "1", "t", null, "ai", payload, null); + + when(processedMessageRepository.existsById("m-ph-normal")).thenReturn(false); + when(sessionRepository.findById(20L)).thenReturn(Optional.of(session)); + when(messageRepository.findById(200L)).thenReturn(Optional.of(parentMessageFixture(session))); + when(messageRepository.findById(301L)).thenReturn(Optional.of(placeholder)); + when(messageRepository.save(any(InterviewMessage.class))).thenAnswer(inv -> inv.getArgument(0)); + + service.apply(env); + + // placeholder 가 in-place 로 업데이트되어야 함 + assertThat(placeholder.getContent()).isEqualTo("꼬리질문 내용?"); + assertThat(placeholder.getStatus()).isEqualTo(com.stackup.stackup.session.domain.MessageStatus.COMPLETED); + // 질문 수 카운트 증가 + assertThat(session.getTotalQuestionCount()).isEqualTo(1); + // save 는 placeholder 자체에 대해 호출됨 (새 InterviewMessage INSERT 아님) + ArgumentCaptor cap = ArgumentCaptor.forClass(InterviewMessage.class); + verify(messageRepository).save(cap.capture()); + assertThat(cap.getValue()).isSameAs(placeholder); + } + + @Test + void apply_followupClarification_updatesPlaceholderWithoutCounting() { + InterviewSession session = sessionFixture(21L, SessionStatus.IN_PROGRESS); + InterviewMessage placeholder = InterviewMessage.followupPlaceholder( + session, 3, parentMessageFixture(session)); + ReflectionTestUtils.setField(placeholder, "id", 302L); + + QuestionsCallbackPayload payload = new QuestionsCallbackPayload( + 21L, "FOLLOWUP", null, 200L, null, "다시 설명드리면…", + null, "CLARIFICATION", 302L + ); + QuestionsCallbackEnvelope env = new QuestionsCallbackEnvelope( + "m-ph-clar", "callback.questions", "1", "t", null, "ai", payload, null); + + when(processedMessageRepository.existsById("m-ph-clar")).thenReturn(false); + when(sessionRepository.findById(21L)).thenReturn(Optional.of(session)); + when(messageRepository.findById(200L)).thenReturn(Optional.of(parentMessageFixture(session))); + when(messageRepository.findById(302L)).thenReturn(Optional.of(placeholder)); + when(messageRepository.save(any(InterviewMessage.class))).thenAnswer(inv -> inv.getArgument(0)); + + service.apply(env); + + // clarification 플래그가 세팅되어야 함 + assertThat(placeholder.isClarification()).isTrue(); + assertThat(placeholder.getContent()).isEqualTo("다시 설명드리면…"); + // 질문 수 카운트 미증가 + assertThat(session.getTotalQuestionCount()).isEqualTo(0); + } + + @Test + void apply_followupDontKnow_deletesPlaceholderAndAdvances() { + InterviewSession session = sessionFixture(22L, SessionStatus.IN_PROGRESS); + InterviewMessage placeholder = InterviewMessage.followupPlaceholder( + session, 3, parentMessageFixture(session)); + ReflectionTestUtils.setField(placeholder, "id", 303L); + + QuestionsCallbackPayload payload = new QuestionsCallbackPayload( + 22L, "FOLLOWUP", null, 200L, null, null, + null, "DONT_KNOW", 303L + ); + QuestionsCallbackEnvelope env = new QuestionsCallbackEnvelope( + "m-ph-dk", "callback.questions", "1", "t", null, "ai", payload, null); + + when(processedMessageRepository.existsById("m-ph-dk")).thenReturn(false); + when(sessionRepository.findById(22L)).thenReturn(Optional.of(session)); + when(messageRepository.findById(200L)).thenReturn(Optional.of(parentMessageFixture(session))); + when(messageRepository.findById(303L)).thenReturn(Optional.of(placeholder)); + // advanceToNextGeneral 내부에서 sessionRepository.findById 재호출 + when(poolRepository.findFirstBySessionIdAndUsedFalseOrderByIdxAsc(22L)) + .thenReturn(java.util.Optional.empty()); + + service.apply(env); + + // placeholder 가 삭제되어야 함 + verify(messageRepository).delete(placeholder); + verify(messageRepository).flush(); + // 세션이 종료됨 (풀이 비어있어 POOL_EXHAUSTED) + assertThat(session.getStatus()).isEqualTo(SessionStatus.COMPLETED); + } + private QuestionsCallbackEnvelope poolEnvelope(Long sessionId, List questions) { QuestionsCallbackPayload payload = new QuestionsCallbackPayload( - sessionId, "POOL", questions, null, null, null, null, null + sessionId, "POOL", questions, null, null, null, null, null, null ); return new QuestionsCallbackEnvelope("m-1", "callback.questions", "1", "t", null, "ai", payload, null); } private QuestionsCallbackEnvelope followupEnvelope(Long sessionId, Long parentId, String followup) { QuestionsCallbackPayload payload = new QuestionsCallbackPayload( - sessionId, "FOLLOWUP", null, parentId, null, followup, null, "NORMAL" + sessionId, "FOLLOWUP", null, parentId, null, followup, null, "NORMAL", null ); return new QuestionsCallbackEnvelope("m-2", "callback.questions", "1", "t", null, "ai", payload, null); } diff --git a/backend/src/test/java/com/stackup/stackup/session/domain/InterviewMessageTest.java b/backend/src/test/java/com/stackup/stackup/session/domain/InterviewMessageTest.java new file mode 100644 index 00000000..e9cb2ccc --- /dev/null +++ b/backend/src/test/java/com/stackup/stackup/session/domain/InterviewMessageTest.java @@ -0,0 +1,60 @@ +package com.stackup.stackup.session.domain; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.stackup.stackup.user.domain.User; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; + +class InterviewMessageTest { + + private InterviewSession session; + + @BeforeEach + void setUp() { + User user = User.createGithubUser(1L, "u", null, null, "t"); + ReflectionTestUtils.setField(user, "id", 1L); + session = InterviewSession.create( + user, "테스트 세션", null, SessionMode.TECHNICAL, JobCategory.BACKEND, 5, 30, null, null + ); + ReflectionTestUtils.setField(session, "id", 10L); + } + + @Test + void followupPlaceholder_hasSentinelContent_createdStatus_interviewerRole_ttsNotRequested() { + InterviewMessage parent = InterviewMessage.interviewer(session, 1, "원질문"); + + InterviewMessage ph = InterviewMessage.followupPlaceholder(session, 2, parent); + + assertThat(ph.getContent()).isEqualTo(InterviewMessage.FOLLOWUP_GENERATING_TEXT); + assertThat(ph.getStatus()).isEqualTo(MessageStatus.CREATED); + assertThat(ph.getTtsStatus()).isEqualTo(TtsStatus.NOT_REQUESTED); + assertThat(ph.getRole()).isEqualTo(MessageRole.INTERVIEWER); + } + + @Test + void completeFollowup_fillsContentMarksCompletedAndTtsPending() { + InterviewMessage parent = InterviewMessage.interviewer(session, 1, "원질문"); + InterviewMessage ph = InterviewMessage.followupPlaceholder(session, 2, parent); + + ph.completeFollowup("진짜 꼬리질문?", false); + + assertThat(ph.getContent()).isEqualTo("진짜 꼬리질문?"); + assertThat(ph.isClarification()).isFalse(); + assertThat(ph.getStatus()).isEqualTo(MessageStatus.COMPLETED); + assertThat(ph.getTtsStatus()).isEqualTo(TtsStatus.PENDING); + } + + @Test + void completeFollowup_withClarificationTrue_setsClarificationFlag() { + InterviewMessage parent = InterviewMessage.interviewer(session, 1, "원질문"); + InterviewMessage ph2 = InterviewMessage.followupPlaceholder(session, 3, parent); + + ph2.completeFollowup("다시 설명하면…", true); + + assertThat(ph2.isClarification()).isTrue(); + assertThat(ph2.getStatus()).isEqualTo(MessageStatus.COMPLETED); + assertThat(ph2.getTtsStatus()).isEqualTo(TtsStatus.PENDING); + } +} From d03726cce3537c227e61c9c0e83ec408f76456fb Mon Sep 17 00:00:00 2001 From: SeoHyeon Date: Thu, 4 Jun 2026 21:15:06 +0900 Subject: [PATCH 3/8] =?UTF-8?q?feat(fe):=20=EA=BC=AC=EB=A6=AC=EC=A7=88?= =?UTF-8?q?=EB=AC=B8=20=ED=86=A0=ED=81=B0=20=EC=8A=A4=ED=8A=B8=EB=A6=AC?= =?UTF-8?q?=EB=B0=8D=20=EB=A0=8C=EB=8D=94=20+=20=EC=A0=9C=EC=B6=9C=20?= =?UTF-8?q?=EA=B2=8C=EC=9D=B4=ED=8C=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SESSION_MESSAGE_DELTA 를 messageId별 버퍼에 누적해 placeholder 버블 content 를 실시간 덮어 렌더. sentinel((생성 중)) 인지 currentTurn 으로 '생각 중→텍스트 차오름' 전환. questionStreaming 으로 입력은 허용·제출만 잠금(AnswerComposer submitLocked). sentinel 상수는 domain/session 단일 출처. --- frontend/src/domain/session/index.ts | 1 + frontend/src/domain/session/lib/turn.test.ts | 5 ++- frontend/src/domain/session/lib/turn.ts | 14 +++++-- .../interview/model/interviewEvent.test.ts | 3 ++ .../interview/model/interviewEvent.ts | 3 ++ .../interview/model/streamingBuffer.test.ts | 16 ++++++++ .../interview/model/streamingBuffer.ts | 21 ++++++++++ .../interview/model/useLiveInterview.ts | 41 ++++++++++++++++++- .../interview/ui/live/AnswerComposer.tsx | 14 +++++-- .../interview/ui/live/InterviewStage.tsx | 3 ++ .../interview/ui/live/LiveInterview.tsx | 2 + .../interview/ui/live/QuestionBubble.tsx | 11 ++++- 12 files changed, 124 insertions(+), 10 deletions(-) create mode 100644 frontend/src/features/interview/model/streamingBuffer.test.ts create mode 100644 frontend/src/features/interview/model/streamingBuffer.ts diff --git a/frontend/src/domain/session/index.ts b/frontend/src/domain/session/index.ts index 8b359c08..a2aa7441 100644 --- a/frontend/src/domain/session/index.ts +++ b/frontend/src/domain/session/index.ts @@ -12,6 +12,7 @@ export { isAnswer, isTranscribing, VOICE_TRANSCRIBING_TEXT, + FOLLOWUP_GENERATING_TEXT, currentTurn, canSubmitAnswer, sessionProgress, diff --git a/frontend/src/domain/session/lib/turn.test.ts b/frontend/src/domain/session/lib/turn.test.ts index 95b86db2..e6e00c3d 100644 --- a/frontend/src/domain/session/lib/turn.test.ts +++ b/frontend/src/domain/session/lib/turn.test.ts @@ -1,6 +1,6 @@ import { describe, it, expect } from 'vitest' import type { Message, Session } from '../model/types' -import { currentTurn, canSubmitAnswer, sessionProgress, isQuestion, isAnswer } from './turn' +import { currentTurn, canSubmitAnswer, sessionProgress, isQuestion, isAnswer, FOLLOWUP_GENERATING_TEXT } from './turn' const msg = (role: 'INTERVIEWER' | 'INTERVIEWEE', content = '내용'): Message => ({ role, content } as Message) @@ -19,6 +19,9 @@ describe('currentTurn', () => { it('내용 없는 질문(생성 중)이면 질문 대기', () => { expect(currentTurn([msg('INTERVIEWER', '')])).toBe('WAITING_FOR_QUESTION') }) + it('생성 중 sentinel 질문이면 질문 대기', () => { + expect(currentTurn([msg('INTERVIEWER', FOLLOWUP_GENERATING_TEXT)])).toBe('WAITING_FOR_QUESTION') + }) }) describe('canSubmitAnswer', () => { diff --git a/frontend/src/domain/session/lib/turn.ts b/frontend/src/domain/session/lib/turn.ts index f3f455b0..d958b043 100644 --- a/frontend/src/domain/session/lib/turn.ts +++ b/frontend/src/domain/session/lib/turn.ts @@ -14,6 +14,10 @@ export function isAnswer(message: Message): boolean { // STT 완료 전까지 이 값이 들어 있고, 완료되면 실제 transcript 로 교체된다. export const VOICE_TRANSCRIBING_TEXT = '(transcribing)' +// 꼬리질문 생성 중인 placeholder content (백엔드 sentinel 과 동일). +// 스트리밍이 완료되면 실제 질문 텍스트로 교체된다. +export const FOLLOWUP_GENERATING_TEXT = '(생성 중)' + // STT 대기 중(transcript 아직 없음)인 음성 답변인지. export function isTranscribing(message: Message): boolean { if (!isAnswer(message)) return false @@ -23,10 +27,14 @@ export function isTranscribing(message: Message): boolean { export function currentTurn(messages: Message[]): Turn { const last = messages[messages.length - 1] - // 마지막이 '내용이 있는' 면접관 질문일 때만 답변 차례. 질문 생성 중(빈 content) - // 메시지가 먼저 도착해도 컴포저가 섣불리 활성화되지 않도록 content 유무로 가드한다. + // 마지막이 '내용이 있는' 면접관 질문일 때만 답변 차례. 질문 생성 중(빈 content 또는 sentinel) + // 메시지가 먼저 도착해도 컴포저가 섣불리 활성화되지 않도록 content 유무 + sentinel 로 가드한다. // (질문 status 는 CREATED 로 들어오므로 status 기준으로 거르면 안 된다.) - const askable = last?.role === 'INTERVIEWER' && (last.content ?? '').trim().length > 0 + const content = (last?.content ?? '').trim() + const askable = + last?.role === 'INTERVIEWER' && + content.length > 0 && + content !== FOLLOWUP_GENERATING_TEXT return askable ? 'AWAITING_ANSWER' : 'WAITING_FOR_QUESTION' } diff --git a/frontend/src/features/interview/model/interviewEvent.test.ts b/frontend/src/features/interview/model/interviewEvent.test.ts index 4fbb83d4..d6561d4c 100644 --- a/frontend/src/features/interview/model/interviewEvent.test.ts +++ b/frontend/src/features/interview/model/interviewEvent.test.ts @@ -17,4 +17,7 @@ describe('interviewEventAction', () => { it('알 수 없는 이벤트는 무시', () => { expect(interviewEventAction('whatever')).toEqual({ kind: 'ignore' }) }) + it('SESSION_MESSAGE_DELTA → 델타 append', () => { + expect(interviewEventAction('SESSION_MESSAGE_DELTA')).toEqual({ kind: 'append-delta' }) + }) }) diff --git a/frontend/src/features/interview/model/interviewEvent.ts b/frontend/src/features/interview/model/interviewEvent.ts index 6f4ccd42..60936e54 100644 --- a/frontend/src/features/interview/model/interviewEvent.ts +++ b/frontend/src/features/interview/model/interviewEvent.ts @@ -2,6 +2,7 @@ export type InterviewAction = | { kind: 'refetch-messages' } | { kind: 'refetch-session' } | { kind: 'redirect-feedback' } + | { kind: 'append-delta' } | { kind: 'ignore' } export function interviewEventAction(event: string): InterviewAction { @@ -12,6 +13,8 @@ export function interviewEventAction(event: string): InterviewAction { return { kind: 'refetch-session' } case 'FEEDBACK_READY': return { kind: 'redirect-feedback' } + case 'SESSION_MESSAGE_DELTA': + return { kind: 'append-delta' } default: return { kind: 'ignore' } } diff --git a/frontend/src/features/interview/model/streamingBuffer.test.ts b/frontend/src/features/interview/model/streamingBuffer.test.ts new file mode 100644 index 00000000..38e92320 --- /dev/null +++ b/frontend/src/features/interview/model/streamingBuffer.test.ts @@ -0,0 +1,16 @@ +import { describe, it, expect } from 'vitest' +import { applyDelta, isStreamingMessage, FOLLOWUP_GENERATING_TEXT } from './streamingBuffer' + +describe('streamingBuffer', () => { + it('applyDelta 는 messageId별로 누적한다', () => { + let buf: Record = {} + buf = applyDelta(buf, { messageId: 5, seq: 0, text: '안녕' }) + buf = applyDelta(buf, { messageId: 5, seq: 1, text: '하세요' }) + expect(buf[5]).toBe('안녕하세요') + }) + it('isStreamingMessage 는 sentinel content + 버퍼 유무로 판별', () => { + expect(isStreamingMessage({ content: FOLLOWUP_GENERATING_TEXT }, undefined)).toBe(true) + expect(isStreamingMessage({ content: FOLLOWUP_GENERATING_TEXT }, '안녕')).toBe(true) + expect(isStreamingMessage({ content: '완성된 질문' }, undefined)).toBe(false) + }) +}) diff --git a/frontend/src/features/interview/model/streamingBuffer.ts b/frontend/src/features/interview/model/streamingBuffer.ts new file mode 100644 index 00000000..6bc99062 --- /dev/null +++ b/frontend/src/features/interview/model/streamingBuffer.ts @@ -0,0 +1,21 @@ +import { FOLLOWUP_GENERATING_TEXT } from '@/domain/session' +export { FOLLOWUP_GENERATING_TEXT } + +export type DeltaPayload = { messageId: number; seq: number; text: string } + +// messageId -> 누적 텍스트. seq 는 순서 보조(현재는 단순 append, gap 은 종료 reconcile 로 자기치유). +export function applyDelta( + buffer: Record, + delta: DeltaPayload, +): Record { + const prev = buffer[delta.messageId] ?? '' + return { ...buffer, [delta.messageId]: prev + delta.text } +} + +export function isStreamingMessage( + message: { content?: string | null }, + bufferedText: string | undefined, +): boolean { + if (bufferedText !== undefined) return true + return message.content === FOLLOWUP_GENERATING_TEXT +} diff --git a/frontend/src/features/interview/model/useLiveInterview.ts b/frontend/src/features/interview/model/useLiveInterview.ts index 101f5108..d06babc5 100644 --- a/frontend/src/features/interview/model/useLiveInterview.ts +++ b/frontend/src/features/interview/model/useLiveInterview.ts @@ -11,6 +11,8 @@ import { useInterviewSocket } from './useInterviewSocket' import { interviewEventAction } from './interviewEvent' import { pendingAnswers, toOptimisticMessage } from './optimistic' import type { OptimisticAnswer } from './optimistic' +import { applyDelta, isStreamingMessage, FOLLOWUP_GENERATING_TEXT } from './streamingBuffer' +import type { DeltaPayload } from './streamingBuffer' export type ThreadItem = Message & { key: string } export type ConnectionStatus = 'connecting' | 'open' | 'closed' @@ -25,14 +27,25 @@ export function useLiveInterview(sessionId: number) { const [optimistic, setOptimistic] = useState([]) const [connection, setConnection] = useState('connecting') + const [deltaBuffer, setDeltaBuffer] = useState>({}) // 서버가 sequenceNumber asc 로 주지만, 순서를 코드에서 명시적으로 보장한다. const serverMessages = [...(messagesQuery.data ?? [])].sort( (a, b) => (a.sequenceNumber ?? 0) - (b.sequenceNumber ?? 0), ) const pending = pendingAnswers(optimistic, serverMessages) + + // 스트리밍 중인 메시지는 deltaBuffer의 누적 텍스트로 content를 오버라이드한다. + const mergedMessages = serverMessages.map((m) => { + const buffered = deltaBuffer[m.id ?? -1] + if (buffered !== undefined && isStreamingMessage(m, buffered)) { + return { ...m, content: buffered } + } + return m + }) + const items: ThreadItem[] = [ - ...serverMessages.map((m) => ({ ...m, key: `m-${m.id}` })), + ...mergedMessages.map((m) => ({ ...m, key: `m-${m.id}` })), ...pending.map((o) => ({ ...toOptimisticMessage(o), key: `opt-${o.tempId}` })), ] @@ -44,6 +57,22 @@ export function useLiveInterview(sessionId: number) { }) }, [messagesQuery.data]) + // 서버에서 최종 content가 도착한 메시지의 버퍼를 정리한다. + useEffect(() => { + setDeltaBuffer((prev) => { + const next = { ...prev } + let changed = false + for (const m of serverMessages) { + if (m.id !== undefined && next[m.id] !== undefined && m.content !== FOLLOWUP_GENERATING_TEXT) { + delete next[m.id] + changed = true + } + } + return changed ? next : prev + }) + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [messagesQuery.data]) + const { submitAnswer: socketSubmit } = useInterviewSocket({ sessionId, enabled: status === 'IN_PROGRESS', @@ -56,6 +85,11 @@ export function useLiveInterview(sessionId: number) { void queryClient.invalidateQueries({ queryKey: sessionKeys.detail(sessionId) }) } else if (action.kind === 'redirect-feedback') { navigate(`/sessions/${sessionId}/feedback`) + } else if (action.kind === 'append-delta') { + const payload = (frame.data as { data?: DeltaPayload } | undefined)?.data + if (payload && typeof payload.messageId === 'number') { + setDeltaBuffer((b) => applyDelta(b, payload)) + } } }, }) @@ -83,6 +117,10 @@ export function useLiveInterview(sessionId: number) { [voiceMutation], ) + // 서버 메시지 기준으로 가장 최근 면접관 메시지가 여전히 sentinel이면 스트리밍 진행 중. + const latestServerQuestion = [...serverMessages].reverse().find((m) => m.role === 'INTERVIEWER') + const questionStreaming = latestServerQuestion?.content === FOLLOWUP_GENERATING_TEXT + return { session: sessionQuery.data, status, @@ -95,5 +133,6 @@ export function useLiveInterview(sessionId: number) { voiceError: voiceMutation.isError, endSession: () => end.mutate(), isLoading: sessionQuery.isLoading, + questionStreaming, } } diff --git a/frontend/src/features/interview/ui/live/AnswerComposer.tsx b/frontend/src/features/interview/ui/live/AnswerComposer.tsx index a93a3856..57294dec 100644 --- a/frontend/src/features/interview/ui/live/AnswerComposer.tsx +++ b/frontend/src/features/interview/ui/live/AnswerComposer.tsx @@ -25,11 +25,13 @@ function MicIcon() { export function AnswerComposer({ disabled = false, + submitLocked = false, onSubmit, onSubmitVoice, voiceUploading = false, }: { disabled?: boolean + submitLocked?: boolean onSubmit: (content: string) => void onSubmitVoice?: (audio: Blob) => void voiceUploading?: boolean @@ -41,7 +43,7 @@ export function AnswerComposer({ const submit = () => { const trimmed = value.trim() - if (!trimmed || disabled) return + if (!trimmed || disabled || submitLocked) return onSubmit(trimmed) setValue('') } @@ -95,21 +97,25 @@ export function AnswerComposer({ maxLength={8000} aria-label="답변 입력" placeholder={ - disabled ? '질문을 기다리는 중…' : '답변을 입력하세요 (Enter 전송, Shift+Enter 줄바꿈)' + disabled + ? '질문을 기다리는 중…' + : submitLocked + ? '질문이 끝나면 전송할 수 있어요' + : '답변을 입력하세요 (Enter 전송, Shift+Enter 줄바꿈)' } /> {voiceSupported && ( )} - diff --git a/frontend/src/features/interview/ui/live/InterviewStage.tsx b/frontend/src/features/interview/ui/live/InterviewStage.tsx index 01da3ddb..d13ae1b6 100644 --- a/frontend/src/features/interview/ui/live/InterviewStage.tsx +++ b/frontend/src/features/interview/ui/live/InterviewStage.tsx @@ -48,6 +48,7 @@ export function InterviewStage({ connection, items, awaitingQuestion, + questionStreaming, onSubmit, onSubmitVoice, voiceUploading, @@ -57,6 +58,7 @@ export function InterviewStage({ connection: ConnectionStatus items: ThreadItem[] awaitingQuestion: boolean + questionStreaming: boolean onSubmit: (content: string) => void onSubmitVoice: (audio: Blob) => void voiceUploading: boolean @@ -126,6 +128,7 @@ export function InterviewStage({
)}
-

{message.content}

+ {message.content === FOLLOWUP_GENERATING_TEXT ? ( + + ● + ● + ● + + ) : ( +

{message.content}

+ )} {ttsReady && (
diff --git a/frontend/src/features/interview/ui/live/LiveInterview.tsx b/frontend/src/features/interview/ui/live/LiveInterview.tsx index 56c1e0e5..db3cfdca 100644 --- a/frontend/src/features/interview/ui/live/LiveInterview.tsx +++ b/frontend/src/features/interview/ui/live/LiveInterview.tsx @@ -17,6 +17,7 @@ export function LiveInterview({ sessionId }: { sessionId: number }) { endSession, isLoading, questionStreaming, + wasSegmented, } = useLiveInterview(sessionId) if (isLoading || !session) { @@ -45,6 +46,7 @@ export function LiveInterview({ sessionId }: { sessionId: number }) { onSubmitVoice={submitVoice} voiceUploading={voiceUploading} onEnd={endSession} + wasSegmented={wasSegmented} /> ) } diff --git a/frontend/src/features/interview/ui/live/StageQuestion.tsx b/frontend/src/features/interview/ui/live/StageQuestion.tsx index d91f1406..c9ad8886 100644 --- a/frontend/src/features/interview/ui/live/StageQuestion.tsx +++ b/frontend/src/features/interview/ui/live/StageQuestion.tsx @@ -11,7 +11,7 @@ function PlayIcon({ playing }: { playing: boolean }) { } // 면접관이 지금 막 던진 한 질문에만 집중시키는 카드. -export function StageQuestion({ question }: { question: Message }) { +export function StageQuestion({ question, segmented = false }: { question: Message; segmented?: boolean }) { const label = categoryLabel(question.category) const ttsReady = question.ttsStatus === 'SUCCEEDED' @@ -19,7 +19,7 @@ export function StageQuestion({ question }: { question: Message }) { sessionId: question.sessionId, messageId: question.id, enabled: ttsReady, - autoPlay: true, + autoPlay: !segmented, }) return ( From b1c4c24e515a737edaaf4168ac9948b14a265e34 Mon Sep 17 00:00:00 2001 From: SeoHyeon Date: Thu, 4 Jun 2026 22:50:43 +0900 Subject: [PATCH 8/8] =?UTF-8?q?docs:=20=EB=AC=B8=EC=9E=A5=20=EB=8B=A8?= =?UTF-8?q?=EC=9C=84=20TTS=20=EC=9D=B4=EB=B2=A4=ED=8A=B8/=EB=A9=94?= =?UTF-8?q?=EC=8B=9C=EC=A7=95/=EC=8A=A4=ED=86=A0=EB=A6=AC=EC=A7=80/?= =?UTF-8?q?=EC=BB=A8=ED=85=8D=EC=8A=A4=ED=8A=B8=20=EA=B0=B1=EC=8B=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit event-stream.md SESSION_MESSAGE_AUDIO 카탈로그, messaging.md 세그먼트 발행+프록시, storage.md 세그먼트 키 규칙(AI·Core SSOT), ai/backend CLAUDE.md 현재 상태. --- ai/CLAUDE.md | 1 + backend/CLAUDE.md | 1 + docs/event-stream.md | 14 +++++++++++++- docs/messaging.md | 5 +++++ docs/storage.md | 4 ++++ 5 files changed, 24 insertions(+), 1 deletion(-) diff --git a/ai/CLAUDE.md b/ai/CLAUDE.md index d4b74996..214efcbb 100644 --- a/ai/CLAUDE.md +++ b/ai/CLAUDE.md @@ -324,6 +324,7 @@ docker run --env-file .env -p 8000:8000 stackup-ai - 꼬리질문 + 답변 평가 (Flash 모델, `chain/followup_generation_chain.py`) - 콜백: `callback.questions` (`kind=POOL|FOLLOWUP`) - **꼬리질문 토큰 스트리밍 본 구현**: followup 출력을 `……{json}` 구분자 포맷으로 바꾸고(`chain/prompts/followup_generation.py`), `StreamingFollowupGenerator`(`astream`)가 `` 토큰만 `SessionRealtimeNotifier`(`messaging/session_notify.py`)로 `SESSION_MESSAGE_DELTA` 발행(`stackup.realtime`/`realtime.session.notify`, Core 우회). `DONT_KNOW` 면 델타 미발행. 종료 후 `parse_followup_result` 로 검증해 기존 `callback.questions(FOLLOWUP, followupMessageId)` 발행. 와이어링은 `messaging/runner.py`(분석 진행 publisher 재사용). +- **문장 단위 TTS 본 구현 (Part B)**: followup consumer 스트림 루프가 `chain/sentence_split.next_sentences` 로 문장 경계를 잡아, 문장마다 `TtsProvider` 인라인 합성(`asyncio.create_task` 백그라운드, 텍스트 델타 비차단)→S3 `interview/tts/{sid}/{mid}/seg-{seq}.{ext}` PUT→`SessionRealtimeNotifier.emit_audio`(`SESSION_MESSAGE_AUDIO`). 콜백 전 `gather` 로 수거. 라이브 세그먼트는 휘발성(DB 미기록). - **임베딩 본 구현** (`rag/`): `MarkdownChunker` + `GeminiEmbeddingProvider` (1536d, `gemini-embedding-001`). 운영/개발 default 는 gemini, 테스트는 `MockEmbeddingProvider`. - 청크를 `EMBEDDING_BATCH_SIZE`(기본 32) 단위로 쪼개 순차 호출 — 한 요청에 몰면 분당 토큰 한도(429 `RESOURCE_EXHAUSTED`)에 걸린다. diff --git a/backend/CLAUDE.md b/backend/CLAUDE.md index 90060393..be5ba071 100644 --- a/backend/CLAUDE.md +++ b/backend/CLAUDE.md @@ -362,6 +362,7 @@ docker compose up -d - 질문 TTS 발행 본 구현: 질문 영속 후 `QuestionPersistedEvent`(AFTER_COMMIT) → `SessionTtsRequester` 가 `generate.tts` 발행, `callback.tts` 수신해 메시지에 오디오 경로 반영(`TtsCallbackService`) - **꼬리질문 토큰 스트리밍 placeholder 본 구현**: `SessionFollowupRequester` 가 AI 호출 분기에서 INTERVIEWER placeholder(`InterviewMessage.followupPlaceholder`, content=`"(생성 중)"`, status=`CREATED`)를 선INSERT + `SESSION_MESSAGE` 발행 + `generate.followup.followupMessageId` 동봉. AI 가 토큰을 `SESSION_MESSAGE_DELTA` 로 흘린 뒤 `callback.questions(FOLLOWUP)` 도착 시 `QuestionsCallbackService` 가 분기: NORMAL→`completeFollowup` UPDATE+카운트, CLARIFICATION→UPDATE(카운트 X), DONT_KNOW→placeholder DELETE 후 `advanceToNextGeneral`. placeholder 없는 레거시 콜백은 기존 INSERT 폴백. +- **문장 단위 TTS 세그먼트 프록시 본 구현 (Part B)**: `InterviewMessageService.streamAudioSegment` + `GET /api/sessions/{sid}/messages/{mid}/audio/segments/{seq}?ext=`. AI 가 휘발성으로 쓴 라이브 세그먼트를 규칙(`interview/tts/{sid}/{mid}/seg-{seq}.{ext}`)으로 재구성해 프록시(DB 미기록). 소유권+ext 화이트리스트+seq>=0 검증으로 임의 키 노출 차단. - AI 호출 로깅 (US-30) 본 구현: `/api/internal/ai-logs` + `ai_request_logs` INSERT - **Spring AI 미사용** — LLM·임베딩 호출은 모두 AI 서버 위임. Core는 RabbitMQ 발행만 담당. - **Redis 미사용** — 휘발성 데이터는 DB short-lived 레코드 또는 인메모리로. diff --git a/docs/event-stream.md b/docs/event-stream.md index 758a3a82..0ab34bbe 100644 --- a/docs/event-stream.md +++ b/docs/event-stream.md @@ -40,7 +40,7 @@ WS /realtime/sessions/{sessionId}/audio # RT3 실시간 음성 답변 스트 ## 2. 이벤트 포맷 -> - `event` 이름은 **`SseEventType` enum 이름(대문자)**: `DOC_STATE`·`REPO_STATE`·`SESSION_MESSAGE`·`SESSION_STATE`·`FEEDBACK_READY`·`ERROR`·`KEEP_ALIVE` (+ AI 가 직접 발행하는 `ANALYSIS_PROGRESS`·`SESSION_MESSAGE_DELTA`). **`session.message` 같은 소문자 점표기가 아니다.** (RealTime `bridge/dispatcher.go` `Type: env.Payload.EventType` → `sse.go`/`ws.go` 가 그대로 전달.) RealTime 디스패처는 이벤트 타입을 화이트리스트 없이 투명 전달하므로, AI 가 새 이벤트 타입(`SESSION_MESSAGE_DELTA` 등)을 발행해도 RealTime 코드 변경이 필요 없다. +> - `event` 이름은 **`SseEventType` enum 이름(대문자)**: `DOC_STATE`·`REPO_STATE`·`SESSION_MESSAGE`·`SESSION_STATE`·`FEEDBACK_READY`·`ERROR`·`KEEP_ALIVE` (+ AI 가 직접 발행하는 `ANALYSIS_PROGRESS`·`SESSION_MESSAGE_DELTA`·`SESSION_MESSAGE_AUDIO`). **`session.message` 같은 소문자 점표기가 아니다.** (RealTime `bridge/dispatcher.go` `Type: env.Payload.EventType` → `sse.go`/`ws.go` 가 그대로 전달.) RealTime 디스패처는 이벤트 타입을 화이트리스트 없이 투명 전달하므로, AI 가 새 이벤트 타입(`SESSION_MESSAGE_DELTA` 등)을 발행해도 RealTime 코드 변경이 필요 없다. > - `data` 봉투는 `{"data": , "traceId": "..."}` 다 (`realtime/CLAUDE.md §8`). payload 필드는 camelCase. > - 클라는 SSE `addEventListener(, …)` / WS `frame.event === ''` 로 매칭한다. @@ -129,6 +129,18 @@ WS(RT1)는 같은 내용을 JSON 한 줄 프레임으로: `{ "id": , "e - 흐름: 답변 → (Core) placeholder INSERT + `SESSION_MESSAGE` → (AI) `astream` 으로 `SESSION_MESSAGE_DELTA` 연속 발행 → (AI) `callback.questions(FOLLOWUP, followupMessageId)` → (Core) placeholder UPDATE(content/COMPLETED) + `SESSION_MESSAGE`(종료) → 프론트 `GET …/messages` 재조회로 정본 reconcile. - `answer_intent=DONT_KNOW` 면 AI 가 델타를 **발행하지 않고**, Core 가 placeholder 삭제 후 다음 일반질문으로 진행한다. 이때 프론트는 placeholder 를 "생각 중"으로만 표시하다 일반질문으로 교체. +### 3.3-2 꼬리질문 문장 단위 TTS (`event: SESSION_MESSAGE_AUDIO`) + +**휘발성** 이벤트. 꼬리질문이 토큰으로 차오르는 동안, **문장이 완성될 때마다 그 문장만 TTS 합성**해 첫 소리까지의 지연을 단축한다. AI 서버가 인라인 합성 후 `stackup.realtime`(`realtime.session.notify`)으로 직접 발행. + +```json +{ "data": { "messageId": 503, "seq": 0, "ext": "mp3", "durationSec": 1.2 }, "traceId": "..." } +``` + +- `messageId` = `SESSION_MESSAGE_DELTA` 와 동일 placeholder id. `seq` = 오디오 세그먼트 순번(0부터, **델타 seq 와 독립**). `ext` ∈ `wav|mp3|ogg|m4a`. +- 세그먼트는 S3 `interview/tts/{sessionId}/{messageId}/seg-{seq}.{ext}` 에 저장(DB 미기록, 휘발성). 프론트는 Core 프록시 `GET /api/sessions/{sid}/messages/{mid}/audio/segments/{seq}?ext=` 로 받아 **seq 순서대로 순차 재생**. +- `DONT_KNOW` 면 발행 안 함. 라이브 세그먼트를 재생한 메시지는 완료 후 whole-message TTS **autoPlay 억제**(중복 재생 방지) — 수동 "다시 듣기"만 동작. + ### 3.4 세션 상태 (`event: SESSION_STATE`) ```json { diff --git a/docs/messaging.md b/docs/messaging.md index 687ed260..dddb11d8 100644 --- a/docs/messaging.md +++ b/docs/messaging.md @@ -399,6 +399,11 @@ 분석 진행(ANALYSIS_PROGRESS)과 동일 패턴으로, AI 서버가 꼬리질문 생성 중 토큰을 `stackup.realtime` exchange 의 `realtime.session.notify` 로 **직접** 발행(Core 미경유)해 세션 채널로 흘린다. 휘발성(영속 불필요)이며 발행 실패는 무시(경고만). - `context.sessionId` 로 세션 채널 라우팅. `payload.eventType` = `SESSION_MESSAGE_DELTA`, `payload.data` = `{ messageId, seq, text }` (§event-stream.md 3.3-1). - `messageId` = `generate.followup` 의 `followupMessageId`(placeholder). 종료/정본은 기존대로 Core 가 `callback.questions` 수신 후 `SESSION_MESSAGE`(messageId) 로 통지. + +### 5.12-2 `realtime.session.notify` — 문장 단위 TTS 세그먼트 (AI 직접 발행) +AI followup consumer 가 토큰 스트림 중 문장 경계마다 그 문장만 인라인 TTS 합성 → S3 세그먼트 PUT → `SESSION_MESSAGE_AUDIO` 를 `realtime.session.notify` 로 직접 발행(휘발성). 문장 합성은 `asyncio.create_task` 백그라운드(텍스트 델타 비차단), 콜백 발행 전 `gather` 로 수거. +- `payload.eventType` = `SESSION_MESSAGE_AUDIO`, `payload.data` = `{ messageId, seq, ext, durationSec }`. `seq` 는 오디오 전용(델타 seq 와 독립). +- 세그먼트 S3 키 규칙(AI·Core 공유): `interview/tts/{sessionId}/{messageId}/seg-{seq}.{ext}`. Core 는 DB 미기록, `GET …/messages/{mid}/audio/segments/{seq}?ext=` 프록시에서 소유권 검증 후 규칙으로 키 재구성(ext 화이트리스트 wav|mp3|ogg|m4a). - 상세 SSE 스펙: [`event-stream.md §3.2-1`](./event-stream.md). --- diff --git a/docs/storage.md b/docs/storage.md index a95ac3b6..ee7fd10c 100644 --- a/docs/storage.md +++ b/docs/storage.md @@ -24,10 +24,14 @@ analyzed/resume/{resume_id}/summary.md analyzed/repository/{repository_id}/summary.md session/{session_id}/audio/{message_id}.webm session/{session_id}/audio/{message_id}.transcript.txt +interview/tts/{session_id}/{message_id}.{ext} # 질문 whole-message TTS (영속, ttsAudioPath) +interview/tts/{session_id}/{message_id}/seg-{seq}.{ext} # 꼬리질문 문장 단위 TTS 세그먼트 (휘발성, DB 미기록) feedback/{session_id}/report.md feedback/{session_id}/charts/{name}.png ``` +> 세그먼트 키 규칙은 **AI(발행)·Core(프록시 재구성) 양측이 하드코딩으로 공유**하는 SSOT다 — 변경 시 양측 동시 수정. `{ext}` ∈ `wav|mp3|ogg|m4a`, `{seq}` 는 0부터의 오디오 세그먼트 순번. 세그먼트는 DB에 기록하지 않고 Core 프록시(`GET …/audio/segments/{seq}?ext=`)가 소유권 검증 후 규칙으로 키를 만든다. + 규칙: - 슬래시(`/`) 계층 구조로 탐색성 확보 - 사용자 식별자는 user_id (정수) 사용 — github_username은 변경 가능하므로 회피