eric/bid-doctor
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(evals): 查标评测集生成器、注入、评分和只检索基线

Browse Source 
- build.py：从「招标文件档案」和读标清单生成规矩投标文件稿；技术参数按 op / value / unit 算满足的值，
  分项报价按分项上限比例分摊，落款日期取截止日前 3 天，偏离表按格式表头决定列（含设备名称列），
  读标漏抽的技术参数也补上响应；routes 支持「+」组合和 prose 散文响应。
- mutate.py：按词库注入要求项问题（单点 / 连续块，带不带标志词）和规则问题，写响应清单；
  符合性条件响应表的问题按表级记一个。
- score.py：按单元去重、类型族比对、读标漏抽单列、端到端召回、无法对照单列。
- run_check.py：加只检索不调模型的基线；render.py：扫描件按最后一次出现定位。
- vocab-dev.json：开发集注入词库（留出集的另行封存，不进 git）。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
YANG JIANKUAN
2026-11-04 12:18:48 +08:00
parent 01ddc86281
commit a07663cbdb
6 changed files with 1070 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

390
evals/check/mutate.py Normal file
View File

@@ -0,0 +1,390 @@
"""往规矩的投标文件稿里注入问题，写出马虎的投标文件稿和响应清单（evals/check/GUIDELINES.md §3、§6.2）。
cd server && ../.venv/bin/python ../evals/check/mutate.py <规矩的投标文件> <输出的投标文件> --split dev [--seed 1] [--density high|low]
cd server && ../.venv/bin/python ../evals/check/mutate.py <规矩的投标文件> --clean --split dev # 只给规矩的投标文件写响应清单
- 词库：开发集用 evals/check/vocab-dev.json；留出集用封存的 evals/out/check/_sealed/vocab-holdout.json；
- 注入是确定性的（同一个 seed 结果相同），每个问题记 defect、variant、granularity、worded；
- 问题只挂在标注单元上；挂上的单元先记 reviewed: false，复核（决策 E1）后改成 true，评分只算复核过的；
- 响应方式按单元在投标文件里的响应位置推出来：逐条响应表 → explicit，只列偏离的表 → implicit，附件 → evidence，格式文件 → presence。
"""
from __future__ import annotations
import argparse
import copy
import json
import pathlib
import random
import re
import sys
from decimal import Decimal
from typing import Any
EVALS = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(EVALS))
sys.path.insert(0, str(EVALS / "check"))
import gold # noqa: E402
from build import rmb_upper # noqa: E402
OUT = EVALS / "out" / "check"
VOCAB = {"dev": EVALS / "check" / "vocab-dev.json", "holdout": OUT / "_sealed" / "vocab-holdout.json"}
UNIT_DOWN = {"TB": ("GB", 1024), "GB": ("MB", 1024), "Gbps": ("Mbps", 1000), "s": ("ms", 1000), "秒": ("毫秒", 1000)}
class Mutator:
def __init__(self, spec: dict[str, Any], vocab: dict[str, Any], rng: random.Random) -> None:
self.spec, self.vocab, self.rng = spec, vocab, rng
self.unit_defects: dict[str, dict[str, Any]] = {}
self.rule_defects: list[dict[str, Any]] = []
self.blocks = spec["volumes"][0]["blocks"]
# ---------- 定位 ----------
def tables(self, *, explicit: bool | None = None, fixed: bool = False) -> list[dict[str, Any]]:
out = []
for b in self.blocks:
if b.get("type") != "table" or not b.get("form"):
continue
if fixed:
if b.get("fixed"):
out.append(b)
continue
if b.get("fixed") or "only_deviations" not in b:
continue
if explicit is None or b["only_deviations"] != explicit:
out.append(b)
return out
def free(self, units: list[str]) -> bool:
return bool(units) and not any(u in self.unit_defects for u in units)
def mark(
self, units: list[str], defect: str, variant: str, granularity: str = "single", worded: bool = True
) -> None:
for u in units:
self.unit_defects[u] = {"defect": defect, "variant": variant, "granularity": granularity, "worded": worded}
def phrase(self, key: str, field: str) -> str:
return str(self.rng.choice(self.vocab[key][field]))
# ---------- 要求项问题 ----------
def d01_block(self) -> None:
"""连续漏掉 5～10 行（逐条响应表）。"""
for t in self.rng.sample(self.tables(explicit=True), k=len(self.tables(explicit=True))):
rows = t["rows"]
if len(rows) < 12:
continue
n = self.rng.randint(5, 10)
start = self.rng.randint(0, len(rows) - n)
chunk = rows[start : start + n]
if not all(self.free(r.get("responds") or []) for r in chunk if r.get("responds")):
continue
for r in chunk:
self.mark(r.get("responds") or [], "D01", "删连续一段", "block", worded=False)
del rows[start : start + n]
return
def row_ops(self, n: int) -> None:
"""逐条响应表里的单点问题：漏行、数值不满足、写明不支持、负偏离未说明、只写满足、对象错位。"""
cands = [(t, r) for t in self.tables(explicit=True) for r in t["rows"] if self.free(r.get("responds") or [])]
self.rng.shuffle(cands)
ops = ["D01", "D03", "D03", "D05", "D06", "D08", "D09"]
done = 0
for t, r in cands:
if done >= n:
break
if r not in t["rows"] or not self.free(r.get("responds") or []):
continue
# 从轮到的类型开始，这一行做不了就换下一种（软件类项目几乎没有数值参数，D03 常常做不了）
for k in range(len(ops)):
if self._apply(ops[(done + k) % len(ops)], t, r, cands):
done += 1
break
@staticmethod
def col(t: dict[str, Any], field: str) -> int | None:
cols = t.get("columns") or ["no", "requirement", "answer", "deviation", "note"]
return cols.index(field) if field in cols else None
def _apply(self, op: str, t: dict[str, Any], r: dict[str, Any], cands: list[Any]) -> bool:
ans, dev, note, req = (self.col(t, f) for f in ("answer", "deviation", "note", "requirement"))
if ans is None:
return False
tp = r.get("tp") or {}
units = r.get("responds") or []
if op == "D01":
t["rows"].remove(r)
self.mark(units, "D01", "删一行", worded=False)
elif op == "D03":
return self._unmet(r, tp, ans, dev)
elif op == "D05":
param = tp.get("param") or (r["cells"][req][:12] if req is not None else "该功能")
r["cells"][ans] = self.phrase("D05", "phrases").format(param=param)
self.mark(units, "D05", r["cells"][ans])
elif op == "D06":
if dev is None:
return False
r["cells"][dev] = self.phrase("D06", "deviations")
if note is not None:
r["cells"][note] = ""
self.mark(units, "D06", r["cells"][dev])
elif op == "D08":
if not tp.get("value"):
return False
r["cells"][ans] = self.phrase("D08", "phrases")
self.mark(units, "D08", r["cells"][ans], worded=False)
elif op == "D09":
others = [
x
for _, x in cands
if x is not r
and (x.get("tp") or {}).get("param")
and x["cells"][x.get("anchor", ans)] != r["cells"][ans]
]
if not others:
return False
o = self.rng.choice(others)
r["cells"][ans] = o["cells"][o.get("anchor", ans)]
self.mark(units, "D09", "另一条参数的响应", worded=False)
return True
def _unmet(self, r: dict[str, Any], tp: dict[str, Any], ans: int, dev: int | None) -> bool:
"""把响应值改成不满足；一部分换成更小的单位（换算后不满足）。不带标志词。"""
op, unit = tp.get("op"), tp.get("unit") or ""
try:
v = Decimal(str(tp.get("value")))
except Exception: # noqa: BLE001
return False
if op not in (">=", "<=") or v == 0:
return False
bad = (v / 2) if op == ">=" else (v * Decimal("1.6"))
variant = "低于下限" if op == ">=" else "高于上限"
if unit in UNIT_DOWN and self.rng.random() < 0.4:
small, k = UNIT_DOWN[unit]
text, variant = f"{(bad * k).normalize():f}{small}", "换算后不满足"
else:
text = f"{bad.normalize():f}{unit}"
r["cells"][ans] = f"{tp.get('param') or ''}：{text}"
if dev is not None and self.rng.random() < 0.5:
r["cells"][dev] = "无偏离" # 一半在偏离栏仍写「无偏离」（另一半保持原样，也是无偏离或正偏离）
self.mark(r.get("responds") or [], "D03", variant, worded=False)
return True
def implicit_deviation(self, n: int) -> None:
"""只列偏离的表：挑默认接受的要求写成偏离（D06 负偏离未说明 / D05 写明不支持）。"""
for t in self.tables(explicit=False):
pool = [r for r in t.get("pool") or [] if self.free(r.get("responds") or [])]
self.rng.shuffle(pool)
ans, dev, note, req = (self.col(t, f) for f in ("answer", "deviation", "note", "requirement"))
if ans is None or dev is None:
continue
for k, r in enumerate(pool[:n]):
row = copy.deepcopy(r)
if k % 2 == 0:
row["cells"][dev] = self.phrase("D06", "deviations")
if note is not None:
row["cells"][note] = ""
self.mark(row["responds"], "D06", row["cells"][dev])
else:
param = (row.get("tp") or {}).get("param") or (
row["cells"][req][:12] if req is not None else "该功能"
)
row["cells"][ans] = self.phrase("D05", "phrases").format(param=param)
row["cells"][dev] = "负偏离"
self.mark(row["responds"], "D05", row["cells"][ans])
if t["rows"] and "—" in t["rows"][0]["cells"]:
t["rows"].clear()
if self.col(t, "no") is not None:
row["cells"][self.col(t, "no")] = str(len(t["rows"]) + 1)
t["rows"].append(row)
def attachments(self, n: int, cats: dict[str, str]) -> None:
"""删掉附件：资格条件 → D11，评分项 → D13，其他 → D02。"""
cands = [b for b in self.blocks if b.get("attach") and self.free(b.get("responds") or [])]
self.rng.shuffle(cands)
for b in cands[:n]:
units = b.get("responds") or []
kinds = {cats.get(u) for u in units}
defect = "D11" if "qualification" in kinds else "D13" if "scoring" in kinds else "D02"
self.mark(units, defect, "附件缺失", worded=False)
self.blocks.remove(b)
def compliance(self) -> None:
for t in self.tables(fixed=True):
rows = [r for r in t["rows"] if self.free(r.get("responds") or [])]
if not rows:
continue
r = self.rng.choice(rows)
at = r.get("anchor", len(r["cells"]) - 2)
r["cells"][at] = self.phrase("D10", "cells")
# 符合性条件响应表的每一行没法对到单个标注单元（整张表共用一组），按表级问题记一个
self.rule("D10", r["cells"][at] or "空着", r["cells"][1] if len(r["cells"]) > 1 else "")
return
# ---------- 规则问题 ----------
def rule(self, kind: str, variant: str, snippet: str) -> None:
self.rule_defects.append(
{"id": f"{kind}-{len(self.rule_defects) + 1}", "type": kind, "variant": variant, "t": snippet}
)
def price_table(self) -> dict[str, Any] | None:
return next((b for b in self.blocks if b.get("price_summary")), None)
def p04(self) -> None:
t = self.price_table()
if not t or len(t["rows"][0]["cells"]) < 3: # 开标一览表没有大写金额栏（S04）：做不了
return
variant = self.phrase("P04", "variants")
total = Decimal(self.spec["price"]["total"])
wrong = {"drop_digit": total / 10, "extra_shi": total + 10, "decimal_shift": total / 100}.get(
variant, total + 1
)
t["rows"][0]["cells"][2] = rmb_upper(wrong.quantize(Decimal("0.01")))
self.rule("P04", variant, t["rows"][0]["cells"][2])
def p05(self) -> None:
t = next((b for b in self.blocks if b.get("price_items")), None)
if not t or not t["rows"]:
return
r = self.rng.choice(t["rows"])
r["cells"][-1] = f"{Decimal(r['cells'][-1].replace(',', '')) + 1000:,.2f}"
self.rule("P05", "分项总价多 1000 元", r["cells"][-1])
def p06(self) -> None:
fp = self.rng.choice(self.vocab["P06"]["foreign_projects"])
self.spec["volumes"][0]["header"] = fp["name"]
self.rule("P06", "页眉留旧项目名", fp["name"])
def text_paras(self, form_hint: str) -> list[dict[str, Any]]:
return [b for b in self.blocks if b.get("type") == "para" and form_hint in str(b.get("form") or "")]
def p07(self) -> None:
bidder = self.spec["bidder"]["bidder"]
paras = [b for b in self.blocks if b.get("type") == "para" and b.get("form") and bidder in b["text"]]
if paras:
b = self.rng.choice(paras)
blank = self.phrase("P07", "blanks")
b["text"] = b["text"].replace(bidder, blank, 1)
self.rule("P07", "投标人名称空", b["text"][:24])
def p08(self) -> None:
signs = [b for b in self.blocks if b.get("type") == "sign"]
if signs:
date = self.phrase("P08", "dates")
b = self.rng.choice(signs)
b["lines"] = [re.sub(r"日期：.*", f"日期：{date}", x) for x in b["lines"]]
self.rule("P08", date, date)
def p09(self) -> None:
days = str(self.spec.get("validity_days") or 90)
for b in self.blocks:
if b.get("type") == "para" and f"{days}天" in b["text"].replace(" ", ""):
new = str(self.phrase("P09", "days"))
b["text"] = b["text"].replace(f"{days}天", f"{new}天").replace(f"{days} 天", f"{new}天")
self.rule("P09", f"{days} → {new} 天", f"{new}天")
return
def p10(self) -> None:
letter = next((i for i, b in enumerate(self.blocks) if b.get("type") == "para" and b.get("form")), None)
if letter is not None:
text = self.phrase("P10", "phrases")
self.blocks.insert(letter + 1, {"type": "para", "text": text, "form": self.blocks[letter]["form"]})
self.rule("P10", text, text)
def p12(self) -> None:
t = self.price_table()
if t:
alt = f"{Decimal(self.spec['price']['total']) * Decimal('0.97'):,.2f}"
text = self.phrase("P12", "phrases").format(alt=alt)
self.blocks.insert(self.blocks.index(t) + 1, {"type": "para", "text": text})
self.rule("P12", text, text)
def unit_modes(spec: dict[str, Any]) -> dict[str, str]:
modes: dict[str, str] = {}
for b in spec["volumes"][0]["blocks"]:
if b.get("type") == "table" and "only_deviations" in b:
mode = "implicit" if b["only_deviations"] else "explicit"
for r in [*b["rows"], *(b.get("pool") or [])]:
for u in r.get("responds") or []:
modes.setdefault(u, mode)
elif b.get("attach"):
for u in b.get("responds") or []:
modes.setdefault(u, "evidence")
elif b.get("form"):
for r in b.get("rows") or [b]:
for u in r.get("responds") or []:
modes.setdefault(u, "presence")
return modes
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("clean")
ap.add_argument("out", nargs="?")
ap.add_argument("--clean", dest="manifest_only", action="store_true", help="只给规矩的投标文件写响应清单")
ap.add_argument("--split", required=True, choices=["dev", "holdout"])
ap.add_argument("--seed", type=int, default=1)
ap.add_argument("--density", default="high", choices=["high", "low"])
a = ap.parse_args()
spec = json.loads((OUT / a.clean / "spec.json").read_text())
tender = spec["tender"]
units, _ = gold.units(tender)
cats = {u.id: u.cat for u in units}
optional = {u.id for u in units if u.optional}
modes = unit_modes(spec) # 注入前算：被删掉的行也要知道它本来该怎么响应
m = Mutator(spec, json.loads(VOCAB[a.split].read_text()), random.Random(a.seed)) # noqa: S311 评测数据，不用于安全
if not a.manifest_only:
if a.density == "high":
m.d01_block()
m.row_ops(18)
m.implicit_deviation(4)
m.attachments(4, cats)
m.compliance()
for op in (m.p04, m.p05, m.p06, m.p07, m.p08, m.p09, m.p10, m.p12):
op()
else:
m.row_ops(4)
m.attachments(1, cats)
m.p04()
m.p07()
bid = a.clean if a.manifest_only else a.out
assert bid
manifest_units: dict[str, Any] = {}
na = set(spec.get("na_units") or [])
for u in units:
mode = modes.get(u.id) or ("na" if u.id in na else "unknown")
entry: dict[str, Any] = {"cat": u.cat, "mode": mode, "optional": u.id in optional}
if u.id in m.unit_defects:
entry.update({"expect": "defect", "reviewed": False, **m.unit_defects[u.id]})
elif mode == "implicit":
entry.update({"expect": "ok", "trap": "T06"})
else:
entry["expect"] = {"na": "na", "unknown": "unknown"}.get(mode, "ok")
manifest_units[u.id] = entry
manifest = {
"bid": bid,
"tender": tender,
"style": spec["style"],
"split": a.split,
"kind": "clean" if a.manifest_only else "sloppy",
"seed": a.seed,
"density": None if a.manifest_only else a.density,
"units": manifest_units,
"rule_defects": m.rule_defects,
}
d = OUT / bid
d.mkdir(parents=True, exist_ok=True)
spec["bid"] = bid
(d / "spec.json").write_text(json.dumps(spec, ensure_ascii=False, indent=1))
(d / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=1))
n_def = sum(1 for e in manifest_units.values() if e.get("expect") == "defect")
by: dict[str, int] = {}
for e in manifest_units.values():
if e.get("expect") == "defect":
by[e["defect"]] = by.get(e["defect"], 0) + 1
print(f"{bid}：要求项问题 {n_def} 个 {by}，规则问题 {len(m.rule_defects)} 个")
if __name__ == "__main__":
main()