diff --git a/core/main_terminal_parts/tools_definition/core_tools.py b/core/main_terminal_parts/tools_definition/core_tools.py index c0cbc4f..5151ec8 100644 --- a/core/main_terminal_parts/tools_definition/core_tools.py +++ b/core/main_terminal_parts/tools_definition/core_tools.py @@ -103,7 +103,7 @@ class ToolsDefinitionCoreToolsMixin: "type": "function", "function": { "name": "sleep", - "description": "等待工具。两种模式二选一：1) seconds：短暂延迟；2) wait_runcommand_id：等待指定后台 run_command 结束并直接返回结果。若同时提供多个等待参数会报错。", + "description": "等待工具。三种模式三选一：1) seconds：短暂延迟；2) wait_runcommand_id：等待指定后台 run_command 结束并直接返回结果；3) wait_sub_agent_output_ids：等待指定子智能体下一次输出并直接返回该消息（多智能体模式专用）。若同时提供多个等待参数会报错。", "parameters": { "type": "object", "properties": self._inject_intent({ @@ -115,6 +115,11 @@ class ToolsDefinitionCoreToolsMixin: "type": "string", "description": "等待指定后台 run_command 的 command_id 结束后返回。" }, + "wait_sub_agent_output_ids": { + "type": "array", + "items": {"type": "integer"}, + "description": "等待指定子智能体下一次输出并直接返回该消息。多智能体模式专用，且一次只能包含一个编号。" + }, "reason": { "type": "string", "description": "等待的原因说明（可选）" diff --git a/core/main_terminal_parts/tools_execution.py b/core/main_terminal_parts/tools_execution.py index fe33eaa..0b503f5 100644 --- a/core/main_terminal_parts/tools_execution.py +++ b/core/main_terminal_parts/tools_execution.py @@ -1080,6 +1080,7 @@ class MainTerminalToolsExecutionMixin: elif tool_name == "sleep": seconds = arguments.get("seconds") wait_sub_agent_ids = arguments.get("wait_sub_agent_ids") + wait_sub_agent_output_ids = arguments.get("wait_sub_agent_output_ids") wait_runcommand_id = arguments.get("wait_runcommand_id") reason = arguments.get("reason", "等待操作完成") @@ -1088,23 +1089,60 @@ class MainTerminalToolsExecutionMixin: provided += 1 if wait_sub_agent_ids: provided += 1 + if wait_sub_agent_output_ids: + provided += 1 if wait_runcommand_id: provided += 1 if provided == 0: result = { "success": False, - "error": "sleep 至少需要提供一个参数：seconds / wait_sub_agent_ids / wait_runcommand_id" + "error": "sleep 至少需要提供一个参数：seconds / wait_sub_agent_ids / wait_sub_agent_output_ids / wait_runcommand_id" } elif provided > 1: result = { "success": False, - "error": "sleep 的等待参数互斥：seconds / wait_sub_agent_ids / wait_runcommand_id 只能提供一个" + "error": "sleep 的等待参数互斥：seconds / wait_sub_agent_ids / wait_sub_agent_output_ids / wait_runcommand_id 只能提供一个" } + elif wait_sub_agent_output_ids: + if not getattr(self, "multi_agent_mode", False): + result = {"success": False, "error": "wait_sub_agent_output_ids 仅在多智能体模式下可用"} + elif not isinstance(wait_sub_agent_output_ids, list) or len(wait_sub_agent_output_ids) != 1: + result = {"success": False, "error": "wait_sub_agent_output_ids 必须包含且仅包含一个子智能体编号"} + else: + try: + agent_id = int(wait_sub_agent_output_ids[0]) + if agent_id <= 0: + raise ValueError() + except Exception: + result = {"success": False, "error": "wait_sub_agent_output_ids 必须是正整数"} + else: + manager = getattr(self, "sub_agent_manager", None) + if not manager: + result = {"success": False, "error": "子智能体管理器不可用"} + else: + state = manager.get_multi_agent_state(getattr(self.context_manager, "current_conversation_id", None)) + if not state: + result = {"success": False, "error": "当前对话没有多智能体状态"} + else: + try: + loop = asyncio.get_running_loop() + fut = state.register_output_wait(agent_id, loop) + msg = await asyncio.wait_for(fut, timeout=300) + result = { + "success": True, + "mode": "wait_sub_agent_output", + "agent_id": agent_id, + "message": msg, + } + except asyncio.TimeoutError: + result = {"success": False, "error": f"等待子智能体 {agent_id} 输出超时（5分钟）"} + except Exception as exc: + result = {"success": False, "error": f"等待子智能体 {agent_id} 输出失败: {exc}"} elif wait_sub_agent_ids: if getattr(self, "multi_agent_mode", False): result = { "success": False, - "error": "多智能体模式下 sleep 工具不支持 wait_sub_agent_ids。请通过 list_active_sub_agents / get_sub_agent_status 查看状态，或直接向子智能体发送消息。" + "error": "多智能体模式下 sleep 工具不支持 wait_sub_agent_ids，请使用 wait_sub_agent_output_ids。" } elif not isinstance(wait_sub_agent_ids, list) or not wait_sub_agent_ids: result = {"success": False, "error": "wait_sub_agent_ids 必须是非空数组"} diff --git a/modules/multi_agent/state.py b/modules/multi_agent/state.py index 1cafa65..ed6932d 100644 --- a/modules/multi_agent/state.py +++ b/modules/multi_agent/state.py @@ -286,6 +286,9 @@ class MultiAgentState: # 一个 agent 可能同时只阻塞在一个 ask 工具上（最简实现） # key = agent_id, value = question_id（表示当前 agent 正阻塞等待） self.agent_blocking_question: Dict[int, str] = {} + # 主智能体通过 sleep(wait_sub_agent_output_ids) 等待某个子智能体下一次输出 + # key = agent_id, value = asyncio.Future（结果为完整消息文本） + self.output_waits: Dict[int, asyncio.Future] = {} # 角色实例计数：role_id -> 已分配的最大 agent_id（数字） # 用于创建新实例时自动递增编号，但允许调用方显式指定 self.role_counters: Dict[str, int] = {} @@ -353,6 +356,76 @@ class MultiAgentState: def has_pending_master_messages(self) -> bool: return len(self.pending_master_messages) > 0 + # ----- 等待子智能体输出（sleep wait_sub_agent_output_ids） ----- + def register_output_wait( + self, agent_id: int, loop: AbstractEventLoop + ) -> asyncio.Future: + """注册等待指定子智能体的下一次输出。 + + 如果 `pending_master_messages` 里已经有该子智能体的未消费输出， + 则立即消费并返回该消息；否则返回一个 future，由后续输出唤醒。 + """ + fut: asyncio.Future = loop.create_future() + inst = self.agents.get(agent_id) + if not inst: + fut.set_exception(ValueError(f"未找到子智能体 {agent_id}")) + return fut + + if inst.status in ("terminated", "failed"): + fut.set_exception( + RuntimeError( + f"子智能体 {agent_id} 当前状态为 {inst.status}，无法等待输出" + ) + ) + return fut + + # 优先消费 pending_master_messages 里已有的未消费输出 + display_name = inst.display_name + for i, msg in enumerate(self.pending_master_messages): + parsed = parse_multi_agent_message(msg) + if parsed and parsed.get("display_name") == display_name: + self.pending_master_messages.pop(i) + fut.set_result(msg) + ma_debug( + "state_output_wait_claimed_pending", + conversation_id=self.conversation_id, + agent_id=agent_id, + display_name=display_name, + ) + return fut + + # 没有未消费输出，注册等待 + self.output_waits[agent_id] = fut + ma_debug( + "state_output_wait_registered", + conversation_id=self.conversation_id, + agent_id=agent_id, + display_name=display_name, + ) + return fut + + def claim_output_wait(self, agent_id: int, message_text: str) -> bool: + """如果该 agent 正被 sleep 等待输出，则把消息交给等待方并阻止进入主对话。""" + fut = self.output_waits.pop(agent_id, None) + if not fut: + return False + loop: Optional[AbstractEventLoop] = None + try: + loop = fut.get_loop() + except Exception: + pass + if loop and not loop.is_closed(): + try: + loop.call_soon_threadsafe(fut.set_result, message_text) + return True + except Exception: + pass + try: + fut.set_result(message_text) + except Exception: + pass + return True + # ----- 阻塞问答 ----- async def wait_for_answer(self, question_id: str, agent_id: int, timeout: float = 600.0) -> str: """子智能体 ask_* 工具调用后阻塞等待答案。 @@ -496,6 +569,17 @@ class MultiAgentState: loop.call_soon_threadsafe(fut.cancel) except Exception: pass + for fut in list(self.output_waits.values()): + if not fut.done(): + try: + loop = fut.get_loop() + if loop and not loop.is_closed(): + loop.call_soon_threadsafe(fut.cancel) + else: + fut.cancel() + except Exception: + pass + self.output_waits.clear() self.agents.clear() self.task_id_to_agent_id.clear() self.pending_master_messages.clear() diff --git a/modules/sub_agent/task.py b/modules/sub_agent/task.py index be4bf01..7f4d382 100644 --- a/modules/sub_agent/task.py +++ b/modules/sub_agent/task.py @@ -396,6 +396,27 @@ class SubAgentTask: try: from modules.multi_agent.state import build_sub_agent_output_text msg = build_sub_agent_output_text(self.display_name, output_text.strip(), is_final=is_final) + # 如果该 agent 正被 sleep(wait_sub_agent_output_ids) 等待，把输出交给等待方， + # 不再插入主对话（避免同一条消息出现两遍） + if self.multi_agent_state.claim_output_wait(self.agent_id, msg): + ma_debug( + "sub_agent_output_claimed_by_wait", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + msg_preview=msg[:200], + ) + # 同时记录到实例状态，供 list_active_sub_agents 使用 + inst = self.multi_agent_state.get_instance(self.agent_id) + if inst: + inst.last_output = output_text[:500] + self.emit("progress", { + "subtype": "output", + "content": output_text, + "is_final": is_final, + "ts": int(time.time() * 1000), + }) + return self.multi_agent_state.push_master_message(msg) ma_debug( "sub_agent_forward_pushed", diff --git a/prompts/multi_agent/master.txt b/prompts/multi_agent/master.txt index 71fd9d0..776af50 100644 --- a/prompts/multi_agent/master.txt +++ b/prompts/multi_agent/master.txt @@ -25,15 +25,40 @@ - **监督与引导**：掌握团队全局进度，在子智能体偏离方向时及时干预纠正 - **综合与决策**：收集子智能体的产出，综合形成最终结果，对用户负责 -- **除非用户要求，或必须为之，你需要尽可能避免自己亲自去执行任务** +你不是单打独斗的执行者，但你的主要价值不在于亲自执行。你拥有完整的工具集（文件读写、终端、搜索、MCP、skill、memory 等），只在必要的统筹、协调和判断中使用它们；具体的读、写、改、验证等操作默认交给子智能体。 -你不是单打独斗的执行者。除非任务极其简单或明确不需要子智能体，否则你应该主动把任务拆解并指派给合适的角色。但同时，你仍然拥有完整的工具集（文件读写、终端、搜索、MCP、skill、memory 等），在需要时可以自己动手。 +## 你自己做 vs 派给子智能体 + +你的核心定位是 **统筹者、协调者、决策者和汇报者**，而不是执行者。你的上下文非常宝贵，必须省着用。 + +### 你必须自己做（不可替代） + +1. **理解用户意图**：澄清模糊需求、追问关键信息、把用户的话翻译成可执行目标。 +2. **整体规划与任务拆解**：决定要做哪些事、按什么顺序、需要哪些角色。 +3. **分配与协调子智能体**：创建子智能体、给它们下指令、在它们之间传递信息、纠正方向。 +4. **回答子智能体提问**：通过 `answer_sub_agent_question` 及时回应子智能体的 `ask_master`。 +5. **综合与最终汇报**：把各子智能体的产出整合成统一结论，用清晰结构向用户汇报。 + +### 你应该派给子智能体（默认不自己做） + +1. **项目初期调查**：从对话开始就可以创建子智能体并行调查项目结构、关键文件、代码位置、运行方式。不要等"正式开始"才创建它们。 +2. **具体实现与修改**：写代码、改配置、调整前端、调试 bug 等。 +3. **专项调研与对比**：技术选型、竞品分析、文档阅读、方案搜索。 +4. **验证与检查**：代码审查、测试执行、结果验证。 +5. **大量文件阅读**：需要读多个文件或长文件才能得出结论时，优先派给子智能体，而不是自己逐行读。 + +### 核心原则 + +- **默认派出去**：只要不是"理解、规划、协调、决策、汇报"这几类工作，就默认创建子智能体去做。 +- **尽早并行**：任务一开始就可以创建 2-3 个子智能体并行摸底，你再根据反馈调整方向。 +- **节省上下文**：尽量避免自己调用 `read_file`、`write_file`、`edit_file`、`run_command` 等消耗上下文的工具；这些操作优先由子智能体完成，你只要看结论。 +- **只做统筹**：你的价值在于判断"谁做什么、何时干预、怎么综合"，而不是亲自敲代码。 ## 任务复杂度评估与投入预算 -在开始执行前，先评估任务复杂度，决定子智能体数量和工具调用预算： +在决定派多少子智能体时，评估任务复杂度，决定子智能体数量和工具调用预算： -- **简单任务**（单一明确目标，1-2 步即可完成）：1 个子智能体，约 3-10 次工具调用。甚至可以自己直接做，不需要创建子智能体。 +- **简单任务**（单一明确目标，1-2 步即可完成）：1 个子智能体，约 3-10 次工具调用。 - **中等任务**（需要多方面调查或 2-3 个不同角色协作）：2-4 个子智能体，每个约 10-15 次工具调用。 - **复杂任务**（涉及多个领域、需要深度调查或多阶段执行）：可以创建更多子智能体，每个有明确分工的职责范围。 @@ -57,7 +82,7 @@ ## 工作原则 - **明确你的身份**：和子智能体交流的对象是你（Team Leader），不是用户。对子智能体来说，你才是它们的 user。你在委派任务、回答提问、发送引导消息时，代表的是 Team Leader，不要替用户转述或以用户自居。 -- **主动分工**：除非任务极其简单或明确不需要子智能体，否则主动把任务拆解并指派给合适的角色。 +- **主动分工**：根据「你自己做 vs 派给子智能体」的分类，把执行性、调研性、验证性工作默认派给子智能体；你自己专注于理解、规划、协调、决策和汇报。 - **明确指令**：委派时写清楚任务目标、范围、产出要求和边界（见上方"委派质量标准"）。 - **及时回答**：当子智能体通过 `ask_master` 提问时，必须尽快通过 `answer_sub_agent_question` 回答。子智能体在阻塞等待你的回答，拖延会卡住整个团队。 - **监督进度**：通过 `list_active_sub_agents` / `get_sub_agent_status` 掌握全局，并在合适的时机引导子智能体。 @@ -71,7 +96,7 @@ 当子智能体正在工作时，你的行为必须遵循以下规则： - **严格禁止反复查看状态**：不要在子智能体运行期间反复调用 `get_sub_agent_status` 或 `list_active_sub_agents` 来查看进度。子智能体的输出会自动以消息形式插入你的对话，你不需要主动轮询。反复查状态是浪费时间的行为。 -- **严格禁止用 sleep 填充等待**：不要在子智能体运行期间反复调用 `sleep` 工具来等待。如果你没有需要立即处理的事情，**立刻停止输出**，等待子智能体的汇报消息自动到达。 +- **等待子智能体输出的正确方式**：如果你没有需要立即处理的事情，可以选择**立刻停止输出**（等待子智能体的汇报消息自动到达），也可以选择使用 `sleep` 工具的 `wait_sub_agent_output_ids` 主动等待指定子智能体的下一次输出。不要在子智能体运行期间反复调用 `sleep` 做无意义的短延迟。 - **无事可做时立刻停止输出**：如果你已经把任务委派出去，当前没有需要回答的 `ask_master` 提问，也没有需要干预的情况，就**停止输出**，不要输出任何文字。子智能体的产出会以 user 消息自动插入你的对话，届时你再继续工作。 - **有事可做时主动干预**：如果你在子智能体的输出中发现了需要纠正的错误、可以提供的指导、或需要传递给其他子智能体的信息，立刻用 `send_message_to_sub_agent` 行动。这是你运行期间最有价值的工作。 diff --git a/utils/tool_result_formatter/terminal.py b/utils/tool_result_formatter/terminal.py index 17a0ee8..3c32b8b 100644 --- a/utils/tool_result_formatter/terminal.py +++ b/utils/tool_result_formatter/terminal.py @@ -142,6 +142,13 @@ def _format_sleep(result_data: Dict[str, Any]) -> str: if not result_data.get("success"): return _format_failure("sleep", result_data) mode = result_data.get("mode") + if mode == "wait_sub_agent_output": + message = result_data.get("message") or "" + agent_id = result_data.get("agent_id") + header = f"已收到子智能体 {agent_id} 的输出" + if message: + return f"{header}\n\n{message}" + return header if mode == "wait_sub_agent_ids": results = result_data.get("results") or [] header = result_data.get("message") or f"已等待 {len(results)} 个子智能体结束"