AI Agent 重试机制：模型调用失败后，退避策略怎么写才靠谱

一、模型也会"抽风"，重试是刚需

在用 Rust 做 AI Agent 的时候，调用大模型 API 是最核心的环节。但这个环节偏偏很不稳定——偶尔超时、偶尔被限流、偶尔返回一个无意义的空响应。这些情况不是 bug，而是分布式服务的常态。

刚开始我的处理方式很粗暴：失败了就报错，让用户自己重试。但用户体验显然不好——用户等了半天结果告诉你"调用失败，请重试"，换谁都会烦。

更好的做法是在 Agent 内部自己处理重试，对用户透明。但重试不是简单地"失败了就再来一次"——如果 API 正在被限流，你立刻重试只会加剧问题。这就是为什么需要退避策略（backoff strategy）。

二、四种退避策略的对比

退避策略的核心是"等待多久再重试"。不同的等待方式对下游服务和用户体验的影响完全不同：

stateDiagram-v2 [*] --> 尝试调用 尝试调用 --> 调用成功: 成功 尝试调用 --> 立即重试: 第1次失败 立即重试 --> 调用成功: 成功 立即重试 --> 线性退避: 第2次失败 线性退避 --> 调用成功: 成功 线性退避 --> 指数退避: 第3次失败 指数退避 --> 调用成功: 成功 指数退避 --> 熔断: 连续失败超阈值 熔断 --> 尝试调用: 冷却时间后 调用成功 --> [*]: 返回结果 熔断 --> [*]: 返回错误 note right of 熔断: 避免雪崩 note right of 指数退避: 等待时间指数增长

四种策略的特点：

三、Rust 实现：从简单到完整

3.1 手工实现三种退避策略

use std::time::Duration; use tokio::time::sleep; /// 退避策略的抽象 trait #[async_trait::async_trait] pub trait BackoffStrategy: Send + Sync { /// 根据重试次数计算等待时长 /// 返回 None 表示不再重试 fn wait_duration(&self, attempt: u32) -> Option<Duration>; /// 执行等待 async fn wait(&self, attempt: u32) -> bool { match self.wait_duration(attempt) { Some(duration) => { sleep(duration).await; true // 可以继续重试 } None => false, // 不再重试 } } } /// 线性退避：每次等待时间线性增长 /// 公式: base_delay * attempt pub struct LinearBackoff { base_delay: Duration, max_attempts: u32, } impl LinearBackoff { pub fn new(base_delay: Duration, max_attempts: u32) -> Self { Self { base_delay, max_attempts } } } #[async_trait::async_trait] impl BackoffStrategy for LinearBackoff { fn wait_duration(&self, attempt: u32) -> Option<Duration> { if attempt > self.max_attempts { return None; // 超过最大尝试次数，不再重试 } // 第1次等1秒，第2次等2秒，第3次等3秒... Some(self.base_delay * attempt) } } /// 指数退避：每次等待时间翻倍 /// 公式: base_delay * 2^(attempt - 1) pub struct ExponentialBackoff { base_delay: Duration, max_delay: Duration, // 上限，防止等待时间过长 max_attempts: u32, } impl ExponentialBackoff { pub fn new(base_delay: Duration, max_delay: Duration, max_attempts: u32) -> Self { Self { base_delay, max_delay, max_attempts } } } #[async_trait::async_trait] impl BackoffStrategy for ExponentialBackoff { fn wait_duration(&self, attempt: u32) -> Option<Duration> { if attempt > self.max_attempts { return None; } // 计算指数退避时间: 1s -> 2s -> 4s -> 8s -> ... let delay = self.base_delay * 2u32.pow(attempt - 1); // 不超过最大等待时间 Some(std::cmp::min(delay, self.max_delay)) } } /// 带抖动的指数退避：在指数退避的基础上加入随机因子 /// 防止多个客户端在同一时刻同时重试（"惊群效应"） pub struct JitteredBackoff { base: ExponentialBackoff, } impl JitteredBackoff { pub fn new(base_delay: Duration, max_delay: Duration, max_attempts: u32) -> Self { Self { base: ExponentialBackoff::new(base_delay, max_delay, max_attempts), } } } #[async_trait::async_trait] impl BackoffStrategy for JitteredBackoff { fn wait_duration(&self, attempt: u32) -> Option<Duration> { self.base.wait_duration(attempt).map(|base_duration| { // 在 [base * 0.5, base * 1.5] 范围内随机抖动 let jitter_factor = 0.5 + rand::random::<f64>() * 1.0; Duration::from_secs_f64(base_duration.as_secs_f64() * jitter_factor) }) } }

3.2 集成到 AI Agent 的调用中

use std::sync::Arc; /// AI 模型调用的重试包装器 pub struct RetryableAiClient<C> { /// 底层的 AI 客户端 client: C, /// 退避策略 backoff: Arc<dyn BackoffStrategy>, } impl<C: AiClient + Send + Sync> RetryableAiClient<C> { pub fn new(client: C, backoff: Arc<dyn BackoffStrategy>) -> Self { Self { client, backoff } } /// 带重试的模型调用 pub async fn chat_with_retry( &self, system_prompt: &str, user_message: &str, model: &str, ) -> Result<String, RetryError> { let mut attempt = 0u32; let mut last_error = None; loop { attempt += 1; println!("[Agent] 第 {} 次尝试调用模型 {}...", attempt, model); match self.client.chat(system_prompt, user_message, model).await { Ok(response) => { // 成功，返回结果 println!("[Agent] 调用成功，共尝试 {} 次", attempt); return Ok(response); } Err(err) => { last_error = Some(err); // 判断是否应该重试 if !self.should_retry(&last_error) { return Err(RetryError::NonRetryable( format!("不可重试的错误: {:?}", last_error) )); } // 执行退避等待 if !self.backoff.wait(attempt).await { return Err(RetryError::MaxRetriesExceeded( format!("已达最大重试次数 ({})", attempt) )); } } } } } /// 判断错误是否可以重试 fn should_retry(&self, error: &Option<anyhow::Error>) -> bool { if let Some(err) = error { let err_str = format!("{}", err); // API 限流（429）可以重试 // 服务器错误（5xx）可以重试 // 客户端错误（4xx 除 429 外）不应重试 // 网络超时可以重试 err_str.contains("429") || err_str.contains("5") || err_str.contains("timeout") || err_str.contains("connection") } else { false } } } /// 重试相关的错误类型 #[derive(Debug)] pub enum RetryError { MaxRetriesExceeded(String), NonRetryable(String), } impl std::fmt::Display for RetryError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { RetryError::MaxRetriesExceeded(msg) => write!(f, "重试耗尽: {}", msg), RetryError::NonRetryable(msg) => write!(f, "不可重试: {}", msg), } } }

3.3 使用retrycrate 的简化版本

如果不想自己写这么多代码，可以用社区成熟的retrycrate：

use retry::{retry, delay::Exponential}; use std::time::Duration; /// 使用 retry crate 的简洁实现 pub async fn call_model_with_retry( client: &impl AiClient, system: &str, user: &str, model: &str, ) -> Result<String, retry::Error<anyhow::Error>> { // 配置指数退避参数 let delay = Exponential::from_millis(1000) // 初始等待 1 秒 .factor(2) // 每次翻倍 .max_delay(Duration::from_secs(30)); // 最大等待 30 秒 // retry 宏自动处理重试逻辑 retry(delay, || { // 注意：retry crate 的闭包需要是同步的 // 对于异步调用，需要配合 tokio 的 block_on 或使用 async-retry let rt = tokio::runtime::Handle::current(); rt.block_on(async { client.chat(system, user, model).await }) }) }

3.4 熔断器模式

当连续失败超过阈值时，应该暂时停止所有调用，给下游服务恢复的时间：

use std::sync::atomic::{AtomicU32, Ordering}; use tokio::time::{sleep, Duration, Instant}; /// 简单的熔断器实现 pub struct CircuitBreaker { /// 连续失败次数 failure_count: AtomicU32, /// 触发熔断的失败阈值 threshold: u32, /// 熔断后的冷却时间 cooldown: Duration, /// 熔断开始的时间 opened_at: tokio::sync::Mutex<Option<Instant>>, /// 熔断器状态 state: tokio::sync::RwLock<CircuitState>, } #[derive(Debug, PartialEq)] enum CircuitState { Closed, // 正常，允许调用 Open, // 熔断中，拒绝调用 HalfOpen, // 半开，允许少量调用用于探测 } impl CircuitBreaker { pub fn new(threshold: u32, cooldown: Duration) -> Self { Self { failure_count: AtomicU32::new(0), threshold, cooldown, opened_at: tokio::sync::Mutex::new(None), state: tokio::sync::RwLock::new(CircuitState::Closed), } } /// 在调用前检查是否允许 pub async fn allow_request(&self) -> bool { let state = self.state.read().await; match *state { CircuitState::Closed => true, CircuitState::HalfOpen => true, CircuitState::Open => { // 检查冷却时间是否已过 let opened = self.opened_at.lock().await; if let Some(opened_time) = *opened { if opened_time.elapsed() >= self.cooldown { drop(opened); drop(state); // 进入半开状态，允许探测 *self.state.write().await = CircuitState::HalfOpen; return true; } } false } } } /// 调用成功后重置 pub async fn on_success(&self) { self.failure_count.store(0, Ordering::SeqCst); *self.state.write().await = CircuitState::Closed; } /// 调用失败后记录 pub async fn on_failure(&self) { let count = self.failure_count.fetch_add(1, Ordering::SeqCst) + 1; if count >= self.threshold { *self.state.write().await = CircuitState::Open; *self.opened_at.lock().await = Some(Instant::now()); eprintln!("[熔断器] 连续失败 {} 次，进入熔断状态", count); } } }

四、边界与陷阱

重试风暴。如果 Agent 有多个并发的模型调用，每个都在重试，会瞬间给 API 服务器施加成倍的压力。使用带抖动的指数退避（jitter）可以有效缓解这个问题。

幂等性要求。重试前需要确认操作是幂等的。对于 AI 模型调用，要求返回一个文本回答通常是幂等的（相同输入得到类似输出），但如果 Agent 在调用模型的同时还做了数据库写入或其他副作用操作，就必须小心处理事务边界。

速率限制的配合。很多模型 API 有明确的速率限制（比如每分钟最多 60 次请求）。退避策略应该配合速率限制器使用，而不是把限流错误当作普通失败来重试。

最大重试时间的控制。用户不能永远等下去。一个好的做法是设置总超时时间，比如"最多重试 30 秒，之后无论结果如何都返回"。

// 使用 tokio::time::timeout 限制总重试时间 let result = tokio::time::timeout( Duration::from_secs(30), client.chat_with_retry(system, user, model), ).await;

还有一个监控盲区：重试次数和重试耗时如果不做埋点，很难发现某些接口正在变慢。建议把每次调用的重试次数、退避等待的时长都记录到指标里。如果某个模型的重试率从 2% 突然涨到 15%，大概率是上游出了问题。指标比日志更快帮你锁定根因。

还发现过一个边界：指数退避的max_delay不能设为 0，否则 jitter 计算会出错；也不要跟max_retries组合出超过接口超时的时间窗口，免得重试还没结束，上层调用方先超时了。

五、总结

Agent 的重试机制看起来简单，但要做对并不容易。核心的几个点：

1. 退避策略要适配错误类型——超时可以快速重试，限流需要慢慢等
2. 加上抖动——防止多个客户端同时重试造成二次伤害
3. 设置熔断——当服务明确不可用时不要死磕
4. 控制总超时——用户的耐心是有限的

实际项目中我建议先用retrycrate 快速搭建，后续如果发现策略不够灵活再自己封装。大部分场景下，指数退避 + 抖动 + 上限就已经足够好用了。毕竟对于 AI Agent 来说，偶尔的调用失败是常态，重试机制的好坏直接决定了用户体验的稳定性。