模型评估与 LLMOps：从离线评测到生产监控的闭环体系

┌──────────┐   ┌──────────┐   ┌──────────┐   ┌──────────┐
│ 1. 评测集  │──▶│ 2. 离线评估 │──▶│ 3. 质量门禁 │──▶│ 4. 生产监控 │
│ Golden    │   │ 全量跑分   │   │ CI 拦截回归 │   │ 漂移告警   │
│ Set 设计  │   │ 版本对比   │   │ 分数≥阈值  │   │ 采样抽检   │
└──────────┘   └──────────┘   └──────────┘   └──────────┘
     ▲                                              │
     └────── 5. 反馈回流：线上坏例沉淀进评测集 ◀───────┘

# golden_set.py — 评测集管理
from dataclasses import dataclass, asdict
import json, random

@dataclass
class EvalCase:
    id: str
    category: str          # 业务场景分类
    query: str
    reference: str         # 参考答案（人工标注）
    difficulty: str        # easy / medium / hard
    tags: list[str] = None # 交叉标签（如 "multilingual", "edge-case"）

class GoldenSet:
    def __init__(self, cases: list[EvalCase]):
        self.cases = cases
        self._validate_distribution()

    def _validate_distribution(self):
        """检查每个分类是否达到最小样本量。"""
        from collections import Counter
        by_cat = Counter(c.category for c in self.cases)
        under = {cat: n for cat, n in by_cat.items() if n < 30}
        if under:
            raise ValueError(f"样本不足的分类: {under}（每类需 ≥30）")

    def stratified_sample(self, ratio: float = 0.3, seed: int = 42):
        """按分类分层抽样，保证 CI 快评估的代表性。"""
        rng = random.Random(seed)
        by_cat = {}
        for c in self.cases:
            by_cat.setdefault(c.category, []).append(c)
        sampled = []
        for cat, cs in by_cat.items():
            sampled += rng.sample(cs, max(1, int(len(cs) * ratio)))
        return GoldenSet(sampled)

    def save(self, path: str):
        json.dump([asdict(c) for c in self.cases],
                  open(path, "w"), ensure_ascii=False, indent=2)

// golden_set.json — 样例格式
{
  "id": "order-tracking-001",
  "category": "物流查询",
  "query": "我的包裹显示签收了但我没收到，怎么办？",
  "reference": "先核对签收人与地址，建议查看门卫/代收点；若确认异常可申请物流核查，提供订单号可查询具体节点。",
  "difficulty": "medium",
  "tags": ["dispute", "proactive"]
}

# metric_bundles.py — 不同任务类型的指标组合
TASK_METRICS = {
    # 摘要任务：忠实 + 信息覆盖 + 简洁
    "summarization": ["faithfulness", "information_coverage", "conciseness"],
    # 开放问答：相关 + 忠实 + 完整
    "qa": ["answer_relevancy", "faithfulness", "completeness"],
    # 分类/抽取：精确匹配 + F1（可确定性）
    "classification": ["exact_match", "f1"],
    # 对话：有用 + 无害 + 上下文一致
    "conversation": ["helpfulness", "harmlessness", "coherence"],
    # 代码生成：通过率 + 编译 + 测试
    "code_gen": ["pass_rate", "compile_ok", "test_pass"],
}

def run_task_eval(task: str, predictions: list[str], references: list[str]):
    """按任务类型选择指标并运行。"""
    metrics = TASK_METRICS[task]
    results = {}
    if task == "classification":
        results["exact_match"] = sum(p == r for p, r in zip(predictions, references)) / len(predictions)
    # 语义类指标委派给 RAGAS / DeepEval
    return results

from ragas import EvaluationDataset, SingleTurnSample, evaluate
from ragas.metrics import (
    faithfulness, answer_relevancy, context_precision, context_recall,
)

def eval_rag_offline(rag_system, golden_cases, judge_llm):
    samples = [
        SingleTurnSample(
            user_input=g["query"],
            response=rag_system.answer(g["query"]),
            retrieved_contexts=rag_system.retrieve(g["query"]),
            reference=g["reference"],
        )
        for g in golden_cases
    ]
    result = evaluate(
        EvaluationDataset(samples=samples),
        metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
        llm=judge_llm,
    )
    df = result.to_pandas()
    return {
        "faithfulness": df["faithfulness"].mean(),
        "answer_relevancy": df["answer_relevancy"].mean(),
        "context_precision": df["context_precision"].mean(),
        "context_recall": df["context_recall"].mean(),
        "samples": len(df),
    }

# judge_reliability.py — Judge 可靠性保障
import statistics

def judge_with_calibration(question, answer, reference,
                           judge_fn, n_runs=3) -> dict:
    """多次评分取中位数，降低单次抖动。"""
    scores = [judge_fn(question, answer, reference) for _ in range(n_runs)]
    median = {k: statistics.median(s[k] for s in scores)
              for k in scores[0].keys()}
    median["raw_runs"] = scores
    return median


def test_judge_distinguishes_good_bad(judge_fn):
    """Judge 自检：必须能区分高质量与低质量回答，否则无区分度。"""
    good = judge_fn("什么是幂等性？",
                    "幂等性指多次执行结果一致，重试不会产生副作用。",
                    "幂等性指重复调用产生相同结果")
    bad = judge_fn("什么是幂等性？",
                   "不知道，可能是关于数据库的东西。",
                   "幂等性指重复调用产生相同结果")
    assert good["total"] > bad["total"], "Judge 无法区分优劣"


def test_judge_self_consistency(judge_fn):
    """Judge 自检：同一对样本重复评分应稳定。"""
    q, a, r = "HTTP 缓存有哪些？", "Cache-Control、ETag……", "Cache-Control 等"
    scores = [judge_fn(q, a, r)["total"] for _ in range(5)]
    assert max(scores) - min(scores) <= 1, f"Judge 抖动过大: {scores}"

JUDGE_SYSTEM = """你是严格的 AI 输出质量评审员。请按 1-5 分评估助手回答：
- relevance 相关度：是否切题
- correctness 正确性：事实是否准确（对照参考）
- completeness 完整度：是否覆盖要点
- faithfulness 忠实度：是否基于给定资料、有无编造
- readability 可读性：表达是否清晰
输出严格 JSON：{"relevance":x,"correctness":x,"completeness":x,
                "faithfulness":x,"readability":x,"reason":"一句话"}
注意：分数要有区分度，避免一律打高分。"""

def structured_judge(question, answer, reference=None, context=None) -> dict:
    import json
    payload = json.dumps({
        "问题": question, "回答": answer,
        "参考": reference or "无", "资料": context or "无",
    }, ensure_ascii=False)
    raw = judge_model(JUDGE_SYSTEM, payload)   # 建议用结构化输出强制 Schema
    result = json.loads(raw)
    result["total"] = sum(result[k] for k in
        ["relevance", "correctness", "completeness",
         "faithfulness", "readability"])
    return result

def cross_validate(question, answer, reference, judge_a, judge_b):
    score_a = judge_a(question, answer, reference)
    score_b = judge_b(question, answer, reference)
    diff = abs(score_a["total"] - score_b["total"])
    if diff >= 5:   # 总分 25，分歧超过 5 分 → 人工复核
        return {"verdict": "REVIEW", "a": score_a, "b": score_b, "diff": diff}
    return {"verdict": "PASS" if max(score_a["total"], score_b["total"]) >= 18
            else "FAIL", "a": score_a, "b": score_b}

# regression_gate.py — 变更前后的对比门禁
def run_regression(new_pipeline, baseline_pipeline,
                   golden: GoldenSet, judge_fn,
                   thresholds: dict = None) -> dict:
    """对比新老 pipeline 在评测集上的分数，报告每个维度的 Δ。"""
    defaults = {
        "avg_semantic_drop": 0.03,   # 平均语义分下降上限
        "keyword_pass_drop": 0.05,   # 关键字通过率下降上限
        "min_score_drop": 0.05,      # 最差样例得分下降上限
    }
    thresholds = thresholds or defaults

    baseline_scores = evaluate_pipeline(baseline_pipeline, golden, judge_fn)
    new_scores = evaluate_pipeline(new_pipeline, golden, judge_fn)

    # 逐维度计算 Δ，并识别哪些样例显著退步
    regression_cases = []
    for c in golden.cases:
        base = baseline_scores[c.id]
        new = new_scores[c.id]
        if new["total"] - base["total"] <= -4:   # 单项退步超过 4/25
            regression_cases.append({"case": c.id, "query": c.query,
                                     "drop": base["total"] - new["total"]})

    verdict = "PASS"
    if baseline_scores["avg"] - new_scores["avg"] > thresholds["avg_semantic_drop"]:
        verdict = "FAIL"
    if len(regression_cases) > len(golden.cases) * 0.05:
        verdict = "FAIL"

    return {"verdict": verdict, "baseline": baseline_scores,
            "new": new_scores, "regression_cases": regression_cases[:20]}

# .github/workflows/llm-eval-gate.yml
name: LLM Evaluation Gate
on:
  pull_request:
    paths:
      - "prompts/**"
      - "golden_set/**"
      - "src/**"

concurrency:
  group: llm-eval-${{ github.ref }}
  cancel-in-progress: true

env:
  OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
  EVAL_BUDGET: "2.0"   # 每次 CI 评估预算上限（美元）

jobs:
  offline-eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install ragas deepeval openai
      - name: 运行回归评估
        run: |
          python scripts/run_regression.py \
            --golden golden_set/qa.json \
            --budget ${{ env.EVAL_BUDGET }}
      - name: 上传评估报告
        uses: actions/upload-artifact@v4
        with:
          name: eval-report
          path: reports/**/*.json
      - name: 门禁失败则标记 PR
        if: failure()
        run: echo "评估未通过，请检查回归报告"

def budget_aware_sample(golden: GoldenSet, budget_usd: float,
                        cost_per_case: float = 0.01) -> GoldenSet:
    """根据预算决定评估量。优先保核心分类，边缘分类可抽样。"""
    full_cost = len(golden.cases) * cost_per_case
    if full_cost <= budget_usd:
        return golden
    ratio = budget_usd / full_cost
    return golden.stratified_sample(ratio=ratio, seed=42)

# shadow_monitor.py — 生产请求的采样质量监控
import random, time

class QualityMonitor:
    def __init__(self, sample_rate: float = 0.01, judge_fn=None):
        self.sample_rate = sample_rate
        self.judge_fn = judge_fn or structured_judge
        self.buffer = []           # 攒批评估，降低 Judge 调用
        self.BATCH_SIZE = 20

    def observe(self, query, answer, context=None):
        """以固定概率采样一条生产请求入评估队列。"""
        if random.random() < self.sample_rate:
            self.buffer.append((query, answer, context, time.time()))
            if len(self.buffer) >= self.BATCH_SIZE:
                self.flush()

    def flush(self):
        """批量评分并推送指标。"""
        batch = self.buffer
        self.buffer = []
        for query, answer, context, ts in batch:
            score = self.judge_fn(query, answer, reference=None, context=context)
            push_metric("llm_quality.total", score["total"], ts)
            push_metric("llm_quality.relevance", score["relevance"], ts)

    def daily_report(self) -> dict:
        """与基线对比，报告当日漂移。"""
        today = query_daily_avg("llm_quality.total")
        baseline = load_baseline("llm_quality_baseline.json")
        drift = today - baseline["avg_total"]
        alert = drift < -3.0    # 总分 25，日漂移超 -3 分告警
        if alert:
            notify_slack(f"质量漂移告警：{drift:+.1f} 分 vs 基线")
        return {"today": today, "baseline": baseline, "drift": drift, "alert": alert}

def feedback_sink(query, answer, thumbs_down: bool, conversation):
    """收集负反馈：沉淀进评测集，形成回归防线。"""
    if thumbs_down:
        # 入"回归快照"，防止相同失败复发
        golden.add(EvalCase(
            id=f"feedback-{int(time.time())}",
            category="user_feedback",
            query=query, reference="", difficulty="hard",
            tags=["regression"]))
        # 触发人工复核
        enqueue_human_review(query, answer, conversation)

# cost_tracking.py — 按用户/功能/模型维度记账
class CostTracker:
    def __init__(self, redis_client):
        self.r = redis_client
        self.HASH = "llm_cost"

    def record(self, user_id, feature, model, prompt_tokens, completion_tokens,
               cached_tokens=0):
        """以 token 为基本单位记账，按需换算金额。"""
        incr(self.r, f"{self.HASH}:{feature}:total_tokens",
             prompt_tokens + completion_tokens)
        incr(self.r, f"{self.HASH}:{user_id}:total_tokens",
             prompt_tokens + completion_tokens)
        # 缓存命中单独统计：缓存省的钱可量化
        incr(self.r, f"{self.HASH}:cached_tokens", cached_tokens)

    def daily_cost_report(self) -> dict:
        """汇总按功能维度排序的每日消耗。"""
        features = scan_keys(f"{self.HASH}:*:total_tokens")
        return {f.split(":")[1]: get(self.r, f) for f in features}

def validate_cost_optimization(optimized_pipeline, baseline_pipeline,
                               golden, judge_fn,
                               max_quality_drop: float = 0.03) -> bool:
    """成本优化方案必须通过质量对比才能上线。"""
    cost_before = estimate_cost(baseline_pipeline, golden)
    cost_after = estimate_cost(optimized_pipeline, golden)
    quality_before = avg_total(baseline_pipeline, golden, judge_fn)
    quality_after = avg_total(optimized_pipeline, golden, judge_fn)

    saving = (cost_before - cost_after) / cost_before
    quality_drop = quality_before - quality_after

    print(f"节省 {saving:.0%}，质量下降 {quality_drop:.2f}")
    assert quality_drop <= max_quality_drop, "成本优化牺牲了质量，拒绝上线"
    return True

def route_model(query: str, complexity_classifier) -> str:
    """按查询复杂度路由模型：简单→小模型，复杂→大模型。"""
    label = complexity_classifier(query)   # easy / medium / hard
    return {
        "easy": "gpt-4o-mini",
        "medium": "gpt-4o-mini",
        "hard": "gpt-4o",
    }[label]

# 路由本身也要监控：评估"路由错误"（简单题误送大模型浪费钱 / 难题误送小模型答错）
def test_router_quality(golden, router, judge_fn):
    routed = [(g, router(g["query"])) for g in golden.cases]
    misroutes = [g for g, model in routed
                 if g["difficulty"] == "easy" and model == "gpt-4o"]
    assert len(misroutes) / len(routed) < 0.10, "简单题过多路由到大模型"

from langsmith import Client

client = Client()

# 创建评测集并灌入 Golden Set
dataset = client.create_dataset(
    dataset_name="insurance-qa-v3",
    description="保险问答回归评测集 v3")

client.create_examples(
    dataset_name="insurance-qa-v3",
    inputs=[{"query": g["query"]} for g in golden.cases],
    outputs=[{"reference": g["reference"]} for g in golden.cases],
)

# 对候选版本跑评估
from langsmith.evaluation import evaluate
results = evaluate(
    lambda inputs: candidate_pipeline.answer(inputs["query"]),
    data="insurance-qa-v3",
    evaluators=[judge_correctness, embedding_distance],
)
for r in results:
    print(r.evaluation_results)

import pytest
from deepeval import assert_test
from deepeval.metrics import AnswerRelevancyMetric, FaithfulnessMetric
from deepeval.test_case import LLMTestCase

@pytest.mark.parametrize("case", load_golden_cases())
def test_llm_output_quality(case):
    test_case = LLMTestCase(
        input=case["query"],
        actual_output=pipeline.answer(case["query"]),
        expected_output=case["reference"],
    )
    assert_test(test_case, [
        AnswerRelevancyMetric(threshold=0.75),
        FaithfulnessMetric(threshold=0.85),
    ])

阶段一（1-2 周）        阶段二（2-4 周）           阶段三（1-2 月）
──────────────────  ────────────────────────   ────────────────────────
· 收集 100+ 真实查询   · 引入 RAGAS / DeepEval    · 全自动 CI 回归门禁
· 人工标注参考答案     · 建立语义 + 事实指标体系     · 生产采样监控 + 漂移告警
· 确定基础指标基线     · LLM-as-a-Judge 校准       · 成本追踪与路由优化
   └ 能度量           └ 能对比                    └ 能拦截、能止损