AI Agent 的降级策略模型不可用时怎么让用户至少得到缓存结果这次经历让我真正理解了那句话AI Agent 的降级策略不是锦上添花是救命稻草。今天这篇文章我就来详细拆解我在 Rust 中实现的四层降级方案——从模型选择到缓存兜底每一层逐级 fallback。一、降级金字塔从最优到最底线我设计了一个四层的降级金字塔每一层都是对上一层的降级use std::collections::HashMap; use std::time::{Duration, Instant}; /// 降级策略的四个层次 /// 从最优到最底线逐级 fallback #[derive(Debug, Clone, PartialEq, Eq)] enum DegradationLevel { /// L0最优路径 —— 主模型正常工作 Primary, /// L1主模型挂了 → 切到备用模型比如 GPT-4o → Claude FallbackModel, /// L2所有模型都挂了 → 返回缓存的相似结果 CachedResponse, /// L3缓存也没有 → 返回预置的默认回复 DefaultResponse, } /// 每一次 AI 调用的完整降级链路 struct DegradationChain { /// 模型优先级列表按优先级排序 model_chain: VecString, /// 调用超时每个模型的单次调用时限 per_model_timeout: Duration, /// 缓存存储把 prompt 的哈希映射到之前的回复 response_cache: HashMapu64, CachedResponse, /// 默认回复的模板 default_responses: VecString, } /// 缓存的回复及其元信息 #[derive(Debug, Clone)] struct CachedResponse { /// 缓存的文本 text: String, /// 缓存时间 cached_at: Instant, /// 生成该回复的模型 source_model: String, /// 缓存有效期 ttl: Duration, } impl CachedResponse { fn is_expired(self) - bool { self.cached_at.elapsed() self.ttl } } /// 简单哈希函数用于 prompt 去重匹配 fn hash_prompt(prompt: str) - u64 { use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; let mut hasher DefaultHasher::new(); prompt.hash(mut hasher); hasher.finish() }这个降级金字塔的直观图示flowchart TD subgraph L0[L0: 最优路径] A1[GPT-4o] --|调用成功| R1[返回最优结果] end L0 --|GPT-4o 超时/错误| L1 subgraph L1[L1: 模型降级] B1[尝试 Claude Sonnet] B2[尝试 DeepSeek] B3[尝试本地 Ollama] B1 -- B2 B2 -- B3 B3 --|某个成功| R2[返回次优结果br/附注使用了备用模型] end L1 --|所有模型都失败| L2 subgraph L2[L2: 缓存降级] C1[查找相似 promptbr/的历史回复] C1 --|命中| R3[返回缓存结果br/附注为历史回复] end L2 --|缓存未命中| L3 subgraph L3[L3: 最终兜底] D1[返回预置默认回复] D1 -- R4[基本可用br/附注AI 暂不可用] end style R1 fill:#4caf50,color:#fff style R2 fill:#8bc34a,color:#fff style R3 fill:#ffc107,color:#333 style R4 fill:#ff9800,color:#fff二、模型级降级从 GPT-4o 到本地 Ollama 的接力赛第一层降级是在模型之间切换。核心思路按优先级遍历模型列表碰到第一个可用的就停。use std::time::Duration; /// 模型降级器按优先级尝试不同的模型 struct ModelFallback { /// 模型优先级链越靠前的模型越优先 models: VecModelConfig, /// 每个模型的超时时间 timeout: Duration, } struct ModelConfig { name: String, api_endpoint: String, api_key: String, /// 该模型是否暂时被标记为不可用 circuit_breaker_open: bool, /// 连续失败次数3次失败后熔断 consecutive_failures: u32, } impl ModelFallback { /// 尝试用降级链调用 LLM /// 按优先级尝试每个模型遇到第一个成功的就返回 async fn call_with_fallback( mut self, prompt: str, ) - ResultModelResponse, FallbackError { let mut last_error None; for (index, model) in self.models.iter_mut().enumerate() { // 检查熔断器连续失败 3 次就跳过该模型 if model.circuit_breaker_open { tracing::warn!( model %model.name, consecutive_failures model.consecutive_failures, 模型已熔断跳过 ); continue; } tracing::info!( model %model.name, priority index, 尝试调用模型 ); // 带超时的模型调用 match tokio::time::timeout( self.timeout, call_model_api(model, prompt), ).await { Ok(Ok(response)) { // 成功了 tracing::info!(model %model.name, 模型调用成功); model.consecutive_failures 0; // 重置失败计数 // 如果用的不是第一个模型主模型给用户一个提示 let response if index 0 { ModelResponse { content: response.content, model_used: model.name.clone(), is_degraded: true, degradation_note: Some( format!( 注主模型暂不可用当前回复由 {} 生成质量可能略有差异, model.name ) ), } } else { ModelResponse { content: response.content, model_used: model.name.clone(), is_degraded: false, degradation_note: None, } }; return Ok(response); } Ok(Err(e)) { // API 返回错误比如 429、503 tracing::error!(model %model.name, error %e, 模型调用失败); model.consecutive_failures 1; // 连续失败 3 次 → 打开熔断器 if model.consecutive_failures 3 { tracing::warn!( model %model.name, 熔断器打开该模型已连续失败 3 次 ); model.circuit_breaker_open true; // 5 分钟后自动恢复 let model_name model.name.clone(); tokio::spawn(async move { tokio::time::sleep(Duration::from_secs(300)).await; tracing::info!(model %model_name, 熔断器恢复); // 实际代码中需要能访问到 model 来重置 }); } last_error Some(e); continue; // 试下一个模型 } Err(_elapsed) { // 超时了 tracing::warn!( model %model.name, timeout_secs self.timeout.as_secs(), 模型调用超时 ); model.consecutive_failures 1; last_error Some(请求超时.into()); continue; // 试下一个模型 } } } // 所有模型都失败了 Err(FallbackError::AllModelsFailed { last_error: last_error.map(|e| e.to_string()) .unwrap_or_else(|| 未知错误.to_string()), models_tried: self.models.iter() .map(|m| m.name.clone()) .collect(), }) } } #[derive(Debug)] struct ModelResponse { content: String, model_used: String, is_degraded: bool, degradation_note: OptionString, } #[derive(Debug)] enum FallbackError { AllModelsFailed { last_error: String, models_tried: VecString, }, } impl std::fmt::Display for FallbackError { fn fmt(self, f: mut std::fmt::Formatter_) - std::fmt::Result { match self { FallbackError::AllModelsFailed { last_error, models_tried } { write!( f, 所有模型均不可用尝试了: {}最后错误: {}, models_tried.join(, ), last_error ) } } } } impl std::error::Error for FallbackError {} // 模拟模型 API 调用 async fn call_model_api( _model: ModelConfig, _prompt: str, ) - ResultModelResponse, Boxdyn std::error::Error { // 实际对接 openai / anthropic / ollama 等 SDK unimplemented!(实际项目中实现具体的 API 调用) }这里有一个生产环境特别实用的设计熔断器Circuit Breaker。如果某个模型连续失败 3 次就暂时跳过它5 分钟后自动恢复。这避免了每次都去尝试一个已知挂了的模型白白浪费时间的问题。三、缓存层降级上次能回复这次至少给上次的答案如果所有模型都不可用了第二层降级是查缓存。缓存策略有讲究——不是所有 prompt 都能从缓存里找到所以需要做相似度匹配。use std::collections::HashMap; /// 语义缓存不是精确匹配而是找最相似的 prompt struct SemanticCache { /// 存储: prompt 哈希 → 缓存的回复 entries: HashMapu64, CachedEntry, /// 最大缓存条目数 max_entries: usize, /// 缓存过期时间 ttl: Duration, } #[derive(Debug, Clone)] struct CachedEntry { prompt: String, response: String, model: String, created_at: Instant, /// 该条目被命中的次数用于淘汰不常用的 hit_count: u64, } impl SemanticCache { /// 查找与当前 prompt 最相似的缓存条目 fn find_similar(self, prompt: str, threshold: f64) - OptionCachedEntry { let mut best_match: Option(CachedEntry, f64) None; for entry in self.entries.values() { // 跳过过期的条目 if entry.created_at.elapsed() self.ttl { continue; } // 计算 prompt 的相似度 let similarity compute_similarity(prompt, entry.prompt); if similarity threshold { match best_match { Some((_, best_score)) if similarity best_score { best_match Some((entry, similarity)); } None { best_match Some((entry, similarity)); } _ {} } } } best_match.map(|(entry, _)| entry) } /// 存入缓存 fn insert(mut self, prompt: String, response: String, model: str) { let hash hash_prompt(prompt); // 如果缓存满了淘汰命中次数最少的条目 if self.entries.len() self.max_entries { let evict_key self.entries.iter() .min_by_key(|(_, entry)| entry.hit_count) .map(|(k, _)| *k); if let Some(key) evict_key { self.entries.remove(key); } } self.entries.insert(hash, CachedEntry { prompt, response, model: model.to_string(), created_at: Instant::now(), hit_count: 0, }); } } /// 计算两个字符串的相似度 /// 这里用简单的 Jaccard 相似度基于词交集/并集 /// 生产环境建议用 embedding 向量数据库 fn compute_similarity(a: str, b: str) - f64 { // 分词简单按空格和标点分割 let words_a: Vecstr a.split(|c: char| c.is_whitespace() || c.is_ascii_punctuation()) .filter(|w| !w.is_empty()) .collect(); let words_b: Vecstr b.split(|c: char| c.is_whitespace() || c.is_ascii_punctuation()) .filter(|w| !w.is_empty()) .collect(); if words_a.is_empty() || words_b.is_empty() { return 0.0; } // 计算交集大小 let intersection words_a.iter() .filter(|w| words_b.contains(w)) .count(); // Jaccard 相似度 交集 / 并集 let union words_a.len() words_b.len() - intersection; if union 0 { 0.0 } else { intersection as f64 / union as f64 } }四、最终兜底预置的智能默认回复当所有模型都不可用、缓存也没命中时最后一层降级是返回预置的默认回复。但这个默认回复不应该是冷冰冰的服务不可用应该是根据聊天上下文生成的个性化默认回复。/// 最底层降级预置回复 /// 不是写死的字符串而是根据用户意图动态匹配 struct DefaultFallback { /// 按意图分类的默认回复 responses: VecIntentResponse, } struct IntentResponse { /// 用户意图关键词匹配用的 keywords: VecString, /// 匹配后的默认回复 response: String, } impl DefaultFallback { fn new() - Self { DefaultFallback { responses: vec![ IntentResponse { keywords: vec![ 你好.to_string(), hello.to_string(), hi.to_string() ], response: 你好AI 助手当前正忙稍后会自动恢复。你可以先浏览其他信息 .to_string(), }, IntentResponse { keywords: vec![ 错误.to_string(), bug.to_string(), 报错.to_string(), error.to_string() ], response: 你遇到了错误信息。虽然 AI 暂时不可用但你可以尝试以下常规排查步骤\ \n1. 检查日志输出\n2. 确认配置文件正确\n3. 搜索相关错误信息\nAI 恢复后会第一时间帮你分析。.to_string(), }, IntentResponse { keywords: vec![ 写代码.to_string(), code.to_string(), 编程.to_string() ], response: AI 助手暂时不可用无法帮你编写代码。你可以先描述清楚需求AI 恢复后会自动继续处理。\ \n\n提示如果是常见问题可以查看文档或搜索相关教程。.to_string(), }, // 通用默认回复匹配不到任何意图时用 IntentResponse { keywords: vec![], response: AI 助手当前不可用正在自动恢复中。上次恢复通常需要 1-5 分钟。\ \n\n你可以继续输入AI 恢复后会自动处理积压的消息。.to_string(), }, ], } } /// 根据用户输入匹配最合适的默认回复 fn get_response(self, user_input: str) - String { let input_lower user_input.to_lowercase(); // 匹配关键词最多的意图 let best self.responses.iter() .filter(|ir| !ir.keywords.is_empty()) // 排除通用回复 .max_by_key(|ir| { ir.keywords.iter() .filter(|kw| input_lower.contains(kw.as_str())) .count() }); match best { Some(ir) if ir.keywords.iter().any(|kw| input_lower.contains(kw.as_str())) { ir.response.clone() } _ { // 使用最后的通用回复 self.responses.last() .map(|ir| ir.response.clone()) .unwrap_or_else(|| 服务暂不可用.to_string()) } } } } /// 完整的降级 Agent 执行流程 async fn execute_with_full_degradation( prompt: str, model_fallback: mut ModelFallback, cache: mut SemanticCache, default_fallback: DefaultFallback, ) - String { // L0 L1尝试模型调用包括模型级降级 match model_fallback.call_with_fallback(prompt).await { Ok(response) { // 成功了把结果存进缓存后续可以用 cache.insert( prompt.to_string(), response.content.clone(), response.model_used, ); // 如果是降级结果附上降级说明 match response.degradation_note { Some(note) format!({}\n\n{}, response.content, note), None response.content, } } Err(e) { tracing::warn!(error %e, 所有模型都失败了尝试缓存降级); // L2尝试缓存 let threshold 0.3; // 30% 相似度就算匹配 if let Some(cached) cache.find_similar(prompt, threshold) { tracing::info!(hit_count cached.hit_count, 缓存命中); return format!( {}\n\n---\n *以上为历史缓存回复。AI 助手当前不可用回复可能不是最新结果。*, cached.response ); } // L3最终兜底 tracing::warn!(缓存未命中使用默认回复); default_fallback.get_response(prompt) } } }缓存降级还有一个隐蔽问题如果用户连续发帮我写一个冒泡排序和帮我写一个快速排序Jaccard 相似度可能高达 70%但给用户的缓存回复完全不对。我的解法是加一层意图分类——先判断 prompt 的意图标签只在同一意图内做相似度匹配跨意图直接跳过。还有一个兜底策略缓存降级的回复里应标注该回复来自缓存可能不是最新结果让用户知道信息来源。否则用户以为是实时生成的结果会追问后续问题造成更大误导。加一行标注的成本几乎为零但体验差距很大。五、总结这篇文章我从一次真实的宕机经历出发拆解了 AI Agent 的四层降级策略模型降级主模型 → 备用模型 → 本地模型逐级尝试缓存降级所有模型不可用时从历史回复中找最相似的结果默认回复缓存也没命中时根据用户意图匹配预置回复熔断保护在模型级别防止反复重试打爆已挂的服务作为自学者搭建这个降级系统之前我的心里其实是用户量又不大搞这么复杂干啥。但经历了那次 40 分钟的全平台宕机后我彻底改变了想法——降级策略不是在用户多的时候才需要而是在你一次都输不起的时候就需要。用户不需要理解为什么 AI 不可用他们只需要知道这个工具在这种情况下至少还能给点有用的东西。如果你也在做 AI 应用今天就可以检查一下你的服务挂了的时候用户看到的是什么是一片空白还是一个友好的降级提示我们下篇见