标注规范、人工 Rubric 与仲裁

mkdir l7-labeling
cd l7-labeling

"""用预置标注记录演示分歧队列、证据仲裁和原始一致率。"""

from copy import deepcopy


RUBRIC_VERSION = "refund-rubric-v1"
# 记录是合成评审夹具：pass=证据齐且动作符合规则，fail=违反规则，
# unknown=证据缺失，无法判断。A/B并非本次参与的真人标注者。
RATINGS = [
    {"id": "r1", "a": "pass", "b": "pass", "evidence": "7天内、身份匹配且回执齐全"},
    {"id": "r2", "a": "fail", "b": "fail", "evidence": "订单属于另一用户"},
    {"id": "r3", "a": "fail", "b": "pass", "evidence": "第8天仍批准退款"},
    {"id": "r4", "a": "unknown", "b": "fail", "evidence": "缺少订单查询回执"},
]
# 仲裁保留规则版本和理由，不能只把两票改成一个平均分。
DECISIONS = {
    "r3": {"label": "fail", "reason": "超过7天；礼貌措辞不改变退款条件",
           "rubric": RUBRIC_VERSION},
    "r4": {"label": "unknown", "reason": "缺少回执，不能证明动作正确或错误",
           "rubric": RUBRIC_VERSION},
}


def reconcile(ratings, decisions):
    """一致记录直接保留；分歧没有有效仲裁时进入pending队列。"""
    results, seen = [], set()
    for row in ratings:
        if row["id"] in seen:
            raise ValueError("重复样本编号")
        seen.add(row["id"])
        if row["a"] not in ("pass", "fail", "unknown") or row["b"] not in (
                "pass", "fail", "unknown"):
            raise ValueError("未知标注值")
        if row["a"] == row["b"]:
            label, source, reason = row["a"], "agreement", row["evidence"]
        elif row["id"] not in decisions:
            label, source, reason = "pending", "queue", "等待证据仲裁"
        else:
            decision = decisions[row["id"]]
            if decision["rubric"] != RUBRIC_VERSION:
                raise ValueError("仲裁规则版本不一致")
            if decision["label"] not in ("pass", "fail", "unknown") or not (
                    decision["reason"].strip()):
                raise ValueError("仲裁需要合法标签和理由")
            label, source, reason = decision["label"], "arbitration", decision["reason"]
        results.append({"id": row["id"], "label": label, "source": source, "reason": reason})
    return results


def agreement(ratings):
    """原始一致率在仲裁前计算，防止仲裁把分歧记录抹掉。"""
    if not ratings:
        return None
    return sum(row["a"] == row["b"] for row in ratings) / len(ratings)


def main():
    pending = reconcile(deepcopy(RATINGS), {})
    agreed = sum(row["a"] == row["b"] for row in RATINGS)
    print(f"规范={RUBRIC_VERSION} 原始一致={agreed}/{len(RATINGS)}={agreement(RATINGS):.0%}")
    print(f"仲裁前待处理={sum(row['label'] == 'pending' for row in pending)}")
    resolved = reconcile(RATINGS, DECISIONS)
    for row in resolved:
        print(f"{row['id']}: {row['label']} 来源={row['source']} 理由={row['reason']}")
    print(f"仲裁后待处理={sum(row['label'] == 'pending' for row in resolved)}")


if __name__ == "__main__":
    main()

ls
python -X utf8 labeling_lab.py

规范=refund-rubric-v1 原始一致=2/4=50%
仲裁前待处理=2
r1: pass 来源=agreement 理由=7天内、身份匹配且回执齐全
r2: fail 来源=agreement 理由=订单属于另一用户
r3: fail 来源=arbitration 理由=超过7天；礼貌措辞不改变退款条件
r4: unknown 来源=arbitration 理由=缺少回执，不能证明动作正确或错误
仲裁后待处理=0

ls
python -X utf8 -m unittest -v test_labeling_lab.py