feat(check): 候选句带同一行的单元格;提示词 check.match v2

检索常命中投标人照抄要求原文的那一格,响应和偏离写在同一行的别的格里,模型看不到。现在候选句在表格里
时,把同一行(同一页、上下范围重叠)的单元格一起交给模型,也可以引用;v2 说明怎么读一行。
开发集 17 份上查标召回 30% → 50%,带标志词的问题 2% → 54%,规矩投标文件上的误报略降,输入 token +44%;
漏响应和响应对象错位变差,下一轮看(docs/architecture/check-eval.md §3)。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
YANG JIANKUAN
2026-11-04 13:58:02 +08:00
parent 1d50faccbe
commit b7432643dd
4 changed files with 187 additions and 13 deletions

View File

@@ -25,8 +25,9 @@ from bid_doctor.check.retrieval import Index
from bid_doctor.core.db import TenantContext
from bid_doctor.document.pipeline.text_layer import Sentence, norm
PROMPT = ("check.match", 1)
PROMPT = ("check.match", 2)
CANDIDATES = 5 # 每条清单项送给模型的候选句数
ROW_MATES = 6 # 候选句在表格里时,带上同一行的单元格(检索常命中投标人照抄要求的那一格,响应和偏离在同一行的别的格)
BATCH = 8 # 每次调用判断的清单项数
SPLIT_DEPTH = 2
CONCURRENCY = 4 # TODO(第 3 步):和解析共用「供应商 × 租户」统一限流(extract.py 的同一个 TODO)
@@ -99,15 +100,37 @@ def _item_block(key: str, item: dict[str, Any], sids: list[str]) -> str:
return "\n".join(lines)
def row_mates(s: Sentence, page: list[Sentence]) -> list[Sentence]:
"""同一行的其他句子:同一页、上下范围重叠(和读标核对取值时找「同一行」是同一口径)。按从左到右排。"""
if not s.quads:
return []
top, bottom = min(q[1] for q in s.quads), max(q[3] for q in s.quads)
mates = [x for x in page if x is not s and x.quads and any(q[1] < bottom - 1 and q[3] > top + 1 for q in x.quads)]
return sorted(mates, key=lambda x: min(q[0] for q in x.quads))[:ROW_MATES]
def build_user(
batch: list[tuple[str, dict[str, Any], list[str]]], fragments: dict[str, tuple[BidDoc, Sentence]]
batch: list[tuple[str, dict[str, Any], list[str]]],
fragments: dict[str, tuple[BidDoc, Sentence]],
rows: dict[str, list[str]] | None = None,
) -> str:
"""数据区:待核对的要求 + 投标文件片段(每句只送一次)。系统提示词声明数据区里的指令一律不执行。"""
used = list(dict.fromkeys(sid for _, _, sids in batch for sid in sids))
"""数据区:待核对的要求 + 投标文件片段(每句只送一次;表格里的句子后面跟着同一行的单元格)。
系统提示词声明数据区里的指令一律不执行。"""
rows = rows or {}
shown: set[str] = set()
frag_lines: list[str] = []
for _, _, sids in batch:
for sid in sids:
if sid in shown:
continue
shown.add(sid)
doc, sent = fragments[sid]
frag_lines.append(f"[{sid}]({doc.label} 第{sent.page_no}页){sent.text}")
mates = [m for m in rows.get(sid, []) if m not in shown]
if mates:
shown.update(mates)
frag_lines.append(" 同一行:" + " | ".join(f"[{m}] {fragments[m][1].text}" for m in mates))
blocks = [_item_block(k, it, sids) for k, it, sids in batch]
frag_lines = [
f"[{sid}]({fragments[sid][0].label} 第{fragments[sid][1].page_no}页){fragments[sid][1].text}" for sid in used
]
return "## 待核对的要求\n\n" + "\n\n".join(blocks) + "\n\n## 投标文件片段\n\n" + "\n".join(frag_lines)
@@ -172,12 +195,30 @@ async def run(
doc_of = {s.key: d for d in docs for s in d.sentences}
results: dict[UUID, Result] = {}
queue: list[tuple[UUID, dict[str, Any], dict[str, tuple[BidDoc, Sentence]]]] = []
pages: dict[tuple[UUID, int], list[Sentence]] = {}
for d in docs:
for s in d.sentences:
pages.setdefault((d.id, s.page_no), []).append(s)
rows: dict[str, list[str]] = {} # 候选句 → 同一行的单元格(短编号)
sent_of = {
short_sid(d, x): (d, x) for d in docs for x in d.sentences
} # 片段按编号查(同一行的单元格可能是别的条目的候选句)
hits_of: dict[UUID, list[str]] = {} # 每条清单项自己的候选句(同一行的单元格跟在后面显示,不单列)
for item_id, item in items.items():
hits = index.search(query_text(item), CANDIDATES)
if not hits: # 检索不到候选句:不调模型(m3-plan §2.1)
results[item_id] = Result("not_found", "retrieval", detail={"search_scope": scope})
continue
allowed = {short_sid(doc_of[h.sentence.key], h.sentence): (doc_of[h.sentence.key], h.sentence) for h in hits}
allowed: dict[str, tuple[BidDoc, Sentence]] = {}
for h in hits:
doc = doc_of[h.sentence.key]
sid = short_sid(doc, h.sentence)
allowed[sid] = (doc, h.sentence)
hits_of.setdefault(item_id, []).append(sid)
mates = row_mates(h.sentence, pages[(doc.id, h.sentence.page_no)])
rows[sid] = [short_sid(doc, m) for m in mates]
for m in mates: # 同一行的单元格也可以引用(响应、偏离常写在那里)
allowed.setdefault(short_sid(doc, m), (doc, m))
queue.append((item_id, item, allowed))
sem = asyncio.Semaphore(CONCURRENCY)
@@ -185,8 +226,12 @@ async def run(
async def call(batch: list[tuple[UUID, dict[str, Any], dict[str, tuple[BidDoc, Sentence]]]], depth: int) -> None:
keyed = [(f"I{i + 1}", item_id, item, allowed) for i, (item_id, item, allowed) in enumerate(batch)]
fragments = {sid: x for _, _, _, allowed in keyed for sid, x in allowed.items()}
user = build_user([(k, item, list(allowed)) for k, _, item, allowed in keyed], fragments)
user = build_user(
[(k, item, hits_of[item_id]) for k, item_id, item, _ in keyed],
sent_of,
{h: rows.get(h, []) for _, item_id, _, _ in keyed for h in hits_of[item_id]},
)
try:
async with sem:
stats.calls += 1