bid-doctor/evals/run_extract.py

"""离线抽取评测：对一份招标文件跑全部抽取任务，缓存模型原始输出，统计结果和模型输出里的不合理模式。

用法（在 server/ 下，模型配置读根目录 .env）：
  ../.venv/bin/python ../evals/run_extract.py <文件> ../evals/out/<名字> [--v extract.tech_param=2 ...] [--only extract.substantive]
  BD_AI_MODELS='{"reasoning": "deepseek-v4-pro"}' 换模型对比。
同一模型 + 同一提示词版本 + 同一输入只调用一次模型（缓存在 <输出目录>/../cache/），只改规则时重放不花钱。
输出：report.md（结果统计 + 模型输出里不合理模式的审计）、items.json（每条结果和模型原始输出）、info.json（项目基本信息和统计）、normalized.pdf。
真实招标文件和输出都不进 git（evals/out/ 已忽略）。
"""

from __future__ import annotations

import argparse
import asyncio
import dataclasses
import hashlib
import json
import pathlib
import re
import sys
from collections import Counter, defaultdict
from typing import Any
from uuid import UUID

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 provider
from bid_doctor.checklist import extract
from bid_doctor.core.db import TenantContext
from bid_doctor.document import formats
from bid_doctor.document.pipeline import convert, text_layer

ap = argparse.ArgumentParser()
ap.add_argument("file")
ap.add_argument("out")
ap.add_argument("--v", action="append", default=[], help="task=version")
ap.add_argument("--only", action="append", default=[])
args = ap.parse_args()

out = pathlib.Path(args.out)
out.mkdir(parents=True, exist_ok=True)
cache_dir = pathlib.Path(__file__).resolve().parent / "out" / "cache"  # 各次评测共用
cache_dir.mkdir(exist_ok=True)

versions = dict(v.split("=") for v in args.v)
extract.TASKS[:] = [
    dataclasses.replace(t, prompt_version=int(versions.get(t.name, t.prompt_version)))
    for t in extract.TASKS
    if not args.only or t.name in args.only
]
calls: list[dict[str, Any]] = []


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)
    from bid_doctor.core.config import get_settings

    s = get_settings()
    model = s.ai_models.get(p.capability) or s.ai_model
    tag = "" if model == "deepseek-chat" else f"{model}\n"  # 旧缓存没带模型名：deepseek-chat 保持原来的键
    key = hashlib.sha256(f"{tag}{prompt_id}\n{version}\n{p.body}\n{user}".encode()).hexdigest()[:32]
    f = cache_dir / f"{key}.json"
    if f.exists():
        data = json.loads(f.read_text())
    else:
        for attempt in range(3):
            try:
                r = await provider().complete_json(p.body, user, p.capability, task)
                break
            except Exception as e:  # noqa: BLE001
                if attempt == 2:
                    raise
                print("retry", task, e, file=sys.stderr)
                await asyncio.sleep(3)
        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))
    calls.append({"task": task, "version": version, "user": user, **data})
    if data.get("incomplete") or near_limit(data.get("tokens_out")):
        raise ai_service.OutputIncomplete(task, data["data"])  # 和线上一样：输出被截断时由 extract.run 对半重跑
    return data["data"]


extract.ai.run_json = cached_run_json  # type: ignore[attr-defined]


async def main() -> None:
    raw = pathlib.Path(args.file).read_bytes()
    det = formats.detect(raw, pathlib.Path(args.file).name)
    pdf = raw if det.format == "pdf" else await convert.to_pdf(raw, det.format)
    (out / "normalized.pdf").write_bytes(pdf)
    pages = text_layer.extract("D", pdf)
    body = text_layer.body_sentences(pages)
    doc = extract.build_doc(body)
    ctx = TenantContext(tenant_id=UUID(int=1), user_id=UUID(int=1))
    items, info, run_stats = await extract.run(ctx, UUID(int=2), doc)

    _sid = extract._sid  # noqa: SLF001
    by_sid = {_sid(s): s for s in body}
    # ---------- 结果 ----------
    zone = Counter((i.zone, i.pending_reason) for i in items)
    cats = Counter((c, i.zone) for i in items for c in i.categories)
    res = [
        {
            "task": i.task,
            "zone": i.zone,
            "reason": i.pending_reason,
            "categories": i.categories,
            "title": i.summary,
            "importance": i.importance,
            "importance_basis": i.importance_basis,
            "basis": i.basis,
            "sentences": [f"p{s.page_no}s{s.seq}" for s in i.sentences],
            "quote_pages": sorted({s.page_no for s in i.sentences}),
            "raw": i.ai_values,
        }
        for i in items
    ]
    (out / "items.json").write_text(json.dumps(res, ensure_ascii=False, indent=1))
    info_out = {
        "fields": {k: {"value": v.value, "unit": v.unit, "sentence": _sid(v.sentence)} for k, v in info.fields.items()},
        "status": info.status,
        "lots": [{k: v for k, v in lot.items() if k != "sentence"} for lot in info.lots],
        "stats": run_stats,
        "tokens": [sum(c.get("tokens_in") or 0 for c in calls), sum(c.get("tokens_out") or 0 for c in calls)],
        "models": sorted({c.get("model") or "" for c in calls}),
    }
    (out / "info.json").write_text(json.dumps(info_out, ensure_ascii=False, indent=1))

    # ---------- 模型原始输出的审计 ----------
    audit: dict[str, Counter[str]] = defaultdict(Counter)
    samples: dict[str, list[str]] = defaultdict(list)
    per_sentence: dict[str, Counter[str]] = defaultdict(Counter)

    def note(task: str, kind: str, example: str) -> None:
        audit[task][kind] += 1
        if len(samples[f"{task}|{kind}"]) < 4:
            samples[f"{task}|{kind}"].append(example)

    for c in calls:
        task = c["task"]
        chunk_ids = set(re.findall(r"^\[(p\d+s\d+)\]", c["user"], re.M))
        for it in (c["data"] or {}).get("items") or []:
            if not isinstance(it, dict):
                continue
            audit[task]["items"] += 1
            evs = [e for e in it.get("evidence") or [] if isinstance(e, dict)]
            if not evs:
                note(task, "没有证据", json.dumps(it, ensure_ascii=False)[:120])
            if len(evs) > 3:
                note(task, "证据超过 3 段", f"{len(evs)} 段：{(it.get('summary') or it.get('requirement') or it.get('factor'))}")
            for e in evs:
                sid, q = str(e.get("sid")), extract.norm(str(e.get("quote", "")))
                per_sentence[task][sid] += 1
                if sid not in chunk_ids:
                    note(task, "编号不在本分块", f"{sid} {q[:30]}")
                    continue
                s = by_sid.get(sid)
                if s is None:
                    continue
                t = extract.norm(s.text)
                if q not in t:
                    joined = "".join(extract.norm(x.text) for x in body[body.index(s): body.index(s) + 3])
                    if q and q in joined:
                        note(task, "引文跨句", f"{sid} {q[:40]}")
                    elif any(q and q in extract.norm(x.text) for x in body):
                        note(task, "编号和引文不是同一句", f"{sid} 引文「{q[:30]}」原句「{t[:30]}」")
                    else:
                        note(task, "引文不是原文", f"{sid} 引文「{q[:30]}」原句「{t[:30]}」")
                elif len(q) < min(8, len(t)):
                    note(task, "引文太短", f"{sid} 「{q}」← {t[:40]}")
                elif len(t) >= 20 and len(q) / len(t) < 0.3:
                    note(task, "引文只摘了原句一小段（<30%）", f"{sid} 「{q[:30]}」← {t[:40]}")
                if s.location == "header_footer":
                    note(task, "引了页眉页脚", sid)
                if s.key in doc.note_keys:
                    note(task, "引了标注规则说明", sid)
            basis = it.get("basis")
            if basis in ("italic_underline", "star", "asterisk"):
                ok = any((s := by_sid.get(str(e.get("sid")))) and extract.is_marked(s, doc.declared) == basis for e in evs)
                if not ok:
                    marked = "".join(by_sid[str(e.get("sid"))].marked[:50] for e in evs[:1] if str(e.get("sid")) in by_sid)
                    note(task, f"声称 {basis} 但原文没有该标注", marked)
    for task, cnt in per_sentence.items():
        multi = {sid: n for sid, n in cnt.items() if n > 1}
        if multi:
            audit[task]["同一句被拆成多条"] = len(multi)
            top = sorted(multi.items(), key=lambda kv: -kv[1])[:3]
            samples[f"{task}|同一句被拆成多条"] = [f"{sid} ×{n}：{by_sid[sid].text[:40]}" for sid, n in top if sid in by_sid]

    tokens = sum((c.get("tokens_in") or 0) for c in calls), sum((c.get("tokens_out") or 0) for c in calls)
    lines = [
        f"# {pathlib.Path(args.file).name}",
        f"页数 {len(pages)}，正文句子 {len(body)}，分块 {len(extract.chunks(body, doc.continuations))}，模型调用 {len(calls)} 次，tokens {tokens[0]} / {tokens[1]}",
        f"版本：{ {t.name: t.prompt_version for t in extract.TASKS} }",
        "",
        "## 结果",
        f"条目 {len(items)}：" + "，".join(f"{z}/{r or '-'} {n}" for (z, r), n in zone.most_common()),
        "按类别：" + "，".join(f"{c}/{z}:{n}" for (c, z), n in sorted(cats.items())),
        "项目信息：" + "，".join(f"{k}={getattr(v, 'value', v)}" for k, v in info.fields.items())[:600],
        "项目信息结论：" + "，".join(f"{k}:{v['status']}{v.get('codes') or ''}" for k, v in info.status.items()),
        "对半重跑 {} 次，拆到底仍不完整 {} 块".format(run_stats.get("chunks_split", 0), run_stats.get("output_incomplete", 0)),
        "没通过校验的原因：" + "，".join(f"{k.removeprefix('fail:')} {n}" for k, n in sorted(run_stats.items()) if k.startswith("fail:")),
        "",
        "## 模型输出审计",
    ]
    for task, cnt in audit.items():
        lines.append(f"### {task}（{cnt['items']} 条）")
        for k, n in cnt.most_common():
            if k == "items":
                continue
            lines.append(f"- {k}：{n}")
            for ex in samples.get(f"{task}|{k}", []):
                lines.append(f"    - {ex}")
    report = "\n".join(lines)
    (out / "report.md").write_text(report)
    print(report)


asyncio.run(main())