diff --git a/core/main_terminal_parts/tools_execution.py b/core/main_terminal_parts/tools_execution.py index e6838bd..9dbe060 100644 --- a/core/main_terminal_parts/tools_execution.py +++ b/core/main_terminal_parts/tools_execution.py @@ -2,6 +2,7 @@ import asyncio import json import os import time +import uuid from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, Set @@ -1755,16 +1756,13 @@ class MainTerminalToolsExecutionMixin: # 构造 task_message(作为 Team Leader 的任务发布) from modules.multi_agent.state import build_master_dispatch_text task_message = build_master_dispatch_text(arguments.get("task", "")) - # 加载并覆盖作业:multi_agent_mode/sub_agent_manager.create_sub_agent 的作业路径 summary_text = (arguments.get("summary") or f"{role.name}作业")[:80] - deliverables_dir = arguments.get("deliverables_dir", f"sub_agent_results/agent_{agent_id}") thinking_mode = arguments.get("thinking_mode") or role.thinking_mode or "fast" # 走原行 发事件创建(避免后期重建提供重复工能重费,直接使用 multi_agent_mode=True 调用) result = self.sub_agent_manager.create_sub_agent( agent_id=int(agent_id), summary=summary_text, task=arguments.get("task", ""), - deliverables_dir=deliverables_dir, run_in_background=bool(arguments.get("run_in_background", True)), timeout_seconds=arguments.get("timeout_seconds"), conversation_id=conv_id, @@ -1773,6 +1771,8 @@ class MainTerminalToolsExecutionMixin: multi_agent_mode=True, role_id=role_id, display_name=display_name, + system_prompt=system_prompt, + task_message=task_message, ) # 在多智能体模式下,主进程 create_sub_agent 总是后台启动, # 主智能体不需要阻塞等待,而是通过子智能体输出转发拿进度。 @@ -1887,11 +1887,13 @@ class MainTerminalToolsExecutionMixin: if not state: result = {"success": False, "error": "多智能体状态未就绪"} else: - # 构造提问并插入子对话(子智能体下一轮 assistant 输出作为回答插入主对话) + question_id = f"ask_sub_agent_{int(time.time() * 1000)}_{uuid.uuid4().hex[:6]}" + # 构造提问并插入子对话(子智能体下一轮 assistant 输出作为回答返回到工具结果) text = format_multi_agent_message( display_name="Team Leader", msg_type=TYPE_ASK, content=question, + msg_id=question_id, target=state.get_instance(agent_id).display_name if state.get_instance(agent_id) else f"Agent_{agent_id}", ) ok = self.sub_agent_manager.inject_message_to_sub_agent(agent_id, text) @@ -1900,7 +1902,7 @@ class MainTerminalToolsExecutionMixin: else: # 阻塞等待子智能体下一轮输出作为回答 answer = await state.wait_for_answer( - question_id=f"ask_sub_agent_{int(time.time())}", + question_id=question_id, agent_id=agent_id, timeout=timeout, ) diff --git a/modules/multi_agent/state.py b/modules/multi_agent/state.py index ddfdece..04f9586 100644 --- a/modules/multi_agent/state.py +++ b/modules/multi_agent/state.py @@ -20,6 +20,7 @@ from __future__ import annotations import asyncio import json import uuid +from asyncio import AbstractEventLoop from dataclasses import dataclass, field from datetime import datetime from pathlib import Path @@ -48,6 +49,7 @@ def format_multi_agent_message( msg_id: Optional[str] = None, target: Optional[str] = None, extra_attrs: Optional[Dict[str, str]] = None, + msg_type_text: Optional[str] = None, ) -> str: """按统一格式构造 user 消息字符串。 @@ -58,15 +60,17 @@ def format_multi_agent_message( msg_id: 消息 id;不传则自动生成 target: 接收方显示名(用于子→子 提问时标明对谁提问) extra_attrs: 额外标签属性(如 question_id="ask_xxx") + msg_type_text: 覆盖默认的中文消息类型文案(如"任务结束汇报") """ if not msg_id: msg_id = f"msg_{uuid.uuid4().hex[:10]}" + type_label = msg_type_text or msg_type_to_text(msg_type) # 第一行:自然语言前缀(含 target 标识) if target: - prefix = f"来自 {display_name} 向 {target} 的{msg_type_to_text(msg_type)}" + prefix = f"来自 {display_name} 向 {target} 的{type_label}" else: - prefix = f"来自 {display_name} 的{msg_type_to_text(msg_type)}" + prefix = f"来自 {display_name} 的{type_label}" # 第二行:id id_line = f"id: {msg_id}" @@ -114,13 +118,14 @@ def build_master_dispatch_text(task: str, msg_id: Optional[str] = None) -> str: ) -def build_sub_agent_output_text(display_name: str, content: str, msg_id: Optional[str] = None) -> str: +def build_sub_agent_output_text(display_name: str, content: str, msg_id: Optional[str] = None, *, is_final: bool = False) -> str: """子智能体输出(进度或完成)插入到主对话的 user 消息文本。""" return format_multi_agent_message( display_name=display_name, msg_type=TYPE_OUTPUT, content=content, msg_id=msg_id, + msg_type_text="任务结束汇报" if is_final else "任务进度输出", ) @@ -223,6 +228,10 @@ class MultiAgentState: # ask_master / ask_other_agent 的等待 future # key = question_id, value = asyncio.Future (结果为 answer str 或 Exception) self.pending_questions: Dict[str, asyncio.Future] = {} + # question_id -> 创建 future 时所在的事件循环,用于跨循环安全 set_result + self.pending_question_loops: Dict[str, AbstractEventLoop] = {} + # 回答早于 wait_for_answer 注册时先暂存 + self.pending_answers: Dict[str, str] = {} # 一个 agent 可能同时只阻塞在一个 ask 工具上(最简实现) # key = agent_id, value = question_id(表示当前 agent 正阻塞等待) self.agent_blocking_question: Dict[int, str] = {} @@ -285,25 +294,26 @@ class MultiAgentState: 返回 answer 字符串;超时/取消抛 asyncio.TimeoutError 或 CancelledError。 """ + # 如果回答已经提前到达,直接返回 + if question_id in self.pending_answers: + return self.pending_answers.pop(question_id) if question_id in self.pending_questions: raise RuntimeError(f"question_id 已存在: {question_id}") - fut: asyncio.Future = asyncio.get_event_loop().create_future() + loop = asyncio.get_running_loop() + fut: asyncio.Future = loop.create_future() self.pending_questions[question_id] = fut + self.pending_question_loops[question_id] = loop self.agent_blocking_question[agent_id] = question_id try: return await asyncio.wait_for(fut, timeout=timeout) - except asyncio.TimeoutError: - raise finally: self.pending_questions.pop(question_id, None) + self.pending_question_loops.pop(question_id, None) if self.agent_blocking_question.get(agent_id) == question_id: self.agent_blocking_question.pop(agent_id, None) - def provide_answer(self, question_id: str, answer: str) -> bool: - """主/其他子智能体 answer_* 工具调用时回写答案。 - - 返回 True 表示找到等待中的 future;False 表示无等待方或已超时。 - """ + async def _do_provide_answer(self, question_id: str, answer: str) -> bool: + """在同 future 所属事件循环内设置结果。""" fut = self.pending_questions.get(question_id) if not fut or fut.done(): return False @@ -313,6 +323,35 @@ class MultiAgentState: return False return True + def provide_answer(self, question_id: str, answer: str) -> bool: + """主/其他子智能体 answer_* 工具调用时回写答案。 + + 返回 True 表示找到等待中的 future;False 表示无等待方或已超时。 + 支持跨事件循环调用(例如主对话循环回答子智能体循环里的提问)。 + """ + # 如果 wait_for_answer 还没注册,先把答案暂存 + if question_id not in self.pending_questions: + self.pending_answers[question_id] = answer + return True + fut = self.pending_questions.get(question_id) + if not fut or fut.done(): + self.pending_answers[question_id] = answer + return False + loop = self.pending_question_loops.get(question_id) + if loop is None: + try: + loop = fut.get_loop() + except Exception: + pass + if loop is not None: + try: + asyncio.run_coroutine_threadsafe(self._do_provide_answer(question_id, answer), loop) + return True + except Exception: + pass + # 同循环回退 + return asyncio.run_coroutine_threadsafe(self._do_provide_answer(question_id, answer), asyncio.get_event_loop()).result(timeout=5) + def is_agent_blocking(self, agent_id: int) -> bool: return agent_id in self.agent_blocking_question diff --git a/modules/multi_agent/tools.py b/modules/multi_agent/tools.py index 741259a..bd055e2 100644 --- a/modules/multi_agent/tools.py +++ b/modules/multi_agent/tools.py @@ -53,10 +53,6 @@ def _master_tool_create_sub_agent() -> Dict[str, Any]: "type": "integer", "description": "(可选)手动指定实例编号;不传时自动递增。", }, - "deliverables_dir": { - "type": "string", - "description": "(可选)交付目录相对路径,留空则用 sub_agent_results/agent_{N}。", - }, "timeout_seconds": {"type": "integer", "description": "超时秒数,默认 600。"}, "thinking_mode": { "type": "string", @@ -64,7 +60,7 @@ def _master_tool_create_sub_agent() -> Dict[str, Any]: "description": "(可选)覆盖角色默认思考模式。不填使用角色配置。", }, }), - "required": ["role_id", "task", "thinking_mode"], + "required": ["role_id", "task"], }, }, } diff --git a/modules/sub_agent/creation.py b/modules/sub_agent/creation.py index f2e0bfb..5abad99 100644 --- a/modules/sub_agent/creation.py +++ b/modules/sub_agent/creation.py @@ -48,7 +48,7 @@ class SubAgentCreationMixin: if agent_id not in used: used.append(agent_id) - def _validate_create_params(self, agent_id: Optional[int], summary: str, task: str, target_dir: str) -> Optional[str]: + def _validate_create_params(self, agent_id: Optional[int], summary: str, task: str, target_dir: Optional[str], *, multi_agent_mode: bool = False) -> Optional[str]: if agent_id is None: return "子智能体代号不能为空" try: @@ -61,7 +61,8 @@ class SubAgentCreationMixin: return "任务摘要不能为空" if not task or not task.strip(): return "任务详情不能为空" - if target_dir is None: + # 多智能体模式不需要交付目录 + if not multi_agent_mode and target_dir is None: return "指定文件夹不能为空" return None @@ -69,8 +70,11 @@ class SubAgentCreationMixin: suffix = uuid.uuid4().hex[:6] return f"sub_{agent_id}_{int(time.time())}_{suffix}" - def _resolve_deliverables_dir(self, relative_dir: str) -> Path: - relative_dir = relative_dir.strip() if relative_dir else "" + def _resolve_deliverables_dir(self, relative_dir: Optional[str], *, multi_agent_mode: bool = False) -> Path: + relative_dir = (relative_dir or "").strip() + # 多智能体模式:没有交付目录概念,直接使用项目根目录 + if multi_agent_mode and not relative_dir: + return self.project_path.resolve() if not relative_dir: raise ValueError("交付目录不能为空,必须指定") deliverables_path = (self.project_path / relative_dir).resolve() diff --git a/modules/sub_agent/manager.py b/modules/sub_agent/manager.py index 47c7e44..b3139c3 100644 --- a/modules/sub_agent/manager.py +++ b/modules/sub_agent/manager.py @@ -136,7 +136,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi agent_id: int, summary: str, task: str, - deliverables_dir: str, + deliverables_dir: Optional[str] = None, timeout_seconds: Optional[int] = None, conversation_id: Optional[str] = None, run_in_background: bool = False, @@ -145,6 +145,8 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi multi_agent_mode: bool = False, role_id: Optional[str] = None, display_name: Optional[str] = None, + system_prompt: Optional[str] = None, + task_message: Optional[str] = None, ) -> Dict: """创建子智能体任务并启动协程。 @@ -152,7 +154,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi 参数 role_id: 多智能体模式下的角色标诶。 参数 display_name: 多智能体模式下的显示名(如 UI Operator_1)。 """ - validation_error = self._validate_create_params(agent_id, summary, task, deliverables_dir) + validation_error = self._validate_create_params(agent_id, summary, task, deliverables_dir, multi_agent_mode=multi_agent_mode) if validation_error: return {"success": False, "error": validation_error} @@ -181,7 +183,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi task_root.mkdir(parents=True, exist_ok=True) try: - deliverables_path = self._resolve_deliverables_dir(deliverables_dir) + deliverables_path = self._resolve_deliverables_dir(deliverables_dir, multi_agent_mode=multi_agent_mode) except ValueError as exc: return {"success": False, "error": str(exc)} @@ -194,11 +196,17 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi prompt_workspace = self._get_runtime_path(self.project_path) deliverables_display = self._get_runtime_path(deliverables_path) - user_message = build_user_message(agent_id, summary, task, deliverables_display, timeout_seconds or SUB_AGENT_DEFAULT_TIMEOUT) + if task_message: + user_message = task_message + else: + user_message = build_user_message(agent_id, summary, task, deliverables_display, timeout_seconds or SUB_AGENT_DEFAULT_TIMEOUT) task_file.write_text(user_message, encoding="utf-8") - system_prompt = build_system_prompt(prompt_workspace) - system_prompt_file.write_text(system_prompt, encoding="utf-8") + if system_prompt: + final_system_prompt = system_prompt + else: + final_system_prompt = build_system_prompt(prompt_workspace) + system_prompt_file.write_text(final_system_prompt, encoding="utf-8") timeout_seconds = timeout_seconds or SUB_AGENT_DEFAULT_TIMEOUT @@ -250,7 +258,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi manager=self, task_record=task_record, task_message=user_message, - system_prompt=system_prompt, + system_prompt=final_system_prompt, model_key=model_key, thinking_mode=thinking_mode, multi_agent_mode=multi_agent_mode, @@ -360,7 +368,12 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi task_id = task["task_id"] running_task = self._running_tasks.pop(task_id, None) if running_task and not running_task.done(): - running_task.cancel() + # 子智能体运行在独立事件循环线程中,取消操作必须投递到该循环 + try: + loop = running_task.get_loop() + loop.call_soon_threadsafe(running_task.cancel) + except Exception: + running_task.cancel() deadline = time.time() + 5 while not running_task.done() and time.time() < deadline: time.sleep(0.05) @@ -574,7 +587,7 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi sub_agent = self._find_sub_agent_task_by_agent_id(agent_id) if not sub_agent: return False - sub_agent.messages.append({"role": "user", "content": message_text}) + sub_agent.inject_message(message_text) return True def _find_sub_agent_task_by_agent_id(self, agent_id: int) -> Optional[Any]: diff --git a/modules/sub_agent/state.py b/modules/sub_agent/state.py index 07ee4de..181668c 100644 --- a/modules/sub_agent/state.py +++ b/modules/sub_agent/state.py @@ -257,6 +257,11 @@ class SubAgentStateMixin: running_task.result(timeout=0) except Exception: pass + # 多智能体模式:任务自然进入 idle 时不写输出文件,不应标记为失败 + if task.get("multi_agent_mode") and not Path(task.get("output_file", "")).exists(): + task["status"] = "idle" + task["updated_at"] = time.time() + return {"status": "idle", "task_id": task_id} return self._check_task_status(task) return {"status": "running", "task_id": task_id} @@ -264,6 +269,12 @@ class SubAgentStateMixin: if output_file.exists(): return self._check_task_status(task) + # 多智能体模式:没有输出文件表示 idle,不强制清理 + if task.get("multi_agent_mode"): + task["status"] = "idle" + task["updated_at"] = time.time() + return {"status": "idle", "task_id": task_id} + if self._should_force_cleanup_stale_task(task): return self._mark_task_terminated( task, diff --git a/modules/sub_agent/task.py b/modules/sub_agent/task.py index bf20b16..e90b7e7 100644 --- a/modules/sub_agent/task.py +++ b/modules/sub_agent/task.py @@ -4,11 +4,15 @@ import asyncio import base64 import json import mimetypes +import re +import threading import time import uuid from datetime import datetime from pathlib import Path -from typing import Any, Dict, List, Optional, TYPE_CHECKING +from typing import Any, Dict, List, Optional, Set, TYPE_CHECKING + +_QUESTION_ID_RE = re.compile(r"^id:\s*(\S+)", re.MULTILINE) from modules.sub_agent.toolkit import ( SUB_AGENT_TOOLS, @@ -90,8 +94,14 @@ class SubAgentTask: # 多智能体模式相关字段 self.multi_agent_mode = bool(multi_agent_mode) self.multi_agent_state = multi_agent_state - # display_name 不传时回退为 'Agent_{agent_id}' + # display_name 不传时回退为 'Agent_{self.agent_id}' self.display_name = display_name or f"Agent_{self.agent_id}" + # 多智能体运行期控制 + # 使用 threading.Event 避免跨事件循环唤醒问题 + self._continue_event = threading.Event() + self._idle = False + self._pending_answer_question_id: Optional[str] = None + self._answered_question_ids: Set[str] = set() def emit(self, type_: str, data: Dict[str, Any]) -> None: """输出一行 JSONL 到 progress 文件并缓存。""" @@ -123,34 +133,57 @@ class SubAgentTask: if self.multi_agent_mode: tools = list(SUB_AGENT_TOOLS) tools.extend(_load_multi_agent_sub_agent_tools()) - # 多智能体模式下不要求 finish_task,自然输出结束即本轮任务结束 + # 多智能体模式下不要求 finish_task,自然输出结束即进入 idle,可继续接收消息 else: tools = list(SUB_AGENT_TOOLS) tools.append(FINISH_TOOL) start_time = time.time() max_turns = 50 + turn = 0 - for turn in range(1, max_turns + 1): - if self._cancelled: - break + while not self._cancelled: elapsed = time.time() - start_time if elapsed > self.timeout_seconds: await self._write_timeout(elapsed) return + # 多智能体模式下,idle 时等待新消息或外部回答;超时后继续循环检查 + if self.multi_agent_mode and self._idle: + try: + await asyncio.wait_for( + asyncio.get_event_loop().run_in_executor(None, self._continue_event.wait), + timeout=1.0, + ) + except asyncio.TimeoutError: + pass + self._continue_event.clear() + if self._cancelled: + break + self._idle = False + continue + + turn += 1 + if turn > max_turns: + await self._write_failure("任务执行超过最大轮次限制", max_turns_exceeded=True) + return + self.stats["api_calls"] += 1 self.stats["turn_count"] = turn self.stats["runtime_seconds"] = int(elapsed) self.emit("stats", {**self.stats, "turn_count": turn}) + # 多智能体模式:在模型调用前识别是否有待回答的提问 + if self.multi_agent_mode: + self._pending_answer_question_id = self._peek_pending_question_id() + assistant_message, reasoning, tool_calls, usage = await self._call_model(client, model_key, tools) if usage: self._apply_usage(usage) # 多智能体模式:把 assistant 文本输出作为进度/完成 output 转发到主对话 if self.multi_agent_mode and self.multi_agent_state and assistant_message.strip(): - self._forward_output_to_master(assistant_message) + self._forward_output_to_master(assistant_message, is_final=not tool_calls) final_message: Dict[str, Any] = {"role": "assistant", "content": assistant_message} if reasoning: @@ -160,10 +193,11 @@ class SubAgentTask: self.messages.append(final_message) if not tool_calls: - # 多智能体模式:没有 tool_calls 表示本轮结束,进入 idle 状态 + # 多智能体模式:没有 tool_calls 表示本轮结束,进入 idle 等待 if self.multi_agent_mode: self._mark_idle() - return + self._idle = True + continue # 普通模式:prompt 并要求继续 / finish_task self.messages.append({ "role": "user", @@ -197,15 +231,21 @@ class SubAgentTask: "content": content, }) - await self._write_failure("任务执行超过最大轮次限制", max_turns_exceeded=True) + # 循环结束(取消或 idle 被外部终止)后的清理 + if self.multi_agent_mode and self._cancelled: + if self.multi_agent_state: + self.multi_agent_state.mark_status(self.agent_id, "terminated") - def _forward_output_to_master(self, output_text: str) -> None: + def _forward_output_to_master(self, output_text: str, *, is_final: bool = False) -> None: """把子智能体的 assistant 文本输出转发成主对话的 user 消息。""" if not self.multi_agent_state: return + # 如果这是对 pending 提问的回答,不走主对话转发,而是返回到 ask 工具结果 + if self._provide_answer(output_text): + return try: from modules.multi_agent.state import build_sub_agent_output_text - msg = build_sub_agent_output_text(self.display_name, output_text.strip()) + msg = build_sub_agent_output_text(self.display_name, output_text.strip(), is_final=is_final) self.multi_agent_state.push_master_message(msg) # 同时记录到实例状态,供 list_active_sub_agents 使用 inst = self.multi_agent_state.get_instance(self.agent_id) @@ -215,10 +255,43 @@ class SubAgentTask: logger.warning(f"[SubAgentTask] forward output to master failed: {exc}") def _mark_idle(self) -> None: - """多智能体模式下,子智能体自然结束']=本轮任务 结束,进入 idle 状态。""" + """多智能体模式下,子智能体自然结束即本轮任务结束,进入 idle 状态。""" if self.multi_agent_state: self.multi_agent_state.mark_status(self.agent_id, "idle") + def inject_message(self, message_text: str) -> None: + """外部向子智能体上下文插入 user 消息,并唤醒 idle 状态。""" + self.messages.append({"role": "user", "content": message_text}) + self._continue_event.set() + + def _peek_pending_question_id(self) -> Optional[str]: + """检查最后一条 user 消息是否是向本智能体提问,返回 question_id。""" + if not self.multi_agent_mode or not self.messages: + return None + for msg in reversed(self.messages): + if msg.get("role") == "user": + content = msg.get("content") or "" + if "的提问" in content: + m = _QUESTION_ID_RE.search(content) + if m: + qid = m.group(1) + if qid not in self._answered_question_ids: + return qid + break + return None + + def _provide_answer(self, output_text: str) -> bool: + """如果当前输出是对 pending 提问的回答,把回答写回 future 并阻止转发到主对话。""" + if not self._pending_answer_question_id or not self.multi_agent_state: + return False + self.multi_agent_state.provide_answer( + self._pending_answer_question_id, + output_text.strip(), + ) + self._answered_question_ids.add(self._pending_answer_question_id) + self._pending_answer_question_id = None + return True + def _build_client(self) -> tuple: """加载模型配置并初始化 DeepSeekClient。""" config_path = self.manager.models_config_file diff --git a/modules/sub_agent/toolkit.py b/modules/sub_agent/toolkit.py index 79a87a5..55b6422 100644 --- a/modules/sub_agent/toolkit.py +++ b/modules/sub_agent/toolkit.py @@ -6,6 +6,8 @@ import json from datetime import datetime from typing import Any, Dict, List, Optional +from config.model_profiles import _parse_env_ref + # 子智能体可用工具定义(与前端进度展示兼容) SUB_AGENT_TOOLS: List[Dict[str, Any]] = [ { @@ -267,8 +269,8 @@ def _format_tool_result(name: str, raw: Any) -> str: def _build_sub_agent_profile(model_raw: Dict[str, Any]) -> Optional[Dict[str, Any]]: """把 sub_agent_models.json 中的模型条目转成 DeepSeekClient.apply_profile 所需格式。""" name = str(model_raw.get("name") or model_raw.get("model_name") or model_raw.get("model") or "").strip() - url = str(model_raw.get("url") or model_raw.get("base_url") or "").strip() - api_key = str(model_raw.get("apikey") or model_raw.get("api_key") or "").strip() + url = str(_parse_env_ref(model_raw.get("url") or model_raw.get("base_url") or "") or "").strip() + api_key = str(_parse_env_ref(model_raw.get("apikey") or model_raw.get("api_key") or "") or "").strip() if not name or not url or not api_key: return None diff --git a/server/chat_flow_task_main.py b/server/chat_flow_task_main.py index cbefa60..6dd0395 100644 --- a/server/chat_flow_task_main.py +++ b/server/chat_flow_task_main.py @@ -130,7 +130,7 @@ from .chat_flow_runtime import ( detect_malformed_tool_call, ) -from .chat_flow_task_support import process_sub_agent_updates, process_background_command_updates +from .chat_flow_task_support import process_sub_agent_updates, process_background_command_updates, process_multi_agent_master_messages from .chat_flow_tool_loop import execute_tool_calls from .chat_flow_stream_loop import run_streaming_attempts from .deep_compression import run_deep_compression @@ -778,6 +778,25 @@ def _collect_pending_completion_notices(*, web_terminal, conversation_id: str) - "sort_key": update.get("updated_at") or time.time(), }) + # 3) 多智能体模式:把子智能体转发到主对话的 pending 消息也作为通知池项消费 + if getattr(web_terminal, "multi_agent_mode", False): + sub_manager = getattr(web_terminal, "sub_agent_manager", None) + if sub_manager: + state = sub_manager.get_multi_agent_state(conversation_id) + if state: + ma_messages = state.drain_master_messages() + for msg_text in ma_messages: + notices.append({ + "kind": "multi_agent", + "message": msg_text, + "payload": { + "sub_agent_notice": True, + "message_source": "sub_agent", + "multi_agent_output": True, + }, + "sort_key": time.time(), + }) + notices.sort(key=lambda item: item.get("sort_key") or 0) return notices @@ -787,18 +806,37 @@ def _has_pending_completion_work(*, web_terminal, conversation_id: str) -> bool: sub_manager = getattr(web_terminal, "sub_agent_manager", None) if sub_manager: announced = getattr(web_terminal, "_announced_sub_agent_tasks", set()) + has_running_non_ma = False + has_unnotified_non_ma = False + has_running_ma = False for task in sub_manager.tasks.values(): if not isinstance(task, dict): continue - if not task.get("run_in_background"): - continue if task.get("conversation_id") != conversation_id: continue status = task.get("status") + multi_agent_flag = task.get("multi_agent_mode") or False if status not in TERMINAL_STATUSES.union({"terminated"}): - return True # 仍在运行 - if status != "terminated" and (task.get("task_id") not in announced) and not task.get("notified"): - return True # 已完成但未通知 + if multi_agent_flag: + has_running_ma = True + elif task.get("run_in_background"): + has_running_non_ma = True + continue + if not multi_agent_flag and task.get("run_in_background") and (task.get("task_id") not in announced) and not task.get("notified"): + has_unnotified_non_ma = True + if has_running_non_ma or has_unnotified_non_ma: + return True + # 多智能体模式:有未消费的主对话消息 或 有运行中(非 idle)实例时继续轮询 + if getattr(web_terminal, "multi_agent_mode", False) and has_running_ma: + state = sub_manager.get_multi_agent_state(conversation_id) + if state: + if state.has_pending_master_messages(): + return True + # 只要还有非 idle 实例就继续;全部 idle 且无 pending 则结束轮询 + if any(a.status not in {"idle", "terminated"} for a in state.list_all()): + return True + return False + return True bg_manager = getattr(web_terminal, "background_command_manager", None) if bg_manager: try: @@ -854,6 +892,11 @@ async def poll_completion_notifications(*, web_terminal, workspace, conversation await asyncio.sleep(1) continue + # 多智能体模式:主对话任务仍在运行时,由主循环自己消费 pending 消息 + if getattr(web_terminal, "_multi_agent_main_task_active", False): + await asyncio.sleep(1) + continue + notices = _collect_pending_completion_notices( web_terminal=web_terminal, conversation_id=conversation_id, @@ -928,6 +971,9 @@ async def handle_task_with_sender( web_terminal = terminal conversation_id = getattr(web_terminal.context_manager, "current_conversation_id", None) + # 多智能体模式:标记主对话任务正在运行,供后台通知池判断是否可以安全消费 + if getattr(web_terminal, "multi_agent_mode", False): + web_terminal._multi_agent_main_task_active = True videos = videos or [] raw_sender = sender @@ -1638,6 +1684,22 @@ async def handle_task_with_sender( debug_log(f"[Goal] 目标停止:{goal_result.get('reason')}") except Exception as exc: debug_log(f"[Goal] turn 结束处理失败: {exc}") + + # 多智能体模式:没有工具调用时,先消费子智能体待转发到主对话的消息; + # 如果有新消息注入,继续迭代让 Team Leader 响应,而不是直接结束任务。 + if getattr(web_terminal, "multi_agent_mode", False): + injected_count = await process_multi_agent_master_messages( + messages=messages, + inline=False, + web_terminal=web_terminal, + sender=sender, + debug_log=debug_log, + ) + if injected_count: + debug_log(f"[MultiAgent] no-tool-call turn 注入 {injected_count} 条子智能体消息,继续迭代") + is_first_iteration = False + continue + break # 目标模式:本轮段确实产生了工具调用 @@ -1699,6 +1761,7 @@ async def handle_task_with_sender( last_tool_call_time=last_tool_call_time, process_sub_agent_updates=process_sub_agent_updates, process_background_command_updates=process_background_command_updates, + process_multi_agent_master_messages=process_multi_agent_master_messages, maybe_mark_failure_from_message=maybe_mark_failure_from_message, mark_force_thinking=mark_force_thinking, get_stop_flag=get_stop_flag, @@ -1749,8 +1812,22 @@ async def handle_task_with_sender( # 标记不再是第一次迭代 is_first_iteration = False + # 多智能体模式:在进入下一轮模型调用前,消费子智能体最新输出到主对话 + if getattr(web_terminal, "multi_agent_mode", False): + await process_multi_agent_master_messages( + messages=messages, + inline=False, + web_terminal=web_terminal, + sender=sender, + debug_log=debug_log, + ) + # 最终统计 + # 多智能体模式:主对话任务结束,解除活跃标记 + if getattr(web_terminal, "multi_agent_mode", False): + web_terminal._multi_agent_main_task_active = False + debug_log(f"\n{'='*40}") debug_log(f"任务完成统计:") debug_log(f" 总迭代次数: {total_iterations}") diff --git a/server/chat_flow_task_support.py b/server/chat_flow_task_support.py index 5b9c31d..48ff7d8 100644 --- a/server/chat_flow_task_support.py +++ b/server/chat_flow_task_support.py @@ -314,6 +314,110 @@ async def process_background_command_updates(*, messages: List[Dict], inline: bo +def inject_multi_agent_master_message( + *, + web_terminal, + messages, + text: str, + sender, + conversation_id: Optional[str] = None, + inline: bool = True, + after_tool_call_id: Optional[str] = None, +) -> Optional[str]: + """把多智能体子智能体输出/消息以原生格式注入主对话,不添加 [系统通知|xxx] 前缀。""" + raw = "" if text is None else str(text).strip() + if not raw: + return None + + metadata = { + "runtime_injected": True, + "source": "sub_agent", + "message_source": "sub_agent", + "inline": inline, + "is_auto_generated": True, + "auto_message_type": "multi_agent_output", + "visibility": "compact", + "starts_work": False, + } + + try: + ctx_manager = getattr(web_terminal, "context_manager", None) + if ctx_manager is not None: + ctx_manager.add_conversation("user", raw, metadata=metadata) + except Exception: + pass + + if messages is not None: + insert_index = len(messages) + if after_tool_call_id: + for idx, msg in enumerate(messages): + if msg.get("role") == "tool" and msg.get("tool_call_id") == after_tool_call_id: + end = idx + 1 + while end < len(messages) and messages[end].get("role") == "tool": + end += 1 + insert_index = end + break + messages.insert(insert_index, {"role": "user", "content": raw}) + + if callable(sender): + payload = { + "message": raw, + "content": raw, + "conversation_id": conversation_id, + "inline": inline, + "source": "sub_agent", + "message_source": "sub_agent", + "visibility": "compact", + "starts_work": False, + "metadata": metadata, + "runtime_injected": True, + } + try: + sender("user_message", payload) + except Exception: + pass + + return raw + + +async def process_multi_agent_master_messages( + *, + web_terminal, + messages, + sender, + debug_log, + inline: bool = False, + after_tool_call_id: Optional[str] = None, +) -> int: + """从 MultiAgentState 取出待插入主对话的消息并注入。返回注入条数。""" + if not getattr(web_terminal, "multi_agent_mode", False): + return 0 + manager = getattr(web_terminal, "sub_agent_manager", None) + if not manager: + return 0 + conversation_id = getattr(getattr(web_terminal, "context_manager", None), "current_conversation_id", None) + if not conversation_id: + return 0 + state = manager.get_multi_agent_state(conversation_id) + if not state: + return 0 + pending = state.drain_master_messages() + if not pending: + return 0 + debug_log(f"[MultiAgent] draining {len(pending)} pending master messages") + for msg in pending: + inject_multi_agent_master_message( + web_terminal=web_terminal, + messages=messages, + text=msg, + sender=sender, + conversation_id=conversation_id, + inline=inline, + after_tool_call_id=after_tool_call_id, + ) + return len(pending) + + async def wait_retry_delay(*, delay_seconds: int, client_sid: str, username: str, sender, get_stop_flag, clear_stop_flag) -> bool: """等待重试间隔,同时检查是否收到停止请求。""" if delay_seconds <= 0: diff --git a/server/chat_flow_tool_loop.py b/server/chat_flow_tool_loop.py index 3124376..0e3d5a4 100644 --- a/server/chat_flow_tool_loop.py +++ b/server/chat_flow_tool_loop.py @@ -24,7 +24,7 @@ from modules.personalization_manager import load_personalization_config, resolve from modules.auto_approval_service import run_auto_approval from modules.user_question_manager import format_user_question_answer from .deep_compression import run_deep_compression -from .chat_flow_task_support import inject_runtime_user_message +from .chat_flow_task_support import inject_runtime_user_message, process_multi_agent_master_messages def _format_numbered_lines(lines: List[str], start_line_no: int) -> List[Dict[str, Any]]: @@ -261,7 +261,7 @@ async def _wait_for_user_questions(*, question_ids: List[str], username: str, ti return answered -async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, client_sid: str, username: str, iteration: int, conversation_id: Optional[str], last_tool_call_time: float, process_sub_agent_updates, process_background_command_updates, maybe_mark_failure_from_message, mark_force_thinking, get_stop_flag, clear_stop_flag, workspace=None): +async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, client_sid: str, username: str, iteration: int, conversation_id: Optional[str], last_tool_call_time: float, process_sub_agent_updates, process_background_command_updates, process_multi_agent_master_messages=process_multi_agent_master_messages, maybe_mark_failure_from_message, mark_force_thinking, get_stop_flag, clear_stop_flag, workspace=None): previous_tool_loop_active = getattr(web_terminal, "_tool_loop_active", False) web_terminal._tool_loop_active = True allowed_tool_names = set() @@ -1323,6 +1323,14 @@ async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, clie debug_log=debug_log, maybe_mark_failure_from_message=maybe_mark_failure_from_message, ) + await process_multi_agent_master_messages( + messages=messages, + inline=True, + after_tool_call_id=last_completed_tool_call_id, + web_terminal=web_terminal, + sender=sender, + debug_log=debug_log, + ) # 运行期模式通知:必须等待同一轮全部 tool_call 都完成后再注入, # 避免在 assistant.tool_calls 与对应 tool 消息之间插入 user 消息导致 API 报错。 diff --git a/server/multi_agent.py b/server/multi_agent.py index 5c033b8..b7ad90a 100644 --- a/server/multi_agent.py +++ b/server/multi_agent.py @@ -13,19 +13,47 @@ from typing import Any, Dict, List from flask import Blueprint, current_app, jsonify, request, session -from server.auth_helpers import api_login_required, get_current_username +from server.auth_helpers import api_login_required, login_required, get_current_username from server.context import get_user_resources multi_agent_bp = Blueprint("multi_agent", __name__) @multi_agent_bp.route("/multiagent/new") -@api_login_required +@login_required def multi_agent_new_page(): """多智能体模式入口,返回与 /new 相同的 SPA index.html。""" return current_app.send_static_file("index.html") +@multi_agent_bp.route("/multiagent/") +@login_required +def multi_agent_conversation_page(conversation_id: str): + """多智能体模式 指定会话 URL,返回 SPA index.html 让前端路由处理。""" + return current_app.send_static_file("index.html") + + +@multi_agent_bp.route("/api/multiagent/rebuild-index", methods=["POST"]) +@api_login_required +def rebuild_conversation_index_api(): + """强制从磁盘重建对话索引,补全 multi_agent_mode 等新字段。""" + try: + username = get_current_username() + if not username: + return jsonify({"success": False, "error": "未登录"}), 401 + terminal, _ = get_user_resources(username) + if not terminal: + return jsonify({"success": False, "error": "工作区未就绪"}), 503 + cm = getattr(getattr(terminal, "context_manager", None), "conversation_manager", None) + if not cm: + return jsonify({"success": False, "error": "对话管理器未初始化"}), 503 + rebuilt = cm._rebuild_index_from_files() + cm._save_index(rebuilt) + return jsonify({"success": True, "index_size": len(rebuilt)}) + except Exception as exc: + return jsonify({"success": False, "error": str(exc)}), 500 + + @multi_agent_bp.route("/api/multiagent/roles", methods=["GET"]) @api_login_required def list_roles_api(): diff --git a/static/src/app/methods/conversation/load.ts b/static/src/app/methods/conversation/load.ts index d0cbbd3..e65fcff 100644 --- a/static/src/app/methods/conversation/load.ts +++ b/static/src/app/methods/conversation/load.ts @@ -180,6 +180,9 @@ export const loadMethods = { if (typeof result.model_key === 'string' && result.model_key) { this.modelSet(result.model_key); } + if (typeof result.multi_agent_mode === 'boolean') { + this.multiAgentMode = result.multi_agent_mode; + } // 2. 更新当前对话信息 this.skipConversationHistoryReload = true; @@ -193,10 +196,11 @@ export const loadMethods = { if (!preserveListPosition) { this.promoteConversationToTop(conversationId); } + const urlPrefix = this.multiAgentMode ? '/multiagent/' : '/'; history.pushState( { conversationId }, '', - `/${this.stripConversationPrefix(conversationId)}` + `${urlPrefix}${this.stripConversationPrefix(conversationId)}` ); this.skipConversationLoadedEvent = true; diff --git a/static/src/app/methods/ui/route.ts b/static/src/app/methods/ui/route.ts index e221e50..ebfbb6f 100644 --- a/static/src/app/methods/ui/route.ts +++ b/static/src/app/methods/ui/route.ts @@ -41,6 +41,12 @@ export const routeMethods = { this.startTitleTyping('多智能体模式', { animate: false }); this.initialRouteResolved = true; this.refreshBlankHeroState(); + // 进入多智能体模式时触发后端重建索引以补全 multi_agent_mode 字段 + try { + await fetch('/api/multiagent/rebuild-index', { method: 'POST' }); + } catch (_e) { + // 重建失败不阻断主流程 + } // 多智能体模式下自动创建一个带 metadata.multi_agent_mode=true 的新对话 try { const resp = await fetch('/api/multiagent/conversations', { diff --git a/static/src/app/watchers.ts b/static/src/app/watchers.ts index 4929f6f..3fa75c4 100644 --- a/static/src/app/watchers.ts +++ b/static/src/app/watchers.ts @@ -1,7 +1,11 @@ // @ts-nocheck import { debugLog, traceLog } from './methods/common'; +import { useConversationStore } from '../stores/conversation'; export const watchers = { + multiAgentMode(newValue) { + useConversationStore().$patch({ multiAgentMode: !!newValue }); + }, inputMessage() { this.autoResizeInput(); if (typeof this.scheduleComposerDraftPersist === 'function') { diff --git a/static/src/auth/LoginApp.vue b/static/src/auth/LoginApp.vue index c9c7633..a785223 100644 --- a/static/src/auth/LoginApp.vue +++ b/static/src/auth/LoginApp.vue @@ -69,7 +69,7 @@ const submitting = ref(false); const hostSubmitting = ref(false); const hostModeEnabled = ref(false); -const login = async () => { +const doLogin = async (redirectUrl = '/') => { if (!email.value || !password.value) { error.value = '请输入邮箱和密码'; return; @@ -92,7 +92,7 @@ const login = async () => { const data = await resp.json(); if (data.success) { - window.location.href = '/'; + window.location.href = redirectUrl; return; } @@ -104,12 +104,9 @@ const login = async () => { } }; -const enterMultiAgent = () => { - // 直接跳转,后端 @login_required 会确保已登录后重定向回来 - window.location.href = '/multiagent/new'; -}; +const login = () => doLogin('/'); -const hostLogin = async () => { +const doHostLogin = async (redirectUrl = '/') => { hostSubmitting.value = true; error.value = ''; @@ -122,7 +119,7 @@ const hostLogin = async () => { const data = await resp.json(); if (data.success) { - window.location.href = '/'; + window.location.href = redirectUrl; return; } @@ -134,6 +131,20 @@ const hostLogin = async () => { } }; +const hostLogin = () => doHostLogin('/'); + +const enterMultiAgent = () => { + if (hostModeEnabled.value) { + doHostLogin('/multiagent/new'); + return; + } + if (!email.value || !password.value) { + error.value = '请输入邮箱和密码,或选择宿主机模式'; + return; + } + doLogin('/multiagent/new'); +}; + onMounted(async () => { applyTheme(loadTheme()); diff --git a/static/src/stores/conversation.ts b/static/src/stores/conversation.ts index 134b158..e4aab2f 100644 --- a/static/src/stores/conversation.ts +++ b/static/src/stores/conversation.ts @@ -45,6 +45,7 @@ interface ConversationState { runningWorkspaceTasks: any[]; acknowledgedCompletedTaskIds: string[]; workspaceGroups: WorkspaceConversationGroup[]; + multiAgentMode: boolean; } export const useConversationStore = defineStore('conversation', { @@ -70,7 +71,8 @@ export const useConversationStore = defineStore('conversation', { conversationsLimit: 20, runningWorkspaceTasks: [], acknowledgedCompletedTaskIds: [], - workspaceGroups: [] + workspaceGroups: [], + multiAgentMode: false }), actions: { resetConversations() { @@ -192,8 +194,9 @@ export const useConversationStore = defineStore('conversation', { const fetchOffset = refresh ? 0 : group.offset; group.loading = true; try { + const maParam = this.multiAgentMode ? '&multi_agent_mode=1' : '&multi_agent_mode=0'; const response = await fetch( - `/api/conversations?workspace_id=${encodeURIComponent(workspaceId)}&limit=${group.fetchLimit}&offset=${fetchOffset}` + `/api/conversations?workspace_id=${encodeURIComponent(workspaceId)}&limit=${group.fetchLimit}&offset=${fetchOffset}${maParam}` ); const data = await response.json(); if (data.success) { diff --git a/utils/conversation_manager/crud_mixin.py b/utils/conversation_manager/crud_mixin.py index f2f2afb..a984444 100644 --- a/utils/conversation_manager/crud_mixin.py +++ b/utils/conversation_manager/crud_mixin.py @@ -233,7 +233,8 @@ class CrudMixin: "has_videos": conversation_data["metadata"].get("has_videos", False), "total_messages": metadata.total_messages, "total_tools": metadata.total_tools, - "status": metadata.status + "status": metadata.status, + "multi_agent_mode": bool(conversation_data["metadata"].get("multi_agent_mode", False)) } self._save_index(index) diff --git a/utils/conversation_manager/index_mixin.py b/utils/conversation_manager/index_mixin.py index b448e48..b16df1c 100644 --- a/utils/conversation_manager/index_mixin.py +++ b/utils/conversation_manager/index_mixin.py @@ -163,6 +163,7 @@ class IndexMixin: "total_messages": metadata.get("total_messages", 0), "total_tools": metadata.get("total_tools", 0), "status": metadata.get("status", "active"), + "multi_agent_mode": bool(metadata.get("multi_agent_mode", False)), } elapsed_ms = (time.perf_counter() - t0) * 1000 if perf_log: