diff --git a/nanobot/agent/context.py b/nanobot/agent/context.py index 19ee935c4..e80c5e511 100644 --- a/nanobot/agent/context.py +++ b/nanobot/agent/context.py @@ -2,6 +2,7 @@ import base64 import mimetypes +import os import platform from contextlib import suppress from importlib.resources import files as pkg_files @@ -10,11 +11,16 @@ from typing import Any, Mapping, Sequence from nanobot.agent.memory import MemoryStore from nanobot.agent.skills import SkillsLoader +from nanobot.config.schema import InputLimitsConfig from nanobot.session.goal_state import goal_state_runtime_lines from nanobot.utils.helpers import ( + audio_format_for_api, + audio_mime_compat, current_time_str, + detect_audio_mime, detect_image_mime, truncate_text, + video_mime_compat, ) from nanobot.utils.prompt_templates import render_template @@ -28,11 +34,12 @@ class ContextBuilder: _MAX_HISTORY_CHARS = 32_000 # hard cap on recent history section size _RUNTIME_CONTEXT_END = "[/Runtime Context]" - def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | None = None): + def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | None = None, input_limits: InputLimitsConfig | None = None): self.workspace = workspace self.timezone = timezone self.memory = MemoryStore(workspace) self.skills = SkillsLoader(workspace, disabled_skills=set(disabled_skills) if disabled_skills else None) + self.input_limits = input_limits or InputLimitsConfig() def build_system_prompt( self, @@ -142,6 +149,28 @@ class ContextBuilder: return content.strip() == tpl.read_text(encoding="utf-8").strip() return False + @staticmethod + def _file_size_ok(p: Path, max_bytes: int) -> bool | None: + """Check file size via stat without reading into memory. + + Returns True if size is within limit, False if oversized, + None if file cannot be stat'd (caller should try read_bytes instead). + """ + try: + return os.stat(p).st_size <= max_bytes + except OSError: + return None + + @staticmethod + def _encode_image_block(raw: bytes, mime: str, path: Path) -> dict[str, Any]: + """Base64-encode file bytes into an image_url content block.""" + b64 = base64.b64encode(raw).decode() + return { + "type": "image_url", + "image_url": {"url": f"data:{mime};base64,{b64}"}, + "_meta": {"path": str(path)}, + } + def build_messages( self, history: list[dict[str, Any]], @@ -154,6 +183,9 @@ class ContextBuilder: sender_id: str | None = None, session_summary: str | None = None, session_metadata: Mapping[str, Any] | None = None, + supports_vision: bool | None = None, + supports_audio: bool | None = None, + supports_video: bool | None = None, ) -> list[dict[str, Any]]: """Build the complete message list for an LLM call.""" extra = goal_state_runtime_lines(session_metadata) @@ -164,7 +196,12 @@ class ContextBuilder: sender_id=sender_id, supplemental_lines=extra or None, ) - user_content = self._build_user_content(current_message, media) + user_content = self._build_user_content( + current_message, media, + supports_vision=supports_vision, + supports_audio=supports_audio, + supports_video=supports_video, + ) # Merge runtime context and user content into a single user message # to avoid consecutive same-role messages that some providers reject. @@ -186,28 +223,171 @@ class ContextBuilder: messages.append({"role": current_role, "content": merged}) return messages - def _build_user_content(self, text: str, media: list[str] | None) -> str | list[dict[str, Any]]: - """Build user message content with optional base64-encoded images.""" + def _build_user_content( + self, + text: str, + media: list[str] | None, + *, + supports_vision: bool | None = None, + supports_audio: bool | None = None, + supports_video: bool | None = None, + ) -> str | list[dict[str, Any]]: + """Build user message content with optional media blocks. + + Args: + text: The user text message. + media: List of file paths to media files. + supports_vision: True=model supports images, False=use placeholder, + None=unconfigured (send images as before, let + provider/retry handle degradation). + supports_audio: True=model supports native audio, False/None=skip + (channel layer already transcribed). + supports_video: True=model supports native video, False/None=use + [file: path] placeholder. + """ if not media: return text - images = [] + blocks: list[dict[str, Any]] = [] + notes: list[str] = [] + limits = self.input_limits + + # Enforce image count limit + max_images = limits.max_input_images + image_count = 0 + image_media = [] + non_image_media = [] for path in media: + p = Path(path) + guessed_mime = mimetypes.guess_type(path)[0] or "" + if guessed_mime.startswith("image/"): + image_count += 1 + if image_count <= max_images: + image_media.append(path) + else: + non_image_media.append(path) + + if image_count > max_images: + extra = image_count - max_images + noun = "image" if extra == 1 else "images" + notes.append( + f"[Skipped {extra} {noun}: " + f"only the first {max_images} images are included]" + ) + + # Process images + for path in image_media: p = Path(path) if not p.is_file(): continue - raw = p.read_bytes() - mime = detect_image_mime(raw) or mimetypes.guess_type(path)[0] - if not mime or not mime.startswith("image/"): + + # When explicitly marked as non-vision, downgrade to text placeholder + if supports_vision is False: + blocks.append({"type": "text", "text": f"[image: {p}]"}) continue - b64 = base64.b64encode(raw).decode() - images.append({ - "type": "image_url", - "image_url": {"url": f"data:{mime};base64,{b64}"}, - "_meta": {"path": str(p)}, - }) - if not images: - return text - return images + [{"type": "text", "text": text}] + size_ok = self._file_size_ok(p, limits.max_input_image_bytes) + if size_ok is False: + size_mb = limits.max_input_image_bytes // (1024 * 1024) + notes.append(f"[Skipped image: file too large ({p.name}, limit {size_mb} MB)]") + continue + try: + raw = p.read_bytes() + except OSError: + notes.append(f"[Skipped image: unable to read ({p.name or path})]") + continue + img_mime = detect_image_mime(raw[:32]) or mimetypes.guess_type(path)[0] + if not img_mime or not img_mime.startswith("image/"): + notes.append(f"[Skipped image: unsupported or invalid image format ({p.name})]") + continue + blocks.append(self._encode_image_block(raw, img_mime, p)) + + # Process non-image media (audio, video, unknown) + audio_count = 0 + video_count = 0 + for path in non_image_media: + p = Path(path) + if not p.is_file(): + continue + guessed_mime = mimetypes.guess_type(path)[0] or "" + is_audio = guessed_mime.startswith("audio/") + is_video = guessed_mime.startswith("video/") + + # Pre-check file size via stat to avoid reading oversized files into memory. + # Determine the relevant byte limit based on detected media type. + _size_limit = 0 + if is_audio or is_video: + _size_limit = limits.max_input_audio_bytes if is_audio else limits.max_input_video_bytes + _stat_size_ok = self._file_size_ok(p, _size_limit) if _size_limit else None + if _stat_size_ok is False: + size_mb = _size_limit // (1024 * 1024) + label = "audio" if is_audio else "video" + notes.append(f"[Skipped {label}: file too large ({p.name}, limit {size_mb} MB)]") + continue + + try: + raw = p.read_bytes() + except OSError: + notes.append(f"[Skipped file: unable to read ({p.name or path})]") + continue + + # Audio detection: by magic bytes or by filename + # Always pass filename so fallback can match when magic bytes fail + audio_mime = detect_audio_mime(raw[:32], filename=path) + if audio_mime or is_audio: + if supports_audio is True and audio_mime_compat(audio_mime): + audio_count += 1 + if audio_count > limits.max_input_audios: + if audio_count == limits.max_input_audios + 1: + notes.append( + f"[Skipped audio: only {limits.max_input_audios} audio file(s) allowed]" + ) + continue + if len(raw) > limits.max_input_audio_bytes: + size_mb = limits.max_input_audio_bytes // (1024 * 1024) + notes.append(f"[Skipped audio: file too large ({p.name}, limit {size_mb} MB)]") + continue + b64 = base64.b64encode(raw).decode() + blocks.append({ + "type": "input_audio", + "input_audio": {"data": b64, "format": audio_format_for_api(audio_mime)}, + "_meta": {"path": str(p)}, + }) + else: + blocks.append({"type": "text", "text": f"[audio: {p}]"}) + continue + + # Video detection (already classified above) + if is_video: + if supports_video is True and video_mime_compat(guessed_mime): + video_count += 1 + if video_count > limits.max_input_videos: + if video_count == limits.max_input_videos + 1: + notes.append( + f"[Skipped video: only {limits.max_input_videos} video file(s) allowed]" + ) + continue + if len(raw) > limits.max_input_video_bytes: + size_mb = limits.max_input_video_bytes // (1024 * 1024) + notes.append(f"[Skipped video: file too large ({p.name}, limit {size_mb} MB)]") + continue + b64 = base64.b64encode(raw).decode() + blocks.append({ + "type": "video_url", + "video_url": {"url": f"data:{guessed_mime};base64,{b64}"}, + "_meta": {"path": str(p)}, + }) + else: + blocks.append({"type": "text", "text": f"[video: {p}]"}) + continue + + # Unknown files are silently ignored (preserves pre-multimodal behaviour) + continue + + note_text = "\n".join(notes).strip() + text_block = text if not note_text else (f"{note_text}\n\n{text}" if text else note_text) + + if not blocks: + return text_block + return blocks + [{"type": "text", "text": text_block}] diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index abc6449de..3505fb496 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -190,6 +190,10 @@ class AgentLoop: model_preset: str | None = None, preset_snapshot_loader: preset_helpers.PresetSnapshotLoader | None = None, runtime_model_publisher: Callable[[str, str | None], None] | None = None, + input_limits: Any = None, + supports_vision: bool | None = None, + supports_audio: bool | None = None, + supports_video: bool | None = None, ): from nanobot.config.schema import ToolsConfig @@ -227,6 +231,10 @@ class AgentLoop: self.tools_config = _tc self.web_config = _tc.web self.exec_config = _tc.exec + self.input_limits = input_limits or _tc.input_limits + self._supports_vision = supports_vision + self._supports_audio = supports_audio + self._supports_video = supports_video self._image_generation_provider_configs = dict(image_generation_provider_configs or {}) if ( image_generation_provider_config is not None @@ -240,7 +248,7 @@ class AgentLoop: self._pending_turn_latency_ms: dict[str, int] = {} self._extra_hooks: list[AgentHook] = hooks or [] - self.context = ContextBuilder(workspace, timezone=timezone, disabled_skills=disabled_skills) + self.context = ContextBuilder(workspace, timezone=timezone, disabled_skills=disabled_skills, input_limits=self.input_limits) self.sessions = session_manager or SessionManager(workspace) self._webui_turns = WebuiTurnCoordinator( bus=self.bus, @@ -366,6 +374,10 @@ class AgentLoop: model_preset=defaults.model_preset, provider_snapshot_loader=provider_snapshot_loader, preset_snapshot_loader=preset_snapshot_loader, + input_limits=config.tools.input_limits, + supports_vision=defaults.supports_vision(defaults.model), + supports_audio=defaults.supports_audio(defaults.model), + supports_video=defaults.supports_video(defaults.model), **extra, ) @@ -594,6 +606,9 @@ class AgentLoop: sender_id=msg.sender_id, session_summary=pending_summary, session_metadata=session.metadata, + supports_vision=self._supports_vision, + supports_audio=self._supports_audio, + supports_video=self._supports_video, ) async def _dispatch_command_inline( @@ -1059,6 +1074,9 @@ class AgentLoop: sender_id=msg.sender_id, session_summary=pending, session_metadata=session.metadata, + supports_vision=self._supports_vision, + supports_audio=self._supports_audio, + supports_video=self._supports_video, ) t_wall = time.time() final_content, _, all_msgs, stop_reason, _ = await self._run_agent_loop( diff --git a/nanobot/channels/feishu.py b/nanobot/channels/feishu.py index c5e085972..b21ccdf0b 100644 --- a/nanobot/channels/feishu.py +++ b/nanobot/channels/feishu.py @@ -172,19 +172,22 @@ def _extract_element_content(element: dict) -> list[str]: return parts -def _extract_post_content(content_json: dict) -> tuple[str, list[str]]: - """Extract text and image keys from Feishu post (rich text) message. +def _extract_post_content(content_json: dict) -> tuple[str, list[str], list[dict]]: + """Extract text and media info from Feishu post (rich text) message. Handles three payload shapes: - Direct: {"title": "...", "content": [[...]]} - Localized: {"zh_cn": {"title": "...", "content": [...]}} - Wrapped: {"post": {"zh_cn": {"title": "...", "content": [...]}}} + + Returns (text, image_keys, media_items) where media_items is a list of + {"tag": "media", "file_key": "..."} dicts for video/file attachments. """ - def _parse_block(block: dict) -> tuple[str | None, list[str]]: + def _parse_block(block: dict) -> tuple[str | None, list[str], list[dict]]: if not isinstance(block, dict) or not isinstance(block.get("content"), list): - return None, [] - texts, images = [], [] + return None, [], [] + texts, images, medias = [], [], [] if title := block.get("title"): texts.append(title) for row in block["content"]: @@ -204,43 +207,36 @@ def _extract_post_content(content_json: dict) -> tuple[str, list[str]]: texts.append(f"\n```{lang}\n{code_text}\n```\n") elif tag == "img" and (key := el.get("image_key")): images.append(key) - return (" ".join(texts).strip() or None), images + elif tag == "media" and el.get("file_key"): + medias.append({"tag": "media", "file_key": el["file_key"]}) + return (" ".join(texts).strip() or None), images, medias # Unwrap optional {"post": ...} envelope root = content_json if isinstance(root, dict) and isinstance(root.get("post"), dict): root = root["post"] if not isinstance(root, dict): - return "", [] + return "", [], [] # Direct format if "content" in root: - text, imgs = _parse_block(root) - if text or imgs: - return text or "", imgs + text, imgs, medias = _parse_block(root) + if text or imgs or medias: + return text or "", imgs, medias # Localized: prefer known locales, then fall back to any dict child for key in ("zh_cn", "en_us", "ja_jp"): if key in root: - text, imgs = _parse_block(root[key]) - if text or imgs: - return text or "", imgs + text, imgs, medias = _parse_block(root[key]) + if text or imgs or medias: + return text or "", imgs, medias for val in root.values(): if isinstance(val, dict): - text, imgs = _parse_block(val) - if text or imgs: - return text or "", imgs + text, imgs, medias = _parse_block(val) + if text or imgs or medias: + return text or "", imgs, medias - return "", [] - - -def _extract_post_text(content_json: dict) -> str: - """Extract plain text from Feishu post (rich text) message content. - - Legacy wrapper for _extract_post_content, returns only text. - """ - text, _ = _extract_post_content(content_json) - return text + return "", [], [] class FeishuConfig(Base): @@ -1156,7 +1152,7 @@ class FeishuChannel(BaseChannel): if msg_type == "text": text = content_json.get("text", "").strip() elif msg_type == "post": - text, _ = _extract_post_content(content_json) + text, _, _ = _extract_post_content(content_json) text = text.strip() else: text = "" @@ -1751,7 +1747,7 @@ class FeishuChannel(BaseChannel): content_parts.append(text) elif msg_type == "post": - text, image_keys = _extract_post_content(content_json) + text, image_keys, media_items = _extract_post_content(content_json) if text: content_parts.append(text) # Download images embedded in post @@ -1762,6 +1758,14 @@ class FeishuChannel(BaseChannel): if file_path: media_paths.append(file_path) content_parts.append(content_text) + # Download media (video/file) embedded in post + for media_item in media_items: + file_path, content_text = await self._download_and_save_media( + "media", media_item, message_id + ) + if file_path: + media_paths.append(file_path) + content_parts.append(content_text) elif msg_type in ("image", "audio", "file", "media"): file_path, content_text = await self._download_and_save_media( diff --git a/nanobot/config/schema.py b/nanobot/config/schema.py index c0ad7e758..47a7f0e17 100644 --- a/nanobot/config/schema.py +++ b/nanobot/config/schema.py @@ -155,8 +155,35 @@ class AgentDefaults(Base): validation_alias=AliasChoices("consolidationRatio"), serialization_alias="consolidationRatio", ) # Consolidation target ratio (0.5 = 50% of budget retained after compression) + vision_models: list[str] = Field(default_factory=list) # Models that support image input + audio_models: list[str] = Field(default_factory=list) # Models that support native audio input + video_models: list[str] = Field(default_factory=list) # Models that support native video input dream: DreamConfig = Field(default_factory=DreamConfig) + @staticmethod + def _bare_model(model: str) -> str: + """Strip provider prefix, e.g. 'openai/gpt-4o' -> 'gpt-4o'.""" + return model.split("/", 1)[-1].lower() if "/" in model else model.lower() + + def _supports_capability(self, model: str, patterns: list[str]) -> bool | None: + """Check if model matches any pattern. Returns None if patterns is empty.""" + if not patterns: + return None + bare = self._bare_model(model) + return any(p.lower() in bare for p in patterns) + + def supports_vision(self, model: str) -> bool | None: + """Check if model supports vision. None if unconfigured.""" + return self._supports_capability(model, self.vision_models) + + def supports_audio(self, model: str) -> bool | None: + """Check if model supports native audio. None if unconfigured.""" + return self._supports_capability(model, self.audio_models) + + def supports_video(self, model: str) -> bool | None: + """Check if model supports native video. None if unconfigured.""" + return self._supports_capability(model, self.video_models) + class AgentsConfig(Base): """Agent configuration.""" @@ -258,6 +285,17 @@ class MCPServerConfig(Base): enabled_tools: list[str] = Field(default_factory=lambda: ["*"]) # Only register these tools; accepts raw MCP names or wrapped mcp__ names; ["*"] = all tools; [] = no tools +class InputLimitsConfig(Base): + """Limits for user-provided multimodal inputs.""" + + max_input_images: int = 3 + max_input_image_bytes: int = 10 * 1024 * 1024 # 10 MB + max_input_audios: int = 1 + max_input_audio_bytes: int = 10 * 1024 * 1024 # 10 MB + max_input_videos: int = 1 + max_input_video_bytes: int = 20 * 1024 * 1024 # 20 MB + + def _lazy_default(module_path: str, class_name: str) -> Any: """Deferred import helper for ToolsConfig default factories.""" import importlib @@ -279,6 +317,7 @@ class ToolsConfig(Base): image_generation: ImageGenerationToolConfig = Field( default_factory=lambda: _lazy_default("nanobot.agent.tools.image_generation", "ImageGenerationToolConfig"), ) + input_limits: InputLimitsConfig = Field(default_factory=InputLimitsConfig) restrict_to_workspace: bool = False # restrict all tool access to workspace directory mcp_servers: dict[str, MCPServerConfig] = Field(default_factory=dict) ssrf_whitelist: list[str] = Field(default_factory=list) # CIDR ranges to exempt from SSRF blocking (e.g. ["100.64.0.0/10"] for Tailscale) diff --git a/nanobot/providers/anthropic_provider.py b/nanobot/providers/anthropic_provider.py index 31f2bc2f1..eb33fa80d 100644 --- a/nanobot/providers/anthropic_provider.py +++ b/nanobot/providers/anthropic_provider.py @@ -212,7 +212,7 @@ class AnthropicProvider(LLMProvider): @staticmethod def _convert_user_content(content: Any) -> Any: - """Convert user message content, translating image_url blocks.""" + """Convert user message content, translating image_url and input_audio blocks.""" if isinstance(content, str) or content is None: return content or "(empty)" if not isinstance(content, list): @@ -228,6 +228,14 @@ class AnthropicProvider(LLMProvider): if converted: result.append(converted) continue + if item.get("type") == "input_audio": + # Anthropic doesn't support native audio → text placeholder + result.append(LLMProvider._media_placeholder("input_audio", item)) + continue + if item.get("type") == "video_url": + # Anthropic doesn't support native video → text placeholder + result.append(LLMProvider._media_placeholder("video_url", item)) + continue result.append(item) return result or "(empty)" diff --git a/nanobot/providers/base.py b/nanobot/providers/base.py index 87697650a..14b245513 100644 --- a/nanobot/providers/base.py +++ b/nanobot/providers/base.py @@ -13,8 +13,6 @@ from typing import Any from loguru import logger -from nanobot.utils.helpers import image_placeholder_text - @dataclass class ToolCallRequest: @@ -439,9 +437,23 @@ class LLMProvider(ABC): return merged + _MEDIA_LABEL_MAP = {"image_url": "image", "input_audio": "audio", "video_url": "video"} + _STRIP_MEDIA_TYPES = frozenset({"image_url", "input_audio", "video_url"}) + @staticmethod - def _strip_image_content(messages: list[dict[str, Any]]) -> list[dict[str, Any]] | None: - """Replace image_url blocks with text placeholder. Returns None if no images found.""" + def _media_placeholder(btype: str, block: dict[str, Any]) -> dict[str, str]: + """Build a text placeholder for a media block.""" + path = (block.get("_meta") or {}).get("path", "") + label = LLMProvider._MEDIA_LABEL_MAP.get(btype, "media") + text = f"[{label}: {path}]" if path else f"[{label}]" + return {"type": "text", "text": text} + + @staticmethod + def _strip_media_content(messages: list[dict[str, Any]]) -> list[dict[str, Any]] | None: + """Replace image_url, input_audio, and video_url blocks with text placeholders. + + Returns None if no media blocks were found (no changes needed). + """ found = False result = [] for msg in messages: @@ -449,10 +461,8 @@ class LLMProvider(ABC): if isinstance(content, list): new_content = [] for b in content: - if isinstance(b, dict) and b.get("type") == "image_url": - path = (b.get("_meta") or {}).get("path", "") - placeholder = image_placeholder_text(path, empty="[image omitted]") - new_content.append({"type": "text", "text": placeholder}) + if isinstance(b, dict) and b.get("type") in LLMProvider._STRIP_MEDIA_TYPES: + new_content.append(LLMProvider._media_placeholder(b["type"], b)) found = True else: new_content.append(b) @@ -462,8 +472,13 @@ class LLMProvider(ABC): return result if found else None @staticmethod - def _strip_image_content_inplace(messages: list[dict[str, Any]]) -> bool: - """Replace image_url blocks with text placeholder *in-place*. + def _strip_image_content(messages: list[dict[str, Any]]) -> list[dict[str, Any]] | None: + """Replace image_url blocks with text placeholder. Returns None if no images found.""" + return LLMProvider._strip_media_content(messages) + + @staticmethod + def _strip_media_content_inplace(messages: list[dict[str, Any]]) -> bool: + """Replace media blocks with text placeholder *in-place*. Mutates the content lists of the original message dicts so that callers holding references to those dicts also see the stripped @@ -474,13 +489,16 @@ class LLMProvider(ABC): content = msg.get("content") if isinstance(content, list): for i, b in enumerate(content): - if isinstance(b, dict) and b.get("type") == "image_url": - path = (b.get("_meta") or {}).get("path", "") - placeholder = image_placeholder_text(path, empty="[image omitted]") - content[i] = {"type": "text", "text": placeholder} + if isinstance(b, dict) and b.get("type") in LLMProvider._STRIP_MEDIA_TYPES: + content[i] = LLMProvider._media_placeholder(b["type"], b) found = True return found + @staticmethod + def _strip_image_content_inplace(messages: list[dict[str, Any]]) -> bool: + """Replace image_url blocks with text placeholder *in-place*.""" + return LLMProvider._strip_media_content_inplace(messages) + async def _safe_chat(self, **kwargs: Any) -> LLMResponse: """Call chat() and convert unexpected exceptions to error responses.""" try: @@ -738,18 +756,18 @@ class LLMProvider(ABC): identical_error_count = 1 if error_key else 0 if not self._is_transient_response(response): - stripped = self._strip_image_content(original_messages) + stripped = self._strip_media_content(original_messages) if stripped is not None and stripped != kw["messages"]: logger.warning( - "Non-transient LLM error with image content, retrying without images" + "Non-transient LLM error with media content, retrying without media" ) retry_kw = dict(kw) retry_kw["messages"] = stripped result = await call(**retry_kw) - # Permanently strip images from the original messages so + # Permanently strip media from the original messages so # subsequent iterations do not repeat the error-retry cycle. if result.finish_reason != "error": - self._strip_image_content_inplace(original_messages) + self._strip_media_content_inplace(original_messages) return result return response diff --git a/nanobot/providers/openai_responses/converters.py b/nanobot/providers/openai_responses/converters.py index e0bfe832d..e4c0552f8 100644 --- a/nanobot/providers/openai_responses/converters.py +++ b/nanobot/providers/openai_responses/converters.py @@ -5,6 +5,8 @@ from __future__ import annotations import json from typing import Any +from nanobot.providers.base import LLMProvider + def convert_messages(messages: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]: """Convert Chat Completions messages to Responses API input items. @@ -58,8 +60,10 @@ def convert_messages(messages: list[dict[str, Any]]) -> tuple[str, list[dict[str def convert_user_message(content: Any) -> dict[str, Any]: """Convert a user message's content to Responses API format. - Handles plain strings, ``text`` blocks -> ``input_text``, and - ``image_url`` blocks -> ``input_image``. + Handles plain strings, ``text`` blocks -> ``input_text``, + ``image_url`` blocks -> ``input_image``, and ``input_audio`` blocks. + ``video_url`` is downgraded to a text placeholder because Codex does + not support native video. """ if isinstance(content, str): return {"role": "user", "content": [{"type": "input_text", "text": content}]} @@ -74,6 +78,18 @@ def convert_user_message(content: Any) -> dict[str, Any]: url = (item.get("image_url") or {}).get("url") if url: converted.append({"type": "input_image", "image_url": url, "detail": "auto"}) + elif item.get("type") == "input_audio": + audio_info = item.get("input_audio") or {} + audio_data = audio_info.get("data") + if audio_data: + converted.append({ + "type": "input_audio", + "input_audio": {"data": audio_data, "format": audio_info.get("format", "wav")}, + }) + elif item.get("type") == "video_url": + # Codex doesn't support native video → text placeholder + placeholder = LLMProvider._media_placeholder("video_url", item) + converted.append({"type": "input_text", "text": placeholder["text"]}) if converted: return {"role": "user", "content": converted} return {"role": "user", "content": [{"type": "input_text", "text": ""}]} diff --git a/nanobot/utils/helpers.py b/nanobot/utils/helpers.py index 2a969298c..2c648f1f2 100644 --- a/nanobot/utils/helpers.py +++ b/nanobot/utils/helpers.py @@ -171,6 +171,79 @@ def detect_image_mime(data: bytes) -> str | None: return None +# Audio formats supported by OpenAI input_audio block +_AUDIO_MIME_COMPAT = {"audio/wav", "audio/mpeg", "audio/mp3", "audio/aac", + "audio/ogg", "audio/flac", "audio/x-m4a", "audio/mp4"} + +# Map MIME types to the format token expected by OpenAI-compatible input_audio APIs. +_AUDIO_FORMAT_MAP: dict[str, str] = { + "audio/wav": "wav", + "audio/x-wav": "wav", + "audio/mpeg": "mp3", + "audio/mp3": "mp3", + "audio/aac": "aac", + "audio/ogg": "ogg", + "audio/flac": "flac", + "audio/x-m4a": "m4a", + "audio/mp4": "m4a", +} + + +def detect_audio_mime(data: bytes, filename: str = "") -> str | None: + """Detect audio MIME type from magic bytes; fallback to filename guess.""" + if data[:4] == b"RIFF" and data[8:12] == b"WAVE": + return "audio/wav" + if data[:2] in (b"\xff\xfb", b"\xff\xf3", b"\xff\xf2", b"\xff\xfa"): + return "audio/mpeg" + if data[:4] == b"fLaC": + return "audio/flac" + if data[:4] == b"OggS": + return "audio/ogg" + if len(data) > 8 and data[4:8] == b"ftyp": + # Only claim audio for M4A-specific brands; avoid matching MP4 video. + brand = data[8:12] + if brand in (b"M4A ", b"M4AB", b"M4AC"): + return "audio/x-m4a" + if filename: + import mimetypes as _mt + guessed = _mt.guess_type(filename)[0] + if guessed and guessed.startswith("audio/"): + return guessed + return None + + +def audio_mime_compat(mime: str | None) -> bool: + """Check if the audio MIME is compatible with OpenAI input_audio block.""" + if not mime: + return False + return mime in _AUDIO_MIME_COMPAT + + +def audio_format_for_api(mime: str) -> str: + """Convert an audio MIME type to the format token expected by the API. + + Falls back to the subtype portion of the MIME (e.g. "x-m4a" from + "audio/x-m4a") when no explicit mapping exists. + """ + if not mime: + return "wav" + return _AUDIO_FORMAT_MAP.get(mime, mime.split("/")[-1]) + + +# Video formats commonly supported by LLM APIs (data URI inline) +_VIDEO_MIME_COMPAT = { + "video/mp4", "video/quicktime", "video/x-m4v", + "video/webm", "video/x-matroska", +} + + +def video_mime_compat(mime: str | None) -> bool: + """Check if the video MIME is in the commonly-supported set.""" + if not mime: + return False + return mime in _VIDEO_MIME_COMPAT + + def build_image_content_blocks( raw: bytes, mime: str, path: str, label: str ) -> list[dict[str, Any]]: diff --git a/tests/agent/test_multimodal.py b/tests/agent/test_multimodal.py new file mode 100644 index 000000000..455a8e0c0 --- /dev/null +++ b/tests/agent/test_multimodal.py @@ -0,0 +1,178 @@ +from pathlib import Path + +from nanobot.agent.context import ContextBuilder +from nanobot.config.schema import InputLimitsConfig +from nanobot.utils.helpers import detect_audio_mime, video_mime_compat + +PNG_BYTES = ( + b"\x89PNG\r\n\x1a\n" + b"\x00\x00\x00\rIHDR" + b"\x00\x00\x00\x01\x00\x00\x00\x01\x08\x02\x00\x00\x00" + b"\x90wS\xde" + b"\x00\x00\x00\x0cIDATx\x9cc``\x00\x00\x00\x04\x00\x01" + b"\x0b\x0e-\xb4" + b"\x00\x00\x00\x00IEND\xaeB`\x82" +) + +WAV_BYTES = b"RIFF\x24\x00\x00\x00WAVEfmt \x10\x00\x00\x00\x01\x00\x01\x00" + + +def _builder(tmp_path: Path, input_limits: InputLimitsConfig | None = None) -> ContextBuilder: + return ContextBuilder(tmp_path, input_limits=input_limits) + + +class TestAudioDetection: + def test_detect_wav_from_magic_bytes(self) -> None: + assert detect_audio_mime(WAV_BYTES) == "audio/wav" + + def test_detect_mp3_from_magic_bytes(self) -> None: + mp3 = b"\xff\xfb\x90\x00" + assert detect_audio_mime(mp3) == "audio/mpeg" + + def test_detect_fallback_to_filename(self) -> None: + assert detect_audio_mime(b"unknown", filename="song.mp3") == "audio/mpeg" + + def test_returns_none_for_non_audio(self) -> None: + assert detect_audio_mime(PNG_BYTES) is None + + +class TestVideoMimeCompat: + def test_mp4_is_compatible(self) -> None: + assert video_mime_compat("video/mp4") is True + + def test_unknown_is_not_compatible(self) -> None: + assert video_mime_compat("video/avi") is False + + def test_none_is_not_compatible(self) -> None: + assert video_mime_compat(None) is False + + +class TestBuildUserContentMultimodal: + def test_audio_block_when_supported(self, tmp_path: Path) -> None: + builder = _builder(tmp_path) + path = tmp_path / "voice.wav" + path.write_bytes(WAV_BYTES) + + content = builder._build_user_content("transcribe", [str(path)], supports_audio=True) + + assert isinstance(content, list) + audio_blocks = [b for b in content if b.get("type") == "input_audio"] + assert len(audio_blocks) == 1 + assert audio_blocks[0]["input_audio"]["format"] == "wav" + + def test_audio_placeholder_when_not_supported(self, tmp_path: Path) -> None: + builder = _builder(tmp_path) + path = tmp_path / "voice.wav" + path.write_bytes(WAV_BYTES) + + content = builder._build_user_content("transcribe", [str(path)], supports_audio=False) + + assert isinstance(content, list) + assert any("[audio:" in b.get("text", "") for b in content) + + def test_video_block_when_supported(self, tmp_path: Path) -> None: + builder = _builder(tmp_path) + path = tmp_path / "clip.mp4" + path.write_bytes(b"\x00" * 64) + + content = builder._build_user_content("describe", [str(path)], supports_video=True) + + assert isinstance(content, list) + video_blocks = [b for b in content if b.get("type") == "video_url"] + assert len(video_blocks) == 1 + assert video_blocks[0]["video_url"]["url"].startswith("data:video/mp4;base64,") + + def test_video_placeholder_when_not_supported(self, tmp_path: Path) -> None: + builder = _builder(tmp_path) + path = tmp_path / "clip.mp4" + path.write_bytes(b"\x00" * 64) + + content = builder._build_user_content("describe", [str(path)], supports_video=False) + + assert isinstance(content, list) + assert any("[video:" in b.get("text", "") for b in content) + + def test_vision_fallback_downgrades_image(self, tmp_path: Path) -> None: + builder = _builder(tmp_path) + path = tmp_path / "pic.png" + path.write_bytes(PNG_BYTES) + + content = builder._build_user_content("look", [str(path)], supports_vision=False) + + assert isinstance(content, list) + assert any("[image:" in b.get("text", "") for b in content) + assert not any(b.get("type") == "image_url" for b in content) + + def test_image_limit_count(self, tmp_path: Path) -> None: + builder = _builder(tmp_path) + paths = [] + for i in range(5): + path = tmp_path / f"img{i}.png" + path.write_bytes(PNG_BYTES) + paths.append(str(path)) + + content = builder._build_user_content("describe", paths) + + assert isinstance(content, list) + image_count = sum(1 for b in content if b.get("type") == "image_url") + assert image_count == 3 # default max_input_images + assert any("only the first 3 images" in b.get("text", "") for b in content) + + def test_image_limit_bytes(self, tmp_path: Path) -> None: + builder = _builder(tmp_path) + big = tmp_path / "big.png" + big.write_bytes(PNG_BYTES + b"x" * builder.input_limits.max_input_image_bytes) + + content = builder._build_user_content("analyze", [str(big)]) + + assert isinstance(content, str) + assert "file too large" in content + + def test_audio_limit_count(self, tmp_path: Path) -> None: + limits = InputLimitsConfig(max_input_audios=1) + builder = _builder(tmp_path, input_limits=limits) + for i in range(2): + path = tmp_path / f"snd{i}.wav" + path.write_bytes(WAV_BYTES) + + content = builder._build_user_content( + "compare", [str(tmp_path / "snd0.wav"), str(tmp_path / "snd1.wav")], supports_audio=True + ) + + assert isinstance(content, list) + audio_count = sum(1 for b in content if b.get("type") == "input_audio") + assert audio_count == 1 + assert any("only 1 audio" in b.get("text", "") for b in content) + + def test_audio_limit_bytes(self, tmp_path: Path) -> None: + limits = InputLimitsConfig(max_input_audio_bytes=32) + builder = _builder(tmp_path, input_limits=limits) + path = tmp_path / "big.wav" + path.write_bytes(WAV_BYTES + b"x" * 64) + + content = builder._build_user_content("analyze", [str(path)], supports_audio=True) + + assert isinstance(content, str) + assert "file too large" in content + + def test_mixed_media_types(self, tmp_path: Path) -> None: + builder = _builder(tmp_path) + img = tmp_path / "pic.png" + img.write_bytes(PNG_BYTES) + snd = tmp_path / "voice.wav" + snd.write_bytes(WAV_BYTES) + vid = tmp_path / "clip.mp4" + vid.write_bytes(b"\x00" * 64) + + content = builder._build_user_content( + "analyze", + [str(img), str(snd), str(vid)], + supports_vision=True, + supports_audio=True, + supports_video=True, + ) + + assert isinstance(content, list) + assert any(b.get("type") == "image_url" for b in content) + assert any(b.get("type") == "input_audio" for b in content) + assert any(b.get("type") == "video_url" for b in content) diff --git a/tests/channels/test_feishu_post_content.py b/tests/channels/test_feishu_post_content.py index a4c5bae19..a354234c4 100644 --- a/tests/channels/test_feishu_post_content.py +++ b/tests/channels/test_feishu_post_content.py @@ -27,10 +27,11 @@ def test_extract_post_content_supports_post_wrapper_shape() -> None: } } - text, image_keys = _extract_post_content(payload) + text, image_keys, media_items = _extract_post_content(payload) assert text == "日报 完成" assert image_keys == ["img_1"] + assert media_items == [] def test_extract_post_content_keeps_direct_shape_behavior() -> None: @@ -45,10 +46,29 @@ def test_extract_post_content_keeps_direct_shape_behavior() -> None: ], } - text, image_keys = _extract_post_content(payload) + text, image_keys, media_items = _extract_post_content(payload) assert text == "Daily report" assert image_keys == ["img_a", "img_b"] + assert media_items == [] + + +def test_extract_post_content_extracts_media_tags() -> None: + payload = { + "title": "Video", + "content": [ + [ + {"tag": "text", "text": "see this"}, + {"tag": "media", "file_key": "vid_1"}, + ] + ], + } + + text, image_keys, media_items = _extract_post_content(payload) + + assert text == "Video see this" + assert image_keys == [] + assert media_items == [{"tag": "media", "file_key": "vid_1"}] def test_register_optional_event_keeps_builder_when_method_missing() -> None: diff --git a/tests/providers/test_provider_retry.py b/tests/providers/test_provider_retry.py index 4b72c163a..f3532e4c5 100644 --- a/tests/providers/test_provider_retry.py +++ b/tests/providers/test_provider_retry.py @@ -242,7 +242,7 @@ async def test_image_fallback_returns_error_on_second_failure() -> None: @pytest.mark.asyncio async def test_image_fallback_without_meta_uses_default_placeholder() -> None: - """When _meta is absent, fallback placeholder is '[image omitted]'.""" + """When _meta is absent, fallback placeholder is '[image]'.""" provider = ScriptedProvider([ LLMResponse(content="error", finish_reason="error"), LLMResponse(content="ok"), @@ -256,7 +256,7 @@ async def test_image_fallback_without_meta_uses_default_placeholder() -> None: for msg in msgs_on_retry: content = msg.get("content") if isinstance(content, list): - assert any("[image omitted]" in (b.get("text") or "") for b in content) + assert any("[image]" in (b.get("text") or "") for b in content) @pytest.mark.asyncio