47c6a1258e
- check/presence.py：「提供《授权委托书》」「须提交投标函」这类要求按标题里有没有这个文件名判断，不调模型。
- 扫描件兜底：要附证明材料的要求没找到、投标文件里有扫描件页时，改为提示级「可能在扫描件里，请人工核对」。
- 响应方式解析时识别、带依据句子存进 checklist.response_modes（迁移 0008）；清单页显示、可改（写审计，
重新解析不覆盖人工改过的）；查标用存下来的。新接口 PUT /projects/{id}/checklist/response-modes/{category}。
- check_item_result.layer 加 presence（迁移 0007）；契约同步默认接受的两种结论。
开发集 17 份：规矩投标文件的致命问题 120 → 51，致命级召回不变（22 / 32）（docs/architecture/check-eval.md §5）。
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
209 lines
8.7 KiB
Python
209 lines
8.7 KiB
Python
"""离线查标评测：拿读标评测的清单（evals/out/runs/<运行>/<招标文件>/items.json）对一份虚构投标文件跑响应匹配。
|
||
|
||
用法（在 server/ 下，模型配置读根目录 .env）：
|
||
../.venv/bin/python ../evals/check/run_check.py <投标文件编号> [--checklist fix3] [--name <运行名>]
|
||
输入：evals/out/check/<投标文件编号>/v*.pdf（render.py 渲染的）、manifest.json 里的 tender。
|
||
输出：evals/out/check/<投标文件编号>/runs/<运行名>/results.json（每条清单项的结论、投标侧证据、生成的问题）。
|
||
同一模型 + 同一提示词版本 + 同一输入只调用一次模型（和读标评测共用 evals/out/cache/），只改规则时重放不花钱。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import hashlib
|
||
import json
|
||
import pathlib
|
||
import re
|
||
import sys
|
||
from typing import Any
|
||
from uuid import UUID, uuid4
|
||
|
||
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.checklist import response_mode
|
||
from bid_doctor.core.config import get_settings
|
||
from bid_doctor.core.db import TenantContext
|
||
from bid_doctor.document.pipeline import text_layer
|
||
|
||
EVALS = pathlib.Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(EVALS))
|
||
import gold # noqa: E402
|
||
|
||
OUT = EVALS / "out" / "check"
|
||
CACHE = EVALS / "out" / "cache"
|
||
usage = {"calls": 0, "cached": 0, "tokens_in": 0, "tokens_out": 0}
|
||
|
||
|
||
async def cached_run_json(
|
||
ctx: Any, *, job_type: str, job_id: Any, task: str, prompt_id: str, version: int, user: str
|
||
) -> dict[str, Any]:
|
||
del ctx, job_type, job_id
|
||
p = prompts.get(prompt_id, version)
|
||
s = get_settings()
|
||
model = s.ai_models.get(p.capability) or s.ai_model
|
||
key = hashlib.sha256(f"{model}\n{prompt_id}\n{version}\n{p.body}\n{user}".encode()).hexdigest()[:32]
|
||
f = CACHE / f"{key}.json"
|
||
if f.exists():
|
||
data = json.loads(f.read_text())
|
||
usage["cached"] += 1
|
||
else:
|
||
r = await provider().complete_json(p.body, user, p.capability, task)
|
||
data = {
|
||
"data": r.data,
|
||
"model": r.model,
|
||
"tokens_in": r.tokens_in,
|
||
"tokens_out": r.tokens_out,
|
||
"incomplete": r.incomplete,
|
||
}
|
||
f.write_text(json.dumps(data, ensure_ascii=False))
|
||
usage["calls"] += 1
|
||
usage["tokens_in"] += data.get("tokens_in") or 0
|
||
usage["tokens_out"] += data.get("tokens_out") or 0
|
||
if data.get("incomplete") or near_limit(data.get("tokens_out")):
|
||
raise OutputIncomplete(task, data["data"])
|
||
return data["data"]
|
||
|
||
|
||
ITEM = re.compile(r"^\[(I\d+)\] ")
|
||
CANDS = re.compile(r"^候选句：(.*)$")
|
||
FRAG = re.compile(r"^\[(d\d+p\d+s\d+)\]（[^）]*）(.*)$")
|
||
|
||
|
||
async def baseline_run_json(
|
||
ctx: Any, *, job_type: str, job_id: Any, task: str, prompt_id: str, version: int, user: str
|
||
) -> dict[str, Any]:
|
||
"""只检索基线（GUIDELINES §7）：不调模型，检索第一名就当作已响应。"""
|
||
del ctx, job_type, job_id, task, prompt_id, version
|
||
frags = {m.group(1): m.group(2) for line in user.splitlines() if (m := FRAG.match(line.strip()))}
|
||
results, key = [], None
|
||
for line in user.splitlines():
|
||
if m := ITEM.match(line.strip()):
|
||
key = m.group(1)
|
||
elif (m := CANDS.match(line.strip())) and key:
|
||
first = m.group(1).split("、")[0]
|
||
results.append(
|
||
{"item": key, "verdict": "responded", "evidence": [{"sid": first, "quote": frags.get(first, "")}]}
|
||
)
|
||
return {"results": results}
|
||
|
||
|
||
def checklist(tender: str, run: str) -> dict[UUID, dict[str, Any]]:
|
||
"""读标评测输出的正式清单 → 查标的清单项（和检查快照的 payload 同样的字段）。"""
|
||
rows = json.loads((EVALS / "out" / "runs" / run / tender / "items.json").read_text())
|
||
by_sid = {gold.sid(s): s for s in gold.sentences(tender)}
|
||
out: dict[UUID, dict[str, Any]] = {}
|
||
for i, r in enumerate(rows):
|
||
if r.get("zone") != "formal":
|
||
continue
|
||
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 {}
|
||
out[UUID(int=i + 1)] = {
|
||
"title": r.get("title"),
|
||
"requirement_text": text or None,
|
||
"categories": r.get("categories") or [],
|
||
"importance": r.get("importance"),
|
||
"importance_basis": r.get("importance_basis"),
|
||
"spec": spec,
|
||
"sentences": r.get("sentences") or [],
|
||
}
|
||
return out
|
||
|
||
|
||
async def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("bid")
|
||
ap.add_argument("--checklist", default="fix3")
|
||
ap.add_argument("--name", default="base")
|
||
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 之前的做法）")
|
||
args = ap.parse_args()
|
||
d = OUT / args.bid
|
||
manifest = json.loads((d / "manifest.json").read_text())
|
||
items = checklist(manifest["tender"], args.checklist)
|
||
docs = []
|
||
for n, pdf in enumerate(sorted(d.glob("v*.pdf")), start=1):
|
||
pages = text_layer.extract(f"{args.bid}-{pdf.stem}", pdf.read_bytes())
|
||
docs.append(
|
||
match.BidDoc(
|
||
uuid4(),
|
||
n,
|
||
pdf.stem,
|
||
text_layer.body_sentences(pages),
|
||
[p.page_no for p in pages if p.text_source == "empty"],
|
||
)
|
||
)
|
||
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)
|
||
match.ROW_MATES = 0
|
||
|
||
async def progress(_: float) -> None:
|
||
return None
|
||
|
||
ctx = TenantContext(tenant_id=UUID(int=0), user_id=UUID(int=0))
|
||
all_items = items
|
||
decisions: dict[UUID, applicability.Decision] = {}
|
||
if not args.no_modes:
|
||
modes = response_mode.detect(gold.sentences(manifest["tender"]))
|
||
decisions = {i: applicability.decide(p, modes) for i, p in items.items()}
|
||
items = {i: p for i, p in items.items() if decisions[i].match}
|
||
results, stats = await match.run(ctx, uuid4(), items, docs, progress)
|
||
if decisions:
|
||
match.apply_modes(results, items, decisions)
|
||
match.scanned_fallback(results, items, decisions, docs)
|
||
rows = []
|
||
for item_id, item in all_items.items():
|
||
if item_id not in items:
|
||
rows.append(
|
||
{
|
||
"title": item["title"],
|
||
"sentences": item["sentences"],
|
||
"categories": item["categories"],
|
||
"importance": item["importance"],
|
||
"verdict": f"skipped:{decisions[item_id].skip}",
|
||
"layer": None,
|
||
"evidence_status": "none",
|
||
"reason_code": None,
|
||
"evidence": [],
|
||
"detail": {},
|
||
"finding": None,
|
||
}
|
||
)
|
||
continue
|
||
res = results[item_id]
|
||
spec = findings.spec(res, item)
|
||
rows.append(
|
||
{
|
||
"title": item["title"],
|
||
"sentences": item["sentences"],
|
||
"categories": item["categories"],
|
||
"importance": item["importance"],
|
||
"verdict": res.verdict,
|
||
"layer": res.layer,
|
||
"evidence_status": res.evidence_status,
|
||
"reason_code": res.reason_code,
|
||
"evidence": [{"volume": doc.label, "p": s.page_no, "text": s.text} for doc, s in res.evidence],
|
||
"detail": res.detail,
|
||
"finding": {"type": spec.type, "severity": spec.severity, "title": spec.title} if spec else 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))
|
||
findings_n = sum(r["finding"] is not None for r in rows)
|
||
print(
|
||
f"{args.bid}：清单 {len(rows)} 条，问题 {findings_n} 个，模型调用 {usage['calls']} 次（缓存 {usage['cached']}），"
|
||
f"token {usage['tokens_in']} / {usage['tokens_out']}"
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|