Teknium
378 lines
17 KiB
Python
378 lines
17 KiB
Python
"""Image-part handling for ``AIAgent`` API messages.
|
|
|
|
Vision capability probes, non-vision text fallbacks (cached ``vision_analyze`` descriptions), tool-result
|
|
image stripping, and provider quirks (Anthropic dot preservation, Qwen portal message shaping).
|
|
"""
|
|
import logging
|
|
import asyncio
|
|
import base64
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any, List, Optional
|
|
|
|
from agent.lazy_forward import forward_static as _forward_static
|
|
from agent.tool_dispatch_helpers import _is_multimodal_tool_result, _multimodal_text_summary
|
|
from utils import base_url_host_matches, base_url_hostname
|
|
|
|
# Same logger name as the origin module so log records / caplog filters are unchanged.
|
|
logger = logging.getLogger("run_agent")
|
|
|
|
_IMAGE_PART_TYPES = {"image_url", "input_image"}
|
|
_TEXT_PART_TYPES = {"text", "input_text"}
|
|
_DATA_URL_SUFFIXES = {
|
|
"image/png": ".png", "image/gif": ".gif", "image/webp": ".webp", "image/jpeg": ".jpg", "image/jpg": ".jpg"
|
|
}
|
|
|
|
|
|
def _is_image_part(part: Any) -> bool:
|
|
return isinstance(part, dict) and part.get("type") in _IMAGE_PART_TYPES
|
|
|
|
|
|
def _provider_model_key(agent: Any) -> tuple[str, str]:
|
|
"""``(provider.lower(), model)`` as recorded in ``_no_list_tool_content_models``.
|
|
Module-level so ``MagicMock(spec=AIAgent)`` agents in tests don't swallow it."""
|
|
return (
|
|
(getattr(agent, "provider", "") or "").strip().lower(),
|
|
(getattr(agent, "model", "") or "").strip(),
|
|
)
|
|
|
|
|
|
class VisionMessagePrepMixin:
|
|
"""Vision probes + image-part fallbacks for outgoing messages (see module docstring)."""
|
|
|
|
@staticmethod
|
|
def _content_has_image_parts(content: Any) -> bool:
|
|
return isinstance(content, list) and any(_is_image_part(part) for part in content)
|
|
|
|
# 20 MB base64 ≈ 15 MB decoded — prevents OOM from an oversized data: URL in a shared gateway process.
|
|
_MAX_DATA_URL_BASE64_BYTES = 20 * 1024 * 1024
|
|
|
|
@staticmethod
|
|
def _materialize_data_url_for_vision(image_url: str) -> tuple[str, Optional[Path]]:
|
|
header, _, data = str(image_url or "").partition(",")
|
|
if len(data) > VisionMessagePrepMixin._MAX_DATA_URL_BASE64_BYTES:
|
|
logger.warning("data-URL payload too large (%d bytes), skipping", len(data))
|
|
return "", None
|
|
mime = header[len("data:"):].split(";", 1)[0].strip() if header.startswith("data:") else ""
|
|
suffix = _DATA_URL_SUFFIXES.get(mime if mime.startswith("image/") else "image/jpeg", ".jpg")
|
|
tmp = tempfile.NamedTemporaryFile(prefix="anthropic_image_", suffix=suffix, delete=False)
|
|
try:
|
|
with tmp:
|
|
tmp.write(base64.b64decode(data))
|
|
except Exception:
|
|
# delete=False means a corrupt/unsupported data URL would otherwise
|
|
# leak a zero-byte temp file on every failed materialization.
|
|
try:
|
|
os.unlink(tmp.name)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
return tmp.name, Path(tmp.name)
|
|
|
|
def _describe_image_for_anthropic_fallback(self, image_url: str, role: str) -> str:
|
|
cache_key = hashlib.sha256(str(image_url or "").encode("utf-8")).hexdigest()
|
|
cached = self._anthropic_image_fallback_cache.get(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
role_label = {"assistant": "assistant", "tool": "tool result"}.get(role, "user")
|
|
analysis_prompt = (
|
|
"Describe everything visible in this image in thorough detail. "
|
|
"Include any text, code, UI, data, objects, people, layout, colors, "
|
|
"and any other notable visual information."
|
|
)
|
|
|
|
vision_source = str(image_url or "")
|
|
is_data_url = vision_source.startswith("data:")
|
|
cleanup_path: Optional[Path] = None
|
|
if is_data_url:
|
|
vision_source, cleanup_path = self._materialize_data_url_for_vision(vision_source)
|
|
|
|
try:
|
|
from tools.vision_tools import vision_analyze_tool
|
|
|
|
result_json = asyncio.run(vision_analyze_tool(image_url=vision_source, user_prompt=analysis_prompt))
|
|
result = json.loads(result_json) if isinstance(result_json, str) else {}
|
|
description = (result.get("analysis") or "").strip()
|
|
except Exception as e:
|
|
description = f"Image analysis failed: {e}"
|
|
finally:
|
|
if cleanup_path and cleanup_path.exists():
|
|
try:
|
|
cleanup_path.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
note = f"[The {role_label} attached an image. Here's what it contains:\n{description or 'Image analysis failed.'}]"
|
|
if vision_source and not is_data_url:
|
|
note += f"\n[If you need a closer look, use vision_analyze with image_url: {vision_source}]"
|
|
|
|
self._anthropic_image_fallback_cache[cache_key] = note
|
|
return note
|
|
|
|
def _model_supports_vision(self) -> bool:
|
|
"""Return True if the active provider+model reports native vision.
|
|
|
|
Resolution: ``model.supports_vision`` > ``providers.<p>.models.<m>.supports_vision`` > models.dev
|
|
lookup (see ``image_routing._supports_vision_override``). Custom/local models absent from models.dev
|
|
would otherwise be misclassified and have their images stripped.
|
|
"""
|
|
try:
|
|
from hermes_cli.config import load_config
|
|
from agent.image_routing import _lookup_supports_vision
|
|
provider = (getattr(self, "provider", "") or "").strip()
|
|
model = (getattr(self, "model", "") or "").strip()
|
|
return _lookup_supports_vision(provider, model, load_config()) is True
|
|
except Exception:
|
|
return False
|
|
|
|
def _provider_supports_vision_tool_messages(self) -> bool:
|
|
"""Return True if the active provider accepts list-type tool content.
|
|
|
|
Some providers (Xiaomi MiMo) accept multimodal user messages but 400 on list-type tool content;
|
|
reads the provider profile's ``supports_vision_tool_messages``.
|
|
"""
|
|
try:
|
|
from providers import get_provider_profile
|
|
profile = get_provider_profile((getattr(self, "provider", "") or "").strip())
|
|
if profile is not None:
|
|
return getattr(profile, "supports_vision_tool_messages", True)
|
|
except Exception:
|
|
pass
|
|
return True # default: assume compatible
|
|
|
|
def _preprocess_anthropic_content(self, content: Any, role: str) -> Any:
|
|
if not self._content_has_image_parts(content):
|
|
return content
|
|
|
|
text_parts: List[str] = []
|
|
image_notes: List[str] = []
|
|
for part in content:
|
|
if isinstance(part, str):
|
|
if part.strip():
|
|
text_parts.append(part.strip())
|
|
continue
|
|
if not isinstance(part, dict):
|
|
continue
|
|
|
|
if _is_image_part(part):
|
|
image_data = part.get("image_url", {})
|
|
image_url = image_data.get("url", "") if isinstance(image_data, dict) else str(image_data or "")
|
|
image_notes.append(
|
|
self._describe_image_for_anthropic_fallback(image_url, role) if image_url
|
|
else "[An image was attached but no image source was available.]"
|
|
)
|
|
continue
|
|
# Text parts and unknown dict types both contribute their ``text``.
|
|
text = str(part.get("text", "") or "").strip()
|
|
if text:
|
|
text_parts.append(text)
|
|
|
|
prefix = "\n\n".join(note for note in image_notes if note).strip()
|
|
suffix = "\n".join(text for text in text_parts if text).strip()
|
|
if prefix and suffix:
|
|
return f"{prefix}\n\n{suffix}"
|
|
return prefix or suffix or "[A multimodal message was converted to text for Anthropic compatibility.]"
|
|
|
|
def _get_transport(self, api_mode: str = None):
|
|
"""Return the cached transport for the given (or current) api_mode (lazy; None if unregistered)."""
|
|
mode = api_mode or self.api_mode
|
|
cache = getattr(self, "_transport_cache", None)
|
|
if cache is None:
|
|
cache = self._transport_cache = {}
|
|
if cache.get(mode) is None:
|
|
from agent.transports import get_transport
|
|
cache[mode] = get_transport(mode)
|
|
return cache[mode]
|
|
|
|
def _prepare_messages_for_non_vision_model(self, api_messages: list) -> list:
|
|
"""Replace native image parts with cached vision_analyze text when the active model lacks vision.
|
|
|
|
Vision-capable models pass through unchanged (the provider adapter — including the Anthropic one —
|
|
handles image parts natively). The text fallback is the historically Anthropic-named preprocessor.
|
|
"""
|
|
if not any(
|
|
isinstance(msg, dict) and self._content_has_image_parts(msg.get("content")) for msg in api_messages
|
|
) or self._model_supports_vision():
|
|
return api_messages
|
|
|
|
transformed = copy.deepcopy(api_messages)
|
|
for msg in transformed:
|
|
if isinstance(msg, dict):
|
|
msg["content"] = self._preprocess_anthropic_content(
|
|
msg.get("content"), str(msg.get("role", "user") or "user")
|
|
)
|
|
return transformed
|
|
|
|
# Same transform for the Anthropic route (callers/tests patch this name independently).
|
|
_prepare_anthropic_messages_for_api = _prepare_messages_for_non_vision_model
|
|
|
|
def _tool_result_content_for_active_model(self, tool_name: str, result: Any) -> Any:
|
|
"""Return the tool message content that is safe for the active model.
|
|
|
|
Text-only providers must not receive image parts: a rejected tool result becomes canonical history
|
|
and can make the next user turn fail before the agent can recover.
|
|
"""
|
|
if not _is_multimodal_tool_result(result):
|
|
return result
|
|
|
|
content = result.get("content") or []
|
|
if not self._content_has_image_parts(content):
|
|
return content
|
|
|
|
if self._model_supports_vision():
|
|
# Vision on paper, but the provider rejects list-type tool content (or we already learned that
|
|
# in-session): short-circuit to a text summary.
|
|
if not self._provider_supports_vision_tool_messages():
|
|
logger.debug(
|
|
"Tool %s: provider %s does not accept list-type tool "
|
|
"content — sending text summary",
|
|
tool_name, getattr(self, "provider", ""),
|
|
)
|
|
return _multimodal_text_summary(result)
|
|
key = _provider_model_key(self)
|
|
no_list = getattr(self, "_no_list_tool_content_models", None)
|
|
if no_list and key in no_list:
|
|
logger.debug(
|
|
"Tool %s: model %s/%s known to reject list-type tool "
|
|
"content this session — sending text summary",
|
|
tool_name, key[0], key[1],
|
|
)
|
|
return _multimodal_text_summary(result)
|
|
return content
|
|
|
|
summary = _multimodal_text_summary(result)
|
|
if tool_name == "computer_use":
|
|
return json.dumps({
|
|
"error": (
|
|
"computer_use returned screenshot/image content, but the active "
|
|
"model/provider does not support image input. Switch to a "
|
|
"vision-capable model for desktop computer use, or use browser "
|
|
"tools for browser tasks."
|
|
),
|
|
"text_summary": summary,
|
|
})
|
|
|
|
logger.warning(
|
|
"Tool %s returned image content for non-vision model %s/%s; "
|
|
"falling back to text summary",
|
|
tool_name, self.provider, self.model,
|
|
)
|
|
return summary
|
|
|
|
_try_shrink_image_parts_in_messages = _forward_static("agent.conversation_compression", "try_shrink_image_parts_in_messages")
|
|
|
|
def _try_strip_image_parts_from_tool_messages(
|
|
self, api_messages: list, *, remember_model: bool = True
|
|
) -> bool:
|
|
"""Downgrade list-type tool messages to text summaries in place; returns True if any were downgraded.
|
|
|
|
Recovery for providers that 400 on list-type tool content (e.g. MiMo "text is not set"). By default
|
|
records the (provider, model) in ``_no_list_tool_content_models`` so later results downgrade without a
|
|
round-trip; 413 recovery passes ``remember_model=False`` (body too large ≠ provider rejects lists).
|
|
"""
|
|
if not isinstance(api_messages, list):
|
|
return False
|
|
|
|
if remember_model:
|
|
# Record (provider, model) so we don't relearn this lesson.
|
|
key = _provider_model_key(self)
|
|
if not hasattr(self, "_no_list_tool_content_models"):
|
|
self._no_list_tool_content_models = set()
|
|
if key[1]: # only record when we actually have a model id
|
|
self._no_list_tool_content_models.add(key)
|
|
|
|
changed = False
|
|
for msg in api_messages:
|
|
if not isinstance(msg, dict) or msg.get("role") != "tool":
|
|
continue
|
|
content = msg.get("content")
|
|
# List content without image parts is left alone; stripping wouldn't reduce ambiguity.
|
|
if not self._content_has_image_parts(content):
|
|
continue
|
|
|
|
# Salvage any text parts so the model still sees some signal.
|
|
text_parts: List[str] = []
|
|
for part in content:
|
|
if isinstance(part, str):
|
|
text = part.strip()
|
|
elif isinstance(part, dict) and part.get("type") in _TEXT_PART_TYPES:
|
|
text = str(part.get("text") or "").strip()
|
|
else:
|
|
continue
|
|
if text:
|
|
text_parts.append(text)
|
|
|
|
msg["content"] = "\n\n".join(text_parts) or (
|
|
"[image content removed — provider does not accept "
|
|
"list-type tool message content]"
|
|
)
|
|
changed = True
|
|
|
|
return changed
|
|
|
|
def _anthropic_preserve_dots(self) -> bool:
|
|
"""True when using an anthropic-compatible endpoint that preserves dots in model names.
|
|
|
|
DashScope, MiniMax, Xiaomi MiMo, OpenCode Go/Zen (non-Claude), ZAI/Zhipu keep dots; AWS Bedrock uses
|
|
dotted inference-profile IDs and rejects the hyphenated form with HTTP 400.
|
|
"""
|
|
if (getattr(self, "provider", "") or "").lower() in {
|
|
"alibaba", "minimax", "minimax-cn", "opencode-go", "opencode-zen", "zai", "bedrock", "xiaomi", "vertex",
|
|
}:
|
|
return True
|
|
base = (getattr(self, "base_url", "") or "").lower()
|
|
host = base_url_hostname(base)
|
|
return (
|
|
"dashscope" in host
|
|
or base_url_host_matches(base, "aliyuncs.com")
|
|
or "minimax" in host
|
|
or (base_url_host_matches(base, "opencode.ai") and "/zen/" in base)
|
|
or base_url_host_matches(base, "bigmodel.cn")
|
|
or base_url_host_matches(base, "xiaomimimo.com")
|
|
# Vertex AI OpenAI-compat endpoint — Gemini model ids keep dots
|
|
# (e.g. google/gemini-3.5-flash); the hyphenated form is wrong.
|
|
or base_url_host_matches(base, "aiplatform.googleapis.com")
|
|
# AWS Bedrock runtime endpoints — defense-in-depth when
|
|
# ``provider`` is unset but ``base_url`` still names Bedrock.
|
|
or host.startswith("bedrock-runtime.")
|
|
)
|
|
|
|
def _is_qwen_portal(self) -> bool:
|
|
"""Return True when the base URL targets Qwen Portal."""
|
|
return base_url_host_matches(self._base_url_lower, "portal.qwen.ai")
|
|
|
|
def _qwen_prepare_chat_messages(self, api_messages: list) -> list:
|
|
"""Deep-copy ``api_messages`` and shape them for Qwen Portal (see the in-place variant)."""
|
|
prepared = copy.deepcopy(api_messages)
|
|
self._qwen_prepare_chat_messages_inplace(prepared)
|
|
return prepared
|
|
|
|
def _qwen_prepare_chat_messages_inplace(self, messages: list) -> None:
|
|
"""Qwen Portal shaping, in place: every content becomes a list of parts (bare strings → text
|
|
dicts, dicts kept), then ``cache_control`` is injected on the last part of the system message."""
|
|
for msg in messages:
|
|
if not isinstance(msg, dict):
|
|
continue
|
|
content = msg.get("content")
|
|
if isinstance(content, str):
|
|
msg["content"] = [{"type": "text", "text": content}]
|
|
elif isinstance(content, list):
|
|
normalized_parts = [
|
|
{"type": "text", "text": part} if isinstance(part, str) else part
|
|
for part in content if isinstance(part, (str, dict))
|
|
]
|
|
if normalized_parts:
|
|
msg["content"] = normalized_parts
|
|
|
|
for msg in messages:
|
|
if isinstance(msg, dict) and msg.get("role") == "system":
|
|
content = msg.get("content")
|
|
if isinstance(content, list) and content and isinstance(content[-1], dict):
|
|
content[-1]["cache_control"] = {"type": "ephemeral"}
|
|
break
|