feat(feedback): 질문별 복기 — 답변마다 모범 답안·리라이트·코칭

ai/CLAUDE.md

@@ -334,6 +334,10 @@ docker run --env-file .env -p 8000:8000 stackup-ai
   덧붙인다(종합 generate 와 `asyncio.gather` 병렬, 실패해도 피드백 계속). 이 항목은 **종합 점수 집계에
   미포함** — 메인 generator 가 모른 채 overall 을 계산한 뒤 표시용으로만 append 한다. 레거시 세션(자기소개
   없음)·빈 답변은 건너뛴다.
+- **질문별 복기(답변 코칭) 본 구현**: `FeedbackConsumer` 가 자기소개 제외 모든 (질문,답변) 쌍을 찾아
+  `LlmAnswerCoach`(Flash, `chain/prompts/answer_coaching.py`)로 **답변별 병렬** 코칭 — 모범 답안 + 내 답변
+  리라이트 + 한 줄 코칭. `callback.feedback.answerCoaching[{messageId,…}]` 로 보내고 Core 가 각 답변 메시지에
+  기록(종료 세션 조회에서만 노출). 종합 generate·첫인상·직무 적합도와 `asyncio.gather` 병렬.
 - **직무 적합도 + 직무 이해도 평가 본 구현**: `mode=JOB_TAILORED` + JD 있을 때 `LlmJobFitEvaluator`(Pro,
   `chain/prompts/job_fit_evaluation.py`)가 면접 전사·자료를 채용공고(JD)와 대조해 **두 축**을 한 번의
   구조화 호출(`JobFitResult{fit, understanding}`)로 평가: `직무 적합도`(JD 요구 역량 매칭) + `직무 이해도`

ai/src/ai_server/chain/feedback_generation_chain.py

@@ -12,6 +12,7 @@
 
 from ai_server.chain.prompts.feedback_generation import HUMAN_PROMPT, SYSTEM_PROMPT
 from ai_server.chain.prompts import (
+    answer_coaching,
     feedback_panel,
     feedback_synthesis,
     job_fit_evaluation,
@@ -485,6 +486,100 @@ async def evaluate(
         return result
 
 
+# ── 질문별 복기 (답변 코칭) ───────────────────────────────────────────────────
+# 답변 1건당 모범 답안 + 리라이트 + 한 줄 코칭. 점수가 아니라 "어떻게 더 잘하는지"를 준다.
+# 답변별 병렬 호출(Flash). 자기소개 답변은 제외(첫인상 평가가 커버).
+
+
+class CoachingResult(BaseModel):
+    model_answer: str | None = Field(None, description="이 질문에 대한 강한 답변 예시")
+    answer_rewrite: str | None = Field(
+        None, description="지원자 답변을 더 좋게 고쳐 쓴 버전"
+    )
+    coaching_comment: str | None = Field(None, description="가장 중요한 보완점 한 문장")
+
+
+def build_answer_coaching_chain(
+    settings: Settings, core_client: CoreClient | None = None
+) -> Runnable:
+    """답변 1건을 코칭(모범 답안·리라이트·한 줄 코칭)하는 체인(Flash — 답변 수만큼 병렬)."""
+    from langchain_openai import ChatOpenAI
+
+    parser = PydanticOutputParser(pydantic_object=CoachingResult)
+    prompt = ChatPromptTemplate.from_messages(
+        [
+            ("system", answer_coaching.SYSTEM_PROMPT),
+            ("human", answer_coaching.HUMAN_PROMPT),
+        ]
+    ).partial(format_instructions=parser.get_format_instructions())
+
+    callbacks = []
+    if core_client is not None:
+        callbacks.append(
+            CoreAiLogCallback(
+                core_client=core_client,
+                request_type="generate.feedback.coaching",
+                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,
+        callbacks=callbacks,
+    )
+    return prompt | llm | parser
+
+
+class AnswerCoach(Protocol):
+    async def coach(
+        self,
+        *,
+        job_category: str,
+        mode: str,
+        target_role: str,
+        question: str,
+        expected_signal: str,
+        answer: str,
+        rag_context: str = "(none)",
+    ) -> CoachingResult: ...
+
+
+class LlmAnswerCoach:
+    def __init__(self, chain: Runnable) -> None:
+        self._chain = chain
+
+    async def coach(
+        self,
+        *,
+        job_category: str,
+        mode: str,
+        target_role: str,
+        question: str,
+        expected_signal: str,
+        answer: str,
+        rag_context: str = "(none)",
+    ) -> CoachingResult:
+        result = await self._chain.ainvoke(
+            {
+                "job_category": job_category,
+                "mode": mode,
+                "target_role": target_role or "",
+                "question": question,
+                "expected_signal": expected_signal or "(명시 없음)",
+                "answer": answer or "(빈 답변)",
+                "rag_context": rag_context or "(none)",
+            }
+        )
+        if not isinstance(result, CoachingResult):
+            raise TypeError(
+                f"chain returned {type(result).__name__}, expected CoachingResult"
+            )
+        return result
+
+
 def _weighted_overall(pairs: list[tuple[float | None, float]]) -> float | None:
     """(score, weight) 중 score 가 있는 것만 가중평균. 전부 None 이면 None."""
     present = [(s, w) for s, w in pairs if s is not None and w > 0]

ai/src/ai_server/chain/prompts/answer_coaching.py

@@ -0,0 +1,26 @@
+# 질문별 복기 프롬프트 — 답변 1건을 코칭한다.
+# 모범 답안(이 질문에 강한 답) + 지원자 답변 리라이트(이 답을 이렇게 고치면 더 좋다) + 한 줄 코칭.
+# 면접 준비 도구의 핵심: "점수"가 아니라 "어떻게 더 잘하는지"를 보여준다.
+
+SYSTEM_PROMPT = (
+    "당신은 IT 면접 코치입니다. 면접관 질문 하나와 지원자의 실제 답변을 보고, 지원자가 "
+    "**어떻게 더 잘 답할 수 있는지**를 구체적으로 코칭합니다.\n"
+    "- **model_answer (모범 답안)**: 이 질문에 대한 강한 답변의 예시. 질문이 기대하는 핵심(기대 신호)을 "
+    "짚고, 가능하면 지원자의 실제 경험/자료(아래 컨텍스트)를 근거로 구체적으로. 1~2문단, 면접에서 말할 분량.\n"
+    "- **answer_rewrite (내 답변 리라이트)**: 지원자의 *실제 답변을 출발점으로* 더 좋게 고쳐 쓴 버전. "
+    "없는 경험을 지어내지 말고, 있는 내용을 구조(두괄식·근거·결과)와 구체성으로 보강. 답변이 비었거나 "
+    "'모르겠다'면 '이렇게 접근했다면' 식으로 짧게.\n"
+    "- **coaching_comment (한 줄 코칭)**: 이 답변에서 가장 중요한 보완점 하나를 한 문장으로.\n"
+    "- 한국어로, 기술 용어는 영문 원어 유지. 과장·미사여구 금지, 실전 조언만.\n"
+    "- 응답은 반드시 지정된 JSON 스키마를 따릅니다."
+)
+
+HUMAN_PROMPT = (
+    "직군: {job_category} / 면접 모드: {mode}\n"
+    "{target_role}\n\n"
+    "=== 면접관 질문 ===\n{question}\n"
+    "질문이 기대한 핵심(기대 신호): {expected_signal}\n\n"
+    "=== 지원자의 실제 답변 ===\n{answer}\n\n"
+    "=== 지원자 자료 근거(이력서/레포 RAG) ===\n{rag_context}\n\n"
+    "{format_instructions}"
+)

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

@@ -12,6 +12,7 @@
     ROLE_UNDERSTANDING_LABEL,
     SELF_INTRO_DIMENSION,
     SELF_INTRO_EVALUATOR_LABEL,
+    AnswerCoach,
     EvaluatorResult,
     FeedbackGenerator,
     JobFitEvaluator,
@@ -22,6 +23,7 @@
 from ai_server.messaging.publisher import CallbackPublisher
 from ai_server.model.envelope import Envelope
 from ai_server.model.messages.feedback import (
+    AnswerCoachingItem,
     FeedbackCallbackPayload,
     FeedbackMessageItem,
     GenerateFeedbackRequest,
@@ -59,6 +61,8 @@ def __init__(
         rag_top_k: int = 5,
         self_intro_evaluator: SelfIntroEvaluator | None = None,
         job_fit_evaluator: JobFitEvaluator | None = None,
+        answer_coach: AnswerCoach | None = None,
+        coaching_max_answers: int = 30,
     ) -> None:
         self._generator = generator
         self._publisher = publisher
@@ -69,6 +73,8 @@ def __init__(
         self._rag_top_k = rag_top_k
         self._self_intro_evaluator = self_intro_evaluator
         self._job_fit_evaluator = job_fit_evaluator
+        self._answer_coach = answer_coach
+        self._coaching_max_answers = coaching_max_answers
 
     async def handle(self, message: AbstractIncomingMessage) -> None:
         async with message.process(requeue=False):
@@ -111,20 +117,23 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
 
             # 종합 피드백 + 자기소개 첫인상 + 직무 적합도(직무 맞춤 모드)를 병렬 실행.
             # 첫인상·직무 적합도는 종합 점수(overall)에 미포함 — generator 가 모른 채 계산한 뒤 표시용으로 덧붙인다.
-            result, self_intro_item, job_fit_items = await asyncio.gather(
-                self._generator.generate(
-                    job_category=req.job_category,
-                    mode=req.mode,
-                    total_question_count=req.total_question_count,
-                    end_reason=req.end_reason,
-                    transcript=transcript,
-                    score_basis=score_basis,
-                    rag_context=rag_context,
-                    voice_analysis_summary=voice_analysis_summary,
-                    domain_question_counts=req.domain_question_counts,
-                ),
-                self._evaluate_self_intro(req, voice_analysis_summary),
-                self._evaluate_job_fit(req, transcript, rag_context),
+            result, self_intro_item, job_fit_items, answer_coaching = (
+                await asyncio.gather(
+                    self._generator.generate(
+                        job_category=req.job_category,
+                        mode=req.mode,
+                        total_question_count=req.total_question_count,
+                        end_reason=req.end_reason,
+                        transcript=transcript,
+                        score_basis=score_basis,
+                        rag_context=rag_context,
+                        voice_analysis_summary=voice_analysis_summary,
+                        domain_question_counts=req.domain_question_counts,
+                    ),
+                    self._evaluate_self_intro(req, voice_analysis_summary),
+                    self._evaluate_job_fit(req, transcript, rag_context),
+                    self._coach_answers(req, rag_context),
+                )
             )
             if self_intro_item is not None:
                 result.panel_breakdown.append(self_intro_item)
@@ -141,6 +150,7 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 improvement_keywords=result.improvement_keywords,
                 study_plan=result.study_plan,
                 panel_breakdown=result.panel_breakdown,
+                answer_coaching=answer_coaching,
                 report_s3_key=None,
             )
 
@@ -227,6 +237,56 @@ async def _evaluate_job_fit(
             ),
         ]
 
+    async def _coach_answers(
+        self, req: GenerateFeedbackRequest, rag_context: str
+    ) -> list[AnswerCoachingItem]:
+        """자기소개 제외 답변마다 모범 답안·리라이트·코칭을 병렬 생성 → 메시지별 복기 리스트."""
+        if self._answer_coach is None:
+            return []
+        pairs = _collect_coachable_pairs(req.messages)
+        if not pairs:
+            return []
+        if len(pairs) > self._coaching_max_answers:
+            log.info(
+                "feedback.coaching.capped",
+                session_id=req.session_id,
+                total=len(pairs),
+                cap=self._coaching_max_answers,
+            )
+            pairs = pairs[: self._coaching_max_answers]
+        target_role = _coaching_target_role(req)
+
+        async def _one(
+            question: FeedbackMessageItem, answer: FeedbackMessageItem
+        ) -> AnswerCoachingItem | None:
+            try:
+                res = await self._answer_coach.coach(
+                    job_category=req.job_category,
+                    mode=req.mode,
+                    target_role=target_role,
+                    question=question.content,
+                    expected_signal=question.expected_signal or "",
+                    answer=answer.content,
+                    rag_context=rag_context,
+                )
+            except Exception as exc:  # noqa: BLE001
+                log.warning(
+                    "feedback.coaching.failed",
+                    error=str(exc),
+                    session_id=req.session_id,
+                    message_id=answer.id,
+                )
+                return None
+            return AnswerCoachingItem(
+                message_id=answer.id,
+                model_answer=res.model_answer,
+                answer_rewrite=res.answer_rewrite,
+                coaching_comment=res.coaching_comment,
+            )
+
+        items = await asyncio.gather(*(_one(q, a) for q, a in pairs))
+        return [it for it in items if it is not None]
+
     async def _build_rag_context(self, req: GenerateFeedbackRequest) -> str:
         if not self._embedder or not req.context_document_ids:
             return "(none)"
@@ -256,6 +316,36 @@ async def _build_rag_context(self, req: GenerateFeedbackRequest) -> str:
         )
 
 
+def _collect_coachable_pairs(
+    messages: list[FeedbackMessageItem],
+) -> list[tuple[FeedbackMessageItem, FeedbackMessageItem]]:
+    """복기 대상 (질문, 답변) 쌍. 자기소개 질문에 대한 답변과 빈 답변은 제외, 시퀀스 순."""
+    by_id = {m.id: m for m in messages}
+    pairs: list[tuple[FeedbackMessageItem, FeedbackMessageItem]] = []
+    for m in messages:
+        if m.role != "INTERVIEWEE" or not (m.content or "").strip():
+            continue
+        question = by_id.get(m.parent_message_id) if m.parent_message_id else None
+        if question is None or question.role != "INTERVIEWER":
+            continue
+        if (question.category or "") == _SELF_INTRO_CATEGORY:
+            continue  # 자기소개는 첫인상 평가가 커버
+        pairs.append((question, m))
+    return pairs
+
+
+def _coaching_target_role(req: GenerateFeedbackRequest) -> str:
+    """직무 맞춤 모드일 때만 코칭 프롬프트에 실을 회사/JD 발췌. 그 외 빈 문자열."""
+    if (req.mode or "") != _JOB_TAILORED_MODE:
+        return ""
+    jd = (req.target_job_description or "").strip()
+    if not jd:
+        return ""
+    company = (req.target_company_name or "").strip()
+    head = f"지원 회사: {company}\n" if company else ""
+    return f"{head}채용공고(JD) 발췌: {jd[:800]}"
+
+
 def _to_panel_item(
     label: str, dimension: str, ev: EvaluatorResult
 ) -> PanelBreakdownItem:

ai/src/ai_server/messaging/runner.py

@@ -20,9 +20,11 @@
     build_streaming_followup_generator,
 )
 from ai_server.chain.feedback_generation_chain import (
+    LlmAnswerCoach,
     LlmJobFitEvaluator,
     LlmSelfIntroEvaluator,
     PanelFeedbackGenerator,
+    build_answer_coaching_chain,
     build_feedback_synthesis_chain,
     build_job_fit_evaluation_chain,
     build_panel_evaluator_chain,
@@ -239,6 +241,10 @@ def __init__(self, settings: Settings) -> None:
             job_fit_evaluator=LlmJobFitEvaluator(
                 build_job_fit_evaluation_chain(settings, core_client=core_client)
             ),
+            # 질문별 복기(Flash, 답변 수만큼 병렬). 모범 답안+리라이트+한 줄 코칭.
+            answer_coach=LlmAnswerCoach(
+                build_answer_coaching_chain(settings, core_client=core_client)
+            ),
         )
 
         # 음성 답변 STT + 분석 (Phase 2)

ai/src/ai_server/model/messages/feedback.py

@@ -85,6 +85,17 @@ class PanelBreakdownItem(BaseModel):
     score_rationale: str | None = None
 
 
+class AnswerCoachingItem(BaseModel):
+    """질문별 복기 1건 (AI → Core). 해당 답변(INTERVIEWEE) 메시지에 기록된다."""
+
+    model_config = camel_config()
+
+    message_id: int
+    model_answer: str | None = None
+    answer_rewrite: str | None = None
+    coaching_comment: str | None = None
+
+
 class FeedbackCallbackPayload(BaseModel):
     """AI → Core 종합 피드백. 점수는 0~100 (NULL 허용)."""
 
@@ -102,4 +113,6 @@ class FeedbackCallbackPayload(BaseModel):
     study_plan: list[str] = Field(default_factory=list)
     # 평가위원별 분해(패널). 비어 있으면 단일/레거시 경로.
     panel_breakdown: list[PanelBreakdownItem] = Field(default_factory=list)
+    # 질문별 복기(답변 메시지별 모범 답안·리라이트·코칭). 비면 복기 없음.
+    answer_coaching: list[AnswerCoachingItem] = Field(default_factory=list)
     report_s3_key: str | None = None