203 lines
8.1 KiB
Python
203 lines
8.1 KiB
Python
"""查标评分：对照响应清单给一次离线检查打分（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()
|