feat(adapters): GovernedVLMClient — VLMProvider 最小可用实现
- 包装 GovernedLLMClient，注入 base64 图片到 OpenAI Vision API 格式 - 复用 LLM 治理栈全部能力（熔断、缓存、重试、遥测） - 8 项单元测试覆盖协议满足、图片编码、注入逻辑、不可变性
This commit is contained in:
+128
@@ -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/<type>;base64,<data> 格式的字符串。
|
||||
"""
|
||||
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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user