From 99f90c2ee8a43083d92c710ceca89a7d7b9ddd55 Mon Sep 17 00:00:00 2001 From: minorQAQ <202200120184@mail.sdu.edu.cn> Date: Fri, 28 Aug 2026 11:32:04 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E4=B8=8A=E4=B8=8B=E6=96=87?= =?UTF-8?q?=E7=BC=96=E6=8E=92=E9=97=AE=E9=A2=98=E3=80=81=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E5=AD=90agent=E6=B6=88=E6=81=AF=E9=98=9F=E5=88=97=E5=B1=95?= =?UTF-8?q?=E7=A4=BAUI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- electron/main.js | 23 ++- src/agent/agents/agent_runtime.py | 6 +- src/agent/agents/config_dist.json | 1 - src/agent/core/config_manager.py | 11 +- src/agent/core/db.py | 6 +- src/agent/core/loop_detector.py | 173 -------------------- src/agent/core/nodes.py | 199 ++++-------------------- src/agent/core/storage.py | 145 ++++++++++++----- src/agent/history/tool_call_recorder.py | 85 +++++++--- src/agent/tools/__init__.py | 49 +++++- src/agent/tools/rag.py | 8 +- src/agent/tools/todo_list.py | 23 ++- src/agent/utils/agent_utils.py | 97 ++++++------ src/agent/utils/tool_call_utils.py | 2 +- src/web/js/toolcalls.js | 10 +- src/web/server.py | 12 +- 16 files changed, 369 insertions(+), 481 deletions(-) delete mode 100644 src/agent/core/loop_detector.py diff --git a/electron/main.js b/electron/main.js index fc19da5..b0e335f 100644 --- a/electron/main.js +++ b/electron/main.js @@ -331,6 +331,21 @@ function stopFastAPI() { } // ==================== 窗口管理 ==================== +// 缓存清理标记：仅首次运行清一次（应用页资源经 FastAPI no-cache 头加载， +// 后续启动无需反复清磁盘缓存，省掉每次启动的等待） +const CACHE_CLEARED_FLAG = () => path.join(app.getPath('userData'), '.cache-cleared'); + +/** 返回 true 表示本次应清理缓存（首次运行且已写标记）；标记写入失败时保持旧行为 */ +function ensureInitialCacheClear() { + try { + if (fs.existsSync(CACHE_CLEARED_FLAG())) return false; + fs.writeFileSync(CACHE_CLEARED_FLAG(), String(Date.now()), 'utf-8'); + return true; + } catch { + return true; + } +} + async function createMainWindow(themeMode) { themeMode = themeMode || getThemeMode(); @@ -350,9 +365,11 @@ async function createMainWindow(themeMode) { show: false, }); - // 清除缓存后再加载，确保获取最新 JS/CSS - await mainWindow.webContents.session.clearCache(); - await mainWindow.webContents.session.clearStorageData({ storages: ['caches', 'serviceworkers'] }); + // 仅首次运行清除缓存（本地文件与应用页资源不走 HTTP 缓存，无需每次清理） + if (ensureInitialCacheClear()) { + await mainWindow.webContents.session.clearCache(); + await mainWindow.webContents.session.clearStorageData({ storages: ['caches', 'serviceworkers'] }); + } // 先显示本地过渡页：毛玻璃 + 中央图标（无任何按钮），主题随 theme_config.json 变化； // FastAPI 就绪后再切到应用页；应用页自带的同款玻璃覆盖层会无缝接手 diff --git a/src/agent/agents/agent_runtime.py b/src/agent/agents/agent_runtime.py index 03dbf54..27b4043 100644 --- a/src/agent/agents/agent_runtime.py +++ b/src/agent/agents/agent_runtime.py @@ -17,7 +17,7 @@ from agent.core.config_manager import load_agent_configs, load_models, AgentConf from agent.core.graph import build_agent_graph from agent.core.runtime import execute_agent as execute_runtime_agent from agent.core.runtime import AgentRuntime -from agent.tools import get_tools +from agent.tools import get_tools, ensure_tool_registered AgentMode = Literal["agent", "plan", "cron"] @@ -75,6 +75,10 @@ def _build_agent_runtime(cfg: AgentConfig) -> AgentRuntime: # 2. 获取所有可用工具并按配置过滤 all_tools = get_tools() if cfg.tools: + # 动态注册：配置显式引用但未静态注册的按需工具（如 rag_tool） + # 在此现场加载（其 langchain/chromadb 重依赖不占用启动时间） + for name in cfg.tools: + ensure_tool_registered(name) agent_tools = [t for t in all_tools if t.name in cfg.tools] else: agent_tools = all_tools diff --git a/src/agent/agents/config_dist.json b/src/agent/agents/config_dist.json index 71af9fb..111f15d 100644 --- a/src/agent/agents/config_dist.json +++ b/src/agent/agents/config_dist.json @@ -99,7 +99,6 @@ "RAG_CHUNK_OVERLAP": "50", "GROUNDING_WIDTH": "1000", "GROUNDING_HEIGHT": "1000", - "LOOP_DETECT_REPEATED_TOOL_WARN": "3", "SEND_FILE_SIZE_LIMIT": "30", "STORAGE_BACKEND": "json", "CRON_TIME_PERIOD_MINUTES": 30, diff --git a/src/agent/core/config_manager.py b/src/agent/core/config_manager.py index f7d9499..8e4173b 100644 --- a/src/agent/core/config_manager.py +++ b/src/agent/core/config_manager.py @@ -453,7 +453,6 @@ def _default_env_config() -> dict: "RAG_CHUNK_OVERLAP": "50", "GROUNDING_WIDTH": "1000", "GROUNDING_HEIGHT": "1000", - "LOOP_DETECT_REPEATED_TOOL_WARN": "3", "SEND_FILE_SIZE_LIMIT": "30", # 浏览器控制工具持久 Profile 目录（空 = 每次启动一次性临时 Profile） "BROWSER_PROFILE_DIR": "", @@ -511,9 +510,13 @@ def save_env_config(config: dict) -> bool: # ---------- 获取所有已注册工具的基础信息 ---------- def get_registered_tool_names() -> list[str]: - """返回所有可用工具名称（含数据库工具）。""" - from agent.tools import get_all_available_tool_names - return get_all_available_tool_names() + """返回所有可选工具名称（静态注册 + 按需工具名，含数据库工具）。 + + 按需工具（如 rag_tool）仅返回名字供前端勾选，不触发其重依赖加载； + 实际构建 Agent 时如该工具在配置中被引用，才由 ensure_tool_registered 现场注册。 + """ + from agent.tools import get_selectable_tool_names + return get_selectable_tool_names() # ---------- GUI 显示器配置 ---------- diff --git a/src/agent/core/db.py b/src/agent/core/db.py index 3e9f7d3..a3908f4 100644 --- a/src/agent/core/db.py +++ b/src/agent/core/db.py @@ -374,7 +374,8 @@ _SQLITE_SESSION_TABLE_DDLS = [ session_id TEXT NOT NULL, turn_id TEXT NOT NULL, tool_calls TEXT NOT NULL, - meta TEXT + meta TEXT, + assistant_units TEXT ) """, "CREATE INDEX IF NOT EXISTS idx_sess_tc_turn ON session_tool_calls (session_id, turn_id)", @@ -418,7 +419,8 @@ _SQLITE_CRON_TABLE_DDLS = [ task_id TEXT NOT NULL, turn_id TEXT NOT NULL, tool_calls TEXT NOT NULL, - meta TEXT + meta TEXT, + assistant_units TEXT ) """, "CREATE INDEX IF NOT EXISTS idx_task_tc_turn ON cron_tool_calls (task_id, turn_id)", diff --git a/src/agent/core/loop_detector.py b/src/agent/core/loop_detector.py deleted file mode 100644 index d0addf2..0000000 --- a/src/agent/core/loop_detector.py +++ /dev/null @@ -1,173 +0,0 @@ -"""Agent 重复工具调用检测模块。 - -系统定位: - 在 ReAct 循环的 agent 节点调用 LLM 前，检测连续相同工具调用的异常模式， - 提供 Level 1 响应：注入反思提示引导模型自我纠正（不再强制终止，Level 2 已移除）。 - -配置来源: - env_config.json 中的 LOOP_DETECT_REPEATED_TOOL_WARN 参数，可由前端动态修改。 - -可扩展性： - 可增加更多循环模式检测器、自定义检测规则。 -""" -from __future__ import annotations - -import hashlib -import json -from typing import Any - -from langchain_core.messages import AIMessage - -from agent.utils.tool_call_utils import normalize_tool_call, format_args_summary - -# ============================================================ -# 配置加载（每次检测时实时读取，支持前端热更新） -# ============================================================ - -def get_loop_config() -> dict: - """从 env_config.json 读取循环检测参数（每次调用实时读取）。""" - try: - from agent.core.config_manager import load_env_config - env = load_env_config() - except Exception: - env = {} - return { - "repeated_tool_warn": int(env.get("LOOP_DETECT_REPEATED_TOOL_WARN", 3)), - } - -# ============================================================ -# 工具调用历史分析 -# ============================================================ - -def _tool_call_fingerprint(tc: dict) -> str: - """计算单个工具调用的指纹 (tool_name, args_hash)。 - - 输入: - tc: 已 normalize 的工具调用 dict。 - - 输出: - "tool_name:md5hash" 格式的指纹字符串。 - """ - tc_name = tc.get("name", "") if isinstance(tc, dict) else getattr(tc, "name", "") - tc_args = tc.get("args", {}) if isinstance(tc, dict) else getattr(tc, "args", {}) - args_str = json.dumps(tc_args, sort_keys=True, default=str) - args_hash = hashlib.md5(args_str.encode()).hexdigest()[:12] - return f"{tc_name}:{args_hash}" - - -def get_recent_tool_calls_from_messages( - messages: list, - count: int = 10, -) -> list[dict]: - """从消息列表中提取最近 N 个工具调用（含指纹）。 - - 输入: - messages: 消息列表。 - count: 需要提取的最近工具调用数量。 - - 输出: - [{"name": str, "args": dict, "fingerprint": str}, ...]，按时间正序排列。 - """ - result = [] - for msg in messages: - if isinstance(msg, AIMessage) and getattr(msg, "tool_calls", None): - for tc in msg.tool_calls: - tc_dict = normalize_tool_call(tc) - fingerprint = _tool_call_fingerprint(tc_dict) - result.append({ - "name": tc_dict.get("name", "unknown"), - "args": tc_dict.get("args", {}), - "fingerprint": fingerprint, - }) - # 按时间正序，取最近 count 个 - return result[-count:] if len(result) > count else result - - -def _compute_consecutive_same(recent_calls: list[dict]) -> int: - """计算从末尾向前连续相同的工具调用次数（基于指纹）。 - - 输入: - recent_calls: 工具调用列表（时间正序）。 - - 输出: - 末尾连续相同指纹的次数。 - """ - if not recent_calls: - return 0 - last_fp = recent_calls[-1]["fingerprint"] - count = 0 - for call in reversed(recent_calls): - if call["fingerprint"] == last_fp: - count += 1 - else: - break - return count - - -def detect_repeated_tool_calls( - messages: list, - session_id: str = "", -) -> tuple[int, str, str | None]: - """检测连续相同工具调用。 - - 输入: - messages: 当前消息列表。 - session_id: 会话 ID。 - - 输出: - (连续次数, 工具名, 参数摘要 | None) - 连续次数为 0 表示无重复。 - """ - cfg = get_loop_config() - warn_threshold = max(cfg["repeated_tool_warn"], 2) - lookback = warn_threshold + 2 - - recent = get_recent_tool_calls_from_messages(messages, count=lookback) - consecutive = _compute_consecutive_same(recent) - - if consecutive < warn_threshold: - return 0, "", None - - last_call = recent[-1] if recent else {} - args_summary = format_args_summary(last_call.get("args", {})) - return consecutive, last_call.get("name", "unknown"), args_summary - - -# ============================================================ -# 反循环反思提示生成 -# ============================================================ - -def build_loop_reflection_prompt( - messages: list, - session_id: str = "", -) -> str | None: - """根据当前状态构建反循环反思提示（Level 1 响应）。 - - 输入: - messages: 当前消息列表。 - session_id: 会话 ID。 - - 输出: - 反思提示文本；无需注入时返回 None。 - """ - parts = [] - - # 检查 1: 连续相同工具调用 (Level 1 - 警告) - tool_count, tool_name, args_summary = detect_repeated_tool_calls(messages, session_id) - cfg = get_loop_config() - warn_threshold = max(cfg["repeated_tool_warn"], 2) - - if tool_count >= warn_threshold: - parts.append( - f"【循环警告】你已经连续 {tool_count} 次调用工具 `{tool_name}` " - f"且参数基本相同（{args_summary}），但似乎没有取得实质进展。\n" - f"请反思当前策略是否有效，考虑：\n" - f"1. 更换工具或策略\n" - f"2. 检查之前的工具返回结果是否有误\n" - f"3. 如果确实无法推进，向用户说明当前困难并请求帮助" - ) - - if not parts: - return None - - return "\n\n".join(parts) diff --git a/src/agent/core/nodes.py b/src/agent/core/nodes.py index 649ea8b..cf6e06b 100644 --- a/src/agent/core/nodes.py +++ b/src/agent/core/nodes.py @@ -20,7 +20,6 @@ from typing import Any, Callable from langchain_core.messages import AIMessage, HumanMessage, ToolMessage -from agent.core.loop_detector import build_loop_reflection_prompt from agent.core.state import AgentState from agent.memory.system_prompt import SUB_AGENT_OOM_PROMPT from agent.core.llm import LLMRetryExhausted, is_timeout_error @@ -64,7 +63,8 @@ def _filter_short_term_memory(messages: list, trajectory_rounds: int) -> list: - 仅 process_tool_artifact 生成的合成图片 HumanMessage 按 trajectory_rounds 裁剪 - 只对实际包含图片截图的轮次计数，无截图的轮次跳过不计数 - 上一/几轮的反思提示词（reflection 标记）不进入本轮上下文 - - 带 tool_calls 的历史 AIMessage 仅保留工具调用，content 中的思考文本不进入上下文 + - 带 tool_calls 的 AIMessage 完整保留（content + reasoning_content + tool_calls + 三元组原样进入上下文，不做剥离/合成，保证前缀缓存一致） """ if trajectory_rounds <= 0: return list(messages) @@ -103,93 +103,7 @@ def _filter_short_term_memory(messages: list, trajectory_rounds: int) -> list: kept_indices.add(i) - result = [messages[i] for i in sorted(kept_indices)] - - # 剥离历史模型思考文本：上一/几轮的思考不进入本轮上下文。 - # 带 tool_calls 的 AIMessage，其 content 仅是思维链，保留 tool_calls 即可 - # （思考/反思已由 extract_thought/extract_reflection 单独记录到 think 面板， - # 并由 build_native_history_messages 以原生 AIMessage 形式注入）。 - cleaned: list = [] - for m in result: - if isinstance(m, AIMessage) and getattr(m, "tool_calls", None) and getattr(m, "content", None): - cleaned.append(m.model_copy(update={"content": ""})) - else: - cleaned.append(m) - return cleaned - - -def _build_dynamic_tail_prompt(messages: list, session_id: str = "") -> str: - """构建每轮追加到消息末尾的动态尾部提示（仅变化部分）。 - - 输入: - messages: 当前消息列表（已过滤短期记忆），供循环检测使用。 - session_id: 会话 ID，供循环检测与 TodoList 状态重建使用。 - - 输出: - 动态尾部文本；无 TodoList 且无循环提醒时返回空串 - （此时不加任何占位提示，省 token）。 - - 系统定位: - - 固定反思引导已移入系统提示词前缀（system_prompt._reflection_section）， - 不再在节点中间构造固定提示； - - 循环警告：思考关闭档位注入（Level 1 仅提醒，不强制终止）； - - TodoList 状态：思考开/关均注入（事实状态，供模型跟踪计划进度）。 - """ - from agent.utils.env_utils import thinking_enabled - - parts: list[str] = [] - - # 思考关闭：循环警告（Level 1 仅提醒，不强制终止） - if not thinking_enabled(): - loop_warning = build_loop_reflection_prompt(messages, session_id) - if loop_warning: - parts.append(loop_warning) - - # TodoList 状态：思考开/关均注入（事实状态，供模型跟踪计划进度） - try: - from agent.tools.todo_list import get_todo_status_text - todo_status = get_todo_status_text(session_id) - except Exception: - todo_status = "" - if todo_status: - parts.append(todo_status) - - return "\n\n".join(parts) - - -def _dynamic_tail_agent_key() -> str: - """动态尾部长期记忆的 agent 键：主 Agent 为 "main"，子 Agent 用其名称。 - - 主/子 Agent 的动态尾部各自隔离，避免子 Agent 覆盖主 Agent 的长期记忆。 - """ - try: - from agent.core.human_request import get_current_agent_name - name = get_current_agent_name() - except Exception: - name = "" - return "main" if name in ("", "__main__") else name - - -def _persist_dynamic_tail(session_id: str, text: str) -> None: - """持久化当前轮动态尾部到 session_meta（长期记忆，供后续轮次注入查看）。""" - if not session_id: - return - try: - from agent.utils.agent_utils import update_session_dynamic_tail - update_session_dynamic_tail(session_id, _dynamic_tail_agent_key(), text) - except Exception: - pass - - -def _load_dynamic_tail_history(session_id: str) -> list[str]: - """读取当前 Agent 的动态尾部长期记忆历史（时间正序，最新在末尾）。""" - if not session_id: - return [] - try: - from agent.utils.agent_utils import get_session_dynamic_tail_history - return get_session_dynamic_tail_history(session_id, _dynamic_tail_agent_key()) - except Exception: - return [] + return [messages[i] for i in sorted(kept_indices)] # ---------- ReAct 循环内上下文压缩 ---------- @@ -539,58 +453,6 @@ def maybe_compress_context(state: AgentState) -> dict: return {"_need_compress": False} -def _inject_turn_thinking(messages: list[Any]) -> list[Any]: - """把本轮图消息中携带的思考/反思（reasoning_content / content）以合成消息插回上下文。 - - reasoning_content 不被 LLM API 回传（additional_kwargs 不在标准消息 - schema 内），跨步骤调用时模型会丢失自己的思考链；此处在其所属的 - 工具调用 AIMessage 之前插入 ``[思考]`` / ``[反思]`` 合成消息 - （因果交错），使模型在后续步骤中仍能看到本轮已产生的思考与反思。 - - 注意: - - 仅注入到本次调用的 messages_for_llm，不进图状态，无副作用； - - 思考与反思独立提取（extract_thought / extract_reflection）， - 可同时存在（既有深度思考又有反思的通用情况），分别注入； - - 思考文本与位置确定性构建（同一图状态下结果恒定），id 稳定， - 不破坏 LLM 前缀缓存； - - trim_messages 已剥空 content 时反思自然跳过（思考存于 - additional_kwargs，不受影响）。 - """ - out: list[Any] = [] - think_seq = 0 - reflect_seq = 0 - for m in messages: - if isinstance(m, AIMessage) and getattr(m, "tool_calls", None): - # 原生重建历史（hist_*）已是 AIMessage 格式：思考在 - # additional_kwargs.reasoning_content、反思在 content， - # 由 llm._get_request_payload 以原生字段透传 → 不在此合成文本。 - # 图中当前轮响应同理：reasoning_content 随之透传后 - # [思考] 合成为重复内容，跳过；[反思] 保持原行为兜底。 - if not _msg_id(m).startswith(HISTORY_MSG_PREFIX): - try: - thought = extract_thought(m) - reflection = extract_reflection(m) - except Exception: - thought = reflection = "" - has_reasoning = bool((getattr(m, "additional_kwargs", None) or {}).get("reasoning_content")) - if thought and not has_reasoning: - out.append(HumanMessage( - content=f"[思考] {thought}", - additional_kwargs={"synthetic": True}, - id=f"turn_think_{think_seq}", - )) - think_seq += 1 - if reflection: - out.append(HumanMessage( - content=f"[反思] {reflection}", - additional_kwargs={"synthetic": True}, - id=f"turn_reflect_{reflect_seq}", - )) - reflect_seq += 1 - out.append(m) - return out - - def _invoke_llm_with_retry(model: Any, messages: list[Any], session_id: str = "") -> Any: """调用已绑定工具的模型，LLM 服务超时时自动重试（默认 3 次，LLM_MAX_RETRIES 配置）。 @@ -647,12 +509,13 @@ def make_call_model_node( model_with_tools: Any, trajectory_rounds: int = 3, ) -> Callable[[AgentState], dict]: - """创建带按思考档位注入提示 + 上下文中压缩的调用已绑定工具的 LLM 的图节点闭包。 + """创建调用已绑定工具的 LLM 的图节点闭包（含上下文中压缩）。 - 每次调用 LLM 前按思考档位注入提示（思考关闭：反思引导 + TodoList 状态 + - 循环警告；思考开启：仅 TodoList 状态）；调用后提取模型产出的思考/反思内容 - （思考开启取深度思考，思考关闭取一句话反思），记录到 think store， - 供前端实时展示思维链。 + 消息上下文 = 图内消息原样（AIMessage 三元组 content/reasoning_content/ + tool_calls 完整保留，json 原生透传，不做合成/剥离/尾部注入）； + TodoList 状态由 process_tool_artifact 拼入各轮工具返回值。 + 调用后提取模型产出的思考/反思内容（思考开启取深度思考，思考关闭取 + 一句话反思），记录到 think store 供前端实时展示思维链。 在每次 LLM 调用后，根据 response 的 ``usage_metadata.total_tokens`` 判断 是否触发消息列表就地压缩——确保 ReAct 循环中的上下文永不溢出。 @@ -675,31 +538,17 @@ def make_call_model_node( filtered_messages = _filter_short_term_memory(all_messages, trajectory_rounds) - # 获取 session_id 供循环检测使用 + # 获取 session_id（token 统计、工具调用记录使用） try: from agent.history.tool_call_recorder import get_current_session _sid = get_current_session() except Exception: _sid = "" - # 构建动态尾部提示并注入到消息列表末尾（仅作引导，不记录到前端）。 - # 固定反思引导已并入系统提示词前缀，此处仅注入变化部分： - # TodoList 状态与循环提醒；无 TodoList 且无循环提醒时不注入（省 token）。 + # 上下文 = 图内消息原样（AIMessage 三元组 content/reasoning_content/tool_calls + # 完整保留，由 llm._get_request_payload 原生透传，不做合成/剥离/尾部注入）。 + # TodoList 状态已由 process_tool_artifact 拼入各轮工具返回值。 messages_for_llm = list(filtered_messages) - # 注入本轮已产生的思考（reasoning_content 不被 API 回传，跨步骤需回插； - # 因果交错：思考位于其工具调用 AIMessage 之前） - messages_for_llm = _inject_turn_thinking(messages_for_llm) - tail = _build_dynamic_tail_prompt(filtered_messages, _sid) - if tail: - # 长期记忆：持久化本轮动态尾部，供后续轮次注入查看 - _persist_dynamic_tail(_sid, tail) - # 注入全部历史动态尾部（含本轮），使 Agent 能看到之前所有轮的 - # todo 状态与循环提醒；这些消息仅存在于本次调用，不进图状态。 - for hist_text in _load_dynamic_tail_history(_sid): - messages_for_llm.append(HumanMessage( - content=hist_text, - additional_kwargs={"synthetic": True, "reflection": True}, - )) # 超时自动重试：仅 ReAct 调用的模型调用（工具内部 LLM 与压缩节点不受影响） response = _invoke_llm_with_retry(model_with_tools, messages_for_llm, _sid) @@ -790,13 +639,33 @@ def process_tool_artifact(state: AgentState) -> dict: if not isinstance(msg, ToolMessage): break trailing.append(msg) - # 按原始顺序处理（保证 record_tool_result_live 的 reverse 查找正确） + # TodoList 状态拼入本轮最后一条工具返回值（存在未完成计划时）。 + # 状态文本随工具返回值一起被 record/save_turn 持久化， + # build_native_history_messages 重建时原样带出——无额外存储与注入逻辑； + # 计划全部完成后不再拼接，之后的任务状态变更（新建计划）自动恢复。 + if trailing: + try: + from agent.tools.todo_list import get_todo_status + status_text, all_done = get_todo_status(sid) + if status_text and not all_done: + last_tool_msg = trailing[0] + raw = str(last_tool_msg.content) if last_tool_msg.content else "" + suffix = f"\n\n【任务进度】\n{status_text}" + last_tool_msg.content = raw + suffix if raw else status_text + except Exception: + pass + # 按原始顺序处理（保证 record_tool_result_live 的 reverse 查找正确）； + # 最后一条（拼接了任务进度）强制按 tool_call_id 更新，覆盖后台任务等 + # 已标记 done 的 live 记录，确保进度文本同步进实时记录与存档。 for msg in reversed(trailing): + is_last = msg is trailing[0] record_tool_result_live( sid, msg.name or "unknown", str(msg.content) if msg.content else "", msg.artifact, + force_update=is_last, + tool_call_id=getattr(msg, "tool_call_id", ""), ) except Exception: pass diff --git a/src/agent/core/storage.py b/src/agent/core/storage.py index 1c86330..8b7548d 100644 --- a/src/agent/core/storage.py +++ b/src/agent/core/storage.py @@ -86,7 +86,7 @@ class ConversationRecordRepository(Protocol): def list_turn_ids(self, session_id: str) -> list[str]: ... def save_tool_calls( self, session_id: str, turn_id: str, tool_calls: list[dict], meta: dict | None = None, - reflections: list[dict] | None = None, + reflections: list[dict] | None = None, assistant_units: list[dict] | None = None, ) -> None: ... def get_turn_record(self, session_id: str, turn_id: str) -> dict[str, Any] | None: ... def get_latest_turn_record(self, session_id: str) -> dict[str, Any] | None: ... @@ -149,6 +149,7 @@ _SESSION_TABLE_DDLS = [ turn_id VARCHAR(64) NOT NULL, tool_calls JSON NOT NULL, meta JSON DEFAULT NULL, + assistant_units JSON DEFAULT NULL, INDEX idx_sess_turn (session_id, turn_id) ) CHARACTER SET utf8mb4 """, @@ -192,6 +193,7 @@ _CRON_TABLE_DDLS = [ task_id VARCHAR(64) NOT NULL, turn_id VARCHAR(64) NOT NULL, tool_calls JSON NOT NULL, + assistant_units JSON DEFAULT NULL, INDEX idx_task_turn (task_id, turn_id) ) CHARACTER SET utf8mb4 """, @@ -353,7 +355,7 @@ class JsonRecordRepository: # ---- 工具调用记录 ---- def save_tool_calls( self, session_id: str, turn_id: str, tool_calls: list[dict], meta: dict | None = None, - reflections: list[dict] | None = None, + reflections: list[dict] | None = None, assistant_units: list[dict] | None = None, ) -> None: if not session_id or not turn_id: return @@ -369,8 +371,12 @@ class JsonRecordRepository: record.update(meta) # meta 全量透传，消费方新增字段无需改动存储层 if reflections: # 尾部思考记录（未被任何工具调用消费的反思），随 turn 一并持久化， - # 供后续轮次 build_native_history_messages 重建原生消息注入上下文 + # 供前端 think 面板历史回放 record["reflections"] = reflections + if assistant_units: + # 每轮 AIMessage 三元组（content/reasoning_content/tool_calls）， + # 供 build_native_history_messages 1:1 重建上下文 + record["assistant_units"] = assistant_units with open(self._tool_path(session_id, turn_id), "w", encoding="utf-8") as f: json.dump(record, f, ensure_ascii=False, indent=2) @@ -596,6 +602,7 @@ class MysqlRecordRepository: else: _ensure_session_tables() self._ensure_extra_column() + self._ensure_assistant_units_column() # ---- 表名 / 连接 ---- def _messages_table(self) -> str: @@ -627,6 +634,21 @@ class MysqlRecordRepository: finally: conn.close() + def _ensure_assistant_units_column(self) -> None: + """幂等添加 *_tool_calls.assistant_units JSON 列（每轮 AIMessage 三元组）。""" + conn = mysql_conn() + try: + with conn.cursor() as cur: + cur.execute( + f"ALTER TABLE {self._tool_table()} " + "ADD COLUMN assistant_units JSON DEFAULT NULL" + ) + conn.commit() + except Exception: + pass # 列已存在 → 忽略 + finally: + conn.close() + @staticmethod def _loads(raw: Any) -> Any: """pymysql 对 JSON 列可能返回 dict 或 str，统一反序列化。""" @@ -773,11 +795,13 @@ class MysqlRecordRepository: # ---- 工具调用记录 ---- def save_tool_calls( self, session_id: str, turn_id: str, tool_calls: list[dict], meta: dict | None = None, - reflections: list[dict] | None = None, + reflections: list[dict] | None = None, assistant_units: list[dict] | None = None, ) -> None: if not session_id or not turn_id: return sid_col = self._sid_col() + # 每轮 AIMessage 三元组（content/reasoning_content/tool_calls），独立列存储 + units_json = json.dumps(assistant_units or [], ensure_ascii=False) conn = mysql_conn() try: with conn.cursor() as cur: @@ -789,9 +813,13 @@ class MysqlRecordRepository: # cron_tool_calls 无 meta 列（与既有表结构一致）， # 尾部思考记录无法入库；JSON 后端下完整持久化 cur.execute( - f"INSERT INTO {self._tool_table()} ({sid_col}, turn_id, tool_calls) " - "VALUES (%s,%s,%s)", - (session_id, turn_id, json.dumps(tool_calls or [], ensure_ascii=False)), + f"INSERT INTO {self._tool_table()} " + f"({sid_col}, turn_id, tool_calls, assistant_units) VALUES (%s,%s,%s,%s)", + ( + session_id, turn_id, + json.dumps(tool_calls or [], ensure_ascii=False), + units_json, + ), ) else: # 尾部思考并入 meta（_row_to_record 全量透传，读取时合并回记录顶层） @@ -799,12 +827,14 @@ class MysqlRecordRepository: meta = {**(meta or {}), "reflections": reflections} cur.execute( f"INSERT INTO {self._tool_table()} " - "(session_id, turn_id, tool_calls, meta) VALUES (%s,%s,%s,%s)", + f"({sid_col}, turn_id, tool_calls, meta, assistant_units) " + "VALUES (%s,%s,%s,%s,%s)", ( session_id, turn_id, json.dumps(tool_calls or [], ensure_ascii=False), json.dumps(meta, ensure_ascii=False) if meta else None, + units_json, ), ) conn.commit() @@ -816,7 +846,7 @@ class MysqlRecordRepository: meta_key: str | None = None) -> dict[str, Any]: """把工具调用表行还原为与 tool_*.json 同构的 record dict。 - meta 全量透传（不做字段白名单）：消费方新增 meta 字段无需改动存储层。 + meta / assistant_units 全量透传（不做字段白名单）：消费方新增字段无需改动存储层。 """ tool_calls = cls._loads(row.get("tool_calls")) or [] record: dict[str, Any] = { @@ -829,6 +859,9 @@ class MysqlRecordRepository: meta = cls._loads(row.get(meta_key)) if isinstance(meta, dict): record.update(meta) + if row.get("assistant_units") is not None: + units = cls._loads(row.get("assistant_units")) + record["assistant_units"] = units if isinstance(units, list) else [] return record def get_turn_record(self, session_id: str, turn_id: str) -> dict[str, Any] | None: @@ -838,22 +871,29 @@ class MysqlRecordRepository: with conn.cursor() as cur: if self.scope == "cron": cur.execute( - f"SELECT tool_calls FROM {self._tool_table()} " + f"SELECT tool_calls, assistant_units FROM {self._tool_table()} " f"WHERE {sid_col}=%s AND turn_id=%s ORDER BY id", (session_id, turn_id), ) rows = cur.fetchall() or [] out: list[dict] = [] + units: list[dict] = [] for r in rows: tc = self._loads(r.get("tool_calls")) if isinstance(tc, list): out.extend(tc) + u = self._loads(r.get("assistant_units")) + if isinstance(u, list): + units.extend(u) if not out: return None - return {"session_id": session_id, "turn_id": turn_id, - "tool_calls": out, "timestamp": turn_id} + rec = {"session_id": session_id, "turn_id": turn_id, + "tool_calls": out, "timestamp": turn_id} + if units: + rec["assistant_units"] = units + return rec cur.execute( - f"SELECT tool_calls, meta FROM {self._tool_table()} " + f"SELECT tool_calls, meta, assistant_units FROM {self._tool_table()} " f"WHERE session_id=%s AND turn_id=%s", (session_id, turn_id), ) @@ -864,6 +904,21 @@ class MysqlRecordRepository: finally: conn.close() + def _cron_record_from_row(self, session_id: str, row: dict[str, Any]) -> dict[str, Any]: + """cron 行 → record（无 meta 列，assistant_units 独立列带出）。""" + tc = self._loads(row.get("tool_calls")) + tid = row.get("turn_id", "") + rec: dict[str, Any] = { + "session_id": session_id, + "turn_id": tid, + "tool_calls": tc if isinstance(tc, list) else [], + "timestamp": tid, + } + units = self._loads(row.get("assistant_units")) + if isinstance(units, list) and units: + rec["assistant_units"] = units + return rec + def get_latest_turn_record(self, session_id: str) -> dict[str, Any] | None: sid_col = self._sid_col() conn = mysql_conn() @@ -871,19 +926,16 @@ class MysqlRecordRepository: with conn.cursor() as cur: if self.scope == "cron": cur.execute( - f"SELECT turn_id, tool_calls FROM {self._tool_table()} " + f"SELECT turn_id, tool_calls, assistant_units FROM {self._tool_table()} " f"WHERE {sid_col}=%s ORDER BY turn_id DESC LIMIT 1", (session_id,), ) row = cur.fetchone() if not row: return None - tc = self._loads(row.get("tool_calls")) - return {"session_id": session_id, "turn_id": row.get("turn_id", ""), - "tool_calls": tc if isinstance(tc, list) else [], - "timestamp": row.get("turn_id", "")} + return self._cron_record_from_row(session_id, row) cur.execute( - f"SELECT turn_id, tool_calls, meta FROM {self._tool_table()} " + f"SELECT turn_id, tool_calls, meta, assistant_units FROM {self._tool_table()} " f"WHERE session_id=%s ORDER BY turn_id DESC LIMIT 1", (session_id,), ) @@ -902,20 +954,14 @@ class MysqlRecordRepository: with conn.cursor() as cur: if self.scope == "cron": cur.execute( - f"SELECT turn_id, tool_calls FROM {self._tool_table()} " + f"SELECT turn_id, tool_calls, assistant_units FROM {self._tool_table()} " f"WHERE {sid_col}=%s ORDER BY turn_id DESC LIMIT %s", (session_id, limit), ) rows = cur.fetchall() or [] - out: list[dict] = [] - for r in rows: - tc = self._loads(r.get("tool_calls")) - out.append({"session_id": session_id, "turn_id": r.get("turn_id", ""), - "tool_calls": tc if isinstance(tc, list) else [], - "timestamp": r.get("turn_id", "")}) - return out + return [self._cron_record_from_row(session_id, r) for r in rows] cur.execute( - f"SELECT turn_id, tool_calls, meta FROM {self._tool_table()} " + f"SELECT turn_id, tool_calls, meta, assistant_units FROM {self._tool_table()} " f"WHERE session_id=%s ORDER BY turn_id DESC LIMIT %s", (session_id, limit), ) @@ -1117,10 +1163,25 @@ class SqliteRecordRepository: def __init__(self, scope: str = "session", db_path: str | None = None) -> None: self.scope = scope if scope == "cron" else "session" self._path_override = db_path + self._db_path_actual = self._db_path() if self.scope == "cron": - ensure_sqlite_tables(self._db_path(), _SQLITE_CRON_TABLE_DDLS) + ensure_sqlite_tables(self._db_path_actual, _SQLITE_CRON_TABLE_DDLS) else: - ensure_sqlite_tables(self._db_path(), _SQLITE_SESSION_TABLE_DDLS) + ensure_sqlite_tables(self._db_path_actual, _SQLITE_SESSION_TABLE_DDLS) + self._ensure_assistant_units_column() + + def _ensure_assistant_units_column(self) -> None: + """幂等添加 *_tool_calls.assistant_units TEXT 列（旧库升级）。""" + conn = sqlite_conn(self._db_path_actual) + try: + conn.execute( + f"ALTER TABLE {self._tool_table()} ADD COLUMN assistant_units TEXT " + ) + conn.commit() + except Exception: + pass # 列已存在 → 忽略 + finally: + conn.close() # ---- 表名 / 连接 ---- def _db_path(self) -> str: @@ -1297,7 +1358,7 @@ class SqliteRecordRepository: # ---- 工具调用记录 ---- def save_tool_calls( self, session_id: str, turn_id: str, tool_calls: list[dict], meta: dict | None = None, - reflections: list[dict] | None = None, + reflections: list[dict] | None = None, assistant_units: list[dict] | None = None, ) -> None: if not session_id or not turn_id: return @@ -1314,12 +1375,14 @@ class SqliteRecordRepository: ) conn.execute( f"INSERT INTO {self._tool_table()} " - f"({sid_col}, turn_id, tool_calls, meta) VALUES (?,?,?,?)", + f"({sid_col}, turn_id, tool_calls, meta, assistant_units) " + "VALUES (?,?,?,?,?)", ( session_id, turn_id, self._dumps(tool_calls or []), self._dumps(meta) if meta else None, + self._dumps(assistant_units or []), ), ) finally: @@ -1330,7 +1393,7 @@ class SqliteRecordRepository: meta_key: str | None = None) -> dict[str, Any]: """把工具调用表行还原为与 tool_*.json 同构的 record dict。 - meta 全量透传（不做字段白名单）：消费方新增 meta 字段无需改动存储层。 + meta / assistant_units 全量透传（不做字段白名单）：消费方新增字段无需改动存储层。 """ tool_calls = cls._loads(row.get("tool_calls")) or [] record: dict[str, Any] = { @@ -1343,6 +1406,9 @@ class SqliteRecordRepository: meta = cls._loads(row.get(meta_key)) if isinstance(meta, dict): record.update(meta) + if row.get("assistant_units") is not None: + units = cls._loads(row.get("assistant_units")) + record["assistant_units"] = units if isinstance(units, list) else [] return record def get_turn_record(self, session_id: str, turn_id: str) -> dict[str, Any] | None: @@ -1350,7 +1416,7 @@ class SqliteRecordRepository: conn = self._conn() try: row = conn.execute( - f"SELECT turn_id, tool_calls, meta FROM {self._tool_table()} " + f"SELECT turn_id, tool_calls, meta, assistant_units FROM {self._tool_table()} " f"WHERE {sid_col}=? AND turn_id=?", (session_id, turn_id), ).fetchone() @@ -1366,7 +1432,7 @@ class SqliteRecordRepository: conn = self._conn() try: row = conn.execute( - f"SELECT turn_id, tool_calls, meta FROM {self._tool_table()} " + f"SELECT turn_id, tool_calls, meta, assistant_units FROM {self._tool_table()} " f"WHERE {sid_col}=? ORDER BY turn_id DESC LIMIT 1", (session_id,), ).fetchone() @@ -1383,7 +1449,7 @@ class SqliteRecordRepository: conn = self._conn() try: rows = conn.execute( - f"SELECT turn_id, tool_calls, meta FROM {self._tool_table()} " + f"SELECT turn_id, tool_calls, meta, assistant_units FROM {self._tool_table()} " f"WHERE {sid_col}=? ORDER BY turn_id DESC LIMIT ?", (session_id, limit), ).fetchall() or [] @@ -2078,11 +2144,12 @@ def list_turn_ids(session_id: str) -> list[str]: def save_tool_calls( session_id: str, turn_id: str, tool_calls: list[dict], meta: dict | None = None, - reflections: list[dict] | None = None, + reflections: list[dict] | None = None, assistant_units: list[dict] | None = None, ) -> None: - """持久化一轮工具调用记录（meta 全量透传；reflections 为尾部思考记录）。""" + """持久化一轮工具调用记录（meta 全量透传；reflections 为尾部思考记录； + assistant_units 为每轮 AIMessage 三元组，供历史重建）。""" try: - _repo().save_tool_calls(session_id, turn_id, tool_calls, meta, reflections) + _repo().save_tool_calls(session_id, turn_id, tool_calls, meta, reflections, assistant_units) except Exception as e: print(f"[storage] save_tool_calls 失败: {e}", file=sys.stderr) diff --git a/src/agent/history/tool_call_recorder.py b/src/agent/history/tool_call_recorder.py index ce51f25..96270df 100644 --- a/src/agent/history/tool_call_recorder.py +++ b/src/agent/history/tool_call_recorder.py @@ -361,8 +361,15 @@ def record_tool_result_live( name: str, result_text: str, artifact: Any | None = None, + force_update: bool = False, + tool_call_id: str = "", ) -> None: - """记录工具返回结果，供前端实时展示文本、图片预览、音频信息与文件信息。""" + """记录工具返回结果，供前端实时展示文本、图片预览、音频信息与文件信息。 + + 默认仅更新 status 仍为 "running" 的条目（幂等）；``force_update=True`` 时 + 按 ``tool_call_id`` 精确匹配并允许覆盖已 done 条目 —— 供 process_tool_artifact + 在工具返回值末尾拼接任务进度后强制同步 live 记录。 + """ # 子 Agent 上下文：路由到 _sub_live_records（按 (session, tool_call_id) 隔离） if is_sub_agent_context(): key = _sub_key(session_id) @@ -389,24 +396,30 @@ def record_tool_result_live( with _live_lock: records = _live_records.get(session_id, []) for rec in reversed(records): - if rec["name"] == name and rec["status"] == "running": - rec["status"] = "done" - rec["result_text"] = (result_text or "")[:4000] - # 图片/文件字节立即落盘为会话目录下的文件，记录只存文件名， - # 不内嵌 base64（实时推送由 get_live_tool_calls 转成 - # /api/tool-image URL；aborted 轮次经 save_aborted_turn 落盘时 - # 文件已存在，历史回放按 result_files 即可还原） - _persist_artifact_files(rec, artifact, records, session_id) - audio_info = _artifact_audio_info(artifact) - if audio_info: - rec["result_audio"] = audio_info - file_infos = _artifact_file_infos(artifact) - if file_infos: - rec["result_file_info"] = file_infos - # 标记变更（增量推送用），并增量持久化到磁盘防止崩溃丢失 - _live_dirty.setdefault(session_id, set()).add(records.index(rec)) - _schedule_live_persist(session_id) - break + # force_update：按 tool_call_id 精确匹配（允许覆盖已 done 的返回值） + if force_update and tool_call_id and rec.get("id") == tool_call_id: + pass + elif rec["name"] == name and rec["status"] == "running": + pass + else: + continue + rec["status"] = "done" + rec["result_text"] = (result_text or "")[:4000] + # 图片/文件字节立即落盘为会话目录下的文件，记录只存文件名， + # 不内嵌 base64（实时推送由 get_live_tool_calls 转成 + # /api/tool-image URL；aborted 轮次经 save_aborted_turn 落盘时 + # 文件已存在，历史回放按 result_files 即可还原） + _persist_artifact_files(rec, artifact, records, session_id) + audio_info = _artifact_audio_info(artifact) + if audio_info: + rec["result_audio"] = audio_info + file_infos = _artifact_file_infos(artifact) + if file_infos: + rec["result_file_info"] = file_infos + # 标记变更（增量推送用），并增量持久化到磁盘防止崩溃丢失 + _live_dirty.setdefault(session_id, set()).add(records.index(rec)) + _schedule_live_persist(session_id) + break _notify_live(session_id) @@ -1226,6 +1239,29 @@ def extract_tool_calls_from_messages(messages: list[BaseMessage]) -> list[dict[s return records +def extract_assistant_units(messages: list[BaseMessage]) -> list[dict[str, Any]]: + """从图执行后的消息列表中提取每轮模型返回的 AIMessage 三元组单元。 + + 每个 AIMessage（含无 tool_calls 的最终回复）按原始顺序保存为一条单元： + {"content": str, "reasoning_content": str, "tool_calls": [{id,name,args}, ...]} + + 上下文历史重建（agent_utils._rebuild_turn_messages）按此单元 1:1 还原 + AIMessage（content + reasoning_content + tool_calls），不做内容剥离与合成， + 保证送入模型的历史与当轮转入时逐字节一致（前缀缓存命中）。 + """ + units: list[dict[str, Any]] = [] + for msg in messages: + if not isinstance(msg, AIMessage): + continue + rc = (getattr(msg, "additional_kwargs", None) or {}).get("reasoning_content") + units.append({ + "content": str(getattr(msg, "content", "") or ""), + "reasoning_content": str(rc or ""), + "tool_calls": [_json_safe(normalize_tool_call(tc)) for tc in (msg.tool_calls or [])], + }) + return units + + def _json_safe(value: Any) -> Any: try: json.dumps(value, ensure_ascii=False) @@ -1348,9 +1384,16 @@ def save_turn(session_id: str, turn_id: str, user_message: dict[str, Any], "duration": (ended_at - started_at) if started_at else None, } # 尾部思考（未被任何工具调用消费的反思）：end_live_turn 暂存 → 此处持久化， - # 供后续轮次 build_native_history_messages 重建原生消息注入上下文。 + # 供前端历史回放/think 面板展示（上下文重建已改由 assistant_units 承载）。 _reflections = _take_stashed_reflections(session_id) - session_storage.save_tool_calls(session_id, turn_id, tool_calls, meta=_meta, reflections=_reflections) + # 每轮模型返回的 AIMessage 三元组（content/reasoning_content/tool_calls）原样保存， + # 供 build_native_history_messages 1:1 重建上下文（前缀缓存一致）。 + _assistant_units = extract_assistant_units(messages) + session_storage.save_tool_calls( + session_id, turn_id, tool_calls, + meta=_meta, reflections=_reflections, + assistant_units=_assistant_units, + ) return None diff --git a/src/agent/tools/__init__.py b/src/agent/tools/__init__.py index f7b8f1a..67207eb 100644 --- a/src/agent/tools/__init__.py +++ b/src/agent/tools/__init__.py @@ -6,7 +6,6 @@ from .human_interaction import HumanInteraction from .todo_list import TodoList from .software_control import SoftwareControl from .gui import GUITool -from .rag import RAGTool from .agent_call import CallSubAgentTool from .doc import DocTool from .skill_router import SkillRouter @@ -28,7 +27,6 @@ _all_tools = [ WebPage(), HumanInteraction(), SoftwareControl(), - RAGTool(), TodoList(), GUITool(), CallSubAgentTool(), @@ -68,3 +66,50 @@ def get_all_available_tool_descriptions(): desc = getattr(inst, "description", "") result[name] = (cls.__name__, (desc or "").split("\n")[0][:200]) return result + + +# ---------- 按需工具（动态注册） ---------- +# 重依赖工具（如 RAG 的 langchain/chromadb）不在启动时加载： +# 仅当 Agent 配置显式引用其名字时，由 ensure_tool_registered 现场实例化并注册。 +def _load_rag_tool(): + """构造 RAGTool（现场导入 langchain/chromadb，仅 rag_tool 被配置引用时执行）。""" + from .rag import RAGTool + return RAGTool() + + +_LAZY_TOOL_LOADERS: dict[str, object] = { + "rag_tool": _load_rag_tool, +} + + +def ensure_tool_registered(tool_name: str): + """确保按需工具已注册（动态注册）。 + + 输入: + tool_name: 工具名字符串（如 "rag_tool"）。 + + 输出: + 已注册的工具实例；非按需工具返回 None。 + + 常规工具启动时静态注册；本函数用于配置显式引用但未静态注册的 + 按需工具，其重依赖不占用启动时间。 + """ + loader = _LAZY_TOOL_LOADERS.get(tool_name) + if loader is None: + return None + existing = next((t for t in _all_tools if t.name == tool_name), None) + if existing is not None: + return existing + tool = loader() + _all_tools.append(tool) + _ALL_AVAILABLE_TOOL_CLASSES[tool.name] = type(tool) + return tool + + +def get_selectable_tool_names(): + """返回前端设置面板可选工具名单（静态注册 + 按需工具名）。 + + 仅返回名称字符串，不触发按需工具加载（langchain/chromadb 等 + 重依赖在 Agent 实际配置引用时才现场注册）。 + """ + return list(_ALL_AVAILABLE_TOOL_CLASSES.keys()) + list(_LAZY_TOOL_LOADERS.keys()) diff --git a/src/agent/tools/rag.py b/src/agent/tools/rag.py index 4d7e43c..e74d6e6 100644 --- a/src/agent/tools/rag.py +++ b/src/agent/tools/rag.py @@ -9,7 +9,6 @@ import os from datetime import datetime from typing import List, Optional, Literal, Union, Tuple -import chromadb import requests from langchain.tools import BaseTool from pydantic import BaseModel, Field, create_model @@ -118,9 +117,14 @@ GLOBAL_COLLECTION_NAME = "rag_global" def get_chroma_client(): - """获取或懒初始化 Chroma 持久化客户端单例。""" + """获取或懒初始化 Chroma 持久化客户端单例。 + + chromadb 仅在真正使用 RAG 工具时才导入（体积大、拖慢应用启动）， + 模块级不再提前加载。 + """ global _chroma_client if _chroma_client is None: + import chromadb _chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR) return _chroma_client diff --git a/src/agent/tools/todo_list.py b/src/agent/tools/todo_list.py index 131e42c..56a9b30 100644 --- a/src/agent/tools/todo_list.py +++ b/src/agent/tools/todo_list.py @@ -84,27 +84,36 @@ def _reconstruct_state(sid: str, steps_arg: Optional[list[str]], done_step_arg: return steps, done_steps -def get_todo_status_text(session_id: str) -> str: - """构建当前会话的 TodoList 状态文本（供每轮注入到模型上下文）。 +def get_todo_status(session_id: str) -> tuple[str, bool]: + """构建当前会话的 TodoList 状态文本与是否全部完成标记。 从已落盘 turn + 当前 live 记录重建累计状态，输出形如： 当前TodoList进度：2/5 已完成 1. [已完成] 打开哔哩哔哩网站 2. [待完成] 点击顶部搜索框，输入"燕三嘤嘤嘤"并点击搜索 - 无计划时返回空字符串。 + 无计划时返回 ("", False)。 + + 返回: (状态文本, 是否全部完成)。全部完成指存在计划且 done_steps 覆盖全部步骤； + 此后调用方不再把状态拼入工具返回值。 """ if not session_id: - return "" + return "", False steps, done_steps = _accumulate_todo_args(_collect_todo_args(session_id)) if not steps: - return "" + return "", False total = len(steps) done_count = len(done_steps) + all_done = done_count >= total lines = [f"当前TodoList进度：{done_count}/{total} 已完成"] for i, s in enumerate(steps): status = "[已完成]" if i in done_steps else "[待完成]" lines.append(f" {i + 1}. {status} {s}") - return "\n".join(lines) + return "\n".join(lines), all_done + + +def get_todo_status_text(session_id: str) -> str: + """构建当前会话的 TodoList 状态文本（兼容入口，供历史注入路径使用）。""" + return get_todo_status(session_id)[0] class TodoListInput(BaseModel): @@ -140,7 +149,7 @@ class TodoList(BaseTool): 调用时会随 live 工具记录推送状态，前端据此实时渲染 Todo 浮窗； 返回值只包含创建/更新结果（成功或失败+原因），**不返回完整步骤列表** - ——当前计划状态由系统每轮注入模型上下文（get_todo_status_text）。 + ——当前计划状态由系统每轮拼入工具返回值供模型感知（见 process_tool_artifact）。 """ args_schema: type[BaseModel] = TodoListInput diff --git a/src/agent/utils/agent_utils.py b/src/agent/utils/agent_utils.py index 4ac4c85..2b9a2b6 100644 --- a/src/agent/utils/agent_utils.py +++ b/src/agent/utils/agent_utils.py @@ -227,63 +227,19 @@ def set_session_thinking_level(session_id: str, level: str) -> None: _save_session_meta(session_id, meta) -# ---------- 动态尾部长期记忆（TodoList 状态 + 循环提醒） ---------- -def update_session_dynamic_tail(session_id: str, agent_key: str, text: str, max_entries: int = 100) -> None: - """持久化当前轮的动态尾部提示（TodoList 状态 + 循环提醒）为长期记忆。 - - 输入: - session_id: 会话 ID。 - agent_key: session_meta 中的 agent 键（主 Agent 为 "main"，子 Agent 用其名称）。 - text: 本轮动态尾部文本。 - max_entries: 历史条目上限（防止无界增长），超出丢弃最旧。 - - 说明: - 独立于 compressed_context 保存（不覆盖）；连续相同内容去重， - 使 todo 进度等状态变更形成可追溯的历史记录，供后续轮次注入查看 - （充分利用前缀缓存：变化部分始终位于消息末尾）。 - """ - if not session_id or not agent_key or not text: - return - meta = _load_session_meta(session_id) - entry = dict(meta.get(agent_key) or {}) - history = list(entry.get("dynamic_tail_history") or []) - if history and history[-1] == text: - return # 去重：内容未变化则不重复记录 - history.append(text) - if len(history) > max_entries: - history = history[-max_entries:] - entry["dynamic_tail_history"] = history - meta[agent_key] = entry - _save_session_meta(session_id, meta) - - -def get_session_dynamic_tail_history(session_id: str, agent_key: str) -> list[str]: - """读取指定 Agent 的动态尾部长期记忆历史（时间正序，最新在末尾）。 - - 输出: - [tail_text, ...]；无记录返回空列表。 - """ - if not session_id or not agent_key: - return [] - meta = _load_session_meta(session_id) - entry = meta.get(agent_key) or {} - return list(entry.get("dynamic_tail_history") or []) - - def build_native_history_messages(session_id: str, msg_cursor: int = 0) -> list: """重建压缩游标之后的历史为原生 LangChain 消息（链式压缩块的输入/输出视图）。 单轮拼接顺序（每条消息 = 1 个重建序号单位，压缩块按此区间计数）: HumanMessage(用户原生 content 部分: text / image_url / file) hist_user_{turn_id} - AIMessage(思考→additional_kwargs.reasoning_content，反思→content， - tool_calls=[真实 id/name/args]) × 每个 assistant 组 hist_assist_{turn_id}_{g} - —— 同一模型响应（AIMessage）的并行调用共享 group_index，合并为一条 + AIMessage(content + reasoning_content + tool_calls) × 每个 assistant 单元 + —— 1:1 复现当轮模型返回的三元组（assistant_units 原样保存） hist_assist_{turn_id}_{i} ToolMessage(result_text, tool_call_id/name 严格匹配) × 每个 call hist_tool_{turn_id}_{g}_{i} - ...未消费的尾部反思（与组内思考/反思去重后） hist_tail_{turn_id}_{i} - AIMessage(最终无工具回复文本) hist_final_{turn_id} + AIMessage(最终无工具回复，三元组无 tool_calls) hist_final_{turn_id}_{i} 所有消息带稳定 id 与 synthetic/tool_ctx 标记（兼容 _filter_short_term_memory - 与压缩节点的 RemoveMessage）。旧记录无 group_index 时每组回退为单 call。 + 与压缩节点的 RemoveMessage）。无 assistant_units 的记录（子 Agent 轨迹、 + abort 轮、旧数据）回退为"按 group_index 拼装 thinking/reflection 字段"的旧逻辑。 输入: session_id: 会话 ID。 @@ -319,7 +275,8 @@ def build_native_history_messages(session_id: str, msg_cursor: int = 0) -> list: record = records_by_turn.get(turn_id) or {} calls = [c for c in (record.get("tool_calls") or []) if isinstance(c, dict)] reflections = record.get("reflections") or [] - for m in _rebuild_turn_messages(turn_id, turn_msgs, calls, reflections): + units = [u for u in (record.get("assistant_units") or []) if isinstance(u, dict)] + for m in _rebuild_turn_messages(turn_id, turn_msgs, calls, reflections, units): if msg_cursor > 0: msg_cursor -= 1 continue @@ -381,8 +338,14 @@ def _tool_result_content(entry: dict) -> str: return "\n".join(x for x in lines if x) -def _rebuild_turn_messages(turn_id: str, turn_msgs: list, calls: list[dict], reflections: list) -> list: - """按原生格式重建单轮消息序列（顺序：user → (组AI+ToolMessage)* → 尾部反思 → 最终回复）。""" +def _rebuild_turn_messages(turn_id: str, turn_msgs: list, calls: list[dict], reflections: list, assistant_units: list[dict] | None = None) -> list: + """按原生格式重建单轮消息序列（顺序：user → (组AI+ToolMessage)* → 尾部反思 → 最终回复）。 + + 存在 assistant_units（当前记录）时按其 1:1 复现当轮 AIMessage 三元组 + （content + reasoning_content + tool_calls），工具结果按 unit 内 tool_calls + 的数量顺序消费 calls —— 不做内容剥离与合成，保证重建字节与当轮转入一致； + 无 assistant_units（子 Agent 轨迹 / abort 轮 / 旧数据）回退为旧拼装逻辑。 + """ from langchain_core.messages import AIMessage, ToolMessage # noqa: F401 msgs: list = [] # 1) 用户消息 @@ -391,6 +354,36 @@ def _rebuild_turn_messages(turn_id: str, turn_msgs: list, calls: list[dict], ref hu = _native_user_message(turn_id, (user_entry.get("user") or {}).get("content") or "") if hu: msgs.append(hu) + + units = [u for u in (assistant_units or []) if isinstance(u, dict)] + if units: + # 三元组 1:1 重建：第 i 个 unit = 第 i 条 AIMessage； + # 带 tool_calls 的 unit 后接与该 unit 等数量、顺序一致的工具结果消息。 + call_iter = iter(calls) + for i, u in enumerate(units): + rc = str(u.get("reasoning_content") or "") + kwargs = {"additional_kwargs": {"synthetic": True, "tool_ctx": True}} + if rc: + kwargs["additional_kwargs"]["reasoning_content"] = rc + content = str(u.get("content") or "") + tc_list = [t for t in (u.get("tool_calls") or []) if isinstance(t, dict)] + if tc_list: + msgs.append(AIMessage(content=content, tool_calls=tc_list, id=f"hist_assist_{turn_id}_{i}", **kwargs)) + for j in range(len(tc_list)): + c = next(call_iter, None) + content_t = _tool_result_content(c) if c else "[无返回内容]" + msgs.append(ToolMessage( + content=content_t if content_t else "[无返回内容]", + tool_call_id=str((c or {}).get("id") or ""), + name=str((c or {}).get("name") or "unknown"), + id=f"hist_tool_{turn_id}_{i}_{j}", + additional_kwargs={"synthetic": True, "tool_ctx": True}, + )) + else: + msgs.append(AIMessage(content=content, id=f"hist_final_{turn_id}_{i}", **kwargs)) + return msgs + + # ---- 回退：无 assistant_units（子 Agent 轨迹 / abort 轮 / 旧数据）---- # 2) 工具调用分组：group_index 相同 → 同一模型响应（并行调用合并为一条 AIMessage） group_order: list = [] group_map: dict = {} diff --git a/src/agent/utils/tool_call_utils.py b/src/agent/utils/tool_call_utils.py index 37f11ac..405907d 100644 --- a/src/agent/utils/tool_call_utils.py +++ b/src/agent/utils/tool_call_utils.py @@ -341,7 +341,7 @@ def format_args_summary(args: dict, max_len: int = 80) -> str: 截断后的 JSON 字符串。 系统定位: - 供 ``core/loop_detector`` 死循环检测、``history/tool_call_recorder`` 工具调用记录展示。 + 供 ``history/tool_call_recorder`` 工具调用记录展示。 """ s = json.dumps(args, ensure_ascii=False, default=str) if len(s) > max_len: diff --git a/src/web/js/toolcalls.js b/src/web/js/toolcalls.js index 3614819..1cd6ccc 100644 --- a/src/web/js/toolcalls.js +++ b/src/web/js/toolcalls.js @@ -22,8 +22,8 @@ const TOOL_META = { gui_tool: { icon: '/image/gui.svg' }, rag: { icon: '/image/rag.svg' }, human_interaction: { icon: '/image/人为干预.svg' }, - call_subagent: { icon: '/image/人机交互.svg', defaultExpanded: true }, - agent_call: { icon: '/image/人机交互.svg', defaultExpanded: true }, + call_subagent: { icon: '/image/人机交互.svg' }, + agent_call: { icon: '/image/人机交互.svg' }, date_query: { icon: '/image/时钟.svg', noParam: true }, cron_manager: { icon: '/image/时钟.svg' }, text2sql: { icon: '/image/SQL.svg' }, @@ -592,7 +592,7 @@ function _renderSubReplyRow(rawText) { const isEmpty = !text.trim(); const row = document.createElement("div"); - row.className = "sub-reply-row" + (isEmpty ? "" : " sub-reply-row--expanded"); + row.className = "sub-reply-row"; const icon = document.createElement("img"); icon.className = "sub-reply-row-icon"; @@ -839,7 +839,9 @@ function _calcLiveDuration(toolCalls, turnStartedAt) { /* ===== 思考行（独立可展开折叠，类似工具调用行） ===== */ function renderThinkRow(content, options = {}) { const isLive = !!options.live; - const defaultExpanded = isLive || !!options.defaultExpanded; + // 思考/反思默认折叠（live 与历史一致）：live 时仅可见 Thought/Reflection 行， + // 内容按需点击展开，避免主 Agent 思考链实时占满气泡 + const defaultExpanded = !!options.defaultExpanded; const index = options.index || 1; const isReflection = options.type === "reflection"; diff --git a/src/web/server.py b/src/web/server.py index eae4697..e43a2cb 100644 --- a/src/web/server.py +++ b/src/web/server.py @@ -535,13 +535,14 @@ async def lifespan(app): """应用生命周期：预热重依赖 + 启动定时任务调度器，退出时关闭。 预热目的: - agent.tools（含 chromadb 等重依赖）与 Agent 运行时原先在首次 - 工具调用/首次打开设置页时惰性导入，首次请求要现场付数秒导入成本。 + agent.tools 重依赖与 Agent 运行时原先在首次工具调用/首次打开设置页 + 时惰性导入，首次请求要现场付数秒导入成本。 这里在启动阶段统一预热（成本被 Electron 健康检查吸收）， 使设置页与首次聊天即时响应。 + chromadb 等按需重型依赖（RAG 工具）保持惰性导入，不占用启动时间。 """ try: - # 预热：构建全部 Agent 运行时（顺带导入 agent.tools/chromadb 等重依赖） + # 预热：构建全部 Agent 运行时（顺带导入 agent.tools 常规依赖） from agent.agents import agent_runtime # noqa: F401 # 预热工具参数缓存（实例化全部工具类读取 args_schema） from agent.core.config_manager import _get_tool_parameters @@ -768,8 +769,11 @@ def api_session_branch( rec = session_storage.get_turn_record(session_id, tid) if rec: _meta = {k: rec[k] for k in ("started_at", "ended_at", "duration", "aborted") if k in rec} + # 保留 AIMessage 三元组与尾部思考：分支会话的历史重建与原会话一致 session_storage.save_tool_calls( - new_sid, tid, rec.get("tool_calls") or [], meta=_meta or None + new_sid, tid, rec.get("tool_calls") or [], meta=_meta or None, + reflections=rec.get("reflections"), + assistant_units=rec.get("assistant_units"), ) except Exception: shutil.rmtree(dst_dir, ignore_errors=True)