feat(review): 取证层多源化 + 视觉多图（整体+2x2 象限）

- 新增 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 归一

conftest.py

@@ -15,11 +15,12 @@ if str(_ROOT) not in sys.path:
 
 
 @pytest.fixture(autouse=True)
-def _no_render_wait(monkeypatch):
-    """消除 collect_frame 里每个图框的 RENDER_WAIT 等待（单元测试不需要真实渲染）。"""
+def _fast_defaults(monkeypatch):
+    """单元测试默认：不等待渲染、取证只取整体单图（多图用例自行覆盖 tiles）。"""
     from dwg_review import config
 
     monkeypatch.setattr(config, "RENDER_WAIT", 0)
+    monkeypatch.setattr(config, "EVIDENCE_DETAIL_TILES", 0)
 
 
 @pytest.fixture

dwg_review/__main__.py

@@ -11,6 +11,7 @@ from . import findings as findings_mod
 from .annotator import annotate, clear
 from .checklist import build_system_prompt, checklist_summary
 from .context import ReviewContext
+from .evidence_model import evidence_images
 from .mcp_client import McpError, McpToolError, result_json
 from .phases import evidence as evidence_phase
 from .phases import planning, report, review
@@ -20,7 +21,8 @@ def _dump_evidence(evidences):
     out_dir = Path(config.REPORT_DIR)
     out_dir.mkdir(parents=True, exist_ok=True)
     path = out_dir / "evidence.json"
-    trimmed = [{k: v for k, v in e.items() if k != "image_b64"} for e in evidences]
+    drop = ("images", "image_b64", "image_mime")
+    trimmed = [{k: v for k, v in e.items() if k not in drop} for e in evidences]
     path.write_text(json.dumps(trimmed, ensure_ascii=False, indent=2), encoding="utf-8")
     return path
 
@@ -67,14 +69,14 @@ def run_review(ctx, args):
             f"  - 图框 {idx}/{len(frames)} kind={frame.get('kind')} "
             f"sheet={frame.get('sheet_size')}"
         )
-        ev = evidence_phase.collect_frame(ctx, idx, frame)
+        ev = evidence_phase.collect(ctx, idx, frame)
         if ev is None:
             print("    [跳过] 缺少有效包围盒")
             continue
         evidences.append(ev)
         print(
             f"    图元 {(ev.get('entities') or {}).get('total')} 条，"
-            f"截图 {'有' if ev.get('image_b64') else '无'}"
+            f"截图 {len(evidence_images(ev))} 张"
         )
 
     if args.plan_only:

dwg_review/config.py

@@ -121,6 +121,11 @@ IMAGE_FORMAT = os.environ.get("IMAGE_FORMAT", "png").strip().lower()
 # JPEG 质量（仅 IMAGE_FORMAT=jpeg 时生效）
 IMAGE_JPEG_QUALITY = int(os.environ.get("IMAGE_JPEG_QUALITY", "80"))
 
+# 取证阶段每图框的局部截图数（0=仅整体；4=整体+2x2 象限）
+EVIDENCE_DETAIL_TILES = int(os.environ.get("EVIDENCE_DETAIL_TILES", "4"))
+# 每图框图片硬上限（整体 + 局部总张数）
+EVIDENCE_MAX_IMAGES = int(os.environ.get("EVIDENCE_MAX_IMAGES", "5"))
+
 
 def has_api_key() -> bool:
     return bool(API_KEY.strip())

dwg_review/evidence_model.py

@@ -0,0 +1,88 @@
+"""统一证据模型：多来源（CAD / PDF / 图片）证据的字段常量与访问器。
+
+设计要点：
+  - evidence 是跨阶段 + 落盘的「线格式」数据，统一用 dict 承载（不用 dataclass）。
+  - 图片统一为 `images: [{"b64","mime","label"}]`；访问器兼容旧的 `image_b64/image_mime`
+    单图字段（历史 evidence.json 与旧测试构造仍可读）。
+  - 本模块不含任何 CAD / 工具逻辑，可被 phases、findings、__main__、tests 共同引用。
+"""
+
+SOURCE_CAD = "cad"
+SOURCE_PDF = "pdf"
+SOURCE_IMAGE = "image"
+
+# 结构化证据键（CAD 侧各查询工具产出的信封）；下游 handle 收集 / 派生检查统一引用，
+# 避免在多处硬编码同一组键名。
+STRUCTURED_SOURCES = ("entities", "texts", "dimensions", "blocks", "polylines")
+
+
+def make_image(b64, mime="image/png", label=None):
+    """构造 images 列表元素。"""
+    return {"b64": b64, "mime": mime or "image/png", "label": label}
+
+
+def source_type_of(container):
+    """读取来源类型，缺省视为 CAD。
+
+    container 可以是审查单元（unit）/ 图框 / evidence 等任何 dict。
+    """
+    if isinstance(container, dict):
+        value = container.get("source_type")
+        if isinstance(value, str) and value:
+            return value
+    return SOURCE_CAD
+
+
+def evidence_images(evidence):
+    """规范化证据里的图片列表（单点归一）。
+
+    优先读 `images`；否则回退旧的 `image_b64` / `image_mime` 单图字段。
+    统一返回 `[{"b64","mime","label"}]`，过滤掉空 b64。
+    """
+    if not isinstance(evidence, dict):
+        return []
+    images = []
+    raw = evidence.get("images")
+    if isinstance(raw, list):
+        for item in raw:
+            if not isinstance(item, dict):
+                continue
+            b64 = item.get("b64")
+            if not b64:
+                continue
+            images.append(
+                {
+                    "b64": b64,
+                    "mime": item.get("mime") or "image/png",
+                    "label": item.get("label"),
+                }
+            )
+    if images:
+        return images
+    # 回退：旧的单图字段
+    b64 = evidence.get("image_b64")
+    if b64:
+        return [
+            {
+                "b64": b64,
+                "mime": evidence.get("image_mime") or "image/png",
+                "label": None,
+            }
+        ]
+    return []
+
+
+def primary_image(evidence):
+    """返回首图（label == "overall" 优先），无图返回 None。"""
+    images = evidence_images(evidence)
+    if not images:
+        return None
+    for img in images:
+        if img.get("label") == "overall":
+            return img
+    return images[0]
+
+
+def image_labels(evidence):
+    """返回图片标签列表（用于日志 / payload 说明）。"""
+    return [img.get("label") for img in evidence_images(evidence)]

dwg_review/phases/evidence.py

@@ -1,14 +1,20 @@
-"""阶段 2：逐图框取证（确定性、串行）。
+"""阶段 2：逐审查单元取证（多源证据层 + 视觉多图）。
 
-对每个图框：缩放到该框 → 取结构化数据（实体/文字/标注/块/多段线）→ 截取视口。
+对每个审查单元：取结构化数据（实体/文字/标注/块/多段线）+ 视觉截图（整体 + 可选局部）。
 结构化数据是权威依据，截图仅作视觉上下文。
+
+多源设计：`collect()` 按审查单元来源分发；当前仅实现 CAD 源，PDF / 图片源预留 seam
+（未来在对应分支产出**同形** evidence，下游无需改动）。
 """
 
+import math
 import time
 
-from .. import config
+from .. import config, evidence_model
 from ..mcp_client import result_images, result_json
 
+_LABEL_OVERALL = "overall"
+
 
 def bbox_to_window(bbox):
     """[minx,miny,maxx,maxy] -> window 对象。"""
@@ -20,8 +26,77 @@ def bbox_to_window(bbox):
     }
 
 
+def sub_windows(window, tiles):
+    """把 window 等分为 tiles 个子窗口（tiles=4 → 2x2 象限）。
+
+    tiles<=1 返回 []。返回 [{"min_x",..,"max_x",..}, ...]，行优先（左上→右上→左下→右下）。
+    """
+    if not tiles or tiles <= 1:
+        return []
+    cols = int(math.ceil(math.sqrt(tiles)))
+    rows = int(math.ceil(tiles / cols))
+    min_x, min_y = window["min_x"], window["min_y"]
+    max_x, max_y = window["max_x"], window["max_y"]
+    w = (max_x - min_x) / cols
+    h = (max_y - min_y) / rows
+    out = []
+    for r in range(rows):
+        for c in range(cols):
+            if len(out) >= tiles:
+                break
+            out.append(
+                {
+                    "min_x": min_x + c * w,
+                    "min_y": max_y - (r + 1) * h,
+                    "max_x": min_x + (c + 1) * w,
+                    "max_y": max_y - r * h,
+                }
+            )
+    return out
+
+
+def _zoom_and_shot(ctx, window, margin=0.02):
+    """缩放到窗口、等待渲染、截一张图，返回 (b64, mime) 或 None。"""
+    ctx.tools.call("zoom_to_window", dict(window, margin=margin))
+    time.sleep(config.RENDER_WAIT)
+    shot = ctx.tools.call("get_viewport_screenshot", {})
+    images = result_images(shot)
+    if not images:
+        return None
+    return images[0]
+
+
+def capture_images(ctx, window, *, tiles=0, max_images=None):
+    """采集视觉证据：整体一张 + 局部 tiles 张。
+
+    - 整体 label="overall"，局部 label="tile-1"..；行优先。
+    - 结束时**复位到整体视口**，保证后续流程从整幅视图起手。
+    - 受 max_images（默认 config.EVIDENCE_MAX_IMAGES）截断。
+    """
+    if max_images is None:
+        max_images = config.EVIDENCE_MAX_IMAGES
+    images = []
+
+    def _add(shot, label):
+        if shot and len(images) < max_images:
+            images.append(evidence_model.make_image(shot[0], shot[1], label))
+
+    # 整体
+    _add(_zoom_and_shot(ctx, window), _LABEL_OVERALL)
+
+    # 局部（等分象限）
+    for i, sub in enumerate(sub_windows(window, tiles), 1):
+        if len(images) >= max_images:
+            break
+        _add(_zoom_and_shot(ctx, sub), f"tile-{i}")
+
+    # 复位整体视口（不截图）
+    ctx.tools.call("zoom_to_window", dict(window, margin=0.02))
+    return images
+
+
 def collect_frame(ctx, index, frame):
-    """采集单个图框的证据；bbox 无效时返回 None。"""
+    """采集单个图框（CAD 源）的证据；bbox 无效时返回 None。"""
     bbox = frame.get("bbox")
     if not bbox or len(bbox) != 4:
         return None
@@ -29,11 +104,7 @@ def collect_frame(ctx, index, frame):
     win = bbox_to_window(bbox)
     limit = ctx.frame_limit
 
-    # a. 缩放到图框（给 CAD 留渲染时间）
-    ctx.tools.call("zoom_to_window", dict(win, margin=0.02))
-    time.sleep(config.RENDER_WAIT)
-
-    # b. 结构化数据（权威）
+    # a. 结构化数据（权威；按 window 查询，不依赖缩放）
     entities = result_json(ctx.tools.call("query_entities", {"window": win, "limit": limit}), {})
     texts = result_json(ctx.tools.call("get_text_content", {"window": win, "limit": limit}), {})
     dimensions = result_json(ctx.tools.call("get_dimensions", {"window": win, "limit": limit}), {})
@@ -44,12 +115,11 @@ def collect_frame(ctx, index, frame):
         ctx.tools.call("get_polylines_info", {"window": win, "limit": limit}), {}
     )
 
-    # c. 视口截图（视觉上下文）
-    shot = ctx.tools.call("get_viewport_screenshot", {})
-    images = result_images(shot)
-    image_b64, image_mime = images[0] if images else (None, None)
+    # b. 视觉：整体 + 可选局部（多图）
+    images = capture_images(ctx, win, tiles=config.EVIDENCE_DETAIL_TILES)
 
     return {
+        "source_type": evidence_model.source_type_of(frame),
         "index": index,
         "frame": frame,
         "window": win,
@@ -58,6 +128,18 @@ def collect_frame(ctx, index, frame):
         "dimensions": dimensions,
         "blocks": blocks,
         "polylines": polylines,
-        "image_b64": image_b64,
-        "image_mime": image_mime,
+        "images": images,
     }
+
+
+def collect(ctx, index, unit):
+    """按审查单元来源分发取证。
+
+    当前仅实现 CAD 源；PDF / 图片源预留（未来在对应分支接入离线解析 / 直接读图，
+    产出与 CAD **同形**的 evidence：window=None、结构化键给空信封、images 带 label）。
+    """
+    source = evidence_model.source_type_of(unit)
+    if source == evidence_model.SOURCE_CAD:
+        return collect_frame(ctx, index, unit)
+    # TODO(pdf/image): 接入离线解析（DWG/PDF）与直接读图。
+    return None

dwg_review/phases/review.py

@@ -11,6 +11,7 @@
 
 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
@@ -24,6 +25,7 @@ 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)
@@ -72,9 +74,11 @@ def review_frame(ctx, evidence, system_prompt, log=print):
             + json.dumps(payload, ensure_ascii=False, indent=2),
         }
     ]
-    image_b64 = evidence.get("image_b64")
-    if image_b64:
-        nb, nm = downsample_image_b64(image_b64, evidence.get("image_mime") or "image/png")
+    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 = [

tests/fakes.py

@@ -10,6 +10,7 @@
 
 import json
 
+from dwg_review import evidence_model
 from dwg_review.cad_tools import CadTools
 from dwg_review.checklist import DEFAULT_CHECKLIST
 from dwg_review.context import ReviewContext
@@ -313,10 +314,23 @@ def make_evidence(
     bbox=(0.0, 0.0, 100.0, 50.0),
     image_b64=None,
     image_mime=None,
+    images=None,
+    source_type=None,
     **over,
 ):
-    """构造 collect_frame 的返回值（evidence）。"""
+    """构造 collect_frame 的返回值（evidence）。
+
+    兼容旧的 `image_b64`/`image_mime` 单图入参：若给了 image_b64 而未给 images，
+    则归一进 `images` 列表（与 `evidence_model.evidence_images` 的归一保持一致）。
+    """
+    if images is None:
+        images = []
+        if image_b64:
+            images.append(
+                evidence_model.make_image(image_b64, image_mime or "image/png")
+            )
     ev = {
+        "source_type": source_type or evidence_model.SOURCE_CAD,
         "index": index,
         "frame": frame(bbox),
         "window": {
@@ -330,8 +344,7 @@ def make_evidence(
         "dimensions": {"total": 0, "items": []},
         "blocks": {"total": 0, "items": []},
         "polylines": {"total": 0, "items": []},
-        "image_b64": image_b64,
-        "image_mime": image_mime,
+        "images": images,
     }
     ev.update(over)
     return ev

tests/test_evidence.py

@@ -55,8 +55,8 @@ def test_collect_frame_assembles_evidence():
     assert ev["dimensions"]["total"] == 2
     assert ev["blocks"]["total"] == 0
     assert ev["polylines"]["total"] == 3
-    assert ev["image_b64"] == "QUJD"
-    assert ev["image_mime"] == "image/png"
+    assert ev["source_type"] == "cad"
+    assert ev["images"] == [{"b64": "QUJD", "mime": "image/png", "label": "overall"}]
 
 
 def test_collect_frame_calls_expected_tools():
@@ -80,12 +80,64 @@ def test_collect_frame_calls_expected_tools():
 def test_collect_frame_without_image():
     ctx = make_ctx(client=build_evidence_client(image=ok_text({})))
     ev = collect_frame(ctx, 1, frame([0, 0, 10, 10]))
-    assert ev["image_b64"] is None
-    assert ev["image_mime"] is None
+    assert ev["images"] == []
 
 
 def test_collect_frame_tolerates_empty_structured_data():
     ctx = make_ctx(client=FakeMcpClient())  # 全部默认 ok_text({})
     ev = collect_frame(ctx, 1, frame([0, 0, 10, 10]))
     assert ev["entities"] == {}
-    assert ev["image_b64"] is None
+    assert ev["images"] == []
+
+
+def test_sub_windows_2x2():
+    from dwg_review.phases.evidence import sub_windows
+
+    win = {"min_x": 0.0, "min_y": 0.0, "max_x": 10.0, "max_y": 10.0}
+    subs = sub_windows(win, 4)
+    assert len(subs) == 4
+    # 行优先：左上 → 右上 → 左下 → 右下
+    assert subs[0] == {"min_x": 0.0, "min_y": 5.0, "max_x": 5.0, "max_y": 10.0}
+    assert subs[1] == {"min_x": 5.0, "min_y": 5.0, "max_x": 10.0, "max_y": 10.0}
+    assert subs[2] == {"min_x": 0.0, "min_y": 0.0, "max_x": 5.0, "max_y": 5.0}
+    assert subs[3] == {"min_x": 5.0, "min_y": 0.0, "max_x": 10.0, "max_y": 5.0}
+
+
+def test_sub_windows_disabled():
+    from dwg_review.phases.evidence import sub_windows
+
+    win = {"min_x": 0.0, "min_y": 0.0, "max_x": 10.0, "max_y": 10.0}
+    assert sub_windows(win, 0) == []
+    assert sub_windows(win, 1) == []
+
+
+def test_collect_frame_captures_multiple_images(monkeypatch):
+    from dwg_review import config
+    from dwg_review.phases.evidence import collect_frame
+
+    monkeypatch.setattr(config, "EVIDENCE_DETAIL_TILES", 4)
+    ctx = make_ctx(client=build_evidence_client())
+    ev = collect_frame(ctx, 1, frame([0, 0, 100, 50]))
+    assert [img["label"] for img in ev["images"]] == [
+        "overall",
+        "tile-1",
+        "tile-2",
+        "tile-3",
+        "tile-4",
+    ]
+
+
+def test_collect_dispatches_cad_source():
+    from dwg_review.phases.evidence import collect
+
+    ctx = make_ctx(client=build_evidence_client())
+    ev = collect(ctx, 1, frame([0, 0, 10, 10]))
+    assert ev is not None
+    assert ev["source_type"] == "cad"
+
+
+def test_collect_returns_none_for_unsupported_source():
+    from dwg_review.phases.evidence import collect
+
+    ctx = make_ctx(client=FakeMcpClient())
+    assert collect(ctx, 1, {"source_type": "pdf", "bbox": [0, 0, 1, 1]}) is None

tests/test_evidence_model.py

@@ -0,0 +1,100 @@
+"""`dwg_review.evidence_model` 单元测试。"""
+
+from dwg_review import evidence_model as em
+
+
+def test_make_image_defaults():
+    assert em.make_image("QUJD") == {"b64": "QUJD", "mime": "image/png", "label": None}
+
+
+def test_make_image_custom():
+    assert em.make_image("QUJD", "image/jpeg", "tile-1") == {
+        "b64": "QUJD",
+        "mime": "image/jpeg",
+        "label": "tile-1",
+    }
+
+
+def test_make_image_empty_mime_falls_back():
+    assert em.make_image("x", "")["mime"] == "image/png"
+
+
+def test_source_type_of_default_cad():
+    assert em.source_type_of({}) == em.SOURCE_CAD
+    assert em.source_type_of(None) == em.SOURCE_CAD
+    assert em.source_type_of({"source_type": ""}) == em.SOURCE_CAD
+
+
+def test_source_type_of_explicit():
+    assert em.source_type_of({"source_type": "pdf"}) == "pdf"
+    assert em.source_type_of({"source_type": "image"}) == "image"
+
+
+def test_evidence_images_new_form():
+    ev = {
+        "images": [
+            em.make_image("A", label="overall"),
+            em.make_image("B", label="tile-1"),
+        ]
+    }
+    out = em.evidence_images(ev)
+    assert [i["b64"] for i in out] == ["A", "B"]
+    assert [i["label"] for i in out] == ["overall", "tile-1"]
+
+
+def test_evidence_images_filters_empty_b64():
+    ev = {"images": [{"b64": "", "mime": "image/png"}, {"b64": "A"}]}
+    out = em.evidence_images(ev)
+    assert len(out) == 1
+    assert out[0]["b64"] == "A"
+
+
+def test_evidence_images_legacy_fallback():
+    ev = {"image_b64": "LEG", "image_mime": "image/jpeg"}
+    assert em.evidence_images(ev) == [
+        {"b64": "LEG", "mime": "image/jpeg", "label": None}
+    ]
+
+
+def test_evidence_images_legacy_default_mime():
+    assert em.evidence_images({"image_b64": "LEG"})[0]["mime"] == "image/png"
+
+
+def test_evidence_images_empty():
+    assert em.evidence_images({}) == []
+    assert em.evidence_images(None) == []
+    assert em.evidence_images({"images": []}) == []
+
+
+def test_evidence_images_prefers_new_over_legacy():
+    ev = {"images": [em.make_image("NEW")], "image_b64": "OLD"}
+    assert [i["b64"] for i in em.evidence_images(ev)] == ["NEW"]
+
+
+def test_primary_image_prefers_overall():
+    ev = {
+        "images": [
+            em.make_image("A", label="tile-1"),
+            em.make_image("B", label="overall"),
+        ]
+    }
+    assert em.primary_image(ev)["b64"] == "B"
+
+
+def test_primary_image_first_when_no_overall():
+    ev = {"images": [em.make_image("A", label="tile-1"), em.make_image("B")]}
+    assert em.primary_image(ev)["b64"] == "A"
+
+
+def test_primary_image_none_when_empty():
+    assert em.primary_image({}) is None
+
+
+def test_image_labels():
+    ev = {
+        "images": [
+            em.make_image("A", label="overall"),
+            em.make_image("B", label="tile-1"),
+        ]
+    }
+    assert em.image_labels(ev) == ["overall", "tile-1"]

tests/test_review.py

@@ -28,8 +28,8 @@ from tests.fakes import (
 def test_evidence_payload_strips_image():
     ev = make_evidence(image_b64="QUJD", image_mime="image/png")
     payload = evidence_payload(ev)
+    assert "images" not in payload
     assert "image_b64" not in payload
-    assert "image_mime" not in payload
     assert payload["frame_index"] == 1
 
 
@@ -243,3 +243,20 @@ def test_review_frame_intermediate_tool_turn_json_not_parsed():
     out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None)
     assert out["summary"] == "最终结论"
     assert out["issues"] == []
+
+
+def test_review_frame_includes_multiple_image_blocks():
+    llm = FakeLlmClient(script=[llm_text("ok")])
+    ctx = make_ctx(client=FakeMcpClient(), llm=llm)
+    ev = make_evidence(
+        images=[
+            {"b64": "A", "mime": "image/png", "label": "overall"},
+            {"b64": "B", "mime": "image/png", "label": "tile-1"},
+        ]
+    )
+    review_frame(ctx, ev, "sys", log=lambda *a: None)
+    user_content = llm.calls[0]["messages"][1]["content"]
+    image_blocks = [b for b in user_content if b.get("type") == "image_url"]
+    assert len(image_blocks) == 2
+    assert any(b.get("text") == "（截图：overall）" for b in user_content)
+    assert any(b.get("text") == "（截图：tile-1）" for b in user_content)