"""查标评分：对照响应清单给一次离线检查打分（evals/check/GUIDELINES.md §7）。 cd server && ../.venv/bin/python ../evals/check/score.py <投标文件> [--run base] [--include-unreviewed] - 清单项 → 标注单元：清单项的招标侧句子和单元的句子有重叠就对上（和读标评分同一口径）； - 按单元去重：同一要求在清单里有几条，只要一条报出就算召回，几条都报不算误报； - 问题类型按类型族比（D01 / D02 / D09 / D11 / D13 → unresponded，D03 / D07 → suspected_unmet，……）； - 只算复核过的问题（决策 E1）；--include-unreviewed 时也算没复核的，报告里标明； - 读标没抽到的问题单列「读标漏抽」：不算查标漏报，但计入端到端召回； - 对不上任何标注单元的问题单列「无法对照」。 输出：evals/out/check/<投标文件>/runs/<运行>/score.json、score.md。 """ from __future__ import annotations import argparse import json import pathlib import sys from collections import Counter, defaultdict from typing import Any EVALS = pathlib.Path(__file__).resolve().parents[1] sys.path.insert(0, str(EVALS)) import gold # noqa: E402 OUT = EVALS / "out" / "check" FAMILY = { "D01": {"unresponded"}, "D02": {"unresponded"}, "D09": {"unresponded"}, "D11": {"unresponded"}, "D13": {"unresponded"}, "D16": {"unresponded"}, "D03": {"suspected_unmet"}, "D07": {"suspected_unmet"}, "D08": {"suspected_unmet"}, "D05": {"negative_deviation", "suspected_unmet"}, "D06": {"negative_deviation"}, "D10": {"negative_deviation"}, "D12": {"qualification"}, "D14": {"qualification"}, "D15": {"qualification", "unresponded"}, } FATAL_CATS = {"substantive", "qualification"} def pct(a: int, b: int) -> str: return f"{a}/{b}（{a / b:.0%}）" if b else "—" def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("bid") ap.add_argument("--run", default="base") ap.add_argument("--include-unreviewed", action="store_true") a = ap.parse_args() d = OUT / a.bid manifest = json.loads((d / "manifest.json").read_text()) res = json.loads((d / "runs" / a.run / "results.json").read_text()) units, _ = gold.units(manifest["tender"]) sid_units: dict[str, set[str]] = defaultdict(set) for u in units: for s in u.sids: sid_units[s].add(u.id) # 标注集里同一条要求有时记了两个单元（两个类别各一条，或表格一行拆成几条）：句子有重叠的单元合成一组，按组计 parent = {u.id: u.id for u in units} def find(x: str) -> str: while parent[x] != x: x = parent[x] return x for ids in sid_units.values(): first, *rest = sorted(ids) for o in rest: parent[find(o)] = find(first) group = {u.id: find(u.id) for u in units} per_unit: dict[str, list[dict[str, Any]]] = defaultdict(list) unmapped: list[dict[str, Any]] = [] for it in res["items"]: hit = {u for s in it["sentences"] for u in sid_units.get(s, ())} for u in hit: per_unit[u].append(it) if not hit and it["finding"]: unmapped.append(it) rec: Counter[str] = Counter() rec_total: Counter[str] = Counter() sev_ok = sev_n = 0 misses: list[str] = [] extraction_miss: list[str] = [] fp: list[str] = [] fp_by_trap: Counter[str] = Counter() ok_units = trap_units = 0 other_findings = 0 seen_groups: set[str] = set() for uid, e in sorted(manifest["units"].items(), key=lambda kv: kv[1].get("expect") != "defect"): g = group.get(uid, uid) if g in seen_groups: continue # 同组的单元只算一次（挂了问题的那个优先） seen_groups.add(g) items = [it for x in manifest["units"] if group.get(x, x) == g for it in per_unit.get(x, [])] found = [it for it in items if it["finding"]] if e["expect"] == "defect": if not e.get("reviewed") and not a.include_unreviewed: continue keys = [e["defect"], f"粒度:{e['granularity']}", "带标志词" if e.get("worded") else "不带标志词"] fatal = e["cat"] in FATAL_CATS if fatal: keys.append("致命级") for k in keys: rec_total[k] += 1 rec_total["端到端"] += 1 if not items: extraction_miss.append(f"{uid}（{e['defect']}，{e['cat']}）") continue rec_total["查标"] += 1 good = [it for it in found if it["finding"]["type"] in FAMILY.get(e["defect"], set())] if good: for k in [*keys, "端到端", "查标"]: rec[k] += 1 if fatal: sev_n += 1 sev_ok += any(it["finding"]["severity"] == "fatal" for it in good) else: got = ", ".join(sorted({it["verdict"] for it in items})) misses.append(f"{uid}（{e['defect']} {e.get('variant', '')}）→ 查标结论：{got}") elif e["expect"] == "ok": ok_units += 1 if e.get("trap"): trap_units += 1 if found: fp.append(f"{uid}（{e['cat']}，{e['mode']}）：{found[0]['finding']['title'][:40]}") if e.get("trap"): fp_by_trap[e["trap"]] += 1 elif found: other_findings += 1 findings = [it for it in res["items"] if it["finding"]] lines = [ f"# 查标评分：{a.bid} · 运行 {a.run}", "", f"- 投标文件：{manifest['kind']}，风格 {manifest['style']}，{manifest['split']}；清单 {len(res['items'])} 条，报出问题 {len(findings)} 个", f"- 金标准：{'含未复核的问题（试跑口径）' if a.include_unreviewed else '只算复核过的问题'}；复核方式见决策 E1", f"- 模型：{res['summary']['usage']}", "", "## 召回", "", f"- 查标召回（读标抽到的要求上）：{pct(rec['查标'], rec_total['查标'])}", f"- 端到端召回（含读标漏抽）：{pct(rec['端到端'], rec_total['端到端'])}", f"- 致命级：{pct(rec['致命级'], rec_total['致命级'])}；其中级别也判成致命的 {pct(sev_ok, sev_n)}", "", "| 分类 | 召回 |", "|---|---|", *[f"| {k} | {pct(rec[k], rec_total[k])} |" for k in sorted(rec_total) if k not in ("端到端", "查标", "致命级")], "", "## 误报", "", f"- 规矩响应的要求 {ok_units} 条，报了问题的 {len(fp)} 条（每 100 条 {100 * len(fp) / ok_units:.1f}）" if ok_units else "- 没有规矩响应的要求", f"- 干扰项 {trap_units} 条，误报 {dict(fp_by_trap)}", f"- 要求本来不需要响应或无法判断（na / unknown）却报了问题：{other_findings}", f"- 无法对照（对不上任何标注单元）的问题：{len(unmapped)}", "", f"## 漏报（{len(misses)}）", "", *[f"- {x}" for x in misses], "", f"## 读标漏抽（{len(extraction_miss)}）", "", *[f"- {x}" for x in extraction_miss], "", f"## 误报明细（{len(fp)}）", "", *[f"- {x}" for x in fp[:80]], *(["- ……"] if len(fp) > 80 else []), "", f"## 规则问题（第 2 步才查，这里只列出）：{len(manifest['rule_defects'])} 个", "", ] out = d / "runs" / a.run (out / "score.md").write_text("\n".join(lines)) summary = { "recall": {k: [rec[k], rec_total[k]] for k in rec_total}, "fp": len(fp), "ok_units": ok_units, "fp_by_trap": dict(fp_by_trap), "unmapped": len(unmapped), "other_findings": other_findings, "extraction_miss": len(extraction_miss), "include_unreviewed": a.include_unreviewed, } (out / "score.json").write_text(json.dumps(summary, ensure_ascii=False, indent=1)) print("\n".join(lines[:20])) if __name__ == "__main__": main()