167 lines
5.1 KiB
Python
167 lines
5.1 KiB
Python
"""
|
|
千问视觉：只读图上可见原文。
|
|
|
|
本文件职责：把图片交给 qwen3-vl-plus，抄出看得见的文字。
|
|
属于 llm 包。依赖：LlmHttpClient（生产）；单测注入 chat_fn。
|
|
禁止：填询价字段 JSON、算最终缺项、发明港口/重量/日期；禁止在回调线程长等（由 Worker 槽调用）。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, Optional
|
|
|
|
from agent.llm.modes import LlmMode
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MODE = LlmMode.VISION
|
|
|
|
VISION_SYSTEM = (
|
|
"只读出图片上能看见的文字和数字，原样抄写，一行不漏。"
|
|
"英文、机场三字码、航线（如 SZX-CGK、SZX/CGK）必须保持原样，禁止翻译成中文城市名。"
|
|
"看不清的字不要猜。不要补询价缺项，不要发明港口、重量或日期。"
|
|
"只有整张图完全没有文字时才输出空；只要看得见字就必须抄出来。"
|
|
)
|
|
|
|
|
|
def sniff_image_mime(raw: bytes) -> str:
|
|
"""
|
|
按文件头认图片类型。企微截图经常是 PNG，不能一律当 JPEG 送给千问。
|
|
"""
|
|
blob = raw or b""
|
|
if len(blob) < 12:
|
|
return ""
|
|
if blob.startswith(b"\x89PNG\r\n\x1a\n"):
|
|
return "image/png"
|
|
if blob.startswith(b"\xff\xd8\xff"):
|
|
return "image/jpeg"
|
|
if blob[:4] == b"RIFF" and blob[8:12] == b"WEBP":
|
|
return "image/webp"
|
|
if blob.startswith(b"GIF87a") or blob.startswith(b"GIF89a"):
|
|
return "image/gif"
|
|
if blob[4:8] == b"ftyp":
|
|
return "image/heic"
|
|
if blob.startswith(b"BM"):
|
|
return "image/bmp"
|
|
return ""
|
|
|
|
|
|
def mode_meta() -> dict[str, Any]:
|
|
return {
|
|
"mode": MODE.value,
|
|
"provider": "qwen",
|
|
"purpose": "vision",
|
|
"implemented": True,
|
|
}
|
|
|
|
|
|
def _pack_text(text: str, *, stub: bool = False) -> dict[str, Any]:
|
|
body = (text or "").strip()
|
|
return {
|
|
"ok": True,
|
|
"stub": stub,
|
|
"mode": MODE.value,
|
|
"text": body,
|
|
"empty": not bool(body),
|
|
"error": "",
|
|
}
|
|
|
|
|
|
def invoke_shell(*, payload: dict[str, Any]) -> dict[str, Any]:
|
|
"""
|
|
无网/单测入口：assistant_text 当作图上可见原文。
|
|
|
|
没有该键时保持失败码，避免假装已经读图。
|
|
"""
|
|
if "assistant_text" in (payload or {}):
|
|
packed = _pack_text(str(payload.get("assistant_text") or ""), stub=True)
|
|
return packed
|
|
return {
|
|
"ok": False,
|
|
"stub": True,
|
|
"mode": MODE.value,
|
|
"text": "",
|
|
"empty": True,
|
|
"error": "vision_shell_not_implemented",
|
|
"echo_keys": sorted((payload or {}).keys()),
|
|
}
|
|
|
|
|
|
def _vision_messages(*, image_b64: str, mime: str, image_url: str) -> list[dict[str, Any]]:
|
|
"""拼兼容 OpenAI 的图文消息。优先 data URL，其次 http 图链。"""
|
|
url = (image_url or "").strip()
|
|
if not url and image_b64:
|
|
kind = (mime or "image/jpeg").strip() or "image/jpeg"
|
|
url = f"data:{kind};base64,{image_b64}"
|
|
user_content: list[dict[str, Any]] = []
|
|
if url:
|
|
user_content.append({"type": "image_url", "image_url": {"url": url}})
|
|
user_content.append({"type": "text", "text": "请只抄写图上可见文字。"})
|
|
return [
|
|
{"role": "system", "content": VISION_SYSTEM},
|
|
{"role": "user", "content": user_content},
|
|
]
|
|
|
|
|
|
def invoke_vision(
|
|
*,
|
|
payload: dict[str, Any],
|
|
chat_fn: Optional[Any] = None,
|
|
allow_network: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
读一张图，返回可见原文。
|
|
|
|
chat_fn：单测注入，签名同 LlmHttpClient.chat。
|
|
allow_network=False 且无 chat_fn：走 invoke_shell，不打真网。
|
|
"""
|
|
data = dict(payload or {})
|
|
image_b64 = str(data.get("image_b64") or "")
|
|
image_url = str(data.get("image_url") or "")
|
|
mime = str(data.get("mime") or "").strip() or "image/jpeg"
|
|
if chat_fn is None and not allow_network:
|
|
return invoke_shell(payload=data)
|
|
if not image_b64 and not image_url:
|
|
logger.info("vision 拒绝：没有图")
|
|
return {
|
|
"ok": False,
|
|
"stub": False,
|
|
"mode": MODE.value,
|
|
"text": "",
|
|
"empty": True,
|
|
"error": "no_image",
|
|
}
|
|
if chat_fn is None:
|
|
from agent.llm.http_client import LlmHttpClient
|
|
|
|
client = LlmHttpClient.from_settings()
|
|
chat_fn = client.chat
|
|
messages = _vision_messages(
|
|
image_b64=image_b64,
|
|
mime=mime,
|
|
image_url=image_url,
|
|
)
|
|
result = chat_fn(mode=MODE, messages=messages)
|
|
if not getattr(result, "ok", False):
|
|
err = str(getattr(result, "error", "") or "vision_failed")
|
|
logger.info("vision 失败 err=%s", err)
|
|
return {
|
|
"ok": False,
|
|
"stub": False,
|
|
"mode": MODE.value,
|
|
"text": "",
|
|
"empty": True,
|
|
"error": err,
|
|
}
|
|
packed = _pack_text(str(getattr(result, "content", "") or ""))
|
|
packed["stub"] = False
|
|
logger.info(
|
|
"vision 读完 ok=%s empty=%s text_len=%s mime=%s",
|
|
True,
|
|
packed.get("empty"),
|
|
len(str(packed.get("text") or "")),
|
|
mime,
|
|
)
|
|
return packed
|