评测集按失败类型切分评测集的实现方法

评测集只看一个总分，最容易漏掉局部回归：拒答率下降了，可能是格式错误突然变多；总体准确率上升了，也可能只是简单样本占比更高。更稳妥的做法是给每条样本保留唯一的主失败类型，把评测集切成固定分桶，再对每个分桶单独计算指标。

官方文档：https://huggingface.co/docs/evaluate/index

一、先固定样本字段和失败类型

分桶的第一步不是写统计代码，而是先定义一条样本的最小记录。建议至少保存 sample_id、task_type、reference、prediction、judge_result 和 primary_failure。其中 primary_failure 只能有一个主值，否则同一条错误会被多个分母重复计算。

失败类型可以按业务调整，但要先定义优先级。例如工具调用已经失败时，不再把它同时记成格式错误；模型明确拒答时，优先记录拒答。这样每个分桶的数量相加才等于纳入统计的样本总数。

def classify_failure(row):
    # 先处理会影响后续判断的硬失败，保证主标签唯一
    if row["tool_call_ok"] is False:
        return "tool_call_failure"
    if row["refusal"] is True:
        return "refusal"
    if row["format_ok"] is False:
        return "format_error"
    if row["fact_ok"] is False:
        return "factual_error"
    # 没有失败时保留成功标签，便于计算全量分布
    return "pass"

如果一条样本可能同时命中多个规则，应该把“主失败类型”和“附加标签”分开：前者服务于互斥统计，后者用于排查组合问题。

二、按主失败类型切分评测集

切分时保留原始记录，不要只保存样本编号。调试需要回看提示词、参考结果和模型输出；只留下计数，后面无法解释失败率为什么变化。

from collections import defaultdict

def split_by_failure(rows):
    # defaultdict 让每种失败类型都得到一个独立样本列表
    buckets = defaultdict(list)
    for row in rows:
        failure = row.get("primary_failure", "unknown")
        buckets[failure].append(row)
    return dict(buckets)

# 每个桶仍保留完整 row，后续可直接回放和抽样
buckets = split_by_failure(evaluation_rows)

切完后先做三个检查：所有桶的样本数之和是否等于输入数；是否出现空字符串、拼写漂移或未知标签；同一个 sample_id 是否被重复写入。线上回归还要固定一份桶清单，新增样本进入下一版，不要悄悄改变旧桶的分母。

三、让每个分桶独立计算指标

Hugging Face Evaluate 的评测模块可以通过 add_batch 累积预测值和参考值，再用 compute 返回指标。把这套调用封装成函数，就能对全量集和每个失败桶使用相同的计算逻辑。

import evaluate

accuracy = evaluate.load("accuracy")

def score_bucket(rows):
    # 只把当前桶的预测和参考答案交给指标模块
    predictions = [row["prediction_label"] for row in rows]
    references = [row["reference_label"] for row in rows]
    if not rows:
        return {"count": 0, "accuracy": None}
    result = accuracy.compute(predictions=predictions, references=references)
    # count 与指标一起保存，避免小样本高分被误读
    return {"count": len(rows), "accuracy": result["accuracy"]}

bucket_scores = {name: score_bucket(rows) for name, rows in buckets.items()}

不同失败类型未必适合完全相同的指标：格式错误可以看 schema 通过率，事实错误可以看人工或规则判定通过率，工具调用失败则要看调用成功率和参数完整性。可以共用结果结构，但不要为了横向比较而强行共用一个指标。

四、用固定分桶做版本回归

回归比较要锁定同一批 sample_id，再分别运行基线版本和新版本。最有用的输出不是“新版本总分 0.82”，而是“事实错误桶下降了多少、工具调用失败桶是否反弹、每个结论有多少样本支撑”。

def compare_versions(baseline, candidate, min_count=30):
    report = []
    for failure, old in baseline.items():
        new = candidate.get(failure, {"count": 0, "accuracy": None})
        # 样本太少时只记录变化，不把它升级为稳定结论
        delta = None
        if old["accuracy"] is not None and new["accuracy"] is not None:
            delta = new["accuracy"] - old["accuracy"]
        report.append({
            "failure": failure,
            "baseline_count": old["count"],
            "candidate_count": new["count"],
            "accuracy_delta": delta,
            "enough_samples": old["count"] >= min_count,
        })
    return report

建议把阈值和结论分开：样本数低于门槛时标记为观察项；达到门槛且失败率变化超过业务容忍线时，才进入发布阻断或人工复核。若桶内样本被重新标注，也要记录标注版本，否则“回归”可能只是标签规则变了。

常见问题

一条样本同时有事实错误和格式错误怎么办？

按预先定义的优先级选择一个 primary_failure，再把另一个写入附加标签。主桶用于互斥统计，附加标签用于组合分析。

为什么不能只比较总准确率？

总准确率会被大样本桶主导，少量但关键的工具调用失败可能被掩盖。分桶指标能指出具体回归发生在哪里。

新版本增加了样本，旧桶还能直接比较吗？

可以新增独立版本，但基线比较应先使用两版共有的固定样本；新增样本另列为扩展集，避免改变原回归分母。