From e3b027ce34a0a810639338b746068e470672d45d Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 7 Jul 2026 01:49:55 -0400 Subject: [PATCH] =?UTF-8?q?feat(adapters):=20GovernedVLMClient=20=E2=80=94?= =?UTF-8?q?=20VLMProvider=20=E6=9C=80=E5=B0=8F=E5=8F=AF=E7=94=A8=E5=AE=9E?= =?UTF-8?q?=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 包装 GovernedLLMClient，注入 base64 图片到 OpenAI Vision API 格式 - 复用 LLM 治理栈全部能力（熔断、缓存、重试、遥测） - 8 项单元测试覆盖协议满足、图片编码、注入逻辑、不可变性 --- adapters/vlm.py | 128 +++++++++++++++++++++++++++++++++ tests/unit/test_vlm_adapter.py | 73 +++++++++++++++++++ 2 files changed, 201 insertions(+) create mode 100644 adapters/vlm.py create mode 100644 tests/unit/test_vlm_adapter.py diff --git a/adapters/vlm.py b/adapters/vlm.py new file mode 100644 index 0000000..cf3342a --- /dev/null +++ b/adapters/vlm.py @@ -0,0 +1,128 @@ +"""GovernedVLMClient -- VLMProvider 最小可用实现。 + +将图片编码为 base64，构造 OpenAI Vision API 格式的 messages， +委托给已有的 GovernedLLMClient 发送。复用 LLM 治理栈的全部能力 +（熔断、缓存、重试、遥测）。 +""" + +from __future__ import annotations + +import base64 +import mimetypes +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from loguru import logger + +if TYPE_CHECKING: + from adapters.llm import GovernedLLMClient + from core.types import LLMResponse + + +class GovernedVLMClient: + """VLMProvider 实现——包装 GovernedLLMClient，注入 base64 图片。 + + 参数: + governed_llm: 已初始化的 GovernedLLMClient 实例。 + """ + + def __init__(self, governed_llm: GovernedLLMClient) -> None: + self._llm = governed_llm + + 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: + """图文调用：将图片编码为 base64 嵌入 messages，委托给 LLM 客户端。 + + 参数: + messages: 对话消息列表。最后一条 user message 的 content 会被扩展为 + 包含图片的多模态格式。 + images: 图片文件路径列表。 + session_id: 会话 ID（遥测用）。 + parent_call_id: 父调用 ID（遥测用）。 + + 返回: + LLMResponse。 + """ + vision_messages = self._inject_images(messages, images) + return await self._llm.chat( + vision_messages, + session_id=session_id, + parent_call_id=parent_call_id, + ) + + @staticmethod + def _encode_image(image_path: str | Path) -> str: + """将图片文件编码为 base64 data URL。 + + 参数: + image_path: 图片文件路径。 + + 返回: + data:image/;base64, 格式的字符串。 + """ + path = Path(image_path) + mime_type = mimetypes.guess_type(str(path))[0] or "image/jpeg" + with open(path, "rb") as f: + b64 = base64.b64encode(f.read()).decode("utf-8") + return f"data:{mime_type};base64,{b64}" + + @staticmethod + def _inject_images( + messages: list[dict[str, Any]], + images: list[str | Path], + ) -> list[dict[str, Any]]: + """将图片注入最后一条 user message，构造 OpenAI Vision API 格式。 + + 参数: + messages: 原始消息列表。 + images: 图片路径列表。 + + 返回: + 新消息列表（不修改原列表）。 + """ + if not images: + return messages + + result = [m.copy() for m in messages] + + # 找到最后一条 user message + last_user_idx = -1 + for i in range(len(result) - 1, -1, -1): + if result[i].get("role") == "user": + last_user_idx = i + break + + if last_user_idx == -1: + logger.warning("messages 中无 user 角色消息，图片未注入") + return result + + user_msg = result[last_user_idx] + original_content = user_msg.get("content", "") + + # 构造多模态 content + content_parts: list[dict[str, Any]] = [] + + # 图片在前 + for img_path in images: + data_url = GovernedVLMClient._encode_image(img_path) + content_parts.append( + { + "type": "image_url", + "image_url": {"url": data_url}, + } + ) + + # 文本在后 + if isinstance(original_content, str) and original_content: + content_parts.append({"type": "text", "text": original_content}) + elif isinstance(original_content, list): + content_parts.extend(original_content) + + result[last_user_idx] = {**user_msg, "content": content_parts} + return result diff --git a/tests/unit/test_vlm_adapter.py b/tests/unit/test_vlm_adapter.py new file mode 100644 index 0000000..e63b9b2 --- /dev/null +++ b/tests/unit/test_vlm_adapter.py @@ -0,0 +1,73 @@ +"""VLM 适配器单元测试。""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from adapters.vlm import GovernedVLMClient + +if TYPE_CHECKING: + from pathlib import Path + + +class TestGovernedVLMClientProtocol: + def test_has_chat_with_images(self): + assert hasattr(GovernedVLMClient, "chat_with_images") + + def test_satisfies_vlm_protocol(self): + """GovernedVLMClient 应满足 VLMProvider Protocol。""" + assert hasattr(GovernedVLMClient, "chat_with_images") + + +class TestImageEncoding: + def test_encode_jpeg(self, tmp_path: Path): + img = tmp_path / "test.jpg" + img.write_bytes(b"\xff\xd8\xff\xe0fake_jpeg_data") + result = GovernedVLMClient._encode_image(img) + assert result.startswith("data:image/jpeg;base64,") + + def test_encode_png(self, tmp_path: Path): + img = tmp_path / "test.png" + img.write_bytes(b"\x89PNG\r\n\x1a\nfake_png_data") + result = GovernedVLMClient._encode_image(img) + assert result.startswith("data:image/png;base64,") + + +class TestInjectImages: + def test_inject_single_image(self, tmp_path: Path): + img = tmp_path / "frame.jpg" + img.write_bytes(b"\xff\xd8\xff\xe0data") + messages = [{"role": "user", "content": "描述这帧画面"}] + result = GovernedVLMClient._inject_images(messages, [img]) + + assert len(result) == 1 + content = result[0]["content"] + assert isinstance(content, list) + assert len(content) == 2 # 1 image + 1 text + assert content[0]["type"] == "image_url" + assert content[1]["type"] == "text" + assert content[1]["text"] == "描述这帧画面" + + def test_inject_does_not_mutate_original(self, tmp_path: Path): + img = tmp_path / "frame.jpg" + img.write_bytes(b"\xff\xd8\xff\xe0data") + messages = [{"role": "user", "content": "text"}] + original_content = messages[0]["content"] + GovernedVLMClient._inject_images(messages, [img]) + assert messages[0]["content"] == original_content # 原列表未变 + + def test_no_images_passthrough(self): + messages = [{"role": "user", "content": "hello"}] + result = GovernedVLMClient._inject_images(messages, []) + assert result[0]["content"] == "hello" + + def test_multiple_images(self, tmp_path: Path): + imgs = [] + for i in range(3): + img = tmp_path / f"frame_{i}.jpg" + img.write_bytes(b"\xff\xd8\xff\xe0data") + imgs.append(img) + messages = [{"role": "user", "content": "描述"}] + result = GovernedVLMClient._inject_images(messages, imgs) + content = result[0]["content"] + assert len(content) == 4 # 3 images + 1 text