"""Recompute a paired report from saved rows. Standard library; no model calls.""" import argparse from collections import Counter, defaultdict import hashlib import json import math from pathlib import Path import random import statistics from mini_eval import grade def paired_interval(differences): n = len(differences) if n < 2: raise ValueError("at least two paired items required") delta = statistics.mean(differences) sem = statistics.stdev(differences) / math.sqrt(n) return {"delta": delta, "sem": sem, "normal_95": [delta - 1.96 * sem, delta + 1.96 * sem]} def bootstrap_interval(differences, repeats=10000, seed=20261017): # Resample whole A/B pairs, represented by their item-level differences. rng = random.Random(seed) n = len(differences) estimates = sorted(statistics.mean(rng.choices(differences, k=n)) for _ in range(repeats)) # Inverse empirical CDF: order statistic ceil(q * repeats), one-based. return [estimates[math.ceil(q * repeats) - 1] for q in (.025, .975)] def analyze(payload): if payload["protocol"] != "json-numeric-answer-v1": raise ValueError("unsupported grading protocol") rows = payload["rows"] by_model = defaultdict(dict) for row in rows: model, ident = row["model"], row["id"] if ident in by_model[model]: raise ValueError("duplicate item; reduce repeated samples per item first") if row["status"] not in {"correct", "wrong_answer", "invalid_format", "invalid_number", "infrastructure_error"}: raise ValueError("unknown status") if row["status"] == "infrastructure_error": if row["score"] is not None: raise ValueError("infrastructure errors need score=null") elif type(row["score"]) is not int or row["score"] != int(row["status"] == "correct"): raise ValueError("score and status disagree") elif grade(row["raw"], row["target"]) != {"status": row["status"], "score": row["score"]}: raise ValueError("saved grade does not match raw response; regrade first") by_model[model][ident] = row if len(by_model) != 2: raise ValueError("exactly two models required") a, b = sorted(by_model) if set(by_model[a]) != set(by_model[b]): raise ValueError("item sets differ; complete planned rows before comparing") summary = {} for model, items in by_model.items(): valid = [r for r in items.values() if r["status"] != "infrastructure_error"] groups = defaultdict(list) for row in valid: groups[row["slice"]].append(row["score"]) summary[model] = { "requests": len(items), "responses": len(valid), "accuracy_responses": statistics.mean(r["score"] for r in valid) if valid else None, "success_all_requests": sum(r["score"] for r in valid) / len(items), "statuses": dict(Counter(r["status"] for r in items.values())), "slices": {s: {"n": len(v), "accuracy": statistics.mean(v)} for s, v in groups.items()}, } pairs, excluded = [], [] counts = Counter() for ident in sorted(by_model[a]): ra, rb = by_model[a][ident], by_model[b][ident] if ra["slice"] != rb["slice"]: raise ValueError("paired slice labels differ") if "infrastructure_error" in (ra["status"], rb["status"]): excluded.append(ident) continue pairs.append(rb["score"] - ra["score"]) counts[f'{ra["score"]}{rb["score"]}'] += 1 if excluded: # Conditional subset analysis is allowed, but is not a full-test estimate. limitation = "共同有效子集；缺失可能有偏，需补齐计划试验或解释缺失机制。" else: limitation = "完整配对；区间仍假设题目独立且代表目标分布。" paired = paired_interval(pairs) paired.update({"n": len(pairs), "counts_A_B": dict(counts), "excluded_ids": excluded, "bootstrap_95": bootstrap_interval(pairs), "bootstrap_repeats": 10000, "seed": 20261017}) return {"data_kind": payload["data_kind"], "protocol": payload["protocol"], "model_order": [a, b], "summary": summary, "paired": paired, "limitation": limitation, "scope": "一次作答/题；不处理聚类或重复生成；成本与时延未提供。零方差或极端比例下，退化区间不代表确定性。"} def report_markdown(result): p = result["paired"] a, b = result["model_order"] lo, hi = p["normal_95"] blo, bhi = p["bootstrap_95"] lines = ["# 配对评测报告", "", f'数据性质：{result["data_kind"]}', f'协议：{result["protocol"]}', "", f'以 {a} 为 A，{b} 为 B；共同有效题数 {p["n"]}。', f'B−A：{p["delta"]*100:+.1f} 个百分点。', f'配对正态近似 95% 区间：[{lo*100:.1f}, {hi*100:.1f}] 个百分点。', f'配对 bootstrap 百分位区间：[{blo*100:.1f}, {bhi*100:.1f}] 个百分点。', 'bootstrap：10000 次，固定随机种子 20261017；不与正态近似混为同一算法。', ""] for model, summary in result["summary"].items(): accuracy = summary["accuracy_responses"] value = f'{accuracy:.1%}' if accuracy is not None else "无响应" lines += [f'## {model}', "", f'收到响应后的准确率：{value}；分母 {summary["responses"]}。', f'全部请求成功率：{summary["success_all_requests"]:.1%}；分母 {summary["requests"]}。', "", "| 判定类别 | 条数 |", "| --- | ---: |"] labels = {"correct": "正确", "wrong_answer": "数值错误", "invalid_format": "格式错误", "invalid_number": "非法数值", "infrastructure_error": "基础设施错误"} lines += [f'| {labels[status]} | {count} |' for status, count in summary["statuses"].items()] lines += ["", "| 分组 | 响应数 | 准确率 |", "| --- | ---: | ---: |"] lines += [f'| {s} | {v["n"]} | {v["accuracy"]:.1%} |' for s, v in summary["slices"].items()] lines += [""] lines += ["## 结论与限制", "", "当前区间包含 0，不能据此断言 B 提升；也不能证明两者等效。" if lo <= 0 <= hi else "该近似区间不含 0；仍需检查代表性、评分与实验设计，不能直接推广。", result["limitation"], result["scope"], "构造数据只用于检查计算流程，不是模型能力证据。" if "teaching" in result["data_kind"] else "结果仅适用于记录的任务与协议。"] if p["n"] < 30: lines += ["样本很少；这里的区间仅演示算法，正态近似和 bootstrap 都可能不可靠。"] if p["excluded_ids"]: lines += ["未进入配对分析的题目 ID：" + ", ".join(p["excluded_ids"])] if "input_sha256" in result: lines += ["", "## 复算记录", "", f'输入文件：{result["input_file"]}', f'输入 SHA-256：{result["input_sha256"]}'] return "\n".join(lines) + "\n" if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("input", type=Path) parser.add_argument("--output", type=Path, default=Path("comparison-report")) args = parser.parse_args() input_bytes = args.input.read_bytes() result = analyze(json.loads(input_bytes.decode("utf-8"))) result.update(input_file=args.input.name, input_sha256=hashlib.sha256(input_bytes).hexdigest()) args.output.with_suffix(".json").write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") args.output.with_suffix(".md").write_text(report_markdown(result), encoding="utf-8") print(report_markdown(result))