feat(review): 新增确定性检查前置与结论结构化内核

- checks.py:确定性检查(find_zero_length / find_duplicate_entities 传 window
  + 纯 Python 派生空文字、未闭合多段线)。问题带 source=deterministic,handle 来自
  工具/数据本身,不依赖 LLM 回忆
- findings.py:解析 LLM 的 JSON 结论 + handle 交叉校验(引用未知 handle 时保留问题、
  剥掉 handle、verified=False)+ findings 装配
- config.py:新增 DETERMINISTIC_CHECKS / VALIDATE_HANDLES / CHECK_UNCLOSED_POLYLINES
  开关与 ZERO_LENGTH_TOLERANCE / DUP_TOLERANCE / DUP_GROUP_LIMIT

实机验证:find_zero_length / find_duplicate_entities 的 window 参数在基座上生效
(schema 未声明,C++ 侧 parseCommonQuery 实际支持)。
This commit is contained in:
gjm committed 2026-10-14 19:53:31 +08:00
1 parent a0ea902b50
commit f6d61f8d07
6 files changed
+884

No files matched your search

+212
View File
@@ -0,0 +1,212 @@
"""`dwg_review.checks` 单元测试:确定性检查前置。"""
from dwg_review import checks
from dwg_review.checks import (
check_duplicate_entities,
check_zero_length,
derive_from_evidence,
make_issue,
run_deterministic_checks,
)
from dwg_review.mcp_client import McpError, McpToolError
from tests.fakes import FakeMcpClient, make_ctx, make_evidence, ok_text
QUIET = lambda *a: None # noqa: E731
WINDOW = {"min_x": 0.0, "min_y": 0.0, "max_x": 10.0, "max_y": 10.0}
# ---------------------------------------------------------------------------
# make_issue
# ---------------------------------------------------------------------------
def test_make_issue_minimal():
issue = make_issue("几何", "medium", "零长度")
assert issue == {"category": "几何", "severity": "medium",
"message": "零长度", "source": "deterministic"}
def test_make_issue_with_handles():
issue = make_issue("几何", "low", "重复", handle="A", handles=["A", "B"])
assert issue["handle"] == "A"
assert issue["handles"] == ["A", "B"]
def test_make_issue_omits_empty_handle():
assert "handle" not in make_issue("文字", "low", "空文字", handle=None)
# ---------------------------------------------------------------------------
# check_zero_length
# ---------------------------------------------------------------------------
def test_zero_length_turns_items_into_issues():
client = FakeMcpClient(responses={"find_zero_length": ok_text({
"total": 2, "returned": 2, "offset": 0,
"items": [
{"handle": "A1", "type": "LINE", "layer": "0", "length": 0.0},
{"handle": "A2", "type": "ARC", "layer": "0", "length": 0.0001},
],
})})
ctx = make_ctx(client=client)
issues = check_zero_length(ctx, WINDOW, tolerance=0.001, limit=200, log=QUIET)
assert [i["handle"] for i in issues] == ["A1", "A2"]
assert all(i["category"] == "几何" and i["severity"] == "medium" for i in issues)
def test_zero_length_skips_items_without_handle():
client = FakeMcpClient(responses={"find_zero_length": ok_text(
{"items": [{"type": "LINE"}, {"handle": "B1"}]}
)})
ctx = make_ctx(client=client)
assert [i["handle"] for i in check_zero_length(ctx, WINDOW, tolerance=0.001,
limit=200, log=QUIET)] == ["B1"]
def test_zero_length_passes_window_and_params():
client = FakeMcpClient(responses={"find_zero_length": ok_text({"items": []})})
ctx = make_ctx(client=client)
check_zero_length(ctx, WINDOW, tolerance=0.5, limit=77, log=QUIET)
assert client.arguments_of("find_zero_length") == {
"window": WINDOW, "tolerance": 0.5, "limit": 77,
}
def test_zero_length_swallows_tool_error():
logs = []
client = FakeMcpClient(responses={"find_zero_length": McpToolError("炸了")})
ctx = make_ctx(client=client)
assert check_zero_length(ctx, WINDOW, tolerance=0.001, limit=200, log=logs.append) == []
assert any("find_zero_length 失败" in line for line in logs)
def test_zero_length_tolerates_non_dict_envelope():
client = FakeMcpClient(responses={"find_zero_length": ok_text("oops")})
ctx = make_ctx(client=client)
assert check_zero_length(ctx, WINDOW, tolerance=0.001, limit=200, log=QUIET) == []
# ---------------------------------------------------------------------------
# check_duplicate_entities
# ---------------------------------------------------------------------------
def test_duplicate_group_becomes_single_issue_with_all_handles():
client = FakeMcpClient(responses={"find_duplicate_entities": ok_text({
"count": 1,
"groups": [{"reason": "identical_geometry", "type": "LINE",
"handles": ["B1", "B2", "B3"]}],
})})
ctx = make_ctx(client=client)
issues = check_duplicate_entities(ctx, WINDOW, tolerance=0.001, group_limit=50, log=QUIET)
assert len(issues) == 1 # 一组 = 一个问题
assert issues[0]["handle"] == "B1"
assert issues[0]["handles"] == ["B1", "B2", "B3"]
assert issues[0]["severity"] == "low"
def test_duplicate_skips_empty_groups():
client = FakeMcpClient(responses={"find_duplicate_entities": ok_text({
"groups": [{"handles": []}, {"handles": None}, {"handles": ["C1", "C2"]}],
})})
ctx = make_ctx(client=client)
issues = check_duplicate_entities(ctx, WINDOW, tolerance=0.001, group_limit=50, log=QUIET)
assert [i["handle"] for i in issues] == ["C1"]
def test_duplicate_passes_group_limit():
client = FakeMcpClient(responses={"find_duplicate_entities": ok_text({"groups": []})})
ctx = make_ctx(client=client)
check_duplicate_entities(ctx, WINDOW, tolerance=0.2, group_limit=9, log=QUIET)
assert client.arguments_of("find_duplicate_entities") == {
"window": WINDOW, "tolerance": 0.2, "limit": 9,
}
def test_duplicate_swallows_tool_error():
logs = []
client = FakeMcpClient(responses={"find_duplicate_entities": McpError(-32603, "x")})
ctx = make_ctx(client=client)
assert check_duplicate_entities(ctx, WINDOW, tolerance=0.001, group_limit=50,
log=logs.append) == []
assert any("find_duplicate_entities 失败" in line for line in logs)
# ---------------------------------------------------------------------------
# derive_from_evidence
# ---------------------------------------------------------------------------
def test_derive_flags_empty_and_whitespace_texts():
ev = make_evidence(texts={"items": [
{"handle": "T1", "content": ""},
{"handle": "T2", "content": " "},
{"handle": "T3", "content": None},
{"handle": "T4", "content": "正常文字"},
]})
issues = derive_from_evidence(ev)
assert [i["handle"] for i in issues] == ["T1", "T2", "T3"]
assert all(i["category"] == "文字" for i in issues)
def test_derive_flags_unclosed_polyline():
ev = make_evidence(polylines={"items": [
{"handle": "P1", "vertex_count": 4, "closed": False},
{"handle": "P2", "vertex_count": 4, "closed": True},
{"handle": "P3", "vertex_count": 2, "closed": False},
{"handle": "P4", "closed": False},
]})
issues = derive_from_evidence(ev)
assert [i["handle"] for i in issues] == ["P1"]
def test_derive_can_disable_unclosed_check():
ev = make_evidence(polylines={"items": [{"handle": "P1", "vertex_count": 4, "closed": False}]})
assert derive_from_evidence(ev, check_unclosed=False) == []
def test_derive_tolerates_missing_sources():
assert derive_from_evidence({}) == []
assert derive_from_evidence(make_evidence()) == []
# ---------------------------------------------------------------------------
# run_deterministic_checks
# ---------------------------------------------------------------------------
def build_ctx():
client = FakeMcpClient(responses={
"find_zero_length": ok_text({"items": [{"handle": "Z1", "type": "LINE", "length": 0.0}]}),
"find_duplicate_entities": ok_text({"groups": [
{"type": "LINE", "handles": ["D1", "D2"]}]}),
})
return make_ctx(client=client)
def test_run_combines_derived_and_mcp():
ctx = build_ctx()
ev = make_evidence(
texts={"items": [{"handle": "T1", "content": ""}]},
polylines={"items": [{"handle": "P1", "vertex_count": 3, "closed": False}]},
)
issues = run_deterministic_checks(ctx, ev, log=QUIET)
by_handle = {i.get("handle"): i for i in issues}
assert set(by_handle) == {"T1", "P1", "Z1", "D1"}
assert by_handle["D1"]["handles"] == ["D1", "D2"]
def test_run_passes_window_to_mcp_tools():
ctx = build_ctx()
ev = make_evidence(bbox=(0.0, 0.0, 10.0, 10.0))
run_deterministic_checks(ctx, ev, log=QUIET)
assert ctx.client.arguments_of("find_zero_length")["window"] == ev["window"]
assert ctx.client.arguments_of("find_duplicate_entities")["window"] == ev["window"]
def test_run_without_window_skips_mcp_but_keeps_derived():
ctx = build_ctx()
ev = make_evidence()
ev["window"] = None
ev["texts"] = {"items": [{"handle": "T1", "content": ""}]}
issues = run_deterministic_checks(ctx, ev, log=QUIET)
assert [i["handle"] for i in issues] == ["T1"]
assert ctx.client.calls == []
def test_run_respects_unclosed_switch(monkeypatch):
monkeypatch.setattr(checks.config, "CHECK_UNCLOSED_POLYLINES", False)
ctx = build_ctx()
ev = make_evidence(polylines={"items": [{"handle": "P1", "vertex_count": 4, "closed": False}]})
assert [i["handle"] for i in run_deterministic_checks(ctx, ev, log=QUIET)] == ["Z1", "D1"]
+36
View File
@@ -27,6 +27,26 @@ def test_read_env_missing_returns_empty(monkeypatch):
assert config._read_env("CAD_TEST_DEFINITELY_MISSING_XYZ") == ""
def test_read_bool_defaults_when_unset(monkeypatch):
monkeypatch.delenv("CAD_TEST_BOOL_UNSET", raising=False)
assert config._read_bool("CAD_TEST_BOOL_UNSET", True) is True
assert config._read_bool("CAD_TEST_BOOL_UNSET", False) is False
def test_read_bool_empty_string_uses_default(monkeypatch):
monkeypatch.setenv("CAD_TEST_BOOL_EMPTY", " ")
assert config._read_bool("CAD_TEST_BOOL_EMPTY", True) is True
def test_read_bool_truthy_and_falsy(monkeypatch):
for raw in ("1", "true", "TRUE", "Yes", "on"):
monkeypatch.setenv("CAD_TEST_BOOL", raw)
assert config._read_bool("CAD_TEST_BOOL", False) is True
for raw in ("0", "false", "no", "off", "nope"):
monkeypatch.setenv("CAD_TEST_BOOL", raw)
assert config._read_bool("CAD_TEST_BOOL", True) is False
def test_has_api_key_reflects_module_value(monkeypatch):
monkeypatch.setattr(config, "API_KEY", "")
assert config.has_api_key() is False
@@ -50,6 +70,22 @@ def test_constants_have_expected_types():
assert isinstance(config.CHECKLIST_PATH, type(config.PROJECT_ROOT))
def test_credibility_constants_have_expected_types():
assert isinstance(config.DETERMINISTIC_CHECKS, bool)
assert isinstance(config.VALIDATE_HANDLES, bool)
assert isinstance(config.CHECK_UNCLOSED_POLYLINES, bool)
assert isinstance(config.ZERO_LENGTH_TOLERANCE, float)
assert isinstance(config.DUP_TOLERANCE, float)
assert isinstance(config.DUP_GROUP_LIMIT, int)
def test_credibility_defaults_are_on():
# 默认开启,保证默认拿到可信度收益
assert config.DETERMINISTIC_CHECKS is True
assert config.VALIDATE_HANDLES is True
assert config.CHECK_UNCLOSED_POLYLINES is True
def test_checklist_path_resolves_relative_to_root():
assert config.CHECKLIST_PATH.name == "review_checklist.json"
assert config.CHECKLIST_PATH.parent == config.PROJECT_ROOT
+241
View File
@@ -0,0 +1,241 @@
"""`dwg_review.findings` 单元测试:结构化解析 + handle 交叉校验 + 装配。"""
from dwg_review.findings import (
build_finding,
collect_known_handles,
extract_json_block,
flatten_issues,
normalize_issue,
parse_conclusion,
)
from tests.fakes import make_evidence
QUIET = lambda *a: None # noqa: E731
# ---------------------------------------------------------------------------
# extract_json_block
# ---------------------------------------------------------------------------
def test_extract_bare_object():
assert extract_json_block('{"a": 1}') == '{"a": 1}'
def test_extract_bare_array():
assert extract_json_block('[{"a": 1}]') == '[{"a": 1}]'
def test_extract_fenced_object():
assert extract_json_block('```json\n{"a": 1}\n```') == '{"a": 1}'
def test_extract_single_line_fence():
# 旧 extract_issues 的已知瑕疵(单行围栏剥离失败)在此不再复现
assert extract_json_block('```json{"a": 1}```') == '{"a": 1}'
def test_extract_from_surrounding_prose():
text = '好的,结论如下:\n{"summary": "ok", "issues": []}\n以上。'
assert extract_json_block(text) == '{"summary": "ok", "issues": []}'
def test_extract_handles_braces_inside_strings():
text = 'prefix {"message": "包含 } 和 { 的文字", "n": 1} suffix'
assert extract_json_block(text) == '{"message": "包含 } 和 { 的文字", "n": 1}'
def test_extract_returns_none_for_garbage():
assert extract_json_block("未发现明显问题") is None
assert extract_json_block("") is None
assert extract_json_block(None) is None
assert extract_json_block("{ not json") is None
# ---------------------------------------------------------------------------
# parse_conclusion
# ---------------------------------------------------------------------------
def test_parse_object_with_issues():
text = '{"summary": "整体尚可", "issues": [{"category": "几何", "message": "x"}]}'
out = parse_conclusion(text, log=QUIET)
assert out["summary"] == "整体尚可"
assert out["issues"] == [{"category": "几何", "message": "x"}]
assert out["raw"] == text
def test_parse_object_without_issues_key():
out = parse_conclusion('{"summary": "无问题"}', log=QUIET)
assert out["summary"] == "无问题"
assert out["issues"] == []
def test_parse_bare_array_becomes_issues():
out = parse_conclusion('[{"message": "a"}, {"message": "b"}]', log=QUIET)
assert out["summary"] == ""
assert [i["message"] for i in out["issues"]] == ["a", "b"]
def test_parse_filters_non_dict_issues():
out = parse_conclusion('{"summary": "s", "issues": [{"message": "ok"}, "junk", 3]}', log=QUIET)
assert out["issues"] == [{"message": "ok"}]
def test_parse_degrades_prose_to_summary():
logs = []
out = parse_conclusion("未发现明显问题。", log=logs.append)
assert out["summary"] == "未发现明显问题。"
assert out["issues"] == []
assert any("降级" in line for line in logs)
def test_parse_degrades_empty_text():
out = parse_conclusion("", log=QUIET)
assert out["summary"] == "" and out["issues"] == []
# ---------------------------------------------------------------------------
# collect_known_handles
# ---------------------------------------------------------------------------
def test_collect_known_handles_from_all_sources():
ev = make_evidence(
entities={"total": 1, "items": [{"handle": "AA"}]},
texts={"total": 1, "items": [{"handle": "BB"}]},
dimensions={"total": 1, "items": [{"handle": "CC"}]},
blocks={"total": 1, "items": [{"handle": "DD"}]},
polylines={"total": 1, "items": [{"handle": "EE"}]},
)
assert collect_known_handles(ev) == {"AA", "BB", "CC", "DD", "EE"}
def test_collect_known_handles_uppercases():
ev = make_evidence(entities={"items": [{"handle": "2f3a"}]})
assert collect_known_handles(ev) == {"2F3A"}
def test_collect_known_handles_includes_extra():
ev = make_evidence()
assert collect_known_handles(ev, extra_handles=["Ff", None, ""]) == {"FF"}
def test_collect_known_handles_tolerates_non_dict_envelope():
# result_json 解析失败会返回原始字符串
assert collect_known_handles({"entities": "oops", "texts": None}) == set()
def test_collect_known_handles_skips_items_without_handle():
ev = make_evidence(entities={"items": [{"type": "LINE"}, {"handle": ""}]})
assert collect_known_handles(ev) == set()
# ---------------------------------------------------------------------------
# normalize_issue
# ---------------------------------------------------------------------------
def test_normalize_llm_issue_with_known_handle_is_verified():
out = normalize_issue(
{"category": "文字", "severity": "high", "message": "重叠", "handle": "2f3a"},
frame_index=1, source="llm", known_handles={"2F3A"}, validate_handles=True,
)
assert out["handle"] == "2f3a" # 保留原样(大小写归一仅用于比对)
assert out["verified"] is True
assert out["source"] == "llm"
assert out["frame"] == 1
def test_normalize_llm_issue_with_unknown_handle_strips_it():
out = normalize_issue(
{"message": "疑似问题", "handle": "DEAD"},
frame_index=2, source="llm", known_handles={"2F3A"}, validate_handles=True,
)
assert out["handle"] is None
assert out["verified"] is False
assert out["message"] == "疑似问题" # 问题保留
def test_normalize_deterministic_never_stripped():
out = normalize_issue(
{"message": "零长度", "handle": "ZZZZ"},
frame_index=1, source="deterministic", known_handles=set(), validate_handles=True,
)
assert out["handle"] == "ZZZZ" and out["verified"] is True
def test_normalize_validation_disabled_keeps_handle():
out = normalize_issue(
{"message": "x", "handle": "DEAD"},
frame_index=1, source="llm", known_handles=set(), validate_handles=False,
)
assert out["handle"] == "DEAD" and out["verified"] is True
def test_normalize_llm_issue_without_handle_is_verified():
out = normalize_issue({"message": "整体偏空"}, frame_index=1, source="llm",
known_handles=set(), validate_handles=True)
assert out["handle"] is None and out["verified"] is True
def test_normalize_falls_back_on_invalid_severity_and_category():
out = normalize_issue({"severity": "critical", "message": "x"},
frame_index=1, source="llm", known_handles=set(),
validate_handles=True)
assert out["severity"] == "low"
assert out["category"] == "图面整体"
def test_normalize_preserves_handles_list():
out = normalize_issue({"message": "重复组", "handle": "A", "handles": ["A", "B"]},
frame_index=1, source="deterministic", known_handles=set(),
validate_handles=True)
assert out["handles"] == ["A", "B"]
# ---------------------------------------------------------------------------
# build_finding
# ---------------------------------------------------------------------------
def test_build_finding_merges_deterministic_then_llm():
ev = make_evidence(index=3, entities={"items": [{"handle": "AA"}]})
checks = [{"category": "几何", "severity": "medium", "message": "零长度", "handle": "ZZ"}]
conclusion = {
"summary": "有 2 处问题",
"issues": [
{"category": "文字", "severity": "high", "message": "重叠", "handle": "AA"},
{"category": "几何", "severity": "low", "message": "臆测", "handle": "NOPE"},
],
}
finding = build_finding(ev, conclusion, checks, validate_handles=True, log=QUIET)
assert finding["frame"] == 3
assert finding["sheet"] == "A1"
assert finding["summary"] == "有 2 处问题"
assert [i["source"] for i in finding["issues"]] == ["deterministic", "llm", "llm"]
assert finding["counts"] == {"deterministic": 1, "llm": 2, "unverified": 1}
# 被剥掉 handle 的那条仍在
assert finding["issues"][2]["verified"] is False
def test_build_finding_deterministic_handle_counts_as_known():
# 确定性检查命中的 handle 应进入已知集合,供同框 LLM 引用
ev = make_evidence(entities={"items": []})
checks = [{"message": "零长度", "handle": "ZZ"}]
conclusion = {"summary": "", "issues": [{"message": "同一条", "handle": "zz"}]}
finding = build_finding(ev, conclusion, checks, validate_handles=True, log=QUIET)
assert finding["issues"][1]["verified"] is True
def test_build_finding_without_checks_or_conclusion():
finding = build_finding(make_evidence(), {}, None, log=QUIET)
assert finding["issues"] == []
assert finding["counts"] == {"deterministic": 0, "llm": 0, "unverified": 0}
# ---------------------------------------------------------------------------
# flatten_issues
# ---------------------------------------------------------------------------
def test_flatten_issues_concatenates():
findings = [
{"issues": [{"message": "a"}, {"message": "b"}]},
{"issues": [{"message": "c"}]},
{},
]
assert [i["message"] for i in flatten_issues(findings)] == ["a", "b", "c"]
def test_flatten_issues_empty():
assert flatten_issues([]) == []
assert flatten_issues(None) == []