feat(eval): Chat Eval Pipeline Phase 1-4 + fix

apps/chat_worker/application/services/eval/calibration_monitor.py

@@ -225,6 +225,23 @@ async def check_coverage(self) -> dict[str, Any]:
 
         return result
 
+    async def recalibrate(self) -> dict[str, Any]:
+        """재교정 실행 (stub).
+
+        HITL 인프라 구축 전까지 경고 로그만 남기고
+        stub 결과를 반환합니다.
+
+        실제 구현은 Phase 4+에서:
+        - Calibration Set 재채점
+        - Baseline 갱신
+        - Version bump
+
+        Returns:
+            {"status": "RECALIBRATING", "action": "stub_logged"}
+        """
+        logger.warning("recalibrate() called — stub only, HITL infra not yet available")
+        return {"status": STATUS_RECALIBRATING, "action": "stub_logged"}
+
     @staticmethod
     def _compute_cusum(scores: list[float]) -> tuple[float, float]:
         """CUSUM 양방향 누적합 산출.

apps/chat_worker/infrastructure/assets/data/calibration_set.json

@@ -0,0 +1,119 @@
+{
+  "version": "v1.0-2026-02-10",
+  "description": "Initial calibration set for Chat Eval Pipeline (intent-stratified, manually curated)",
+  "min_kappa": 0.6,
+  "samples": [
+    {
+      "query": "페트병 분리배출 방법 알려줘",
+      "intent": "waste",
+      "context": "페트병은 내용물을 비우고 라벨을 제거한 후 압착하여 투명 페트병 전용 수거함에 배출합니다.",
+      "reference_answer": "페트병 분리배출 방법을 안내드릴게요.\n\n1. 내용물을 깨끗이 비워주세요\n2. 라벨(비닐)을 제거해 주세요\n3. 뚜껑을 분리해 주세요\n4. 납작하게 압착한 후 투명 페트 전용 수거함에 배출해 주세요\n\n주의: 유색 페트병은 플라스틱류로 별도 배출합니다.",
+      "human_scores": {
+        "faithfulness": 5,
+        "relevance": 5,
+        "completeness": 5,
+        "safety": 5,
+        "communication": 4
+      },
+      "annotator_agreement": 0.85
+    },
+    {
+      "query": "형광등 어떻게 버려?",
+      "intent": "waste",
+      "context": "형광등은 유해폐기물로 분류되며, 깨지지 않도록 전용 수거함에 배출합니다.",
+      "reference_answer": "형광등은 유해폐기물이에요.\n\n배출 방법:\n- 깨지지 않도록 조심히 전용 수거함에 넣어주세요\n- 아파트: 단지 내 폐형광등 수거함\n- 단독주택: 동주민센터 또는 폐형광등 수거함\n\n주의: 깨진 형광등은 종량제 봉투에 넣고 '유해폐기물' 표시 후 배출합니다.",
+      "human_scores": {
+        "faithfulness": 5,
+        "relevance": 5,
+        "completeness": 4,
+        "safety": 5,
+        "communication": 5
+      },
+      "annotator_agreement": 0.78
+    },
+    {
+      "query": "우리 동네 대형폐기물 신청하려면 어떻게 해?",
+      "intent": "bulk_waste",
+      "context": "대형폐기물은 관할 구청 또는 온라인으로 배출 신청 후 수수료를 납부하고 지정 장소에 배출합니다.",
+      "reference_answer": "대형폐기물 배출 신청 방법이에요.\n\n1. 관할 구청 홈페이지 또는 앱에서 신청\n2. 품목과 크기를 선택하면 수수료가 안내됩니다\n3. 수수료 납부 후 스티커를 부착하세요\n4. 지정된 날짜에 문 앞 또는 지정 장소에 배출하세요\n\n수수료는 품목과 크기에 따라 1,000원~10,000원 정도입니다.",
+      "human_scores": {
+        "faithfulness": 4,
+        "relevance": 5,
+        "completeness": 4,
+        "safety": 5,
+        "communication": 5
+      },
+      "annotator_agreement": 0.72
+    },
+    {
+      "query": "근처에 재활용 수거함이 어디 있어?",
+      "intent": "collection_point",
+      "context": "재활용 수거함은 아파트 단지, 동주민센터, 대형마트 등에 설치되어 있습니다.",
+      "reference_answer": "재활용 수거함 위치를 찾는 방법이에요.\n\n일반적으로 다음 장소에 있어요:\n- 아파트 단지 내 분리수거장\n- 동주민센터\n- 대형마트 입구\n\n정확한 위치를 안내해 드리려면 현재 위치(동 이름)를 알려주시겠어요?",
+      "human_scores": {
+        "faithfulness": 4,
+        "relevance": 4,
+        "completeness": 3,
+        "safety": 5,
+        "communication": 5
+      },
+      "annotator_agreement": 0.68
+    },
+    {
+      "query": "안녕하세요!",
+      "intent": "general",
+      "context": "",
+      "reference_answer": "안녕하세요! 분리배출 도우미 에코에코입니다 🌿\n\n분리배출 방법이나 재활용 관련 궁금한 점이 있으시면 물어봐 주세요!",
+      "human_scores": {
+        "faithfulness": 5,
+        "relevance": 5,
+        "completeness": 5,
+        "safety": 5,
+        "communication": 5
+      },
+      "annotator_agreement": 0.92
+    },
+    {
+      "query": "비닐은 어떻게 분리배출 하나요?",
+      "intent": "waste",
+      "context": "비닐류는 이물질을 제거하고 투명/불투명 구분 없이 비닐류 전용 수거함에 배출합니다.",
+      "reference_answer": "비닐 분리배출 방법을 알려드릴게요.\n\n1. 내용물을 비우고 이물질을 제거해 주세요\n2. 물로 간단히 헹궈주세요\n3. 비닐류 전용 수거함에 배출하세요\n\n참고:\n- 과자봉지, 라면봉지, 택배 비닐 등 모두 비닐류\n- 오염이 심한 비닐은 종량제 봉투에 배출합니다",
+      "human_scores": {
+        "faithfulness": 5,
+        "relevance": 5,
+        "completeness": 4,
+        "safety": 5,
+        "communication": 5
+      },
+      "annotator_agreement": 0.80
+    },
+    {
+      "query": "오늘 날씨가 어때요?",
+      "intent": "weather",
+      "context": "",
+      "reference_answer": "현재 날씨 정보를 확인해 볼게요.\n\n날씨에 따라 분리배출 팁도 달라지니 참고해 주세요:\n- 비 오는 날: 종이류는 젖지 않게 보관 후 맑은 날 배출\n- 겨울철: 음식물 쓰레기가 얼 수 있으니 배출 시간을 확인하세요",
+      "human_scores": {
+        "faithfulness": 4,
+        "relevance": 4,
+        "completeness": 3,
+        "safety": 5,
+        "communication": 4
+      },
+      "annotator_agreement": 0.65
+    },
+    {
+      "query": "캐릭터 보여줘",
+      "intent": "character",
+      "context": "",
+      "reference_answer": "에코에코 캐릭터를 소개할게요!\n\n저는 분리배출을 도와주는 AI 도우미 에코에코에요. 재활용과 환경 보호에 관심이 많답니다. 분리배출 방법이 궁금하시면 언제든 물어봐 주세요!",
+      "human_scores": {
+        "faithfulness": 5,
+        "relevance": 5,
+        "completeness": 4,
+        "safety": 5,
+        "communication": 5
+      },
+      "annotator_agreement": 0.75
+    }
+  ]
+}

apps/chat_worker/infrastructure/orchestration/langgraph/factory.py

@@ -237,6 +237,7 @@ def create_chat_graph(
     llm_grader: "LLMGraderService | None" = None,
     score_aggregator: "ScoreAggregatorService | None" = None,
     calibration_monitor: "CalibrationMonitorService | None" = None,
+    eval_counter: Any | None = None,
 ) -> StateGraph:
     """Chat 파이프라인 그래프 생성.
 
@@ -663,6 +664,7 @@ async def general_node(state: dict[str, Any]) -> dict[str, Any]:
             llm_grader=llm_grader,
             score_aggregator=score_aggregator,
             calibration_monitor=calibration_monitor,
+            eval_counter=eval_counter,
         )
         graph.add_node("eval", eval_subgraph)
         graph.add_edge("answer", "eval")

apps/chat_worker/infrastructure/persistence/__init__.py

@@ -0,0 +1 @@
+"""Persistence Adapters."""

apps/chat_worker/infrastructure/persistence/eval/__init__.py

@@ -0,0 +1,29 @@
+"""Eval Persistence Adapters.
+
+Gateway 구현체: Redis (L2 Hot Storage), PostgreSQL (L3 Cold Storage).
+Composite Gateway: Redis-first, PG fallback.
+
+See: docs/plans/chat-eval-pipeline-plan.md §5.1
+"""
+
+from chat_worker.infrastructure.persistence.eval.composite_eval_gateway import (
+    CompositeEvalCommandGateway,
+    CompositeEvalQueryGateway,
+)
+from chat_worker.infrastructure.persistence.eval.json_calibration_adapter import (
+    JsonCalibrationDataAdapter,
+)
+from chat_worker.infrastructure.persistence.eval.redis_eval_counter import (
+    RedisEvalCounter,
+)
+from chat_worker.infrastructure.persistence.eval.redis_eval_result_adapter import (
+    RedisEvalResultAdapter,
+)
+
+__all__ = [
+    "CompositeEvalCommandGateway",
+    "CompositeEvalQueryGateway",
+    "JsonCalibrationDataAdapter",
+    "RedisEvalCounter",
+    "RedisEvalResultAdapter",
+]

apps/chat_worker/infrastructure/persistence/eval/composite_eval_gateway.py

@@ -0,0 +1,177 @@
+"""Composite Eval Gateway - Redis + PostgreSQL Layered Memory.
+
+Redis (L2 Hot) + PostgreSQL (L3 Cold) 동시 저장/조회.
+PG pool=None → Redis-only 모드 (로컬 개발).
+
+CQS(Command-Query Separation) 패턴:
+- CompositeEvalCommandGateway: save_result, save_drift_log
+- CompositeEvalQueryGateway: get_recent_scores, get_daily_cost, get_intent_distribution
+
+See: docs/plans/chat-eval-pipeline-plan.md §5.1
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from chat_worker.application.dto.eval_result import EvalResult
+    from chat_worker.infrastructure.persistence.eval.postgres_eval_result_adapter import (
+        PostgresEvalResultAdapter,
+    )
+    from chat_worker.infrastructure.persistence.eval.redis_eval_result_adapter import (
+        RedisEvalResultAdapter,
+    )
+
+logger = logging.getLogger(__name__)
+
+
+class CompositeEvalCommandGateway:
+    """Eval 결과 저장 Gateway (Redis + PG 동시 저장).
+
+    EvalResultCommandGateway Protocol 구현.
+    PG 실패 시 non-blocking (Redis는 항상 저장).
+    pg_adapter=None → Redis-only 모드.
+    """
+
+    def __init__(
+        self,
+        redis_adapter: "RedisEvalResultAdapter",
+        pg_adapter: "PostgresEvalResultAdapter | None" = None,
+    ) -> None:
+        self._redis = redis_adapter
+        self._pg = pg_adapter
+
+    async def save_result(self, eval_result: "EvalResult") -> None:
+        """평가 결과 저장 (Redis + PG).
+
+        Redis: 축별 점수 + 일일 비용 (hot path)
+        PG: 전체 결과 INSERT (cold storage, non-blocking)
+
+        Args:
+            eval_result: 평가 결과 DTO
+        """
+        result_dict = eval_result.to_dict()
+
+        # Redis: 축별 점수 캐시
+        axis_scores = result_dict.get("axis_scores", {})
+        if axis_scores:
+            flat_scores = {}
+            for axis, score_data in axis_scores.items():
+                if isinstance(score_data, dict) and "score" in score_data:
+                    flat_scores[axis] = float(score_data["score"])
+            if flat_scores:
+                await self._redis.push_axis_scores(flat_scores)
+
+        # Redis: 일일 비용 누적
+        cost = result_dict.get("eval_cost_usd")
+        if cost is not None and cost > 0:
+            await self._redis.increment_daily_cost(cost)
+
+        # PG: 전체 결과 저장 (non-blocking)
+        if self._pg is not None:
+            try:
+                await self._pg.save_result(result_dict)
+            except Exception as e:
+                logger.warning(
+                    "PG save_result failed (non-blocking): %s",
+                    e,
+                    exc_info=True,
+                )
+
+    async def save_drift_log(self, drift_entry: dict[str, Any]) -> None:
+        """Calibration Drift 로그 저장 (PG only).
+
+        Args:
+            drift_entry: drift 탐지 정보
+        """
+        if self._pg is not None:
+            try:
+                await self._pg.save_drift_log(drift_entry)
+            except Exception as e:
+                logger.warning(
+                    "PG save_drift_log failed (non-blocking): %s",
+                    e,
+                    exc_info=True,
+                )
+        else:
+            logger.debug(
+                "save_drift_log skipped (PG not configured)",
+                extra={"drift_entry": drift_entry},
+            )
+
+
+class CompositeEvalQueryGateway:
+    """Eval 결과 조회 Gateway (Redis-first, PG fallback).
+
+    EvalResultQueryGateway Protocol 구현.
+    Redis에서 먼저 조회, miss 시 PG fallback.
+    pg_adapter=None → Redis-only 모드.
+    """
+
+    def __init__(
+        self,
+        redis_adapter: "RedisEvalResultAdapter",
+        pg_adapter: "PostgresEvalResultAdapter | None" = None,
+    ) -> None:
+        self._redis = redis_adapter
+        self._pg = pg_adapter
+
+    async def get_recent_scores(self, axis: str, n: int = 10) -> list[float]:
+        """특정 축의 최근 N개 점수 조회.
+
+        Redis-first, PG fallback.
+
+        Args:
+            axis: 평가 축 이름
+            n: 조회할 개수
+
+        Returns:
+            최근 N개 점수 리스트
+        """
+        scores = await self._redis.get_recent_scores(axis, n)
+        if scores:
+            return scores
+
+        # PG fallback
+        if self._pg is not None:
+            try:
+                return await self._pg.get_recent_scores(axis, n)
+            except Exception as e:
+                logger.warning("PG get_recent_scores fallback failed: %s", e)
+
+        return []
+
+    async def get_daily_cost(self) -> float:
+        """당일 평가 비용 합계 (Redis only).
+
+        Returns:
+            당일 누적 평가 비용 (USD)
+        """
+        return await self._redis.get_daily_cost()
+
+    async def get_intent_distribution(self, days: int = 7) -> dict[str, float]:
+        """최근 N일간 Intent별 트래픽 비율.
+
+        PG only (Redis는 집계 데이터 미보관).
+
+        Args:
+            days: 조회 기간
+
+        Returns:
+            intent -> 비율 매핑
+        """
+        if self._pg is not None:
+            try:
+                return await self._pg.get_intent_distribution(days)
+            except Exception as e:
+                logger.warning("PG get_intent_distribution failed: %s", e)
+
+        return {}
+
+
+__all__ = [
+    "CompositeEvalCommandGateway",
+    "CompositeEvalQueryGateway",
+]