158 lines
5.5 KiB
Python
158 lines
5.5 KiB
Python
"""视觉模型调用模块 -- 两轮 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以下是另一个模型基于这些图片生成的描述，请核实：\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[验证] 跳过（调用失败）")
|