"""将工具执行结果转换为对话上下文可用的纯文本摘要。""" from __future__ import annotations import json from typing import Any, Dict, List, Optional, Tuple from utils.tool_result_formatter.mcp import ( _looks_like_mcp_payload, extract_mcp_content_for_context, ) from utils.tool_result_formatter.file import ( format_read_file_result, _format_write_file_diff, _format_create_file, _format_write_file, _format_edit_file, _format_delete_file, _format_rename_file, _format_create_folder, ) from utils.tool_result_formatter.terminal import ( _format_terminal_snapshot, _format_terminal_session, _format_terminal_input, _format_sleep, _format_run_command, _format_command_result, ) from utils.tool_result_formatter.agent_context import ( _format_conversation_search, _format_conversation_review, _format_todo_create, _format_todo_update_task, _format_update_memory, _format_create_sub_agent, _format_close_sub_agent, _format_get_sub_agent_status, ) from utils.tool_result_formatter.web_media import ( _format_extract_webpage, _format_vlm_analyze, _format_ocr_image, _format_trigger_easter_egg, _format_create_skill, _format_manage_personalization, ) from utils.tool_result_formatter.common import _format_failure def format_tool_result_for_context(function_name: str, result_data: Any, raw_text: str = "") -> str: """根据工具名称输出纯文本摘要,必要时附加关键信息。""" # 自定义工具统一格式 if isinstance(result_data, dict) and result_data.get("custom_tool"): tool_id = result_data.get("tool_id") or function_name success = result_data.get("success") status = "成功" if success else "失败" message = result_data.get("message") or result_data.get("summary") or "" output = result_data.get("output") or "" parts = [f"[自定义工具] {tool_id} 执行{status}"] if message: parts.append(str(message)) if output: 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 return "" if function_name in {"read_file", "read_skill", "recall_project_memory"} and isinstance(result_data, dict): return format_read_file_result(result_data) if function_name == "write_file_diff" and isinstance(result_data, dict): return _format_write_file_diff(result_data, raw_text) if function_name == "conversation_search" and isinstance(result_data, dict): return _format_conversation_search(result_data) if function_name == "conversation_review" and isinstance(result_data, dict): return _format_conversation_review(result_data) if not isinstance(result_data, dict): return raw_text handler = TOOL_FORMATTERS.get(function_name) if handler: return handler(result_data) summary = result_data.get("summary") or result_data.get("message") error_msg = result_data.get("error") parts: List[str] = [] if summary: parts.append(str(summary)) if error_msg: parts.append(f"⚠️ 错误: {error_msg}") return "\n".join(parts) if parts else raw_text TOOL_FORMATTERS = { "create_file": _format_create_file, "write_file": _format_write_file, "edit_file": _format_edit_file, "delete_file": _format_delete_file, "rename_file": _format_rename_file, "create_folder": _format_create_folder, "terminal_snapshot": _format_terminal_snapshot, "terminal_session": _format_terminal_session, "terminal_input": _format_terminal_input, "sleep": _format_sleep, "run_command": _format_run_command, "extract_webpage": _format_extract_webpage, "vlm_analyze": _format_vlm_analyze, "ocr_image": _format_ocr_image, "trigger_easter_egg": _format_trigger_easter_egg, "todo_create": _format_todo_create, "todo_update_task": _format_todo_update_task, "update_memory": _format_update_memory, "create_skill": _format_create_skill, "manage_personalization": _format_manage_personalization, "create_sub_agent": _format_create_sub_agent, "close_sub_agent": _format_close_sub_agent, "get_sub_agent_status": _format_get_sub_agent_status, }