From d4c3e731346a1f9140a270d47821f9eac1dd4c82 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Fri, 17 Jul 2026 14:05:35 +0800 Subject: [PATCH] =?UTF-8?q?fix(sub=5Fagent):=20=E7=A7=BB=E9=99=A4=20search?= =?UTF-8?q?=5Fworkspace=20=E5=B7=A5=E5=85=B7=E5=B9=B6=E6=B8=85=E7=90=86?= =?UTF-8?q?=E5=86=85=E5=AD=98=E8=B0=83=E8=AF=95=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 删除子智能体 search_workspace 工具定义、路由、formatter 与测试 - 传统/多智能体模式共用 SUB_AGENT_TOOLS，一处删除同时生效 - 老对话恢复后再调用 search_workspace 会返回 未知工具 错误 - 清理多智能体排查期间加入的 memory_debug 模块及所有调用点 （modules/memory_debug.py 删除，server/ core/ utils/ modules/sub_agent/ 中相关埋点全部移除） - 顺手删除 SubAgentActivityDialog 中 search_workspace 的历史显示文案 --- core/main_terminal_parts/context/messages.py | 15 - core/web_terminal.py | 4 - modules/memory_debug.py | 336 ------------------ modules/sub_agent/manager.py | 48 +-- modules/sub_agent/task.py | 90 +---- modules/sub_agent/toolkit.py | 38 -- modules/sub_agent/tools.py | 223 +----------- prompts/sub_agent/system.txt | 3 +- server/chat_flow_task_main.py | 15 - server/chat_flow_task_support.py | 16 - server/deep_compression.py | 16 - .../overlay/SubAgentActivityDialog.vue | 4 - test/test_sub_agent_regression.py | 58 --- utils/api_client/chat_mixin.py | 31 -- 14 files changed, 14 insertions(+), 883 deletions(-) delete mode 100644 modules/memory_debug.py diff --git a/core/main_terminal_parts/context/messages.py b/core/main_terminal_parts/context/messages.py index c6879bf..1c6a4ba 100644 --- a/core/main_terminal_parts/context/messages.py +++ b/core/main_terminal_parts/context/messages.py @@ -80,7 +80,6 @@ from utils.context_manager import ContextManager, AUTO_SHALLOW_PLACEHOLDER from utils.host_workspace_debug import write_host_workspace_debug from utils.tool_result_formatter import format_tool_result_for_context from utils.logger import setup_logger -from modules.memory_debug import log_memory_event, estimate_messages_size from config.model_profiles import ( get_model_profile, get_model_prompt_replacements, @@ -98,12 +97,6 @@ class MessagesMixin: def build_messages(self, context: Dict, user_input: str) -> List[Dict]: """构建消息列表（添加终端内容注入）""" - log_memory_event( - "build_messages_enter", - terminal_id=id(self), - conversation_id=getattr(getattr(self, "context_manager", None), "current_conversation_id", None), - conversation_size=estimate_messages_size(context.get("conversation")), - ) try: file_tree_preview = (context.get("project_info", {}).get("file_tree") or "").splitlines() write_host_workspace_debug( @@ -423,12 +416,4 @@ class MessagesMixin: print(f"[ContextCompression] build_messages 替换tool占位符: {replaced_tool_count} 条") if deep_compacted_skipped: print(f"[ContextCompression] build_messages 跳过已深压缩消息: {deep_compacted_skipped} 条") - log_memory_event( - "build_messages_exit", - terminal_id=id(self), - conversation_id=getattr(getattr(self, "context_manager", None), "current_conversation_id", None), - conversation_size=estimate_messages_size(context.get("conversation")), - result_messages_size=estimate_messages_size(messages), - deep_compacted_skipped=deep_compacted_skipped, - ) return messages diff --git a/core/web_terminal.py b/core/web_terminal.py index fe717c9..7cbd9f8 100644 --- a/core/web_terminal.py +++ b/core/web_terminal.py @@ -20,7 +20,6 @@ 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 @@ -102,9 +101,6 @@ 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 deleted file mode 100644 index c5d7d1e..0000000 --- a/modules/memory_debug.py +++ /dev/null @@ -1,336 +0,0 @@ -"""内存调试工具：在关键路径记录内存变化，帮助定位暴涨点。 - -使用方式： - from modules.memory_debug import mem_snapshot, log_memory_event, get_memory_info - - # 简单记录 - log_memory_event("my_event", foo="bar") - - # 上下文：记录进入/退出内存差 - with mem_snapshot("deep_compress", conversation_id=conv_id): - await run_deep_compression(...) - -输出： - ~/.astrion/astrion//logs/memory_debug.log（JSONL） -""" -from __future__ import annotations - -import json -import os -import sys -import threading -import time -import tracemalloc -from contextlib import asynccontextmanager, contextmanager -from pathlib import Path -from typing import Any, AsyncIterator, Dict, Optional - -from utils.log_rotation import append_line - -# 开关：默认关闭，避免生产环境大量写日志。 -# 用户需要排查时设置环境变量 ASTRION_MEMORY_DEBUG=1 开启。 -ENABLED = str(os.environ.get("ASTRION_MEMORY_DEBUG", "1")).strip().lower() in { - "1", "true", "yes", "on", -} - -_log_lock = threading.Lock() -_log_path: Optional[Path] = None - - -def _resolve_log_path() -> Path: - """解析内存调试日志路径，复用项目运行态 logs 目录。""" - global _log_path - if _log_path is not None: - return _log_path - try: - from config import LOGS_DIR - base = Path(LOGS_DIR) - except Exception: - base = Path.home() / ".astrion" / "astrion" / "host" / "logs" - _log_path = base / "memory_debug.log" - 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），优先使用 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": - 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 - - -def _current_vms_mb() -> Optional[float]: - """当前进程 VMS（MB），失败返回 None。""" - try: - import psutil - proc = psutil.Process() - return proc.memory_info().vms / (1024 * 1024) - except Exception: - return None - - -def _get_detailed_memory_info() -> Dict[str, Optional[float]]: - """获取更详细的内存指标，更接近 macOS 活动监视器。""" - result: Dict[str, Optional[float]] = { - "rss_mb": _current_rss_mb(), - "vms_mb": _current_vms_mb(), - "uss_mb": None, - "shared_mb": None, - "private_mb": None, - "data_mb": None, - "dirty_mb": None, - "compressed_mb": None, - } - try: - import psutil - proc = psutil.Process() - info = proc.memory_info() - # uss 只在 memory_full_info() 中提供 - try: - full = proc.memory_full_info() - result["uss_mb"] = getattr(full, "uss", None) / (1024 * 1024) if getattr(full, "uss", None) else None - result["compressed_mb"] = getattr(full, "compressed", None) / (1024 * 1024) if getattr(full, "compressed", None) else None - except Exception: - pass - result["shared_mb"] = getattr(info, "shared", None) / (1024 * 1024) if getattr(info, "shared", None) else None - result["private_mb"] = getattr(info, "private", None) / (1024 * 1024) if getattr(info, "private", None) else None - result["data_mb"] = getattr(info, "data", None) / (1024 * 1024) if getattr(info, "data", None) else None - result["dirty_mb"] = getattr(info, "dirty", None) / (1024 * 1024) if getattr(info, "dirty", None) else None - except Exception: - pass - return result - - -def get_memory_info() -> Dict[str, Optional[float]]: - return _get_detailed_memory_info() - - -def _estimate_text_size(text: Any) -> int: - """估算文本/列表/字典在 JSON 序列化后的大小（字符数）。""" - try: - return len(json.dumps(text, ensure_ascii=False)) - except Exception: - return len(str(text)) - - -def estimate_messages_size(messages: Any) -> Dict[str, Any]: - """估算消息列表的规模和体积。""" - if not messages: - return {"count": 0, "total_chars": 0, "avg_chars": 0} - if not isinstance(messages, list): - return {"count": 0, "total_chars": _estimate_text_size(messages), "avg_chars": 0} - total = 0 - for m in messages: - total += _estimate_text_size(m) - return { - "count": len(messages), - "total_chars": total, - "avg_chars": total // max(len(messages), 1), - } - - -def log_memory_event(event: str, **kwargs: Any) -> None: - """写一条内存调试日志。""" - if not ENABLED: - return - try: - info = get_memory_info() - payload: Dict[str, Any] = { - "t": time.time(), - "event": event, - "rss_mb": info.get("rss_mb"), - "vms_mb": info.get("vms_mb"), - } - # 过滤掉不可序列化的对象 - for k, v in kwargs.items(): - try: - json.dumps({k: v}, ensure_ascii=False) - payload[k] = v - except Exception: - payload[k] = str(v)[:500] - append_line(_resolve_log_path(), json.dumps(payload, ensure_ascii=False)) - except Exception: - pass - - -@contextmanager -def mem_snapshot(event: str, **kwargs: Any): - """同步上下文管理器：记录进入和退出时的内存变化。""" - if not ENABLED: - yield - return - - start_info = get_memory_info() - start_t = time.time() - log_memory_event(f"{event}_enter", duration_ms=0, **kwargs) - try: - yield - finally: - end_info = get_memory_info() - delta_rss = None - delta_vms = None - if start_info.get("rss_mb") is not None and end_info.get("rss_mb") is not None: - delta_rss = round(end_info["rss_mb"] - start_info["rss_mb"], 2) - if start_info.get("vms_mb") is not None and end_info.get("vms_mb") is not None: - delta_vms = round(end_info["vms_mb"] - start_info["vms_mb"], 2) - log_memory_event( - f"{event}_exit", - duration_ms=round((time.time() - start_t) * 1000, 2), - delta_rss_mb=delta_rss, - delta_vms_mb=delta_vms, - start_rss_mb=start_info.get("rss_mb"), - end_rss_mb=end_info.get("rss_mb"), - **kwargs, - ) - - -@asynccontextmanager -async def async_mem_snapshot(event: str, **kwargs: Any) -> AsyncIterator[None]: - """异步上下文管理器：记录进入和退出时的内存变化。""" - if not ENABLED: - yield - return - - start_info = get_memory_info() - start_t = time.time() - log_memory_event(f"{event}_enter", duration_ms=0, **kwargs) - try: - yield - finally: - end_info = get_memory_info() - delta_rss = None - delta_vms = None - if start_info.get("rss_mb") is not None and end_info.get("rss_mb") is not None: - delta_rss = round(end_info["rss_mb"] - start_info["rss_mb"], 2) - if start_info.get("vms_mb") is not None and end_info.get("vms_mb") is not None: - delta_vms = round(end_info["vms_mb"] - start_info["vms_mb"], 2) - log_memory_event( - f"{event}_exit", - duration_ms=round((time.time() - start_t) * 1000, 2), - delta_rss_mb=delta_rss, - delta_vms_mb=delta_vms, - start_rss_mb=start_info.get("rss_mb"), - end_rss_mb=end_info.get("rss_mb"), - **kwargs, - ) - - -class TopAllocator: - """基于 tracemalloc 的 Top 分配点采样（开销较大，按需显式调用）。""" - - def __init__(self, top_n: int = 10): - self.top_n = top_n - - def start(self) -> None: - try: - tracemalloc.start() - except Exception: - pass - - def snapshot(self, label: str = "") -> None: - try: - snap = tracemalloc.take_snapshot() - top = snap.statistics("lineno")[: self.top_n] - lines = [] - for stat in top: - lines.append( - f"{stat.size / (1024 * 1024):.2f}MB {stat.count} {stat.traceback.format()[-1]}" - ) - log_memory_event( - "tracemalloc_top", - label=label, - top=lines, - ) - except Exception: - pass - - def stop(self) -> None: - try: - tracemalloc.stop() - except Exception: - pass - - -def _periodic_sampling( - interval_seconds: float = 0.1, - spike_threshold_mb: float = 50.0, - max_records: int = 10, -) -> None: - """后台线程：每 100ms 采样一次内存，发现快速增长或峰值时记录。""" - global _last_sample_rss_mb, _peak_rss_mb - records_since_spike = 0 - last_regular_log = 0.0 - while ENABLED: - try: - info = get_memory_info() - rss = info.get("rss_mb") - if rss is not None: - spike = False - delta = None - 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", - **info, - delta_from_last_mb=delta, - ) - else: - now = time.time() - if now - last_regular_log >= 2.0: - last_regular_log = now - log_memory_event( - "memory_periodic_sample", - **info, - ) - except Exception: - pass - time.sleep(interval_seconds) - - -def start_periodic_sampling( - interval_seconds: float = 0.1, - spike_threshold_mb: float = 50.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 dcf99bc..76ecc3a 100644 --- a/modules/sub_agent/manager.py +++ b/modules/sub_agent/manager.py @@ -24,12 +24,11 @@ from config import ( from utils.logger import setup_logger from modules.sub_agent.task import SubAgentTask from modules.sub_agent.prompts import build_user_message, build_system_prompt -from modules.sub_agent.tools import handle_search_workspace, handle_read_mediafile +from modules.sub_agent.tools import handle_read_mediafile from modules.sub_agent.state import SubAgentStateMixin from modules.sub_agent.stats import SubAgentStatsMixin from modules.sub_agent.creation import SubAgentCreationMixin from modules.multi_agent.debug_logger import ma_debug -from modules.memory_debug import log_memory_event from server.utils_common import debug_log if TYPE_CHECKING: @@ -285,16 +284,6 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi conversation_id=conversation_id, state_id=id(multi_agent_state) if multi_agent_state else None, ) - log_memory_event( - "sub_agent_manager_create_sub_agent", - task_id=task_id, - agent_id=agent_id, - conversation_id=conversation_id, - display_name=display_name, - multi_agent_mode=multi_agent_mode, - total_tasks=len(self.tasks), - running_tasks=len(self._running_tasks), - ) task_coro = sub_agent.run() asyncio_task = self._run_coro(task_coro) sub_agent._task = asyncio_task @@ -304,13 +293,6 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi def _on_done(fut): try: - log_memory_event( - "sub_agent_task_done_callback", - task_id=task_id, - agent_id=agent_id, - conversation_id=conversation_id, - multi_agent_mode=multi_agent_mode, - ) self._running_tasks.pop(task_id, None) # 多智能体模式下 failed 视为可复活状态，保留实例引用供后续 send_message_to_sub_agent 重新激活 if multi_agent_mode: @@ -711,18 +693,11 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi try: # 多智能体模式常见问答工具已在 SubAgentTask._execute_multi_agent_tool 中处理 # 这里只处理实际通过主进程执行的工具 - if tool_name == "search_workspace": - return await handle_search_workspace(self.project_path, self.terminal, arguments) if tool_name == "read_mediafile": return await handle_read_mediafile(self.project_path, arguments) # 其余工具直接走主进程 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: @@ -766,27 +741,8 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi continue try: - 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) + conversation_data = json.loads(conversation_file.read_text(encoding="utf-8")) 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 diff --git a/modules/sub_agent/task.py b/modules/sub_agent/task.py index f6f3c3b..29e7bd0 100644 --- a/modules/sub_agent/task.py +++ b/modules/sub_agent/task.py @@ -28,7 +28,6 @@ if TYPE_CHECKING: from modules.multi_agent.state import MultiAgentState from modules.multi_agent.debug_logger import ma_debug -from modules.memory_debug import async_mem_snapshot, log_memory_event, estimate_messages_size logger = setup_logger(__name__) @@ -121,15 +120,6 @@ class SubAgentTask: self._pending_answer_question_id: Optional[str] = None self._answered_question_ids: Set[str] = set() - log_memory_event( - "sub_agent_task_created", - task_id=self.task_id, - agent_id=self.agent_id, - display_name=self.display_name, - multi_agent_mode=self.multi_agent_mode, - initial_messages_size=estimate_messages_size(self.messages), - ) - def emit(self, type_: str, data: Dict[str, Any]) -> None: """输出一行 JSONL 到 progress 文件并缓存。""" line = json.dumps({"type": type_, **data}, ensure_ascii=False) @@ -273,30 +263,13 @@ class SubAgentTask: pending_answer_question_id=self._pending_answer_question_id, ) - async with async_mem_snapshot( - "sub_agent_call_model", - task_id=self.task_id, - agent_id=self.agent_id, - display_name=self.display_name, - turn=turn, - messages_size=estimate_messages_size(self.messages), - ): - assistant_message, reasoning, tool_calls, usage = await self._call_model(client, model_key, tools) + assistant_message, reasoning, tool_calls, usage = await self._call_model(client, model_key, tools) if usage: self._apply_usage(usage) # 上下文压缩检查：超过阈值时触发深度压缩 if self.current_context_tokens > 0 and self.current_context_tokens >= self.compress_threshold_tokens: - with mem_snapshot( - "sub_agent_deep_compress", - task_id=self.task_id, - agent_id=self.agent_id, - display_name=self.display_name, - turn=turn, - messages_size=estimate_messages_size(self.messages), - current_context_tokens=self.current_context_tokens, - ): - compressed = self._deep_compress_messages() + compressed = self._deep_compress_messages() if compressed: ma_debug( "sub_agent_deep_compressed", @@ -414,15 +387,6 @@ class SubAgentTask: output_len=len(output_text), is_final=is_final, ) - log_memory_event( - "sub_agent_forward_output", - task_id=self.task_id, - agent_id=self.agent_id, - display_name=self.display_name, - output_len=len(output_text), - is_final=is_final, - messages_size=estimate_messages_size(self.messages), - ) if not self.multi_agent_state: ma_debug("sub_agent_forward_no_state", task_id=self.task_id, agent_id=self.agent_id) return @@ -638,15 +602,6 @@ 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]: @@ -662,19 +617,11 @@ class SubAgentTask: 多智能体模式下，对于通信工具（ask_master / ask_other_agent / answer_other_agent/ list_active_sub_agents）在主进程内直接处理，不再转发到 WebTerminal。 """ - async with async_mem_snapshot( - "sub_agent_execute_tool", - task_id=self.task_id, - agent_id=self.agent_id, - display_name=self.display_name, - tool_name=name, - messages_size=estimate_messages_size(self.messages), - ): - if self.multi_agent_mode and self.multi_agent_state: - result = await self._execute_multi_agent_tool(name, args) - if result is not None: - return result - return await self.manager.execute_tool_for_sub_agent(name, args) + if self.multi_agent_mode and self.multi_agent_state: + result = await self._execute_multi_agent_tool(name, args) + if result is not None: + return result + return await self.manager.execute_tool_for_sub_agent(name, args) async def _execute_multi_agent_tool(self, name: str, args: Dict[str, Any]) -> Optional[Dict[str, Any]]: """处理多智能体模式专属的通信工具。返回 None 表示不 属于多智能体工具。""" @@ -768,8 +715,6 @@ class SubAgentTask: self.stats["write_files"] += 1 elif name == "edit_file": self.stats["edit_files"] += 1 - elif name == "search_workspace": - self.stats["searches"] += 1 elif name in ("web_search", "extract_webpage", "save_webpage"): self.stats["web_pages"] += 1 elif name == "run_command": @@ -1016,21 +961,7 @@ class SubAgentTask: "stats": {**self.stats, "runtime_seconds": runtime_seconds, "turn_count": self.stats.get("turn_count", 0)}, } self.conversation_file.parent.mkdir(parents=True, exist_ok=True) - log_memory_event( - "sub_agent_persist_before_dumps", - task_id=self.task_id, - agent_id=self.agent_id, - display_name=self.display_name, - messages_size=estimate_messages_size(self.messages), - ) conversation_json = json.dumps(conversation_data, ensure_ascii=False) - log_memory_event( - "sub_agent_persist_after_conversation_dumps", - task_id=self.task_id, - agent_id=self.agent_id, - display_name=self.display_name, - conversation_json_chars=len(conversation_json), - ) self.conversation_file.write_text(conversation_json, encoding="utf-8") output_data = { @@ -1042,13 +973,6 @@ class SubAgentTask: self.output_file.parent.mkdir(parents=True, exist_ok=True) output_json = json.dumps(output_data, ensure_ascii=False) self.output_file.write_text(output_json, encoding="utf-8") - log_memory_event( - "sub_agent_persist_after_output_dumps", - task_id=self.task_id, - agent_id=self.agent_id, - display_name=self.display_name, - output_json_chars=len(output_json), - ) except Exception as exc: logger.warning(f"[SubAgentTask] 增量保存失败: {exc}") diff --git a/modules/sub_agent/toolkit.py b/modules/sub_agent/toolkit.py index 70f6cbb..356de3b 100644 --- a/modules/sub_agent/toolkit.py +++ b/modules/sub_agent/toolkit.py @@ -145,29 +145,6 @@ SUB_AGENT_TOOLS: List[Dict[str, Any]] = [ }, }, }, - { - "type": "function", - "function": { - "name": "search_workspace", - "description": "在本地目录内搜索文件名或文件内容（跨文件）。仅返回摘要。", - "parameters": { - "type": "object", - "properties": { - "mode": {"type": "string", "enum": ["file", "content"]}, - "query": {"type": "string"}, - "root": {"type": "string", "description": "搜索起点目录，默认 '.'。"}, - "use_regex": {"type": "boolean"}, - "case_sensitive": {"type": "boolean"}, - "max_results": {"type": "integer"}, - "max_matches_per_file": {"type": "integer"}, - "include_glob": {"type": "array", "items": {"type": "string"}}, - "exclude_glob": {"type": "array", "items": {"type": "string"}}, - "max_file_size": {"type": "integer"}, - }, - "required": ["mode", "query"], - }, - }, - }, { "type": "function", "function": { @@ -338,21 +315,6 @@ def _format_tool_result(name: str, raw: Any) -> str: if raw.get("mode") == "save": return f"已保存: {raw.get('path')}" return f"URL: {raw.get('url')}\n{raw.get('content') or ''}" - if name == "search_workspace": - if raw.get("mode") == "file": - lines = [f"命中文件({len(raw.get('matches') or [])}):"] - for idx, p in enumerate(raw.get("matches") or [], 1): - lines.append(f"{idx}) {p}") - return "\n".join(lines) - if raw.get("mode") == "content": - parts = [] - for item in raw.get("results") or []: - parts.append(item.get("file") or "") - for m in item.get("matches") or []: - parts.append(f"- L{m.get('line')}: {m.get('snippet')}") - parts.append("") - return "\n".join(parts).strip() - return "" if name == "read_mediafile": return raw.get("message") or "媒体已读取" if name == "read_skill": diff --git a/modules/sub_agent/tools.py b/modules/sub_agent/tools.py index 89d7b61..9ef7c9a 100644 --- a/modules/sub_agent/tools.py +++ b/modules/sub_agent/tools.py @@ -1,231 +1,16 @@ -"""子智能体特有工具实现（search_workspace / read_mediafile）。 +"""子智能体特有工具实现（read_mediafile）。 -这些工具在主进程中没有同名工具，因此由子智能体管理器本地实现。 -search_workspace 优先复用主进程的 run_command 调用 rg，回退到 Python 遍历； read_mediafile 直接读取项目内媒体文件并返回 base64。 +（原 search_workspace 工具已移除：子智能体需要搜索时改用 run_command +执行 rg / grep / find，走主进程沙箱链路，避免进程内遍历大目录导致内存失控。） """ from __future__ import annotations import base64 -import fnmatch -import json import mimetypes -import re -import shlex -import shutil from pathlib import Path -from typing import Any, Dict, List, Optional - - -async def handle_search_workspace( - project_path: Path, - terminal: Any, - arguments: Dict[str, Any], -) -> Dict[str, Any]: - """search_workspace 实现。""" - mode = arguments.get("mode") - query = arguments.get("query") or "" - root = arguments.get("root") or "." - use_regex = bool(arguments.get("use_regex", False)) - case_sensitive = bool(arguments.get("case_sensitive", False)) - max_results = int(arguments.get("max_results", 20)) - max_matches_per_file = int(arguments.get("max_matches_per_file", 3)) - include_glob = arguments.get("include_glob") or ["**/*"] - exclude_glob = arguments.get("exclude_glob") or [] - max_file_size = arguments.get("max_file_size") - - if mode not in {"file", "content"}: - return {"success": False, "error": "search_workspace mode 必须是 file 或 content"} - - root_path = (project_path / root).resolve() - try: - root_path.relative_to(project_path) - except Exception: - return {"success": False, "error": "搜索根目录必须位于项目目录内"} - - if mode == "file": - return _search_workspace_file(project_path, root_path, query, use_regex, case_sensitive, max_results, include_glob, exclude_glob) - - rg_result = await _search_workspace_content_rg( - project_path, root_path, query, use_regex, case_sensitive, max_results, max_matches_per_file, - include_glob, exclude_glob, max_file_size, terminal - ) - if rg_result is not None: - return rg_result - - return _search_workspace_content_python( - project_path, root_path, query, use_regex, case_sensitive, max_results, max_matches_per_file, - include_glob, exclude_glob, max_file_size - ) - - -def _search_workspace_file( - project_path: Path, - root_path: Path, - query: str, - use_regex: bool, - case_sensitive: bool, - max_results: int, - include_glob: List[str], - exclude_glob: List[str], -) -> Dict[str, Any]: - try: - pattern = re.compile(query, 0 if case_sensitive else re.IGNORECASE) if use_regex else None - needle = query if case_sensitive else query.lower() - matches = [] - for item in _iter_files(project_path, root_path, include_glob, exclude_glob): - name = item.name - hay = name if case_sensitive else name.lower() - ok = pattern.search(name) if pattern else needle in hay - if ok: - matches.append(str(item.relative_to(project_path))) - if len(matches) >= max_results: - break - return { - "success": True, - "mode": "file", - "root": str(root_path.relative_to(project_path)), - "query": query, - "matches": matches, - } - except Exception as exc: - return {"success": False, "error": str(exc)} - - -async def _search_workspace_content_rg( - project_path: Path, - root_path: Path, - query: str, - use_regex: bool, - case_sensitive: bool, - max_results: int, - max_matches_per_file: int, - include_glob: List[str], - exclude_glob: List[str], - max_file_size: Optional[int], - terminal: Any, -) -> Optional[Dict[str, Any]]: - if not shutil.which("rg"): - return None - try: - args = ["rg", "--line-number", "--no-heading", "--color=never"] - if not use_regex: - args.append("--fixed-strings") - if not case_sensitive: - args.append("-i") - if max_matches_per_file: - args.append(f"--max-count={max_matches_per_file}") - if max_file_size: - args.append(f"--max-filesize={max_file_size}") - for g in include_glob: - args.extend(["-g", g]) - for g in exclude_glob: - args.extend(["-g", f"!{g}"]) - args.extend([query, str(root_path)]) - - result_text = await terminal.handle_tool_call("run_command", {"command": shlex.join(args), "timeout": 60}) - parsed = json.loads(result_text) if isinstance(result_text, str) else result_text - if not parsed.get("success"): - return None - - files_map: Dict[str, List[Dict[str, Any]]] = {} - for line in (parsed.get("output") or "").splitlines(): - if not line.strip(): - continue - parts = line.split(":", 2) - if len(parts) < 3: - continue - file_path, line_num, snippet = parts[0], parts[1], parts[2] - try: - line_no = int(line_num) - except ValueError: - continue - files_map.setdefault(file_path, []) - if len(files_map[file_path]) < max_matches_per_file: - files_map[file_path].append({"line": line_no, "snippet": snippet.strip()}) - if len(files_map) >= max_results and len(files_map[file_path]) >= max_matches_per_file: - break - - results = [{"file": f, "matches": ms} for f, ms in list(files_map.items())[:max_results]] - return { - "success": True, - "mode": "content", - "root": str(root_path.relative_to(project_path)), - "query": query, - "results": results, - } - except Exception: - return None - - -def _search_workspace_content_python( - project_path: Path, - root_path: Path, - query: str, - use_regex: bool, - case_sensitive: bool, - max_results: int, - max_matches_per_file: int, - include_glob: List[str], - exclude_glob: List[str], - max_file_size: Optional[int], -) -> Dict[str, Any]: - try: - pattern = re.compile(query, 0 if case_sensitive else re.IGNORECASE) if use_regex else None - needle = query if case_sensitive else query.lower() - results = [] - for file_path in _iter_files(project_path, root_path, include_glob, exclude_glob): - if max_file_size: - try: - if file_path.stat().st_size > max_file_size: - continue - except Exception: - continue - try: - text = file_path.read_text(encoding="utf-8", errors="ignore") - except Exception: - continue - lines = text.splitlines() - matches = [] - for idx, line in enumerate(lines, 1): - hay = line if case_sensitive else line.lower() - ok = pattern.search(line) if pattern else needle in hay - if ok: - matches.append({"line": idx, "snippet": line.strip()}) - if len(matches) >= max_matches_per_file: - break - if matches: - results.append({"file": str(file_path.relative_to(project_path)), "matches": matches}) - if len(results) >= max_results: - break - return { - "success": True, - "mode": "content", - "root": str(root_path.relative_to(project_path)), - "query": query, - "results": results, - } - except Exception as exc: - return {"success": False, "error": str(exc)} - - -def _iter_files(project_path: Path, root_path: Path, include_glob: List[str], exclude_glob: List[str]): - """按 glob 规则遍历文件。""" - for item in root_path.rglob("*"): - if not item.is_file(): - continue - try: - rel = item.relative_to(project_path) - except Exception: - continue - rel_str = str(rel) - included = any(fnmatch.fnmatch(rel_str, g) for g in include_glob) - if not included: - continue - if any(fnmatch.fnmatch(rel_str, g) for g in exclude_glob): - continue - yield item +from typing import Any, Dict async def handle_read_mediafile(project_path: Path, arguments: Dict[str, Any]) -> Dict[str, Any]: diff --git a/prompts/sub_agent/system.txt b/prompts/sub_agent/system.txt index 33714d8..f01bf86 100644 --- a/prompts/sub_agent/system.txt +++ b/prompts/sub_agent/system.txt @@ -54,8 +54,7 @@ 你拥有以下工具能力： - read_file: 读取文件内容 - write_file / edit_file: 创建或修改文件 -- search_workspace: 搜索文件和代码 -- run_command: 执行终端命令 +- run_command: 执行终端命令（需要搜索文件/代码时，用它执行 rg 或 grep -rn / find；务必排除 node_modules、.git 等重目录，例如 grep -rn --exclude-dir=node_modules --exclude-dir=.git） - web_search / extract_webpage: 搜索和提取网页内容 - read_mediafile: 读取图片/视频文件 - finish_task: 完成任务并退出（必须调用） diff --git a/server/chat_flow_task_main.py b/server/chat_flow_task_main.py index 8fd2bf7..e9c9186 100644 --- a/server/chat_flow_task_main.py +++ b/server/chat_flow_task_main.py @@ -46,7 +46,6 @@ from modules.user_manager import UserWorkspace from modules.usage_tracker import QUOTA_DEFAULTS from modules.sub_agent import TERMINAL_STATUSES from modules.multi_agent.debug_logger import ma_debug -from modules.memory_debug import log_memory_event, estimate_messages_size from modules.versioning_manager import ConversationVersioningManager, VersioningError from modules.shallow_versioning import ShallowVersioningManager from core.web_terminal import WebTerminal @@ -1362,12 +1361,6 @@ async def handle_task_with_sender( message_preview=str(message)[:300] if message else None, pending_master_messages_count=pending_count, ) - log_memory_event( - "handle_task_with_sender_start", - conversation_id=conversation_id, - multi_agent_mode=True, - pending_master_messages_count=pending_count, - ) videos = videos or [] raw_sender = sender @@ -2508,14 +2501,6 @@ async def handle_task_with_sender( finalize_run_versioning_checkpoint("completed") else: finalize_run_versioning_checkpoint("waiting_background") - log_memory_event( - "handle_task_with_sender_complete", - conversation_id=conversation_id, - multi_agent_mode=bool(getattr(web_terminal, "multi_agent_mode", False)), - total_iterations=total_iterations, - total_tool_calls=total_tool_calls, - has_running_multi_agent=has_running_multi_agent or has_pending_ma_messages, - ) sender('task_complete', { 'total_iterations': total_iterations, 'total_tool_calls': total_tool_calls, diff --git a/server/chat_flow_task_support.py b/server/chat_flow_task_support.py index 628eb34..c48f6ea 100644 --- a/server/chat_flow_task_support.py +++ b/server/chat_flow_task_support.py @@ -7,7 +7,6 @@ from typing import Any, Callable, Dict, List, Optional from modules.sub_agent import TERMINAL_STATUSES from modules.multi_agent.debug_logger import ma_debug -from modules.memory_debug import log_memory_event, estimate_messages_size _VALID_SOURCES = { @@ -351,14 +350,6 @@ def inject_multi_agent_master_message( inline=inline, after_tool_call_id=after_tool_call_id, ) - log_memory_event( - "inject_multi_agent_master_message", - conversation_id=conversation_id, - inline=inline, - text_len=len(raw), - messages_size=estimate_messages_size(messages), - ) - # 解析标准格式以获取 subtype；解析失败时回退到通用类型 try: from modules.multi_agent.state import parse_multi_agent_message @@ -475,13 +466,6 @@ async def process_multi_agent_master_messages( return 0 ma_debug("process_ma_messages_got_state", conversation_id=conversation_id, state_id=id(state)) pending = state.drain_master_messages() - log_memory_event( - "process_multi_agent_master_messages", - conversation_id=conversation_id, - inline=inline, - pending_count=len(pending), - messages_size=estimate_messages_size(messages), - ) if not pending: ma_debug("process_ma_messages_empty", conversation_id=conversation_id, state_id=id(state)) return 0 diff --git a/server/deep_compression.py b/server/deep_compression.py index fb87ba5..0808465 100644 --- a/server/deep_compression.py +++ b/server/deep_compression.py @@ -6,9 +6,6 @@ from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, Tuple -from modules.memory_debug import log_memory_event, estimate_messages_size - - def _load_summary_prompt(web_terminal) -> str: """从 prompts/deep_compression_summary.txt 加载压缩总结提示词。""" try: @@ -313,11 +310,6 @@ async def run_deep_compression( mode: str, sender=None, ) -> Dict[str, Any]: - log_memory_event( - "run_deep_compression_enter", - conversation_id=conversation_id, - mode=mode, - ) cm = web_terminal.context_manager target_manager = ( cm._get_conversation_manager_for_id(conversation_id) @@ -516,14 +508,6 @@ async def run_deep_compression( "compress_behavior": compress_behavior, "job_id": job_id, }) - log_memory_event( - "run_deep_compression_exit", - conversation_id=conversation_id, - mode=mode, - marked_count=marked_count, - history_size=estimate_messages_size(cm.conversation_history), - ) - return { "success": True, "in_place": True, diff --git a/static/src/components/overlay/SubAgentActivityDialog.vue b/static/src/components/overlay/SubAgentActivityDialog.vue index 41df30e..b5aeb02 100644 --- a/static/src/components/overlay/SubAgentActivityDialog.vue +++ b/static/src/components/overlay/SubAgentActivityDialog.vue @@ -104,10 +104,6 @@ const buildText = (entry: ActivityEntry) => { const skillName = args.skill_name || ''; return `阅读技能 ${skillName}`; } - if (tool === 'search_workspace') { - const query = args.query || args.keyword || ''; - return `在工作区搜索 ${query}`; - } if (tool === 'web_search') { const query = args.query || args.q || ''; return `在互联网中搜索 ${query}`; diff --git a/test/test_sub_agent_regression.py b/test/test_sub_agent_regression.py index 16d30ba..5648169 100644 --- a/test/test_sub_agent_regression.py +++ b/test/test_sub_agent_regression.py @@ -185,63 +185,6 @@ def test_write_file_and_edit(): shutil.rmtree(tmp, ignore_errors=True) -def test_search_workspace(): - """测试 search_workspace 走 run_command rg 回退。""" - tmp = Path(tempfile.mkdtemp(prefix="sub_agent_test_")) - try: - manager = SubAgentManager(str(tmp), str(tmp / "data")) - terminal = FakeTerminal(tmp) - manager.set_terminal(terminal) - - (tmp / "findme.txt").write_text("hello world unique string", encoding="utf-8") - - async def fake_search(project_path, terminal_, arguments): - query = arguments["query"] - if arguments.get("mode") == "file": - matches = [p.name for p in project_path.rglob("*") if p.is_file() and query in p.name] - return {"success": True, "mode": "file", "matches": matches} - matches = [] - for p in project_path.rglob("*.txt"): - text = p.read_text(encoding="utf-8") - if query in text: - matches.append({"file": str(p.relative_to(project_path)), "matches": [{"line": 1, "snippet": text}]}) - return {"success": True, "mode": "content", "results": matches} - - import modules.sub_agent.tools as sat - original_search = sat.handle_search_workspace - sat.handle_search_workspace = fake_search - - responses = [ - [make_tool_chunk([{ - "name": "search_workspace", - "args": json.dumps({"mode": "content", "query": "unique string"}) - }])], - [make_tool_chunk([{ - "name": "finish_task", - "args": json.dumps({"success": True, "summary": "搜索完成"}) - }])], - ] - _, original_model = install_fake_model(responses) - try: - result = manager.create_sub_agent( - agent_id=2, - summary="搜索测试", - task="测试 search_workspace", - deliverables_dir="deliverables2", - timeout_seconds=30, - conversation_id="conv-test-2", - thinking_mode="fast", - ) - assert result["success"], result - final = manager.wait_for_completion(task_id=result["task_id"], timeout_seconds=10) - assert final["status"] == "completed", final - finally: - restore_model(original_model) - sat.handle_search_workspace = original_search - finally: - shutil.rmtree(tmp, ignore_errors=True) - - def test_read_mediafile(): """测试 read_mediafile 返回 base64。""" tmp = Path(tempfile.mkdtemp(prefix="sub_agent_test_")) @@ -341,7 +284,6 @@ def run_all(): print("=" * 60) tests = [ ("write_file + edit_file", test_write_file_and_edit), - ("search_workspace", test_search_workspace), ("read_mediafile", test_read_mediafile), ("terminate", test_terminate), ] diff --git a/utils/api_client/chat_mixin.py b/utils/api_client/chat_mixin.py index b1d2400..fead686 100644 --- a/utils/api_client/chat_mixin.py +++ b/utils/api_client/chat_mixin.py @@ -35,7 +35,6 @@ from utils.log_rotation import append_line, prune_dir from utils.api_client.utils import _api_dump_enabled -from modules.memory_debug import log_memory_event, estimate_messages_size class DeepSeekClientChatMixin: async def chat( @@ -101,13 +100,6 @@ class DeepSeekClientChatMixin: final_messages = self._merge_system_messages(messages) final_messages = self._sanitize_messages_for_model_capability(final_messages) final_messages = self._sanitize_message_fields_for_api(final_messages) - log_memory_event( - "api_client_chat_request_prepared", - model_key=self.model_key, - input_messages_size=estimate_messages_size(messages), - final_messages_size=estimate_messages_size(final_messages), - ) - payload = { "model": api_config["model_id"], "messages": final_messages, @@ -143,11 +135,6 @@ class DeepSeekClientChatMixin: except Exception: pass dump_path = self._dump_request_payload(payload, api_config, headers) - log_memory_event( - "api_client_chat_request_payload", - model_key=self.model_key, - payload_chars=len(json.dumps(payload, ensure_ascii=False)), - ) try: async with httpx.AsyncClient(http2=True, timeout=300) as client: @@ -197,15 +184,7 @@ 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]": @@ -222,16 +201,6 @@ 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 = {