From faef998a9a2b2dd7bf76ce4a048c4c6daee357f7 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Wed, 15 Jul 2026 17:33:47 +0800 Subject: [PATCH] =?UTF-8?q?perf(debug):=20=E5=A2=9E=E5=BC=BA=E5=86=85?= =?UTF-8?q?=E5=AD=98=E7=9B=91=E6=8E=A7=EF=BC=9A=E6=AF=8F=E7=A7=92=E9=87=87?= =?UTF-8?q?=E6=A0=B7=20+=20=E6=9B=B4=E5=A4=9A=E5=85=B3=E9=94=AE=E8=B7=AF?= =?UTF-8?q?=E5=BE=84=E5=9F=8B=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 改用 psutil 实时 RSS 取代历史最大 ru_maxrss - 新增后台每秒采样线程，检测到 >100MB/s 增长立即记录 - WebTerminal 初始化时自动启动采样 - 增加子智能体恢复加载 conversation.json、httpx 流式/非流式响应、 工具返回结果、模型输出组装等埋点 --- core/web_terminal.py | 4 + modules/memory_debug.py | 94 ++++++++++-- modules/sub_agent/manager.py | 251 ++++++++++++++++++++++++++++++++- modules/sub_agent/task.py | 18 +++ utils/api_client/chat_mixin.py | 18 +++ 5 files changed, 371 insertions(+), 14 deletions(-) diff --git a/core/web_terminal.py b/core/web_terminal.py index 7cbd9f8..fe717c9 100644 --- a/core/web_terminal.py +++ b/core/web_terminal.py @@ -20,6 +20,7 @@ except ImportError: sys.path.insert(0, str(project_root)) from config import MAX_TERMINALS, TERMINAL_BUFFER_SIZE, TERMINAL_DISPLAY_SIZE from modules.terminal_manager import TerminalManager +from modules.memory_debug import start_periodic_sampling if TYPE_CHECKING: from modules.user_container_manager import ContainerHandle @@ -101,6 +102,9 @@ class WebTerminal(MainTerminal): print(f"[WebTerminal] 实时token统计已启用") else: print(f"[WebTerminal] 警告：message_callback为None，无法启用实时token统计") + + # 启动后台内存采样，帮助定位偶发内存暴涨 + start_periodic_sampling() # =========================================== # 新增：对话管理相关方法（Web版本） # =========================================== diff --git a/modules/memory_debug.py b/modules/memory_debug.py index 0788b85..20d8d1c 100644 --- a/modules/memory_debug.py +++ b/modules/memory_debug.py @@ -51,21 +51,34 @@ def _resolve_log_path() -> Path: return _log_path +_peak_rss_mb: Optional[float] = None +_last_sample_rss_mb: Optional[float] = None +_sample_thread: Optional[threading.Thread] = None + + def _current_rss_mb() -> Optional[float]: - """当前进程 RSS（MB），失败返回 None。""" + """当前进程 RSS（MB），优先使用 psutil（实时），失败回退到 resource。""" + global _peak_rss_mb + try: + import psutil + proc = psutil.Process() + rss = proc.memory_info().rss / (1024 * 1024) + if _peak_rss_mb is None or rss > _peak_rss_mb: + _peak_rss_mb = rss + return rss + except Exception: + pass try: import resource usage = resource.getrusage(resource.RUSAGE_SELF) # macOS: ru_maxrss 单位是 bytes；Linux: 单位是 KB if sys.platform == "darwin": - return usage.ru_maxrss / (1024 * 1024) - return usage.ru_maxrss / 1024.0 - except Exception: - pass - try: - import psutil - proc = psutil.Process() - return proc.memory_info().rss / (1024 * 1024) + rss = usage.ru_maxrss / (1024 * 1024) + else: + rss = usage.ru_maxrss / 1024.0 + if _peak_rss_mb is None or rss > _peak_rss_mb: + _peak_rss_mb = rss + return rss except Exception: return None @@ -231,3 +244,66 @@ class TopAllocator: tracemalloc.stop() except Exception: pass + + +def _periodic_sampling( + interval_seconds: float = 1.0, + spike_threshold_mb: float = 100.0, + max_records: int = 3, +) -> None: + """后台线程：每秒采样一次内存，发现快速增长或峰值时记录。""" + global _last_sample_rss_mb, _peak_rss_mb + records_since_spike = 0 + while ENABLED: + try: + info = get_memory_info() + rss = info.get("rss_mb") + vms = info.get("vms_mb") + if rss is not None: + spike = False + if _last_sample_rss_mb is not None: + delta = rss - _last_sample_rss_mb + if delta >= spike_threshold_mb: + spike = True + records_since_spike = max_records + _last_sample_rss_mb = rss + if records_since_spike > 0: + records_since_spike -= 1 + log_memory_event( + "memory_periodic_spike", + rss_mb=rss, + vms_mb=vms, + peak_rss_mb=_peak_rss_mb, + delta_from_last_mb=(delta if spike else None), + ) + else: + # 每 5 秒记录一次常规采样，避免日志过多 + if int(time.time()) % 5 == 0: + log_memory_event( + "memory_periodic_sample", + rss_mb=rss, + vms_mb=vms, + peak_rss_mb=_peak_rss_mb, + ) + except Exception: + pass + time.sleep(interval_seconds) + + +def start_periodic_sampling( + interval_seconds: float = 1.0, + spike_threshold_mb: float = 100.0, +) -> None: + """启动后台内存采样线程。幂等：重复调用不会启动多个线程。""" + global _sample_thread + if not ENABLED: + return + if _sample_thread is not None and _sample_thread.is_alive(): + return + _sample_thread = threading.Thread( + target=_periodic_sampling, + args=(interval_seconds, spike_threshold_mb), + name="memory-debug-sampler", + daemon=True, + ) + _sample_thread.start() diff --git a/modules/sub_agent/manager.py b/modules/sub_agent/manager.py index 3027136..dcf99bc 100644 --- a/modules/sub_agent/manager.py +++ b/modules/sub_agent/manager.py @@ -312,7 +312,13 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi multi_agent_mode=multi_agent_mode, ) self._running_tasks.pop(task_id, None) - self._sub_agent_instances.pop(agent_id, None) + # 多智能体模式下 failed 视为可复活状态，保留实例引用供后续 send_message_to_sub_agent 重新激活 + if multi_agent_mode: + final_task = self.tasks.get(task_id) or {} + if final_task.get("status") != "failed": + self._sub_agent_instances.pop(agent_id, None) + else: + self._sub_agent_instances.pop(agent_id, None) self.reconcile_task_states(conversation_id=conversation_id) # 多智能体模式：结束时把状态写回 MultiAgentState if multi_agent_mode and multi_agent_state: @@ -452,6 +458,59 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi ) return count + def stop_sub_agent( + self, + *, + task_id: Optional[str] = None, + agent_id: Optional[int] = None, + ) -> Dict: + """暂停指定子智能体，使其进入 idle 状态而不终结。""" + task = self._select_task(task_id, agent_id) + if not task: + return {"success": False, "error": "未找到对应的子智能体任务"} + + real_task_id = task["task_id"] + real_agent_id = task.get("agent_id") + if not task.get("multi_agent_mode"): + return {"success": False, "error": "stop_sub_agent 仅在多智能体模式下可用"} + + # 查找或复活实例，确保能接收软停止信号 + if real_agent_id is not None: + sub_agent = self._find_or_revive_sub_agent_task(real_agent_id) + else: + sub_agent = None + + if sub_agent and hasattr(sub_agent, "request_soft_stop"): + try: + sub_agent.request_soft_stop() + except Exception as exc: + return {"success": False, "error": f"暂停子智能体失败: {exc}"} + else: + # 没有活实例时直接修改任务状态为 idle + task["status"] = "idle" + task["updated_at"] = time.time() + self._save_state() + + # 同步更新 MultiAgentState + conversation_id = task.get("conversation_id") + if conversation_id: + state = self.get_multi_agent_state(conversation_id) + if state and real_agent_id is not None: + state.mark_status(real_agent_id, "idle") + + ma_debug( + "manager_stop_sub_agent", + task_id=real_task_id, + agent_id=real_agent_id, + had_instance=bool(sub_agent), + ) + return { + "success": True, + "task_id": real_task_id, + "agent_id": real_agent_id, + "message": f"子智能体{real_agent_id} 已暂停，可用 send_message_to_sub_agent 重新激活。", + } + def terminate_sub_agent( self, *, @@ -659,6 +718,11 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi # 其余工具直接走主进程 handle_tool_call，自然经过沙箱/容器/权限链路 result_text = await self.terminal.handle_tool_call(tool_name, arguments) + log_memory_event( + "sub_agent_tool_result", + tool_name=tool_name, + result_chars=len(result_text), + ) try: return json.loads(result_text) except Exception: @@ -702,8 +766,27 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi continue try: - conversation_data = json.loads(conversation_file.read_text(encoding="utf-8")) + log_memory_event( + "sub_agent_restore_load_conversation_start", + task_id=task_id, + agent_id=task.get("agent_id"), + conversation_file=str(conversation_file), + ) + conversation_text = conversation_file.read_text(encoding="utf-8") + log_memory_event( + "sub_agent_restore_after_read", + task_id=task_id, + agent_id=task.get("agent_id"), + conversation_file_chars=len(conversation_text), + ) + conversation_data = json.loads(conversation_text) messages = list(conversation_data.get("messages") or []) + log_memory_event( + "sub_agent_restore_after_parse", + task_id=task_id, + agent_id=task.get("agent_id"), + messages_count=len(messages), + ) except Exception as exc: logger.warning(f"[restore] 读取任务 {task_id} 对话文件失败: {exc}") continue @@ -925,12 +1008,14 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi def inject_message_to_sub_agent(self, agent_id: int, message_text: str) -> bool: """同事件循环中向子智能体上下文插入 user 消息。 - + 适用于 ask_other_agent / send_message_to_sub_agent / answer_sub_agent_question_ （非阻塞到工具结果的路径）。返回 True 表示成功注入。 + 若内存中无运行实例（如 failed 后保留的实例已结束），会尝试从 conversation + 文件重建子智能体（保留原 agent_id 和 role_id）后再注入消息。 """ - # 查找该 agent_id 对应的 running SubAgentTask - sub_agent = self._find_sub_agent_task_by_agent_id(agent_id) + # 查找或复活该 agent_id 对应的 SubAgentTask + sub_agent = self._find_or_revive_sub_agent_task(agent_id) ma_debug( "manager_inject_message_to_sub_agent", agent_id=agent_id, @@ -943,6 +1028,162 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi sub_agent.inject_message(message_text) return True + def _find_or_revive_sub_agent_task(self, agent_id: int) -> Optional[Any]: + """查找内存中的 SubAgentTask；不存在或已结束时从磁盘复活（多智能体模式）。""" + inst = self._find_sub_agent_task_by_agent_id(agent_id) + if inst is not None: + task = getattr(inst, "_task", None) + if task is None or not task.done(): + return inst + # 实例存在但已结束，需要复活前先清理旧引用 + self._sub_agent_instances.pop(agent_id, None) + revived = self._revive_sub_agent(agent_id) + return revived + + def _revive_sub_agent(self, agent_id: int) -> Optional[Any]: + """从 conversation.json 重建一个多智能体子智能体实例（保留原 agent_id/role_id）。 + + 用于 failed/idle 等可复活状态被 send_message_to_sub_agent 重新激活的场景。 + """ + from modules.sub_agent.task import SubAgentTask + + candidates = [ + t for t in self.tasks.values() + if isinstance(t, dict) and t.get("agent_id") == agent_id and t.get("multi_agent_mode") + ] + if not candidates: + return None + # 按创建时间取最新一条 + candidates.sort(key=lambda item: item.get("created_at", 0), reverse=True) + task = candidates[0] + task_id = task.get("task_id") + if not task_id: + return None + # 已在运行中则不重复重建 + if task_id in self._running_tasks: + existing_inst = self._sub_agent_instances.get(agent_id) + if existing_inst is not None: + return existing_inst + + task_root = Path(task.get("task_root", "")) + conversation_file = Path(task.get("conversation_file", "")) + system_prompt_file = task_root / "system_prompt.txt" + task_message_file = task_root / "task.txt" + + if not conversation_file.exists(): + logger.warning(f"[revive] 任务 {task_id} 的对话文件缺失，无法复活") + return None + + try: + conversation_data = json.loads(conversation_file.read_text(encoding="utf-8")) + messages = list(conversation_data.get("messages") or []) + except Exception as exc: + logger.warning(f"[revive] 读取任务 {task_id} 对话文件失败: {exc}") + return None + + system_prompt = "" + if system_prompt_file.exists(): + try: + system_prompt = system_prompt_file.read_text(encoding="utf-8") + except Exception: + pass + + task_message = "" + if task_message_file.exists(): + try: + task_message = task_message_file.read_text(encoding="utf-8") + except Exception: + pass + + if not messages: + messages = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": task_message}, + ] + + conversation_id = task.get("conversation_id") + multi_agent_state = None + if conversation_id: + multi_agent_state = self.get_or_create_multi_agent_state(conversation_id) + if multi_agent_state and not multi_agent_state.get_instance(agent_id): + from modules.multi_agent.state import AgentInstance + inst = AgentInstance( + agent_id=agent_id, + role_id=task.get("role_id") or "", + display_name=task.get("display_name") or f"Agent_{agent_id}", + task_id=task_id, + status="idle", + summary=task.get("summary", ""), + ) + try: + multi_agent_state.register_instance(inst) + except ValueError: + pass + + sub_agent = SubAgentTask( + manager=self, + task_record=task, + task_message=task_message, + system_prompt=system_prompt, + model_key=task.get("model_key"), + thinking_mode=task.get("thinking_mode") or "fast", + multi_agent_mode=True, + multi_agent_state=multi_agent_state, + display_name=task.get("display_name"), + ) + sub_agent.messages = messages + sub_agent._idle = True + task["status"] = "idle" + task["updated_at"] = time.time() + if multi_agent_state: + multi_agent_state.mark_status(agent_id, "idle") + # 同步落盘 output.json + try: + output_file = Path(task.get("output_file", "")) + if output_file.exists(): + output_data = json.loads(output_file.read_text(encoding="utf-8")) + else: + output_data = {} + output_data["status"] = "idle" + output_data["success"] = None + output_file.parent.mkdir(parents=True, exist_ok=True) + output_file.write_text(json.dumps(output_data, ensure_ascii=False), encoding="utf-8") + except Exception as exc: + logger.warning(f"[revive] 更新任务 {task_id} output 文件失败: {exc}") + + task_coro = sub_agent.run() + asyncio_task = self._run_coro(task_coro) + sub_agent._task = asyncio_task + self._running_tasks[task_id] = asyncio_task + self._sub_agent_instances[agent_id] = sub_agent + + def _on_done(fut, tid=task_id, aid=agent_id, state=multi_agent_state, sa=sub_agent): + try: + self._running_tasks.pop(tid, None) + # failed 保留实例供复活；其余终态清理 + if state: + final_task = self.tasks.get(tid) or {} + if final_task.get("status") != "failed": + self._sub_agent_instances.pop(aid, None) + else: + self._sub_agent_instances.pop(aid, None) + self.reconcile_task_states(conversation_id=conversation_id) + if state: + self._on_multi_agent_task_done(tid, aid, state, sa) + except Exception as exc: + logger.exception(f"[SubAgent] revived task {tid} 完成回调异常: {exc}") + ma_debug("manager_revive_on_done_exception", task_id=tid, agent_id=aid, error=str(exc)) + + asyncio_task.add_done_callback(_on_done) + ma_debug( + "manager_revive_sub_agent", + task_id=task_id, + agent_id=agent_id, + display_name=task.get("display_name"), + message_count=len(messages), + ) + return sub_agent + def _find_sub_agent_task_by_agent_id(self, agent_id: int) -> Optional[Any]: """通过遍历创建中的 task 查找活 SubAgentTask 实例。 diff --git a/modules/sub_agent/task.py b/modules/sub_agent/task.py index d7b5b40..f6f3c3b 100644 --- a/modules/sub_agent/task.py +++ b/modules/sub_agent/task.py @@ -638,6 +638,15 @@ class SubAgentTask: if chunk.get("usage"): usage = chunk["usage"] + log_memory_event( + "sub_agent_call_model_assembled", + task_id=self.task_id, + agent_id=self.agent_id, + display_name=self.display_name, + assistant_message_chars=len(assistant_message), + reasoning_chars=len(reasoning), + tool_calls_count=len(tool_calls), + ) return assistant_message, reasoning, tool_calls, usage def _parse_args(self, tool_call: Dict[str, Any]) -> Dict[str, Any]: @@ -1120,8 +1129,17 @@ class SubAgentTask: idle=self._idle, cancelled=self._cancelled, ) + if timeout: + status = "timeout" + elif max_turns_exceeded: + status = "failed" + elif success: + status = "completed" + else: + status = "failed" output_data = { "success": success, + "status": status, "summary": summary, "timeout": timeout, "max_turns_exceeded": max_turns_exceeded, diff --git a/utils/api_client/chat_mixin.py b/utils/api_client/chat_mixin.py index 909a0e2..b1d2400 100644 --- a/utils/api_client/chat_mixin.py +++ b/utils/api_client/chat_mixin.py @@ -197,7 +197,15 @@ class DeepSeekClientChatMixin: yield {"error": self.last_error_info} return + chunk_count = 0 async for line in response.aiter_lines(): + chunk_count += 1 + if chunk_count % 100 == 0: + log_memory_event( + "api_client_stream_chunk", + model_key=self.model_key, + chunk_count=chunk_count, + ) if line.startswith("data:"): json_str = line[5:].strip() if json_str == "[DONE]": @@ -214,6 +222,16 @@ class DeepSeekClientChatMixin: json=payload, headers=headers ) + try: + response_text = response.text + except Exception: + response_text = "" + log_memory_event( + "api_client_nonstream_response", + model_key=self.model_key, + status_code=response.status_code, + response_chars=len(response_text), + ) if response.status_code != 200: error_text = response.text self.last_error_info = {