feat: 读标补齐五类清单与项目基本信息,修复国动样本暴露的 4 个问题
- 文字层 v2:页眉页脚带单独成句;写到页尾的句子和下一页首句接续(出处两页各一条) - 结构识别(最小版):十进制编号、冒号引出的列表找上级条款;上级条款写明无效的,条目照样认定 - 抽取:资格、评分、日期与递交、技术参数、项目基本信息各一份提示词;实质性条款出 v2 (认上级条款、补「做无效响应处理」等写法);各类取值核对;按首句合并; 拆开被合成一条的列举条款;描述标注方式的句子和表头碎片不当条款;* 号标注 - 项目:解析出的名称、编号、采购人、预算、截止时间、分包写回项目,用户改过的不覆盖 - 前端:清单项显示分值、参数要求、日期、证明材料 - 测试 40 个,新增两页虚构样例 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -25,4 +25,4 @@ class TransientError(Exception):
|
||||
class Provider(Protocol):
|
||||
name: str
|
||||
|
||||
async def complete_json(self, system: str, user: str, capability: str) -> Completion: ...
|
||||
async def complete_json(self, system: str, user: str, capability: str, task: str) -> Completion: ...
|
||||
|
||||
@@ -1,40 +1,165 @@
|
||||
"""确定性的假模型:开发和测试用,CI 不调用真实模型(架构 §8)。
|
||||
|
||||
只认 extract.substantive:把带 ★ 或 <iu> 的句子当作实质性条款,逐字摘一段原文作为引文。
|
||||
按任务用关键词规则给出结构化结果,输出格式和真实模型一致,好让三道校验、合并、写库这些环节在测试里都跑到。
|
||||
它不理解文字,召回和准确率都没有参考价值。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
from bid_doctor.ai.providers.base import Completion
|
||||
|
||||
LINE = re.compile(r"^\[([^\]]+)\] (.*)$")
|
||||
TAG = re.compile(r"</?(iu|u|i)>")
|
||||
PARENT = re.compile(r"〔上级:(.*)〕$")
|
||||
INVALID = re.compile(r"无效|否决|未实质性响应")
|
||||
DATE = re.compile(r"(\d{4})年(\d{1,2})月(\d{1,2})日\s*(\d{1,2})[时::](\d{2})?")
|
||||
NUM_REQ = re.compile(
|
||||
r"(不低于|不少于|≥|不小于|不大于|不超过|≤)\s*(\d+(?:\.\d+)?)\s*([A-Za-z%]+|[个台套项条次人天年月日分秒])?"
|
||||
)
|
||||
POINTS = re.compile(r"[((](\d+(?:\.\d+)?)\s*分[))]")
|
||||
|
||||
|
||||
def _lines(user: str) -> list[tuple[str, str, str, str]]:
|
||||
"""(句子编号, 带格式标记的文本, 纯文本, 上级条款)"""
|
||||
out = []
|
||||
for raw in user.split("## 待审分块", 1)[-1].splitlines():
|
||||
m = LINE.match(raw.strip())
|
||||
if not m:
|
||||
continue
|
||||
sid, marked = m.groups()
|
||||
parent = ""
|
||||
if pm := PARENT.search(marked):
|
||||
parent, marked = pm.group(1), marked[: pm.start()]
|
||||
out.append((sid, marked, TAG.sub("", marked), parent))
|
||||
return out
|
||||
|
||||
|
||||
def _ev(sid: str, plain: str) -> list[dict[str, str]]:
|
||||
return [{"sid": sid, "quote": plain.lstrip("★** ")[:20] or plain}]
|
||||
|
||||
|
||||
def _substantive(lines: list[tuple[str, str, str, str]]) -> dict[str, Any]:
|
||||
items = []
|
||||
for sid, marked, plain, parent in lines:
|
||||
if "★" in plain:
|
||||
basis = "star"
|
||||
elif "<iu>" in marked:
|
||||
basis = "italic_underline"
|
||||
elif INVALID.search(plain) or INVALID.search(parent):
|
||||
basis = "explicit_invalid"
|
||||
else:
|
||||
continue
|
||||
if any(k in plain for k in ("标注", "实质性要求", "条款响应", "斜体")) and len(plain) < 60:
|
||||
continue # 只是在说明标注规则的句子,不算条款(和提示词的「不要输出」一致)
|
||||
items.append({"summary": plain.lstrip("★** ")[:40], "basis": basis, "evidence": _ev(sid, plain)})
|
||||
return {"items": items}
|
||||
|
||||
|
||||
def _qualification(lines: list[tuple[str, str, str, str]]) -> dict[str, Any]:
|
||||
items = [
|
||||
{
|
||||
"summary": p[:40],
|
||||
"kind": "credit" if "信用" in p else "specific",
|
||||
"evidence_required": None,
|
||||
"lot_no": None,
|
||||
"evidence": _ev(sid, p),
|
||||
}
|
||||
for sid, _, p, _ in lines
|
||||
if "资格" in p and any(k in p for k in ("具备", "须", "应", "具有"))
|
||||
]
|
||||
return {"items": items}
|
||||
|
||||
|
||||
def _scoring(lines: list[tuple[str, str, str, str]]) -> dict[str, Any]:
|
||||
items = []
|
||||
for sid, _, p, _ in lines:
|
||||
if m := POINTS.search(p):
|
||||
items.append(
|
||||
{
|
||||
"factor": p[: m.start()].strip(":: ")[-12:] or "评分项",
|
||||
"part": "technical",
|
||||
"points": m.group(1),
|
||||
"criteria": p[:80],
|
||||
"evidence_required": None,
|
||||
"evidence": _ev(sid, p),
|
||||
}
|
||||
)
|
||||
return {"items": items}
|
||||
|
||||
|
||||
def _iso(y: str, mo: str, d: str, h: str, mi: str | None) -> str:
|
||||
return f"{y}-{int(mo):02d}-{int(d):02d}T{int(h):02d}:{mi or '00'}"
|
||||
|
||||
|
||||
def _schedule(lines: list[tuple[str, str, str, str]]) -> dict[str, Any]:
|
||||
items = []
|
||||
for sid, _, p, _ in lines:
|
||||
if m := DATE.search(p):
|
||||
y, mo, d, h, mi = m.groups()
|
||||
kind = "bid_deadline" if "截止" in p else "opening" if "开标" in p or "磋商时间" in p else "other"
|
||||
items.append(
|
||||
{
|
||||
"summary": p[:40],
|
||||
"date_kind": kind,
|
||||
"at": _iso(y, mo, d, h, mi),
|
||||
"deposit_amount": None,
|
||||
"submission_mode": None,
|
||||
"evidence": _ev(sid, p),
|
||||
}
|
||||
)
|
||||
return {"items": items}
|
||||
|
||||
|
||||
def _tech_param(lines: list[tuple[str, str, str, str]]) -> dict[str, Any]:
|
||||
ops = {"不低于": ">=", "不少于": ">=", "≥": ">=", "不小于": ">=", "不大于": "<=", "不超过": "<=", "≤": "<="}
|
||||
items = []
|
||||
for sid, _, p, _ in lines:
|
||||
if m := NUM_REQ.search(p):
|
||||
items.append(
|
||||
{
|
||||
"param": p[: m.start()].lstrip("★** ")[-10:] or "参数",
|
||||
"requirement": p.lstrip("★ ")[:60],
|
||||
"op": ops[m.group(1)],
|
||||
"value": m.group(2),
|
||||
"unit": m.group(3) or None,
|
||||
"device": None,
|
||||
"lot_no": None,
|
||||
"evidence": _ev(sid, p),
|
||||
}
|
||||
)
|
||||
return {"items": items}
|
||||
|
||||
|
||||
def _project_info(lines: list[tuple[str, str, str, str]]) -> dict[str, Any]:
|
||||
fields: dict[str, Any] = {}
|
||||
for sid, _, p, _ in lines:
|
||||
for key, label in (("name", "项目名称"), ("code", "项目编号"), ("purchaser", "采购人")):
|
||||
if key not in fields and (m := re.search(label + r"[::]\s*([^,。;]+)", p)):
|
||||
fields[key] = {"value": m.group(1).strip(), "sid": sid, "quote": p[:30]}
|
||||
if "budget_total" not in fields and (m := re.search(r"预算[^0-9]{0,8}(\d+(?:\.\d+)?)\s*(万元|元)", p)):
|
||||
fields["budget_total"] = {"value": m.group(1), "unit": m.group(2), "sid": sid, "quote": p[:30]}
|
||||
return {"fields": fields, "lots": []}
|
||||
|
||||
|
||||
TASKS: dict[str, Callable[[list[tuple[str, str, str, str]]], dict[str, Any]]] = {
|
||||
"extract.substantive": _substantive,
|
||||
"extract.qualification": _qualification,
|
||||
"extract.scoring": _scoring,
|
||||
"extract.schedule": _schedule,
|
||||
"extract.tech_param": _tech_param,
|
||||
"extract.project_info": _project_info,
|
||||
}
|
||||
|
||||
|
||||
class FakeProvider:
|
||||
name = "fake"
|
||||
|
||||
async def complete_json(self, system: str, user: str, capability: str) -> Completion:
|
||||
async def complete_json(self, system: str, user: str, capability: str, task: str) -> Completion:
|
||||
del system, capability
|
||||
items = []
|
||||
for line in user.split("## 待审分块", 1)[-1].splitlines():
|
||||
m = LINE.match(line.strip())
|
||||
if not m:
|
||||
continue
|
||||
sid, marked = m.groups()
|
||||
plain = TAG.sub("", marked)
|
||||
if "★" in plain:
|
||||
basis = "star"
|
||||
elif "<iu>" in marked:
|
||||
basis = "italic_underline"
|
||||
else:
|
||||
continue
|
||||
if any(k in plain for k in ("标注", "实质性要求", "条款响应")) and len(plain) < 60:
|
||||
continue # 只是在说明标注规则的句子,不算条款(和提示词的「不要输出」一致)
|
||||
body = plain.lstrip("★** ").strip()
|
||||
items.append({"summary": body[:40], "basis": basis, "evidence": [{"sid": sid, "quote": body[:20]}]})
|
||||
data = {"items": items}
|
||||
data = TASKS.get(task, lambda _lines: {"items": []})(_lines(user))
|
||||
return Completion(data=data, raw=json.dumps(data, ensure_ascii=False), model="fake-extractor")
|
||||
|
||||
@@ -17,8 +17,8 @@ from bid_doctor.core.config import get_settings
|
||||
class OpenAICompatProvider:
|
||||
name = "openai_compat"
|
||||
|
||||
async def complete_json(self, system: str, user: str, capability: str) -> Completion:
|
||||
del capability
|
||||
async def complete_json(self, system: str, user: str, capability: str, task: str) -> Completion:
|
||||
del capability, task
|
||||
s = get_settings()
|
||||
body = {
|
||||
"model": s.ai_model,
|
||||
|
||||
Reference in New Issue
Block a user