修复上下文编排问题、优化子agent消息队列展示UI

This commit is contained in:
minorQAQ committed 2026-08-28 11:32:04 +08:00
1 parent 0ce72f74f1
commit 99f90c2ee8
16 files changed
+369 -481

No files matched your search

+20 -3
View File
@@ -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) { async function createMainWindow(themeMode) {
themeMode = themeMode || getThemeMode(); themeMode = themeMode || getThemeMode();
@@ -350,9 +365,11 @@ async function createMainWindow(themeMode) {
show: false, show: false,
}); });
// 清除缓存后再加载,确保获取最新 JS/CSS // 仅首次运行清除缓存(本地文件与应用页资源不走 HTTP 缓存,无需每次清理)
await mainWindow.webContents.session.clearCache(); if (ensureInitialCacheClear()) {
await mainWindow.webContents.session.clearStorageData({ storages: ['caches', 'serviceworkers'] }); await mainWindow.webContents.session.clearCache();
await mainWindow.webContents.session.clearStorageData({ storages: ['caches', 'serviceworkers'] });
}
// 先显示本地过渡页:毛玻璃 + 中央图标(无任何按钮),主题随 theme_config.json 变化; // 先显示本地过渡页:毛玻璃 + 中央图标(无任何按钮),主题随 theme_config.json 变化;
// FastAPI 就绪后再切到应用页;应用页自带的同款玻璃覆盖层会无缝接手 // FastAPI 就绪后再切到应用页;应用页自带的同款玻璃覆盖层会无缝接手
+5 -1
View File
@@ -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.graph import build_agent_graph
from agent.core.runtime import execute_agent as execute_runtime_agent from agent.core.runtime import execute_agent as execute_runtime_agent
from agent.core.runtime import AgentRuntime 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"] AgentMode = Literal["agent", "plan", "cron"]
@@ -75,6 +75,10 @@ def _build_agent_runtime(cfg: AgentConfig) -> AgentRuntime:
# 2. 获取所有可用工具并按配置过滤 # 2. 获取所有可用工具并按配置过滤
all_tools = get_tools() all_tools = get_tools()
if cfg.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] agent_tools = [t for t in all_tools if t.name in cfg.tools]
else: else:
agent_tools = all_tools agent_tools = all_tools
-1
View File
@@ -99,7 +99,6 @@
"RAG_CHUNK_OVERLAP": "50", "RAG_CHUNK_OVERLAP": "50",
"GROUNDING_WIDTH": "1000", "GROUNDING_WIDTH": "1000",
"GROUNDING_HEIGHT": "1000", "GROUNDING_HEIGHT": "1000",
"LOOP_DETECT_REPEATED_TOOL_WARN": "3",
"SEND_FILE_SIZE_LIMIT": "30", "SEND_FILE_SIZE_LIMIT": "30",
"STORAGE_BACKEND": "json", "STORAGE_BACKEND": "json",
"CRON_TIME_PERIOD_MINUTES": 30, "CRON_TIME_PERIOD_MINUTES": 30,
+7 -4
View File
@@ -453,7 +453,6 @@ def _default_env_config() -> dict:
"RAG_CHUNK_OVERLAP": "50", "RAG_CHUNK_OVERLAP": "50",
"GROUNDING_WIDTH": "1000", "GROUNDING_WIDTH": "1000",
"GROUNDING_HEIGHT": "1000", "GROUNDING_HEIGHT": "1000",
"LOOP_DETECT_REPEATED_TOOL_WARN": "3",
"SEND_FILE_SIZE_LIMIT": "30", "SEND_FILE_SIZE_LIMIT": "30",
# 浏览器控制工具持久 Profile 目录(空 = 每次启动一次性临时 Profile) # 浏览器控制工具持久 Profile 目录(空 = 每次启动一次性临时 Profile)
"BROWSER_PROFILE_DIR": "", "BROWSER_PROFILE_DIR": "",
@@ -511,9 +510,13 @@ def save_env_config(config: dict) -> bool:
# ---------- 获取所有已注册工具的基础信息 ---------- # ---------- 获取所有已注册工具的基础信息 ----------
def get_registered_tool_names() -> list[str]: 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 显示器配置 ---------- # ---------- GUI 显示器配置 ----------
+4 -2
View File
@@ -374,7 +374,8 @@ _SQLITE_SESSION_TABLE_DDLS = [
session_id TEXT NOT NULL, session_id TEXT NOT NULL,
turn_id TEXT NOT NULL, turn_id TEXT NOT NULL,
tool_calls 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)", "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, task_id TEXT NOT NULL,
turn_id TEXT NOT NULL, turn_id TEXT NOT NULL,
tool_calls 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)", "CREATE INDEX IF NOT EXISTS idx_task_tc_turn ON cron_tool_calls (task_id, turn_id)",
-173
View File
@@ -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)
+34 -165
View File
@@ -20,7 +20,6 @@ from typing import Any, Callable
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage 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.core.state import AgentState
from agent.memory.system_prompt import SUB_AGENT_OOM_PROMPT from agent.memory.system_prompt import SUB_AGENT_OOM_PROMPT
from agent.core.llm import LLMRetryExhausted, is_timeout_error 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 裁剪 - 仅 process_tool_artifact 生成的合成图片 HumanMessage 按 trajectory_rounds 裁剪
- 只对实际包含图片截图的轮次计数,无截图的轮次跳过不计数 - 只对实际包含图片截图的轮次计数,无截图的轮次跳过不计数
- 上一/几轮的反思提示词(reflection 标记)不进入本轮上下文 - 上一/几轮的反思提示词(reflection 标记)不进入本轮上下文
- 带 tool_calls 的历史 AIMessage 仅保留工具调用,content 中的思考文本不进入上下文 - 带 tool_calls 的 AIMessage 完整保留(content + reasoning_content + tool_calls
三元组原样进入上下文,不做剥离/合成,保证前缀缓存一致)
""" """
if trajectory_rounds <= 0: if trajectory_rounds <= 0:
return list(messages) return list(messages)
@@ -103,93 +103,7 @@ def _filter_short_term_memory(messages: list, trajectory_rounds: int) -> list:
kept_indices.add(i) kept_indices.add(i)
result = [messages[i] for i in sorted(kept_indices)] return [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 []
# ---------- ReAct 循环内上下文压缩 ---------- # ---------- ReAct 循环内上下文压缩 ----------
@@ -539,58 +453,6 @@ def maybe_compress_context(state: AgentState) -> dict:
return {"_need_compress": False} 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: def _invoke_llm_with_retry(model: Any, messages: list[Any], session_id: str = "") -> Any:
"""调用已绑定工具的模型,LLM 服务超时时自动重试(默认 3 次,LLM_MAX_RETRIES 配置)。 """调用已绑定工具的模型,LLM 服务超时时自动重试(默认 3 次,LLM_MAX_RETRIES 配置)。
@@ -647,12 +509,13 @@ def make_call_model_node(
model_with_tools: Any, model_with_tools: Any,
trajectory_rounds: int = 3, trajectory_rounds: int = 3,
) -> Callable[[AgentState], dict]: ) -> Callable[[AgentState], dict]:
"""创建带按思考档位注入提示 + 上下文中压缩的调用已绑定工具的 LLM 的图节点闭包。 """创建调用已绑定工具的 LLM 的图节点闭包(含上下文中压缩)。
每次调用 LLM 前按思考档位注入提示(思考关闭:反思引导 + TodoList 状态 + 消息上下文 = 图内消息原样(AIMessage 三元组 content/reasoning_content/
循环警告;思考开启:仅 TodoList 状态);调用后提取模型产出的思考/反思内容 tool_calls 完整保留,json 原生透传,不做合成/剥离/尾部注入);
(思考开启取深度思考,思考关闭取一句话反思),记录到 think store, TodoList 状态由 process_tool_artifact 拼入各轮工具返回值。
供前端实时展示思维链。 调用后提取模型产出的思考/反思内容(思考开启取深度思考,思考关闭取
一句话反思),记录到 think store 供前端实时展示思维链。
在每次 LLM 调用后,根据 response 的 ``usage_metadata.total_tokens`` 判断 在每次 LLM 调用后,根据 response 的 ``usage_metadata.total_tokens`` 判断
是否触发消息列表就地压缩——确保 ReAct 循环中的上下文永不溢出。 是否触发消息列表就地压缩——确保 ReAct 循环中的上下文永不溢出。
@@ -675,31 +538,17 @@ def make_call_model_node(
filtered_messages = _filter_short_term_memory(all_messages, trajectory_rounds) filtered_messages = _filter_short_term_memory(all_messages, trajectory_rounds)
# 获取 session_id 供循环检测使用 # 获取 session_id(token 统计、工具调用记录使用)
try: try:
from agent.history.tool_call_recorder import get_current_session from agent.history.tool_call_recorder import get_current_session
_sid = get_current_session() _sid = get_current_session()
except Exception: except Exception:
_sid = "" _sid = ""
# 构建动态尾部提示并注入到消息列表末尾(仅作引导,不记录到前端)。 # 上下文 = 图内消息原样(AIMessage 三元组 content/reasoning_content/tool_calls
# 固定反思引导已并入系统提示词前缀,此处仅注入变化部分: # 完整保留,由 llm._get_request_payload 原生透传,不做合成/剥离/尾部注入)。
# TodoList 状态与循环提醒;无 TodoList 且无循环提醒时不注入(省 token)。 # TodoList 状态已由 process_tool_artifact 拼入各轮工具返回值。
messages_for_llm = list(filtered_messages) 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 与压缩节点不受影响) # 超时自动重试:仅 ReAct 调用的模型调用(工具内部 LLM 与压缩节点不受影响)
response = _invoke_llm_with_retry(model_with_tools, messages_for_llm, _sid) 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): if not isinstance(msg, ToolMessage):
break break
trailing.append(msg) 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): for msg in reversed(trailing):
is_last = msg is trailing[0]
record_tool_result_live( record_tool_result_live(
sid, sid,
msg.name or "unknown", msg.name or "unknown",
str(msg.content) if msg.content else "", str(msg.content) if msg.content else "",
msg.artifact, msg.artifact,
force_update=is_last,
tool_call_id=getattr(msg, "tool_call_id", ""),
) )
except Exception: except Exception:
pass pass
+106 -39
View File
@@ -86,7 +86,7 @@ class ConversationRecordRepository(Protocol):
def list_turn_ids(self, session_id: str) -> list[str]: ... def list_turn_ids(self, session_id: str) -> list[str]: ...
def save_tool_calls( def save_tool_calls(
self, session_id: str, turn_id: str, tool_calls: list[dict], meta: dict | None = None, 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: ... ) -> None: ...
def get_turn_record(self, session_id: str, turn_id: str) -> dict[str, Any] | 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: ... 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, turn_id VARCHAR(64) NOT NULL,
tool_calls JSON NOT NULL, tool_calls JSON NOT NULL,
meta JSON DEFAULT NULL, meta JSON DEFAULT NULL,
assistant_units JSON DEFAULT NULL,
INDEX idx_sess_turn (session_id, turn_id) INDEX idx_sess_turn (session_id, turn_id)
) CHARACTER SET utf8mb4 ) CHARACTER SET utf8mb4
""", """,
@@ -192,6 +193,7 @@ _CRON_TABLE_DDLS = [
task_id VARCHAR(64) NOT NULL, task_id VARCHAR(64) NOT NULL,
turn_id VARCHAR(64) NOT NULL, turn_id VARCHAR(64) NOT NULL,
tool_calls JSON NOT NULL, tool_calls JSON NOT NULL,
assistant_units JSON DEFAULT NULL,
INDEX idx_task_turn (task_id, turn_id) INDEX idx_task_turn (task_id, turn_id)
) CHARACTER SET utf8mb4 ) CHARACTER SET utf8mb4
""", """,
@@ -353,7 +355,7 @@ class JsonRecordRepository:
# ---- 工具调用记录 ---- # ---- 工具调用记录 ----
def save_tool_calls( def save_tool_calls(
self, session_id: str, turn_id: str, tool_calls: list[dict], meta: dict | None = None, 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: ) -> None:
if not session_id or not turn_id: if not session_id or not turn_id:
return return
@@ -369,8 +371,12 @@ class JsonRecordRepository:
record.update(meta) # meta 全量透传,消费方新增字段无需改动存储层 record.update(meta) # meta 全量透传,消费方新增字段无需改动存储层
if reflections: if reflections:
# 尾部思考记录(未被任何工具调用消费的反思),随 turn 一并持久化, # 尾部思考记录(未被任何工具调用消费的反思),随 turn 一并持久化,
# 供后续轮次 build_native_history_messages 重建原生消息注入上下文 # 供前端 think 面板历史回放
record["reflections"] = reflections 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: with open(self._tool_path(session_id, turn_id), "w", encoding="utf-8") as f:
json.dump(record, f, ensure_ascii=False, indent=2) json.dump(record, f, ensure_ascii=False, indent=2)
@@ -596,6 +602,7 @@ class MysqlRecordRepository:
else: else:
_ensure_session_tables() _ensure_session_tables()
self._ensure_extra_column() self._ensure_extra_column()
self._ensure_assistant_units_column()
# ---- 表名 / 连接 ---- # ---- 表名 / 连接 ----
def _messages_table(self) -> str: def _messages_table(self) -> str:
@@ -627,6 +634,21 @@ class MysqlRecordRepository:
finally: finally:
conn.close() 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 @staticmethod
def _loads(raw: Any) -> Any: def _loads(raw: Any) -> Any:
"""pymysql 对 JSON 列可能返回 dict 或 str,统一反序列化。""" """pymysql 对 JSON 列可能返回 dict 或 str,统一反序列化。"""
@@ -773,11 +795,13 @@ class MysqlRecordRepository:
# ---- 工具调用记录 ---- # ---- 工具调用记录 ----
def save_tool_calls( def save_tool_calls(
self, session_id: str, turn_id: str, tool_calls: list[dict], meta: dict | None = None, 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: ) -> None:
if not session_id or not turn_id: if not session_id or not turn_id:
return return
sid_col = self._sid_col() sid_col = self._sid_col()
# 每轮 AIMessage 三元组(content/reasoning_content/tool_calls),独立列存储
units_json = json.dumps(assistant_units or [], ensure_ascii=False)
conn = mysql_conn() conn = mysql_conn()
try: try:
with conn.cursor() as cur: with conn.cursor() as cur:
@@ -789,9 +813,13 @@ class MysqlRecordRepository:
# cron_tool_calls 无 meta 列(与既有表结构一致), # cron_tool_calls 无 meta 列(与既有表结构一致),
# 尾部思考记录无法入库;JSON 后端下完整持久化 # 尾部思考记录无法入库;JSON 后端下完整持久化
cur.execute( cur.execute(
f"INSERT INTO {self._tool_table()} ({sid_col}, turn_id, tool_calls) " f"INSERT INTO {self._tool_table()} "
"VALUES (%s,%s,%s)", 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)), (
session_id, turn_id,
json.dumps(tool_calls or [], ensure_ascii=False),
units_json,
),
) )
else: else:
# 尾部思考并入 meta(_row_to_record 全量透传,读取时合并回记录顶层) # 尾部思考并入 meta(_row_to_record 全量透传,读取时合并回记录顶层)
@@ -799,12 +827,14 @@ class MysqlRecordRepository:
meta = {**(meta or {}), "reflections": reflections} meta = {**(meta or {}), "reflections": reflections}
cur.execute( cur.execute(
f"INSERT INTO {self._tool_table()} " 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, session_id,
turn_id, turn_id,
json.dumps(tool_calls or [], ensure_ascii=False), json.dumps(tool_calls or [], ensure_ascii=False),
json.dumps(meta, ensure_ascii=False) if meta else None, json.dumps(meta, ensure_ascii=False) if meta else None,
units_json,
), ),
) )
conn.commit() conn.commit()
@@ -816,7 +846,7 @@ class MysqlRecordRepository:
meta_key: str | None = None) -> dict[str, Any]: meta_key: str | None = None) -> dict[str, Any]:
"""把工具调用表行还原为与 tool_*.json 同构的 record dict。 """把工具调用表行还原为与 tool_*.json 同构的 record dict。
meta 全量透传(不做字段白名单):消费方新增 meta 字段无需改动存储层。 meta / assistant_units 全量透传(不做字段白名单):消费方新增字段无需改动存储层。
""" """
tool_calls = cls._loads(row.get("tool_calls")) or [] tool_calls = cls._loads(row.get("tool_calls")) or []
record: dict[str, Any] = { record: dict[str, Any] = {
@@ -829,6 +859,9 @@ class MysqlRecordRepository:
meta = cls._loads(row.get(meta_key)) meta = cls._loads(row.get(meta_key))
if isinstance(meta, dict): if isinstance(meta, dict):
record.update(meta) 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 return record
def get_turn_record(self, session_id: str, turn_id: str) -> dict[str, Any] | None: 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: with conn.cursor() as cur:
if self.scope == "cron": if self.scope == "cron":
cur.execute( 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", f"WHERE {sid_col}=%s AND turn_id=%s ORDER BY id",
(session_id, turn_id), (session_id, turn_id),
) )
rows = cur.fetchall() or [] rows = cur.fetchall() or []
out: list[dict] = [] out: list[dict] = []
units: list[dict] = []
for r in rows: for r in rows:
tc = self._loads(r.get("tool_calls")) tc = self._loads(r.get("tool_calls"))
if isinstance(tc, list): if isinstance(tc, list):
out.extend(tc) out.extend(tc)
u = self._loads(r.get("assistant_units"))
if isinstance(u, list):
units.extend(u)
if not out: if not out:
return None return None
return {"session_id": session_id, "turn_id": turn_id, rec = {"session_id": session_id, "turn_id": turn_id,
"tool_calls": out, "timestamp": turn_id} "tool_calls": out, "timestamp": turn_id}
if units:
rec["assistant_units"] = units
return rec
cur.execute( 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", f"WHERE session_id=%s AND turn_id=%s",
(session_id, turn_id), (session_id, turn_id),
) )
@@ -864,6 +904,21 @@ class MysqlRecordRepository:
finally: finally:
conn.close() 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: def get_latest_turn_record(self, session_id: str) -> dict[str, Any] | None:
sid_col = self._sid_col() sid_col = self._sid_col()
conn = mysql_conn() conn = mysql_conn()
@@ -871,19 +926,16 @@ class MysqlRecordRepository:
with conn.cursor() as cur: with conn.cursor() as cur:
if self.scope == "cron": if self.scope == "cron":
cur.execute( 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", f"WHERE {sid_col}=%s ORDER BY turn_id DESC LIMIT 1",
(session_id,), (session_id,),
) )
row = cur.fetchone() row = cur.fetchone()
if not row: if not row:
return None return None
tc = self._loads(row.get("tool_calls")) return self._cron_record_from_row(session_id, row)
return {"session_id": session_id, "turn_id": row.get("turn_id", ""),
"tool_calls": tc if isinstance(tc, list) else [],
"timestamp": row.get("turn_id", "")}
cur.execute( 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", f"WHERE session_id=%s ORDER BY turn_id DESC LIMIT 1",
(session_id,), (session_id,),
) )
@@ -902,20 +954,14 @@ class MysqlRecordRepository:
with conn.cursor() as cur: with conn.cursor() as cur:
if self.scope == "cron": if self.scope == "cron":
cur.execute( 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", f"WHERE {sid_col}=%s ORDER BY turn_id DESC LIMIT %s",
(session_id, limit), (session_id, limit),
) )
rows = cur.fetchall() or [] rows = cur.fetchall() or []
out: list[dict] = [] return [self._cron_record_from_row(session_id, r) for r in rows]
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
cur.execute( 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", f"WHERE session_id=%s ORDER BY turn_id DESC LIMIT %s",
(session_id, limit), (session_id, limit),
) )
@@ -1117,10 +1163,25 @@ class SqliteRecordRepository:
def __init__(self, scope: str = "session", db_path: str | None = None) -> None: def __init__(self, scope: str = "session", db_path: str | None = None) -> None:
self.scope = scope if scope == "cron" else "session" self.scope = scope if scope == "cron" else "session"
self._path_override = db_path self._path_override = db_path
self._db_path_actual = self._db_path()
if self.scope == "cron": 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: 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: def _db_path(self) -> str:
@@ -1297,7 +1358,7 @@ class SqliteRecordRepository:
# ---- 工具调用记录 ---- # ---- 工具调用记录 ----
def save_tool_calls( def save_tool_calls(
self, session_id: str, turn_id: str, tool_calls: list[dict], meta: dict | None = None, 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: ) -> None:
if not session_id or not turn_id: if not session_id or not turn_id:
return return
@@ -1314,12 +1375,14 @@ class SqliteRecordRepository:
) )
conn.execute( conn.execute(
f"INSERT INTO {self._tool_table()} " 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, session_id,
turn_id, turn_id,
self._dumps(tool_calls or []), self._dumps(tool_calls or []),
self._dumps(meta) if meta else None, self._dumps(meta) if meta else None,
self._dumps(assistant_units or []),
), ),
) )
finally: finally:
@@ -1330,7 +1393,7 @@ class SqliteRecordRepository:
meta_key: str | None = None) -> dict[str, Any]: meta_key: str | None = None) -> dict[str, Any]:
"""把工具调用表行还原为与 tool_*.json 同构的 record dict。 """把工具调用表行还原为与 tool_*.json 同构的 record dict。
meta 全量透传(不做字段白名单):消费方新增 meta 字段无需改动存储层。 meta / assistant_units 全量透传(不做字段白名单):消费方新增字段无需改动存储层。
""" """
tool_calls = cls._loads(row.get("tool_calls")) or [] tool_calls = cls._loads(row.get("tool_calls")) or []
record: dict[str, Any] = { record: dict[str, Any] = {
@@ -1343,6 +1406,9 @@ class SqliteRecordRepository:
meta = cls._loads(row.get(meta_key)) meta = cls._loads(row.get(meta_key))
if isinstance(meta, dict): if isinstance(meta, dict):
record.update(meta) 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 return record
def get_turn_record(self, session_id: str, turn_id: str) -> dict[str, Any] | None: def get_turn_record(self, session_id: str, turn_id: str) -> dict[str, Any] | None:
@@ -1350,7 +1416,7 @@ class SqliteRecordRepository:
conn = self._conn() conn = self._conn()
try: try:
row = conn.execute( 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=?", f"WHERE {sid_col}=? AND turn_id=?",
(session_id, turn_id), (session_id, turn_id),
).fetchone() ).fetchone()
@@ -1366,7 +1432,7 @@ class SqliteRecordRepository:
conn = self._conn() conn = self._conn()
try: try:
row = conn.execute( 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", f"WHERE {sid_col}=? ORDER BY turn_id DESC LIMIT 1",
(session_id,), (session_id,),
).fetchone() ).fetchone()
@@ -1383,7 +1449,7 @@ class SqliteRecordRepository:
conn = self._conn() conn = self._conn()
try: try:
rows = conn.execute( 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 ?", f"WHERE {sid_col}=? ORDER BY turn_id DESC LIMIT ?",
(session_id, limit), (session_id, limit),
).fetchall() or [] ).fetchall() or []
@@ -2078,11 +2144,12 @@ def list_turn_ids(session_id: str) -> list[str]:
def save_tool_calls( def save_tool_calls(
session_id: str, turn_id: str, tool_calls: list[dict], meta: dict | None = None, 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: ) -> None:
"""持久化一轮工具调用记录(meta 全量透传;reflections 为尾部思考记录)。""" """持久化一轮工具调用记录(meta 全量透传;reflections 为尾部思考记录;
assistant_units 为每轮 AIMessage 三元组,供历史重建)。"""
try: 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: except Exception as e:
print(f"[storage] save_tool_calls 失败: {e}", file=sys.stderr) print(f"[storage] save_tool_calls 失败: {e}", file=sys.stderr)
+64 -21
View File
@@ -361,8 +361,15 @@ def record_tool_result_live(
name: str, name: str,
result_text: str, result_text: str,
artifact: Any | None = None, artifact: Any | None = None,
force_update: bool = False,
tool_call_id: str = "",
) -> None: ) -> None:
"""记录工具返回结果,供前端实时展示文本、图片预览、音频信息与文件信息。""" """记录工具返回结果,供前端实时展示文本、图片预览、音频信息与文件信息。
默认仅更新 status 仍为 "running" 的条目(幂等);``force_update=True`` 时
按 ``tool_call_id`` 精确匹配并允许覆盖已 done 条目 —— 供 process_tool_artifact
在工具返回值末尾拼接任务进度后强制同步 live 记录。
"""
# 子 Agent 上下文:路由到 _sub_live_records(按 (session, tool_call_id) 隔离) # 子 Agent 上下文:路由到 _sub_live_records(按 (session, tool_call_id) 隔离)
if is_sub_agent_context(): if is_sub_agent_context():
key = _sub_key(session_id) key = _sub_key(session_id)
@@ -389,24 +396,30 @@ def record_tool_result_live(
with _live_lock: with _live_lock:
records = _live_records.get(session_id, []) records = _live_records.get(session_id, [])
for rec in reversed(records): for rec in reversed(records):
if rec["name"] == name and rec["status"] == "running": # force_update:按 tool_call_id 精确匹配(允许覆盖已 done 的返回值)
rec["status"] = "done" if force_update and tool_call_id and rec.get("id") == tool_call_id:
rec["result_text"] = (result_text or "")[:4000] pass
# 图片/文件字节立即落盘为会话目录下的文件,记录只存文件名, elif rec["name"] == name and rec["status"] == "running":
# 不内嵌 base64(实时推送由 get_live_tool_calls 转成 pass
# /api/tool-image URL;aborted 轮次经 save_aborted_turn 落盘时 else:
# 文件已存在,历史回放按 result_files 即可还原) continue
_persist_artifact_files(rec, artifact, records, session_id) rec["status"] = "done"
audio_info = _artifact_audio_info(artifact) rec["result_text"] = (result_text or "")[:4000]
if audio_info: # 图片/文件字节立即落盘为会话目录下的文件,记录只存文件名,
rec["result_audio"] = audio_info # 不内嵌 base64(实时推送由 get_live_tool_calls 转成
file_infos = _artifact_file_infos(artifact) # /api/tool-image URL;aborted 轮次经 save_aborted_turn 落盘时
if file_infos: # 文件已存在,历史回放按 result_files 即可还原)
rec["result_file_info"] = file_infos _persist_artifact_files(rec, artifact, records, session_id)
# 标记变更(增量推送用),并增量持久化到磁盘防止崩溃丢失 audio_info = _artifact_audio_info(artifact)
_live_dirty.setdefault(session_id, set()).add(records.index(rec)) if audio_info:
_schedule_live_persist(session_id) rec["result_audio"] = audio_info
break 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) _notify_live(session_id)
@@ -1226,6 +1239,29 @@ def extract_tool_calls_from_messages(messages: list[BaseMessage]) -> list[dict[s
return records 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: def _json_safe(value: Any) -> Any:
try: try:
json.dumps(value, ensure_ascii=False) 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, "duration": (ended_at - started_at) if started_at else None,
} }
# 尾部思考(未被任何工具调用消费的反思):end_live_turn 暂存 → 此处持久化, # 尾部思考(未被任何工具调用消费的反思):end_live_turn 暂存 → 此处持久化,
# 供后续轮次 build_native_history_messages 重建原生消息注入上下文。 # 供前端历史回放/think 面板展示(上下文重建已改由 assistant_units 承载)。
_reflections = _take_stashed_reflections(session_id) _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 return None
+47 -2
View File
@@ -6,7 +6,6 @@ from .human_interaction import HumanInteraction
from .todo_list import TodoList from .todo_list import TodoList
from .software_control import SoftwareControl from .software_control import SoftwareControl
from .gui import GUITool from .gui import GUITool
from .rag import RAGTool
from .agent_call import CallSubAgentTool from .agent_call import CallSubAgentTool
from .doc import DocTool from .doc import DocTool
from .skill_router import SkillRouter from .skill_router import SkillRouter
@@ -28,7 +27,6 @@ _all_tools = [
WebPage(), WebPage(),
HumanInteraction(), HumanInteraction(),
SoftwareControl(), SoftwareControl(),
RAGTool(),
TodoList(), TodoList(),
GUITool(), GUITool(),
CallSubAgentTool(), CallSubAgentTool(),
@@ -68,3 +66,50 @@ def get_all_available_tool_descriptions():
desc = getattr(inst, "description", "") desc = getattr(inst, "description", "")
result[name] = (cls.__name__, (desc or "").split("\n")[0][:200]) result[name] = (cls.__name__, (desc or "").split("\n")[0][:200])
return result 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())
+6 -2
View File
@@ -9,7 +9,6 @@ import os
from datetime import datetime from datetime import datetime
from typing import List, Optional, Literal, Union, Tuple from typing import List, Optional, Literal, Union, Tuple
import chromadb
import requests import requests
from langchain.tools import BaseTool from langchain.tools import BaseTool
from pydantic import BaseModel, Field, create_model from pydantic import BaseModel, Field, create_model
@@ -118,9 +117,14 @@ GLOBAL_COLLECTION_NAME = "rag_global"
def get_chroma_client(): def get_chroma_client():
"""获取或懒初始化 Chroma 持久化客户端单例。""" """获取或懒初始化 Chroma 持久化客户端单例。
chromadb 仅在真正使用 RAG 工具时才导入(体积大、拖慢应用启动),
模块级不再提前加载。
"""
global _chroma_client global _chroma_client
if _chroma_client is None: if _chroma_client is None:
import chromadb
_chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR) _chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_DIR)
return _chroma_client return _chroma_client
+16 -7
View File
@@ -84,27 +84,36 @@ def _reconstruct_state(sid: str, steps_arg: Optional[list[str]], done_step_arg:
return steps, done_steps return steps, done_steps
def get_todo_status_text(session_id: str) -> str: def get_todo_status(session_id: str) -> tuple[str, bool]:
"""构建当前会话的 TodoList 状态文本(供每轮注入到模型上下文)。 """构建当前会话的 TodoList 状态文本与是否全部完成标记。
从已落盘 turn + 当前 live 记录重建累计状态,输出形如: 从已落盘 turn + 当前 live 记录重建累计状态,输出形如:
当前TodoList进度:2/5 已完成 当前TodoList进度:2/5 已完成
1. [已完成] 打开哔哩哔哩网站 1. [已完成] 打开哔哩哔哩网站
2. [待完成] 点击顶部搜索框,输入"燕三嘤嘤嘤"并点击搜索 2. [待完成] 点击顶部搜索框,输入"燕三嘤嘤嘤"并点击搜索
无计划时返回空字符串。 无计划时返回 ("", False)。
返回: (状态文本, 是否全部完成)。全部完成指存在计划且 done_steps 覆盖全部步骤;
此后调用方不再把状态拼入工具返回值。
""" """
if not session_id: if not session_id:
return "" return "", False
steps, done_steps = _accumulate_todo_args(_collect_todo_args(session_id)) steps, done_steps = _accumulate_todo_args(_collect_todo_args(session_id))
if not steps: if not steps:
return "" return "", False
total = len(steps) total = len(steps)
done_count = len(done_steps) done_count = len(done_steps)
all_done = done_count >= total
lines = [f"当前TodoList进度:{done_count}/{total} 已完成"] lines = [f"当前TodoList进度:{done_count}/{total} 已完成"]
for i, s in enumerate(steps): for i, s in enumerate(steps):
status = "[已完成]" if i in done_steps else "[待完成]" status = "[已完成]" if i in done_steps else "[待完成]"
lines.append(f" {i + 1}. {status} {s}") 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): class TodoListInput(BaseModel):
@@ -140,7 +149,7 @@ class TodoList(BaseTool):
调用时会随 live 工具记录推送状态,前端据此实时渲染 Todo 浮窗; 调用时会随 live 工具记录推送状态,前端据此实时渲染 Todo 浮窗;
返回值只包含创建/更新结果(成功或失败+原因),**不返回完整步骤列表** 返回值只包含创建/更新结果(成功或失败+原因),**不返回完整步骤列表**
——当前计划状态由系统每轮注入模型上下文(get_todo_status_text)。 ——当前计划状态由系统每轮拼入工具返回值供模型感知(见 process_tool_artifact)。
""" """
args_schema: type[BaseModel] = TodoListInput args_schema: type[BaseModel] = TodoListInput
+45 -52
View File
@@ -227,63 +227,19 @@ def set_session_thinking_level(session_id: str, level: str) -> None:
_save_session_meta(session_id, meta) _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: def build_native_history_messages(session_id: str, msg_cursor: int = 0) -> list:
"""重建压缩游标之后的历史为原生 LangChain 消息(链式压缩块的输入/输出视图)。 """重建压缩游标之后的历史为原生 LangChain 消息(链式压缩块的输入/输出视图)。
单轮拼接顺序(每条消息 = 1 个重建序号单位,压缩块按此区间计数): 单轮拼接顺序(每条消息 = 1 个重建序号单位,压缩块按此区间计数):
HumanMessage(用户原生 content 部分: text / image_url / file) hist_user_{turn_id} HumanMessage(用户原生 content 部分: text / image_url / file) hist_user_{turn_id}
AIMessage(思考→additional_kwargs.reasoning_content,反思→content, AIMessage(content + reasoning_content + tool_calls) × 每个 assistant 单元
tool_calls=[真实 id/name/args]) × 每个 assistant 组 hist_assist_{turn_id}_{g} —— 1:1 复现当轮模型返回的三元组(assistant_units 原样保存) hist_assist_{turn_id}_{i}
—— 同一模型响应(AIMessage)的并行调用共享 group_index,合并为一条
ToolMessage(result_text, tool_call_id/name 严格匹配) × 每个 call hist_tool_{turn_id}_{g}_{i} ToolMessage(result_text, tool_call_id/name 严格匹配) × 每个 call hist_tool_{turn_id}_{g}_{i}
...未消费的尾部反思(与组内思考/反思去重后) hist_tail_{turn_id}_{i} AIMessage(最终无工具回复,三元组无 tool_calls) hist_final_{turn_id}_{i}
AIMessage(最终无工具回复文本) hist_final_{turn_id}
所有消息带稳定 id 与 synthetic/tool_ctx 标记(兼容 _filter_short_term_memory 所有消息带稳定 id 与 synthetic/tool_ctx 标记(兼容 _filter_short_term_memory
与压缩节点的 RemoveMessage)。旧记录无 group_index 时每组回退为单 call。 与压缩节点的 RemoveMessage)。无 assistant_units 的记录(子 Agent 轨迹、
abort 轮、旧数据)回退为"按 group_index 拼装 thinking/reflection 字段"的旧逻辑。
输入: 输入:
session_id: 会话 ID。 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 {} record = records_by_turn.get(turn_id) or {}
calls = [c for c in (record.get("tool_calls") or []) if isinstance(c, dict)] calls = [c for c in (record.get("tool_calls") or []) if isinstance(c, dict)]
reflections = record.get("reflections") or [] 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: if msg_cursor > 0:
msg_cursor -= 1 msg_cursor -= 1
continue continue
@@ -381,8 +338,14 @@ def _tool_result_content(entry: dict) -> str:
return "\n".join(x for x in lines if x) 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: 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)* → 尾部反思 → 最终回复)。""" """按原生格式重建单轮消息序列(顺序: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 from langchain_core.messages import AIMessage, ToolMessage # noqa: F401
msgs: list = [] msgs: list = []
# 1) 用户消息 # 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 "") hu = _native_user_message(turn_id, (user_entry.get("user") or {}).get("content") or "")
if hu: if hu:
msgs.append(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) # 2) 工具调用分组:group_index 相同 → 同一模型响应(并行调用合并为一条 AIMessage)
group_order: list = [] group_order: list = []
group_map: dict = {} group_map: dict = {}
+1 -1
View File
@@ -341,7 +341,7 @@ def format_args_summary(args: dict, max_len: int = 80) -> str:
截断后的 JSON 字符串。 截断后的 JSON 字符串。
系统定位: 系统定位:
供 ``core/loop_detector`` 死循环检测、``history/tool_call_recorder`` 工具调用记录展示。 供 ``history/tool_call_recorder`` 工具调用记录展示。
""" """
s = json.dumps(args, ensure_ascii=False, default=str) s = json.dumps(args, ensure_ascii=False, default=str)
if len(s) > max_len: if len(s) > max_len:
+6 -4
View File
@@ -22,8 +22,8 @@ const TOOL_META = {
gui_tool: { icon: '/image/gui.svg' }, gui_tool: { icon: '/image/gui.svg' },
rag: { icon: '/image/rag.svg' }, rag: { icon: '/image/rag.svg' },
human_interaction: { icon: '/image/人为干预.svg' }, human_interaction: { icon: '/image/人为干预.svg' },
call_subagent: { icon: '/image/人机交互.svg', defaultExpanded: true }, call_subagent: { icon: '/image/人机交互.svg' },
agent_call: { icon: '/image/人机交互.svg', defaultExpanded: true }, agent_call: { icon: '/image/人机交互.svg' },
date_query: { icon: '/image/时钟.svg', noParam: true }, date_query: { icon: '/image/时钟.svg', noParam: true },
cron_manager: { icon: '/image/时钟.svg' }, cron_manager: { icon: '/image/时钟.svg' },
text2sql: { icon: '/image/SQL.svg' }, text2sql: { icon: '/image/SQL.svg' },
@@ -592,7 +592,7 @@ function _renderSubReplyRow(rawText) {
const isEmpty = !text.trim(); const isEmpty = !text.trim();
const row = document.createElement("div"); 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"); const icon = document.createElement("img");
icon.className = "sub-reply-row-icon"; icon.className = "sub-reply-row-icon";
@@ -839,7 +839,9 @@ function _calcLiveDuration(toolCalls, turnStartedAt) {
/* ===== 思考行(独立可展开折叠,类似工具调用行) ===== */ /* ===== 思考行(独立可展开折叠,类似工具调用行) ===== */
function renderThinkRow(content, options = {}) { function renderThinkRow(content, options = {}) {
const isLive = !!options.live; const isLive = !!options.live;
const defaultExpanded = isLive || !!options.defaultExpanded; // 思考/反思默认折叠(live 与历史一致):live 时仅可见 Thought/Reflection 行,
// 内容按需点击展开,避免主 Agent 思考链实时占满气泡
const defaultExpanded = !!options.defaultExpanded;
const index = options.index || 1; const index = options.index || 1;
const isReflection = options.type === "reflection"; const isReflection = options.type === "reflection";
+8 -4
View File
@@ -535,13 +535,14 @@ async def lifespan(app):
"""应用生命周期:预热重依赖 + 启动定时任务调度器,退出时关闭。 """应用生命周期:预热重依赖 + 启动定时任务调度器,退出时关闭。
预热目的: 预热目的:
agent.tools(含 chromadb 等重依赖)与 Agent 运行时原先在首次 agent.tools 重依赖与 Agent 运行时原先在首次工具调用/首次打开设置页
工具调用/首次打开设置页时惰性导入,首次请求要现场付数秒导入成本。 时惰性导入,首次请求要现场付数秒导入成本。
这里在启动阶段统一预热(成本被 Electron 健康检查吸收), 这里在启动阶段统一预热(成本被 Electron 健康检查吸收),
使设置页与首次聊天即时响应。 使设置页与首次聊天即时响应。
chromadb 等按需重型依赖(RAG 工具)保持惰性导入,不占用启动时间。
""" """
try: try:
# 预热:构建全部 Agent 运行时(顺带导入 agent.tools/chromadb 等重依赖) # 预热:构建全部 Agent 运行时(顺带导入 agent.tools 常规依赖)
from agent.agents import agent_runtime # noqa: F401 from agent.agents import agent_runtime # noqa: F401
# 预热工具参数缓存(实例化全部工具类读取 args_schema) # 预热工具参数缓存(实例化全部工具类读取 args_schema)
from agent.core.config_manager import _get_tool_parameters 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) rec = session_storage.get_turn_record(session_id, tid)
if rec: if rec:
_meta = {k: rec[k] for k in ("started_at", "ended_at", "duration", "aborted") if k in rec} _meta = {k: rec[k] for k in ("started_at", "ended_at", "duration", "aborted") if k in rec}
# 保留 AIMessage 三元组与尾部思考:分支会话的历史重建与原会话一致
session_storage.save_tool_calls( 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: except Exception:
shutil.rmtree(dst_dir, ignore_errors=True) shutil.rmtree(dst_dir, ignore_errors=True)