feat: 면접 꼬리질문 토큰 스트리밍 + 문장 단위 TTS (체감 반응 속도 개선)

ai/CLAUDE.md

@@ -323,6 +323,8 @@ docker run --env-file .env -p 8000:8000 stackup-ai
   - 질문 풀 생성 (Pro 모델, `chain/question_generation_chain.py`)
   - 꼬리질문 + 답변 평가 (Flash 모델, `chain/followup_generation_chain.py`)
   - 콜백: `callback.questions` (`kind=POOL|FOLLOWUP`)
+- **꼬리질문 토큰 스트리밍 본 구현**: followup 출력을 `<intent>…</intent><question>…</question><meta>{json}</meta>` 구분자 포맷으로 바꾸고(`chain/prompts/followup_generation.py`), `StreamingFollowupGenerator`(`astream`)가 `<question>` 토큰만 `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`)에 걸린다.

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"<intent>\s*(.*?)\s*</intent>", re.DOTALL)
+_QUESTION_RE = re.compile(r"<question>\s*(.*?)\s*</question>", re.DOTALL)
+_META_RE = re.compile(r"<meta>\s*(\{.*?\})\s*</meta>", re.DOTALL)
+_VALID_INTENTS = {"NORMAL", "DONT_KNOW", "CLARIFICATION"}
+
+
+def extract_question_span(text: str) -> str:
+    """완성 또는 진행 중 텍스트에서 <question> 안쪽만 추출(닫는 태그 없어도 가능)."""
+    closed = _QUESTION_RE.search(text)
+    if closed:
+        return closed.group(1).strip()
+    open_idx = text.find("<question>")
+    if open_idx == -1:
+        return ""
+    tail = text[open_idx + len("<question>") :]
+    tail = tail.split("<meta>", 1)[0]
+    # 닫는/메타 태그가 스트림 청크 경계로 잘려 들어와도(예: "질문?</quest")
+    # 미완성 '<...' 꼬리를 잘라낸다 — 질문 본문에는 '<' 가 없다고 가정.
+    lt = tail.rfind("<")
+    if lt != -1 and ">" 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 으로 흘리며 <question> 토큰만 콜백으로 내보낸다.
+
+    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)

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"
+    "  <intent>NORMAL 또는 DONT_KNOW 또는 CLARIFICATION</intent>\n"
+    "  <question>지원자에게 보여줄 한국어 질문 1개(또는 CLARIFICATION 시 재설명 문장)</question>\n"
+    "  <meta>{{\"specificity\": <0~5>, \"logic\": <0~5>, "
+    "\"structure\": \"FULL_STAR|PARTIAL_STAR|NONE\", \"correctness\": <0~5 또는 null>}}</meta>\n"
+    "- <intent> 를 가장 먼저 확정해 출력합니다. DONT_KNOW 여도 <question> 블록은 비우지 말고 한 문장 넣되, "
+    "표시는 시스템이 결정합니다."
 )
 
 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"
 )

ai/src/ai_server/chain/sentence_split.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import re
+
+# 종결: 문장부호(. ! ? …) 뒤에 공백/끝, 또는 줄바꿈.
+_SENTENCE_END = re.compile(r".*?(?:[.!?…](?=[ \t]|$)|\n)", re.DOTALL)
+
+
+def next_sentences(text: str, consumed: int) -> tuple[list[str], int]:
+    """text[consumed:] 에서 '완성된' 문장들만 떼어 반환하고 새 consumed 를 돌려준다.
+
+    미완성(종결부호 없는) 꼬리는 남겨 다음 호출에서 이어 처리한다.
+
+    반환 정책:
+    - 완성 문장이 없으면 ([], consumed) 반환.
+    - 완성 문장이 1개이거나 마지막 완성 문장이 '\\n' 으로 끝나면 전부 반환.
+    - 그 외(마지막 완성 문장이 공백 앞 구두점으로만 끝나고 2개 이상 존재)에는
+      마지막 문장을 버퍼에 남기고 나머지만 반환한다. 이는 스트리밍 중 아직
+      다음 토큰이 올 수 있는 경계를 보수적으로 처리하기 위함이다.
+    """
+    rest = text[consumed:]
+
+    # 각 완성 문장의 (stripped 텍스트, 줄바꿈 종결 여부, 매치 끝 위치) 수집
+    segments: list[tuple[str, bool, int]] = []
+    pos = 0
+    for m in _SENTENCE_END.finditer(rest):
+        chunk = rest[pos : m.end()].strip()
+        is_newline = m.group().endswith("\n")
+        if chunk:
+            segments.append((chunk, is_newline, m.end()))
+        pos = m.end()
+
+    if not segments:
+        return [], consumed
+
+    # 마지막 문장이 줄바꿈 종결이거나 완성 문장이 1개뿐이면 전부 방출
+    last_is_newline = segments[-1][1]
+    if last_is_newline or len(segments) == 1:
+        out = [s for s, _, _ in segments]
+        new_consumed = consumed + segments[-1][2]
+    else:
+        # 마지막 space-terminated 문장은 버퍼로 남김
+        to_emit = segments[:-1]
+        out = [s for s, _, _ in to_emit]
+        new_consumed = consumed + to_emit[-1][2]
+
+    return out, new_consumed