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

AI Agent 重试机制模型调用失败后退避策略怎么写才靠谱一、模型也会抽风重试是刚需在用 Rust 做 AI Agent 的时候调用大模型 API 是最核心的环节。但这个环节偏偏很不稳定——偶尔超时、偶尔被限流、偶尔返回一个无意义的空响应。这些情况不是 bug而是分布式服务的常态。刚开始我的处理方式很粗暴失败了就报错让用户自己重试。但用户体验显然不好——用户等了半天结果告诉你调用失败请重试换谁都会烦。更好的做法是在 Agent 内部自己处理重试对用户透明。但重试不是简单地失败了就再来一次——如果 API 正在被限流你立刻重试只会加剧问题。这就是为什么需要退避策略backoff strategy。二、四种退避策略的对比退避策略的核心是等待多久再重试。不同的等待方式对下游服务和用户体验的影响完全不同stateDiagram-v2 [*] -- 尝试调用 尝试调用 -- 调用成功: 成功 尝试调用 -- 立即重试: 第1次失败 立即重试 -- 调用成功: 成功 立即重试 -- 线性退避: 第2次失败 线性退避 -- 调用成功: 成功 线性退避 -- 指数退避: 第3次失败 指数退避 -- 调用成功: 成功 指数退避 -- 熔断: 连续失败超阈值 熔断 -- 尝试调用: 冷却时间后 调用成功 -- [*]: 返回结果 熔断 -- [*]: 返回错误 note right of 熔断: 避免雪崩 note right of 指数退避: 等待时间指数增长四种策略的特点策略等待时间适用场景立即重试0网络抖动导致的偶发超时线性退避N × base服务暂时过载指数退避base × 2^N服务严重过载或被限流熔断停止调用服务不可用避免浪费资源三、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) - OptionDuration; /// 执行等待 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) - OptionDuration { 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) - OptionDuration { 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) - OptionDuration { 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 RetryableAiClientC { /// 底层的 AI 客户端 client: C, /// 退避策略 backoff: Arcdyn BackoffStrategy, } implC: AiClient Send Sync RetryableAiClientC { pub fn new(client: C, backoff: Arcdyn BackoffStrategy) - Self { Self { client, backoff } } /// 带重试的模型调用 pub async fn chat_with_retry( self, system_prompt: str, user_message: str, model: str, ) - ResultString, 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: Optionanyhow::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 的简化版本如果不想自己写这么多代码可以用社区成熟的retrycrateuse 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, ) - ResultString, retry::Erroranyhow::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::MutexOptionInstant, /// 熔断器状态 state: tokio::sync::RwLockCircuitState, } #[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 的重试机制看起来简单但要做对并不容易。核心的几个点退避策略要适配错误类型——超时可以快速重试限流需要慢慢等加上抖动——防止多个客户端同时重试造成二次伤害设置熔断——当服务明确不可用时不要死磕控制总超时——用户的耐心是有限的实际项目中我建议先用retrycrate 快速搭建后续如果发现策略不够灵活再自己封装。大部分场景下指数退避 抖动 上限就已经足够好用了。毕竟对于 AI Agent 来说偶尔的调用失败是常态重试机制的好坏直接决定了用户体验的稳定性。