feat(feedback): 자기소개 첫인상 평가 추가 (종합 점수 미포함)

ai/CLAUDE.md

@@ -325,6 +325,12 @@ 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차 근거로 사용한다(없으면 자료만으로).
+- **자기소개 첫인상 평가 본 구현**: `FeedbackConsumer` 가 `messages[]` 에서 `category=SELF_INTRODUCTION`
+  질문+답변을 찾아 `LlmSelfIntroEvaluator`(Flash, `chain/prompts/self_intro_evaluation.py`)로 첫인상
+  (전달력·구조·간결성·직무적합성)을 평가하고, 결과를 `panelBreakdown` 의 `evaluator="첫인상"` 항목으로
+  덧붙인다(종합 generate 와 `asyncio.gather` 병렬, 실패해도 피드백 계속). 이 항목은 **종합 점수 집계에
+  미포함** — 메인 generator 가 모른 채 overall 을 계산한 뒤 표시용으로만 append 한다. 레거시 세션(자기소개
+  없음)·빈 답변은 건너뛴다.
 - **꼬리질문 토큰 스트리밍 본 구현**: 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

@@ -11,7 +11,11 @@
 from pydantic import BaseModel, Field
 
 from ai_server.chain.prompts.feedback_generation import HUMAN_PROMPT, SYSTEM_PROMPT
-from ai_server.chain.prompts import feedback_panel, feedback_synthesis
+from ai_server.chain.prompts import (
+    feedback_panel,
+    feedback_synthesis,
+    self_intro_evaluation,
+)
 from ai_server.config.settings import Settings
 from ai_server.core.client import CoreClient
 from ai_server.model.messages.feedback import PanelBreakdownItem
@@ -194,16 +198,18 @@ def _domain_specs_weighted(
     for dom, cnt in domain_question_counts.items():
         ko = _DOMAIN_KO.get((dom or "").upper(), dom)
         weight = float(cnt) if cnt and cnt > 0 else 1.0
-        specs.append((
-            _EvaluatorSpec(
-                key=f"tech:{dom}",
-                label=ko,
-                persona=f"{ko} 직군 시니어 기술 면접관",
-                dimension_name="기술 정확도·깊이",
-                dimension_guide=_TECH_GUIDE,
-            ),
-            weight,
-        ))
+        specs.append(
+            (
+                _EvaluatorSpec(
+                    key=f"tech:{dom}",
+                    label=ko,
+                    persona=f"{ko} 직군 시니어 기술 면접관",
+                    dimension_name="기술 정확도·깊이",
+                    dimension_guide=_TECH_GUIDE,
+                ),
+                weight,
+            )
+        )
     return specs or [(_domain_spec(job_category, mode), 1.0)]
 
 
@@ -297,6 +303,90 @@ def build_feedback_synthesis_chain(
     return prompt | llm | parser
 
 
+# ── 자기소개(첫인상) 평가 ─────────────────────────────────────────────────────
+# 모든 면접의 첫 질문은 자기소개다. 기술 채점이 아닌 첫인상·전달력만 별도로 평가해
+# 패널의 '첫인상' 항목으로 표시한다. 종합 점수 집계에는 포함하지 않는다(별도 정성 평가).
+
+SELF_INTRO_EVALUATOR_LABEL = "첫인상"
+SELF_INTRO_DIMENSION = "자기소개 전달력·구성·직무적합성"
+
+
+def build_self_intro_evaluation_chain(
+    settings: Settings, core_client: CoreClient | None = None
+) -> Runnable:
+    """자기소개 답변 1건을 첫인상(전달력·구조·간결성·직무적합성)으로 평가하는 경량 체인(Flash)."""
+    from langchain_openai import ChatOpenAI
+
+    parser = PydanticOutputParser(pydantic_object=EvaluatorResult)
+    prompt = ChatPromptTemplate.from_messages(
+        [
+            ("system", self_intro_evaluation.SYSTEM_PROMPT),
+            ("human", self_intro_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.self_intro",
+                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 SelfIntroEvaluator(Protocol):
+    async def evaluate(
+        self,
+        *,
+        job_category: str,
+        mode: str,
+        self_intro_question: str,
+        self_intro_answer: str,
+        voice_analysis_summary: str = "",
+    ) -> EvaluatorResult: ...
+
+
+class LlmSelfIntroEvaluator:
+    def __init__(self, chain: Runnable) -> None:
+        self._chain = chain
+
+    async def evaluate(
+        self,
+        *,
+        job_category: str,
+        mode: str,
+        self_intro_question: str,
+        self_intro_answer: str,
+        voice_analysis_summary: str = "",
+    ) -> EvaluatorResult:
+        result = await self._chain.ainvoke(
+            {
+                "job_category": job_category,
+                "mode": mode,
+                "self_intro_question": self_intro_question or "자기소개를 해주세요.",
+                "self_intro_answer": self_intro_answer or "(빈 답변)",
+                "voice_analysis_summary": voice_analysis_summary
+                or "No voice analysis summary was provided.",
+            }
+        )
+        if not isinstance(result, EvaluatorResult):
+            raise TypeError(
+                f"chain returned {type(result).__name__}, expected EvaluatorResult"
+            )
+        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]
@@ -307,7 +397,9 @@ def _weighted_overall(pairs: list[tuple[float | None, float]]) -> float | None:
 
 
 def _merge_notes(items: list[tuple[str, str | None]]) -> str | None:
-    parts = [f"[{label}] {note.strip()}" for label, note in items if note and note.strip()]
+    parts = [
+        f"[{label}] {note.strip()}" for label, note in items if note and note.strip()
+    ]
     return " ".join(parts) if parts else None
 
 
@@ -351,7 +443,9 @@ async def generate(
         score_basis: str = "(없음)",
         domain_question_counts: dict[str, int] | None = None,
     ) -> FeedbackResult:
-        domain_specs = _domain_specs_weighted(job_category, mode, domain_question_counts or {})
+        domain_specs = _domain_specs_weighted(
+            job_category, mode, domain_question_counts or {}
+        )
         # 평가위원 순서: 직군 기술 평가위원(N) + 논리 + 전달.
         specs = [s for s, _ in domain_specs] + [_LOGIC_SPEC, _COMM_SPEC]
         shared = {
@@ -386,16 +480,26 @@ async def generate(
             if isinstance(r, EvaluatorResult):
                 results.append(r)
             else:
-                log.warning("feedback.panel.evaluator_failed", evaluator=spec.key, error=str(r))
+                log.warning(
+                    "feedback.panel.evaluator_failed", evaluator=spec.key, error=str(r)
+                )
                 results.append(EvaluatorResult())
 
         n_domain = len(domain_specs)
-        domain_results = list(zip([s for s, _ in domain_specs], [w for _, w in domain_specs], results[:n_domain]))
+        domain_results = list(
+            zip(
+                [s for s, _ in domain_specs],
+                [w for _, w in domain_specs],
+                results[:n_domain],
+            )
+        )
         logic = results[n_domain]
         comm = results[n_domain + 1]
 
         # technical_accuracy = 직군 평가위원 점수의 질문수 가중평균.
-        technical_accuracy = _weighted_overall([(r.score, w) for _, w, r in domain_results])
+        technical_accuracy = _weighted_overall(
+            [(r.score, w) for _, w, r in domain_results]
+        )
         overall = _weighted_overall(
             [
                 (technical_accuracy, self._w_tech),
@@ -443,7 +547,12 @@ async def generate(
             )
 
         # 종합 서술형 + 학습 방향(synthesis). 미설정/실패 시 기계적 병합으로 폴백.
-        strengths, weaknesses, keywords, study_plan = fb_strengths, fb_weaknesses, fb_keywords, []
+        strengths, weaknesses, keywords, study_plan = (
+            fb_strengths,
+            fb_weaknesses,
+            fb_keywords,
+            [],
+        )
         if self._synthesis is not None:
             panel_summary = "\n".join(
                 f"- {b.evaluator}({b.dimension}) "

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

@@ -0,0 +1,28 @@
+# 자기소개(첫 질문) 답변 평가 프롬프트.
+# 기술 정답성이 아니라 첫인상·전달력(전달력·구조·간결성·직무적합성)만 본다.
+# 결과는 패널의 '첫인상' 평가위원 한 명으로 표시되며, 종합 점수 집계에는 포함되지 않는다.
+
+SYSTEM_PROMPT = (
+    "당신은 IT 직군 면접의 **첫인상 평가위원**입니다. 지원자의 자기소개 답변 한 건만 보고 "
+    "면접 도입부의 첫인상을 평가합니다.\n"
+    "- **기술 정확성은 평가하지 않습니다.** 오직 전달력·구조·간결성·직무적합성만 봅니다.\n"
+    "  · 전달력/명료성: 메시지가 분명하고 자신감 있게 전달되는가\n"
+    "  · 구조: 두괄식·핵심→근거 흐름 등 짜임새가 있는가\n"
+    "  · 간결성: 장황하지 않고 적정 길이인가\n"
+    "  · 직무적합성: 지원 직군/면접 맥락과 연결되는 강점·경험을 드러내는가\n"
+    "- 점수는 0~100 정수. 답변이 비었거나 너무 짧아 판단 불가하면 null.\n"
+    "- 점수 앵커: 90~100 매우 인상적이고 직무 연결까지 명확 / 70~89 무난하나 일부 아쉬움 / "
+    "50~69 전달은 되나 구조·직무연결 약함 / 30~49 산만하거나 너무 짧음 / 0~29 거의 무응답.\n"
+    "- strength/weakness 는 각각 한 줄(한국어, 구체적으로). keywords 는 보완 키워드 0~3개(짧은 명사구).\n"
+    "- detail: 자기소개의 구체적 부분을 인용/지목하며 2~4문장으로 서술(추상적 총평 금지).\n"
+    "- score_rationale: 그 점수를 준 핵심 근거를 한두 문장으로.\n"
+    "- 응답은 반드시 지정된 JSON 스키마를 따른다."
+)
+
+HUMAN_PROMPT = (
+    "지원 직군: {job_category} / 면접 모드: {mode}\n\n"
+    "=== 면접관 질문(자기소개) ===\n{self_intro_question}\n\n"
+    "=== 지원자 자기소개 답변 ===\n{self_intro_answer}\n\n"
+    "=== (참고) 세션 전체 음성 지표 — 자기소개 단독이 아님 ===\n{voice_analysis_summary}\n\n"
+    "{format_instructions}"
+)

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

@@ -1,9 +1,16 @@
 from __future__ import annotations
 
+import asyncio
+
 import structlog
 from aio_pika.abc import AbstractIncomingMessage
 
-from ai_server.chain.feedback_generation_chain import FeedbackGenerator
+from ai_server.chain.feedback_generation_chain import (
+    SELF_INTRO_DIMENSION,
+    SELF_INTRO_EVALUATOR_LABEL,
+    FeedbackGenerator,
+    SelfIntroEvaluator,
+)
 from ai_server.core.client import CoreClient
 from ai_server.messaging.idempotency import LruIdempotencyStore
 from ai_server.messaging.publisher import CallbackPublisher
@@ -12,12 +19,15 @@
     FeedbackCallbackPayload,
     FeedbackMessageItem,
     GenerateFeedbackRequest,
+    PanelBreakdownItem,
     VoiceAnalysisSummary,
 )
 from ai_server.rag.embedder import EmbeddingProvider
 
 log = structlog.get_logger(__name__)
 
+_SELF_INTRO_CATEGORY = "SELF_INTRODUCTION"
+
 
 class FeedbackConsumer:
     """generate.feedback consumer (US-24).
@@ -40,6 +50,7 @@ def __init__(
         core_client: CoreClient,
         embedder: EmbeddingProvider | None = None,
         rag_top_k: int = 5,
+        self_intro_evaluator: SelfIntroEvaluator | None = None,
     ) -> None:
         self._generator = generator
         self._publisher = publisher
@@ -48,6 +59,7 @@ def __init__(
         self._core = core_client
         self._embedder = embedder
         self._rag_top_k = rag_top_k
+        self._self_intro_evaluator = self_intro_evaluator
 
     async def handle(self, message: AbstractIncomingMessage) -> None:
         async with message.process(requeue=False):
@@ -88,17 +100,24 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 req.voice_analysis_summary
             )
 
-            result = await 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,
+            # 종합 피드백 + 자기소개 첫인상 평가를 병렬 실행(첫인상은 종합 점수에 미포함).
+            result, self_intro_item = 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),
             )
+            if self_intro_item is not None:
+                # 종합 집계는 generator 가 첫인상을 모른 채 계산하므로, 여기서 표시용으로만 덧붙인다.
+                result.panel_breakdown.append(self_intro_item)
 
             payload = FeedbackCallbackPayload(
                 session_id=req.session_id,
@@ -129,6 +148,39 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 trace_id=envelope.trace_id,
             )
 
+    async def _evaluate_self_intro(
+        self, req: GenerateFeedbackRequest, voice_analysis_summary: str
+    ) -> PanelBreakdownItem | None:
+        """자기소개 Q/A 를 찾아 첫인상 평가 → 패널 항목 1개. 없거나 실패하면 None(피드백은 계속)."""
+        if self._self_intro_evaluator is None:
+            return None
+        pair = _find_self_intro(req.messages)
+        if pair is None:
+            return None  # 레거시 세션(자기소개 없음) 또는 빈 답변 — 건너뜀
+        question, answer = pair
+        try:
+            ev = await self._self_intro_evaluator.evaluate(
+                job_category=req.job_category,
+                mode=req.mode,
+                self_intro_question=question.content,
+                self_intro_answer=answer.content,
+                voice_analysis_summary=voice_analysis_summary,
+            )
+        except Exception as exc:  # noqa: BLE001
+            log.warning(
+                "feedback.self_intro.failed", error=str(exc), session_id=req.session_id
+            )
+            return None
+        return PanelBreakdownItem(
+            evaluator=SELF_INTRO_EVALUATOR_LABEL,
+            dimension=SELF_INTRO_DIMENSION,
+            score=ev.score,
+            strength=ev.strength,
+            weakness=ev.weakness,
+            detail=ev.detail,
+            score_rationale=ev.score_rationale,
+        )
+
     async def _build_rag_context(self, req: GenerateFeedbackRequest) -> str:
         if not self._embedder or not req.context_document_ids:
             return "(none)"
@@ -158,6 +210,33 @@ async def _build_rag_context(self, req: GenerateFeedbackRequest) -> str:
         )
 
 
+def _find_self_intro(
+    messages: list[FeedbackMessageItem],
+) -> tuple[FeedbackMessageItem, FeedbackMessageItem] | None:
+    """자기소개 질문(category=SELF_INTRODUCTION)과 그 답변 쌍. 없거나 답변이 비면 None."""
+    question = next(
+        (
+            m
+            for m in messages
+            if m.role == "INTERVIEWER" and (m.category or "") == _SELF_INTRO_CATEGORY
+        ),
+        None,
+    )
+    if question is None:
+        return None
+    answer = next(
+        (
+            m
+            for m in messages
+            if m.role == "INTERVIEWEE" and m.parent_message_id == question.id
+        ),
+        None,
+    )
+    if answer is None or not (answer.content or "").strip():
+        return None
+    return question, answer
+
+
 def _build_transcript(messages: list[FeedbackMessageItem]) -> str:
     if not messages:
         return "(empty)"