LLM as a Judge 완전정리 6편 — 프로덕션 파이프라인: 샘플링·CI 게이트·캘리브레이션 주기

 

평가 파이프라인은 두 가지 잘못된 방향으로 망가집니다. 첫 번째는 "아무것도 안 하기" — 프로덕션에서 품질을 측정하지 않습니다. 두 번째는 "모든 것을 평가하기" — 모든 스팬에 판사 API를 붙여서 비용이 프로덕션 추론과 같아지고, 사용자 응답 지연에 평가 시간이 더해집니다. 2026년 기준으로 대부분의 팀은 레벨 0~1(측정 없음 또는 오프라인 eval만)에 머물고 있습니다. 레벨 2(비동기 프로덕션 모니터링)까지 가는 데 1주일이면 충분하고, 레벨 3(CI 게이트 + 캘리브레이션 주기)까지는 한 달입니다. 이 6편에서 그 구체적인 방법을 전부 코드로 정리합니다.

---

6편이 다루는 것 → 평가 성숙도 5단계 모델 — 현재 어디에 있는지 진단 → 결정론적 샘플링 — 해시 기반 5% 샘플링, 재현 가능성 보장 → 비동기 평가 아키텍처 — 사용자 지연 없이 백그라운드 평가 → OpenTelemetry 스팬 어노테이션 — 판사 결과를 트레이스에 첨부 → CI 게이트 설계 — GitHub Actions + 임계값 기반 블록 → 판사 계약(Judge Contract) — 버전 고정의 4요소 → 60~90일 캘리브레이션 주기 — 표류 탐지 + 자동화 → 프로덕션 인시던트 → 평가 데이터셋 플라이휠

---

1. 평가 성숙도 5단계

지금 어느 단계에 있는지부터 진단합니다.

Level 0: 평가 없음
  - 출시 후 사용자 불만이 유일한 품질 신호
  - HTTP 200 OK가 품질을 보장한다고 믿음

Level 1: 오프라인 eval만
  - 배포 전 고정 데이터셋으로 한 번 측정
  - 프로덕션 트래픽 품질 모니터링 없음
  - 모델 업데이트나 프롬프트 변경 시 회귀 탐지 불가

Level 2: 비동기 프로덕션 모니터링
  - 프로덕션 트래픽 5% 샘플링 + 백그라운드 판사 평가
  - 품질 대시보드 + 임계값 알림
  - 사용자 지연 없음

Level 3: CI 게이트 + 캘리브레이션
  - PR 머지 전 자동 eval 실행
  - 판사 계약 버전 관리
  - 60~90일 황금 세트 재캘리브레이션

Level 4: 런타임 평가 임베딩
  - 판사 점수가 에이전트 행동에 영향
  - 신뢰도 임계값이 툴 접근을 제어
  - 에이전트-as-a-Judge가 크리티컬 결정 경로에서 실행

2026년 현재 대부분의 팀: Level 0~1
현실적 목표: Level 3
최전선: Level 4

---

2. 결정론적 샘플링 — 재현 가능한 5% 트래픽 평가

모든 스팬을 평가하면 비용이 프로덕션과 같아집니다. 판사 비용을 1~5%로 유지하면서 통계적으로 의미 있는 샘플을 얻는 것이 목표입니다.

왜 결정론적이어야 하는가: 무작위 샘플링은 재실행할 때마다 다른 스팬을 평가합니다. 동일한 스팬 ID는 항상 같은 평가 결과를 내야 비교와 재현이 가능합니다.

import hashlib
from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime

class SamplingStrategy(Enum):
    DETERMINISTIC = "deterministic"  # 해시 기반, 재현 가능
    PRIORITY      = "priority"       # 중요도 기반 우선 평가
    RESERVOIR     = "reservoir"      # 저수지 샘플링 (희귀 케이스 보존)

@dataclass
class SamplingConfig:
    base_rate: float = 0.05          # 기본 5% 샘플링
    error_rate: float = 1.00         # 에러는 100% 평가
    low_confidence_rate: float = 0.50 # 낮은 신뢰도 응답 50% 평가
    high_value_rate: float = 0.20    # 고가치 사용자 세션 20% 평가
    min_daily_samples: int = 100     # 하루 최소 평가 수


class DeterministicSampler:
    """
    스팬 ID 해시 기반 결정론적 샘플링.
    같은 스팬은 항상 같은 결정을 내림.
    """

    def __init__(self, config: SamplingConfig = None):
        self.config = config or SamplingConfig()

    def should_evaluate(
        self,
        span_id: str,
        context: dict | None = None,
    ) -> tuple[bool, str]:
        """
        Returns: (평가 여부, 선택 이유)
        """
        context = context or {}

        # 1. 에러 스팬은 항상 평가
        if context.get("has_error") or context.get("status_code", 200) >= 400:
            return True, "error_span"

        # 2. 낮은 신뢰도 (모델 자체 confidence 또는 사용자 피드백)
        if context.get("confidence", 1.0) < 0.5:
            hash_val = int(hashlib.md5(f"lc:{span_id}".encode()).hexdigest(), 16)
            if (hash_val % 100) < (self.config.low_confidence_rate * 100):
                return True, "low_confidence"

        # 3. 고가치 사용자 세션
        if context.get("user_tier") == "premium":
            hash_val = int(hashlib.md5(f"hv:{span_id}".encode()).hexdigest(), 16)
            if (hash_val % 100) < (self.config.high_value_rate * 100):
                return True, "high_value_user"

        # 4. 기본 결정론적 샘플링
        # MD5 해시의 첫 32비트를 100으로 나눈 나머지
        hash_val = int(hashlib.md5(span_id.encode()).hexdigest()[:8], 16)
        if (hash_val % 100) < (self.config.base_rate * 100):
            return True, "base_sample"

        return False, "not_sampled"

    def estimate_daily_cost(
        self,
        daily_requests: int,
        cost_per_eval: float = 0.05,  # $0.05 per judge call
    ) -> dict:
        """일일 예상 평가 비용 계산"""
        expected_samples = max(
            self.config.min_daily_samples,
            int(daily_requests * self.config.base_rate)
        )
        return {
            "expected_daily_samples": expected_samples,
            "estimated_daily_cost_usd": round(expected_samples * cost_per_eval, 2),
            "cost_as_pct_of_production": round(
                (expected_samples * cost_per_eval) /
                (daily_requests * 0.01) * 100, 1  # 가정: 프로덕션 $0.01/req
            ),
        }

---

3. 비동기 평가 아키텍처 — 사용자 지연 없이

판사 평가를 사용자 응답 경로에 넣으면 안 됩니다. 판사 API 호출이 200~500ms를 추가합니다.

import asyncio
import json
from anthropic import Anthropic
from datetime import datetime

client = Anthropic()

class AsyncJudgeEvaluator:
    """
    비동기 판사 평가기.
    사용자 응답 반환 후 백그라운드에서 실행.
    """

    def __init__(
        self,
        judge_contract: dict,   # {model_id, rubric, rubric_version, prompt_hash}
        sampler: DeterministicSampler,
        sink,                   # 평가 결과를 저장하는 저장소 (DB, 큐 등)
    ):
        self.contract = judge_contract
        self.sampler = sampler
        self.sink = sink
        self._semaphore = asyncio.Semaphore(10)  # 동시 판사 호출 최대 10개

    async def evaluate_span(self, span: dict) -> dict | None:
        """
        단일 스팬 비동기 평가.
        샘플링 결정 → 판사 호출 → 결과 저장.
        """
        should_eval, reason = self.sampler.should_evaluate(
            span["span_id"],
            context={
                "has_error": span.get("error") is not None,
                "confidence": span.get("model_confidence", 1.0),
                "user_tier": span.get("user_tier", "free"),
            }
        )

        if not should_eval:
            return None

        async with self._semaphore:
            try:
                # 실제 판사 호출은 스레드에서 실행 (blocking API)
                result = await asyncio.to_thread(
                    self._run_judge_sync, span
                )

                # 결과를 원본 스팬에 어노테이션으로 첨부
                annotated = {
                    **span,
                    "judge_evaluation": {
                        "score": result["overall"],
                        "dimensions": result.get("dimension_scores", {}),
                        "judge_model": self.contract["model_id"],
                        "judge_contract_fingerprint": self._fingerprint(),
                        "sampling_reason": reason,
                        "evaluated_at": datetime.utcnow().isoformat(),
                        "latency_ms": result.get("latency_ms"),
                    }
                }

                # 비동기로 저장소에 기록
                await self.sink.write(annotated)
                return annotated

            except Exception as e:
                # 평가 실패는 프로덕션 흐름에 영향 없음
                await self.sink.write_error(span["span_id"], str(e))
                return None

    def _run_judge_sync(self, span: dict) -> dict:
        """동기 판사 호출 (스레드에서 실행)"""
        start = datetime.utcnow()

        JUDGE_PROMPT = f"""
{self.contract['rubric']}

[질문]
{{question}}

[응답]
{{response}}

JSON으로만 출력:
{{"accuracy": 1-5, "relevance": 1-5, "completeness": 1-5,
  "overall": <가중 평균>, "summary": "2문장 요약"}}
"""
        raw = client.messages.create(
            model=self.contract["model_id"],
            max_tokens=256,
            messages=[{
                "role": "user",
                "content": JUDGE_PROMPT.format(
                    question=span.get("input", ""),
                    response=span.get("output", ""),
                )
            }]
        ).content[0].text.strip()

        result = json.loads(raw)
        result["latency_ms"] = int(
            (datetime.utcnow() - start).total_seconds() * 1000
        )
        return result

    def _fingerprint(self) -> str:
        return (
            f"{self.contract['model_id']}:"
            f"{self.contract['rubric_version']}:"
            f"{self.contract['prompt_hash']}"
        )

    async def evaluate_batch(self, spans: list[dict]) -> list[dict]:
        """배치 스팬 병렬 평가"""
        tasks = [self.evaluate_span(span) for span in spans]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return [r for r in results if r is not None and not isinstance(r, Exception)]

실제 통합 패턴 (FastAPI 예시)

from fastapi import FastAPI, BackgroundTasks
import asyncio

app = FastAPI()
evaluator = AsyncJudgeEvaluator(...)   # 싱글톤

@app.post("/chat")
async def chat_endpoint(
    request: ChatRequest,
    background_tasks: BackgroundTasks,
):
    # 1. 생성 모델 호출 (사용자 경로)
    response = await llm.generate(request.message)

    # 2. 스팬 구성
    span = {
        "span_id": request.request_id,
        "input": request.message,
        "output": response.text,
        "user_tier": request.user.tier,
        "model": response.model_used,
        "latency_ms": response.latency_ms,
    }

    # 3. 판사 평가는 백그라운드 — 사용자 응답에 영향 없음
    background_tasks.add_task(evaluator.evaluate_span, span)

    # 4. 즉시 응답 반환 (판사 완료 대기 없음)
    return {"message": response.text, "request_id": request.request_id}

---

4. OpenTelemetry 스팬 어노테이션

판사 결과를 분산 추적 시스템(OTel)의 스팬에 붙이면, 품질 점수를 레이턴시·에러율·비용과 함께 단일 뷰로 관찰할 수 있습니다.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

tracer = trace.get_tracer(__name__)

class OTelAnnotatedJudge:
    """판사 결과를 OTel 스팬 속성으로 기록"""

    def __init__(self, base_evaluator: AsyncJudgeEvaluator):
        self.evaluator = base_evaluator

    async def evaluate_and_annotate(self, span_data: dict) -> dict | None:
        """평가 결과를 OTel 스팬의 속성으로 첨부"""

        with tracer.start_as_current_span("llm.judge.evaluate") as otel_span:
            # 판사 평가 실행
            result = await self.evaluator.evaluate_span(span_data)

            if result and "judge_evaluation" in result:
                ev = result["judge_evaluation"]

                # OTel 속성으로 기록 (Langfuse, Arize 등과 호환)
                otel_span.set_attributes({
                    "llm.judge.score.overall":     ev["score"],
                    "llm.judge.score.accuracy":    ev.get("dimensions", {}).get("accuracy", 0),
                    "llm.judge.score.relevance":   ev.get("dimensions", {}).get("relevance", 0),
                    "llm.judge.model":             ev["judge_model"],
                    "llm.judge.contract":          ev["judge_contract_fingerprint"],
                    "llm.judge.sampling_reason":   ev["sampling_reason"],
                    "llm.judge.latency_ms":        ev.get("latency_ms", 0),
                    # 품질 알림 트리거용
                    "llm.judge.below_threshold":   ev["score"] < 3.0,
                })

            return result


# 품질 추이 쿼리 예시 (Langfuse/Arize 대시보드에서)
"""
SELECT
    DATE_TRUNC('hour', evaluated_at) as hour,
    AVG(judge_score_overall) as avg_quality,
    COUNT(*) as n_evaluations,
    SUM(CASE WHEN judge_score_overall < 3.0 THEN 1 ELSE 0 END) as low_quality_count
FROM span_evaluations
WHERE evaluated_at > NOW() - INTERVAL '7 days'
GROUP BY 1
ORDER BY 1
"""

---

5. CI 게이트 — PR 머지 전 자동 평가

프롬프트 변경, 모델 업그레이드, 루브릭 수정은 모두 품질 회귀를 일으킬 수 있습니다. CI 게이트는 이것을 코드처럼 테스트합니다.

# .github/workflows/llm-quality-gate.yml
name: LLM Quality Gate

on:
  pull_request:
    paths:
      - "prompts/**"
      - "src/llm/**"
      - "config/judge*.yaml"

jobs:
  judge-eval:
    runs-on: ubuntu-latest
    timeout-minutes: 15

    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: pip install anthropic deepeval pyyaml scipy sklearn

      - name: Run Golden Set Evaluation
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          python eval/ci_judge.py \
            --golden-set eval/golden_set.jsonl \
            --judge-config config/judge_contract.yaml \
            --output results/ci_eval.json \
            --baseline results/baseline_eval.json

      - name: Check Quality Thresholds
        run: |
          python eval/check_thresholds.py \
            --results results/ci_eval.json \
            --fail-on-violation

      - name: Upload Results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: eval-results
          path: results/

CI 평가 스크립트

# eval/ci_judge.py
import json
import yaml
import argparse
from pathlib import Path
from anthropic import Anthropic
from sklearn.metrics import cohen_kappa_score
import numpy as np

client = Anthropic()

def run_ci_evaluation(
    golden_set_path: str,
    judge_config_path: str,
    output_path: str,
    baseline_path: str | None = None,
) -> dict:
    """CI에서 실행되는 판사 평가"""

    # 판사 계약 로드
    with open(judge_config_path) as f:
        contract = yaml.safe_load(f)

    # 황금 세트 로드
    golden_set = []
    with open(golden_set_path) as f:
        for line in f:
            golden_set.append(json.loads(line.strip()))

    # 평가 실행
    judge_scores = []
    for sample in golden_set:
        raw = client.messages.create(
            model=contract["model_id"],
            max_tokens=128,
            messages=[{
                "role": "user",
                "content": contract["prompt_template"].format(
                    question=sample["question"],
                    response=sample["response"],
                )
            }]
        ).content[0].text.strip()

        try:
            result = json.loads(raw)
            judge_scores.append(round(result["overall"]))
        except Exception:
            judge_scores.append(3)  # 파싱 실패 시 중간값

    human_scores = [s["human_score"] for s in golden_set]

    # 지표 계산
    kappa = cohen_kappa_score(human_scores, judge_scores, weights="quadratic")
    mae = float(np.mean(np.abs(np.array(human_scores) - np.array(judge_scores))))
    exact_match = float(np.mean(np.array(human_scores) == np.array(judge_scores)))

    results = {
        "contract_fingerprint": f"{contract['model_id']}:{contract['rubric_version']}:{contract['prompt_hash']}",
        "n_samples": len(golden_set),
        "weighted_kappa": round(kappa, 4),
        "mae": round(mae, 3),
        "exact_match_rate": round(exact_match, 4),
        "timestamp": __import__("datetime").datetime.utcnow().isoformat(),
    }

    # 베이스라인과 비교
    if baseline_path and Path(baseline_path).exists():
        with open(baseline_path) as f:
            baseline = json.load(f)
        results["kappa_delta"] = round(kappa - baseline["weighted_kappa"], 4)
        results["mae_delta"] = round(mae - baseline["mae"], 3)

    # 결과 저장
    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, "w") as f:
        json.dump(results, f, indent=2)

    return results


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--golden-set")
    parser.add_argument("--judge-config")
    parser.add_argument("--output")
    parser.add_argument("--baseline", default=None)
    args = parser.parse_args()

    run_ci_evaluation(args.golden_set, args.judge_config, args.output, args.baseline)

임계값 검사 스크립트

# eval/check_thresholds.py
import json, sys, argparse

THRESHOLDS = {
    "weighted_kappa":     {"min": 0.60, "critical": 0.40},
    "mae":               {"max": 1.00, "critical": 1.50},
    "exact_match_rate":  {"min": 0.55, "critical": 0.40},
    # 베이스라인 대비 회귀 허용 범위
    "kappa_delta":       {"min": -0.05},  # κ 5% 이상 하락 시 실패
    "mae_delta":         {"max": 0.20},   # MAE 0.2 이상 증가 시 실패
}

def check_thresholds(results_path: str) -> bool:
    with open(results_path) as f:
        results = json.load(f)

    violations = []
    critical_violations = []

    for metric, bounds in THRESHOLDS.items():
        if metric not in results:
            continue

        value = results[metric]

        if "min" in bounds and value < bounds["min"]:
            msg = f"{metric}={value:.4f} < min={bounds['min']}"
            if "critical" in bounds and value < bounds["critical"]:
                critical_violations.append(f"🚨 CRITICAL: {msg}")
            else:
                violations.append(f"⚠️  WARNING: {msg}")

        if "max" in bounds and value > bounds["max"]:
            msg = f"{metric}={value:.4f} > max={bounds['max']}"
            if "critical" in bounds and value > bounds["critical"]:
                critical_violations.append(f"🚨 CRITICAL: {msg}")
            else:
                violations.append(f"⚠️  WARNING: {msg}")

    if critical_violations or violations:
        print("❌ Quality Gate FAILED")
        for v in critical_violations + violations:
            print(f"  {v}")
        if critical_violations:
            sys.exit(1)   # CI 블록
        else:
            print("  (warnings only — merge allowed with review)")
            sys.exit(0)
    else:
        print(f"✅ Quality Gate PASSED — κ={results['weighted_kappa']:.3f}")
        sys.exit(0)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--results")
    parser.add_argument("--fail-on-violation", action="store_true")
    args = parser.parse_args()
    check_thresholds(args.results)

---

6. 판사 계약 — 4요소 버전 고정

캘리브레이션 표류의 근본 원인은 판사 계약이 명시적으로 고정되지 않는 것입니다. 4요소가 모두 고정돼야 점수의 시계열 비교가 유의미합니다.

import hashlib
import yaml
from dataclasses import dataclass, asdict
from datetime import datetime

@dataclass
class JudgeContract:
    """
    판사 계약 — 4요소 불변 고정.

    중요: 이 중 하나라도 변경되면 eval suite 마이그레이션으로 취급.
    이전 점수와 새 점수는 직접 비교할 수 없음.
    """

    # 요소 1: 모델 ID (alias 금지, 날짜 버전 필수)
    # ❌ "claude-sonnet-4-6"           (alias — 언제든 바뀔 수 있음)
    # ✅ "claude-sonnet-4-6-20260519"  (날짜 고정)
    model_id: str

    # 요소 2: 루브릭 버전
    rubric_version: str   # "v2.3"

    # 요소 3: 루브릭 내용 (해시로 변경 추적)
    rubric_content: str

    # 요소 4: 프롬프트 템플릿 (해시로 변경 추적)
    prompt_template: str

    # 메타데이터
    created_at: str = ""
    created_by: str = ""
    change_reason: str = ""

    def __post_init__(self):
        if not self.created_at:
            self.created_at = datetime.utcnow().isoformat()

    @property
    def rubric_hash(self) -> str:
        return hashlib.sha256(self.rubric_content.encode()).hexdigest()[:12]

    @property
    def prompt_hash(self) -> str:
        return hashlib.sha256(self.prompt_template.encode()).hexdigest()[:12]

    @property
    def fingerprint(self) -> str:
        """계약의 고유 식별자 — 이것이 바뀌면 별도 평가 스위트"""
        return f"{self.model_id}:{self.rubric_version}:{self.rubric_hash}:{self.prompt_hash}"

    def to_yaml(self) -> str:
        d = asdict(self)
        d["rubric_hash"] = self.rubric_hash
        d["prompt_hash"] = self.prompt_hash
        d["fingerprint"] = self.fingerprint
        return yaml.dump(d, allow_unicode=True)

    def is_compatible_with(self, other: "JudgeContract") -> bool:
        """두 계약이 점수 비교 가능한가"""
        return self.fingerprint == other.fingerprint

    @classmethod
    def from_yaml(cls, path: str) -> "JudgeContract":
        with open(path) as f:
            data = yaml.safe_load(f)
        return cls(
            model_id=data["model_id"],
            rubric_version=data["rubric_version"],
            rubric_content=data["rubric_content"],
            prompt_template=data["prompt_template"],
            created_at=data.get("created_at", ""),
            created_by=data.get("created_by", ""),
            change_reason=data.get("change_reason", ""),
        )

계약 변경 워크플로

계약 변경이 필요할 때 (모델 업그레이드, 루브릭 개선):

1. 새 계약 파일 생성 (judge_contract_v2.yaml)
2. 황금 세트로 새 계약의 κ 측정
3. 이전 계약 점수와 비교 (직접 비교 불가 — 별도 시계열)
4. 이전 계약 결과를 "history" 아카이브로 이동
5. 새 계약으로 마이그레이션 → 새 베이스라인 설정

절대 하면 안 되는 것:
- judge_contract.yaml을 그냥 덮어쓰기
  (이전 점수와 새 점수가 섞여서 의미 없어짐)
- "gpt-4o-latest" 같은 alias 사용
  (6주마다 다른 모델로 교체되는 alias)

---

7. 60~90일 캘리브레이션 주기

import numpy as np
from sklearn.metrics import cohen_kappa_score
from datetime import datetime, timedelta

class CalibrationScheduler:
    """
    판사 캘리브레이션 주기 관리.
    트리거: 시간 경과 | κ 하락 | 평균 이동 | 모델 업데이트 이벤트
    """

    def __init__(
        self,
        contract: JudgeContract,
        golden_set: list[dict],
        baseline_kappa: float,
        baseline_mean_score: float,
        recalibrate_after_days: int = 60,
    ):
        self.contract = contract
        self.golden_set = golden_set
        self.baseline_kappa = baseline_kappa
        self.baseline_mean = baseline_mean_score
        self.last_calibrated = datetime.utcnow()
        self.calibration_history = []

    def run_calibration(self, judge_fn) -> dict:
        """황금 세트로 현재 판사 상태 측정"""
        judge_scores  = [round(judge_fn(s["question"], s["response"])) for s in self.golden_set]
        human_scores  = [s["human_score"] for s in self.golden_set]

        current_kappa = cohen_kappa_score(human_scores, judge_scores, weights="quadratic")
        current_mean  = float(np.mean(judge_scores))
        kappa_drop    = self.baseline_kappa - current_kappa
        mean_shift    = abs(current_mean - self.baseline_mean)
        days_elapsed  = (datetime.utcnow() - self.last_calibrated).days

        # 트리거 결정
        triggers = []
        if days_elapsed >= 90:
            triggers.append(f"hard_deadline ({days_elapsed}일 경과)")
        elif days_elapsed >= 60:
            triggers.append(f"soft_deadline ({days_elapsed}일 경과)")
        if kappa_drop > 0.10:
            triggers.append(f"kappa_drop ({kappa_drop:.3f})")
        if mean_shift > 3.0:
            triggers.append(f"mean_shift ({mean_shift:.1f}점)")
        if current_kappa < 0.50:
            triggers.append(f"kappa_below_threshold ({current_kappa:.3f})")

        alert_level = (
            "critical" if current_kappa < 0.40 or mean_shift > 5.0
            else "warning" if triggers
            else "ok"
        )

        result = {
            "contract_fingerprint": self.contract.fingerprint,
            "calibration_date": datetime.utcnow().isoformat(),
            "current_kappa": round(current_kappa, 4),
            "baseline_kappa": round(self.baseline_kappa, 4),
            "kappa_drop": round(kappa_drop, 4),
            "current_mean_score": round(current_mean, 3),
            "baseline_mean_score": round(self.baseline_mean, 3),
            "mean_shift": round(mean_shift, 3),
            "days_since_calibration": days_elapsed,
            "recalibration_needed": len(triggers) > 0,
            "triggers": triggers,
            "alert_level": alert_level,
            "action": self._recommend_action(alert_level, triggers),
        }

        self.calibration_history.append(result)
        return result

    def _recommend_action(self, alert_level: str, triggers: list) -> str:
        if alert_level == "critical":
            return "즉시 판사 교체 또는 루브릭 재설계 — 현재 점수 신뢰 불가"
        elif "kappa_drop" in str(triggers):
            return "판사 계약 업데이트 검토 — 새 황금 세트로 마이그레이션"
        elif "mean_shift" in str(triggers):
            return "체계적 편향 확인 — 루브릭 앵커 예시 재검토"
        elif triggers:
            return "캘리브레이션 일정 확정 — 2주 내 실행 권장"
        return "정상 — 다음 정기 캘리브레이션 예약"

---

8. 인시던트 → 데이터셋 플라이휠

프로덕션에서 발생하는 모든 품질 인시던트는 황금 세트의 씨앗입니다. 이 피드백 루프가 평가 시스템을 점점 더 강력하게 만듭니다.

class IncidentToDatasetConverter:
    """
    프로덕션 인시던트를 황금 세트 샘플로 변환.
    모든 인시던트 사후 검토(Postmortem)는 새 eval 케이스를 생성.
    """

    def __init__(self, golden_set_path: str):
        self.golden_set_path = golden_set_path

    def add_incident_case(
        self,
        span_id: str,
        question: str,
        response: str,
        human_score: int,
        incident_type: str,
        description: str,
        annotator: str,
    ) -> dict:
        """인시던트를 황금 세트 케이스로 추가"""
        case = {
            "span_id": span_id,
            "question": question,
            "response": response,
            "human_score": human_score,
            "category": incident_type,
            "difficulty": "hard",    # 인시던트 케이스는 항상 hard
            "source": "production_incident",
            "description": description,
            "annotator": annotator,
            "added_at": datetime.utcnow().isoformat(),
        }

        # 황금 세트에 추가
        with open(self.golden_set_path, "a") as f:
            f.write(json.dumps(case, ensure_ascii=False) + "\n")

        return case

    def from_user_feedback(
        self,
        thumbs_down_spans: list[dict],
        judge_fn,
    ) -> list[dict]:
        """
        사용자 thumbs-down 피드백을 황금 세트 후보로 변환.
        인간 레이블러가 검토 후 승인.
        """
        candidates = []
        for span in thumbs_down_spans:
            # 판사 점수와 사용자 불만이 불일치하는 케이스 추출
            # (판사는 높게 줬는데 사용자는 싫어함 → 판사 맹점)
            judge_score = judge_fn(span["input"], span["output"])
            if judge_score >= 4.0:   # 판사는 좋다고 했는데
                candidates.append({
                    **span,
                    "judge_score": judge_score,
                    "user_signal": "thumbs_down",
                    "type": "judge_blind_spot",
                    "priority": "high",
                })

        return candidates


# 플라이휠 완성 — 데이터 흐름
"""
프로덕션 트래픽
    ↓ (5% 샘플링)
비동기 판사 평가
    ↓
낮은 점수 스팬 플래그
    ↓
인간 검토 (주 1회 배치)
    ↓
황금 세트에 추가
    ↓
CI 게이트 재실행 → 판사 개선 감지
    ↓
캘리브레이션 재실행 → 기준선 업데이트
    ↓ (반복)
"""

---

9. 비용 추적 — 판사 비용을 독립적으로 추적

판사 비용이 프로덕션 비용과 섞이면 모니터링이 어렵습니다.

class JudgeCostTracker:
    """판사 비용 독립 추적"""

    # 2026년 5월 기준 가격 (1M 토큰당 USD)
    MODEL_COSTS = {
        "claude-opus-4-7":           {"input": 5.00, "output": 25.00},
        "claude-sonnet-4-6":         {"input": 3.00, "output": 15.00},
        "claude-haiku-4-5-20251001": {"input": 0.25, "output": 1.25},
        "gpt-5.5":                   {"input": 2.50, "output": 10.00},
        "gemini-3.5-flash":          {"input": 1.50, "output":  9.00},
    }

    def __init__(self):
        self._records = []

    def record(
        self,
        model_id: str,
        input_tokens: int,
        output_tokens: int,
        span_id: str,
        rubric_version: str,
    ):
        costs = self.MODEL_COSTS.get(model_id, {"input": 1.0, "output": 5.0})
        cost_usd = (
            input_tokens * costs["input"] / 1_000_000
            + output_tokens * costs["output"] / 1_000_000
        )
        self._records.append({
            "model_id": model_id,
            "span_id": span_id,
            "rubric_version": rubric_version,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost_usd": cost_usd,
            "timestamp": datetime.utcnow().isoformat(),
        })

    def daily_summary(self) -> dict:
        today_records = [
            r for r in self._records
            if r["timestamp"][:10] == datetime.utcnow().date().isoformat()
        ]
        total = sum(r["cost_usd"] for r in today_records)
        by_model = {}
        for r in today_records:
            by_model[r["model_id"]] = by_model.get(r["model_id"], 0) + r["cost_usd"]

        return {
            "date": datetime.utcnow().date().isoformat(),
            "total_cost_usd": round(total, 4),
            "n_evaluations": len(today_records),
            "cost_per_eval": round(total / len(today_records), 5) if today_records else 0,
            "breakdown_by_model": {k: round(v, 4) for k, v in by_model.items()},
        }

---

✅ 6편 정리 — 프로덕션 파이프라인 체크리스트

배포 전
□ 황금 세트 200개 이상 (easy/medium/hard 분층)
□ 판사 계약 YAML 파일 생성 (모델 ID 날짜 고정)
□ CI 워크플로 추가 (.github/workflows/llm-quality-gate.yml)
□ 베이스라인 κ 측정 후 results/baseline_eval.json 저장

배포 후 (즉시)
□ DeterministicSampler 설정 (base_rate=0.05)
□ AsyncJudgeEvaluator 백그라운드 태스크 연결
□ OTel 스팬 어노테이션 활성화
□ 판사 비용 독립 추적 시작

정기 (월별)
□ CalibrationScheduler.run_calibration() 실행
□ κ 하락 > 0.10 또는 mean shift > 3.0이면 계약 리뷰
□ 인시던트 케이스 황금 세트 추가 (Postmortem 후)
□ 비용 대시보드 검토 (판사 비용이 프로덕션의 5% 초과 시 샘플률 조정)

분기 (90일)
□ 판사 계약 전면 재검토
□ 황금 세트 확장 또는 갱신
□ CI 임계값 재조정 (모델 전반적 품질 향상 반영)

안티패턴 결과

gpt-4o-latest alias 사용	6주마다 다른 모델 측정
황금 세트 없음	표류 탐지 불가
동기 인라인 평가	사용자 지연 +200ms
모든 스팬 평가	비용 = 프로덕션 추론 비용
판사 비용 미추적	비용이 조용히 2~3배 증가
캘리브레이션 주기 없음	3개월 후 점수가 의미 없어짐

 

---

LLM-as-a-Judge 완전정리 시리즈 — 완결

1. ✅ 1편 — 왜 기존 지표는 죽었고, 세 패러다임은 무엇인가 https://cell-devlog.tistory.com/265
2. ✅ 2편 — 판사는 어디서 거짓말하나: 7가지 편향 해부 https://cell-devlog.tistory.com/266
3. ✅ 3편 — 편향 잡는 법: Position Swap부터 Cross-family까지 https://cell-devlog.tistory.com/267
4. ✅ 4편 — G-Eval vs Prometheus 2 vs PAJAMA vs Themis https://cell-devlog.tistory.com/268 
5. ✅ 5편 — 판사를 평가하기: Cohen's κ, Bradley-Terry, 황금 세트 설계 https://cell-devlog.tistory.com/269
6. ✅ 6편 — 프로덕션 파이프라인: 샘플링·CI 게이트·캘리브레이션 주기 https://cell-devlog.tistory.com/270
7. ✅ 7편 — 한계와 대안: LLM 판사가 절대 못 하는 것들 https://cell-devlog.tistory.com/271