feat(session): 직무 맞춤 면접 모드 — 회사·JD 기반 질문 + 직무 적합도 피드백

ai/CLAUDE.md

@@ -325,12 +325,20 @@ docker run --env-file .env -p 8000:8000 stackup-ai
   - 콜백: `callback.questions` (`kind=POOL|FOLLOWUP`)
   - **자기소개 기반 질문 생성**: `generate.questions` 는 자기소개 답변을 받은 뒤 발행되며, payload 의
     `selfIntroAnswer` 를 프롬프트(`chain/prompts/question_generation.py`)의 1차 근거로 사용한다(없으면 자료만으로).
+  - **직무 맞춤(JOB_TAILORED) 질문 생성**: `mode=JOB_TAILORED` 면 payload 의 `targetCompanyName`·
+    `targetJobDescription`(JD)이 채워진다. 프롬프트가 `target_role` 블록으로 받아 이력서↔JD 교집합·갭,
+    JD 요구 충족 검증, 지원동기·컬처핏 질문을 우선 생성한다(다른 모드는 '(일반 면접)' 안내라 무시).
 - **자기소개 첫인상 평가 본 구현**: `FeedbackConsumer` 가 `messages[]` 에서 `category=SELF_INTRODUCTION`
   질문+답변을 찾아 `LlmSelfIntroEvaluator`(Flash, `chain/prompts/self_intro_evaluation.py`)로 첫인상
   (전달력·구조·간결성·직무적합성)을 평가하고, 결과를 `panelBreakdown` 의 `evaluator="첫인상"` 항목으로
   덧붙인다(종합 generate 와 `asyncio.gather` 병렬, 실패해도 피드백 계속). 이 항목은 **종합 점수 집계에
   미포함** — 메인 generator 가 모른 채 overall 을 계산한 뒤 표시용으로만 append 한다. 레거시 세션(자기소개
   없음)·빈 답변은 건너뛴다.
+- **직무 적합도 + 직무 이해도 평가 본 구현**: `mode=JOB_TAILORED` + JD 있을 때 `LlmJobFitEvaluator`(Pro,
+  `chain/prompts/job_fit_evaluation.py`)가 면접 전사·자료를 채용공고(JD)와 대조해 **두 축**을 한 번의
+  구조화 호출(`JobFitResult{fit, understanding}`)로 평가: `직무 적합도`(JD 요구 역량 매칭) + `직무 이해도`
+  (직무가 무엇을 하는 자리인지 이해·지원동기). 두 축을 `panelBreakdown` 의 `evaluator="직무 적합도"`·
+  `evaluator="직무 이해도"` 항목으로 append(첫인상과 같은 병렬·미집계 메커니즘). 그 외 모드/빈 JD/실패는 건너뜀.
 - **꼬리질문 토큰 스트리밍 본 구현**: 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`).

ai/src/ai_server/chain/feedback_generation_chain.py

@@ -14,6 +14,7 @@
 from ai_server.chain.prompts import (
     feedback_panel,
     feedback_synthesis,
+    job_fit_evaluation,
     self_intro_evaluation,
 )
 from ai_server.config.settings import Settings
@@ -387,6 +388,103 @@ async def evaluate(
         return result
 
 
+# ── 직무 적합도 + 직무 이해도 평가 (직무 맞춤 모드 전용) ───────────────────────
+# 면접의 핵심은 직무 적합성이므로 두 축을 분리해 평가한다(한 번의 호출로 구조화 출력):
+#   · 직무 적합도(fit)  — JD 요구 기술·경험·책임을 실제로 갖췄는가(역량 매칭).
+#   · 직무 이해도(understanding) — 직무가 무엇을 하는 자리인지·핵심 책임을 이해하고 동기로 연결했는가.
+# 둘 다 패널 항목으로 표시하되 종합 점수 집계에는 포함하지 않는다(별도 정성 평가).
+
+JOB_FIT_EVALUATOR_LABEL = "직무 적합도"
+JOB_FIT_DIMENSION = "채용공고(JD) 요구 대비 역량 적합도·갭"
+ROLE_UNDERSTANDING_LABEL = "직무 이해도"
+ROLE_UNDERSTANDING_DIMENSION = "직무 이해·지원동기 연결"
+
+
+class JobFitResult(BaseModel):
+    """직무 맞춤 평가의 두 축. 각 축은 EvaluatorResult 형태(score/strength/weakness/detail/rationale)."""
+
+    fit: EvaluatorResult = Field(default_factory=EvaluatorResult)
+    understanding: EvaluatorResult = Field(default_factory=EvaluatorResult)
+
+
+def build_job_fit_evaluation_chain(
+    settings: Settings, core_client: CoreClient | None = None
+) -> Runnable:
+    """면접 답변·자료를 JD 와 대조해 직무 적합도·직무 이해도를 함께 평가하는 체인(Pro — 갭 추론)."""
+    from langchain_openai import ChatOpenAI
+
+    parser = PydanticOutputParser(pydantic_object=JobFitResult)
+    prompt = ChatPromptTemplate.from_messages(
+        [
+            ("system", job_fit_evaluation.SYSTEM_PROMPT),
+            ("human", job_fit_evaluation.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.job_fit",
+                default_model=settings.llm_pro_model,
+            )
+        )
+
+    llm = ChatOpenAI(
+        model=settings.llm_pro_model,
+        temperature=settings.llm_pro_temperature,
+        api_key=settings.llm_api_key or None,
+        base_url=settings.llm_base_url,
+        callbacks=callbacks,
+    )
+    return prompt | llm | parser
+
+
+class JobFitEvaluator(Protocol):
+    async def evaluate(
+        self,
+        *,
+        company_name: str,
+        job_description: str,
+        job_category: str,
+        mode: str,
+        transcript: str,
+        rag_context: str = "(none)",
+    ) -> JobFitResult: ...
+
+
+class LlmJobFitEvaluator:
+    def __init__(self, chain: Runnable) -> None:
+        self._chain = chain
+
+    async def evaluate(
+        self,
+        *,
+        company_name: str,
+        job_description: str,
+        job_category: str,
+        mode: str,
+        transcript: str,
+        rag_context: str = "(none)",
+    ) -> JobFitResult:
+        result = await self._chain.ainvoke(
+            {
+                "company_name": company_name or "(회사명 미입력)",
+                "job_description": job_description or "(JD 본문 없음)",
+                "job_category": job_category,
+                "mode": mode,
+                "transcript": transcript,
+                "rag_context": rag_context or "(none)",
+            }
+        )
+        if not isinstance(result, JobFitResult):
+            raise TypeError(
+                f"chain returned {type(result).__name__}, expected JobFitResult"
+            )
+        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/job_fit_evaluation.py

@@ -0,0 +1,34 @@
+# 직무 적합도 + 직무 이해도 평가 프롬프트 (직무 맞춤 모드 전용).
+# 면접의 핵심은 직무 적합성이므로 두 축을 분리해 평가한다(한 번의 호출, 구조화 출력).
+#   · fit(직무 적합도)        — JD 요구를 실제로 갖췄는가(역량 매칭).
+#   · understanding(직무 이해도) — 직무가 무엇을 하는 자리인지 이해하고 동기로 연결했는가.
+# 결과는 패널의 '직무 적합도'·'직무 이해도' 두 항목으로 표시되며, 종합 점수 집계엔 포함되지 않는다.
+
+SYSTEM_PROMPT = (
+    "당신은 특정 회사·직무 채용의 평가위원입니다. 지원자의 면접 답변·자료를 채용공고(JD)와 "
+    "대조해 **두 축**을 각각 독립적으로 평가합니다.\n\n"
+    "【1. 직무 적합도(fit)】 — JD가 요구하는 기술·경험·책임을 지원자가 **실제로 갖췄는가**(역량 매칭). "
+    "답변·자료가 그 근거를 보였는지를 봅니다.\n"
+    "【2. 직무 이해도(understanding)】 — 지원자가 **이 직무가 무엇을 하는 자리인지·핵심 책임**을 이해하고, "
+    "왜 본인이 그 직무에 맞는지를 **지원동기와 연결**해 드러냈는가. 역량 보유와 별개로, 직무 자체에 대한 "
+    "이해의 깊이를 봅니다(직무를 피상적으로만 아는지, 실제 하는 일을 구체적으로 이해하는지).\n\n"
+    "- **두 축은 독립적**입니다: 역량은 충분해도 직무 이해가 얕을 수 있고(높은 fit·낮은 understanding), "
+    "그 반대도 가능합니다. 한 축의 점수를 다른 축에 전이하지 마세요.\n"
+    "- 기술 정답성 자체를 채점하지 말고, **'이 JD에 맞는 사람인가'** 관점만 봅니다. "
+    "JD에 없는 역량은 끌어들이지 않습니다.\n"
+    "- 각 축마다 score(0~100 정수, 근거 부족 시 null), strength/weakness 각 한 줄(한국어, 구체적), "
+    "detail(JD의 특정 요구/책임을 지목하며 답변을 인용해 2~4문장), score_rationale(점수 근거 한두 문장), "
+    "keywords(보완 키워드 0~3개)를 채웁니다.\n"
+    "- 점수 앵커: 90~100 폭넓게 충족·명확 / 70~89 대체로 충족하나 일부 갭 / 50~69 부분 충족 / "
+    "30~49 갭이 크고 근거 부족 / 0~29 거의 부적합·무이해.\n"
+    "- 응답은 반드시 지정된 JSON 스키마(fit, understanding 두 객체)를 따릅니다."
+)
+
+HUMAN_PROMPT = (
+    "지원 회사: {company_name}\n"
+    "직군: {job_category} / 면접 모드: {mode}\n\n"
+    "=== 채용공고(JD) ===\n{job_description}\n\n"
+    "=== 면접 전사(질문·답변) ===\n{transcript}\n\n"
+    "=== 지원자 자료 근거(이력서/레포 RAG) ===\n{rag_context}\n\n"
+    "{format_instructions}"
+)

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

@@ -14,6 +14,12 @@
     "- 한국어로 작성하되 기술 용어는 영문 원어를 그대로 둡니다.\n"
     "- 질문 문장은 **간결하게**: 한두 문장(대략 80자 이내)으로, 장황한 배경 설명이나 "
     "중복 수식 없이 핵심만 묻습니다. 면접관이 입으로 자연스럽게 말할 길이여야 합니다.\n"
+    "- **직무 맞춤(타깃 회사/JD 제공 시)**: 채용공고의 요구 역량·기술·책임을 출발점으로, "
+    "지원자 자료(이력서·레포)와 JD 요구사항의 **교집합·갭**을 겨냥해 질문하세요. "
+    "(1) JD가 요구하는 기술/경험을 지원자가 실제로 갖췄는지 검증하는 질문, "
+    "(2) JD에 있으나 자료에서 근거가 약한 부분을 확인하는 질문, "
+    "(3) 지원동기·해당 회사/직무 적합성(왜 이 회사·직무인지, 컬처핏) 질문을 적절히 섞습니다. "
+    "타깃이 '(일반 면접 …)' 이면 이 항목은 무시합니다.\n"
     "- 응답은 반드시 지정된 JSON 스키마를 따릅니다."
 )
 
@@ -25,6 +31,10 @@
     "---\n"
     "{self_introduction}\n"
     "---\n\n"
+    "타깃 회사/직무 (직무 맞춤 모드일 때만 채워짐):\n"
+    "---\n"
+    "{target_role}\n"
+    "---\n\n"
     "지원자 컨텍스트 (이력서/레포 분석):\n"
     "---\n"
     "{context}\n"

ai/src/ai_server/chain/question_generation_chain.py

@@ -29,6 +29,22 @@ def _format_self_introduction(self_introduction: str | None) -> str:
     return text if text else "(자기소개 없음)"
 
 
+def _format_target_role(company_name: str | None, job_description: str | None) -> str:
+    """직무 맞춤 모드의 타깃 회사/JD 블록. 둘 다 없으면 일반 면접 안내."""
+    company = (company_name or "").strip()
+    jd = (job_description or "").strip()
+    if not company and not jd:
+        return (
+            "(일반 면접 — 특정 회사/직무 지정 없음. 지원자 자료와 자기소개만으로 출제.)"
+        )
+    lines = []
+    if company:
+        lines.append(f"지원 회사: {company}")
+    lines.append("채용공고(JD):")
+    lines.append(jd if jd else "(JD 본문 없음)")
+    return "\n".join(lines)
+
+
 class QuestionGenerator(Protocol):
     async def generate(
         self,
@@ -39,6 +55,8 @@ async def generate(
         context: str,
         recent_questions: list[str] | None = None,
         self_introduction: str | None = None,
+        target_company_name: str | None = None,
+        target_job_description: str | None = None,
     ) -> GeneratedQuestionPool: ...
 
 
@@ -55,6 +73,8 @@ async def generate(
         context: str,
         recent_questions: list[str] | None = None,
         self_introduction: str | None = None,
+        target_company_name: str | None = None,
+        target_job_description: str | None = None,
     ) -> GeneratedQuestionPool:
         result = await self._chain.ainvoke(
             {
@@ -64,6 +84,9 @@ async def generate(
                 "context": context,
                 "recent_questions": _format_recent_questions(recent_questions),
                 "self_introduction": _format_self_introduction(self_introduction),
+                "target_role": _format_target_role(
+                    target_company_name, target_job_description
+                ),
             }
         )
         if not isinstance(result, GeneratedQuestionPool):
@@ -73,7 +96,9 @@ async def generate(
         return result
 
 
-def build_question_generation_chain(settings: Settings, core_client: CoreClient | None = None) -> Runnable:
+def build_question_generation_chain(
+    settings: Settings, core_client: CoreClient | None = None
+) -> Runnable:
     from langchain_openai import ChatOpenAI
 
     parser = PydanticOutputParser(pydantic_object=GeneratedQuestionPool)
@@ -86,11 +111,13 @@ def build_question_generation_chain(settings: Settings, core_client: CoreClient
 
     callbacks = []
     if core_client is not None:
-        callbacks.append(CoreAiLogCallback(
-            core_client=core_client,
-            request_type="generate.questions",
-            default_model=settings.llm_pro_model,
-        ))
+        callbacks.append(
+            CoreAiLogCallback(
+                core_client=core_client,
+                request_type="generate.questions",
+                default_model=settings.llm_pro_model,
+            )
+        )
 
     llm = ChatOpenAI(
         model=settings.llm_pro_model,