From 457f6b2408e088f2e9ed8b4182ec471238b4ab01 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Tue, 4 Aug 2026 12:53:33 +0800 Subject: [PATCH] =?UTF-8?q?feat(retry):=20=E4=B8=BA=20LLM=20=E4=B8=BB?= =?UTF-8?q?=E8=B0=83=E7=94=A8=E6=B7=BB=E5=8A=A0=E5=8F=AF=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E9=87=8D=E8=AF=95=E6=9C=BA=E5=88=B6=EF=BC=8C=E9=BB=98=E8=AE=A4?= =?UTF-8?q?=203=20=E6=AC=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - config: ProviderConfig 增加 max_retries 字段（serde default=3，向后兼容） - config: LLMProviderConfig 透传 max_retries，不进 ProviderRuntimeConfig（保持 provider 构造包纯净） - agent: AgentRuntimeConfig 增加 max_retries，归属 agent 行为层 - agent_loop: 流式 + summary 两个调用点实现重试循环 - 指数退避 1s/2s/4s，仅对 429/502/503/504/timeout/connection reset 重试 - 流式仅在未 emit delta 时重试（AtomicBool 跟踪），避免重复输出 - 退避 sleep 期间响应 cancel_signal，取消优先 - 前端: ProviderConfig 类型和表单增加 max_retries 字段 - 测试: 7 个单元测试（3 判定 + 4 行为），540 个 lib 测试全绿 --- PLAN.md | 129 ++++++ docs/CHANGELOG.md | 7 +- src/agent/agent_loop.rs | 500 ++++++++++++++++----- src/agent/context_compressor.rs | 1 + src/agent/runtime_config.rs | 4 + src/cli/init.rs | 2 + src/config/mod.rs | 12 + src/gateway/agent_prompt_provider.rs | 1 + src/gateway/provider_config_service.rs | 1 + src/gateway/session.rs | 10 + src/tools/task/prompt.rs | 2 + tests/test_integration.rs | 1 + tests/test_tool_calling.rs | 1 + web/src/components/Settings/ConfigPage.tsx | 10 + web/src/components/Settings/types.ts | 1 + 15 files changed, 565 insertions(+), 117 deletions(-) create mode 100644 PLAN.md diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 0000000..2b9527b --- /dev/null +++ b/PLAN.md @@ -0,0 +1,129 @@ +# 模型访问重试机制 — 第一性原理分析 + +## 一、问题本质 + +### 1.1 为什么需要重试？ +LLM 请求存在**客观的瞬态失败**：网络抖动、服务端 502/503/504、限流 429、连接重置。这些故障在秒级内可自愈，但当前代码遇到即终止整个用户回合，用户必须手动重发。这是体验断裂点。 + +### 1.2 重试的本质权衡 +重试 = 用 **资源（额外请求/计费）+ 延迟（等待+退避）** 换取 **成功概率提升**。 + +边界条件： +- **可恢复错误**重试有意义：timeout、502/503/504、429、connection reset +- **不可恢复错误**重试是纯浪费：401/403（认证）、400（参数）、404（模型不存在）、内容审查拒绝、token 超限 +- **用户取消**必须立即生效，重试不能凌驾于取消之上 + +### 1.3 重试的副作用 +| 副作用 | 严重性 | 对策 | +|--------|--------|------| +| 计费翻倍 | 中 | 限制次数，仅对瞬态错误 | +| 请求放大加剧服务过载 | 低（单机场景） | 退避等待 | +| 流式已 emit 内容后重试→重复输出 | 高 | 流式仅在建连阶段重试 | +| 延迟累积（N×请求+退避） | 中 | 退避不宜过长 | + +## 二、架构决策的第一性原理 + +### 2.1 重试决策权归属 +原则：**决策权给最了解错误语义的层**，同时**不破坏既有解耦边界**。 + +**决策：AgentLoop 层重试**。理由： +1. cancel_token 在 AgentLoop，重试与取消协作自然（Provider 内部重试无法响应取消） +2. memory_maintenance 已采用"调用方重试"模式，保持一致 +3. 不修改 LLMProvider trait，零侵入 + +**承认的既有债务**：AgentLoop 的 `is_recoverable_llm_error()` 字符串匹配本身就是解耦缺陷——业务层不该知道 "504" 是 HTTP 错误。根因是 trait 返回 `Box` 丢失类型信息。彻底解耦需给 trait 加类型化错误 enum，超出本次范围，沿用既有字符串匹配作为增量改进。 + +### 2.2 配置层级归属（保留解耦边界） + +**关键约束**：`ProviderRuntimeConfig` 的语义是"构造 provider 实例的最小参数包"——其每个字段都被 `create_provider()` 消费。`max_retries` 不参与 provider 构造，塞进去会破坏该语义。 + +**决策**： +- 用户配置层：`ProviderConfig.max_retries`（与 `llm_timeout_secs` 同级，符合"provider 级网络参数"分组） +- 聚合配置层：`LLMProviderConfig.max_retries`（透传） +- **不进 `ProviderRuntimeConfig`**（保持 provider 构造包纯净） +- 改放进 `AgentRuntimeConfig.max_retries`（该结构本就含 `max_tool_iterations` 等 agent 行为参数，"agent 对 provider 瞬态失败的容忍策略"归属 agent 行为层合理） + +``` +config.json ProviderConfig.max_retries (用户配置) + ↓ +LLMProviderConfig.max_retries (聚合配置) + ↓ +AgentRuntimeConfig.max_retries (agent 行为参数) + ↓ +AgentLoop.runtime_config.max_retries (业务层读取) + ↓ +chat_with_retry() (AgentLoop 内部循环) +``` + +`ProviderRuntimeConfig` 保持不变。 + +### 2.3 退避策略 +原则：**退避长度应匹配故障恢复时间尺度**。 + +LLM 服务瞬态故障通常秒级恢复。指数退避 1s/2s/4s 总等待 7s，对单机本地代理场景已足够。 + +**决策：硬编码指数退避 `[1000, 2000, 4000]` ms，不暴露给用户。** 理由： +- 单机场景无需 jitter（jitter 解决分布式客户端同步重试，单机不存在） +- 退避细节是实现策略，非用户可调参数（YAGNI） +- 用户只关心"重试几次"，不关心"等多久" + +### 2.4 流式重试的边界 +**核心矛盾**：`chat_with_streaming` 一旦 emit delta，重试会重复输出。 + +**决策：流式调用仅在"未 emit 任何 delta"时重试。** 用 `Arc` 跟踪 emit 状态，首次 delta 后置 true，true 时不再重试。 + +这覆盖了最常见的瞬态场景：建连失败、首次响应超时。已开始流式传输后的失败通常是网络中断，重试意义不大且会重复。 + +## 三、实施计划 + +### 后端（4 个文件） + +**1. `src/config/mod.rs`** +- `ProviderConfig` 增 `max_retries: u32`（`#[serde(default = "default_max_retries")]`，默认 3） +- `LLMProviderConfig` 增 `max_retries: u32` +- `resolve_provider_config()` 和 `override_provider_model()` 传递该字段 +- 新增 `fn default_max_retries() -> u32 { 3 }` +- **不改 `ProviderRuntimeConfig`** + +**2. `src/providers/traits.rs`** +- **保持不变**（`ProviderRuntimeConfig` 不增字段，保持 provider 构造包纯净） + +**3. `src/agent/runtime_config.rs`** +- `AgentRuntimeConfig` 增 `max_retries: u32` +- `From` 传递该字段 + +**4. `src/agent/agent_loop.rs`** +- 新增 `RETRY_DELAYS_MS: &[u64] = &[1000, 2000, 4000]` +- 新增 `chat_with_retry()`：包装 `provider.chat()`，循环 `max_retries+1` 次 +- 新增 `chat_with_streaming_with_retry()`：包装 `chat_with_streaming()`，用 `Arc` 跟踪 emit 状态 +- 两处重试循环均 `tokio::select!` 监听 cancel_signal +- 替换 `:1068` 和 `:1424` 两处直接调用 +- tracing 日志：`warn!(attempt, retry_in_ms, error, "LLM request failed, retrying")` +- 单元测试：可恢复错误重试成功、不可恢复错误立即失败、重试中取消生效、流式已 emit 不重试 + +### 前端（2 个文件） + +**5. `web/src/components/Settings/types.ts`** +- `ProviderConfig` 增 `max_retries: number` + +**6. `web/src/components/Settings/ConfigPage.tsx`** +- provider 表单增 "最大重试次数" 输入框（与 "LLM 超时" 同组） +- 新增 provider 默认值 `max_retries: 3` + +## 四、关键设计约束 + +1. **向后兼容**：`#[serde(default)]` 保证旧 config.json 无需修改 +2. **取消优先**：重试 sleep 期间 `select!` 监听 cancel_signal，立即响应 +3. **max_retries=0**：不重试，行为与现状完全一致 +4. **不修改 LLMProvider trait**：零侵入，不影响 channels/subagents 调用链 +5. **memory_maintenance 不受影响**：它有独立重试逻辑，不经过 AgentLoop + +## 五、验证清单 + +- [ ] `cargo build` 通过 +- [ ] `cargo test --lib` 全绿（含新增重试测试） +- [ ] 前端 `npm run build` 类型检查通过 +- [ ] max_retries=0 时行为与现状一致 +- [ ] max_retries=3 时可恢复错误重试 3 次后失败 +- [ ] 不可恢复错误（如 401 模拟）立即失败不重试 +- [ ] 重试 sleep 期间触发取消立即生效 diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 094b53e..453be46 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -4,7 +4,7 @@ ## [0.3.0] - 2026-08-04 -较 [0.2.0] 的 14 个已提交 commit + 本次版本一并提交的锁屏冻结架构修复（12 文件），聚焦 **Agent 执行与显示层解耦**、**工程化基线**、**并发持久化稳定性** 与 **安全加固** 四大方向。架构修复经五轮对抗性审查验证。 +较 [0.2.0] 的 15 个已提交 commit + 本次版本一并提交的锁屏冻结架构修复（12 文件），聚焦 **Agent 执行与显示层解耦**、**工程化基线**、**并发持久化稳定性**、**安全加固** 与 **MCP 兼容性** 五大方向。架构修复经五轮对抗性审查验证。 ### 新增功能 @@ -66,6 +66,11 @@ #### 跨平台构建修复 - 修正依赖分类错误：`rmcp` / `schemars` / `http` / `tower-http` / `rust-embed` 均为跨平台 crate，错放在 `[target.'cfg(windows)'.dependencies]` 下导致 Linux 构建失败，移出 target 段。 +#### MCP 工具名清洗 +- OpenAI 要求 function name 匹配 `^[a-zA-Z0-9_-]+$`，否则整个请求 400。MCP 工具名由 `mcp_{server_key}_{tool_name}` 拼成，两个输入源分别处理： +- 后端：`sanitize_tool_name` 替换 `tool_name` 中的非法字符（`.`, `:`, `/` 等）为 `_`，保留 `server_key` 和 `tool_name` 原值用于路由，仅清洗 LLM 可见的 `full_name`，发生清洗时打 warn 日志便于定位。 +- 前端：MCP 卡片头部改用 `MapEntryHeader` 支持点击重命名，`addMcp` / `renameMcp` 正则校验 server 名称 + toast 提示。顺手修了 `MapEntryHeader` 进入编辑时 `val` 未同步当前 `name` 的 bug。 + ## [0.2.0] - 2026-07-31 较 [0.1.2] 的 47 个 commit 迭代，聚焦 **能力策略**、**模型独立配置**、**话题级并发隔离** 与 **Agent Loop 性能优化** 四大方向。 diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index ba826bd..a03314f 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -17,13 +17,20 @@ use std::collections::VecDeque; use std::hash::{Hash, Hasher}; use std::io::Read; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Instant; +use std::time::Duration; /// Minimum characters to keep when truncating const TRUNCATION_SUFFIX_LEN: usize = 200; const PENDING_USER_ACTION_MARKER: &str = "__PICOBOT_PENDING_USER_ACTION__"; const RECOVERABLE_LLM_ERROR_MESSAGE: &str = "模型服务暂时不可用或响应超时。请稍后重试。"; +/// LLM 请求瞬态失败的重试退避延迟（指数退避：1s, 2s, 4s）。 +/// 仅对 `is_recoverable_llm_error` 判定为真的错误重试。 +/// 索引越界时取最后一个值（避免 attempt > len 时的 panic）。 +const LLM_RETRY_DELAYS_MS: &[u64] = &[1000, 2000, 4000]; + const SUPPORTED_IMAGE_MIME_TYPES: &[&str] = &["image/jpeg", "image/png", "image/gif", "image/webp"]; const TOKEN_ESTIMATE_CHARS_PER_TOKEN: usize = 4; const TOKEN_ESTIMATE_SAFETY_MULTIPLIER: f64 = 1.2; @@ -507,11 +514,18 @@ fn format_error_chain(error: &(dyn std::error::Error + 'static)) -> String { fn is_recoverable_llm_error(error: &str) -> bool { let normalized = error.to_ascii_lowercase(); - normalized.contains("504") + // 瞬态可恢复错误：服务端过载/限流/网关错误/超时/连接重置。 + // 注意：401/403/400/404 等不可恢复错误不含这些子串，不会被误判。 + normalized.contains("429") + || normalized.contains("502") + || normalized.contains("503") + || normalized.contains("504") || normalized.contains("gateway timeout") || normalized.contains("stream timeout") || normalized.contains("timed out") || normalized.contains("timeout") + || normalized.contains("connection reset") + || normalized.contains("connection refused") } fn recoverable_llm_message(error: &str) -> String { @@ -1032,80 +1046,114 @@ impl AgentLoop { handler.set_stream_message_id(&streaming_message_id).await; } - let (delta_tx, mut delta_rx) = tokio::sync::mpsc::channel::(256); - let consumer_handler = self.emitted_message_handler.clone(); - let consumer_task = tokio::spawn(async move { - while let Some(delta) = delta_rx.recv().await { - if let Some(ref handler) = consumer_handler { - handler.handle_stream_delta(&delta).await; - } - } - }); - let stream_callback: StreamCallback = std::sync::Arc::new(move |delta: StreamDelta| { - // try_send is non-blocking and safe to call from within a tokio runtime - let _ = delta_tx.try_send(delta); - }); + let max_retries = self.runtime_config.max_retries as usize; + let mut response: Option = None; - // LLM 调用与取消信号竞速：若取消信号到达，drop LLM future 以 abort HTTP 请求。 - // stream_callback 是 Arc<...>，LLM future 持有其 clone。 - // 取消时需显式 drop 外部 stream_callback 以关闭 mpsc channel， - // 让 consumer_task 自然退出。 - let llm_result: Result< - crate::providers::ChatCompletionResponse, - Box, - >; - if self.cancel_token.is_some() { - tokio::select! { - _ = self.cancel_signal() => { - // LLM future 已被 select! drop → stream_callback clone 已释放。 - // 显式 drop 外部 stream_callback → delta_tx 释放 → channel 关闭。 - drop(stream_callback); - let _ = consumer_task.await; - let cancel = Self::build_cancel_result(iteration, emitted_messages); - self.emit_live_tool_call_message(cancel.final_response.clone()).await; - return Ok(cancel); + 'retry: for attempt in 0..=max_retries { + // 每次重试重建 channel + consumer：上次失败的 channel 可能已关闭。 + let (delta_tx, mut delta_rx) = tokio::sync::mpsc::channel::(256); + let consumer_handler = self.emitted_message_handler.clone(); + let consumer_task = tokio::spawn(async move { + while let Some(delta) = delta_rx.recv().await { + if let Some(ref handler) = consumer_handler { + handler.handle_stream_delta(&delta).await; + } } - result = self.provider.chat_with_streaming(request, stream_callback.clone()) => { - llm_result = result; + }); + // 跟踪是否已 emit delta：流式传输一旦开始（emit 过内容）， + // 重试会重复输出，此时不重试。 + let emitted = Arc::new(AtomicBool::new(false)); + let emitted_for_cb = emitted.clone(); + let stream_callback: StreamCallback = std::sync::Arc::new(move |delta: StreamDelta| { + emitted_for_cb.store(true, Ordering::SeqCst); + let _ = delta_tx.try_send(delta); + }); + + // LLM 调用与取消信号竞速：若取消信号到达，drop LLM future 以 abort HTTP 请求。 + let llm_result: Result< + crate::providers::ChatCompletionResponse, + Box, + >; + if self.cancel_token.is_some() { + tokio::select! { + _ = self.cancel_signal() => { + drop(stream_callback); + let _ = consumer_task.await; + let cancel = Self::build_cancel_result(iteration, emitted_messages); + self.emit_live_tool_call_message(cancel.final_response.clone()).await; + return Ok(cancel); + } + result = self.provider.chat_with_streaming(request.clone(), stream_callback.clone()) => { + llm_result = result; + } + } + drop(stream_callback); + } else { + llm_result = self + .provider + .chat_with_streaming(request.clone(), stream_callback) + .await; + } + + let _ = consumer_task.await; + + match llm_result { + Ok(resp) => { + response = Some(resp); + break 'retry; + } + Err(e) => { + let error_text = e.to_string(); + let can_retry = attempt < max_retries + && !emitted.load(Ordering::SeqCst) + && is_recoverable_llm_error(&error_text); + if can_retry { + let delay = LLM_RETRY_DELAYS_MS + [attempt.min(LLM_RETRY_DELAYS_MS.len() - 1)]; + tracing::warn!( + attempt = attempt + 1, + retry_in_ms = delay, + provider = %self.provider.name(), + model = %self.provider.model_id(), + error = %error_text, + "LLM streaming request failed, retrying" + ); + // 退避等待期间也响应取消信号 + if self.cancel_token.is_some() { + tokio::select! { + _ = self.cancel_signal() => { + let cancel = Self::build_cancel_result(iteration, emitted_messages); + self.emit_live_tool_call_message(cancel.final_response.clone()).await; + return Ok(cancel); + } + _ = tokio::time::sleep(Duration::from_millis(delay)) => {} + } + } else { + tokio::time::sleep(Duration::from_millis(delay)).await; + } + continue 'retry; + } + tracing::error!( + provider = %self.provider.name(), + model = %self.provider.model_id(), + error = %e, + error_details = %format_error_chain(e.as_ref()), + "LLM request failed" + ); + let assistant_message = + ChatMessage::assistant(recoverable_llm_message(&error_text)); + emitted_messages.push(assistant_message.clone()); + self.emit_live_tool_call_message(assistant_message.clone()) + .await; + return Ok(AgentProcessResult { + final_response: assistant_message, + emitted_messages, + }); } } - // LLM 调用正常完成：clone 的 Arc 已被 select! drop， - // 但外部 stream_callback 仍存活。显式 drop 以关闭 mpsc channel， - // 让 consumer_task 能自然退出。 - drop(stream_callback); - } else { - // 无取消令牌：stream_callback 被 move 进 chat_with_streaming，调用完成即释放。 - llm_result = self - .provider - .chat_with_streaming(request, stream_callback) - .await; } - // Close delta channel and wait for consumer to finish processing - // (delta_tx is dropped when the callback closure is dropped) - let _ = consumer_task.await; - - let response = match llm_result { - Ok(response) => response, - Err(e) => { - tracing::error!( - provider = %self.provider.name(), - model = %self.provider.model_id(), - error = %e, - error_details = %format_error_chain(e.as_ref()), - "LLM request failed" - ); - let assistant_message = - ChatMessage::assistant(recoverable_llm_message(&e.to_string())); - emitted_messages.push(assistant_message.clone()); - self.emit_live_tool_call_message(assistant_message.clone()) - .await; - return Ok(AgentProcessResult { - final_response: assistant_message, - emitted_messages, - }); - } - }; + let response = response.expect("retry loop must set response or return"); // Signal stream end if handler exists let had_streaming = self.emitted_message_handler.is_some(); @@ -1408,61 +1456,94 @@ impl AgentLoop { messages.push(summary_request); let request = self.build_llm_request(messages, system_prompt_context, None, 0); + let max_retries = self.runtime_config.max_retries as usize; - // 最终 summary 调用也与取消信号竞速 - let final_result: Result< - crate::providers::ChatCompletionResponse, - Box, - >; - if self.cancel_token.is_some() { - tokio::select! { - _ = self.cancel_signal() => { - let cancel = Self::build_cancel_result(self.max_iterations, std::mem::take(emitted_messages)); - self.emit_live_tool_call_message(cancel.final_response.clone()).await; - return cancel; - } - result = self.provider.chat(request) => { - final_result = result; + for attempt in 0..=max_retries { + // 最终 summary 调用也与取消信号竞速 + let final_result: Result< + crate::providers::ChatCompletionResponse, + Box, + >; + if self.cancel_token.is_some() { + tokio::select! { + _ = self.cancel_signal() => { + let cancel = Self::build_cancel_result(self.max_iterations, std::mem::take(emitted_messages)); + self.emit_live_tool_call_message(cancel.final_response.clone()).await; + return cancel; + } + result = self.provider.chat(request.clone()) => { + final_result = result; + } } + } else { + final_result = self.provider.chat(request.clone()).await; } - } else { - final_result = self.provider.chat(request).await; - } - match final_result { - Ok(response) => { - let assistant_message = if let Some(reasoning_content) = response.reasoning_content - { - ChatMessage::assistant_with_reasoning(response.content, reasoning_content) - } else { - ChatMessage::assistant(response.content) - }; - emitted_messages.push(assistant_message.clone()); - self.emit_live_tool_call_message(assistant_message.clone()) - .await; - AgentProcessResult { - final_response: assistant_message, - emitted_messages: std::mem::take(emitted_messages), + match final_result { + Ok(response) => { + let assistant_message = if let Some(reasoning_content) = response.reasoning_content + { + ChatMessage::assistant_with_reasoning(response.content, reasoning_content) + } else { + ChatMessage::assistant(response.content) + }; + emitted_messages.push(assistant_message.clone()); + self.emit_live_tool_call_message(assistant_message.clone()) + .await; + return AgentProcessResult { + final_response: assistant_message, + emitted_messages: std::mem::take(emitted_messages), + }; } - } - Err(e) => { - tracing::error!( - provider = %self.provider.name(), - model = %self.provider.model_id(), - error = %e, - error_details = %format_error_chain(e.as_ref()), - "Failed to get summary from LLM" - ); - let final_message = ChatMessage::assistant(recoverable_llm_message(&e.to_string())); - emitted_messages.push(final_message.clone()); - self.emit_live_tool_call_message(final_message.clone()) - .await; - AgentProcessResult { - final_response: final_message, - emitted_messages: std::mem::take(emitted_messages), + Err(e) => { + let error_text = e.to_string(); + let can_retry = attempt < max_retries + && is_recoverable_llm_error(&error_text); + if can_retry { + let delay = LLM_RETRY_DELAYS_MS + [attempt.min(LLM_RETRY_DELAYS_MS.len() - 1)]; + tracing::warn!( + attempt = attempt + 1, + retry_in_ms = delay, + provider = %self.provider.name(), + model = %self.provider.model_id(), + error = %error_text, + "Summary LLM request failed, retrying" + ); + if self.cancel_token.is_some() { + tokio::select! { + _ = self.cancel_signal() => { + let cancel = Self::build_cancel_result(self.max_iterations, std::mem::take(emitted_messages)); + self.emit_live_tool_call_message(cancel.final_response.clone()).await; + return cancel; + } + _ = tokio::time::sleep(Duration::from_millis(delay)) => {} + } + } else { + tokio::time::sleep(Duration::from_millis(delay)).await; + } + continue; + } + tracing::error!( + provider = %self.provider.name(), + model = %self.provider.model_id(), + error = %e, + error_details = %format_error_chain(e.as_ref()), + "Failed to get summary from LLM" + ); + let final_message = ChatMessage::assistant(recoverable_llm_message(&error_text)); + emitted_messages.push(final_message.clone()); + self.emit_live_tool_call_message(final_message.clone()) + .await; + return AgentProcessResult { + final_response: final_message, + emitted_messages: std::mem::take(emitted_messages), + }; } } } + + unreachable!("retry loop must return within its body") } /// 构建取消响应，包含已完成的迭代次数和已生成的消息数量。 @@ -1675,6 +1756,7 @@ mod tests { use super::*; use crate::config::LLMProviderConfig; use crate::observability::{MultiObserver, Observer}; + use crate::providers::{ChatCompletionResponse, Usage}; use tempfile::tempdir; struct TestObserver { @@ -1720,6 +1802,7 @@ mod tests { extra_headers: std::collections::HashMap::new(), llm_timeout_secs: 120, memory_maintenance_timeout_secs: 600, + max_retries: 3, model_id: "test-model".to_string(), temperature: Some(0.0), max_tokens: Some(32), @@ -2594,6 +2677,191 @@ mod tests { ); assert_eq!(messages.len(), 3); } + + // ===== LLM 重试机制测试 ===== + + #[test] + fn test_is_recoverable_llm_error_detects_transient_errors() { + // 429 限流 + assert!(is_recoverable_llm_error("HTTP 429 Too Many Requests")); + assert!(is_recoverable_llm_error("rate limited (429)")); + // 502/503/504 网关错误 + assert!(is_recoverable_llm_error("502 Bad Gateway")); + assert!(is_recoverable_llm_error("503 Service Unavailable")); + assert!(is_recoverable_llm_error("504 Gateway Timeout")); + // timeout + assert!(is_recoverable_llm_error("request timed out")); + assert!(is_recoverable_llm_error("stream timeout")); + assert!(is_recoverable_llm_error("operation timed out after 30s")); + // 连接错误 + assert!(is_recoverable_llm_error("connection reset by peer")); + assert!(is_recoverable_llm_error("connection refused")); + } + + #[test] + fn test_is_recoverable_llm_error_rejects_fatal_errors() { + assert!(!is_recoverable_llm_error("401 Unauthorized")); + assert!(!is_recoverable_llm_error("403 Forbidden")); + assert!(!is_recoverable_llm_error("400 Bad Request")); + assert!(!is_recoverable_llm_error("404 Not Found: model not found")); + assert!(!is_recoverable_llm_error("Invalid API key")); + assert!(!is_recoverable_llm_error("content filter blocked the request")); + } + + #[test] + fn test_is_recoverable_llm_error_case_insensitive() { + assert!(is_recoverable_llm_error("GATEWAY TIMEOUT")); + assert!(is_recoverable_llm_error("Timed Out")); + assert!(is_recoverable_llm_error("CONNECTION RESET")); + } + + /// 简单 mock provider：按预设序列依次返回响应/错误。 + /// chat_with_streaming 不 emit delta（保持 emitted=false），以便测试重试逻辑。 + struct MockProvider { + responses: std::sync::Mutex>>, + } + + impl MockProvider { + fn new(responses: Vec>) -> Self { + Self { + responses: std::sync::Mutex::new(responses), + } + } + } + + #[async_trait] + impl LLMProvider for MockProvider { + async fn chat( + &self, + _request: ChatCompletionRequest, + ) -> Result> { + let mut responses = self.responses.lock().unwrap(); + if responses.is_empty() { + return Err("no more mock responses".into()); + } + match responses.remove(0) { + Ok(r) => Ok(r), + Err(e) => Err(e.into()), + } + } + + fn ptype(&self) -> &str { + "mock" + } + fn name(&self) -> &str { + "mock" + } + fn model_id(&self) -> &str { + "mock-model" + } + } + + fn mock_success_response(content: &str) -> ChatCompletionResponse { + ChatCompletionResponse { + id: "test-id".to_string(), + model: "test-model".to_string(), + content: content.to_string(), + reasoning_content: None, + tool_calls: vec![], + usage: Usage { + prompt_tokens: 10, + completion_tokens: 10, + total_tokens: 20, + }, + } + } + + fn make_loop_with_provider(provider: Box, max_retries: u32) -> AgentLoop { + let config = test_runtime_config(); + let mut runtime_config: AgentRuntimeConfig = config.into(); + runtime_config.max_retries = max_retries; + AgentLoop { + runtime_config, + provider, + tools: Arc::new(ToolRegistry::new()), + system_prompt_provider: None, + skills: None, + tool_context: ToolContext::default(), + observer: None, + emitted_message_handler: None, + cancel_token: None, + max_iterations: 1, + } + } + + #[tokio::test] + async fn test_retry_recovers_on_transient_error() { + // 前 2 次返回 504（可恢复），第 3 次成功 + let loop_instance = + make_loop_with_provider(Box::new(MockProvider::new(vec![ + Err("504 Gateway Timeout".to_string()), + Err("504 Gateway Timeout".to_string()), + Ok(mock_success_response("recovered")), + ])), 3); + + let result = loop_instance + .process(vec![ChatMessage::user("hello")], None) + .await + .unwrap(); + + assert_eq!(result.final_response.content, "recovered"); + } + + #[tokio::test] + async fn test_retry_skipped_on_fatal_error() { + // 401 不可恢复 → 立即失败，不重试 + let loop_instance = + make_loop_with_provider(Box::new(MockProvider::new(vec![ + Err("401 Unauthorized: invalid api key".to_string()), + // 如果错误地重试了，第二次会返回成功，但我们期望不会到达 + Ok(mock_success_response("should not reach")), + ])), 3); + + let result = loop_instance + .process(vec![ChatMessage::user("hello")], None) + .await + .unwrap(); + + // 应返回错误消息而非成功内容 + assert_ne!(result.final_response.content, "should not reach"); + assert!(result.final_response.content.contains("暂时不可用") || result.final_response.content.contains("401")); + } + + #[tokio::test] + async fn test_no_retry_when_max_retries_zero() { + // max_retries=0 → 不重试，第一次失败即返回 + let loop_instance = + make_loop_with_provider(Box::new(MockProvider::new(vec![ + Err("504 Gateway Timeout".to_string()), + Ok(mock_success_response("should not reach")), + ])), 0); + + let result = loop_instance + .process(vec![ChatMessage::user("hello")], None) + .await + .unwrap(); + + // 504 是可恢复错误，但 max_retries=0 不重试 → 返回错误消息 + assert_ne!(result.final_response.content, "should not reach"); + } + + #[tokio::test] + async fn test_retry_exhausted_returns_error_message() { + // max_retries=1，两次都返回 504 → 重试1次后仍失败 + let loop_instance = + make_loop_with_provider(Box::new(MockProvider::new(vec![ + Err("504 Gateway Timeout".to_string()), + Err("504 Gateway Timeout".to_string()), + ])), 1); + + let result = loop_instance + .process(vec![ChatMessage::user("hello")], None) + .await + .unwrap(); + + // 重试用完，返回可恢复错误消息 + assert_eq!(result.final_response.content, RECOVERABLE_LLM_ERROR_MESSAGE); + } } #[derive(Debug)] diff --git a/src/agent/context_compressor.rs b/src/agent/context_compressor.rs index b190514..0605f80 100644 --- a/src/agent/context_compressor.rs +++ b/src/agent/context_compressor.rs @@ -1440,6 +1440,7 @@ mod tests { extra_headers: std::collections::HashMap::new(), llm_timeout_secs: 120, memory_maintenance_timeout_secs: 300, + max_retries: 3, model_id: "test-model".to_string(), temperature: None, max_tokens: None, diff --git a/src/agent/runtime_config.rs b/src/agent/runtime_config.rs index 836c72f..63ccf01 100644 --- a/src/agent/runtime_config.rs +++ b/src/agent/runtime_config.rs @@ -12,6 +12,9 @@ pub struct AgentRuntimeConfig { /// 图片上下文限制配置 pub max_images_in_context: usize, pub max_image_age_rounds: usize, + /// LLM 请求瞬态失败的最大重试次数（仅对 timeout/502/503/504/429 等可恢复错误重试）。 + /// 0 表示不重试。归属 agent 行为层，不进 ProviderRuntimeConfig（保持 provider 构造包纯净）。 + pub max_retries: u32, } impl From for AgentRuntimeConfig { @@ -39,6 +42,7 @@ impl From for AgentRuntimeConfig { context_tool_result_trim_chars: config.context_tool_result_trim_chars, max_images_in_context: config.max_images_in_context, max_image_age_rounds: config.max_image_age_rounds, + max_retries: config.max_retries, } } } diff --git a/src/cli/init.rs b/src/cli/init.rs index 88e2679..536579c 100644 --- a/src/cli/init.rs +++ b/src/cli/init.rs @@ -271,6 +271,7 @@ impl InitWizard { extra_headers: HashMap::new(), llm_timeout_secs: 120, memory_maintenance_timeout_secs: 600, + max_retries: 3, }; let mut providers = existing.providers.clone(); @@ -329,6 +330,7 @@ impl InitWizard { extra_headers: current_provider.extra_headers.clone(), llm_timeout_secs: current_provider.llm_timeout_secs, memory_maintenance_timeout_secs: current_provider.memory_maintenance_timeout_secs, + max_retries: current_provider.max_retries, }; let mut providers = existing.providers.clone(); diff --git a/src/config/mod.rs b/src/config/mod.rs index a39dfff..83bcc06 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -451,6 +451,10 @@ pub struct ProviderConfig { pub llm_timeout_secs: u64, #[serde(default = "default_memory_maintenance_timeout_secs")] pub memory_maintenance_timeout_secs: u64, + /// LLM 请求瞬态失败（timeout/502/503/504/429 等）的最大重试次数。 + /// 0 表示不重试。默认 3。 + #[serde(default = "default_max_retries")] + pub max_retries: u32, } #[derive(Debug, Clone, Deserialize, Serialize)] @@ -498,6 +502,10 @@ fn default_memory_maintenance_timeout_secs() -> u64 { 600 } +fn default_max_retries() -> u32 { + 3 +} + #[derive(Debug, Clone, Deserialize, Serialize)] pub struct GatewayConfig { #[serde(default = "default_gateway_host")] @@ -857,6 +865,8 @@ pub struct LLMProviderConfig { pub extra_headers: HashMap, pub llm_timeout_secs: u64, pub memory_maintenance_timeout_secs: u64, + /// LLM 请求瞬态失败的最大重试次数（透传自 ProviderConfig）。 + pub max_retries: u32, pub model_id: String, pub temperature: Option, pub max_tokens: Option, @@ -1004,6 +1014,7 @@ impl Config { extra_headers: provider.extra_headers.clone(), llm_timeout_secs: provider.llm_timeout_secs, memory_maintenance_timeout_secs: provider.memory_maintenance_timeout_secs, + max_retries: provider.max_retries, model_id: model.model_id.clone(), temperature: model.temperature, max_tokens: model.max_tokens, @@ -1098,6 +1109,7 @@ impl ModelResolver { result.extra_headers = provider.extra_headers.clone(); result.llm_timeout_secs = provider.llm_timeout_secs; result.memory_maintenance_timeout_secs = provider.memory_maintenance_timeout_secs; + result.max_retries = provider.max_retries; } if let Some(name) = model_name.map(str::trim).filter(|s| !s.is_empty()) { diff --git a/src/gateway/agent_prompt_provider.rs b/src/gateway/agent_prompt_provider.rs index a678f3e..ef1845b 100644 --- a/src/gateway/agent_prompt_provider.rs +++ b/src/gateway/agent_prompt_provider.rs @@ -129,6 +129,7 @@ mod tests { extra_headers: HashMap::new(), llm_timeout_secs: 120, memory_maintenance_timeout_secs: 600, + max_retries: 3, model_id: "test-model".to_string(), temperature: Some(0.0), max_tokens: Some(32), diff --git a/src/gateway/provider_config_service.rs b/src/gateway/provider_config_service.rs index f498e5d..6aaabb4 100644 --- a/src/gateway/provider_config_service.rs +++ b/src/gateway/provider_config_service.rs @@ -59,6 +59,7 @@ mod tests { extra_headers: HashMap::new(), llm_timeout_secs: 120, memory_maintenance_timeout_secs: 600, + max_retries: 3, model_id: model_id.to_string(), temperature: Some(0.0), max_tokens: Some(32), diff --git a/src/gateway/session.rs b/src/gateway/session.rs index d442186..e2973c6 100644 --- a/src/gateway/session.rs +++ b/src/gateway/session.rs @@ -985,6 +985,7 @@ mod tests { extra_headers: HashMap::new(), llm_timeout_secs: 120, memory_maintenance_timeout_secs: 600, + max_retries: 3, model_id: "test-model".to_string(), temperature: Some(0.0), max_tokens: Some(32), @@ -1265,6 +1266,7 @@ mod tests { max_tool_iterations: 1, llm_timeout_secs: 30, memory_maintenance_timeout_secs: 600, + max_retries: 3, tool_result_max_chars: 100_000, context_tool_result_trim_chars: 100_000, max_images_in_context: 1, @@ -1321,6 +1323,7 @@ mod tests { max_tool_iterations: 1, llm_timeout_secs: 30, memory_maintenance_timeout_secs: 600, + max_retries: 3, tool_result_max_chars: 100_000, context_tool_result_trim_chars: 100_000, max_images_in_context: 1, @@ -1399,6 +1402,7 @@ mod tests { max_tool_iterations: 1, llm_timeout_secs: 30, memory_maintenance_timeout_secs: 600, + max_retries: 3, tool_result_max_chars: 100_000, context_tool_result_trim_chars: 100_000, max_images_in_context: 1, @@ -1495,6 +1499,7 @@ mod tests { max_tool_iterations: 1, llm_timeout_secs: 30, memory_maintenance_timeout_secs: 600, + max_retries: 3, tool_result_max_chars: 100_000, context_tool_result_trim_chars: 100_000, max_images_in_context: 1, @@ -1585,6 +1590,7 @@ mod tests { max_tool_iterations: 1, llm_timeout_secs: 1, memory_maintenance_timeout_secs: 600, + max_retries: 3, tool_result_max_chars: 100_000, context_tool_result_trim_chars: 100_000, max_images_in_context: 1, @@ -1674,6 +1680,7 @@ mod tests { max_tool_iterations: 1, llm_timeout_secs: 30, memory_maintenance_timeout_secs: 600, + max_retries: 3, tool_result_max_chars: 100_000, context_tool_result_trim_chars: 100_000, max_images_in_context: 1, @@ -1745,6 +1752,7 @@ mod tests { max_tool_iterations: 1, llm_timeout_secs: 30, memory_maintenance_timeout_secs: 600, + max_retries: 3, tool_result_max_chars: 100_000, context_tool_result_trim_chars: 100_000, max_images_in_context: 1, @@ -1826,6 +1834,7 @@ mod tests { max_tool_iterations: 1, llm_timeout_secs: 30, memory_maintenance_timeout_secs: 600, + max_retries: 3, tool_result_max_chars: 100_000, context_tool_result_trim_chars: 100_000, max_images_in_context: 1, @@ -1893,6 +1902,7 @@ mod tests { max_tool_iterations: 1, llm_timeout_secs: 1, memory_maintenance_timeout_secs: 600, + max_retries: 3, tool_result_max_chars: 100_000, context_tool_result_trim_chars: 100_000, max_images_in_context: 1, diff --git a/src/tools/task/prompt.rs b/src/tools/task/prompt.rs index 5fb2e6e..94e7570 100644 --- a/src/tools/task/prompt.rs +++ b/src/tools/task/prompt.rs @@ -125,6 +125,7 @@ mod tests { extra_headers: std::collections::HashMap::new(), llm_timeout_secs: 120, memory_maintenance_timeout_secs: 600, + max_retries: 3, model_id: "test".to_string(), temperature: None, max_tokens: None, @@ -158,6 +159,7 @@ mod tests { extra_headers: std::collections::HashMap::new(), llm_timeout_secs: 120, memory_maintenance_timeout_secs: 600, + max_retries: 3, model_id: "test".to_string(), temperature: None, max_tokens: None, diff --git a/tests/test_integration.rs b/tests/test_integration.rs index 84c6bd2..f4a0758 100644 --- a/tests/test_integration.rs +++ b/tests/test_integration.rs @@ -36,6 +36,7 @@ fn load_config() -> Option { extra_headers: HashMap::new(), llm_timeout_secs: 120, memory_maintenance_timeout_secs: 600, + max_retries: 3, model_id: openai_model, temperature: Some(0.0), max_tokens: Some(100), diff --git a/tests/test_tool_calling.rs b/tests/test_tool_calling.rs index 071edfa..44eda80 100644 --- a/tests/test_tool_calling.rs +++ b/tests/test_tool_calling.rs @@ -38,6 +38,7 @@ fn load_openai_config() -> Option { extra_headers: HashMap::new(), llm_timeout_secs: 120, memory_maintenance_timeout_secs: 600, + max_retries: 3, model_id: openai_model, temperature: Some(0.0), max_tokens: Some(100), diff --git a/web/src/components/Settings/ConfigPage.tsx b/web/src/components/Settings/ConfigPage.tsx index bc7d9da..56a531f 100644 --- a/web/src/components/Settings/ConfigPage.tsx +++ b/web/src/components/Settings/ConfigPage.tsx @@ -530,6 +530,7 @@ export function ConfigPage({ onClose, onSaveConnection, initialTab }: ConfigPage extra_headers: {}, llm_timeout_secs: 120, memory_maintenance_timeout_secs: 600, + max_retries: 3, }, }); } @@ -608,6 +609,15 @@ export function ConfigPage({ onClose, onSaveConnection, initialTab }: ConfigPage className={inputCls} /> + + updProvider(name, { max_retries: +e.target.value })} + className={inputCls} + /> + ))} diff --git a/web/src/components/Settings/types.ts b/web/src/components/Settings/types.ts index 5068bad..a725056 100644 --- a/web/src/components/Settings/types.ts +++ b/web/src/components/Settings/types.ts @@ -7,6 +7,7 @@ export interface ProviderConfig { extra_headers: Record; llm_timeout_secs: number; memory_maintenance_timeout_secs: number; + max_retries: number; } export interface ModelConfig { model_id: string;