feat(check): 规则检查(报价形式、形式要件、响应表),表格按单元格分行

- check/rules.py(规则集 2026.11-r1,不调模型):大小写金额、分项合计、超最高限价、
  多个报价、别的项目名称、没填的地方、落款日期、投标有效期、附加条件、响应表里的未响应;
  每条挂投标原文,有依据的挂招标侧出处;缺限价、截止时间、有效期要求时写进覆盖范围
- 文字层的句子带上单元格框,表格按单元格上沿分行
- 按 Fable 审读意见收紧致命级:认不准的降为重要或提示
- 评测:score.py 评规则问题;P07 注入记录改为空白处前后原文,剔除注入到人员表的 D10
- 上传页的检查范围文案同步

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
YANG JIANKUAN
2026-11-04 16:44:22 +08:00
parent 47c6a1258e
commit db1b1f6f9f
17 changed files with 1792 additions and 20 deletions

View File

@@ -334,6 +334,8 @@ class Mutator:
def compliance(self) -> None:
for t in self.tables(fixed=True):
if not any("响应" in str(h) for h in t.get("header") or []):
continue # 只有响应表(表头有「响应情况」栏)才注入;人员表、业绩表不是
rows = [r for r in t["rows"] if self.free(r.get("responds") or [])]
if not rows:
continue
@@ -388,7 +390,7 @@ class Mutator:
b = self.rng.choice(paras)
blank = self.phrase("P07", "blanks")
b["text"] = b["text"].replace(bidder, blank, 1)
self.rule("P07", "投标人名称空", b["text"][:24])
self.rule("P07", "投标人名称空", blank_context(b["text"], blank))
def p08(self) -> None:
signs = [b for b in self.blocks if b.get("type") == "sign"]
@@ -423,6 +425,13 @@ class Mutator:
self.rule("P12", text, text)
def blank_context(text: str, blank: str) -> str:
"""空白处前后的原文(不跨句号),评分时拿它对照问题的投标侧出处。"""
i = text.index(blank)
start = max(i - 10, max(text.rfind(x, 0, i) for x in "。;") + 1)
return text[start : i + len(blank) + 10]
def table_mode(b: dict[str, Any]) -> str:
"""表格的响应方式:档案里表单写了 mode 的以它为准(如 S02 评分要的技术规格表是 presence,不是逐条响应的机制)。"""
return str(b.get("mode") or ("implicit" if b.get("only_deviations") else "explicit"))
@@ -473,7 +482,7 @@ def main() -> None:
for f in OUT.glob(f"{tender}-*/manifest.json"):
for uid, e in json.loads(f.read_text())["units"].items():
if (e.get("review") or {}).get("ok") is False:
m.rejected.add(uid) # noqa: S311 评测数据,不用于安全
m.rejected.add(uid)
if not a.manifest_only:
if a.density == "high":
m.d01_block()

View File

@@ -30,6 +30,10 @@ def main() -> None:
traps: Counter[str] = Counter()
tokens = [0, 0]
findings_clean = items_clean = 0
r_found: Counter[str] = Counter()
r_total: Counter[str] = Counter()
r_fp: Counter[str] = Counter()
r_fp_clean = r_clean = 0
for bid in a.bids:
d = OUT / bid
m = json.loads((d / "manifest.json").read_text())
@@ -43,6 +47,13 @@ def main() -> None:
if "查标" in s["recall"]:
by_style[m["style"]] += s["recall"]["查标"][0]
by_style_tot[m["style"]] += s["recall"]["查标"][1]
rules = s.get("rules") or {}
r_found.update(rules.get("found") or {})
r_total.update(rules.get("total") or {})
r_fp.update(rules.get("fp_by_code") or {})
if m["kind"] == "clean":
r_fp_clean += rules.get("fp", 0)
r_clean += 1
if m["kind"] == "clean":
fp += s["fp"]
ok += s["ok_units"]
@@ -68,6 +79,15 @@ def main() -> None:
f"- 规矩的投标文件里,不需要响应(na / unknown)的要求上报了问题 {other} 个,对不上标注单元的问题 {unmapped} 个",
f"- 干扰项误报:{dict(traps)}",
f"- token:输入 {tokens[0]:,},输出 {tokens[1]:,}({len(a.bids)} 次检查)",
"",
f"## 规则问题:查出 {pct(sum(r_found.values()), sum(r_total.values()))};误报 {sum(r_fp.values())} 个"
f"(其中规矩的投标文件 {r_clean} 份上 {r_fp_clean} 个)",
"",
"| 问题 | 查出 |",
"|---|---|",
*[f"| {k} | {pct(r_found[k], r_total[k])} |" for k in sorted(r_total)],
"",
f"- 误报按规则:{dict(r_fp)}",
]
print("\n".join(lines))

View File

@@ -11,11 +11,13 @@ from __future__ import annotations
import argparse
import asyncio
import datetime as dt
import hashlib
import json
import pathlib
import re
import sys
from decimal import Decimal
from typing import Any
from uuid import UUID, uuid4
@@ -23,8 +25,9 @@ from bid_doctor.ai import prompts
from bid_doctor.ai import service as ai_service
from bid_doctor.ai.providers.base import near_limit
from bid_doctor.ai.service import OutputIncomplete, provider
from bid_doctor.check import applicability, findings, match
from bid_doctor.check import applicability, findings, match, rules
from bid_doctor.checklist import response_mode
from bid_doctor.checklist.extract import amount_yuan
from bid_doctor.core.config import get_settings
from bid_doctor.core.db import TenantContext
from bid_doctor.document.pipeline import text_layer
@@ -102,6 +105,9 @@ def checklist(tender: str, run: str) -> dict[UUID, dict[str, Any]]:
text = "".join(by_sid[k].text for k in r.get("sentences") or [] if k in by_sid)
raw = (r.get("raw") or {}).get("tech_param") or {}
spec = {"tech_param": {k: raw[k] for k in ("param", "op", "value", "unit") if raw.get(k)}} if raw else {}
sched = (r.get("raw") or {}).get("schedule") or {}
if sched:
spec["schedule"] = {k: sched[k] for k in ("date_kind", "at") if sched.get(k)}
out[UUID(int=i + 1)] = {
"title": r.get("title"),
"requirement_text": text or None,
@@ -114,6 +120,53 @@ def checklist(tender: str, run: str) -> dict[UUID, dict[str, Any]]:
return out
def facts(tender: str, run: str, items: dict[UUID, dict[str, Any]]) -> rules.Facts:
"""项目侧的值:读标评测抽出的项目基本信息(和产品里一样来自抽取);所投的包按招标档案(相当于用户选包)。"""
base = EVALS / "out" / "runs" / run / tender
info = json.loads((base / "info.json").read_text())
f = {k: v.get("value") for k, v in info["fields"].items()}
profile = json.loads((OUT / "tenders" / f"{tender}.json").read_text())["project"]
deadline = next(
(
dt.date.fromisoformat(s["at"][:10])
for r in json.loads((base / "items.json").read_text())
if r.get("zone") == "formal"
and (s := (r.get("raw") or {}).get("schedule") or {}).get("date_kind") == "bid_deadline"
and s.get("at")
),
None,
)
lots = info.get("lots") or []
ceiling, scope, note = None, "total", None
if lots:
lot = next(
(x for x in lots if str(x["lot_no"]).lstrip("0") == str(profile.get("bidding_lot", "")).lstrip("0")), None
)
if lot is None and len(lots) == 1:
lot = lots[0]
if lot is not None:
c = amount_yuan((lot.get("ceiling") or {}).get("value"), (lot.get("ceiling") or {}).get("unit"))
ceiling, scope = (Decimal(c) if c else None), "lot" if len(lots) > 1 else "total"
else:
note = "所投采购包不明确,报价没有和限价比对"
else:
c = amount_yuan(f.get("ceiling_total"), info["fields"].get("ceiling_total", {}).get("unit"))
ceiling = Decimal(c) if c else None
days, item = rules.required_validity(items)
return rules.Facts(
name=f.get("name"),
code=f.get("code"),
purchaser=f.get("purchaser"),
agency=f.get("agency"),
deadline=deadline,
ceiling=ceiling,
ceiling_scope=scope,
ceiling_note=note,
validity_days=days,
validity_item=item,
)
async def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("bid")
@@ -122,6 +175,7 @@ async def main() -> None:
ap.add_argument("--baseline", action="store_true", help="只检索、不调模型的基线")
ap.add_argument("--v1", action="store_true", help="第 1 步的做法:提示词 v1、不带同一行的单元格(对比基线)")
ap.add_argument("--no-modes", action="store_true", help="不按响应方式和「要不要回答」筛选(C12 之前的做法)")
ap.add_argument("--rules-only", action="store_true", help="只重跑规则检查,写回已有运行的 results.json")
args = ap.parse_args()
d = OUT / args.bid
manifest = json.loads((d / "manifest.json").read_text())
@@ -136,8 +190,39 @@ async def main() -> None:
pdf.stem,
text_layer.body_sentences(pages),
[p.page_no for p in pages if p.text_source == "empty"],
[s for p in pages for s in p.sentences if s.location == "header_footer"],
)
)
hits, rules_cov = rules.run(
facts(manifest["tender"], args.checklist, items), docs, gold.sentences(manifest["tender"])
)
def rule_rows(covered: set[tuple[str, int, str]]) -> list[dict[str, Any]]:
return [
{
"code": h.code,
"type": h.type,
"severity": h.severity,
"title": h.title,
"explanation": h.explanation,
"evidence": [{"volume": doc.label, "p": s.page_no, "text": s.text} for doc, s in h.evidence],
}
for h in rules.drop_covered(hits, covered)
]
if args.rules_only:
prev = d / "runs" / args.name / "results.json"
data = json.loads(prev.read_text())
covered = {
(e["volume"], e["p"], "".join(e["text"].split()))
for it in data["items"]
if it["verdict"] == "negative_deviation"
for e in it["evidence"]
}
data["rules"], data["rules_coverage"] = rule_rows(covered), rules_cov
prev.write_text(json.dumps(data, ensure_ascii=False, indent=1))
print(f"{args.bid}:规则问题 {len(data['rules'])} 个")
return
ai_service.run_json = baseline_run_json if args.baseline else cached_run_json # type: ignore[assignment]
if args.v1:
match.PROMPT = ("check.match", 1)
@@ -196,7 +281,19 @@ async def main() -> None:
out = d / "runs" / args.name
out.mkdir(parents=True, exist_ok=True)
summary = {"items": len(rows), "usage": usage, "verify_failures": stats.verify_failures, "splits": stats.splits}
(out / "results.json").write_text(json.dumps({"summary": summary, "items": rows}, ensure_ascii=False, indent=1))
covered = {
(doc.label, s.page_no, "".join(s.text.split()))
for res in results.values()
if res.verdict == "negative_deviation"
for doc, s in res.evidence
}
(out / "results.json").write_text(
json.dumps(
{"summary": summary, "items": rows, "rules": rule_rows(covered), "rules_coverage": rules_cov},
ensure_ascii=False,
indent=1,
)
)
findings_n = sum(r["finding"] is not None for r in rows)
print(
f"{args.bid}:清单 {len(rows)} 条,问题 {findings_n} 个,模型调用 {usage['calls']} 次(缓存 {usage['cached']}),"

View File

@@ -44,6 +44,83 @@ FAMILY = {
"D15": {"qualification", "unresponded"},
}
FATAL_CATS = {"substantive", "qualification"}
RULE_CODES = {
"P01": "PRICE.OVER_CEILING_TOTAL",
"P02": "PRICE.OVER_CEILING_LOT",
"P03": "PRICE.OVER_CEILING_ITEM",
"P04": "PRICE.CAPITAL_MISMATCH",
"P05": "PRICE.SUM_MISMATCH",
"P06": "FORM.FOREIGN_PROJECT",
"P07": "FORM.BLANK_FIELD",
"P08": "FORM.DATE_AFTER_DEADLINE",
"P09": "FORM.VALIDITY_SHORT",
"P10": "FORM.CONDITION_ADDED",
"P11": "FORM.MISSING_DOC",
"P12": "PRICE.MULTIPLE_QUOTES",
"P13": "PRICE.ITEM_MISSING",
}
def squash(t: str) -> str:
return "".join(t.split())
def score_rules(manifest: dict[str, Any], res: dict[str, Any]) -> dict[str, Any]:
"""规则问题(P 类):规则编号对上、投标侧出处里有注入的那段原文才算查出;没对上任何注入问题的规则问题算误报。
D10(符合性条件响应表)按表级问题记在 rule_defects 里,它由响应匹配查(negative_deviation),这里一起对照。"""
hits = list(res.get("rules") or [])
used: set[int] = set()
found: Counter[str] = Counter()
total: Counter[str] = Counter()
missed: list[str] = []
for rd in manifest.get("rule_defects") or []:
kind, snip = rd["type"], squash(rd.get("t") or "")
total[kind] += 1
if kind == "D10":
pool = [
(-1, {"evidence": it["evidence"]})
for it in res["items"]
if it["finding"] and it["finding"]["type"] == "negative_deviation"
] + [(i, h) for i, h in enumerate(hits) if h["code"] == "FORM.RESPONSE_NEGATIVE"]
else:
pool = [(i, h) for i, h in enumerate(hits) if h["code"] == RULE_CODES.get(kind)]
key = snip[:12]
cell = squash(rd.get("variant") or "") if kind == "D10" else ""
ok = next(
(
i
for i, h in pool
if any(
(key and (key in squash(e["text"]) or squash(e["text"]) in snip))
or (cell and squash(e["text"]) == cell)
for e in h["evidence"]
)
),
None,
)
if ok is None:
missed.append(f"{rd['id']}({rd.get('variant', '')}):{rd.get('t', '')[:40]}")
continue
found[kind] += 1
if ok >= 0:
used.add(ok)
# 响应表规则报的「负偏离」对上了响应问题(D05 / D06)的注入位置:算那个单元查出(rule_units),不算规则误报
rule_units: set[str] = set()
for i, h in enumerate(hits):
if h["code"] != "FORM.RESPONSE_NEGATIVE":
continue
texts = [squash(e["text"]) for e in h["evidence"]]
for uid, u in manifest["units"].items():
if u.get("expect") != "defect" or "negative_deviation" not in FAMILY.get(u.get("defect", ""), set()):
continue
if any(
(w := squash(x.get("t") or "")[:12]) and any(w in t or (t and t in w) for t in texts)
for x in u.get("where") or []
):
rule_units.add(uid)
used.add(i)
fp = [h for i, h in enumerate(hits) if i not in used]
return {"found": found, "total": total, "missed": missed, "fp": fp, "n": len(hits), "units": rule_units}
def pct(a: int, b: int) -> str:
@@ -79,6 +156,7 @@ def main() -> None:
parent[find(o)] = find(first)
group = {u.id: find(u.id) for u in units}
rs = score_rules(manifest, res)
per_unit: dict[str, list[dict[str, Any]]] = defaultdict(list)
unmapped: list[dict[str, Any]] = []
for it in res["items"]:
@@ -120,6 +198,9 @@ def main() -> None:
continue
rec_total["查标"] += 1
good = [it for it in found if it["finding"]["type"] in FAMILY.get(e["defect"], set())]
if not good and any(x in rs["units"] for x in manifest["units"] if group.get(x, x) == g):
good = [{"finding": {"type": "negative_deviation", "severity": "important"}}]
rec["其中由响应表规则查出"] += 1
if good:
for k in [*keys, "端到端", "查标"]:
rec[k] += 1
@@ -182,7 +263,23 @@ def main() -> None:
*[f"- {x}" for x in fp[:80]],
*(["- ……"] if len(fp) > 80 else []),
"",
f"## 规则问题(第 2 步才查,这里只列出):{len(manifest['rule_defects'])} 个",
]
lines += [
f"## 规则问题:注入 {sum(rs['total'].values())} 个,查出 {sum(rs['found'].values())} 个;规则报出 {rs['n']} 个,误报 {len(rs['fp'])} 个",
"",
"| 问题 | 查出 |",
"|---|---|",
*[f"| {k} | {pct(rs['found'][k], rs['total'][k])} |" for k in sorted(rs["total"])],
"",
*[f"- 漏:{x}" for x in rs["missed"]],
*[
f"- 误报:{h['code']} {h['title']}:{h['evidence'][0]['text'][:50] if h['evidence'] else ''}"
for h in rs["fp"]
],
"",
"### 规则检查没覆盖的",
"",
*[f"- {x}" for x in (res.get("rules_coverage") or {}).get("not_checked", [])],
"",
]
out = d / "runs" / a.run
@@ -196,6 +293,13 @@ def main() -> None:
"other_findings": other_findings,
"extraction_miss": len(extraction_miss),
"include_unreviewed": a.include_unreviewed,
"rules": {
"found": dict(rs["found"]),
"total": dict(rs["total"]),
"fp": len(rs["fp"]),
"fp_by_code": dict(Counter(h["code"] for h in rs["fp"])),
"n": rs["n"],
},
}
(out / "score.json").write_text(json.dumps(summary, ensure_ascii=False, indent=1))
print("\n".join(lines[:20]))