评测的第一性问题

from dataclasses import dataclass, field
from typing import Callable, Any
import statistics, json
@dataclass
class Sample:
    id: str
    input: str
    expected: Any = None        # 有标准答案的样本才有
    tags: list[str] = field(default_factory=list)   # 分层标签：冒烟/回归/边界/对抗
@dataclass
class Metric:
    """
    指标必须自带口径说明。没有口径的指标不可能被复现，
    因此在数据结构层面就强制要求写清楚。
    """
    name: str
    numerator_desc: str         # 分子是什么
    denominator_desc: str       # 分母是什么
    boundary: str               # 边界情况怎么算（超时/中断/空结果）
    fn: Callable[[dict], bool]
    def describe(self) -> str:
        return (f"{self.name}\n"
                f"  分子：{self.numerator_desc}\n"
                f"  分母：{self.denominator_desc}\n"
                f"  边界：{self.boundary}")
@dataclass
class EvalReport:
    metrics: dict[str, float]
    n: int
    failures: list[dict]        # 必须保留失败样本，否则无法归因
    by_tag: dict[str, float]
def run_eval(samples: list[Sample], run_once: Callable[[Sample], dict],
             metrics: list[Metric], repeats: int = 1) -> EvalReport:
    """
    repeats > 1 用于估计方差 —— Agent 有随机性，
    只跑一遍得到的差异无法与噪声区分。
    """
    per_metric: dict[str, list[float]] = {m.name: [] for m in metrics}
    failures: list[dict] = []
    tag_scores: dict[str, list[float]] = {}
    for run_i in range(repeats):
        for s in samples:
            try:
                out = run_once(s)
            except Exception as e:
                # 异常也算一次「未通过」，不能让异常样本静默消失，
                # 否则分母被人为缩小，指标会虚高。
                out = {"error": f"{type(e).__name__}: {e}"}
            for m in metrics:
                per_metric[m.name].append(1.0 if m.fn(out) else 0.0)
            # 记录失败样本：这是最有价值的产出，而不是那个百分比
            main = metrics[0]
            if not main.fn(out):
                failures.append({"sample": s.id, "tags": s.tags, "output": out,
                                 "run": run_i})
                for t in s.tags:
                    tag_scores.setdefault(t, []).append(0.0)
            else:
                for t in s.tags:
                    tag_scores.setdefault(t, []).append(1.0)
    report = EvalReport(
        metrics={m.name: statistics.fmean(v) for m, v in
                 ((m, per_metric[m.name]) for m in metrics)},
        n=len(samples) * repeats,
        failures=failures,
        by_tag={t: statistics.fmean(v) for t, v in tag_scores.items()},
    )
    # 顺带输出方差：方差大的指标在决策时要更保守
    for m in metrics:
        v = per_metric[m.name]
        if len(v) > 1:
            report.metrics[f"{m.name}__stdev"] = statistics.pstdev(v)
    return report
# --- 指标定义示例（注意每一条都把口径写全了） ---
m_parse_ok = Metric(
    name="解析成功率",
    numerator_desc="输出能被 JSON 解析器成功解析的样本数",
    denominator_desc="全部参评样本数（含超时、异常、中断）",
    boundary="超时与异常一律计入分母且计为不通过；空输出计为不通过",
    fn=lambda out: "parsed" in out and out["parsed"] is not None,
)
m_tool_ok = Metric(
    name="工具调用成功率",
    numerator_desc="工具调用返回状态为 ok 或 empty 的样本数（empty 视为成功）",
    denominator_desc="全部参评样本数（含超时、异常、中断）",
    boundary="超时/异常计为不通过；empty 计为通过（查询成功但无数据）",
    fn=lambda out: out.get("tool_status") in ("ok", "empty"),
)
def compare(baseline: EvalReport, candidate: EvalReport) -> str:
    """
    对比报告。原则：
      · 差异小于历史波动（stdev）的不下结论
      · 主指标提升但分层指标恶化，必须显式提示（防刷分）
    """
    lines = [f"样本数：baseline {baseline.n} → candidate {candidate.n}"]
    for k, v_new in candidate.metrics.items():
        if k.endswith("__stdev"):
            continue
        v_old = baseline.metrics.get(k)
        if v_old is None:
            continue
        delta = v_new - v_old
        noise = max(candidate.metrics.get(f"{k}__stdev", 0),
                    baseline.metrics.get(f"{k}__stdev", 0))
        verdict = "差异不显著" if abs(delta) <= noise else ("提升" if delta > 0 else "回归")
        lines.append(f"{k}: {v_old:.1%} → {v_new:.1%}（Δ{delta:+.1%}）{verdict}")
    # 防刷分：整体提升但某个分层恶化
    for tag, v_new in candidate.by_tag.items():
        v_old = baseline.by_tag.get(tag)
        if v_old is not None and v_new < v_old - 0.05:
            lines.append(f"⚠ 分层「{tag}」出现回落：{v_old:.1%} → {v_new:.1%}，"
                         f"整体指标可能掩盖了局部退化，请检查")
    return "\n".join(lines)