From 727cafc37027e6f0e4d0e29b48b4c90dfad762c0 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Mon, 27 Apr 2026 12:21:45 +0800 Subject: [PATCH] fix: persist media via store and preserve MCP content --- core/main_terminal_parts/commands.py | 43 ++- core/main_terminal_parts/context.py | 20 +- modules/mcp_client_manager.py | 40 ++- server/chat_flow_tool_loop.py | 84 +++-- test/test_media_store_and_mcp_content.py | 130 +++++++ utils/context_manager.py | 410 ++++++++++++++++++++--- utils/media_store.py | 369 ++++++++++++++++++++ utils/tool_result_formatter.py | 188 +++++++++++ 8 files changed, 1201 insertions(+), 83 deletions(-) create mode 100644 test/test_media_store_and_mcp_content.py create mode 100644 utils/media_store.py diff --git a/core/main_terminal_parts/commands.py b/core/main_terminal_parts/commands.py index 114a0d3..d85b84a 100644 --- a/core/main_terminal_parts/commands.py +++ b/core/main_terminal_parts/commands.py @@ -73,7 +73,10 @@ from modules.container_monitor import collect_stats, inspect_state from core.tool_config import TOOL_CATEGORIES from utils.api_client import DeepSeekClient from utils.context_manager import ContextManager -from utils.tool_result_formatter import format_tool_result_for_context +from utils.tool_result_formatter import ( + extract_mcp_content_for_context, + format_tool_result_for_context, +) from utils.logger import setup_logger from config.model_profiles import ( get_model_profile, @@ -226,7 +229,37 @@ class MainTerminalCommandMixin: result_data = parsed if isinstance(parsed, dict) else {} except Exception: result_data = {} - tool_result_content = format_tool_result_for_context(tool_name, result_data, result) + tool_metadata = {"tool_payload": result_data} if result_data else None + if isinstance(tool_name, str) and tool_name.startswith("mcp__") and isinstance(result_data, dict): + mcp_parsed = extract_mcp_content_for_context(result_data) + tool_result_content = ( + mcp_parsed.get("text") + or format_tool_result_for_context(tool_name, result_data, result) + ) + tool_metadata = {"tool_payload": mcp_parsed.get("sanitized_payload") or result_data} + mcp_media_items = mcp_parsed.get("media_items") or [] + tool_media_refs = [] + for item in mcp_media_items: + if not isinstance(item, dict): + continue + tool_media_refs.append( + { + "kind": item.get("kind"), + "mime_type": item.get("mime_type"), + "data_base64": item.get("data_base64"), + "source": "mcp_content", + "item_type": item.get("item_type"), + "label": item.get("label"), + "name": item.get("name"), + "title": item.get("title"), + "uri": item.get("uri"), + "url": item.get("url"), + "index": item.get("index"), + } + ) + else: + tool_result_content = format_tool_result_for_context(tool_name, result_data, result) + tool_media_refs = None tool_images = None tool_videos = None if ( @@ -250,8 +283,10 @@ class MainTerminalCommandMixin: "system_message": result_data.get("system_message") if isinstance(result_data, dict) else None, "task_id": result_data.get("task_id") if isinstance(result_data, dict) else None, "raw_result_data": result_data if result_data else None, + "metadata": tool_metadata, "images": tool_images, "videos": tool_videos, + "media_refs": tool_media_refs, }) return result @@ -290,8 +325,10 @@ class MainTerminalCommandMixin: tool_result["content"], tool_call_id=tool_result["tool_call_id"], name=tool_result["name"], + metadata=tool_result.get("metadata"), images=tool_result.get("images"), - videos=tool_result.get("videos") + videos=tool_result.get("videos"), + media_refs=tool_result.get("media_refs"), ) system_message = tool_result.get("system_message") if system_message: diff --git a/core/main_terminal_parts/context.py b/core/main_terminal_parts/context.py index 151e3d8..b80a2fe 100644 --- a/core/main_terminal_parts/context.py +++ b/core/main_terminal_parts/context.py @@ -313,11 +313,17 @@ class MainTerminalContextMixin: # Tool消息需要保留完整结构 images = conv.get("images") or metadata.get("images") or [] videos = conv.get("videos") or metadata.get("videos") or [] + media_refs = conv.get("media_refs") or metadata.get("media_refs") or [] content_value = conv.get("content") if isinstance(content_value, list): content_payload = content_value - elif images or videos: - content_payload = self.context_manager._build_content_with_images(content_value, images, videos) + elif images or videos or media_refs: + content_payload = self.context_manager._build_content_with_images( + content_value, + images, + videos, + media_refs=media_refs, + ) else: content_payload = content_value message = { @@ -338,9 +344,15 @@ class MainTerminalContextMixin: # User 或普通 System 消息 images = conv.get("images") or metadata.get("images") or [] videos = conv.get("videos") or metadata.get("videos") or [] + media_refs = conv.get("media_refs") or metadata.get("media_refs") or [] content_payload = ( - self.context_manager._build_content_with_images(conv["content"], images, videos) - if (images or videos) else conv["content"] + self.context_manager._build_content_with_images( + conv["content"], + images, + videos, + media_refs=media_refs, + ) + if (images or videos or media_refs) else conv["content"] ) # 调试：记录所有 system 消息 if conv["role"] == "system": diff --git a/modules/mcp_client_manager.py b/modules/mcp_client_manager.py index 538b48c..b2d08bf 100644 --- a/modules/mcp_client_manager.py +++ b/modules/mcp_client_manager.py @@ -893,11 +893,47 @@ class MCPClientManager: def _build_call_message(result: Dict[str, Any]) -> str: content = result.get("content") if isinstance(content, list): - for item in content: + lines: List[str] = [] + for idx, item in enumerate(content, start=1): if isinstance(item, dict) and item.get("type") == "text": text = str(item.get("text") or "").strip() if text: - return text[:500] + lines.append(text) + continue + if not isinstance(item, dict): + continue + ctype = str(item.get("type") or "").strip().lower() + if ctype in {"image", "audio"}: + mime = str(item.get("mimeType") or item.get("mime_type") or "").strip() + if not mime: + mime = "image/png" if ctype == "image" else "audio/wav" + data_len = len(str(item.get("data") or "")) + lines.append(f"[MCP {ctype} #{idx}] {mime} ({data_len} base64 chars)") + continue + if ctype in {"resource_link", "resourcelink"}: + uri = str(item.get("uri") or item.get("url") or "").strip() + name = str(item.get("name") or item.get("title") or "").strip() + if uri and name: + lines.append(f"[MCP 资源] {name} -> {uri}") + elif uri: + lines.append(f"[MCP 资源] {uri}") + continue + if ctype == "resource": + resource = item.get("resource") + if isinstance(resource, dict): + text = str(resource.get("text") or "").strip() + if text: + lines.append(text) + continue + uri = str(resource.get("uri") or "").strip() + if uri: + lines.append(f"[MCP 资源] {uri}") + continue + uri = str(item.get("uri") or item.get("url") or "").strip() + if uri: + lines.append(f"[MCP 资源] {uri}") + if lines: + return "\n".join(lines)[:500] structured = result.get("structuredContent") if structured is not None: try: diff --git a/server/chat_flow_tool_loop.py b/server/chat_flow_tool_loop.py index 7bc718b..eab2e8f 100644 --- a/server/chat_flow_tool_loop.py +++ b/server/chat_flow_tool_loop.py @@ -13,7 +13,10 @@ from .monitor import cache_monitor_snapshot from .security import compact_web_search_result from .chat_flow_helpers import detect_tool_failure from .chat_flow_runner_helpers import resolve_monitor_path, resolve_monitor_memory, capture_monitor_snapshot -from utils.tool_result_formatter import format_tool_result_for_context +from utils.tool_result_formatter import ( + extract_mcp_content_for_context, + format_tool_result_for_context, +) from utils.context_manager import AUTO_SHALLOW_PLACEHOLDER from config import TOOL_CALL_COOLDOWN from modules.personalization_manager import load_personalization_config, resolve_context_compression_settings @@ -696,16 +699,48 @@ async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, clie metadata_payload = None tool_images = None tool_videos = None + tool_media_refs = None if isinstance(result_data, dict): - # 特殊处理 web_search：保留可供前端渲染的精简结构，以便历史记录复现搜索结果 - if function_name == "web_search": - try: - tool_result_content = json.dumps(compact_web_search_result(result_data), ensure_ascii=False) - except Exception: - tool_result_content = tool_result + if isinstance(function_name, str) and function_name.startswith("mcp__"): + mcp_parsed = extract_mcp_content_for_context(result_data) + tool_result_content = ( + mcp_parsed.get("text") + or format_tool_result_for_context(function_name, result_data, tool_result) + ) + mcp_media_items = mcp_parsed.get("media_items") or [] + if mcp_media_items: + tool_media_refs = [] + for item in mcp_media_items: + if not isinstance(item, dict): + continue + tool_media_refs.append( + { + "kind": item.get("kind"), + "mime_type": item.get("mime_type"), + "data_base64": item.get("data_base64"), + "source": "mcp_content", + "item_type": item.get("item_type"), + "label": item.get("label"), + "name": item.get("name"), + "title": item.get("title"), + "uri": item.get("uri"), + "url": item.get("url"), + "index": item.get("index"), + } + ) + metadata_payload = { + "tool_payload": mcp_parsed.get("sanitized_payload") or result_data + } else: - tool_result_content = format_tool_result_for_context(function_name, result_data, tool_result) - metadata_payload = {"tool_payload": result_data} + # 特殊处理 web_search：保留可供前端渲染的精简结构，以便历史记录复现搜索结果 + if function_name == "web_search": + try: + tool_result_content = json.dumps(compact_web_search_result(result_data), ensure_ascii=False) + except Exception: + tool_result_content = tool_result + else: + tool_result_content = format_tool_result_for_context(function_name, result_data, tool_result) + metadata_payload = {"tool_payload": result_data} else: tool_result_content = tool_result tool_message_content = tool_result_content @@ -748,27 +783,32 @@ async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, clie 'content': f'系统已记录视频路径（不再附带二进制数据）: {video_path}' }) - # 将工具结果即时组装为多模态消息，确保同一轮 tool loop 的下一次模型请求能真正看到图片/视频 - if tool_images or tool_videos: - try: - tool_message_content = web_terminal.context_manager._build_content_with_images( - str(tool_result_content or ""), - tool_images or [], - tool_videos or [] - ) - except Exception: - tool_message_content = tool_result_content - # 立即保存工具结果 - web_terminal.context_manager.add_conversation( + saved_tool_message = web_terminal.context_manager.add_conversation( "tool", tool_result_content, tool_call_id=tool_call_id, name=function_name, metadata=metadata_payload, images=tool_images, - videos=tool_videos + videos=tool_videos, + media_refs=tool_media_refs, ) + saved_media_refs = [] + if isinstance(saved_tool_message, dict): + saved_media_refs = saved_tool_message.get("media_refs") or [] + + # 将工具结果即时组装为多模态消息，确保同一轮 tool loop 的下一次模型请求能真正看到媒体 + if tool_images or tool_videos or saved_media_refs: + try: + tool_message_content = web_terminal.context_manager._build_content_with_images( + str(tool_result_content or ""), + tool_images or [], + tool_videos or [], + media_refs=saved_media_refs, + ) + except Exception: + tool_message_content = tool_result_content try: workspace_data_dir = getattr(workspace, "data_dir", None) if workspace else None personal_config = load_personalization_config(workspace_data_dir) if workspace_data_dir else {} diff --git a/test/test_media_store_and_mcp_content.py b/test/test_media_store_and_mcp_content.py new file mode 100644 index 0000000..1b40253 --- /dev/null +++ b/test/test_media_store_and_mcp_content.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path + +from utils.context_manager import ContextManager +from utils.tool_result_formatter import ( + extract_mcp_content_for_context, + format_tool_result_for_context, +) + + +class MediaStoreAndMCPContentTest(unittest.TestCase): + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + root = Path(self.temp_dir.name) + self.project_dir = root / "project" + self.project_dir.mkdir(parents=True, exist_ok=True) + self.data_dir = root / "data" + self.data_dir.mkdir(parents=True, exist_ok=True) + + def tearDown(self): + self.temp_dir.cleanup() + + def test_media_store_survives_source_file_deletion(self): + image_path = self.project_dir / "demo.png" + image_path.write_bytes(b"\x89PNG\r\n\x1a\nfake-image-data") + + ctx = ContextManager(project_path=str(self.project_dir), data_dir=str(self.data_dir)) + conv_id = ctx.start_new_conversation(project_path=str(self.project_dir), thinking_mode=False) + ctx.add_conversation("user", "看图", images=["demo.png"]) + message = ctx.conversation_history[-1] + + refs = message.get("media_refs") or [] + self.assertTrue(refs, message) + self.assertEqual(refs[0].get("kind"), "image") + + image_path.unlink() + payload = ctx._build_content_with_images( + message.get("content") or "", + message.get("images") or [], + message.get("videos") or [], + media_refs=refs, + ) + + self.assertIsInstance(payload, list, payload) + image_parts = [item for item in payload if isinstance(item, dict) and item.get("type") == "image_url"] + self.assertTrue(image_parts, payload) + self.assertTrue(str(image_parts[0].get("image_url", {}).get("url", "")).startswith("data:image/")) + + index_file = self.data_dir / "conversations" / "media_store" / "index.json" + self.assertTrue(index_file.exists()) + index_data = json.loads(index_file.read_text(encoding="utf-8")) + self.assertIn(conv_id, index_data.get("message_map", {})) + self.assertIn(message.get("message_id"), (index_data.get("message_map", {}).get(conv_id) or {})) + + def test_add_tool_message_with_inline_mcp_media_refs(self): + ctx = ContextManager(project_path=str(self.project_dir), data_dir=str(self.data_dir)) + ctx.start_new_conversation(project_path=str(self.project_dir), thinking_mode=False) + + saved = ctx.add_conversation( + "tool", + "截图已记录", + tool_call_id="call_1", + name="mcp__browsermcp__browser_screenshot", + media_refs=[ + { + "kind": "image", + "mime_type": "image/png", + "data_base64": "aGVsbG8=", + "source": "mcp_content", + } + ], + ) + + refs = (saved or {}).get("media_refs") or [] + self.assertTrue(refs, saved) + self.assertTrue(str(refs[0].get("media_id") or "").startswith("sha256:")) + + payload = ctx._build_content_with_images( + saved.get("content") or "", + [], + [], + media_refs=refs, + ) + self.assertIsInstance(payload, list, payload) + image_parts = [item for item in payload if isinstance(item, dict) and item.get("type") == "image_url"] + self.assertTrue(image_parts, payload) + self.assertTrue(str(image_parts[0].get("image_url", {}).get("url", "")).startswith("data:image/png;base64,")) + + def test_mcp_formatter_keeps_non_text_information(self): + result_data = { + "success": True, + "server_id": "browsermcp", + "tool_alias": "mcp__browsermcp__browser_screenshot", + "tool_name": "browser_screenshot", + "message": "MCP 工具调用完成", + "content": [ + { + "type": "image", + "mimeType": "image/png", + "data": "aGVsbG8=", + } + ], + "raw_result": { + "content": [ + { + "type": "image", + "mimeType": "image/png", + "data": "aGVsbG8=", + } + ] + }, + } + + text = format_tool_result_for_context("mcp__browsermcp__browser_screenshot", result_data, "") + self.assertIn("MCP image #1", text) + parsed = extract_mcp_content_for_context(result_data) + self.assertTrue(parsed.get("media_items")) + sanitized = parsed.get("sanitized_payload") or {} + content = (sanitized.get("content") or [{}])[0] + self.assertNotIn("data", content) + self.assertTrue(content.get("data_omitted")) + + +if __name__ == "__main__": + unittest.main() + diff --git a/utils/context_manager.py b/utils/context_manager.py index a63dc09..55fb45a 100644 --- a/utils/context_manager.py +++ b/utils/context_manager.py @@ -5,6 +5,7 @@ import json import base64 import mimetypes import io +import uuid import platform import shutil import subprocess @@ -54,6 +55,7 @@ except ImportError: model_supports_video, ) from utils.conversation_manager import ConversationManager +from utils.media_store import MediaStore AUTO_SHALLOW_PLACEHOLDER = "过早的工具结果已经被自动压缩" AUTO_SHALLOW_TOOL_WHITELIST = { @@ -95,6 +97,7 @@ class ContextManager: # 新增：对话持久化管理器 self.conversation_manager = ConversationManager(base_dir=self.data_dir) + self.media_store = MediaStore(self.data_dir) self.current_conversation_id: Optional[str] = None self.auto_save_enabled = True self.main_terminal = None # 由宿主终端在初始化后回填，用于工具定义访问 @@ -837,6 +840,24 @@ class ContextManager: model_key = metadata.get("model_key") self.has_images = metadata.get("has_images", False) self.has_videos = metadata.get("has_videos", False) + if not self.has_images or not self.has_videos: + for msg in self.conversation_history: + if not isinstance(msg, dict): + continue + images = msg.get("images") or [] + videos = msg.get("videos") or [] + if images: + self.has_images = True + if videos: + self.has_videos = True + for ref in (msg.get("media_refs") or []): + kind = str((ref or {}).get("kind") or "").strip().lower() + if kind == "image": + self.has_images = True + elif kind == "video": + self.has_videos = True + if self.has_images and self.has_videos: + break if self.main_terminal: try: if model_key: @@ -1329,6 +1350,189 @@ class ContextManager: print(f"[Debug] 广播token更新失败: {e}") import traceback traceback.print_exc() + + def _generate_message_id(self) -> str: + """生成消息唯一 ID（用于 media_store 映射）。""" + return f"msg_{uuid.uuid4().hex}" + + def _resolve_media_input_path(self, raw_path: str) -> Optional[Path]: + path_text = str(raw_path or "").strip() + if not path_text: + return None + try: + candidate = Path(path_text).expanduser() + if candidate.is_absolute(): + return candidate.resolve() + return (Path(self.project_path) / candidate).resolve() + except Exception: + return None + + def _store_media_item_from_candidate(self, candidate: Dict[str, Any]) -> Optional[Dict[str, Any]]: + if not isinstance(candidate, dict): + return None + kind = str(candidate.get("kind") or "").strip().lower() or "binary" + mime_type = candidate.get("mime_type") + source_path = candidate.get("source_path") + base_ref: Optional[Dict[str, Any]] = None + try: + media_id = str(candidate.get("media_id") or "").strip() + if media_id: + data_url = self.media_store.to_data_url(candidate) + if data_url: + base_ref = { + "media_id": media_id, + "kind": kind, + "mime_type": mime_type, + "store_path": candidate.get("store_path"), + "source_path": source_path, + } + else: + base_ref = None + elif candidate.get("data_url"): + stored = self.media_store.store_data_url( + str(candidate.get("data_url")), + kind=kind, + source_path=source_path, + ) + base_ref = dict(stored) + elif candidate.get("data_base64") or candidate.get("data"): + payload = str(candidate.get("data_base64") or candidate.get("data") or "") + if payload: + stored = self.media_store.store_base64_data( + payload, + kind=kind, + mime_type=mime_type, + source_path=source_path, + ) + base_ref = dict(stored) + elif candidate.get("path"): + abs_path = self._resolve_media_input_path(str(candidate.get("path"))) + if abs_path and abs_path.exists() and abs_path.is_file(): + stored = self.media_store.store_file( + abs_path, + kind=kind, + mime_type=mime_type, + source_path=source_path or str(candidate.get("path")), + ) + base_ref = dict(stored) + except Exception: + base_ref = None + if not base_ref: + return None + # 保留调用侧附加信息（例如视频 fps、MCP item 类型等） + passthrough_keys = { + "label", + "fps", + "source", + "origin", + "item_type", + "name", + "title", + "uri", + "url", + "index", + } + for key in passthrough_keys: + if key in candidate and candidate.get(key) is not None: + base_ref[key] = candidate.get(key) + if base_ref.get("source_path") is None and source_path: + base_ref["source_path"] = source_path + return base_ref + + def _prepare_media_candidates( + self, + images: Optional[List[Any]] = None, + videos: Optional[List[Any]] = None, + media_refs: Optional[List[Dict[str, Any]]] = None, + ) -> List[Dict[str, Any]]: + candidates: List[Dict[str, Any]] = [] + for item in media_refs or []: + if not isinstance(item, dict): + continue + kind = str(item.get("kind") or "").strip().lower() + if not kind: + mime_hint = str(item.get("mime_type") or item.get("mimeType") or "").strip() + if mime_hint.startswith("image/"): + kind = "image" + elif mime_hint.startswith("video/"): + kind = "video" + elif mime_hint.startswith("audio/"): + kind = "audio" + candidate = dict(item) + if kind: + candidate["kind"] = kind + if "mimeType" in candidate and "mime_type" not in candidate: + candidate["mime_type"] = candidate.get("mimeType") + candidates.append(candidate) + + for path in images or []: + if isinstance(path, dict): + candidate = dict(path) + candidate.setdefault("kind", "image") + if "mimeType" in candidate and "mime_type" not in candidate: + candidate["mime_type"] = candidate.get("mimeType") + if "path" not in candidate and path.get("url"): + candidate["path"] = path.get("url") + candidates.append(candidate) + continue + path_text = str(path or "").strip() + if not path_text: + continue + candidates.append( + { + "kind": "image", + "path": path_text, + "source_path": path_text, + "source": "message_images", + } + ) + + for item in videos or []: + if isinstance(item, dict): + path_text = str(item.get("path") or "").strip() + if not path_text: + continue + candidate = { + "kind": "video", + "path": path_text, + "source_path": path_text, + "fps": item.get("fps"), + "source": item.get("source") or "message_videos", + } + if item.get("mime_type"): + candidate["mime_type"] = item.get("mime_type") + elif item.get("mimeType"): + candidate["mime_type"] = item.get("mimeType") + candidates.append(candidate) + else: + path_text = str(item or "").strip() + if not path_text: + continue + candidates.append( + { + "kind": "video", + "path": path_text, + "source_path": path_text, + "source": "message_videos", + } + ) + return candidates + + @staticmethod + def _dedupe_media_refs(media_refs: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + deduped: List[Dict[str, Any]] = [] + seen = set() + for item in media_refs: + if not isinstance(item, dict): + continue + media_id = str(item.get("media_id") or "").strip() + source_path = str(item.get("source_path") or "").strip() + key = (media_id, source_path, str(item.get("kind") or "")) + if key in seen: + continue + seen.add(key) + deduped.append(item) + return deduped def add_conversation( self, @@ -1339,33 +1543,61 @@ class ContextManager: name: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, reasoning_content: Optional[str] = None, - images: Optional[List[str]] = None, - videos: Optional[List[str]] = None + images: Optional[List[Any]] = None, + videos: Optional[List[Any]] = None, + media_refs: Optional[List[Dict[str, Any]]] = None, ): """添加对话记录（改进版：集成自动保存 + 智能token统计）""" timestamp = datetime.now().isoformat() + message_id = self._generate_message_id() if role == "assistant": message = { "role": role, "reasoning_content": reasoning_content if reasoning_content is not None else "", "content": content or "", - "timestamp": timestamp + "timestamp": timestamp, + "message_id": message_id, } else: message = { "role": role, "content": content, - "timestamp": timestamp + "timestamp": timestamp, + "message_id": message_id, } if metadata: - message["metadata"] = metadata + message["metadata"] = dict(metadata) if images: message["images"] = images self.has_images = True if videos: message["videos"] = videos self.has_videos = True + + prepared_candidates = self._prepare_media_candidates(images=images, videos=videos, media_refs=media_refs) + stored_media_refs: List[Dict[str, Any]] = [] + for candidate in prepared_candidates: + stored_ref = self._store_media_item_from_candidate(candidate) + if stored_ref: + stored_media_refs.append(stored_ref) + stored_media_refs = self._dedupe_media_refs(stored_media_refs) + if stored_media_refs: + message["media_refs"] = stored_media_refs + message.setdefault("metadata", {}) + message["metadata"]["media_refs"] = stored_media_refs + if any(str(item.get("kind") or "").strip().lower() == "image" for item in stored_media_refs): + self.has_images = True + if any(str(item.get("kind") or "").strip().lower() == "video" for item in stored_media_refs): + self.has_videos = True + try: + self.media_store.link_message_media( + self.current_conversation_id, + message_id, + stored_media_refs, + ) + except Exception: + pass # 记录当前助手回复所用模型，便于回放时查看 if role == "assistant": @@ -1406,6 +1638,7 @@ class ContextManager: print(f"[Debug] 添加{role}消息后广播token更新") self.safe_broadcast_token_update() + return message def add_tool_result(self, tool_call_id: str, function_name: str, result: str): """添加工具调用结果（保留方法以兼容）""" @@ -1866,65 +2099,140 @@ class ContextManager: except Exception: return None - def _build_content_with_images(self, text: str, images: List[str], videos: Optional[List[Any]] = None) -> Any: - """将文本与图片/视频路径组合成多模态content，图片转换为data URI，视频转换为 data URL。""" + def _build_content_with_images( + self, + text: str, + images: List[Any], + videos: Optional[List[Any]] = None, + media_refs: Optional[List[Dict[str, Any]]] = None, + ) -> Any: + """将文本与媒体组合成多模态 content。 + + 优先使用 media_refs（来自 media_store），回退到历史路径字段（images/videos）。 + """ videos = videos or [] - if not images and not videos: + media_refs = media_refs or [] + if not images and not videos and not media_refs: return text + parts: List[Dict[str, Any]] = [] extra_videos: List[Any] = [] + extra_notes: List[str] = [] current_model = getattr(getattr(self, "main_terminal", None), "model_key", None) supports_video_fps = current_model == "qwen3-vl-plus" qwen_video_fps = 2 - if text: - parts.append({"type": "text", "text": text}) - for path in images: - try: - abs_path = Path(self.project_path) / path - if not abs_path.exists() or not abs_path.is_file(): + + media_payload_attached = False + + if media_refs: + for ref in media_refs: + if not isinstance(ref, dict): continue - mime, _ = mimetypes.guess_type(abs_path.name) - if mime and mime.startswith("video/"): - extra_videos.append(path) - continue - if mime and not mime.startswith("image/"): - continue - data_url = self._compress_image_if_needed(abs_path) + kind = str(ref.get("kind") or "").strip().lower() + data_url = self.media_store.to_data_url(ref) if not data_url: + continue + if kind == "image": + parts.append({"type": "image_url", "image_url": {"url": data_url}}) + media_payload_attached = True + continue + if kind == "video": + payload: Dict[str, Any] = {"type": "video_url", "video_url": {"url": data_url}} + fps_value = ref.get("fps") + if supports_video_fps and fps_value is None: + fps_value = qwen_video_fps + if fps_value is not None: + payload["fps"] = fps_value + parts.append(payload) + media_payload_attached = True + continue + if kind == "audio": + mime = str(ref.get("mime_type") or "audio/*") + size = ref.get("size") + label = f"[已附加音频 {mime}" + if isinstance(size, int): + label += f", {size} bytes" + label += "]" + extra_notes.append(label) + media_payload_attached = True + + # 若已通过 media_refs 成功附加媒体，避免再次从路径重复注入 + should_use_legacy_paths = not media_payload_attached + + if should_use_legacy_paths: + for item in images: + try: + if isinstance(item, dict): + path = str(item.get("path") or "").strip() + else: + path = str(item or "").strip() + if not path: + continue + abs_path = self._resolve_media_input_path(path) + if not abs_path or not abs_path.exists() or not abs_path.is_file(): + continue + mime, _ = mimetypes.guess_type(abs_path.name) + if mime and mime.startswith("video/"): + extra_videos.append(item) + continue + if mime and not mime.startswith("image/"): + continue + data_url = self._compress_image_if_needed(abs_path) + if not data_url: + if not mime: + mime = "image/png" + data = abs_path.read_bytes() + b64 = base64.b64encode(data).decode("utf-8") + data_url = f"data:{mime};base64,{b64}" + parts.append({"type": "image_url", "image_url": {"url": data_url}}) + except Exception: + continue + + for item in [*videos, *extra_videos]: + try: + if isinstance(item, dict): + path = item.get("path") or "" + else: + path = item + path_text = str(path or "").strip() + if not path_text: + continue + abs_path = self._resolve_media_input_path(path_text) + if not abs_path or not abs_path.exists() or not abs_path.is_file(): + continue + if abs_path.stat().st_size > 50 * 1024 * 1024: + continue + mime, _ = mimetypes.guess_type(abs_path.name) if not mime: - mime = "image/png" + mime = "video/mp4" data = abs_path.read_bytes() b64 = base64.b64encode(data).decode("utf-8") data_url = f"data:{mime};base64,{b64}" - parts.append({"type": "image_url", "image_url": {"url": data_url}}) - except Exception: - continue - for item in [*videos, *extra_videos]: - try: - if isinstance(item, dict): - path = item.get("path") or "" - else: - path = item - if not path: + payload = {"type": "video_url", "video_url": {"url": data_url}} + fps_value = item.get("fps") if isinstance(item, dict) else None + if supports_video_fps and fps_value is None: + fps_value = qwen_video_fps + if fps_value is not None: + payload["fps"] = fps_value + parts.append(payload) + except Exception: continue - abs_path = Path(self.project_path) / path - if not abs_path.exists() or not abs_path.is_file(): - continue - if abs_path.stat().st_size > 50 * 1024 * 1024: - continue - mime, _ = mimetypes.guess_type(abs_path.name) - if not mime: - mime = "video/mp4" - data = abs_path.read_bytes() - b64 = base64.b64encode(data).decode("utf-8") - data_url = f"data:{mime};base64,{b64}" - payload: Dict[str, Any] = {"type": "video_url", "video_url": {"url": data_url}} - if supports_video_fps: - payload["fps"] = qwen_video_fps - parts.append(payload) - except Exception: - continue - return parts if parts else text + + text_segments = [] + if text: + text_segments.append(text) + if extra_notes: + text_segments.extend(extra_notes) + merged_text = "\n".join(segment for segment in text_segments if segment) + if merged_text: + parts.insert(0, {"type": "text", "text": merged_text}) + + # 只有纯文本时保持字符串，避免无意义地返回 list[{"type":"text"}] + if not parts: + return text + if len(parts) == 1 and parts[0].get("type") == "text": + return parts[0].get("text", "") + return parts def _build_workspace_system_message(self, context: Dict) -> Optional[str]: """构建独立的工作区系统消息，根据运行模式动态展示环境与资源信息。""" @@ -1971,5 +2279,3 @@ class ContextManager: ) return content - - diff --git a/utils/media_store.py b/utils/media_store.py new file mode 100644 index 0000000..03dbe10 --- /dev/null +++ b/utils/media_store.py @@ -0,0 +1,369 @@ +"""对话媒体存储：将图片/视频/音频二进制与对话 JSON 解耦。""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import mimetypes +import os +import tempfile +import threading +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + + +DEFAULT_MIME_BY_KIND: Dict[str, str] = { + "image": "image/png", + "video": "video/mp4", + "audio": "audio/wav", +} + + +def _normalize_kind(value: Optional[str], mime_type: Optional[str] = None) -> str: + kind = str(value or "").strip().lower() + if kind in {"image", "video", "audio"}: + return kind + mime = str(mime_type or "").strip().lower() + if mime.startswith("image/"): + return "image" + if mime.startswith("video/"): + return "video" + if mime.startswith("audio/"): + return "audio" + return "binary" + + +def _normalize_mime(value: Optional[str], kind: str, fallback_path: Optional[str] = None) -> str: + mime = str(value or "").strip().lower() + if mime: + return mime + if fallback_path: + guessed, _ = mimetypes.guess_type(str(fallback_path)) + if guessed: + return guessed + return DEFAULT_MIME_BY_KIND.get(kind, "application/octet-stream") + + +def _guess_extension(mime_type: str, kind: str) -> str: + ext = mimetypes.guess_extension(mime_type or "") + if ext: + return ext + if kind == "image": + return ".png" + if kind == "video": + return ".mp4" + if kind == "audio": + return ".wav" + return ".bin" + + +def _decode_base64_payload(payload: str) -> bytes: + data = "".join(str(payload or "").strip().split()) + if not data: + raise ValueError("base64 数据为空") + missing_padding = len(data) % 4 + if missing_padding: + data += "=" * (4 - missing_padding) + return base64.b64decode(data.encode("utf-8"), validate=False) + + +def _parse_data_url(data_url: str) -> Tuple[str, str]: + raw = str(data_url or "").strip() + if not raw.startswith("data:") or "," not in raw: + raise ValueError("非法 data URL") + header, payload = raw.split(",", 1) + mime = "application/octet-stream" + if ";" in header: + mime = header[5:].split(";", 1)[0] or mime + elif len(header) > 5: + mime = header[5:] + if ";base64" not in header.lower(): + raise ValueError("仅支持 base64 data URL") + return mime, payload + + +class MediaStore: + """媒体存储中心（data/conversations/media_store）。""" + + INDEX_VERSION = 1 + + def __init__(self, data_dir: Path): + self.data_dir = Path(data_dir).expanduser().resolve() + self.conversations_dir = self.data_dir / "conversations" + self.root_dir = self.conversations_dir / "media_store" + self.blobs_dir = self.root_dir / "blobs" + self.index_file = self.root_dir / "index.json" + self._lock = threading.RLock() + self._ensure_dirs() + + def _ensure_dirs(self) -> None: + self.conversations_dir.mkdir(parents=True, exist_ok=True) + self.root_dir.mkdir(parents=True, exist_ok=True) + self.blobs_dir.mkdir(parents=True, exist_ok=True) + if not self.index_file.exists(): + self._save_index( + { + "version": self.INDEX_VERSION, + "media": {}, + "message_map": {}, + "updated_at": datetime.now().isoformat(), + } + ) + + def _atomic_write_json(self, target_file: Path, payload: Dict[str, Any]) -> None: + target_file.parent.mkdir(parents=True, exist_ok=True) + temp_path: Optional[Path] = None + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=str(target_file.parent), + prefix=f".{target_file.name}.", + suffix=".tmp", + delete=False, + ) as fh: + temp_path = Path(fh.name) + json.dump(payload, fh, ensure_ascii=False, indent=2) + fh.flush() + os.fsync(fh.fileno()) + os.replace(str(temp_path), str(target_file)) + finally: + if temp_path and temp_path.exists(): + try: + temp_path.unlink() + except Exception: + pass + + def _load_index(self) -> Dict[str, Any]: + try: + if not self.index_file.exists(): + return { + "version": self.INDEX_VERSION, + "media": {}, + "message_map": {}, + "updated_at": datetime.now().isoformat(), + } + data = json.loads(self.index_file.read_text(encoding="utf-8") or "{}") + if not isinstance(data, dict): + raise ValueError("index.json 不是对象") + data.setdefault("version", self.INDEX_VERSION) + data.setdefault("media", {}) + data.setdefault("message_map", {}) + data.setdefault("updated_at", datetime.now().isoformat()) + if not isinstance(data.get("media"), dict): + data["media"] = {} + if not isinstance(data.get("message_map"), dict): + data["message_map"] = {} + return data + except Exception: + return { + "version": self.INDEX_VERSION, + "media": {}, + "message_map": {}, + "updated_at": datetime.now().isoformat(), + } + + def _save_index(self, payload: Dict[str, Any]) -> None: + payload["updated_at"] = datetime.now().isoformat() + self._atomic_write_json(self.index_file, payload) + + @staticmethod + def _build_media_id(sha256_hex: str) -> str: + return f"sha256:{sha256_hex}" + + def _blob_path_for(self, sha256_hex: str, ext: str) -> Tuple[Path, str]: + rel = Path("blobs") / sha256_hex[:2] / f"{sha256_hex}{ext}" + return self.root_dir / rel, rel.as_posix() + + def _write_blob_if_missing(self, blob_path: Path, payload: bytes) -> None: + if blob_path.exists(): + return + blob_path.parent.mkdir(parents=True, exist_ok=True) + temp_path: Optional[Path] = None + try: + with tempfile.NamedTemporaryFile( + mode="wb", + dir=str(blob_path.parent), + prefix=f".{blob_path.name}.", + suffix=".tmp", + delete=False, + ) as fh: + temp_path = Path(fh.name) + fh.write(payload) + fh.flush() + os.fsync(fh.fileno()) + os.replace(str(temp_path), str(blob_path)) + finally: + if temp_path and temp_path.exists(): + try: + temp_path.unlink() + except Exception: + pass + + def store_bytes( + self, + payload: bytes, + *, + kind: Optional[str] = None, + mime_type: Optional[str] = None, + source_path: Optional[str] = None, + ) -> Dict[str, Any]: + if not isinstance(payload, (bytes, bytearray)) or len(payload) == 0: + raise ValueError("媒体二进制为空") + sha256_hex = hashlib.sha256(payload).hexdigest() + media_id = self._build_media_id(sha256_hex) + normalized_kind = _normalize_kind(kind, mime_type) + normalized_mime = _normalize_mime(mime_type, normalized_kind, fallback_path=source_path) + ext = _guess_extension(normalized_mime, normalized_kind) + blob_path, blob_rel = self._blob_path_for(sha256_hex, ext) + + with self._lock: + self._write_blob_if_missing(blob_path, bytes(payload)) + index = self._load_index() + media = index.setdefault("media", {}) + entry = media.get(media_id) if isinstance(media.get(media_id), dict) else {} + entry.update( + { + "media_id": media_id, + "sha256": sha256_hex, + "kind": normalized_kind, + "mime_type": normalized_mime, + "size": len(payload), + "blob_rel_path": blob_rel, + "created_at": entry.get("created_at") or datetime.now().isoformat(), + "updated_at": datetime.now().isoformat(), + } + ) + media[media_id] = entry + self._save_index(index) + return { + "media_id": media_id, + "kind": normalized_kind, + "mime_type": normalized_mime, + "size": len(payload), + "sha256": sha256_hex, + "store_path": f"media_store/{blob_rel}", + "source_path": source_path, + } + + def store_file( + self, + file_path: Path, + *, + kind: Optional[str] = None, + mime_type: Optional[str] = None, + source_path: Optional[str] = None, + ) -> Dict[str, Any]: + path = Path(file_path).expanduser().resolve() + payload = path.read_bytes() + return self.store_bytes( + payload, + kind=kind, + mime_type=mime_type, + source_path=source_path or str(file_path), + ) + + def store_base64_data( + self, + data_base64: str, + *, + kind: Optional[str] = None, + mime_type: Optional[str] = None, + source_path: Optional[str] = None, + ) -> Dict[str, Any]: + payload = _decode_base64_payload(data_base64) + return self.store_bytes( + payload, + kind=kind, + mime_type=mime_type, + source_path=source_path, + ) + + def store_data_url( + self, + data_url: str, + *, + kind: Optional[str] = None, + source_path: Optional[str] = None, + ) -> Dict[str, Any]: + mime_type, base64_payload = _parse_data_url(data_url) + return self.store_base64_data( + base64_payload, + kind=kind, + mime_type=mime_type, + source_path=source_path, + ) + + def _resolve_blob_path(self, media_ref: Dict[str, Any]) -> Optional[Tuple[Path, Dict[str, Any]]]: + media_id = str(media_ref.get("media_id") or "").strip() + with self._lock: + index = self._load_index() + media = index.get("media") if isinstance(index.get("media"), dict) else {} + entry = media.get(media_id) if media_id else None + if isinstance(entry, dict): + rel = str(entry.get("blob_rel_path") or "").strip() + if rel: + path = self.root_dir / rel + if path.exists() and path.is_file(): + return path, entry + store_path = str(media_ref.get("store_path") or "").strip() + if store_path.startswith("media_store/"): + candidate = self.conversations_dir / store_path + elif store_path: + candidate = self.root_dir / store_path + else: + candidate = None + if candidate and candidate.exists() and candidate.is_file(): + if not isinstance(entry, dict): + entry = {} + return candidate, entry + return None + + def load_bytes(self, media_ref: Dict[str, Any]) -> Optional[bytes]: + resolved = self._resolve_blob_path(media_ref or {}) + if not resolved: + return None + blob_path, _entry = resolved + try: + return blob_path.read_bytes() + except Exception: + return None + + def to_data_url(self, media_ref: Dict[str, Any]) -> Optional[str]: + payload = self.load_bytes(media_ref) + if payload is None: + return None + mime = _normalize_mime( + media_ref.get("mime_type"), + _normalize_kind(media_ref.get("kind"), media_ref.get("mime_type")), + ) + b64 = base64.b64encode(payload).decode("utf-8") + return f"data:{mime};base64,{b64}" + + def link_message_media( + self, + conversation_id: Optional[str], + message_id: Optional[str], + media_refs: List[Dict[str, Any]], + ) -> None: + conv = str(conversation_id or "").strip() + msg = str(message_id or "").strip() + if not conv or not msg or not media_refs: + return + media_ids = [str((item or {}).get("media_id") or "").strip() for item in media_refs] + media_ids = [item for item in media_ids if item] + if not media_ids: + return + with self._lock: + index = self._load_index() + message_map = index.setdefault("message_map", {}) + conv_map = message_map.get(conv) if isinstance(message_map.get(conv), dict) else {} + conv_map[msg] = { + "media_ids": media_ids, + "updated_at": datetime.now().isoformat(), + } + message_map[conv] = conv_map + self._save_index(index) + diff --git a/utils/tool_result_formatter.py b/utils/tool_result_formatter.py index a2c6a2f..9980991 100644 --- a/utils/tool_result_formatter.py +++ b/utils/tool_result_formatter.py @@ -2,9 +2,191 @@ from __future__ import annotations +import json from typing import Any, Dict, List, Optional, Tuple +def _looks_like_mcp_payload(function_name: str, result_data: Any) -> bool: + if isinstance(function_name, str) and function_name.startswith("mcp__"): + return True + if not isinstance(result_data, dict): + return False + if result_data.get("tool_alias") or result_data.get("server_id"): + return True + content = result_data.get("content") + if isinstance(content, list) and any(isinstance(item, dict) and item.get("type") for item in content): + return True + return False + + +def _format_mcp_resource_reference(item: Dict[str, Any]) -> str: + uri = str(item.get("uri") or item.get("url") or "").strip() + title = str(item.get("title") or item.get("name") or "").strip() + if uri and title: + return f"[MCP 资源] {title} -> {uri}" + if uri: + return f"[MCP 资源] {uri}" + if title: + return f"[MCP 资源] {title}" + return "" + + +def _sanitize_mcp_content_item(item: Dict[str, Any]) -> Dict[str, Any]: + sanitized = dict(item) + ctype = str(sanitized.get("type") or "").strip().lower() + if ctype in {"image", "audio"}: + data = str(sanitized.pop("data", "") or "") + if data: + sanitized["data_omitted"] = True + sanitized["data_length"] = len(data) + if ctype == "resource": + resource = sanitized.get("resource") + if isinstance(resource, dict) and "blob" in resource: + blob_data = str(resource.get("blob") or "") + resource_copy = dict(resource) + resource_copy.pop("blob", None) + if blob_data: + resource_copy["blob_omitted"] = True + resource_copy["blob_length"] = len(blob_data) + sanitized["resource"] = resource_copy + return sanitized + + +def extract_mcp_content_for_context(result_data: Dict[str, Any]) -> Dict[str, Any]: + """提取 MCP content，返回可读文本 + 可落盘媒体列表 + 去二进制后的 payload。""" + if not isinstance(result_data, dict): + return { + "text": "", + "media_items": [], + "sanitized_payload": result_data, + } + + content_items = result_data.get("content") + if not isinstance(content_items, list): + raw_result = result_data.get("raw_result") + if isinstance(raw_result, dict) and isinstance(raw_result.get("content"), list): + content_items = raw_result.get("content") + else: + content_items = [] + + text_lines: List[str] = [] + media_lines: List[str] = [] + media_items: List[Dict[str, Any]] = [] + sanitized_content: List[Dict[str, Any]] = [] + + for index, item in enumerate(content_items, start=1): + if not isinstance(item, dict): + continue + ctype = str(item.get("type") or "").strip().lower() + if ctype == "text": + text_value = str(item.get("text") or "").strip() + if text_value: + text_lines.append(text_value) + sanitized_content.append(_sanitize_mcp_content_item(item)) + continue + + if ctype in {"image", "audio"}: + kind = "image" if ctype == "image" else "audio" + mime_type = str(item.get("mimeType") or item.get("mime_type") or "").strip().lower() + if not mime_type: + mime_type = "image/png" if kind == "image" else "audio/wav" + data_base64 = str(item.get("data") or "") + if data_base64: + media_items.append( + { + "index": index, + "item_type": ctype, + "kind": kind, + "mime_type": mime_type, + "data_base64": data_base64, + "label": item.get("label") or f"{kind}_{index}", + "name": item.get("name"), + "title": item.get("title"), + "uri": item.get("uri"), + "url": item.get("url"), + } + ) + media_lines.append( + f"[MCP {kind} #{index}] {mime_type}（{len(data_base64)} base64 chars）" + ) + else: + media_lines.append(f"[MCP {kind} #{index}] 未包含数据") + sanitized_content.append(_sanitize_mcp_content_item(item)) + continue + + if ctype in {"resource_link", "resourcelink"}: + line = _format_mcp_resource_reference(item) + if line: + text_lines.append(line) + sanitized_content.append(_sanitize_mcp_content_item(item)) + continue + + if ctype == "resource": + resource = item.get("resource") + resource_line = _format_mcp_resource_reference(item) + if isinstance(resource, dict): + resource_line = _format_mcp_resource_reference(resource) or resource_line + resource_text = str(resource.get("text") or "").strip() + if resource_text: + text_lines.append(resource_text) + elif resource_line: + text_lines.append(resource_line) + elif resource_line: + text_lines.append(resource_line) + sanitized_content.append(_sanitize_mcp_content_item(item)) + continue + + fallback_line = _format_mcp_resource_reference(item) + if fallback_line: + text_lines.append(fallback_line) + else: + text_lines.append(f"[MCP content:{ctype or 'unknown'}]") + sanitized_content.append(_sanitize_mcp_content_item(item)) + + message = str(result_data.get("message") or "").strip() + if message and message != "MCP 工具调用完成": + if message not in text_lines: + text_lines.insert(0, message) + + if media_lines: + text_lines.extend(media_lines) + + if not text_lines: + structured = result_data.get("structured_content") + if structured is None: + structured = result_data.get("structuredContent") + if structured is not None: + try: + text_lines.append(json.dumps(structured, ensure_ascii=False)) + except Exception: + text_lines.append(str(structured)) + + if not text_lines and message: + text_lines.append(message) + if not text_lines: + text_lines.append("MCP 工具调用完成") + + sanitized_payload: Dict[str, Any] = dict(result_data) + if isinstance(result_data.get("content"), list): + sanitized_payload["content"] = sanitized_content + raw_result = result_data.get("raw_result") + if isinstance(raw_result, dict): + raw_sanitized = dict(raw_result) + raw_content = raw_result.get("content") + if isinstance(raw_content, list): + raw_sanitized["content"] = [ + _sanitize_mcp_content_item(item) if isinstance(item, dict) else item + for item in raw_content + ] + sanitized_payload["raw_result"] = raw_sanitized + + return { + "text": "\n".join(line for line in text_lines if line), + "media_items": media_items, + "sanitized_payload": sanitized_payload, + } + + def format_read_file_result(result_data: Dict[str, Any]) -> str: """格式化 read_file 工具的输出，兼容读取/搜索/抽取模式。""" if not isinstance(result_data, dict): @@ -83,6 +265,12 @@ def format_tool_result_for_context(function_name: str, result_data: Any, raw_tex parts.append(str(output)) return "\n".join(parts).strip() or raw_text + if _looks_like_mcp_payload(function_name, result_data): + parsed = extract_mcp_content_for_context(result_data if isinstance(result_data, dict) else {}) + text = str(parsed.get("text") or "").strip() + if text: + return text + if function_name == "read_file" and isinstance(result_data, dict): return format_read_file_result(result_data)