""" agents/main_agent.py - 主 Agent ================================= 总指挥 Agent，负责任务规划、Agent 调度、结果汇总。 """ from __future__ import annotations import asyncio import json import os import re from pathlib import Path from typing import Any, Callable, Dict, List, Optional from core.logger import get_logger from core.llm import LLMClient, LLMResponse, Message from agents.base import BaseAgent, AgentContext from core.utils import generate_id, timestamp, truncate_str from core.context_builder import ContextBuilder from core.tool_dispatcher import ToolDispatcher logger = get_logger("myagent.agent.main") class MainAgent(BaseAgent): """ 主 Agent - 总指挥。 职责: - 接收用户消息 - 任务分析与规划 - 调度子 Agent (ToolAgent / MemoryAgent) - 整合结果并回复 - 多轮迭代(计划-执行-反思循环) - 确保不进入死循环 """ name = "main_agent" description = "AI助手主控Agent，负责理解用户意图、规划任务、调度执行" # ========================================================================= # 系统提示词 — [v1.38] 原生 tool_calling 模式（不再要求 XML 输出格式） # ========================================================================= SYSTEM_PROMPT = """你是一个智能AI助手，能够通过工具调用来完成各种任务。 ## 行为准则 1. 直接回应用户的问题和需求，使用 Markdown 格式使内容直观清晰（可包含超链接、表格等） 2. 需要执行操作时，调用相应工具；可以一次调用多个工具以提高效率 3. 复杂任务（超过3步）应先使用 task_plan 工具创建任务计划，然后逐步执行并更新状态 4. 重要信息（用户偏好、关键结论、个人信息等）应主动保存到记忆库 5. 值得长期保存的专业知识、经验法则应保存到知识库 6. 执行过程中简洁展示进展，任务完成后给出详细总结 7. 内容不要与上次回复重复 8. 文档生成工具（docx/xlsx/ppt/pdf-create）执行后会自动发送文件给用户，不需要再调用 send-file ## 工具使用要点 - **command**: 所有系统操作都通过命令行完成，包括文件读写、搜索、OCR、网络请求等 - **多个命令可用 && 连接一次执行**，减少回调次数：`myagent-ai search xxx && myagent-ai read-url https://...` - **web_control**: 浏览器自动化操作（打开网页、点击、填写、截图等） - **task_plan**: 任务规划管理（create/update/get/clear） - **save_memory**: 保存重要信息到记忆库（global=跨会话, session=仅当前会话） - **recall_memory**: 即时搜索记忆库获取相关信息 - **save_knowledge**: 保存专业知识到知识库供未来复用 - **search_knowledge**: 搜索知识库获取专业知识 - **file_send**: 向用户发送文件（文档生成工具自动发送，此工具用于发送其他文件） - **playaudio/playvideo**: 在聊天中嵌入音视频播放器 ## CLI 命令参考（通过 command 工具调用 myagent-ai） 【感知】ocr, analyze-image, transcribe 【搜索】search, read-url, fetch-url 【文件】read, write, ls, rm, grep, mv, send-file 【文档】docx-create/read, xlsx-create/read/edit, ppt-create/read, pdf-create/read 【系统】sysinfo, ps, env, pathinfo 【浏览器】browser-open/screenshot/click/fill/eval/navigate/close 【GUI】screenshot, mouse-click/drag, type-text, hotkey, window-list/focus, screen-element 【记忆】memory --keyword 【Agent间通信】chat --agent <路径> -m "消息" -f "文件" 【压缩】zip/unzip 专业技能指南：系统内置了丰富的专业技能（PDF/DOCX/XLSX/PPT生成、图表绘制等），通过 search_knowledge 搜索获取。""" def __init__(self, tool_agent=None, memory_agent=None, skill_registry=None, **kwargs): super().__init__(**kwargs) self.tool_agent = tool_agent self.memory_agent = memory_agent self.skill_registry = skill_registry # 缓存数字 agent_id（在 memory 初始化后设置） self._iteration_count = 0 self._current_task_id: str = "" self._registered_task: bool = False # [v1.22.0] 统一工具分发器 self.dispatcher: Optional[ToolDispatcher] = None # Context Builder (结构化上下文构建) self.context_builder: Optional[ContextBuilder] = None # 执行事件追踪（用于前端展示命令执行过程） self._execution_events: List[Dict] = [] self._exec_event_counter: int = 0 # 活跃会话上下文追踪（用于消息注入） self.active_contexts: Dict[str, AgentContext] = {} # [v1.23.37] 会话取消标志（由 api_server.handle_chat_stop 设置） self._cancelled_sessions: set = set() # [v1.25.7] 初始化数字 agent_id if self.memory: try: self.agent_id = self.memory.get_agent_id(self.name) except Exception: self.agent_id = 1 # [v1.34.1] TaskPlan 技能实例（用于任务规划） self._task_plan_skill = None if skill_registry: self._task_plan_skill = skill_registry.get("task_plan") def init_context_builder(self, memory_manager=None, skill_registry=None, knowledge_base_dir=None, context_window=None, max_message_chars=None): """初始化 Context Builder（在系统启动后调用，注入依赖）""" if context_window is None and self.llm: context_window = getattr(self.llm, 'context_window', 200000) if context_window is None: context_window = 200000 if max_message_chars is None: max_message_chars = 10000 self.context_builder = ContextBuilder( memory_manager=memory_manager, skill_registry=skill_registry, knowledge_base_dir=knowledge_base_dir, context_window=context_window, max_message_chars=max_message_chars, ) logger.info(f"Context Builder 已初始化 (context_window={context_window}, max_message_chars={max_message_chars})" + (f" (知识库: {knowledge_base_dir})" if knowledge_base_dir else "")) def _get_tools(self) -> List[Dict]: """ [v1.38] 获取所有工具定义（OpenAI function calling 格式）。 将内置工具、记忆/知识库工具、技能注册表中的工具统一为标准格式。 """ tools = [] # ── 1. 内置平台工具 ── tools.append({ "type": "function", "function": { "name": "command", "description": "执行shell命令行。所有系统操作都通过此工具完成，包括文件读写、搜索、OCR、网络请求、文档生成等。使用 myagent-ai CLI 调用子命令。多个命令用 && 连接。", "parameters": { "type": "object", "properties": { "command": { "type": "string", "description": "要执行的命令行。如: myagent-ai search 关键词, myagent-ai docx-create -c '{...}', python3 script.py, ls -la" }, }, "required": ["command"], }, } }) tools.append({ "type": "function", "function": { "name": "web_control", "description": "浏览器自动化操作：打开网页、点击元素、填写表单、截图、执行JS等", "parameters": { "type": "object", "properties": { "action": { "type": "string", "enum": ["open", "navigate", "get_content", "click", "fill", "scroll", "evaluate", "screenshot", "wait", "set_cookies", "get_cookies", "close"], "description": "操作类型" }, "url": {"type": "string", "description": "URL（open/navigate 时必填）"}, "selector": {"type": "string", "description": "CSS选择器（click/fill 时必填）"}, "value": {"type": "string", "description": "填写内容（fill 时必填）"}, "what": {"type": "string", "description": "获取内容类型: text/html/url/title/links/images/forms/inputs"}, "script": {"type": "string", "description": "JS代码（evaluate 时必填）"}, "direction": {"type": "string", "enum": ["up", "down", "top", "bottom"], "description": "滚动方向"}, "distance": {"type": "integer", "description": "滚动距离（像素）"}, "time": {"type": "integer", "description": "等待时间（毫秒）"}, "timeout": {"type": "integer", "description": "等待超时（秒）"}, "cookies": {"type": "array", "description": "Cookie列表（set_cookies 时必填）"}, "session_id": {"type": "string", "description": "浏览器会话ID"}, }, "required": ["action"], }, } }) tools.append({ "type": "function", "function": { "name": "file_send", "description": "向用户发送文件，文件会以卡片形式显示在聊天中。注意：docx/xlsx/ppt/pdf-create 等命令执行后会自动发送文件，不需要再调用此工具。", "parameters": { "type": "object", "properties": { "file_path": {"type": "string", "description": "文件的绝对路径"}, "description": {"type": "string", "description": "文件描述（可选）"}, }, "required": ["file_path"], }, } }) tools.append({ "type": "function", "function": { "name": "playaudio", "description": "在聊天中嵌入音频播放器，支持 YouTube Music、网易云音乐、QQ音乐、B站等", "parameters": { "type": "object", "properties": { "url": {"type": "string", "description": "音频URL（在线链接）"}, "file_path": {"type": "string", "description": "本地音频文件路径"}, "title": {"type": "string", "description": "播放器标题（可选）"}, }, }, } }) tools.append({ "type": "function", "function": { "name": "playvideo", "description": "在聊天中嵌入视频播放器，支持 YouTube、B站、抖音等", "parameters": { "type": "object", "properties": { "url": {"type": "string", "description": "视频URL（在线链接）"}, "file_path": {"type": "string", "description": "本地视频文件路径"}, "title": {"type": "string", "description": "播放器标题（可选）"}, }, }, } }) # ── 2. 任务规划工具 ── tools.append({ "type": "function", "function": { "name": "task_plan", "description": "任务规划管理。复杂任务（超过3步）应先创建计划，然后逐步执行并更新状态。", "parameters": { "type": "object", "properties": { "action": { "type": "string", "enum": ["create", "update", "get", "clear"], "description": "操作类型: create=创建计划, update=更新状态, get=查询计划, clear=清空计划" }, "plan": {"type": "string", "description": "Markdown格式的任务列表（create时必填，如: - [ ] 步骤1\\n- [ ] 步骤2）"}, "task_index": {"type": "integer", "description": "任务索引（update时必填）"}, "completed": {"type": "boolean", "description": "是否完成（update时必填）"}, }, "required": ["action"], }, } }) # ── 3. 记忆与知识库工具 ── tools.append({ "type": "function", "function": { "name": "save_memory", "description": "保存重要信息到记忆库。对话不会自动保存记录，必须主动保存重要信息才能为后续多轮对话提供持续记忆基础。global类型跨会话持久保存，session类型仅当前会话。", "parameters": { "type": "object", "properties": { "content": {"type": "string", "description": "要保存的记忆内容（用户偏好、重要结论、错误经验、个人信息、对话要点等）"}, "type": { "type": "string", "enum": ["global", "session"], "description": "global=跨会话全局记忆（用户个人信息和偏好）, session=仅当前会话（临时信息）。默认session" }, }, "required": ["content"], }, } }) tools.append({ "type": "function", "function": { "name": "recall_memory", "description": "即时搜索记忆库，获取与关键词相关的历史记忆。当你需要回顾之前的信息或用户的偏好时使用。", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词或描述（可包含时间参考，如'2025年1月的项目'）"}, "limit": {"type": "integer", "description": "返回结果数量上限，默认5"}, }, "required": ["query"], }, } }) tools.append({ "type": "function", "function": { "name": "save_knowledge", "description": "保存专业知识到知识库供未来复用。如技术要点、经验法则、事实知识等。", "parameters": { "type": "object", "properties": { "content": {"type": "string", "description": "要保存的知识内容，简洁明确，每条知识一行"}, "category": {"type": "string", "description": "知识分类，如: general, programming, design, api 等。默认general"}, }, "required": ["content"], }, } }) tools.append({ "type": "function", "function": { "name": "search_knowledge", "description": "搜索知识库获取专业知识。当你需要专业技能指南（如PDF/DOCX生成格式、图表绘制方法等）或已有知识时使用。", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词或描述"}, "limit": {"type": "integer", "description": "返回结果数量上限，默认5"}, }, "required": ["query"], }, } }) tools.append({ "type": "function", "function": { "name": "update_conversation_title", "description": "更新当前对话的标题。在对话主题发生变化或新对话开始时调用。", "parameters": { "type": "object", "properties": { "title": {"type": "string", "description": "对话标题（6字以内）"}, }, "required": ["title"], }, } }) # ── 4. 技能注册表中的工具 ── if self.skill_registry: try: for skill_name in self.skill_registry.list_skills(): if skill_name in ("task_plan",): # 已在上面定义 continue skill = self.skill_registry.get(skill_name) if skill and hasattr(skill, 'get_schema'): try: schema = skill.get_schema() if schema and isinstance(schema, dict): tools.append(schema) except Exception as e: logger.debug(f"跳过技能 {skill_name} 的schema: {e}") except Exception as e: logger.warning(f"获取技能工具列表失败: {e}") return tools def _add_exec_event(self, event_type: str, data: Dict): """记录一个执行事件（供前端展示）""" import time as _time self._exec_event_counter += 1 event = { "id": self._exec_event_counter, "type": event_type, "timestamp": _time.time(), **data, } self._execution_events.append(event) logger.debug(f"[exec-event] {event_type}: {data.get('title', '')[:80]}") def get_execution_events(self) -> List[Dict]: """获取本次处理中的所有执行事件""" return self._execution_events def _build_conversation_history(self, context: AgentContext, task_id: str) -> List[Message]: """ [v1.26.9] 统一历史对话处理算法 - 确保流式和非流式输出使用相同的逻辑 从数据库或上下文构建标准化的对话历史列表。 这个算法被用于： 1. 流式输出时的上下文构建 2. 非流式输出时的上下文构建 3. 历史聊天记录的加载和展示 [FIX-失忆] 改进： - 增加 limit (100→200) 以加载更多历史对话 - 在 metadata 中保留 key 字段，用于过滤内部审计条目 - 之前 metadata 只包含 time，丢失了 key，导致无法区分 用户消息和内部条目（如 llm_output、reasoning 等） Args: context: Agent上下文 task_id: 任务ID（用于日志） Returns: 标准化的对话历史列表 """ conversation_history = list(context.conversation_history or []) # 从 DB 加载历史对话（如果 conversation_history 为空且 memory 可用） if self.memory and not conversation_history: try: db_history = self.memory.get_conversation( session_id=context.session_id, limit=200, ) if db_history: conversation_history = [ Message( role=entry.role, content=entry.content, metadata={ "time": (entry.created_at[:19] if entry.created_at else ""), "key": entry.key or "", # [FIX-失忆] 保留 key 用于过滤 } ) for entry in db_history ] logger.info(f"[{task_id}] 从 DB 加载了 {len(conversation_history)} 条历史对话") except Exception as e: logger.warning(f"[{task_id}] 加载历史对话失败: {e}") return conversation_history def clear_execution_events(self): """清空执行事件""" self._execution_events = [] self._exec_event_counter = 0 async def process(self, context: AgentContext, stream_callback=None) -> AgentContext: """ 主处理循环。 流程: 1. 加载相关记忆 2. 构建消息(系统提示 + 记忆上下文 + 对话历史 + 用户消息) 3. 调用 LLM 4. 解析响应(纯文本回复 / 工具调用) 5. 如需执行工具，调度 ToolAgent 6. 如需记忆操作，调度 MemoryAgent 7. 循环直到任务完成或达到最大迭代次数 """ task_id = context.task_id or generate_id("task") context.task_id = task_id self._iteration_count = 0 self._current_task_id = task_id # 记录活跃上下文 self.active_contexts[context.session_id] = context # 清空上一轮的执行事件 self.clear_execution_events() # 注册到配置广播器（用于热重载通知） if self.config_broadcaster: self.config_broadcaster.register(task_id, agent_name=self.name) self._registered_task = True logger.info(f"[{task_id}] 开始处理用户请求: {context.user_message[:100]}") # [v1.23.30] 读取外部注入的 Agent 专属提示词（群聊、多Agent 场景使用） # _try_model_chain_inner 通过此属性注入 group_context 等自定义提示词 _override_prompt = getattr(self, '_agent_override_prompt', None) # [v1.23.35] 读取外部注入的 Agent 名称和描述（群聊路径通过实例属性注入） # 确保群聊场景下 process_v2 也能拿到正确的 agent_name/agent_description， # 而非默认的"助手"/"通用AI助手" _injected_name = getattr(self, '_agent_override_name', None) _injected_desc = getattr(self, '_agent_override_description', None) try: return await self.process_v2( context, agent_name=_injected_name or self.name, agent_description=_injected_desc or self.description, agent_override_prompt=_override_prompt, agent_path=getattr(self, '_agent_override_path', None), stream_callback=stream_callback, ) finally: # 移除活跃上下文 self.active_contexts.pop(context.session_id, None) # 注销广播器 if self.config_broadcaster and self._registered_task: self.config_broadcaster.unregister(task_id) self._registered_task = False # ========================================================================= # 执行循环 — 结构化输出 + Context Builder + Output Parser # ========================================================================= async def _emit_v2_event(self, event_type: str, data: Dict, stream_callback: Optional[Callable] = None): """发送 V2 SSE 事件。如果 stream_callback 不存在则仅记录日志。""" event = {"type": event_type, **data} if stream_callback is not None: try: if asyncio.iscoroutinefunction(stream_callback): await stream_callback(event) else: stream_callback(event) except Exception as e: logger.debug(f"V2 SSE 事件发送失败 ({event_type}): {e}") async def _merge_duplicate_memory( self, old_memory, new_content: str, context: AgentContext, task_id: str, ) -> Optional[str]: """ 当发现新旧记忆高度相似时，调用 LLM API 让其判断最终记忆内容。 将旧记忆、新记忆、当前上下文发送给 LLM，由 LLM 决定: - 合并为一条更完整的记忆 - 保留新记忆（旧记忆已过时） - 保留旧记忆（新记忆无新增信息） Returns: 合并后的记忆内容，或 None（合并失败） """ if not self.llm: logger.warning(f"[{task_id}] 记忆合并: 无 LLM 客户端，跳过合并") return None from datetime import datetime from core.utils import get_config_tz old_time = old_memory.created_at or "未知时间" new_time = datetime.now(get_config_tz()).strftime("%Y-%m-%d %H:%M:%S") user_msg = context.user_message or "" merge_prompt = f"""你是一个记忆管理系统。现在系统检测到两条高度相似的记忆，请你判断如何合并它们。 ## 旧记忆（创建于 {old_time}） {old_memory.content} ## 新记忆（创建于 {new_time}） {new_content} ## 当前用户输入 {user_msg} ## 任务 请分析新旧记忆，输出一条最终的合并记忆。规则： 1. 如果新记忆包含了旧记忆的信息并有更新，则合并为更完整的表述 2. 如果新记忆只是旧记忆的重复或信息量更少，保留旧记忆中更完整的信息 3. 如果新记忆提供了全新的信息，以新记忆为主，补充旧记忆中的有效部分 4. 合并后的记忆应当简洁、准确、包含时间上下文 5. 直接输出合并后的记忆内容，不要输出任何解释或标记 请输出合并后的记忆：""" try: messages = [ Message(role="system", content="你是一个记忆管理系统，负责合并重复或相似的记忆条目。只输出合并后的记忆内容，不要输出任何额外说明。"), Message(role="user", content=merge_prompt), ] response = await self._call_llm(messages) if response.success and response.content: merged = response.content.strip() # 清理可能的引号包裹 if merged.startswith('"') and merged.endswith('"'): merged = merged[1:-1] if merged.startswith("'") and merged.endswith("'"): merged = merged[1:-1] logger.info( f"[{task_id}] 记忆合并成功: 旧({len(old_memory.content)}字) + " f"新({len(new_content)}字) → 合并({len(merged)}字)" ) return merged else: logger.warning(f"[{task_id}] 记忆合并 LLM 调用失败: {response.error}") return None except Exception as e: logger.warning(f"[{task_id}] 记忆合并异常: {e}") return None async def _save_knowledge_to_base( self, content: str, session_id: str, task_id: str, ) -> bool: """ 将 LLM 输出的 内容追加到知识库文件。 存储策略: - 知识按会话 (session_id) 分文件存储 - 文件路径: {knowledge_base_dir}/auto_knowledge/{session_id}.md - 每次追加时检查重复（TF-IDF 相似度 ≥ 0.9 视为重复，跳过） - 追加时带有时间戳标记 Returns: True 表示成功存储了新知识，False 表示跳过（重复）或失败 """ if not self.context_builder: logger.debug(f"[{task_id}] 知识库未配置，跳过 knowledge 存储") return False from datetime import datetime from pathlib import Path from core.utils import get_config_tz # 优先写入 Agent 专属知识库，其次回退到组织知识库 kb_dir = None if self.context_builder.agent_knowledge_dir: kb_dir = Path(self.context_builder.agent_knowledge_dir) elif self.context_builder.knowledge_base_dir: kb_dir = Path(self.context_builder.knowledge_base_dir) if not kb_dir: logger.debug(f"[{task_id}] 知识库目录未配置，跳过 knowledge 存储") return False auto_kb_dir = kb_dir / "auto_knowledge" auto_kb_dir.mkdir(parents=True, exist_ok=True) # 使用 session_id 作为文件名（取前8位避免过长） # 注意: session_id 可能包含 '/' (来自旧版 agent_path 格式)， # 必须替换为安全字符，避免创建意外的子目录 safe_session = session_id.replace("-", "").replace("/", "_")[:12] if session_id else "default" kb_file = auto_kb_dir / f"{safe_session}.md" now_str = datetime.now(get_config_tz()).strftime("%Y-%m-%d %H:%M") # 检查重复：与已有文件内容做相似度比较 existing_content = "" if kb_file.exists(): try: existing_content = kb_file.read_text(encoding="utf-8") except Exception: existing_content = "" if existing_content and content.strip(): # 简单去重：检查新知识是否已存在于文件中 # 使用逐行比对 + 关键词匹配 new_lines = [line.strip() for line in content.strip().split("\n") if line.strip()] existing_lines = [line.strip() for line in existing_content.split("\n") if line.strip() and not line.strip().startswith("- [")] dup_count = 0 for new_line in new_lines: # 精确匹配或高度相似（共现字符占比 > 85%） is_dup = False for ex_line in existing_lines: # 计算字符重叠率 set_new = set(new_line) set_ex = set(ex_line) if not set_new or not set_ex: continue overlap = len(set_new & set_ex) / max(len(set_new), len(set_ex)) if overlap >= 0.85 or new_line == ex_line: is_dup = True break if is_dup: dup_count += 1 if dup_count == len(new_lines): logger.info(f"[{task_id}] 知识全部重复，跳过存储 ({dup_count}/{len(new_lines)} 条)") return False elif dup_count > 0: # 过滤掉重复的行 filtered_lines = [] for new_line in new_lines: is_dup = False for ex_line in existing_lines: set_new = set(new_line) set_ex = set(ex_line) if not set_new or not set_ex: continue overlap = len(set_new & set_ex) / max(len(set_new), len(set_ex)) if overlap >= 0.85 or new_line == ex_line: is_dup = True break if not is_dup: filtered_lines.append(new_line) content = "\n".join(filtered_lines) logger.info(f"[{task_id}] 知识去重: {dup_count}/{len(new_lines)} 条重复，{len(filtered_lines)} 条新增") # 追加写入 try: with open(kb_file, "a", encoding="utf-8") as f: f.write(f"\n## {now_str}\n") f.write(content.strip() + "\n") logger.info( f"[{task_id}] 知识已存入知识库: {kb_file} " f"({len(content)} 字符, {len(content.strip().split(chr(10)))} 条)" ) return True except Exception as e: logger.warning(f"[{task_id}] 知识写入失败: {e}") return False async def process_v2( self, context: AgentContext, agent_name: str = "助手", agent_description: str = "通用AI助手", agent_override_prompt: Optional[str] = None, stream_callback: Optional[Callable] = None, stream_response=None, text_delta_callback=None, agent_path: Optional[str] = None, ) -> AgentContext: """ V2 主处理循环 — 使用原生 tool_calling。 核心流程: 1. 使用 ContextBuilder 构建 XML 2. 将 context 注入 SYSTEM_PROMPT，调用 LLM 3. LLM 通过原生 tool_calling 返回工具调用 4. 根据 tool_calls 依次执行工具 5. 任一工具超时 → 强制回调 LLM 6. 根据 callback 标志决定是否回调 LLM 7. 处理 remember/recall Args: context: Agent 上下文 agent_name: Agent 名称（用于 ContextBuilder） agent_description: Agent 描述 agent_override_prompt: 可选的 Agent 身份覆盖提示词 stream_callback: 可选的 SSE 事件回调 (callable 或 async callable) stream_response: 可选的流式响应对象（用于 LLM 流式输出） text_delta_callback: 可选的文本增量回调 agent_path: Agent 的数字 aid（用于独立工作目录） """ task_id = context.task_id or generate_id("task") context.task_id = task_id self._iteration_count = 0 self._current_task_id = task_id self.clear_execution_events() logger.info(f"[{task_id}] 执行循环启动: {context.user_message[:100]}") try: return await self._process_v2_inner( context, task_id, agent_name, agent_description, agent_override_prompt, stream_callback, stream_response, text_delta_callback, agent_path=agent_path, ) except Exception as e: logger.error(f"[{task_id}] V2 执行循环异常: {e}", exc_info=True) context.working_memory["final_response"] = f"执行异常: {str(e)}" await self._emit_v2_event("v2_reasoning", {"content": f"执行异常: {str(e)}"}, stream_callback) return context async def _process_v2_inner( self, context: AgentContext, task_id: str, agent_name: str, agent_description: str, agent_override_prompt: Optional[str], stream_callback: Optional[Callable] = None, stream_response=None, text_delta_callback=None, agent_path: Optional[str] = None, ) -> AgentContext: """[v1.38] V2 内部循环 — 原生 tool_calling 模式（不再使用 XML 输出格式） 核心流程： 1. 构建 system prompt + context + 对话历史 2. 调用 LLM（带 tools 参数） 3. 如果 LLM 返回 tool_calls → 执行工具 → 将结果喂回 LLM → 继续循环 4. 如果 LLM 只返回文本 → 这就是最终回复 → 结束循环 """ max_iter = self.config.agent.max_iterations current_task_plan = "" # 追踪已发送文件 _sent_files: List[Dict[str, Any]] = [] _effective_agent_id = context.metadata.get("agent_db_id", self.agent_id) # 构建对话历史 conversation_history = self._build_conversation_history(context, task_id) # 计算历史用户消息数 _history_user_msg_count = sum( 1 for m in conversation_history if getattr(m, 'role', '') == 'user' and getattr(m, 'metadata', {}).get('key', '') not in ('llm_output', 'conversation_insight') ) # 保存用户消息到会话记忆 if self.memory: _attachment_meta = {} if context.metadata.get("user_image_files"): _attachment_meta["images"] = context.metadata["user_image_files"] if context.metadata.get("user_file_files"): _attachment_meta["files"] = context.metadata["user_file_files"] self.memory.add_session(agent_id=_effective_agent_id, session_id=context.session_id, role="user", content=context.user_message, key="user_input", metadata=_attachment_meta if _attachment_meta else None, ) # 加载相关记忆 if self.memory_agent and context.user_message: mem_ctx = AgentContext( task_id=task_id, session_id=context.session_id, user_message=context.user_message, metadata={"memory_action": "get_relevant"}, ) await self.memory_agent.process(mem_ctx) if "memory_context_prompt" in mem_ctx.working_memory: context.working_memory["memory_context_prompt"] = \ mem_ctx.working_memory["memory_context_prompt"] # ── 构建初始 messages 列表 ── messages: List[Message] = [] # System prompt _system_content = self.SYSTEM_PROMPT if agent_override_prompt: _system_content = agent_override_prompt + "\n\n" + _system_content # 注入上下文 (context builder) if self.context_builder: try: _memory_ctx = context.working_memory.get("memory_context_prompt", "") ctx_result = self.context_builder.build_context( agent_name=agent_name or "", agent_description=agent_description or "", session_id=context.session_id, conversation_history=conversation_history, user_typed_text=context.user_message or "", user_voice_text="", task_plan=current_task_plan, memory_context_prompt=_memory_ctx, agent_path=str(_effective_agent_id) if _effective_agent_id else None, ) # build_context 返回 (context_xml, static_xml, dynamic_xml) if isinstance(ctx_result, tuple): ctx_text = ctx_result[0] else: ctx_text = ctx_result if ctx_text: _system_content += "\n\n" + ctx_text except Exception as e: logger.warning(f"[{task_id}] 上下文构建失败: {e}") messages.append(Message(role="system", content=_system_content)) # [v1.47.20] VNC 模式下注入浏览器工具使用提示 try: from core.vnc_manager import get_vnc_manager vnc_mgr = get_vnc_manager() if vnc_mgr.is_running: vnc_hint = ( "\n\n## VNC 远程桌面模式提示\n" "当前运行在 VNC 远程桌面环境，浏览器为 Firefox（不支持 Chromium/CDP）。\n" "- **网页浏览**: 优先使用 stealth_browser_start → stealth_browser_navigate → stealth_browser_content\n" "- **获取页面内容**: stealth_browser_content（返回截图+标签页信息），不要使用 browser_open\n" "- **交互操作**: stealth_browser_click / stealth_browser_fill / stealth_browser_key\n" "- **不要使用**: browser_open（需要 Chromium）、web_control（需要前端面板）\n" "- **不要关闭 Firefox**: stealth_browser_close 在 VNC 模式下只释放会话，不关闭浏览器" ) messages[0] = Message(role="system", content=messages[0].content + vnc_hint) except (ImportError, Exception): pass # 注入对话历史 if conversation_history: _history_budget = int(self.context_builder.context_window * 0.25) if self.context_builder else 50000 _history_msgs = [] _history_chars = 0 _reversed_history = list(reversed(conversation_history)) for msg in _reversed_history: _role = msg.role if hasattr(msg, 'role') else msg.get('role', '') _content = msg.content if hasattr(msg, 'content') else msg.get('content', '') _key = getattr(msg, 'metadata', {}).get('key', '') if hasattr(msg, 'metadata') else '' if _role not in ('user', 'assistant'): continue if _key in ('llm_output', 'llm_input', 'reasoning', 'tool_result_raw'): continue if not _content or not _content.strip(): continue _max_msg_chars = 10000 if len(_content) > _max_msg_chars: _content = _content[:_max_msg_chars] + "\n... [内容已截断]" if _history_chars + len(_content) > _history_budget: break _history_msgs.append(Message(role=_role, content=_content)) _history_chars += len(_content) _history_msgs.reverse() if _history_msgs: messages.extend(_history_msgs) logger.info(f"[{task_id}] 注入 {len(_history_msgs)} 条对话历史 (共 {_history_chars} 字符)") # 注入用户消息（首轮） user_images = context.metadata.get("user_images", []) if user_images: multimodal_content = [{"type": "text", "text": context.user_message or "请描述这些图片。"}] for img in user_images: if img.get("url"): multimodal_content.append({ "type": "image_url", "image_url": {"url": img["url"]} }) messages.append(Message(role="user", content=multimodal_content)) else: messages.append(Message( role="user", content=context.user_message or "请处理上述上下文。" )) # ── 获取工具定义 ── tools = self._get_tools() # ── 主循环 ── while self._iteration_count < max_iter: self._iteration_count += 1 logger.info(f"[{task_id}] V2 迭代 {self._iteration_count}/{max_iter}") # 检查取消信号 if context.session_id in self._cancelled_sessions: logger.info(f"[{task_id}] 收到停止信号，终止执行循环") context.working_memory["final_response"] = "⏹️ 任务已被用户停止" await self._emit_v2_event("v2_stopped", {"reason": "用户手动停止"}, stream_callback) self._cancelled_sessions.discard(context.session_id) break # 检查配置热加载 if self.config_broadcaster: reloaded, reload_type = await self.config_broadcaster.check_and_wait(task_id) if reloaded: logger.info(f"[{task_id}] V2 迭代 {self._iteration_count}: {reload_type}已热更新") # 检查注入消息 if context.pending_injected_messages: injected = context.pending_injected_messages.copy() context.pending_injected_messages.clear() for msg_text in injected: logger.info(f"[{task_id}] 注入消息到对话历史: {msg_text[:50]}...") messages.append(Message(role="user", content=f"[用户中断/补充]: {msg_text}")) # ── 调用 LLM（带 tools） ── _reasoning_parts = [] async def _reasoning_delta_cb(full_reasoning, delta_text): _reasoning_parts.append(delta_text) await self._emit_v2_event("v2_reasoning", {"content": delta_text}, stream_callback) # 可重试的 LLM 调用（最多 5 次） _max_llm_retries = 5 _llm_retry_count = 0 response = None while _llm_retry_count < _max_llm_retries: try: if stream_response and self.llm: response = await self._call_llm_stream( messages, tools=tools if tools else None, text_delta_callback=text_delta_callback, reasoning_delta_callback=_reasoning_delta_cb, stream_response=stream_response, ) else: response = await self._call_llm(messages, tools=tools if tools else None) except Exception as _llm_exc: # 异常类错误 → 判断是否可重试 _llm_exc_str = str(_llm_exc).lower() _is_retryable = any(kw in _llm_exc_str for kw in ( "connection", "timeout", "timed out", "429", "500", "502", "503", "504", "rate_limit", "rate limit", "overloaded", "capacity", "network", "eof", )) _llm_retry_count += 1 if _is_retryable and _llm_retry_count < _max_llm_retries: _delay = 2.0 * (2 ** (_llm_retry_count - 1)) # 2s, 4s, 8s, 16s logger.warning( f"[{task_id}] LLM 调用异常 (第 {_llm_retry_count}/{_max_llm_retries} 次)，" f"{_delay:.0f}s 后重试: {_llm_exc}" ) await self._emit_v2_event("v2_reasoning", { "content": f"⏳ 网络不稳定，正在重试 ({_llm_retry_count}/{_max_llm_retries})..." }, stream_callback) await asyncio.sleep(_delay) continue else: # 不可重试 或 已达上限 → 包装为失败 response logger.error(f"[{task_id}] LLM 调用异常 (已重试 {_llm_retry_count} 次): {_llm_exc}") response = type("LLMResponse", (), {"success": False, "error": str(_llm_exc)})() break # 没有异常 → 检查 response.success if response.success: break # 成功 → 跳出重试循环 # response.success == False 但没抛异常 → 判断错误是否可重试 _llm_error = (response.error or "").lower() _is_retryable = any(kw in _llm_error for kw in ( "connection", "timeout", "timed out", "429", "500", "502", "503", "504", "rate_limit", "rate limit", "overloaded", "capacity", "network", "eof", )) _llm_retry_count += 1 if _is_retryable and _llm_retry_count < _max_llm_retries: _delay = 2.0 * (2 ** (_llm_retry_count - 1)) logger.warning( f"[{task_id}] LLM 返回失败 (第 {_llm_retry_count}/{_max_llm_retries} 次)，" f"{_delay:.0f}s 后重试: {response.error}" ) await self._emit_v2_event("v2_reasoning", { "content": f"⏳ 网络不稳定，正在重试 ({_llm_retry_count}/{_max_llm_retries})..." }, stream_callback) await asyncio.sleep(_delay) continue else: # 不可重试 或 已达上限 → 保持失败 response，退出循环 break # LLM 调用失败处理 if not response.success: _llm_error = response.error or "" logger.error(f"[{task_id}] LLM 调用失败: {_llm_error}") # 审查拦截 if "451" in _llm_error or "censorship" in _llm_error.lower() or "blocked" in _llm_error.lower(): _censor_msg = "抱歉，当前对话内容触发了安全审查，无法继续处理。请尝试修改您的请求内容后重试。" context.working_memory["final_response"] = _censor_msg await self._emit_v2_event("v2_reasoning", {"content": _censor_msg}, stream_callback) break # 模型不支持图片 _vision_keywords = [ "doesn't support image", "does not support image", "model_incompatible", "image input", "not support vision", "unsupported multimodal", "不支持图片", "不支持图像", ] _is_vision_error = ( any(kw.lower() in _llm_error.lower() for kw in _vision_keywords) and context.metadata.get("user_images") ) if _is_vision_error: _vision_skip_msg = f"⚠️ 模型 {self.llm.model} 不支持图片，正在切换..." context.working_memory["final_response"] = _vision_skip_msg await self._emit_v2_event("v2_reasoning", {"content": _vision_skip_msg}, stream_callback) break # 其他错误（含已耗尽重试次数的连接错误） error_msg = f"LLM 调用失败 (已重试 {_llm_retry_count} 次): {_llm_error}" context.working_memory["final_response"] = error_msg await self._emit_v2_event("v2_reasoning", {"content": error_msg}, stream_callback) break # 保存推理内容 _reasoning_content = (response.reasoning or "").strip() if _reasoning_content or _reasoning_parts: _full_reasoning = _reasoning_content or "".join(_reasoning_parts) if self.memory: self.memory.add_session(agent_id=_effective_agent_id, session_id=context.session_id, role="assistant", content=_full_reasoning, key="reasoning", importance=0.2, ) await self._emit_v2_event("v2_reasoning", {"content": _reasoning_content}, stream_callback) # ── 检查 tool_calls ── if response.tool_calls: # 有工具调用 → 执行工具 → 喂回 LLM → 继续循环 # 添加 assistant 消息（包含 tool_calls）到消息列表 messages.append(Message( role="assistant", content=response.content or "", tool_calls=response.tool_calls, )) # 保存 LLM 原始输出到会话记忆 if self.memory: _llm_output = response.content or "" if _llm_output: self.memory.add_session(agent_id=_effective_agent_id, session_id=context.session_id, role="assistant", content=_llm_output, key="llm_output", importance=0.3, ) # 逐个执行工具调用 for tool_call in response.tool_calls: tc_id = tool_call.get("id", "") tc_name = tool_call.get("name", "") _tc_args_raw = tool_call.get("arguments", "{}") # 原生 tool_calling 模式下 arguments 可能是 dict（已解析），也可能是 string if isinstance(_tc_args_raw, dict): tc_args_str = json.dumps(_tc_args_raw, ensure_ascii=False) tc_params = _tc_args_raw else: tc_args_str = str(_tc_args_raw) if _tc_args_raw else "{}" try: tc_params = json.loads(tc_args_str) if tc_args_str else {} except (json.JSONDecodeError, TypeError): tc_params = {"raw_input": tc_args_str} logger.info(f"[{task_id}] 执行工具: {tc_name}") # tc_params 已在上面解析完成 # 注入 session_id 供记忆/知识库工具使用 if tc_name in ("save_memory", "recall_memory", "update_conversation_title"): tc_params.setdefault("session_id", context.session_id) # 发送工具开始事件 await self._emit_v2_event( "v2_tool_start", {"tool": { "toolname": tc_name, "parms": truncate_str(tc_args_str, 500), }}, stream_callback, ) self._add_exec_event("tool_call", { "title": f"调用工具: {tc_name}", "tool_name": tc_name, "arguments": tc_args_str, }) # 执行工具 _timeout = 120 # 从参数中提取 timeout（如果有的话） if isinstance(tc_params, dict) and "timeout" in tc_params: try: _timeout = int(tc_params.pop("timeout")) except (ValueError, TypeError): pass tool_result = await self._execute_v2_tool( tc_name, json.dumps(tc_params, ensure_ascii=False), _timeout, context, task_id, stream_callback=stream_callback, sent_files=_sent_files, agent_path=agent_path, ) # task_plan 特殊处理 if tc_name == "task_plan" and tool_result.get("success"): _plan_data = tool_result.get("data", {}) _new_plan = _plan_data.get("plan", "") if isinstance(_plan_data, dict) else "" if _new_plan: current_task_plan = _new_plan await self._emit_v2_event( "v2_task_plan", {"plan": truncate_str(current_task_plan, 2000)}, stream_callback, ) # 提取工具输出 if tool_result is None: tool_result = {"success": False, "error": "工具返回了空结果"} _output_text = ( tool_result.get("output", "") or tool_result.get("message", "") or tool_result.get("stdout", "") or tool_result.get("error", "") ) # 如果 output/message/stdout 均为空但 data 有内容，格式化 data 作为输出 if not _output_text and tool_result.get("data"): try: _output_text = json.dumps(tool_result["data"], ensure_ascii=False, default=str)[:30000] except Exception: _output_text = str(tool_result["data"])[:30000] # 发送工具结果事件 await self._emit_v2_event( "v2_tool_result", {"tool": {"toolname": tc_name}, "result": { "success": tool_result.get("success", False), "output": truncate_str(_output_text, 30000), "error": truncate_str(tool_result.get("error", ""), 30000), }}, stream_callback, ) self._add_exec_event("tool_result", { "title": f"工具结果: {tc_name}", "tool_name": tc_name, "success": tool_result.get("success", False), "summary": truncate_str(_output_text, 30000), }) # 保存工具调用到会话记忆 if self.memory: tool_result_data = { "tool_name": tc_name, "params": truncate_str(tc_args_str, 500), "success": tool_result.get("success", False), "output": truncate_str(_output_text, 5000), } self.memory.add_session(agent_id=_effective_agent_id, session_id=context.session_id, role="assistant", content=f"调用工具: {tc_name}\n参数: {truncate_str(tc_args_str, 1000)}", key="tool_call", importance=0.4, metadata={"tool_result": tool_result_data}, ) # 添加 tool result 消息到消息列表 _result_str = truncate_str(_output_text, 30000) if _output_text else json.dumps(tool_result, ensure_ascii=False, default=str)[:3000] messages.append(Message( role="tool", content=_result_str, tool_call_id=tc_id, name=tc_name, )) # 所有工具执行完毕 → 继续循环（让 LLM 处理工具结果） continue else: # 没有原生工具调用 → 纯文本回复，完全依赖 tool_calling reply_text = (response.content or "").strip() logger.info(f"[{task_id}] 无工具调用，任务完成 (reply长度={len(reply_text)})") if not reply_text: reply_text = "处理完毕。" context.working_memory["final_response"] = reply_text # 发送最终回复事件 await self._emit_v2_event("v2_reasoning", {"content": truncate_str(reply_text, 3000)}, stream_callback) # 保存回复到会话记忆 if self.memory: self.memory.add_session(agent_id=_effective_agent_id, session_id=context.session_id, role="assistant", content=reply_text, key="reply", importance=0.5, ) # 保存 LLM 原始输出 if self.memory and response.content: self.memory.add_session(agent_id=_effective_agent_id, session_id=context.session_id, role="assistant", content=response.content, key="llm_output", importance=0.3, ) break # 循环正常结束（max_iter 耗尽） else: logger.warning(f"[{task_id}] 达到最大迭代次数 ({max_iter})，任务未完成") await self._emit_v2_event("v2_iter_limit", { "message": f"任务已达到最大迭代次数 ({max_iter})", "iterations": self._iteration_count, "max_iterations": max_iter, "task_id": task_id, }, stream_callback) # 持久化已发送文件 if _sent_files and self.memory: try: # [v1.23.75] 分离 chat_agent 类型和普通文件类型 _chat_entries = [f for f in _sent_files if f.get("_type") == "chat_agent"] _file_entries = [f for f in _sent_files if f.get("_type") != "chat_agent"] # 将 chat_agent 条目写入 agent_chat 表（私聊记录） if _chat_entries: logger.info(f"[{task_id}] [私聊保存] main_agent 发现 {len(_chat_entries)} 条私聊条目") try: from groups.manager import GroupManager logger.info(f"[{task_id}] [私聊保存] 正在创建 GroupManager") _gm = GroupManager() _gm.initialize() logger.info(f"[{task_id}] [私聊保存] GroupManager 初始化完成") for _idx, _ce in enumerate(_chat_entries): _from_agent = agent_path or str(_effective_agent_id) _from_name = "" logger.info(f"[{task_id}] [私聊保存] 处理第 {_idx+1} 条私聊: from_agent={_from_agent}, target_agent={_ce.get('target_agent', '')}") try: import sqlite3 _db_path = str(Path.home() / ".myagent" / "data" / "agents.db") logger.info(f"[{task_id}] [私聊保存] 查询 agents.db: {_db_path}") _db = sqlite3.connect(_db_path) _db.row_factory = sqlite3.Row _row = _db.execute("SELECT name FROM agents WHERE path = ? OR id = ?", (_from_agent, _from_agent)).fetchone() if _row and _row["name"]: _from_name = _row["name"] logger.info(f"[{task_id}] [私聊保存] 查询到 from_name: {_from_name}") else: logger.info(f"[{task_id}] [私聊保存] 未查询到 from_name") _db.close() except Exception as _e: logger.warning(f"[{task_id}] [私聊保存] 查询 from_name 失败: {_e}") _to_agent = _ce.get("target_agent", "") _to_name = _ce.get("target_name", "") _content = _ce.get("message", "") logger.info(f"[{task_id}] [私聊保存] 调用 add_agent_chat: from_agent={_from_agent}, from_name={_from_name}, to_agent={_to_agent}, to_name={_to_name}") _msg_id = _gm.add_agent_chat( group_id="", from_agent=_from_agent, from_name=_from_name, to_agent=_to_agent, to_name=_to_name, content=_content, ) logger.info(f"[{task_id}] [私聊保存] add_agent_chat 返回 msg_id={_msg_id}") logger.info(f"[{task_id}] Agent私聊已保存: {_from_name} → {_to_name}: {_content[:80]}") _gm.close() logger.info(f"[{task_id}] [私聊保存] GroupManager 已关闭，处理完成") except Exception as _ce_err: logger.warning(f"[{task_id}] [私聊保存] 保存Agent私聊记录失败: {_ce_err}") logger.exception(f"[{task_id}] [私聊保存] 异常详情") # 普通文件列表照常持久化到 session memory if _file_entries: self.memory.add_session(agent_id=_effective_agent_id, session_id=context.session_id, role="assistant", content="", key="file_send", importance=0.2, metadata={"files": _file_entries}, ) except Exception as _fe: logger.warning(f"[{task_id}] 持久化文件信息失败: {_fe}") context.working_memory["iterations"] = self._iteration_count if current_task_plan: context.working_memory["task_plan"] = current_task_plan logger.info(f"[{task_id}] V2 循环完成 (共 {self._iteration_count} 次迭代)") return context async def _execute_v2_tool( self, tool_name: str, parms_str: str, timeout: int, context: AgentContext, task_id: str, stream_callback: Optional[Callable] = None, sent_files: Optional[List[Dict[str, Any]]] = None, agent_path: Optional[str] = None, ) -> Dict[str, Any]: """[v1.22.0] V2 工具执行 — 统一分发到 ToolDispatcher""" try: import json as _json import html as _html try: # [v1.39] 改进: 先尝试直接解析，失败再 unescape 后重试 # 原生 tool_calling 路径传入的是干净的 JSON，不需要 unescape # 旧的 XML 路径可能包含 HTML 实体，需要 unescape params = _json.loads(parms_str) if parms_str else {} except (_json.JSONDecodeError, TypeError): try: # 二次尝试: unescape HTML 实体后再解析 _clean_parms = _html.unescape(parms_str) if parms_str else "" params = _json.loads(_clean_parms) if _clean_parms else {} except (_json.JSONDecodeError, TypeError): # 三次尝试: 使用 safe_json_parse 进行宽松解析 from core.utils import safe_json_parse params = safe_json_parse(parms_str, default={"raw_input": parms_str}) if self.dispatcher: return await self.dispatcher.dispatch( tool_name=tool_name, params=params, timeout=timeout, task_id=task_id, stream_callback=stream_callback, sent_files=sent_files, agent_path=agent_path, agent_id=agent_path, # agent_path 就是 agent_id ) # 兼容回退: dispatcher 未初始化时使用基础 fallback result = {"success": False, "output": "", "error": f"工具分发器未初始化: {tool_name}"} except Exception as e: result = {"success": False, "output": "", "error": f"工具调用异常: {tool_name} - {e}"} logger.warning(f"[{task_id}] 工具调用异常 ({tool_name}): {e}") return result