diff --git a/test/test_api_client_multimodal_sanitize.py b/test/test_api_client_multimodal_sanitize.py new file mode 100644 index 0000000..cfecd9f --- /dev/null +++ b/test/test_api_client_multimodal_sanitize.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import unittest + +from utils.api_client import DeepSeekClient + + +class APIClientMultimodalSanitizeTest(unittest.TestCase): + def test_text_only_model_strips_image_parts_to_text(self): + client = DeepSeekClient(web_mode=True) + client.model_key = "DeepSeek-V4-Pro High" # custom_models.json: multimodal=none + + messages = [ + { + "role": "tool", + "name": "mcp__playwright__browser_take_screenshot", + "content": [ + {"type": "text", "text": "截图成功"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,abc"}}, + ], + } + ] + + sanitized = client._sanitize_messages_for_model_capability(messages) + self.assertEqual(len(sanitized), 1) + self.assertIsInstance(sanitized[0].get("content"), str, sanitized[0]) + self.assertIn("截图成功", sanitized[0].get("content") or "", sanitized[0]) + + def test_multimodal_model_keeps_image_parts(self): + client = DeepSeekClient(web_mode=True) + client.model_key = "qwen3-vl-plus" # 内置多模态模型 + + messages = [ + { + "role": "user", + "content": [ + {"type": "text", "text": "看图"}, + {"type": "image_url", "image_url": {"url": "data:image/png;base64,abc"}}, + ], + } + ] + + sanitized = client._sanitize_messages_for_model_capability(messages) + self.assertEqual(len(sanitized), 1) + self.assertIsInstance(sanitized[0].get("content"), list, sanitized[0]) + self.assertEqual( + [part.get("type") for part in sanitized[0].get("content") or []], + ["text", "image_url"], + sanitized[0], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/utils/api_client.py b/utils/api_client.py index d43534b..8160bb9 100644 --- a/utils/api_client.py +++ b/utils/api_client.py @@ -316,6 +316,101 @@ class DeepSeekClient: "content": "\n\n".join(c for c in merged_contents if c) } return [merged] + messages[idx:] + + @staticmethod + def _sanitize_content_for_capability( + content: Any, + *, + supports_image: bool, + supports_video: bool, + ) -> Any: + """根据模型多模态能力裁剪 content,避免向纯文本模型发送 image_url/video_url。""" + if not isinstance(content, list): + return content + + kept_parts: List[Dict[str, Any]] = [] + text_segments: List[str] = [] + + for part in content: + if not isinstance(part, dict): + text = str(part).strip() + if text: + text_segments.append(text) + continue + + ctype = str(part.get("type") or "").strip().lower() + if ctype == "text": + text = str(part.get("text") or "") + if text: + text_segments.append(text) + kept_parts.append({"type": "text", "text": text}) + continue + if ctype == "image_url": + if supports_image: + kept_parts.append(part) + continue + if ctype == "video_url": + if supports_video: + kept_parts.append(part) + continue + + # 其他未知/暂不支持类型统一忽略(纯文本模型下只保留 text) + continue + + # 纯文本模型:统一回退为字符串,规避供应商对 content part 类型的严格校验 + if not supports_image and not supports_video: + return "\n".join(seg for seg in text_segments if seg).strip() + + # 半多模态模型(例如只支持图片):尽量保留可支持的 part + if kept_parts: + if len(kept_parts) == 1 and kept_parts[0].get("type") == "text": + return kept_parts[0].get("text", "") + return kept_parts + return "\n".join(seg for seg in text_segments if seg).strip() + + def _sanitize_messages_for_model_capability(self, messages: List[Dict]) -> List[Dict]: + model_key = str(self.model_key or "").strip() + if not model_key: + return messages + + try: + from config.model_profiles import model_supports_image, model_supports_video + except Exception: + return messages + + supports_image = bool(model_supports_image(model_key)) + supports_video = bool(model_supports_video(model_key)) + if supports_image and supports_video: + return messages + + changed = 0 + sanitized: List[Dict[str, Any]] = [] + for message in messages or []: + if not isinstance(message, dict): + continue + msg_copy = dict(message) + original_content = msg_copy.get("content") + new_content = self._sanitize_content_for_capability( + original_content, + supports_image=supports_image, + supports_video=supports_video, + ) + if new_content != original_content: + changed += 1 + msg_copy["content"] = new_content + sanitized.append(msg_copy) + + if changed: + self._debug_log( + { + "event": "sanitize_messages_for_model_capability", + "model_key": model_key, + "supports_image": supports_image, + "supports_video": supports_video, + "changed_messages": changed, + } + ) + return sanitized def set_deep_thinking_mode(self, enabled: bool): """配置深度思考模式(持续使用思考模型)。""" @@ -596,6 +691,7 @@ class DeepSeekClient: is_minimax = self.model_key == "minimax-m2.5" or "minimax" in lower_base_url final_messages = self._merge_system_messages(messages) + final_messages = self._sanitize_messages_for_model_capability(final_messages) payload = { "model": api_config["model_id"],