From 3ae3d5ab504f50be1684926d8b19e75c9ec97d6a Mon Sep 17 00:00:00 2001
From: iomgaa <iomgaaycz@gmail.com>
Date: Tue, 7 Jul 2026 05:52:54 -0400
Subject: [PATCH] =?UTF-8?q?feat(search):=20app/search/vision.py=20?=
 =?UTF-8?q?=E2=80=94=20=E4=B8=A4=E8=BD=AE=20VLM=20=E5=B8=A7=E8=A7=82?=
 =?UTF-8?q?=E5=AF=9F=E6=A8=A1=E5=9D=97?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

从 TRM4 core/tree/vision.py 迁移 observe_frame，关键变更：
- VLM 调用走 VLMProvider.chat_with_images Protocol（images 传 Path）
- OCR 调用走 OCRProvider.transcribe_frames 异步 Protocol
- 遥测字段 session_id / parent_call_id 透传
- 帧文件存在性前置校验

12 个单元测试覆盖：两轮正常、仅提取、OCR 注入/失败降级/None、
VLM 提取失败、VLM 验证失败降级、帧缺失、stats 完整性、
分歧/弃权标记、遥测透传。

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 app/search/vision.py             | 159 +++++++++++
 tests/unit/test_search_vision.py | 458 +++++++++++++++++++++++++++++++
 2 files changed, 617 insertions(+)
 create mode 100644 app/search/vision.py
 create mode 100644 tests/unit/test_search_vision.py

diff --git a/app/search/vision.py b/app/search/vision.py
new file mode 100644
index 0000000..b453d6b
--- /dev/null
+++ b/app/search/vision.py
@@ -0,0 +1,159 @@
+"""视觉模型调用模块 -- 两轮 VLM 调用查看关键帧图像。
+
+提取轮：带防幻觉 system prompt，提取原始视觉证据。
+验证轮：把初稿全文喂回，逐条核实并给置信度。
+
+从 TRM4 ``core/tree/vision.py`` 迁移，关键变更：
+- VLM 调用走 ``VLMProvider.chat_with_images`` Protocol，images 传 Path 列表；
+- OCR 调用走 ``OCRProvider.transcribe_frames`` 异步 Protocol；
+- 遥测字段（session_id / parent_call_id）透传给 VLM 调用。
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from loguru import logger
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+    from pathlib import Path
+
+    from app.ports import OCRProvider
+    from core.protocols import VLMProvider
+
+_OCR_PREFIX = (
+    "以下是 OCR 工具对这些帧的文字转录，仅供参考；"
+    "与你实际看到的不一致时，报告双读数并标注分歧：\n"
+)
+
+
+def _load_prompt(prompts_dir: Path, filename: str) -> str:
+    """从 prompts 目录加载 system prompt 文件。
+
+    参数:
+        prompts_dir: prompt 文件所在目录。
+        filename: prompt 文件名。
+
+    返回:
+        文件内容字符串。
+    """
+    return (prompts_dir / filename).read_text(encoding="utf-8")
+
+
+async def observe_frame(
+    vlm: VLMProvider,
+    frame_paths: list[Path],
+    question: str,
+    prompts_dir: Path,
+    *,
+    ocr: OCRProvider | None,
+    verify: bool,
+    stats_sink: Callable[[dict[str, int]], None] | None = None,
+    session_id: str | None = None,
+    parent_call_id: str | None = None,
+) -> str:
+    """调用 VLM 查看帧图像：可选 OCR 事前并置 + 提取轮 + 可选验证轮。
+
+    参数:
+        vlm: VLM 图文调用端口。
+        frame_paths: 帧文件路径列表。
+        question: 针对帧内容的视觉问题。
+        prompts_dir: prompt 文件目录。
+        ocr: 帧文字转录端口（None=不注入；返回空串视为无结果不注入）。
+        verify: 是否执行验证轮（False 时仅提取轮，输出无 [验证] 段）。
+        stats_sink: 统计回调（None 不收集）；统计严禁写入输出文本。
+        session_id: 遥测会话 ID，透传给 VLM 调用。
+        parent_call_id: 遥测父调用 ID，透传给 VLM 调用。
+
+    返回:
+        verify=True 为 ``"[视觉观察] {证据}\\n[验证] {核实结果}"``，
+        verify=False 为 ``"[视觉观察] {证据}"``，或错误信息。
+
+    关键实现细节:
+        OCR 文本作为额外文本并置于问题之前（事前并置——OCR 误读不进
+        工具输出故零 judge 口径风险）；OCR 异常降级为不注入并计
+        ocr_failed（ocr 是外部注入依赖，任何异常都不得中断工具主流程，
+        故此处 except Exception 是刻意的降级边界）。sink 键：
+        ocr_injected / ocr_chars / ocr_failed / discrepancy（输出含"分歧"词面）/
+        abstain（含 [证据不存在]）。
+    """
+    stats: dict[str, int] = {
+        "ocr_injected": 0,
+        "ocr_chars": 0,
+        "ocr_failed": 0,
+        "discrepancy": 0,
+        "abstain": 0,
+    }
+
+    def _emit(output: str) -> str:
+        """计算语义标记并回调 stats_sink。"""
+        stats["abstain"] = int("[证据不存在]" in output)
+        stats["discrepancy"] = int("分歧" in output)
+        if stats_sink is not None:
+            stats_sink(stats)
+        return output
+
+    # -- 帧文件存在性校验 --
+    for p in frame_paths:
+        if not p.exists():
+            return _emit(f"[VL错误] 帧文件不存在: {p}")
+
+    # -- OCR 转录（可选） --
+    ocr_text = ""
+    if ocr is not None:
+        try:
+            ocr_text = await ocr.transcribe_frames(frame_paths)
+        except Exception as e:  # noqa: BLE001 — 刻意的降级边界
+            logger.warning("OCR 转录失败，降级不注入: {}", e)
+            stats["ocr_failed"] = 1
+
+    # -- 拼装提取轮 user 消息 --
+    user_parts: list[str] = []
+    if ocr_text:
+        stats["ocr_injected"] = 1
+        stats["ocr_chars"] = len(ocr_text)
+        user_parts.append(_OCR_PREFIX + ocr_text)
+    user_parts.append(question)
+    user_text = "\n".join(user_parts)
+
+    extract_messages = [
+        {"role": "system", "content": _load_prompt(prompts_dir, "observe_frame_extract.md")},
+        {"role": "user", "content": user_text},
+    ]
+
+    # -- 提取轮 --
+    try:
+        extract_response = await vlm.chat_with_images(
+            extract_messages,
+            images=frame_paths,
+            session_id=session_id,
+            parent_call_id=parent_call_id,
+        )
+        raw_evidence = extract_response.content
+    except Exception as e:  # noqa: BLE001
+        return _emit(f"[VL错误] {e}")
+
+    if not verify:
+        return _emit(f"[视觉观察] {raw_evidence}")
+
+    # -- 验证轮 --
+    verify_text = (
+        f"问题: {question}\n\n"
+        f"以下是另一个模型基于这些图片生成的描述，请核实：\n{raw_evidence}"
+    )
+    verify_messages = [
+        {"role": "system", "content": _load_prompt(prompts_dir, "observe_frame_verify.md")},
+        {"role": "user", "content": verify_text},
+    ]
+    try:
+        verify_response = await vlm.chat_with_images(
+            verify_messages,
+            images=frame_paths,
+            session_id=session_id,
+            parent_call_id=parent_call_id,
+        )
+        return _emit(f"[视觉观察] {raw_evidence}\n[验证] {verify_response.content}")
+    except Exception as e:  # noqa: BLE001
+        logger.warning("验证轮调用失败，跳过: {}", e)
+        return _emit(f"[视觉观察] {raw_evidence}\n[验证] 跳过（调用失败）")
diff --git a/tests/unit/test_search_vision.py b/tests/unit/test_search_vision.py
new file mode 100644
index 0000000..88f3f69
--- /dev/null
+++ b/tests/unit/test_search_vision.py
@@ -0,0 +1,458 @@
+"""observe_frame 单元测试。
+
+FakeVLMProvider / FakeOCRProvider 实现 Protocol 最小集，
+覆盖：两轮正常、verify=False 仅提取、OCR 注入、OCR 失败降级、
+OCR 为 None、VLM 提取失败、VLM 验证失败降级、帧文件不存在、stats 键完整性。
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from core.types import LLMResponse
+
+# ---------------------------------------------------------------------------
+# Fake 实现
+# ---------------------------------------------------------------------------
+
+_DUMMY_RESPONSE_KWARGS = {
+    "thinking": "",
+    "model": "fake-vlm",
+    "provider": "fake",
+    "prompt_tokens": 10,
+    "completion_tokens": 20,
+    "latency_ms": 100,
+    "ttft_ms": None,
+    "max_inter_token_ms": None,
+    "cache_hit": False,
+    "call_id": "fake-call-id",
+}
+
+
+class FakeVLMProvider:
+    """可编程的 VLM 假实现。
+
+    通过 responses 列表按序返回预设内容；raises 列表对应位置不为 None 时抛异常。
+    """
+
+    def __init__(
+        self,
+        responses: list[str] | None = None,
+        raises: list[Exception | None] | None = None,
+    ) -> None:
+        self._responses = responses or []
+        self._raises = raises or []
+        self._call_idx = 0
+        self.calls: list[dict[str, Any]] = []
+
+    async def chat_with_images(
+        self,
+        messages: list[dict[str, Any]],
+        images: list[str | Path],
+        *,
+        session_id: str | None = None,
+        parent_call_id: str | None = None,
+    ) -> LLMResponse:
+        """记录调用并按序返回预设响应或抛出异常。"""
+        idx = self._call_idx
+        self._call_idx += 1
+        self.calls.append(
+            {
+                "messages": messages,
+                "images": images,
+                "session_id": session_id,
+                "parent_call_id": parent_call_id,
+            }
+        )
+        if idx < len(self._raises) and self._raises[idx] is not None:
+            raise self._raises[idx]  # type: ignore[misc]
+        content = self._responses[idx] if idx < len(self._responses) else ""
+        return LLMResponse(content=content, **_DUMMY_RESPONSE_KWARGS)
+
+
+class FakeOCRProvider:
+    """可编程的 OCR 假实现。"""
+
+    def __init__(
+        self,
+        text: str = "",
+        raise_on_call: Exception | None = None,
+    ) -> None:
+        self._text = text
+        self._raise_on_call = raise_on_call
+
+    async def transcribe_frames(self, frame_paths: list[Path]) -> str:
+        """返回预设文本或抛出异常。"""
+        if self._raise_on_call is not None:
+            raise self._raise_on_call
+        return self._text
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture()
+def frame_files(tmp_path: Path) -> list[Path]:
+    """创建两个最小 JPEG 占位帧文件。"""
+    frames: list[Path] = []
+    for i in range(2):
+        p = tmp_path / f"frame_{i}.jpg"
+        p.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 20)
+        frames.append(p)
+    return frames
+
+
+@pytest.fixture()
+def prompts_dir(tmp_path: Path) -> Path:
+    """创建 observe_frame_extract.md / observe_frame_verify.md 占位文件。"""
+    d = tmp_path / "prompts"
+    d.mkdir()
+    (d / "observe_frame_extract.md").write_text("extract prompt", encoding="utf-8")
+    (d / "observe_frame_verify.md").write_text("verify prompt", encoding="utf-8")
+    return d
+
+
+# ---------------------------------------------------------------------------
+# 测试用例
+# ---------------------------------------------------------------------------
+
+
+class TestObserveFrameNormal:
+    """两轮正常执行（verify=True）。"""
+
+    def test_two_round_normal(
+        self, frame_files: list[Path], prompts_dir: Path
+    ) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider(responses=["raw evidence", "verified ok"])
+        collected: list[dict[str, int]] = []
+
+        result = asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=frame_files,
+                question="what happened?",
+                prompts_dir=prompts_dir,
+                ocr=None,
+                verify=True,
+                stats_sink=collected.append,
+            )
+        )
+
+        assert result == "[视觉观察] raw evidence\n[验证] verified ok"
+        assert len(vlm.calls) == 2
+        # 提取轮使用 extract prompt
+        assert vlm.calls[0]["messages"][0]["content"] == "extract prompt"
+        # 验证轮使用 verify prompt
+        assert vlm.calls[1]["messages"][0]["content"] == "verify prompt"
+        assert len(collected) == 1
+
+
+class TestObserveFrameExtractOnly:
+    """verify=False 仅执行提取轮。"""
+
+    def test_extract_only(
+        self, frame_files: list[Path], prompts_dir: Path
+    ) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider(responses=["only extract"])
+
+        result = asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=frame_files,
+                question="q?",
+                prompts_dir=prompts_dir,
+                ocr=None,
+                verify=False,
+            )
+        )
+
+        assert result == "[视觉观察] only extract"
+        assert len(vlm.calls) == 1
+
+
+class TestObserveFrameOCRInjection:
+    """OCR 注入：文本非空时并置于问题前。"""
+
+    def test_ocr_injected(
+        self, frame_files: list[Path], prompts_dir: Path
+    ) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider(responses=["evidence with ocr"])
+        ocr = FakeOCRProvider(text="帧1: 你好世界")
+        collected: list[dict[str, int]] = []
+
+        result = asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=frame_files,
+                question="q?",
+                prompts_dir=prompts_dir,
+                ocr=ocr,
+                verify=False,
+                stats_sink=collected.append,
+            )
+        )
+
+        assert "[视觉观察]" in result
+        # 验证 OCR 文本被注入到 user message
+        user_msg = vlm.calls[0]["messages"][1]["content"]
+        assert "帧1: 你好世界" in user_msg
+        # stats 中 ocr_injected=1
+        assert collected[0]["ocr_injected"] == 1
+        assert collected[0]["ocr_chars"] == len("帧1: 你好世界")
+
+
+class TestObserveFrameOCRFailDegrades:
+    """OCR 转录抛出异常时降级：不注入 OCR、ocr_failed=1。"""
+
+    def test_ocr_failure_degrades(
+        self, frame_files: list[Path], prompts_dir: Path
+    ) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider(responses=["evidence no ocr"])
+        ocr = FakeOCRProvider(raise_on_call=RuntimeError("OCR service down"))
+        collected: list[dict[str, int]] = []
+
+        result = asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=frame_files,
+                question="q?",
+                prompts_dir=prompts_dir,
+                ocr=ocr,
+                verify=False,
+                stats_sink=collected.append,
+            )
+        )
+
+        assert result == "[视觉观察] evidence no ocr"
+        assert collected[0]["ocr_failed"] == 1
+        assert collected[0]["ocr_injected"] == 0
+
+
+class TestObserveFrameOCRNone:
+    """ocr=None 时不执行转录。"""
+
+    def test_ocr_none(
+        self, frame_files: list[Path], prompts_dir: Path
+    ) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider(responses=["no ocr"])
+        collected: list[dict[str, int]] = []
+
+        result = asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=frame_files,
+                question="q?",
+                prompts_dir=prompts_dir,
+                ocr=None,
+                verify=False,
+                stats_sink=collected.append,
+            )
+        )
+
+        assert result == "[视觉观察] no ocr"
+        assert collected[0]["ocr_injected"] == 0
+        assert collected[0]["ocr_chars"] == 0
+        assert collected[0]["ocr_failed"] == 0
+
+
+class TestObserveFrameVLMExtractFailure:
+    """VLM 提取轮失败 → 返回 [VL错误]。"""
+
+    def test_vlm_extract_failure(
+        self, frame_files: list[Path], prompts_dir: Path
+    ) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider(raises=[RuntimeError("VLM timeout")])
+        collected: list[dict[str, int]] = []
+
+        result = asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=frame_files,
+                question="q?",
+                prompts_dir=prompts_dir,
+                ocr=None,
+                verify=True,
+                stats_sink=collected.append,
+            )
+        )
+
+        assert result.startswith("[VL错误]")
+        assert "VLM timeout" in result
+        assert len(collected) == 1
+
+
+class TestObserveFrameVLMVerifyFailureDegrades:
+    """VLM 验证轮失败 → 降级：保留提取结果 + [验证] 跳过。"""
+
+    def test_vlm_verify_failure_degrades(
+        self, frame_files: list[Path], prompts_dir: Path
+    ) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider(
+            responses=["good evidence", ""],
+            raises=[None, RuntimeError("verify timeout")],
+        )
+        collected: list[dict[str, int]] = []
+
+        result = asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=frame_files,
+                question="q?",
+                prompts_dir=prompts_dir,
+                ocr=None,
+                verify=True,
+                stats_sink=collected.append,
+            )
+        )
+
+        assert "[视觉观察] good evidence" in result
+        assert "[验证] 跳过（调用失败）" in result
+        assert len(collected) == 1
+
+
+class TestObserveFrameFileMissing:
+    """帧文件不存在 → 返回 [VL错误] 帧文件不存在。"""
+
+    def test_frame_file_not_found(self, prompts_dir: Path) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider()
+        missing = [Path("/nonexistent/frame_0.jpg")]
+        collected: list[dict[str, int]] = []
+
+        result = asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=missing,
+                question="q?",
+                prompts_dir=prompts_dir,
+                ocr=None,
+                verify=True,
+                stats_sink=collected.append,
+            )
+        )
+
+        assert "[VL错误] 帧文件不存在" in result
+        assert len(collected) == 1
+
+
+class TestObserveFrameStatsKeys:
+    """stats 包含全部五个预期键。"""
+
+    def test_stats_keys_complete(
+        self, frame_files: list[Path], prompts_dir: Path
+    ) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider(responses=["evidence"])
+        collected: list[dict[str, int]] = []
+
+        asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=frame_files,
+                question="q?",
+                prompts_dir=prompts_dir,
+                ocr=None,
+                verify=False,
+                stats_sink=collected.append,
+            )
+        )
+
+        expected_keys = {"ocr_injected", "ocr_chars", "ocr_failed", "discrepancy", "abstain"}
+        assert set(collected[0].keys()) == expected_keys
+
+
+class TestObserveFrameDiscrepancyAndAbstain:
+    """VLM 返回含 '分歧' 或 '[证据不存在]' 时对应 stats 标记。"""
+
+    def test_discrepancy_flag(
+        self, frame_files: list[Path], prompts_dir: Path
+    ) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider(responses=["发现分歧：OCR 与画面不一致"])
+        collected: list[dict[str, int]] = []
+
+        asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=frame_files,
+                question="q?",
+                prompts_dir=prompts_dir,
+                ocr=None,
+                verify=False,
+                stats_sink=collected.append,
+            )
+        )
+
+        assert collected[0]["discrepancy"] == 1
+
+    def test_abstain_flag(
+        self, frame_files: list[Path], prompts_dir: Path
+    ) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider(responses=["[证据不存在] 无法判断"])
+        collected: list[dict[str, int]] = []
+
+        asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=frame_files,
+                question="q?",
+                prompts_dir=prompts_dir,
+                ocr=None,
+                verify=False,
+                stats_sink=collected.append,
+            )
+        )
+
+        assert collected[0]["abstain"] == 1
+
+
+class TestObserveFrameTelemetryPassthrough:
+    """session_id 和 parent_call_id 透传到 VLM 调用。"""
+
+    def test_telemetry_passthrough(
+        self, frame_files: list[Path], prompts_dir: Path
+    ) -> None:
+        from app.search.vision import observe_frame
+
+        vlm = FakeVLMProvider(responses=["evidence"])
+
+        asyncio.run(
+            observe_frame(
+                vlm=vlm,
+                frame_paths=frame_files,
+                question="q?",
+                prompts_dir=prompts_dir,
+                ocr=None,
+                verify=False,
+                session_id="sess-123",
+                parent_call_id="parent-456",
+            )
+        )
+
+        assert vlm.calls[0]["session_id"] == "sess-123"
+        assert vlm.calls[0]["parent_call_id"] == "parent-456"