diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 628713224..415f3b2dc 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -362,6 +362,159 @@ def _multimodal_validation_error(exc: ValueError, *, param: str) -> "web.Respons ) +def _is_coder_model(model_name: Optional[str]) -> bool: + """Return True if model_name looks like a coder model (case-insensitive 'coder' in name).""" + if not model_name: + return False + return "coder" in str(model_name).lower() + + +def _extract_image_urls_from_messages(messages: List[Dict[str, Any]]) -> List[str]: + """Extract all image URLs from OpenAI-style messages payload.""" + image_urls: List[str] = [] + for msg in messages: + if not isinstance(msg, dict): + continue + content = msg.get("content") + if not content: + continue + # Normalize multimodal content to list of parts + parts: Any = content + if isinstance(content, str): + continue # text-only message has no images + if not isinstance(parts, list): + continue + for part in parts: + if not isinstance(part, dict): + continue + part_type = str(part.get("type", "")).strip().lower() + if part_type in {"image_url", "input_image"}: + img_ref = part.get("image_url") + if isinstance(img_ref, dict): + url = img_ref.get("url") + else: + url = img_ref + if isinstance(url, str) and url.strip(): + image_urls.append(url.strip()) + return image_urls + + +async def _maybe_enrich_with_vision_if_coder( + body: Dict[str, Any], +) -> Dict[str, Any]: + """ + For coder model requests with image attachments, pre-analyze images + with the vision fallback model and replace image parts with descriptive text. + + Returns a modified copy of body with image URLs replaced by text + descriptions in the messages array. + """ + model_name = body.get("model") + if not _is_coder_model(model_name): + return body + + messages = body.get("messages") + if not messages or not isinstance(messages, list): + return body + + image_urls = _extract_image_urls_from_messages(messages) + if not image_urls: + return body + + # Import here to avoid circular dependency issues + try: + from tools.vision_tools import vision_analyze_tool + except ImportError: + logger.warning( + "Vision tool not available for coder model enrichment. " + "Install hermes-agent tools to enable this feature." + ) + return body + + # Use the configured vision model or fall back to VL-30B + # We'll let vision_analyze_tool use its own default (auxiliary.vision) + # but we can optionally override by passing model parameter + # For now, let it auto-resolve via config + analysis_prompt = ( + "Describe everything visible in this image in thorough detail. " + "Include any text, code, data, objects, people, layout, colors, " + "and any other notable visual information." + ) + + # Build a mapping of URL -> description + url_to_desc: Dict[str, str] = {} + for img_url in image_urls: + try: + result_json = await vision_analyze_tool( + image_url=img_url, + user_prompt=analysis_prompt, + ) + result = json.loads(result_json) + if result.get("success"): + desc = result.get("analysis", "") + url_to_desc[img_url] = ( + f"[The user sent an image~ Here's what I can see:\n{desc}]\n" + f"[If you need a closer look, use vision_analyze with image_url: {img_url} ~]" + ) + else: + url_to_desc[img_url] = ( + "[The user sent an image but I couldn't quite see it this time (>_<) " + f"You can try looking at it yourself with vision_analyze using image_url: {img_url}]" + ) + except Exception as e: + logger.error("Vision auto-analysis error for image %s: %s", img_url, e) + url_to_desc[img_url] = ( + f"[The user sent an image but something went wrong when I tried to look at it~ " + f"You can try examining it yourself with vision_analyze using image_url: {img_url}]" + ) + + # Now replace image parts with text in a copy of messages + enriched_messages: List[Dict[str, Any]] = [] + for msg in messages: + msg_copy = dict(msg) + content = msg_copy.get("content") + if isinstance(content, str): + enriched_messages.append(msg_copy) + continue + + if not isinstance(content, list): + enriched_messages.append(msg_copy) + continue + + enriched_parts: List[Any] = [] + for part in content: + if not isinstance(part, dict): + enriched_parts.append(part) + continue + + part_type = str(part.get("type", "")).strip().lower() + if part_type in {"image_url", "input_image"}: + img_ref = part.get("image_url") + if isinstance(img_ref, dict): + img_url = img_ref.get("url") + else: + img_url = img_ref + if isinstance(img_url, str) and img_url in url_to_desc: + # Replace image part with text part + enriched_parts.append({ + "type": "text", + "text": url_to_desc[img_url], + }) + else: + # Keep original if URL not found (shouldn't happen) + enriched_parts.append(part) + else: + enriched_parts.append(part) + + msg_copy["content"] = enriched_parts + enriched_messages.append(msg_copy) + + # Return modified body + body_copy = dict(body) + body_copy["messages"] = enriched_messages + return body_copy + + def _session_chat_user_message(body: Dict[str, Any], *, param: str = "message") -> tuple[Any, Optional["web.Response"]]: """Parse and normalize session chat ``message`` / ``input`` like chat completions.""" user_message = body.get("message") or body.get("input")