602703c47c
- evals/gold：1,887 个标注单元（AI 标注、逐条裁定，未经人工复核），规范见 GUIDELINES.md - prepare / run_all / score / gold：导出文字层、批量抽取、打分、标注定位 - docs/architecture/extract-eval.md：修复前后的结果（三项目标全部达标）和剩余问题 - decisions.md 记 M1（只用 Flash）和 Q1–Q4；架构 §13 SK13；AGENTS.md 进度 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
301 lines
13 KiB
Python
301 lines
13 KiB
Python
"""标注集评测：用 evals/gold/ 的标注给一次抽取（run_all.sh 的输出）打分。
|
||
|
||
用法（在 server/ 下）：../.venv/bin/python ../evals/score.py <运行名> [S01 ...]
|
||
输出 evals/out/score/<运行名>/：summary.md（各类召回率、准确率、项目信息正确率）、<ID>.md（漏抽、多抽、
|
||
被误拦进待确认区的条目，给人工裁定用）、score.json。
|
||
|
||
口径：
|
||
- 一条标注被「召回」= 有一个对应类别的清单项，出处句子和标注的句子有交集。正式清单算一次，正式 + 待确认再算一次
|
||
（待确认区用户也看得到，只是要多点一下）。
|
||
- 「准确」= 正式清单里这一类的条目，出处和这一类的某条标注（含 optional）有交集。
|
||
- 技术参数只在抽样页上算：标注是抽样页上的全部要求，条目只算出处落在抽样页上的。
|
||
- optional 的标注不算进召回的分母，抽到了算对。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import pathlib
|
||
import re
|
||
import sys
|
||
from collections import defaultdict
|
||
from decimal import Decimal, InvalidOperation
|
||
from typing import Any
|
||
|
||
import gold
|
||
|
||
ROOT = pathlib.Path(__file__).resolve().parent
|
||
CAT = {
|
||
"substantive": "invalidity",
|
||
"qualification": "qualification",
|
||
"scoring": "scoring",
|
||
"schedule": "schedule",
|
||
"tech_param": "tech_param",
|
||
}
|
||
NAMES = {
|
||
"substantive": "实质性条款",
|
||
"substantive/marked": " 标注类（★ / * / 斜体下划线）",
|
||
"substantive/explicit_invalid": " 写明无效类",
|
||
"qualification": "资格条件",
|
||
"scoring": "评分项",
|
||
"schedule": "日期与递交",
|
||
"tech_param": "技术参数（抽样页）",
|
||
}
|
||
|
||
|
||
def subgroup(u: gold.Unit) -> str | None:
|
||
if u.cat != "substantive":
|
||
return None
|
||
return "substantive/explicit_invalid" if u.raw.get("basis") == "explicit_invalid" else "substantive/marked"
|
||
|
||
|
||
def pct(a: int, b: int) -> str:
|
||
return f"{a / b:.1%}（{a}/{b}）" if b else "—"
|
||
|
||
|
||
def text_of(doc_id: str, sids: list[str] | set[str], limit: int = 80) -> str:
|
||
by = {gold.sid(s): s for s in gold.sentences(doc_id)}
|
||
return " / ".join(by[k].text[:limit] for k in sorted(sids, key=_order) if k in by)
|
||
|
||
|
||
def _order(k: str) -> tuple[int, int]:
|
||
m = re.match(r"p(\d+)s(\d+)", k)
|
||
return (int(m.group(1)), int(m.group(2))) if m else (0, 0)
|
||
|
||
|
||
def score_doc(run: str, doc_id: str) -> tuple[dict[str, Any], list[str]]:
|
||
out_dir = ROOT / "out" / "runs" / run / doc_id
|
||
items = json.loads((out_dir / "items.json").read_text())
|
||
info = json.loads((out_dir / "info.json").read_text())
|
||
g = gold.load(doc_id)
|
||
units, problems = gold.units(doc_id)
|
||
pages = set(g.get("tech_param_pages") or [])
|
||
star_scope = star_sections(doc_id, g)
|
||
page_of = lambda k: _order(k)[0] # noqa: E731
|
||
|
||
res: dict[str, Any] = {"problems": problems}
|
||
detail: list[str] = [f"# {doc_id} · {run}", ""]
|
||
if problems:
|
||
detail += ["## 标注定位问题", *(f"- {p}" for p in problems), ""]
|
||
|
||
def in_scope(cat: str, it: dict[str, Any]) -> bool:
|
||
return cat != "tech_param" or any(page_of(k) in pages for k in it["sentences"])
|
||
|
||
qual_sids = set().union(*(u.sids for u in units if u.cat == "qualification"))
|
||
for cat, pcat in CAT.items():
|
||
cu = [u for u in units if u.cat == cat]
|
||
preds = [it for it in items if pcat in it["categories"] and in_scope(cat, it)]
|
||
formal = [it for it in preds if it["zone"] == "formal"]
|
||
# 待确认区：待确认的条目，以及正式条目上「待确认的类别」（extract.add_pending_categories）
|
||
pending = [it for it in preds if it["zone"] != "formal"] + [
|
||
it
|
||
for it in items
|
||
if it["zone"] == "formal"
|
||
and pcat not in it["categories"]
|
||
and pcat in ((it.get("raw") or {}).get("pending_categories") or {})
|
||
and in_scope(cat, it)
|
||
]
|
||
gold_sids = set().union(*(u.sids for u in cu)) if cu else set()
|
||
|
||
def hit(u: gold.Unit, pool: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
return [it for it in pool if u.sids & set(it["sentences"])]
|
||
|
||
groups: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
|
||
missed, pend_only, dups = [], [], 0
|
||
for u in cu:
|
||
if u.optional:
|
||
continue
|
||
f, p = hit(u, formal), hit(u, pending)
|
||
other = [it for it in items if u.sids & set(it["sentences"])]
|
||
for key in filter(None, (cat, subgroup(u))):
|
||
groups[key]["gold"] += 1
|
||
groups[key]["formal"] += bool(f)
|
||
groups[key]["any"] += bool(f or p)
|
||
groups[key]["anycat"] += bool(other)
|
||
dups += len(f) > 1
|
||
if not f and not p:
|
||
missed.append((u, other))
|
||
elif not f:
|
||
pend_only.append((u, p))
|
||
def ok(it: dict[str, Any]) -> bool:
|
||
if gold_sids & set(it["sentences"]):
|
||
return True
|
||
if cat != "substantive":
|
||
return False
|
||
if set(it["sentences"]) <= star_scope: # 口径 Q1：★ 节里的条目（节内没有逐条标注）
|
||
return True
|
||
# 口径 Q4：政府采购项目的资格条件按法规（87 号令第六十三条）带「无效」类别
|
||
return it.get("importance_basis") == "statute" and bool(qual_sids & set(it["sentences"]))
|
||
|
||
right = [it for it in formal if ok(it)]
|
||
wrong = [it for it in formal if not ok(it)]
|
||
good_pending = [it for it in pending if gold_sids & set(it["sentences"])]
|
||
for key in [cat] + ([k for k in groups if k.startswith(cat + "/")] if cat == "substantive" else []):
|
||
res[key] = dict(groups[key])
|
||
res[cat].update(
|
||
pred_formal=len(formal),
|
||
formal_right=len(right),
|
||
pending=len(pending),
|
||
pending_right=len(good_pending),
|
||
dups=dups,
|
||
)
|
||
|
||
detail.append(f"## {NAMES[cat]}")
|
||
if missed:
|
||
detail.append(f"### 漏抽（{len(missed)}）")
|
||
for u, other in missed:
|
||
where = (
|
||
"；被别的类别抽到：" + ",".join(sorted({c for it in other for c in it["categories"]}))
|
||
if other
|
||
else ""
|
||
)
|
||
detail.append(
|
||
f"- **{u.id}** {u.raw.get('basis') or u.raw.get('kind') or ''} {u.raw.get('note', '')}{where}"
|
||
)
|
||
detail.append(f" - {text_of(doc_id, u.sids)}")
|
||
if pend_only:
|
||
detail.append(f"### 只在待确认区（{len(pend_only)}）")
|
||
for u, p in pend_only:
|
||
reasons = ",".join(
|
||
sorted(
|
||
{
|
||
r
|
||
for it in p
|
||
for r in (it.get("raw", {}) or {}).get("failures", []) or [it.get("reason") or ""]
|
||
}
|
||
)
|
||
)
|
||
detail.append(f"- **{u.id}** {u.raw.get('note', '')}〔{reasons}〕")
|
||
if wrong:
|
||
detail.append(f"### 正式清单里对不上标注的（{len(wrong)}）")
|
||
for it in wrong:
|
||
detail.append(f"- [{it['task']}] {it['title']}｜{text_of(doc_id, it['sentences'], 60)}")
|
||
detail.append("")
|
||
|
||
res["info"] = score_info(g, info)
|
||
detail.append("## 项目信息")
|
||
for k, (ok, want, got) in res["info"].items():
|
||
detail.append(f"- {'✓' if ok else '✗'} {k}：标注 {want}，抽到 {got}")
|
||
if g.get("marking", {}).get("declared_count"):
|
||
n = sum(1 for u in units if u.cat == "substantive" and u.raw.get("basis") == "star")
|
||
res["declared"] = [g["marking"]["declared_count"], n]
|
||
return res, detail
|
||
|
||
|
||
def star_sections(doc_id: str, g: dict[str, Any]) -> set[str]:
|
||
"""标注集里记的 ★ 节范围（marking.star_sections，from 含、to 不含）里的句子编号。"""
|
||
order = [gold.sid(s) for s in gold.sentences(doc_id)]
|
||
out: set[str] = set()
|
||
for sec in (g.get("marking") or {}).get("star_sections") or []:
|
||
a, _ = gold.locate(doc_id, sec["from"]["p"], sec["from"]["t"])
|
||
b, _ = gold.locate(doc_id, sec["to"]["p"], sec["to"]["t"])
|
||
if a and b:
|
||
out.update(order[order.index(a[0]) : order.index(b[0])])
|
||
return out
|
||
|
||
|
||
def yuan(v: Any, unit: Any) -> Decimal | None:
|
||
try:
|
||
n = Decimal(str(v).replace(",", "").replace("，", ""))
|
||
except (InvalidOperation, ValueError):
|
||
return None
|
||
return n * 10000 if unit and "万" in str(unit) else n
|
||
|
||
|
||
def score_info(g: dict[str, Any], info: dict[str, Any]) -> dict[str, tuple[bool | None, Any, Any]]:
|
||
"""每个字段：(对不对, 标注值, 抽到的值)。标注是 null 的，抽到「没写」才算对；抽到 needs_review 算没答出来。"""
|
||
want_all = g.get("project_info") or {}
|
||
out: dict[str, tuple[bool | None, Any, Any]] = {}
|
||
for k in gold.INFO_KEYS:
|
||
want = want_all.get(k)
|
||
f = info["fields"].get(k)
|
||
status = info["status"].get(k, {}).get("status")
|
||
got: Any = f["value"] if f else f"〔{status}〕"
|
||
if f is None:
|
||
ok = want is None and status == "absent"
|
||
elif want is None:
|
||
ok = False
|
||
elif k in ("budget_total", "ceiling_total"):
|
||
w = yuan(want.get("value"), want.get("unit")) if isinstance(want, dict) else yuan(want, None)
|
||
ok = w is not None and w == yuan(f["value"], f.get("unit"))
|
||
got = f"{f['value']}{f.get('unit') or ''}"
|
||
elif isinstance(want, bool):
|
||
ok = f["value"] is want
|
||
else:
|
||
a, b = gold.norm(str(want)), gold.norm(str(f["value"]))
|
||
ok = a == b if k in ("code", "legal_regime") else bool(a and b and (a in b or b in a))
|
||
if isinstance(want, dict):
|
||
want = f"{want.get('value')}{want.get('unit') or ''}"
|
||
out[k] = (ok, want, got)
|
||
return out
|
||
|
||
|
||
def main() -> None:
|
||
run = sys.argv[1]
|
||
ids = sys.argv[2:] or [p.stem for p in sorted(gold.GOLD.glob("S*.json"))]
|
||
dest = ROOT / "out" / "score" / run
|
||
dest.mkdir(parents=True, exist_ok=True)
|
||
all_res: dict[str, Any] = {}
|
||
for doc_id in ids:
|
||
if not (ROOT / "out" / "runs" / run / doc_id / "items.json").exists():
|
||
print(doc_id, "没有抽取结果，跳过")
|
||
continue
|
||
res, detail = score_doc(run, doc_id)
|
||
all_res[doc_id] = res
|
||
(dest / f"{doc_id}.md").write_text("\n".join(detail) + "\n")
|
||
(dest / "score.json").write_text(json.dumps(all_res, ensure_ascii=False, indent=1, default=str))
|
||
|
||
lines = [f"# 标注集评测 · {run}", "", f"文件：{', '.join(all_res)}", ""]
|
||
lines += [
|
||
"| 类别 | 标注 | 召回（正式清单） | 召回（含待确认） | 准确率（正式清单） | 待确认里是好结果 | 重复条目 |",
|
||
"|---|---|---|---|---|---|---|",
|
||
]
|
||
for key in NAMES:
|
||
tot: dict[str, int] = defaultdict(int)
|
||
for r in all_res.values():
|
||
for k, v in (r.get(key) or {}).items():
|
||
tot[k] += v
|
||
prec = pct(tot["formal_right"], tot["pred_formal"]) if "/" not in key else ""
|
||
pend = pct(tot["pending_right"], tot["pending"]) if "/" not in key else ""
|
||
dups = str(tot["dups"]) if "/" not in key else ""
|
||
lines.append(
|
||
f"| {NAMES[key]} | {tot['gold']} | {pct(tot['formal'], tot['gold'])} | {pct(tot['any'], tot['gold'])} | {prec} | {pend} | {dups} |"
|
||
)
|
||
lines += [
|
||
"",
|
||
"## 按文件",
|
||
"",
|
||
"| 文件 | " + " | ".join(NAMES[k].strip(" ") for k in NAMES) + " |",
|
||
"|---|" + "---|" * len(NAMES),
|
||
]
|
||
for doc_id, r in all_res.items():
|
||
cells = []
|
||
for key in NAMES:
|
||
v = r.get(key) or {}
|
||
cell = f"{v.get('formal', 0)}/{v.get('any', 0)}/{v.get('gold', 0)}"
|
||
if "/" not in key:
|
||
cell += f" · {v.get('formal_right', 0)}/{v.get('pred_formal', 0)}"
|
||
cells.append(cell)
|
||
lines.append(f"| {doc_id} | " + " | ".join(cells) + " |")
|
||
lines += ["", "格子里：正式召回/含待确认召回/标注数 · 正式条目对上标注/正式条目数", ""]
|
||
lines += ["## 项目信息", "", "| 字段 | 正确 | 错的 |", "|---|---|---|"]
|
||
for k in gold.INFO_KEYS:
|
||
oks = [(d, r["info"][k]) for d, r in all_res.items()]
|
||
bad = [f"{d}（标 {w}，抽 {g_}）" for d, (ok, w, g_) in oks if not ok]
|
||
lines.append(f"| {k} | {sum(1 for _, (ok, _, _) in oks if ok)}/{len(oks)} | {'；'.join(bad)} |")
|
||
declared = {d: r["declared"] for d, r in all_res.items() if "declared" in r}
|
||
if declared:
|
||
lines += [
|
||
"",
|
||
"文件写明的 ★ 条款数 / 标注的 ★ 单元数：" + "，".join(f"{d} {a}/{b}" for d, (a, b) in declared.items()),
|
||
]
|
||
problems = {d: r["problems"] for d, r in all_res.items() if r["problems"]}
|
||
if problems:
|
||
lines += ["", "## 标注定位问题", *(f"- {d}：{len(p)} 处" for d, p in problems.items())]
|
||
(dest / "summary.md").write_text("\n".join(lines) + "\n")
|
||
print("\n".join(lines))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|