gjm
1c9fc72839
- 新增 evidence_model.py：统一证据模型（source_type + images 列表），访问器兼容旧 image_b64/image_mime 单图字段 - phases/evidence.py：新增 collect() 多源分发 seam（仅 CAD）、capture_images() 整体+等分象限多图（结束复位视口）、sub_windows() - phases/review.py：多图逐张下采样喂 LLM，payload 增加 source_type - config.py：EVIDENCE_DETAIL_TILES=4 / EVIDENCE_MAX_IMAGES=5（默认开启多图） - 新增 tests/test_evidence_model.py，更新 evidence/review 断言与 fakes 归一
122 lines
4.8 KiB
Python
122 lines
4.8 KiB
Python
"""阶段 3：逐图框审查（LLM + 受控只读取证子循环）。
|
||
|
||
流程：
|
||
1. 用清单生成系统提示，喂入结构化证据 + 截图。
|
||
2. LLM 若请求只读取证工具 → 执行后把观测回灌，再问（有界循环）。
|
||
3. LLM 不再请求工具时，其输出（受约束的 JSON）即为该图框的审查结论。
|
||
|
||
结论是**结构化对象** `{"summary": str, "issues": [dict], "raw": str}`，
|
||
不再依赖第二次 LLM 调用来抽取结构（见 `findings.parse_conclusion`）。
|
||
"""
|
||
|
||
import json
|
||
|
||
from .. import evidence_model
|
||
from ..cad_tools import READ_ONLY_TOOLS, is_write_tool
|
||
from ..findings import parse_conclusion
|
||
from ..llm_client import LlmClient, downsample_image_b64
|
||
from ..mcp_client import McpError, McpToolError, result_images, result_texts
|
||
|
||
# 单次审查回灌给 LLM 的图片数上限（控制视觉 token 成本）
|
||
_MAX_OBS_IMAGES = 3
|
||
|
||
|
||
def evidence_payload(evidence):
|
||
"""把证据整理成给 LLM 的 JSON（剔除 base64 图片，避免撑爆上下文）。"""
|
||
frame = evidence.get("frame") or {}
|
||
return {
|
||
"source_type": evidence.get("source_type"),
|
||
"frame_index": evidence.get("index"),
|
||
"frame": {
|
||
k: frame.get(k)
|
||
for k in ("sheet_size", "kind", "layer", "bbox", "confidence")
|
||
},
|
||
"entities": evidence.get("entities"),
|
||
"texts": evidence.get("texts"),
|
||
"dimensions": evidence.get("dimensions"),
|
||
"blocks": evidence.get("blocks"),
|
||
"polylines": evidence.get("polylines"),
|
||
}
|
||
|
||
|
||
def _run_readonly_tool(ctx, name, args, image_budget):
|
||
"""执行一个只读取证工具，返回回灌给 LLM 的 content 列表。"""
|
||
if name not in READ_ONLY_TOOLS or is_write_tool(name):
|
||
return [{"type": "text", "text": f"工具 {name} 不在只读取证白名单内，已拒绝。"}], image_budget
|
||
|
||
try:
|
||
resp = ctx.tools.call(name, args)
|
||
except (McpError, McpToolError) as exc:
|
||
return [{"type": "text", "text": f"工具 {name} 执行失败: {exc}"}], image_budget
|
||
|
||
content = []
|
||
for text in result_texts(resp):
|
||
content.append({"type": "text", "text": text})
|
||
for b64, mime in result_images(resp):
|
||
if image_budget[0] <= 0:
|
||
content.append({"type": "text", "text": "（后续截图已省略以控制成本）"})
|
||
break
|
||
image_budget[0] -= 1
|
||
nb, nm = downsample_image_b64(b64, mime)
|
||
content.append(LlmClient.image_content(nb, nm))
|
||
if not content:
|
||
content.append({"type": "text", "text": "（无返回内容）"})
|
||
return content, image_budget
|
||
|
||
|
||
def review_frame(ctx, evidence, system_prompt, log=print):
|
||
"""审查单个图框，返回结构化结论 `{"summary", "issues", "raw"}`。"""
|
||
payload = evidence_payload(evidence)
|
||
user_content = [
|
||
{
|
||
"type": "text",
|
||
"text": "以下是该图框的审查输入数据(JSON)：\n"
|
||
+ json.dumps(payload, ensure_ascii=False, indent=2),
|
||
}
|
||
]
|
||
for img in evidence_model.evidence_images(evidence):
|
||
nb, nm = downsample_image_b64(img["b64"], img.get("mime") or "image/png")
|
||
label = img.get("label")
|
||
if label:
|
||
user_content.append({"type": "text", "text": f"（截图：{label}）"})
|
||
user_content.append(LlmClient.image_content(nb, nm))
|
||
|
||
messages = [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_content},
|
||
]
|
||
|
||
tools_schema = ctx.tools.openai_tools(names=READ_ONLY_TOOLS)
|
||
image_budget = [_MAX_OBS_IMAGES]
|
||
|
||
for _ in range(max(0, ctx.max_iters)):
|
||
resp = ctx.llm.chat(messages, tools=tools_schema)
|
||
msg = resp.choices[0].message
|
||
messages.append(msg)
|
||
|
||
# 仅在「不再请求工具」的回合解析结论；中间工具回合的 content 永不解析
|
||
if not msg.tool_calls:
|
||
return parse_conclusion(msg.content or "", log=log)
|
||
|
||
for tool_call in msg.tool_calls:
|
||
name = tool_call.function.name
|
||
try:
|
||
args = json.loads(tool_call.function.arguments or "{}")
|
||
except json.JSONDecodeError:
|
||
args = {}
|
||
log(f" [取证] {name} {args if args else ''}")
|
||
content, image_budget = _run_readonly_tool(ctx, name, args, image_budget)
|
||
messages.append(
|
||
{"role": "tool", "tool_call_id": tool_call.id, "content": content}
|
||
)
|
||
|
||
# 达到回合上限：强制收口（不传 tools，保证不再请求工具，输出必为终稿 JSON）
|
||
messages.append(
|
||
{
|
||
"role": "user",
|
||
"content": "已达到取证回合上限，请基于现有信息直接给出该图框的审查结论（JSON）。",
|
||
}
|
||
)
|
||
resp = ctx.llm.chat(messages)
|
||
return parse_conclusion(resp.choices[0].message.content or "", log=log)
|