feat(review): 审查结论改为结构化 JSON,移除二次 LLM 抽取

- checklist.py:系统提示加入结构化输出契约(只输出 {summary, issues[]},
  引用 handle 必须真实存在、不得编造)
- review.py:review_frame 返回 {summary, issues, raw};仅在「不再请求工具」的回合
  解析结论,中间工具回合的 content 永不解析;强制收口那次 chat 不传 tools
- report.py:删除 extract_issues 与 _EXTRACT_SYSTEM(不再需要第二次 LLM 调用);
  新增 format_finding(用 .get,顺带修复缺 frame 键 KeyError);save_report 的
  issues 省略时从 findings 拍平
- __main__.py:主流程改为 确定性检查 → review_frame → build_finding(合并+校验),
  issues 由 flatten_issues 派生

顺带修复既有 gap:summarize 硬下标 KeyError、extract_issues 单行围栏解析失败。
This commit is contained in:
gjm committed 2026-10-14 19:53:41 +08:00
1 parent f6d61f8d07
commit f7fa8f14d9
8 files changed
+224 -140

No files matched your search

+26 -14
View File
@@ -6,6 +6,8 @@ import sys
from pathlib import Path from pathlib import Path
from . import config from . import config
from . import checks as checks_phase
from . import findings as findings_mod
from .annotator import annotate, clear from .annotator import annotate, clear
from .checklist import build_system_prompt, checklist_summary from .checklist import build_system_prompt, checklist_summary
from .context import ReviewContext from .context import ReviewContext
@@ -80,25 +82,35 @@ def run_review(ctx, args):
print(f"[evidence] 已写入 {path}") print(f"[evidence] 已写入 {path}")
return 0 return 0
print("[3/4] 审查:逐图框 LLM 审查(含受控只读取证子循环) ...") print("[3/4] 审查:确定性检查前置 + 逐图框 LLM 审查(含受控只读取证子循环) ...")
findings = [] frame_findings = []
for ev in evidences: for ev in evidences:
print(f" - 审查图框 {ev['index']} ...") print(f" - 审查图框 {ev['index']} ...")
text = review.review_frame(ctx, ev, system_prompt) checks = (
findings.append( checks_phase.run_deterministic_checks(ctx, ev)
{ if config.DETERMINISTIC_CHECKS
"frame": ev["index"], else []
"sheet": (ev.get("frame") or {}).get("sheet_size"), )
"finding": text, conclusion = review.review_frame(ctx, ev, system_prompt)
} finding = findings_mod.build_finding(
ev, conclusion, checks, validate_handles=config.VALIDATE_HANDLES
)
frame_findings.append(finding)
counts = finding["counts"]
preview = finding["summary"]
if len(preview) > 200:
preview = preview[:200] + "..."
print(
f" 结论:{preview}"
f"(确定性 {counts['deterministic']} / LLM {counts['llm']}"
f" / 未核实 {counts['unverified']})"
) )
preview = text if len(text) <= 200 else text[:200] + "..."
print(f" 结论:{preview}")
print("[4/4] 汇总报告 ...") print("[4/4] 汇总报告 ...")
markdown = report.summarize(ctx, findings) markdown = report.summarize(ctx, frame_findings)
issues = report.extract_issues(ctx, findings) issues = findings_mod.flatten_issues(frame_findings)
report.save_report(markdown, findings, plan, issues) report.save_report(markdown, frame_findings, plan, issues)
print() print()
print(markdown) print(markdown)
+19 -1
View File
@@ -62,9 +62,27 @@ def build_system_prompt(checklist, *, allow_tools=True):
"要求:", "要求:",
"- 只报告你**有依据**的问题,不要臆测;数据与截图冲突时以数据为准。", "- 只报告你**有依据**的问题,不要臆测;数据与截图冲突时以数据为准。",
"- 按清单条目组织结论,未命中的条目可省略。", "- 按清单条目组织结论,未命中的条目可省略。",
"- 若该图框无任何明显问题,明确说“未发现明显问题”。",
"- 不要复述输入数据。", "- 不要复述输入数据。",
"",
"输出格式(**必须严格遵守**):",
"在你完成全部取证、准备给出该图框的最终结论时,**只输出一个 JSON 对象**,"
"不要输出任何解释文字,也不要用 markdown 代码块包裹。结构如下:",
"{",
' "summary": "<该图框的一句话整体结论;若未发现问题,写“未发现明显问题”>",',
' "issues": [',
" {",
' "category": "<图层/文字/标注/几何/图面整体 之一>",',
' "severity": "<high|medium|low>",',
' "message": "<问题描述,简洁具体>",',
' "handle": "<该问题对应的实体句柄;无法定位到具体实体时省略此字段>"',
" }",
" ]",
"}",
"其中 `handle` 必须是你**在取证数据或工具返回里真实见过**的句柄,不要编造;"
"无法确定具体实体时省略 `handle`,该问题仍会被记录。",
] ]
if allow_tools:
lines.append("(需要进一步确认时先调用工具;拿到结果后,最终再输出上述 JSON。)")
return "\n".join(lines) return "\n".join(lines)
+38 -43
View File
@@ -1,10 +1,37 @@
"""阶段 4:汇总报告 + 结构化问题抽取。""" """阶段 4:汇总报告 + 落盘。
审查结论在阶段 3 已是结构化对象(`{summary, issues[]}`),
因此这里**不再**需要第二次 LLM 调用来抽取结构。
"""
import json import json
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from .. import config from .. import config
from ..findings import flatten_issues
_SUMMARY_SYSTEM = (
"你是图纸审查负责人。请把下面各图框的审查意见汇总为一份结构化的审查报告:"
"先给整体结论,再按图框列出问题,最后给出整改建议。语言简洁、条目化。"
)
def format_finding(finding):
"""把一个 findings 元素渲染成给汇总 LLM 的文本块。"""
lines = [f"【图框 {finding.get('frame')}(图幅 {finding.get('sheet')})】"]
summary = finding.get("summary")
if summary:
lines.append(str(summary))
for issue in finding.get("issues") or []:
tag = "确定性" if issue.get("source") == "deterministic" else "LLM"
flag = "" if issue.get("verified", True) else "(未在证据中定位,仅供参考)"
handle = issue.get("handle")
where = f" handle={handle}" if handle else ""
lines.append(
f"- [{issue.get('severity', '?')}][{tag}] {issue.get('message', '')}{where}{flag}"
)
return "\n".join(lines)
def summarize(ctx, findings, log=print): def summarize(ctx, findings, log=print):
@@ -12,54 +39,22 @@ def summarize(ctx, findings, log=print):
if not findings: if not findings:
return "未识别到可审查的图框,或所有图框均无有效包围盒。" return "未识别到可审查的图框,或所有图框均无有效包围盒。"
report_input = "\n\n".join( report_input = "\n\n".join(format_finding(f) for f in findings)
f"【图框 {f['frame']}(图幅 {f.get('sheet')})】\n{f['finding']}" for f in findings
)
system = (
"你是图纸审查负责人。请把下面各图框的审查意见汇总为一份结构化的审查报告:"
"先给整体结论,再按图框列出问题,最后给出整改建议。语言简洁、条目化。"
)
try: try:
return ctx.llm.chat_text(system, report_input) return ctx.llm.chat_text(_SUMMARY_SYSTEM, report_input)
except Exception as exc: # noqa: BLE001 except Exception as exc: # noqa: BLE001
log(f"[汇总失败] {exc}") log(f"[汇总失败] {exc}")
return report_input return report_input
_EXTRACT_SYSTEM = (
"你是结构化信息抽取器。请把下面的图纸审查意见抽取为 JSON 数组,"
"每个元素形如:"
'{"frame": <图框序号>, "handle": "<实体句柄,没有则省略>", '
'"category": "<图层/文字/标注/几何/图面整体>", '
'"severity": "<high|medium|low>", "message": "<问题描述>"}。'
"只输出 JSON 数组,不要输出任何解释、不要用 markdown 代码块包裹。"
"若某条意见无法定位到具体实体,省略 handle 字段。"
)
def extract_issues(ctx, findings, log=print):
"""把自由文本的审查意见抽取为结构化问题列表(用于批注回写)。"""
if not findings:
return []
source = "\n\n".join(
f"图框 {f['frame']}:\n{f['finding']}" for f in findings
)
try:
text = ctx.llm.chat_text(_EXTRACT_SYSTEM, source)
text = text.strip()
if text.startswith("```"):
text = text.strip("`")
text = text.split("\n", 1)[1] if "\n" in text else text
data = json.loads(text)
if isinstance(data, list):
return data
except Exception as exc: # noqa: BLE001
log(f"[结构化抽取失败] {exc}")
return []
def save_report(markdown, findings, plan, issues=None, log=print): def save_report(markdown, findings, plan, issues=None, log=print):
"""落盘 Markdown + JSON 报告,返回两个路径。""" """落盘 Markdown + JSON 报告,返回两个路径。
`issues` 省略时从 findings 里拍平派生(供批注回写)。
"""
if issues is None:
issues = flatten_issues(findings)
out_dir = Path(config.REPORT_DIR) out_dir = Path(config.REPORT_DIR)
out_dir.mkdir(parents=True, exist_ok=True) out_dir.mkdir(parents=True, exist_ok=True)
stamp = datetime.now().strftime("%Y%m%d_%H%M%S") stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -72,7 +67,7 @@ def save_report(markdown, findings, plan, issues=None, log=print):
{ {
"generated_at": stamp, "generated_at": stamp,
"findings": findings, "findings": findings,
"issues": issues or [], "issues": issues,
"frames": plan.get("frames"), "frames": plan.get("frames"),
}, },
ensure_ascii=False, ensure_ascii=False,
+11 -6
View File
@@ -3,12 +3,16 @@
流程: 流程:
1. 用清单生成系统提示,喂入结构化证据 + 截图。 1. 用清单生成系统提示,喂入结构化证据 + 截图。
2. LLM 若请求只读取证工具 → 执行后把观测回灌,再问(有界循环)。 2. LLM 若请求只读取证工具 → 执行后把观测回灌,再问(有界循环)。
3. LLM 不再请求工具时,其输出即为该图框的审查结论。 3. LLM 不再请求工具时,其输出(受约束的 JSON)即为该图框的审查结论。
结论是**结构化对象** `{"summary": str, "issues": [dict], "raw": str}`,
不再依赖第二次 LLM 调用来抽取结构(见 `findings.parse_conclusion`)。
""" """
import json import json
from ..cad_tools import READ_ONLY_TOOLS, is_write_tool from ..cad_tools import READ_ONLY_TOOLS, is_write_tool
from ..findings import parse_conclusion
from ..llm_client import LlmClient, downsample_image_b64 from ..llm_client import LlmClient, downsample_image_b64
from ..mcp_client import McpError, McpToolError, result_images, result_texts from ..mcp_client import McpError, McpToolError, result_images, result_texts
@@ -59,7 +63,7 @@ def _run_readonly_tool(ctx, name, args, image_budget):
def review_frame(ctx, evidence, system_prompt, log=print): def review_frame(ctx, evidence, system_prompt, log=print):
"""审查单个图框,返回审查意见文本。""" """审查单个图框,返回结构化结论 `{"summary", "issues", "raw"}`。"""
payload = evidence_payload(evidence) payload = evidence_payload(evidence)
user_content = [ user_content = [
{ {
@@ -86,8 +90,9 @@ def review_frame(ctx, evidence, system_prompt, log=print):
msg = resp.choices[0].message msg = resp.choices[0].message
messages.append(msg) messages.append(msg)
# 仅在「不再请求工具」的回合解析结论;中间工具回合的 content 永不解析
if not msg.tool_calls: if not msg.tool_calls:
return msg.content or "" return parse_conclusion(msg.content or "", log=log)
for tool_call in msg.tool_calls: for tool_call in msg.tool_calls:
name = tool_call.function.name name = tool_call.function.name
@@ -101,12 +106,12 @@ def review_frame(ctx, evidence, system_prompt, log=print):
{"role": "tool", "tool_call_id": tool_call.id, "content": content} {"role": "tool", "tool_call_id": tool_call.id, "content": content}
) )
# 达到回合上限:强制收口 # 达到回合上限:强制收口(不传 tools,保证不再请求工具,输出必为终稿 JSON)
messages.append( messages.append(
{ {
"role": "user", "role": "user",
"content": "已达到取证回合上限,请基于现有信息直接给出该图框的审查结论。", "content": "已达到取证回合上限,请基于现有信息直接给出该图框的审查结论(JSON)。",
} }
) )
resp = ctx.llm.chat(messages) resp = ctx.llm.chat(messages)
return resp.choices[0].message.content or "" return parse_conclusion(resp.choices[0].message.content or "", log=log)
+1 -1
View File
@@ -101,7 +101,7 @@ class FakeLlmClient:
- `script` :`chat()` 依次弹出的响应队列;弹空后回落到 `text`。 - `script` :`chat()` 依次弹出的响应队列;弹空后回落到 `text`。
- `text` :`chat()` / `chat_text()` 的兜底文本。 - `text` :`chat()` / `chat_text()` 的兜底文本。
- `text_script` :`chat_text()` 专用队列(便于同时喂 summarize 与 extract_issues)。 - `text_script` :`chat_text()` 专用队列(便于连续喂多次 `chat_text` 调用)。
""" """
def __init__(self, script=None, text="", text_script=None, model="fake-model"): def __init__(self, script=None, text="", text_script=None, model="fake-model"):
+16
View File
@@ -65,6 +65,22 @@ def test_build_system_prompt_tool_hint_toggle():
assert "只读" not in without assert "只读" not in without
def test_build_system_prompt_requires_structured_json_output():
# allow_tools=False 以隔离「工具提示块」里的「只读」字样
prompt = build_system_prompt(DEFAULT_CHECKLIST, allow_tools=False)
assert "JSON" in prompt
assert '"summary"' in prompt
assert '"issues"' in prompt
assert '"severity"' in prompt and '"category"' in prompt and '"handle"' in prompt
# 结构化契约本身不得含「只读」,否则会破坏 allow_tools 开关的判别
assert "只读" not in prompt
def test_build_system_prompt_forbids_fabricated_handle():
prompt = build_system_prompt(DEFAULT_CHECKLIST)
assert "不要编造" in prompt
def test_checklist_summary_mentions_name_and_count(): def test_checklist_summary_mentions_name_and_count():
summary = checklist_summary(DEFAULT_CHECKLIST) summary = checklist_summary(DEFAULT_CHECKLIST)
assert DEFAULT_CHECKLIST["name"] in summary assert DEFAULT_CHECKLIST["name"] in summary
+72 -70
View File
@@ -2,19 +2,64 @@
import json import json
import pytest from dwg_review.phases.report import format_finding, save_report, summarize
from dwg_review.phases.report import extract_issues, save_report, summarize
from tests.fakes import FakeLlmClient, make_ctx from tests.fakes import FakeLlmClient, make_ctx
def issue(**over):
base = {
"frame": 1, "category": "文字", "severity": "high", "message": "文字重叠",
"handle": "A1", "source": "llm", "verified": True,
}
base.update(over)
return base
def findings_fixture(): def findings_fixture():
return [ return [
{"frame": 1, "sheet": "A1", "finding": "图框1:文字重叠"}, {
{"frame": 2, "sheet": "A3", "finding": "图框2:未发现明显问题"}, "frame": 1, "sheet": "A1", "summary": "该图框存在文字重叠",
"issues": [issue()],
"counts": {"deterministic": 0, "llm": 1, "unverified": 0},
},
{
"frame": 2, "sheet": "A3", "summary": "未发现明显问题",
"issues": [],
"counts": {"deterministic": 0, "llm": 0, "unverified": 0},
},
] ]
# ---------------------------------------------------------------------------
# format_finding
# ---------------------------------------------------------------------------
def test_format_finding_renders_summary_and_issues():
text = format_finding(findings_fixture()[0])
assert "图框 1" in text and "图幅 A1" in text
assert "该图框存在文字重叠" in text
assert "[high][LLM] 文字重叠" in text
assert "handle=A1" in text
def test_format_finding_marks_deterministic_source():
finding = {"frame": 1, "sheet": "A1", "summary": "s",
"issues": [issue(source="deterministic", handle="Z1")]}
assert "[确定性]" in format_finding(finding)
def test_format_finding_marks_unverified():
finding = {"frame": 1, "sheet": "A1", "summary": "s",
"issues": [issue(handle=None, verified=False)]}
text = format_finding(finding)
assert "未在证据中定位" in text
assert "handle=" not in text
def test_format_finding_tolerates_missing_keys():
text = format_finding({})
assert "图框 None" in text # 不再 KeyError(修复既有硬下标 gap)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# summarize # summarize
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -31,7 +76,16 @@ def test_summarize_uses_chat_text():
ctx = make_ctx(llm=llm) ctx = make_ctx(llm=llm)
assert summarize(ctx, findings_fixture(), log=lambda *a: None) == "## 总报告" assert summarize(ctx, findings_fixture(), log=lambda *a: None) == "## 总报告"
assert len(llm.calls) == 1 assert len(llm.calls) == 1
assert "图框1" in llm.calls[0]["user"] assert "图框 1" in llm.calls[0]["user"]
assert "图框 2" in llm.calls[0]["user"]
def test_summarize_input_includes_issue_lines():
llm = FakeLlmClient(text="x")
ctx = make_ctx(llm=llm)
summarize(ctx, findings_fixture(), log=lambda *a: None)
user = llm.calls[0]["user"]
assert "[high][LLM] 文字重叠" in user
def test_summarize_falls_back_to_raw_input_on_error(): def test_summarize_falls_back_to_raw_input_on_error():
@@ -42,70 +96,16 @@ def test_summarize_falls_back_to_raw_input_on_error():
logs = [] logs = []
ctx = make_ctx(llm=Boom()) ctx = make_ctx(llm=Boom())
out = summarize(ctx, findings_fixture(), log=logs.append) out = summarize(ctx, findings_fixture(), log=logs.append)
assert "图框1" in out and "图框2" in out assert "图框 1" in out and "图框 2" in out
assert any("汇总失败" in line for line in logs) assert any("汇总失败" in line for line in logs)
def test_summarize_missing_frame_key_known_gap(): def test_summarize_tolerates_finding_without_frame_key():
"""已知瑕疵:summarize 用硬下标 f['frame'],findings 缺该键直接 KeyError。""" # 修复既有硬下标 gap:缺 frame 键不再 KeyError
ctx = make_ctx(llm=FakeLlmClient(text="x")) llm = FakeLlmClient(text="ok")
with pytest.raises(KeyError):
summarize(ctx, [{"finding": "缺 frame 键"}], log=lambda *a: None)
# ---------------------------------------------------------------------------
# extract_issues
# ---------------------------------------------------------------------------
def test_extract_issues_empty_returns_empty_list():
llm = FakeLlmClient(text_script=["[]"])
ctx = make_ctx(llm=llm) ctx = make_ctx(llm=llm)
assert extract_issues(ctx, [], log=lambda *a: None) == [] out = summarize(ctx, [{"summary": "无 frame 键", "issues": []}], log=lambda *a: None)
assert llm.calls == [] assert out == "ok"
def test_extract_issues_parses_json_array():
payload = '[{"frame": 1, "handle": "A", "severity": "high", "message": "x"}]'
llm = FakeLlmClient(text_script=[payload])
ctx = make_ctx(llm=llm)
issues = extract_issues(ctx, findings_fixture(), log=lambda *a: None)
assert issues == [{"frame": 1, "handle": "A", "severity": "high", "message": "x"}]
def test_extract_issues_strips_code_fence():
fenced = '```json\n[{"frame": 1, "message": "y"}]\n```'
llm = FakeLlmClient(text_script=[fenced])
ctx = make_ctx(llm=llm)
assert extract_issues(ctx, findings_fixture(), log=lambda *a: None) == [
{"frame": 1, "message": "y"}
]
def test_extract_issues_returns_empty_on_invalid_json():
logs = []
llm = FakeLlmClient(text_script=["这不是 JSON"])
ctx = make_ctx(llm=llm)
assert extract_issues(ctx, findings_fixture(), log=logs.append) == []
assert any("结构化抽取失败" in line for line in logs)
def test_extract_issues_returns_empty_when_not_a_list():
llm = FakeLlmClient(text_script=['{"frame": 1}'])
ctx = make_ctx(llm=llm)
assert extract_issues(ctx, findings_fixture(), log=lambda *a: None) == []
def test_extract_issues_single_line_fence_known_gap():
"""已知瑕疵:围栏剥离依赖换行,单行 ```json[...]``` 会残留 json 前缀 → 解析失败。"""
llm = FakeLlmClient(text_script=['```json[{"frame": 1, "message": "z"}]```'])
ctx = make_ctx(llm=llm)
assert extract_issues(ctx, findings_fixture(), log=lambda *a: None) == []
def test_extract_issues_missing_frame_key_known_gap():
"""已知瑕疵:extract_issues 同样用硬下标 f['frame'],findings 缺该键直接 KeyError。"""
ctx = make_ctx(llm=FakeLlmClient(text="[]"))
with pytest.raises(KeyError):
extract_issues(ctx, [{"finding": "缺 frame 键"}], log=lambda *a: None)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -114,7 +114,7 @@ def test_extract_issues_missing_frame_key_known_gap():
def test_save_report_writes_md_and_json(tmp_reports): def test_save_report_writes_md_and_json(tmp_reports):
findings = findings_fixture() findings = findings_fixture()
plan = {"frames": [{"bbox": [0, 0, 10, 10]}]} plan = {"frames": [{"bbox": [0, 0, 10, 10]}]}
issues = [{"frame": 1, "handle": "A", "message": "x"}] issues = [issue()]
md_path, json_path = save_report("# 报告", findings, plan, issues, log=lambda *a: None) md_path, json_path = save_report("# 报告", findings, plan, issues, log=lambda *a: None)
@@ -127,9 +127,11 @@ def test_save_report_writes_md_and_json(tmp_reports):
assert "generated_at" in data assert "generated_at" in data
def test_save_report_issues_default_to_empty(tmp_reports): def test_save_report_derives_issues_from_findings(tmp_reports):
_, json_path = save_report("md", findings_fixture(), {"frames": []}, log=lambda *a: None) findings = findings_fixture()
assert json.loads(json_path.read_text(encoding="utf-8"))["issues"] == [] _, json_path = save_report("md", findings, {"frames": []}, log=lambda *a: None)
data = json.loads(json_path.read_text(encoding="utf-8"))
assert data["issues"] == [issue()] # 拍平自 findings
def test_save_report_creates_missing_directory(tmp_path, monkeypatch): def test_save_report_creates_missing_directory(tmp_path, monkeypatch):
+41 -5
View File
@@ -111,11 +111,12 @@ def test_readonly_empty_response_marker():
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# review_frame # review_frame
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def test_review_frame_returns_text_when_no_tools(): def test_review_frame_returns_summary_when_no_tools():
llm = FakeLlmClient(script=[llm_text("结论A")]) llm = FakeLlmClient(script=[llm_text("结论A")])
ctx = make_ctx(client=FakeMcpClient(), llm=llm) ctx = make_ctx(client=FakeMcpClient(), llm=llm)
out = review_frame(ctx, make_evidence(), "系统提示", log=lambda *a: None) out = review_frame(ctx, make_evidence(), "系统提示", log=lambda *a: None)
assert out == "结论A" assert out["summary"] == "结论A" # 非 JSON → 降级为纯文本 summary
assert out["issues"] == []
assert len(llm.calls) == 1 assert len(llm.calls) == 1
assert llm.calls[0]["tools"] is not None # 传了只读工具 schema assert llm.calls[0]["tools"] is not None # 传了只读工具 schema
@@ -147,7 +148,7 @@ def test_review_frame_runs_tool_then_concludes():
client = FakeMcpClient(responses={"query_entities": ok_text({"total": 1})}) client = FakeMcpClient(responses={"query_entities": ok_text({"total": 1})})
ctx = make_ctx(client=client, llm=llm) ctx = make_ctx(client=client, llm=llm)
out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
assert out == "结论B" assert out["summary"] == "结论B"
assert len(llm.calls) == 2 assert len(llm.calls) == 2
assert client.called_names() == ["query_entities"] assert client.called_names() == ["query_entities"]
@@ -179,7 +180,7 @@ def test_review_frame_force_closes_at_iteration_limit():
) )
ctx = make_ctx(client=FakeMcpClient(), llm=llm, max_iters=1) ctx = make_ctx(client=FakeMcpClient(), llm=llm, max_iters=1)
out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
assert out == "收口结论" assert out["summary"] == "收口结论"
last_msg = llm.calls[-1]["messages"][-1] last_msg = llm.calls[-1]["messages"][-1]
assert last_msg["role"] == "user" assert last_msg["role"] == "user"
assert "上限" in last_msg["content"] assert "上限" in last_msg["content"]
@@ -189,7 +190,7 @@ def test_review_frame_with_zero_iters_forces_immediate_close():
llm = FakeLlmClient(script=[llm_text("直接结论")]) llm = FakeLlmClient(script=[llm_text("直接结论")])
ctx = make_ctx(client=FakeMcpClient(), llm=llm, max_iters=0) ctx = make_ctx(client=FakeMcpClient(), llm=llm, max_iters=0)
out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
assert out == "直接结论" assert out["summary"] == "直接结论"
assert len(llm.calls) == 1 assert len(llm.calls) == 1
@@ -207,3 +208,38 @@ def test_review_frame_payload_is_json_serializable():
review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
text = llm.calls[0]["messages"][1]["content"][0]["text"] text = llm.calls[0]["messages"][1]["content"][0]["text"]
json.loads(text.split(":\n", 1)[1]) # 不抛异常即通过 json.loads(text.split(":\n", 1)[1]) # 不抛异常即通过
def test_review_frame_parses_structured_json_conclusion():
text = (
'{"summary": "整体尚可", "issues": ['
'{"category": "几何", "severity": "medium", "message": "零长度", "handle": "A1"}]}'
)
llm = FakeLlmClient(script=[llm_text(text)])
ctx = make_ctx(client=FakeMcpClient(), llm=llm)
out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
assert out["summary"] == "整体尚可"
assert out["issues"] == [
{"category": "几何", "severity": "medium", "message": "零长度", "handle": "A1"}
]
def test_review_frame_parses_fenced_json():
llm = FakeLlmClient(script=[llm_text('```json\n{"summary": "s", "issues": []}\n```')])
ctx = make_ctx(client=FakeMcpClient(), llm=llm)
out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
assert out["summary"] == "s"
assert out["issues"] == []
def test_review_frame_intermediate_tool_turn_json_not_parsed():
# 中间工具回合的 content 里即使塞了 JSON,也不能被当成最终结论
intermediate = '{"summary": "这不是最终结论", "issues": [{"message": "x"}]}'
llm = FakeLlmClient(script=[
llm_calls([("query_entities", {})], content=intermediate),
llm_text('{"summary": "最终结论", "issues": []}'),
])
ctx = make_ctx(client=FakeMcpClient(), llm=llm)
out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
assert out["summary"] == "最终结论"
assert out["issues"] == []