// src/agent/runtime/mod.rs // // 科研智能体运行时核心模块。 // 实现 ReAct 循环：Thought -> Action (工具调用) -> Observation -> Thought... // 支持会话持久化、上下文压缩、死循环检测和 SSE 流式输出。 // // 子模块结构： // session — 会话创建/恢复、历史加载 // context — 上下文构建、任务状态恢复 // streaming — LLM 流式响应处理 // executor — 工具调用验证与并行执行 // finalize — 会话收尾、指标持久化 pub mod checkpoint; pub mod circuit_breaker; pub mod context; pub mod denial_tracker; pub mod error_recovery; pub mod executor; pub mod file_cache; pub mod finalize; pub mod hardline; pub mod partitioner; pub mod permission; pub mod permission_explainer; pub mod permission_profile; pub mod session; pub mod streaming; pub mod streaming_executor; pub mod system_prompt; pub mod token_budget; pub mod untrusted; use serde::Serialize; use sqlx::SqlitePool; use std::collections::HashMap; use std::sync::Arc; use tokio::sync::mpsc; use tracing::{error, info, warn}; use super::background::BgNotificationQueue; use super::compact; use super::hooks::{ HookRegistry, SessionStartContext, StepCompleteContext, UserPromptSubmitContext, }; use super::modes::{self, AgentMode, ModeRegistry}; use super::terminal::TurnTerminal; use super::tools::ToolRegistry; use crate::api::AppState; use crate::clients::llm::{ChatMessage, LlmClient, MessageRole, StreamEvent}; use self::error_recovery::{classify_error, ErrorKind, ErrorRecovery}; use self::session::SessionInfo; use self::streaming::{StreamOutput, StreamStatus}; use self::system_prompt::SystemPromptCache; use self::token_budget::TokenBudget; /// Agent 配置参数 #[derive(Debug, Clone)] pub struct AgentConfig { /// 最大 ReAct 迭代次数 pub max_steps: usize, /// 同质调用检测阈值（连续相同调用次数） pub duplicate_call_threshold: usize, /// 工具执行超时时间（秒） pub tool_timeout_secs: u64, /// 工具输出最大字符数 pub max_tool_output_chars: usize, /// 上下文 Token 估算上限（触发自动摘要压缩） pub context_char_limit: usize, /// Token 预算软限制（触发 nudging 提醒） pub token_soft_limit: usize, /// Token 预算硬限制（触发强制动作） pub token_hard_limit: usize, /// 最大消息数（超过此阈值触发 snip_compact 层压缩） pub max_messages: usize, /// 是否启用 LLM 思考模式（前端可控，默认关闭） pub enable_thinking: bool, /// 权限拒绝规则（逗号分隔，格式: ToolName 或 ToolName(content_pattern)） pub permission_deny_rules: Vec, /// 权限允许规则（逗号分隔） pub permission_allow_rules: Vec, /// 权限询问规则（逗号分隔） pub permission_ask_rules: Vec, /// 权限模式: "default" | "accept_edits" | "bypass" | "dont_ask" pub permission_mode: String, /// 拒绝追踪：连续拒绝上限（默认 3） pub denial_max_consecutive: usize, /// 拒绝追踪：总拒绝上限（默认 20） pub denial_max_total: usize, /// 附加允许目录（逗号分隔，扩展文件沙箱范围） pub additional_allowed_dirs: Vec, /// 子代理工具白名单（逗号分隔，空=全部工具可用） pub subagent_allowed_tools: Vec, /// Agent 运行模式 ID（"default" / "deep-research" / "literature-reader"） pub mode: String, } impl AgentConfig { /// 从环境变量加载配置，缺失时使用默认值。 pub fn from_env_optional() -> Self { let mut config = AgentConfig { max_steps: std::env::var("AGENT_MAX_STEPS") .ok() .and_then(|v| v.parse().ok()) .unwrap_or(8), duplicate_call_threshold: 3, tool_timeout_secs: std::env::var("AGENT_TOOL_TIMEOUT_SECS") .ok() .and_then(|v| v.parse().ok()) .unwrap_or(120), max_tool_output_chars: std::env::var("AGENT_MAX_TOOL_OUTPUT_CHARS") .ok() .and_then(|v| v.parse().ok()) .unwrap_or(4000), context_char_limit: std::env::var("AGENT_CONTEXT_CHAR_LIMIT") .ok() .and_then(|v| v.parse().ok()) .unwrap_or(16000), token_soft_limit: std::env::var("AGENT_TOKEN_SOFT_LIMIT") .ok() .and_then(|v| v.parse().ok()) .unwrap_or(32000), token_hard_limit: std::env::var("AGENT_TOKEN_HARD_LIMIT") .ok() .and_then(|v| v.parse().ok()) .unwrap_or(40000), max_messages: std::env::var("AGENT_MAX_MESSAGES") .ok() .and_then(|v| v.parse().ok()) .unwrap_or(50), enable_thinking: false, permission_deny_rules: parse_comma_list("AGENT_PERMISSIONS_DENY"), permission_allow_rules: parse_comma_list("AGENT_PERMISSIONS_ALLOW"), permission_ask_rules: parse_comma_list("AGENT_PERMISSIONS_ASK"), permission_mode: std::env::var("AGENT_PERMISSION_MODE") .unwrap_or_else(|_| "default".to_string()), denial_max_consecutive: std::env::var("AGENT_DENIAL_MAX_CONSECUTIVE") .ok() .and_then(|v| v.parse().ok()) .unwrap_or(3), denial_max_total: std::env::var("AGENT_DENIAL_MAX_TOTAL") .ok() .and_then(|v| v.parse().ok()) .unwrap_or(20), additional_allowed_dirs: parse_comma_list("AGENT_ADDITIONAL_DIRS"), subagent_allowed_tools: parse_comma_list("AGENT_SUBAGENT_ALLOWED_TOOLS"), mode: std::env::var("AGENT_MODE").unwrap_or_else(|_| "default".to_string()), }; // 加载权限档案（AGENT_PERMISSION_PROFILE），追加到现有规则 let profile_name = std::env::var("AGENT_PERMISSION_PROFILE").unwrap_or_default(); if !profile_name.is_empty() { if let Some(profile) = permission_profile::load_profile(&profile_name) { info!( "[AgentConfig] 加载权限档案: {} — {}", profile.name, profile.description ); permission_profile::apply_profile_to_config( &profile, &mut config.permission_deny_rules, &mut config.permission_allow_rules, &mut config.permission_ask_rules, &mut config.permission_mode, ); } else { warn!( "[AgentConfig] 未知的权限档案: {}（可用: {:?}）", profile_name, permission_profile::list_available_profiles() ); } } config } } /// 解析逗号分隔的环境变量为字符串列表 fn parse_comma_list(env_key: &str) -> Vec { std::env::var(env_key) .ok() .map(|v| { v.split(',') .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()) .collect() }) .unwrap_or_default() } impl Default for AgentConfig { fn default() -> Self { Self::from_env_optional() } } // ── SSE Stream Events ── /// SSE 流式事件（发送给前端） #[derive(Debug, Clone, Serialize)] #[serde(tag = "type")] pub enum AgentStreamEvent { /// 会话创建/恢复 #[serde(rename = "session")] Session { session_id: String, title: String }, /// 智能体思考过程 #[serde(rename = "thought")] Thought { content: String, step: usize }, /// 工具调用开始 #[serde(rename = "tool_call")] ToolCall { /// LLM 生成的工具调用 ID，用于全链路关联（前端/审计/持久化） id: String, name: String, arguments: serde_json::Value, step: usize, }, /// 工具执行结果（Observation） #[serde(rename = "tool_result")] ToolResult { /// 对应的工具调用 ID，前端凭此精确匹配 tool_call 条目 tool_call_id: String, name: String, output: String, is_error: bool, metadata: serde_json::Value, step: usize, }, /// 文本增量流式输出（最终回答或工具流式输出） #[serde(rename = "text_delta")] TextDelta { content: String, /// 可选：工具调用 ID。当 set 时，此增量属于对应工具的流式输出， /// 前端应将其渲染到工具结果区域而非主文本区。 #[serde(skip_serializing_if = "Option::is_none")] tool_call_id: Option, }, /// Token 使用统计 #[serde(rename = "usage")] Usage { prompt_tokens: u32, completion_tokens: u32, total_tokens: u32, }, /// 错误通知 #[serde(rename = "error")] Error { message: String }, /// 权限请求（需要用户确认工具执行） #[serde(rename = "permission_request")] PermissionRequest { tool_call_id: String, tool_name: String, message: String, arguments: serde_json::Value, /// 可选的权限风险解释（参考 Claude Code permissionExplainer） explanation: Option, }, /// 权限响应已处理 #[serde(rename = "permission_response")] PermissionResponse { tool_call_id: String, allowed: bool }, /// 完成标记 #[serde(rename = "done")] Done, } // ── Metrics & Detection ── /// Agent 运行指标 #[derive(Debug, Default, Serialize)] pub struct AgentMetrics { pub total_steps: usize, pub compression_count: usize, pub duplicate_detections: usize, /// 各工具调用次数统计 pub tool_calls: HashMap, } /// 同质调用检测器 #[derive(Debug, Default)] pub struct DuplicateDetector { last_call: Option<(String, String)>, // (tool_name, arguments) consecutive_count: usize, } impl DuplicateDetector { /// 记录一次调用，返回是否检测到死循环 pub fn record(&mut self, tool_name: &str, arguments: &str, threshold: usize) -> bool { let key = (tool_name.to_string(), arguments.to_string()); if self.last_call.as_ref() == Some(&key) { self.consecutive_count += 1; if self.consecutive_count >= threshold { return true; } } else { self.last_call = Some(key); self.consecutive_count = 1; } false } } // ── Agent Runtime ── /// 智能体运行时 pub struct AgentRuntime { app_state: Arc, config: AgentConfig, tool_registry: ToolRegistry, /// 后台任务通知队列（支持 bg_task_run/bg_task_check） bg_notification_queue: Arc, /// 指标采集 hook 的数据引用（供 API 查询） metrics_data: Arc>, /// 压缩熔断器（跨 turn 共享，防止无限压缩循环） compaction_breaker: Arc>, /// 权限检查器 permission_checker: Arc, /// 拒绝追踪器（跨 turn 共享） denial_tracker: Arc>, /// 文件状态缓存（跨 turn 共享，用于 Read 去重） read_file_state: Arc>, /// 系统提示词 section 缓存（跨 turn 共享，避免每 turn 重建静态/低频变动内容） prompt_cache: std::sync::Mutex, /// 上下文压缩折叠日志（跨 turn 共享，追踪压缩历史并触发溢出合并） collapse_log: Arc>, /// Checkpoint 管理器（跨 turn 共享，文件变更操作前自动快照） checkpoint_manager: Arc, /// 当前运行模式（从 ModeRegistry 解析的静态引用） mode: &'static AgentMode, /// 模式注册表（持有所有已注册模式） mode_registry: ModeRegistry, } impl AgentRuntime { /// 创建新的运行时实例 pub fn new(app_state: Arc) -> Self { let mut config = AgentConfig::default(); let mode_registry = ModeRegistry::builtins(); let mode = mode_registry.get(&config.mode).copied().unwrap_or_else(|| { tracing::warn!("[AgentRuntime] 未知模式 '{}'，回退到默认模式", config.mode); mode_registry .get(ModeRegistry::default_id()) .copied() .unwrap_or(&modes::default::DEFAULT_MODE) }); // 合并模式配置预设到 AgentConfig apply_mode_config(&mut config, mode); let queue = Arc::new(BgNotificationQueue::new()); let metrics_data = Arc::new(std::sync::Mutex::new(super::hooks::MetricsData::default())); let permission_checker = Arc::new(permission::PermissionChecker::from_config(&config)); let denial_tracker = Arc::new(std::sync::Mutex::new(denial_tracker::DenialTracker::new( config.denial_max_consecutive, config.denial_max_total, ))); let skill_registry = app_state.skill_registry.clone(); let mut tool_registry = ToolRegistry::new_with_queue(Some(queue.clone()), skill_registry); // 注册记忆工具 tool_registry.add_tool(Box::new(crate::agent::tools::memory::SaveMemoryTool::new( app_state.memory_manager.clone(), ))); // 替换 DelegateResearchTool 为带有 permission_checker 的版本（SSE 通道通过 ToolContext 注入） tool_registry.replace_tool(Box::new( crate::agent::tools::subagent::SubAgentTool::new_with_hooks( None, permission_checker.clone(), ), )); // 初始化会话级权限检查器（与 AgentRuntime 使用相同的环境变量规则） if let Ok(mut session_checker) = app_state.session_permission_checker.write() { *session_checker = (*permission_checker).clone(); } // 视觉模型可用时注册 analyze_image 工具 if app_state.vision_llm.is_some() { tool_registry.add_tool(Box::new(crate::agent::tools::astro::AnalyzeImageTool)); } // ── 应用模式的工具集过滤 ── apply_mode_tool_filter(&mut tool_registry, mode); // 初始化 checkpoint 管理器 let checkpoint_enabled = std::env::var("AGENT_CHECKPOINT_ENABLED") .unwrap_or_else(|_| "true".to_string()) .to_lowercase() != "false"; let checkpoint_store = app_state.config.library_dir.join("..").join(".checkpoints"); let checkpoint_manager = Arc::new(checkpoint::CheckpointManager::new( std::fs::canonicalize(&checkpoint_store).unwrap_or(checkpoint_store), checkpoint_enabled, )); AgentRuntime { app_state, config, tool_registry, bg_notification_queue: queue, metrics_data, compaction_breaker: Arc::new(std::sync::Mutex::new( circuit_breaker::CompactionCircuitBreaker::new(), )), permission_checker, denial_tracker, read_file_state: Arc::new(std::sync::Mutex::new(file_cache::FileStateCache::new())), prompt_cache: std::sync::Mutex::new(SystemPromptCache::new()), collapse_log: Arc::new(std::sync::Mutex::new(compact::collapse::CollapseLog::new())), checkpoint_manager, mode, mode_registry, } } /// 创建带自定义配置的运行时实例 pub fn with_config(app_state: Arc, mut config: AgentConfig) -> Self { let mode_registry = ModeRegistry::builtins(); let mode = mode_registry.get(&config.mode).copied().unwrap_or_else(|| { tracing::warn!("[AgentRuntime] 未知模式 '{}'，回退到默认模式", config.mode); mode_registry .get(ModeRegistry::default_id()) .copied() .unwrap_or(&modes::default::DEFAULT_MODE) }); // 合并模式配置预设 apply_mode_config(&mut config, mode); let queue = Arc::new(BgNotificationQueue::new()); let metrics_data = Arc::new(std::sync::Mutex::new(super::hooks::MetricsData::default())); let permission_checker = Arc::new(permission::PermissionChecker::from_config(&config)); let denial_tracker = Arc::new(std::sync::Mutex::new(denial_tracker::DenialTracker::new( config.denial_max_consecutive, config.denial_max_total, ))); let skill_registry = app_state.skill_registry.clone(); let mut tool_registry = ToolRegistry::new_with_queue(Some(queue.clone()), skill_registry); tool_registry.add_tool(Box::new(crate::agent::tools::memory::SaveMemoryTool::new( app_state.memory_manager.clone(), ))); tool_registry.replace_tool(Box::new( crate::agent::tools::subagent::SubAgentTool::new_with_hooks( None, permission_checker.clone(), ), )); // 初始化会话级权限检查器 if let Ok(mut session_checker) = app_state.session_permission_checker.write() { *session_checker = (*permission_checker).clone(); } // 视觉模型可用时注册 analyze_image 工具 if app_state.vision_llm.is_some() { tool_registry.add_tool(Box::new(crate::agent::tools::astro::AnalyzeImageTool)); } // ── 应用模式的工具集过滤 ── apply_mode_tool_filter(&mut tool_registry, mode); // 初始化 checkpoint 管理器 let checkpoint_enabled = std::env::var("AGENT_CHECKPOINT_ENABLED") .unwrap_or_else(|_| "true".to_string()) .to_lowercase() != "false"; let checkpoint_store = app_state.config.library_dir.join("..").join(".checkpoints"); let checkpoint_manager = Arc::new(checkpoint::CheckpointManager::new( std::fs::canonicalize(&checkpoint_store).unwrap_or(checkpoint_store), checkpoint_enabled, )); AgentRuntime { app_state, config, tool_registry, bg_notification_queue: queue, metrics_data, compaction_breaker: Arc::new(std::sync::Mutex::new( circuit_breaker::CompactionCircuitBreaker::new(), )), permission_checker, denial_tracker, read_file_state: Arc::new(std::sync::Mutex::new(file_cache::FileStateCache::new())), prompt_cache: std::sync::Mutex::new(SystemPromptCache::new()), collapse_log: Arc::new(std::sync::Mutex::new(compact::collapse::CollapseLog::new())), checkpoint_manager, mode, mode_registry, } } /// 返回当前运行指标快照（锁异常时返回默认值） pub fn get_metrics(&self) -> super::hooks::MetricsData { self.metrics_data .lock() .ok() .map(|m| m.clone()) .unwrap_or_default() } /// 设置是否启用 LLM 思考模式（向后兼容，优先使用 mode 设置） pub fn with_thinking(mut self, enable: bool) -> Self { self.config.enable_thinking = enable; self } /// 设置运行模式（覆盖 AgentConfig 中的 mode 字段）。 /// /// 调用此方法会重新解析模式并应用对应的配置预设、工具过滤和权限档案。 pub fn with_mode(mut self, mode_id: &str) -> Self { self.config.mode = mode_id.to_string(); let mode = self.mode_registry.get(mode_id).copied().unwrap_or_else(|| { tracing::warn!( "[AgentRuntime] with_mode: 未知模式 '{}'，回退到默认模式", mode_id ); self.mode_registry .get(ModeRegistry::default_id()) .copied() .unwrap_or(&modes::default::DEFAULT_MODE) }); apply_mode_config(&mut self.config, mode); apply_mode_tool_filter(&mut self.tool_registry, mode); self.mode = mode; self } /// 返回当前模式的 ID。 pub fn mode_id(&self) -> &str { self.mode.id } /// 返回模式是否强制固定了 thinking。`Some(true/false)` 表示模式已锁定，用户不可覆盖。 pub fn mode_fixed_thinking(&self) -> Option { self.mode.mode_config.enable_thinking } // ── Private Helpers ── /// 执行文件缓存快照 → 压缩 → 恢复 → 上下文注入 的完整周期。 /// 返回压缩前的消息数（用于调用者判断压缩是否有效）。 async fn snapshot_compress_restore( &self, messages: &mut Vec, llm: &LlmClient, session_id: &str, hook_registry: &HookRegistry, ) -> usize { let before_len = messages.len(); // ── 文件缓存快照（压缩前）── let file_snapshot = { if let Ok(mut cache) = self.read_file_state.lock() { let snap = cache.to_snapshot(); cache.clear(); snap } else { Vec::new() } }; // 压缩前捕获消息快照（用于记忆提取桥接，P3） let pre_compact_snapshot: Vec = messages.to_vec(); compact::compress_context_with_hooks_and_log( messages, llm, self.config.context_char_limit, session_id, Some(hook_registry), Some(&self.collapse_log), ) .await; // 压缩后提取记忆（P3 桥接：将丢弃的消息内容喂给记忆提取子代理） compact::extract_memories_from_compaction( &pre_compact_snapshot, session_id, self.app_state.memory_manager.clone(), self.app_state.clone(), ); // ── 文件缓存恢复（压缩后：重新注入最近文件 + 恢复缓存）── { if let Ok(mut cache) = self.read_file_state.lock() { cache.restore_from_snapshot( &file_snapshot, file_cache::POST_COMPACT_MAX_FILES_TO_RESTORE, ); } let restore_ctx = file_cache::FileStateCache::build_restore_context( &file_snapshot, file_cache::POST_COMPACT_MAX_FILES_TO_RESTORE, ); for block in restore_ctx { messages.push(ChatMessage::user(format!("[压缩后上下文恢复]\n{}", block))); } } before_len } // ── Public API ── /// 执行完整的智能体对话回合（流式 SSE 输出）。 /// /// 流程： /// 1. 创建/恢复会话 /// 2. 构建消息上下文 /// 3. ReAct 循环 /// 4. 会话收尾 pub async fn run_turn( &self, session_id: Option, question: &str, tx: mpsc::UnboundedSender, ) -> anyhow::Result { self.run_turn_with_image_context(session_id, question, None, None, tx) .await } /// 带图片上下文的对话回合。`image_context` 会在用户消息前注入为 system-reminder。 /// `image_path` 会存入用户消息的 metadata 以便前端渲染。 pub async fn run_turn_with_image_context( &self, session_id: Option, question: &str, image_context: Option, image_path: Option, tx: mpsc::UnboundedSender, ) -> anyhow::Result { let db = &self.app_state.db; let llm = &self.app_state.llm; // Phase 1: 创建或恢复会话 let session_info = session::create_or_resume_session(db, session_id.clone(), llm, &self.config.mode) .await?; // 构建 hook 注册表（注入依赖，复用 AgentRuntime 的 metrics_data） let hook_registry = HookRegistry::with_builtins( db.clone(), self.app_state.cancelled_runs.clone(), Some(self.metrics_data.clone()), ); // 触发 OnSessionStart hook_registry .run_on_session_start(&SessionStartContext { session_id: session_info.session_id.clone(), turn_index: session_info.turn_index, is_resume: session_id.is_some(), }) .await; // 触发 UserPromptSubmit —— hooks 可在上下文构建前检查/记录用户输入 hook_registry .run_on_user_prompt_submit(&UserPromptSubmitContext { session_id: session_info.session_id.clone(), prompt: question.to_string(), turn_index: session_info.turn_index, }) .await; let _ = tx.send(AgentStreamEvent::Session { session_id: session_info.session_id.clone(), title: String::new(), }); // Phase 2: 构建初始上下文（历史 + system prompt + 用户消息 + 任务恢复） let mut messages = context::build_initial_context( db, &session_info.session_id, &self.system_prompt(), question, session_info.turn_index, ) .await?; // 图片上下文：在用户消息前注入 system-reminder if let Some(ref img_ctx) = image_context { let reminder = format!("\n{}\n", img_ctx); // 插入到倒数第二条位置（用户消息之前） let user_msg = messages.pop().unwrap(); // 用户消息 messages.push(ChatMessage::user(&reminder)); messages.push(user_msg); } // 保存用户消息到数据库（含图片路径元数据，供前端历史渲染） let user_metadata = image_path .as_ref() .map(|p| serde_json::json!({"image_path": p})); self.save_message( db, &session_info.session_id, session_info.turn_index, 0, &ChatMessage::user(question), None, user_metadata, ) .await?; // Phase 3: ReAct 循环 let (metrics, loop_terminal) = self .run_react_loop(&session_info, &mut messages, &tx, &hook_registry) .await?; // Phase 4: 会话收尾（传入实际的终止原因 + trajectory 导出参数） let system_prompt = self.system_prompt(); let model_name = self.app_state.llm.model(); finalize::finalize_turn( db, &session_info.session_id, &metrics, question, &tx, &hook_registry, loop_terminal, Some(&self.app_state.config.library_dir), Some(&model_name), Some(&system_prompt), Some(self.app_state.clone()), ) .await?; Ok(session_info.session_id) } // ── ReAct Loop ── /// ReAct 循环核心：LLM 调用 → 工具执行 → 结果注入 → 循环... async fn run_react_loop( &self, session_info: &SessionInfo, messages: &mut Vec, tx: &mpsc::UnboundedSender, hook_registry: &HookRegistry, ) -> anyhow::Result<(AgentMetrics, Option)> { let db = &self.app_state.db; let llm = &self.app_state.llm; let sid = &session_info.session_id; let turn_index = session_info.turn_index; let tool_defs = self.tool_registry.definitions(); let mut duplicate_detector = DuplicateDetector::default(); let mut metrics = AgentMetrics::default(); let mut step = 0; let mut loop_terminal: Option = None; // Token 追踪（API 精确值优先，字符估算作近似值） let mut last_api_prompt_tokens: Option = None; let mut msg_count_at_last_call: usize = messages.len(); let mut steps_since_last_todo: usize = 0; let nag_after_steps: usize = 3; let mut pending_manual_compress: bool = false; // Token 预算管理器（用于 diminishing returns 检测和 error recovery） let mut token_budget = TokenBudget::new(self.config.token_soft_limit, self.config.token_hard_limit); loop { step += 1; // ── Checkpoint: 每个 ReAct 迭代开始时重置去重状态 ── self.checkpoint_manager.new_turn(); // 检查用户取消 let is_cancelled = { if let Ok(mut cancelled) = self.app_state.cancelled_runs.lock() { cancelled.remove(sid) } else { false } }; if is_cancelled { warn!("[AgentRuntime] 用户手动中止了会话 {} 的智能体执行", sid); let _ = tx.send(AgentStreamEvent::Error { message: "用户已手动中止执行。".to_string(), }); loop_terminal = Some(TurnTerminal::CancelledByUser { session_id: sid.clone(), at_step: step, }); break; } // ── 上下文压缩检查 ── // ── Token 感知的压缩触发 ── // 优先使用 API 返回的精确 prompt_tokens，辅以简单的增量估算 let estimated_tokens = match last_api_prompt_tokens { Some(last_tokens) => { let new_msg_count = messages.len().saturating_sub(msg_count_at_last_call); let new_tokens_estimate: u32 = messages .iter() .rev() .take(new_msg_count) .map(|m| (m.content.as_ref().map_or(0, |c| c.len()) + 4) as u32) .sum(); (last_tokens + new_tokens_estimate) as usize } None => compact::rough_estimate_tokens(messages), }; // 使用 token 预算的软限制作为压缩触发点（而非粗糙的 context_char_limit * 1.5） let token_limit = token_budget.soft_limit; let mut did_compress = false; // 熔断器检查：如果连续压缩失败多次，跳过自动压缩 let breaker_ok = match self.compaction_breaker.lock() { Ok(mut breaker) => breaker.can_attempt(), Err(e) => { warn!("[AgentRuntime] 熔断器锁异常，跳过自动压缩: {:?}", e); false } }; if estimated_tokens > token_limit && breaker_ok { info!( "[AgentRuntime] 上下文超限 (est. {} tokens > {} limit)，触发压缩", estimated_tokens, token_limit ); let before_len = self .snapshot_compress_restore( messages, llm, &session_info.session_id, hook_registry, ) .await; // 熔断器反馈：压缩后消息数减少 = 成功 if let Ok(mut breaker) = self.compaction_breaker.lock() { if messages.len() < before_len { breaker.record_success(); } else { breaker.record_failure(); } } else { warn!("[AgentRuntime] 熔断器反馈写入失败（锁异常）"); } last_api_prompt_tokens = None; msg_count_at_last_call = messages.len(); metrics.compression_count += 1; did_compress = true; } else if estimated_tokens > token_limit && !breaker_ok { warn!("[AgentRuntime] 熔断器已打开，跳过自动压缩"); } // 处理手动压缩请求（跳过刚自动压缩过的情况，避免双重压缩） // 手动压缩不受熔断器限制 if pending_manual_compress && !did_compress { pending_manual_compress = false; info!("[AgentRuntime] 执行手动压缩（compress_context 工具触发）"); self.snapshot_compress_restore( messages, llm, &session_info.session_id, hook_registry, ) .await; // 手动压缩成功后重置熔断器 if let Ok(mut breaker) = self.compaction_breaker.lock() { breaker.reset(); } last_api_prompt_tokens = None; msg_count_at_last_call = messages.len(); metrics.compression_count += 1; } else if pending_manual_compress { pending_manual_compress = false; info!("[AgentRuntime] 跳过手动压缩（刚已完成自动压缩）"); } // Token 预算 diminishing returns 检测 + 渐进式 nudge 提醒 token_budget.record_continuation(); let mut should_nudge = false; // TodoWrite nag reminder if steps_since_last_todo >= nag_after_steps { messages.push(ChatMessage::user( "提醒：你已经连续多步未更新任务计划。建议调用 todo_write 工具复盘当前进度并规划后续步骤。", )); steps_since_last_todo = 0; should_nudge = true; } // Token 预算 nudge（仅在无 nag 时注入，避免消息过多） if !should_nudge { if let Some(nudge) = token_budget.nudge_message() { messages.push(ChatMessage::user(nudge)); } } // Diminishing returns 检测 — 强制结束 if token_budget.diminishing_returns { warn!("[AgentRuntime] 检测到 diminishing returns，强制结束循环"); let _ = tx.send(AgentStreamEvent::Error { message: "检测到重复操作模式，已自动停止。请查看已收集的信息。".to_string(), }); messages.push(ChatMessage::user( "检测到你的后续步骤未产生新信息（diminishing returns）。\ 请基于已收集的全部信息直接给出最终答案，不要再调用任何工具。", )); let _ = self .final_answer_without_tools(llm, messages, sid, turn_index, step, tx) .await; break; } // 最大步数检查 if step > self.config.max_steps { warn!( "[AgentRuntime] 达到最大步数限制 ({} steps)", self.config.max_steps ); let _ = tx.send(AgentStreamEvent::Error { message: format!( "已达到最大推理步数 ({})，请根据已收集的信息给出最终回答。", self.config.max_steps ), }); messages.push(ChatMessage::user(format!( "你已经执行了 {} 步（最大 {} 步）。请根据已有信息直接给出最终答案，不要再调用工具。", step, self.config.max_steps ))); let _ = self .final_answer_without_tools(llm, messages, sid, turn_index, step, tx) .await; break; } // ── 后台任务通知注入 ── let bg_results = self.bg_notification_queue.drain().await; for result in bg_results { let status = if result.is_error { "❌" } else { "✅" }; messages.push(ChatMessage::user(format!( "[后台任务完成] {} {}: {} ({}): {}", status, result.tool_name, result.bibcode, result.task_id, result.summary, ))); } // ── LLM 流式调用（含错误恢复） ── let stream_output_opt = self .call_llm_with_recovery(llm, messages, &tool_defs, tx, step, sid, &mut token_budget) .await; let stream_output = match stream_output_opt { Some(output) => output, None => { // 所有恢复尝试均失败 loop_terminal = Some(TurnTerminal::ModelError { session_id: sid.clone(), message: "LLM 调用失败，所有恢复步骤已尝试完毕".to_string(), }); break; } }; // 更新 API 精确 token 计数 + token 预算 if let Some(ref u) = stream_output.usage { last_api_prompt_tokens = Some(u.prompt_tokens); msg_count_at_last_call = messages.len(); token_budget.spend_input(u.prompt_tokens as usize); token_budget.spend_output(u.completion_tokens as usize); } // ── 处理 Thought/Reasoning ── let mut thought_content = stream_output.reasoning.clone(); if thought_content.is_none() && stream_output.is_tool_call_step && !stream_output.content.is_empty() { thought_content = Some(stream_output.content.clone()); } if stream_output.is_tool_call_step { if let Some(ref thought_text) = thought_content { let _ = tx.send(AgentStreamEvent::Thought { content: thought_text.clone(), step, }); } } // ── 无工具调用 = 最终回答 ── let mut tool_calls = match stream_output.tool_calls { Some(ref tc) if !tc.is_empty() => tc.clone(), _ => { // 保存最终回答 let assistant_msg = ChatMessage::assistant_with_reasoning( if stream_output.content.is_empty() { None } else { Some(stream_output.content.clone()) }, stream_output.reasoning.clone(), None, ); self.save_message( db, sid, turn_index, step as i32, &assistant_msg, stream_output.reasoning.as_deref(), None, ) .await?; messages.push(assistant_msg); // 发送 reasoning（当模型思考后直接给出答案、未调用工具时， // thought 尚未在上面的 is_tool_call_step 块中发送） if let Some(ref thought_text) = stream_output.reasoning { let _ = tx.send(AgentStreamEvent::Thought { content: thought_text.clone(), step, }); } // Token 使用统计 if let Some(u) = stream_output.usage { let _ = tx.send(AgentStreamEvent::Usage { prompt_tokens: u.prompt_tokens, completion_tokens: u.completion_tokens, total_tokens: u.total_tokens, }); } break; } }; // 修复空 ID（LLM 可能不返回 tool_call id） for tc in tool_calls.iter_mut() { if tc.id.is_empty() { tc.id = format!("call_{}", &uuid::Uuid::new_v4().to_string()[..8]); } } // ── 工具调用处理 ── // 检测 todo_write 和 compress_context let called_todo_write = tool_calls.iter().any(|tc| tc.function.name == "todo_write"); if called_todo_write { steps_since_last_todo = 0; } else { steps_since_last_todo += 1; } if tool_calls .iter() .any(|tc| tc.function.name == "compress_context") { pending_manual_compress = true; } // 更新指标 metrics.total_steps = step; for tc in &tool_calls { *metrics .tool_calls .entry(tc.function.name.clone()) .or_insert(0) += 1; } // 构建 assistant 消息（含 tool_calls） let assistant_msg = ChatMessage::assistant_with_reasoning( if stream_output.content.is_empty() { None } else { Some(stream_output.content.clone()) }, stream_output.reasoning.clone(), Some(tool_calls.clone()), ); self.save_message( db, sid, turn_index, step as i32, &assistant_msg, stream_output.reasoning.as_deref(), None, ) .await?; messages.push(assistant_msg); // 验证 + 准备工具调用 let (prepared_calls, has_duplicate) = executor::validate_and_prepare( &tool_calls, &mut duplicate_detector, self.config.duplicate_call_threshold, messages, tx, db, sid, turn_index, step, ); if has_duplicate { metrics.duplicate_detections += 1; continue; } if prepared_calls.is_empty() { continue; } // 并行执行工具（带权限检查、checkpoint 和分区器） let exec_result = executor::execute_parallel( &prepared_calls, &self.tool_registry, self.app_state.clone(), hook_registry, Some(&self.permission_checker), Some(&self.app_state.session_permission_checker), Some(&self.denial_tracker), Some(&self.checkpoint_manager), tx, db, sid, "lead", turn_index, step, self.config.tool_timeout_secs, self.config.max_tool_output_chars, self.read_file_state.clone(), self.config.enable_thinking, self.config.additional_allowed_dirs.clone(), ) .await; // 拒绝追踪熔断检查：连续/累计拒绝达到阈值则终止循环 if let Ok(dt) = self.denial_tracker.lock() { if dt.should_terminate() { let reason = dt.termination_reason(); warn!("[AgentRuntime] 拒绝熔断触发: {}", reason); let _ = tx.send(AgentStreamEvent::Error { message: reason }); break; } } // 将工具结果推入消息上下文 for tm in exec_result.tool_messages { messages.push(tm.chat_message); } // Hook 注入的附加上下文：包装为 system-reminder 注入 LLM 消息列表 // 使用 ContextDeduplicator 在单步内去重（多个 hook 注入相同内容时只保留一份） let mut dedup = crate::agent::hooks::ContextDeduplicator::new(); for ctx in &exec_result.hook_contexts { if dedup.is_duplicate(ctx) { continue; } let reminder = format!( "\n[Hook 注入上下文]\n{}\n", ctx ); messages.push(ChatMessage::user(&reminder)); } // Hook 阻塞错误：记录到日志用于诊断 for be in &exec_result.blocking_errors { warn!("[AgentRuntime] Hook 阻塞错误: {}", be); } // 持久化 todo_write 任务状态到数据库 if called_todo_write { for prep in &prepared_calls { if prep.tool_name == "todo_write" { if let Some(todos) = prep.args.get("todos").and_then(|t| t.as_array()) { let todos_vec: Vec = todos.to_vec(); let _ = crate::agent::tools::persist_tasks(db, sid, &todos_vec, "lead") .await; } } } } if exec_result.was_cancelled { if let Ok(mut locked) = self.app_state.cancelled_runs.lock() { locked.remove(sid); } warn!( "[AgentRuntime] 工具执行期间被用户手动中止，会话 ID: {}", sid ); let _ = tx.send(AgentStreamEvent::Error { message: "用户已手动中止执行。".to_string(), }); loop_terminal = Some(TurnTerminal::CancelledByUser { session_id: sid.clone(), at_step: step, }); break; } // OnStepComplete hook let step_ctx = StepCompleteContext { session_id: sid.clone(), step, max_steps: self.config.max_steps, messages_count: messages.len(), estimated_tokens, token_limit, }; hook_registry.run_on_step_complete(&step_ctx).await; } Ok((metrics, loop_terminal)) } /// LLM 流式调用，含完整的错误恢复阶梯。 /// /// 首次调用失败后，按顺序尝试： /// 1. AggressiveCompact (keep_recent=2) /// 2. ReactiveCompact (LLM 摘要) /// 3. EscalateTokens (提升 hard_limit → 64k) /// 4. MultiTurn (注入分步消息) /// 5. Surface (放弃) /// /// 每一步后重试 LLM 调用。返回 Some(StreamOutput) 表示成功（可能经过恢复）， /// None 表示所有步骤均已尝试且失败。 #[allow(clippy::too_many_arguments)] async fn call_llm_with_recovery( &self, llm: &LlmClient, messages: &mut Vec, tool_defs: &[crate::clients::llm::ToolDefinition], tx: &mpsc::UnboundedSender, step: usize, session_id: &str, token_budget: &mut TokenBudget, ) -> Option { // 首次尝试 let output = streaming::process_llm_stream( llm, messages, tool_defs, tx, step, session_id, self.app_state.cancelled_runs.clone(), self.config.enable_thinking, ) .await; match output.status { StreamStatus::Success => return Some(output), StreamStatus::Cancelled => { if let Ok(mut cancelled) = self.app_state.cancelled_runs.lock() { cancelled.remove(session_id); } warn!( "[AgentRuntime] 流式调用期间被用户手动中止，会话 ID: {}", session_id ); let _ = tx.send(AgentStreamEvent::Error { message: "用户已手动中止执行。".to_string(), }); return None; } StreamStatus::Error(ref e_str) => { error!("[AgentRuntime] 流式读取错误: {}", e_str); } } // 提取错误字符串（用于分类） let e_str = match &output.status { StreamStatus::Error(s) => s.clone(), _ => return Some(output), // 不应到达，但安全起见 }; let error_kind = classify_error(&e_str); // ── 429/529 瞬态错误：指数退避重试（独立的快速路径） ── if matches!(error_kind, ErrorKind::RateLimited | ErrorKind::Overloaded) { let retry_after_secs = error_recovery::parse_retry_after(&e_str); let mut consecutive_overloads: u32 = 0; const MAX_BACKOFF_RETRIES: u32 = 10; for attempt in 0..MAX_BACKOFF_RETRIES { let delay_ms = error_recovery::backoff_delay(attempt, retry_after_secs); info!( "[AgentRuntime] 退避重试 {}/{} ({}ms, error={:?})", attempt + 1, MAX_BACKOFF_RETRIES, delay_ms, error_kind ); let _ = tx.send(AgentStreamEvent::Thought { content: format!( "⏳ 模型服务暂时不可用，正在重试 ({}/{})...", attempt + 1, MAX_BACKOFF_RETRIES ), step, }); tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; // 529 连续过载检测：3 次后尝试切换备用模型 if matches!(error_kind, ErrorKind::Overloaded) { consecutive_overloads += 1; if consecutive_overloads >= 3 { let fallback = &self.app_state.config.llm_fallback_model; if !fallback.is_empty() { warn!( "[AgentRuntime] 连续 {} 次过载，切换到备用模型: {}", consecutive_overloads, fallback ); llm.set_model(fallback.clone()); consecutive_overloads = 0; } else if !self.app_state.config.llm_fallback_chain.is_empty() { let idx = ((consecutive_overloads as usize - 3) % self.app_state.config.llm_fallback_chain.len()) .min(self.app_state.config.llm_fallback_chain.len() - 1); let alt = &self.app_state.config.llm_fallback_chain[idx]; warn!( "[AgentRuntime] 连续 {} 次过载，从链中切换: {}", consecutive_overloads, alt ); llm.set_model(alt.clone()); consecutive_overloads = 0; } } } // 检查用户取消 if let Ok(cancelled) = self.app_state.cancelled_runs.lock() { if cancelled.contains(session_id) { warn!("[AgentRuntime] 退避重试期间被用户取消"); let _ = tx.send(AgentStreamEvent::Error { message: "用户已手动中止执行。".to_string(), }); return None; } } // 重试 LLM 调用 let retry_output = streaming::process_llm_stream( llm, messages, tool_defs, tx, step, session_id, self.app_state.cancelled_runs.clone(), self.config.enable_thinking, ) .await; match retry_output.status { StreamStatus::Success => { info!("[AgentRuntime] 退避重试成功！(尝试 {})", attempt + 1); return Some(retry_output); } StreamStatus::Cancelled => { if let Ok(mut cancelled) = self.app_state.cancelled_runs.lock() { cancelled.remove(session_id); } return None; } StreamStatus::Error(_) => { // 继续重试 continue; } } } // 所有退避重试失败 warn!( "[AgentRuntime] {} 次退避重试后仍然失败", MAX_BACKOFF_RETRIES ); let _ = tx.send(AgentStreamEvent::Error { message: format!( "模型服务暂时不可用（已重试 {} 次）。请稍后再试或检查模型服务状态。", MAX_BACKOFF_RETRIES ), }); return None; } if !ErrorRecovery::is_recoverable(&error_kind) { let _ = tx.send(AgentStreamEvent::Error { message: format!("大模型流式读取失败: {}", e_str), }); return None; } let mut recovery = ErrorRecovery::new(token_budget.clone()); // 尝试从错误消息中解析 ContextOverflow 信息（参考 Claude Code 自动修复） let overflow_info = error_recovery::parse_context_overflow(&e_str); while let Some(recovery_step) = recovery.try_recover(&error_kind, overflow_info.as_ref()) { match recovery_step { error_recovery::RecoveryStep::AdjustMaxTokens { new_max_tokens } => { info!( "[AgentRuntime] 错误恢复: AdjustMaxTokens → {} (从错误消息自动计算)", new_max_tokens ); // token_budget.hard_limit 已由 try_recover 下调 } error_recovery::RecoveryStep::RetryWithBackoff { attempt, delay_ms } => { // 429/529 本应在 streaming 层处理，若到达此处说明分类逻辑有变更， // 安全降级为 sleep + 直接重试（不依赖 streaming 层重试）。 warn!( "[AgentRuntime] RetryWithBackoff 在 error_recovery 层触发 (attempt={}, delay={}ms)，执行降级重试", attempt, delay_ms ); tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; // 不计入 recovery 计数，由外层循环自然重试 } error_recovery::RecoveryStep::AggressiveCompact => { info!("[AgentRuntime] 错误恢复: 激进压缩 (snip + micro with keep_recent=2)"); compact::snip_compact(messages, self.config.max_messages); compact::micro_compact(messages, 2); } error_recovery::RecoveryStep::ReactiveCompact => { info!("[AgentRuntime] 错误恢复: LLM 摘要压缩"); compact::compress_context( messages, llm, self.config.context_char_limit, session_id, ) .await; } error_recovery::RecoveryStep::EscalateTokens { .. } => { info!( "[AgentRuntime] 错误恢复: 提升 token 硬限制到 {}", recovery.token_budget.hard_limit ); } error_recovery::RecoveryStep::MultiTurn => { info!("[AgentRuntime] 错误恢复: 注入多轮消息"); messages.push(ChatMessage::user(ErrorRecovery::multi_turn_message())); } error_recovery::RecoveryStep::Surface => { warn!("[AgentRuntime] 错误恢复: 所有步骤失败，暴露错误"); break; } } // 重试 LLM 调用 let retry_output = streaming::process_llm_stream( llm, messages, tool_defs, tx, step, session_id, self.app_state.cancelled_runs.clone(), self.config.enable_thinking, ) .await; match retry_output.status { StreamStatus::Success => { info!("[AgentRuntime] 错误恢复成功！"); // 将恢复后的 token_budget 状态同步回去 *token_budget = recovery.token_budget.clone(); return Some(retry_output); } StreamStatus::Cancelled => { if let Ok(mut cancelled) = self.app_state.cancelled_runs.lock() { cancelled.remove(session_id); } warn!("[AgentRuntime] 恢复期间被用户中止"); let _ = tx.send(AgentStreamEvent::Error { message: "用户已手动中止执行。".to_string(), }); return None; } StreamStatus::Error(retry_err) => { info!( "[AgentRuntime] 恢复步骤 {:?} 未能解决，继续下一阶梯: {}", recovery_step, retry_err ); } } } // 所有恢复步骤均已尝试 let _ = tx.send(AgentStreamEvent::Error { message: format!("大模型流式读取失败，且所有恢复步骤均未能解决: {}", e_str), }); None } // ── Helpers ── /// 系统提示词（模块化组装）。 /// /// 设计原则： /// 1. 所有静态 section 在前 → 内容不变，服务端自然缓存 /// 2. 动态 section（environment/tools/skills/memory）在后 /// 3. 使用 SystemPromptCache：首次计算后永久复用，/clear 时失效 /// 4. 模式（AgentMode）可覆盖 identity 和 principles section fn system_prompt(&self) -> String { use self::system_prompt::{ SystemPrompt, IDENTITY_SECTION, PRINCIPLES_SECTION, SAFETY_SECTION, SYSTEM_CONTEXT_SECTION, TOOL_USAGE_SECTION, }; let mut sp = SystemPrompt::new(); // ═══════ 静态 section（首次计算后永久缓存）═══════ // 注意：identity 和 principles 可能被 mode 覆盖 let mut cache = self.prompt_cache.lock().unwrap_or_else(|e| { tracing::warn!("[SystemPrompt] 缓存锁异常: {:?}", e); e.into_inner() }); // Identity：优先使用模式的覆盖，否则使用默认 let identity = match self.mode.identity_override { Some(override_text) => override_text.to_string(), None => cache.get_or_compute("identity", || IDENTITY_SECTION.to_string()), }; sp.add_section("identity", identity); // Principles：优先使用模式的覆盖，否则使用默认 let principles = match self.mode.principles_override { Some(override_text) => override_text.to_string(), None => cache.get_or_compute("principles", || PRINCIPLES_SECTION.to_string()), }; sp.add_section("principles", principles); // 模式的额外 section（追加在静态 section 之后） for (name, content) in self.mode.extra_sections { sp.add_section(name, content.to_string()); } sp.add_section( "system_context", cache.get_or_compute("system_context", || SYSTEM_CONTEXT_SECTION.to_string()), ); sp.add_section( "tool_usage", cache.get_or_compute("tool_usage", || TOOL_USAGE_SECTION.to_string()), ); sp.add_section( "safety", cache.get_or_compute("safety", || SAFETY_SECTION.to_string()), ); // ═══════ 动态 section（首次计算后缓存，session 内不变）═══════ // 环境上下文：CWD/platform/OS/model 在 session 内不变 let env_section = cache.get_or_compute("environment", || self.build_environment_section()); sp.add_section("environment", env_section); // 工具目录：使用 tool_catalog() 列出常驻+延迟工具（P3 defer_loading 集成） let tools_section = cache.get_or_compute("tools", || { let catalog = self.tool_registry.tool_catalog(); format!( "你可以使用以下工具（[deferred] 标记的工具需要通过 load_skill 发现详情）：\n{}", catalog ) }); sp.add_section("tools", tools_section); drop(cache); // 技能列表：通过文件监听热更新，不缓存 if let Some(skills) = self .app_state .skill_registry .read() .ok() .and_then(|r| r.build_reminder()) { sp.add_section("skills", skills); } // 项目记忆：受 save_memory 工具实时影响，不缓存 if let Some(memory) = self .app_state .memory_manager .try_lock() .ok() .and_then(|mgr| mgr.build_system_reminder(5)) { sp.add_section("memory", memory); } sp.assemble() } /// 使提示词缓存中指定 section 失效。 pub fn invalidate_prompt_cache(&self, section_name: &'static str) { if let Ok(mut cache) = self.prompt_cache.lock() { cache.invalidate(section_name); } } /// 使所有提示词缓存失效（`/clear` 或 `/compact` 事件触发）。 pub fn invalidate_all_prompt_cache(&self) { if let Ok(mut cache) = self.prompt_cache.lock() { cache.invalidate_all(); } } /// 构建环境上下文 section（参考 Claude Code `computeEnvInfo()`）。 /// /// 包含：工作目录、git 状态、平台、OS 版本、日期、模型信息。 fn build_environment_section(&self) -> String { let cwd = std::env::current_dir() .map(|p| p.display().to_string()) .unwrap_or_else(|_| "(unknown)".to_string()); let is_git = std::process::Command::new("git") .args(["rev-parse", "--is-inside-work-tree"]) .output() .map(|o| o.status.success()) .unwrap_or(false); let platform = std::env::consts::OS; let os_version = { let output = std::process::Command::new("uname") .args(["-s", "-r"]) .output() .ok() .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string()) .unwrap_or_default(); if output.is_empty() { std::env::consts::ARCH.to_string() } else { output } }; let today = chrono::Utc::now().format("%Y-%m-%d").to_string(); let model_name = self.app_state.llm.model().to_string(); let mut lines = vec![ "# 环境信息".to_string(), format!("- 工作目录: {}", cwd), format!("- Git 仓库: {}", if is_git { "是" } else { "否" }), format!("- 平台: {}", platform), format!("- OS 版本: {}", os_version), format!("- 日期: {}", today), format!("- 当前模型: {}", model_name), ]; // Agent 配置摘要（最大步数、超时等） lines.push(format!("- 最大推理步数: {}", self.config.max_steps)); lines.push(format!("- 工具超时: {} 秒", self.config.tool_timeout_secs)); lines.join("\n") } /// 步数耗尽时的最终答案生成（不带工具调用，强制 LLM 直接回答） async fn final_answer_without_tools( &self, llm: &LlmClient, messages: &[ChatMessage], session_id: &str, turn_index: i32, step: usize, tx: &mpsc::UnboundedSender, ) -> anyhow::Result<()> { let empty_tools: Vec = Vec::new(); let mut stream_rx = match llm .chat_stream(messages, &empty_tools, self.config.enable_thinking) .await { Ok(rx) => rx, Err(e) => { let _ = tx.send(AgentStreamEvent::Error { message: format!("最终回答生成失败: {}", e), }); return Err(anyhow::anyhow!("final_answer LLM call failed: {}", e)); } }; let mut accumulated = String::new(); while let Some(event) = stream_rx.recv().await { match event { StreamEvent::TextDelta(delta) => { accumulated.push_str(&delta); let _ = tx.send(AgentStreamEvent::TextDelta { content: delta, tool_call_id: None, }); } StreamEvent::Usage(u) => { let _ = tx.send(AgentStreamEvent::Usage { prompt_tokens: u.prompt_tokens, completion_tokens: u.completion_tokens, total_tokens: u.total_tokens, }); } StreamEvent::Done => break, StreamEvent::Error(e) => { let _ = tx.send(AgentStreamEvent::Error { message: format!("最终回答流式错误: {}", e), }); break; } _ => {} } } let assistant_msg = ChatMessage::assistant(accumulated.clone()); self.save_message( &self.app_state.db, session_id, turn_index, step as i32, &assistant_msg, None, None, ) .await?; Ok(()) } /// 保存消息到数据库 #[allow(clippy::too_many_arguments)] async fn save_message( &self, db: &SqlitePool, session_id: &str, turn_index: i32, step_index: i32, msg: &ChatMessage, thought: Option<&str>, extra_metadata: Option, ) -> anyhow::Result<()> { self.save_message_as( db, session_id, turn_index, step_index, msg, thought, "lead", extra_metadata, ) .await } /// 保存消息到数据库（指定 agent 身份） #[allow(clippy::too_many_arguments)] async fn save_message_as( &self, db: &SqlitePool, session_id: &str, turn_index: i32, step_index: i32, msg: &ChatMessage, thought: Option<&str>, agent_name: &str, extra_metadata: Option, ) -> anyhow::Result<()> { let role = match msg.role { MessageRole::System => "system", MessageRole::User => "user", MessageRole::Assistant => "assistant", MessageRole::Tool => "tool", }; let content = msg.content.clone().unwrap_or_default(); let tool_calls_json = msg .tool_calls .as_ref() .map(|tc| serde_json::to_string(tc).unwrap_or_default()); let tool_call_id = msg.tool_call_id.as_deref(); let token_count = content.len() as i32 / 4; // metadata: 存储结构化的消息元信息（thought/tool_calls/tool_call_id 等） let mut metadata = serde_json::json!({ "has_thought": thought.is_some(), "has_tool_calls": tool_calls_json.is_some(), "step_index": step_index, }); // 合并额外元数据（如图片路径等） if let Some(ref extra) = extra_metadata { if let (Some(base), Some(extra_obj)) = (metadata.as_object_mut(), extra.as_object()) { for (k, v) in extra_obj { base.insert(k.clone(), v.clone()); } } } let metadata_str = serde_json::to_string(&metadata).unwrap_or_default(); // raw_json: 存储完整消息的 JSON 序列化（调试/审计用） let raw_json = serde_json::to_string(msg).unwrap_or_default(); sqlx::query( "INSERT INTO agent_messages (session_id, turn_index, step_index, role, content, thought, tool_calls, tool_call_id, token_count, metadata, raw_json, agent_name) \ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(session_id) .bind(turn_index) .bind(step_index) .bind(role) .bind(content) .bind(thought) .bind(&tool_calls_json) .bind(tool_call_id) .bind(token_count) .bind(&metadata_str) .bind(&raw_json) .bind(agent_name) .execute(db) .await?; Ok(()) } /// 非致命保存：数据库写入失败时记录日志但不终止 turn #[allow(dead_code)] #[allow(clippy::too_many_arguments)] async fn save_message_non_fatal( &self, db: &SqlitePool, session_id: &str, turn_index: i32, step_index: i32, msg: &ChatMessage, thought: Option<&str>, agent_name: &str, ) { if let Err(e) = self .save_message_as( db, session_id, turn_index, step_index, msg, thought, agent_name, None, ) .await { warn!("[AgentRuntime] 消息持久化失败（非致命）: {}", e); } } } // ── Mode helper functions ──────────────────────────────────────────────── /// 将模式的 AgentConfig 预设合并到给定的 config 中。 /// /// 仅覆盖 mode_config 中 Some 的字段，None 保持原值不变。 fn apply_mode_config(config: &mut AgentConfig, mode: &AgentMode) { if let Some(max_steps) = mode.mode_config.max_steps { config.max_steps = max_steps; } if let Some(enable_thinking) = mode.mode_config.enable_thinking { config.enable_thinking = enable_thinking; } if let Some(tool_timeout_secs) = mode.mode_config.tool_timeout_secs { config.tool_timeout_secs = tool_timeout_secs; } // 加载权限档案（模式绑定的 permission_profile） if let Some(profile_name) = mode.mode_config.permission_profile { if let Some(profile) = permission_profile::load_profile(profile_name) { tracing::info!( "[AgentRuntime] 模式 '{}' 加载权限档案: {} — {}", mode.id, profile.name, profile.description ); permission_profile::apply_profile_to_config( &profile, &mut config.permission_deny_rules, &mut config.permission_allow_rules, &mut config.permission_ask_rules, &mut config.permission_mode, ); } } } /// 根据模式的 ToolSet 设置工具注册表的定义过滤器。 fn apply_mode_tool_filter(tool_registry: &mut ToolRegistry, mode: &AgentMode) { let all_names = tool_registry.tool_names(); if let Some(filter) = modes::tool_set_to_filter(&mode.tool_set, &all_names) { tracing::info!( "[AgentRuntime] 模式 '{}' 工具过滤: {} → {} 个工具", mode.id, all_names.len(), filter.len() ); tool_registry.set_definition_filter(filter); } }