标注一致性 从双人混淆矩阵到 Cohen kappa

两位标注员有 80% 的样本给出相同标签，是否就说明标注质量很好？还不够。如果九成样本都被判为负类，较高的一致率可能主要来自这个类别分布；即使两人完全一致，也可能共同误解了任务。

标注检查需要同时回答：哪些样本有双人标注、分歧集中在哪类、相对于两人的类别倾向，一致程度是多少。 本文限于两位标注员、每个样本一个名义类别；不直接处理多标签、框选 IoU、分词边界或多人标注。

先定义标注单位和判断规则

标签不是天然存在的数字。做情感分类时，要先决定单位是整段评论还是一句话，混合评价怎样处理，讽刺、引用和无法判断的内容是否允许弃权。规则不清晰时，换一个更复杂的一致性指标也不能替代任务定义。

要记录的内容	用途
样本 ID 与版本	确保两人判断的是同一个标注单位，而不是按文件行号误配
标签定义与正反例	使判断条件可以被复核
原始双人标签	计算独立判断的一致性，不能被裁决结果覆盖
缺失与弃权原因	区分未完成、无法判断与正式的“中性”标签
裁决标签与依据	为训练或评测生成最终版本，但单独保存原始分歧

先做一小批盲标试标，再讨论分歧、修改规则并用新样本复查。把裁决后的统一标签重新算成“100% 一致”，会掩盖原来的规则问题。

1. 固定单位与规则

   明确什么算一个样本，标签互斥关系和弃权语义。

2. 保留独立判断与覆盖率

   按 ID 配对，缺失样本不悄悄消失。

3. 定位分歧后再裁决

   先检查类别与边界案例，再生成带版本的最终标签。

混淆矩阵的行列都不是金标准

下面是十条合成数据，行表示标注员 A，列表示标注员 B。标签顺序固定为 negative、positive。

A 的判断 / B 的判断	negative	positive	A 的合计
negative	4	1	5
positive	1	4	5
B 的合计	5	5	10

对角线有 8 条，一致率 p_o=8/10=0.8。这不是分类准确率：这里没有指定谁是真实标签。交换 A 与 B 会转置矩阵，但不改变总体一致率与 κ。

Cohen kappa 怎样修正类别倾向

scikit-learn 的官方定义给出 **κ=(p_o−p_e)/(1−p_e)**，p_e 根据每位标注员各自的经验类别分布计算。

记 A、B 给类别 c 的比例为 p_A(c)、p_B(c)，则 **p_e=Σ_c p_A(c)p_B(c)**。这是在两人的边际分布保持不变、判断独立这一参照模型下的预期一致率，不是观察到他们实际“随机乱标”的证据。

均衡例中，两人的类别比例都是 0.5、0.5，所以 p_e=0.5²+0.5²=0.5，κ=(0.8−0.5)/(1−0.5)=0.6。

量	计算	结果
观察一致率 p_o	(4+4)/10	0.8
边际基线 p_e	(5/10)(5/10)+(5/10)(5/10)	0.5
超出基线的一致	p_o−p_e	0.3
κ	0.3/(1−0.5)	0.6

同样八条一致为什么可能得到负值

第二组也是十条数据，但矩阵变为 [[8,1],[1,0]]。对角线仍为 8，一致率仍为 0.8；两人的边际却都是 9 条负类、1 条正类。

p_e=0.9²+0.1²=0.82，κ=(0.8−0.82)/(1−0.82)=−1/9，约 −0.1111。它表示在这个边际参照模型下，一致程度比基线还低；不表示“两人有 11% 的错误率”。

原创合成计数示意。两组都没有给定金标准，因此不能把一致率称为准确率。查看原图

保留缺失覆盖率的纯 Python 实验

不依赖第三方库。None 表示缺失，不是一个正式类别；未知标签、重复 ID 会被拒绝。代码只对双方都完成的样本计算矩阵，同时报告总量与各自缺失数。没有配对样本，或 p_e=1 时，κ 返回 None，不伪造为 0 或 1。

import json


def agreement(records, labels=("negative", "positive")):
    if not isinstance(labels, (list, tuple)) or not labels:
        raise ValueError("invalid labels")
    if any(type(label) is not str or not label for label in labels):
        raise ValueError("invalid label name")
    if len(set(labels)) != len(labels):
        raise ValueError("duplicate labels")
    records = list(records)
    index = {label: i for i, label in enumerate(labels)}
    matrix = [[0] * len(labels) for _ in labels]
    seen, missing_a, missing_b, paired = set(), 0, 0, 0
    for record in records:
        if not isinstance(record, dict):
            raise ValueError("invalid record")
        if not all(key in record for key in ("id", "a", "b")):
            raise ValueError("missing field")
        item = record["id"]
        if type(item) is not str or not item.strip() or item in seen:
            raise ValueError("invalid or duplicate id")
        seen.add(item)
        a, b = record["a"], record["b"]
        for label in (a, b):
            if label is not None and (type(label) is not str or label not in index):
                raise ValueError("unknown label")
        missing_a += a is None
        missing_b += b is None
        if a is None or b is None:
            continue
        matrix[index[a]][index[b]] += 1
        paired += 1
    row = [sum(values) for values in matrix]
    col = [sum(values[j] for values in matrix) for j in range(len(labels))]
    diagonal = sum(matrix[i][i] for i in range(len(labels)))
    cross = sum(a * b for a, b in zip(row, col))
    denominator = paired * paired - cross
    return {"labels": list(labels), "matrix": matrix,
            "total": len(records), "paired": paired,
            "missing_a": missing_a, "missing_b": missing_b,
            "coverage": paired / len(records) if records else None,
            "observed": diagonal / paired if paired else None,
            "expected": cross / (paired * paired) if paired else None,
            "kappa": (paired * diagonal - cross) / denominator
                     if denominator else None}


if __name__ == "__main__":
    pairs = [("negative", "negative")] * 4 + [("negative", "positive")]
    pairs += [("positive", "negative")] + [("positive", "positive")] * 4
    records = [{"id": f"s{i}", "a": a, "b": b} for i, (a, b) in enumerate(pairs)]
    records.append({"id": "missing", "a": None, "b": "positive"})
    print(json.dumps(agreement(records), ensure_ascii=False, indent=2))

演示输出的 paired=10、total=11、κ=0.6；缺失记录不改变十条配对样本的计数，但覆盖率只有 10/11。边际分布也只由同一个配对子集估计，不能混入另一人尚未完成的额外记录。若困难样本更容易缺失，已配对子集可能不能代表全部任务。需要再检查缺失原因，而不是只把它当作分母问题。

对 n 条记录、k 个标签，计数与 ID 检查平均 O(n)，矩阵汇总 O(k²)，总空间 O(n+k²)。整数形式的分子分母避免了先计算两个接近 1 的小数再相减。

指标不能代替什么

情况	必须保留的判断
两人都只标一个相同类别	p_o=p_e=1，κ 分母为零；报告退化情况
高一致率	不保证标签正确，需独立质量检查与分歧复核
不同数据集比较 κ	同时看类别分布、覆盖率与抽样方式
有序等级标签	可考虑有明确距离语义的加权指标，不随意给名义类别排序
多人、多标签或不同片段边界	本例并不直接适用，需要匹配标注单位的指标
设定通过门槛	依据任务风险与复核流程预先制定，不把某个 κ 阈值当普遍质量标准

自检：两人都把所有样本标成负类，可以说 κ=1 吗？

不能。虽然观察一致率是 1，边际基线也是 1，归一化分母为零。应报告 κ 未定义，并检查类别覆盖，而不是自动判定质量满分。

测试位于项目 tests/verify-annotation-agreement-article.cjs，用独立分数运算核对随机矩阵、交换标注员、标签重命名和缺失统计。进一步阅读 Artstein 与 Poesio 的标注一致性综述。建立评测标签后，可接上RAG 回归评测；训练端则进入反向传播与优化器。