LLM推理服务的容错设计重试策略、降级方案与Fallback链路一、引言几乎所有AI产品都对接了LLM推理服务。OpenAI、Claude、自建模型都可能响应超时、限流或返回错误。如果不在调用层做好容错用户体验会断崖式下跌。容错不是简单地重试几次。需要设计重试策略、降级方案和Fallback链路三层防护。本文给出一个生产环境验证过的容错架构。某AI写作产品的惨痛教训团队只对接了OpenAI一个模型没有降级和Fallback。一次OpenAI全球性限流持续4小时产品完全不可用用户流失率当天达到15%。后来接入三个模型提供商并配置熔断器同样级别的限流再发生时98%的用户请求通过降级模型完成流失率降到0.3%。容错架构的投入产出比远超预期。二、原理三层容错模型三层设计的分工重试策略层同模型重试 跨模型切换。降级方案层从大模型降级到小模型牺牲质量保可用。Fallback链路层缓存 预设回复保障底线体验。对比不同容错策略的覆盖范围单纯重试只能应对瞬态故障网络抖动、临时限流对持续故障无效单纯降级在模型提供商全面故障时也无能为力比如所有云厂商同时限流只有三层叠加才能覆盖从瞬态到持续的完整故障谱。实测数据单层重试的故障恢复率约60%两层重试降级约85%三层叠加达到98%。每多一层投入成本递增但边际收益也在递增——关键在于你的业务能承受多大的不可用率。三、代码容错调度器实现import asyncio import random import time from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, List, Optional, Callable, Awaitable class ResultStatus(Enum): SUCCESS success DEGRADED degraded CACHED cached FALLBACK fallback dataclass class LLMResult: content: str status: ResultStatus model_used: str latency_ms: float retry_count: int 0 dataclass class ModelConfig: name: str endpoint: Callable[..., Awaitable[str]] priority: int # 数字越小优先级越高 weight: float 1.0 # 负载权重 max_retries: int 2 base_delay: float 1.0 # 退避基准时间(秒) max_delay: float 30.0 timeout: float 30.0 def __hash__(self): return hash(self.name) class LLMResilienceDispatcher: LLM推理容错调度器 def __init__(self): self._models: List[ModelConfig] [] self._circuit_breakers: Dict[str, CircuitBreaker] {} # 结果缓存 self._cache: Dict[str, LLMResult] {} def register_model(self, config: ModelConfig): 注册一个模型 self._models.append(config) self._models.sort(keylambda m: m.priority) self._circuit_breakers[config.name] CircuitBreaker( failure_threshold5, recovery_timeout60.0 ) async def _call_with_retry( self, model: ModelConfig, prompt: str ) - Optional[str]: 带指数退避的重试调用 for attempt in range(model.max_retries 1): try: result await asyncio.wait_for( model.endpoint(prompt), timeoutmodel.timeout ) return result except asyncio.TimeoutError: if attempt model.max_retries: raise except Exception: if attempt model.max_retries: raise # 指数退避 随机抖动 delay min( model.base_delay * (2 ** attempt), model.max_delay ) delay random.uniform(0, delay * 0.1) await asyncio.sleep(delay) return None async def dispatch( self, prompt: str, allow_degraded: bool True, cache_ttl: float 300.0 ) - LLMResult: 主调度入口 start_time time.time() # 检查缓存 cached self._check_cache(prompt, cache_ttl) if cached: cached.latency_ms (time.time() - start_time) * 1000 return cached # 遍历模型优先级 for model in self._models: cb self._circuit_breakers[model.name] if cb.is_open(): continue try: content await self._call_with_retry(model, prompt) if content: result LLMResult( contentcontent, statusResultStatus.SUCCESS, model_usedmodel.name, latency_ms(time.time() - start_time) * 1000, ) self._cache[prompt] result cb.record_success() return result except Exception: cb.record_failure() # 是否尝试降级模型 if allow_degraded and model.priority 2: continue break # Fallback链路级 return await self._fallback(prompt) def _check_cache(self, prompt: str, ttl: float) - Optional[LLMResult]: 检查缓存 if prompt in self._cache: result self._cache[prompt] if time.time() - (result.latency_ms / 1000) ttl: result.status ResultStatus.CACHED return result return None async def _fallback(self, prompt: str) - LLMResult: 最终兜底 fallback_msgs [ 抱歉服务暂时不可用。请稍后重试。, ] return LLMResult( contentfallback_msgs[0], statusResultStatus.FALLBACK, model_usedfallback, latency_ms1.0, ) class CircuitBreaker: 熔断器实现 def __init__(self, failure_threshold: int, recovery_timeout: float): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self._failure_count 0 self._last_failure_time 0.0 self._state CLOSED # CLOSED / OPEN / HALF_OPEN def is_open(self) - bool: if self._state CLOSED: return False if self._state OPEN: if time.time() - self._last_failure_time self.recovery_timeout: self._state HALF_OPEN return False return True return False # HALF_OPEN 允许尝试 def record_success(self): self._failure_count 0 self._state CLOSED def record_failure(self): self._failure_count 1 self._last_failure_time time.time() if self._failure_count self.failure_threshold: self._state OPEN关键设计点优先级队列模型按优先级排序优先选择高优模型。指数退避重试间隔翻倍加随机抖动避免惊群。熔断器连续失败5次自动熔断60秒后半开探测。结果缓存缓存成功结果短时间内直接返回。四、权衡容错设计的边界考量重试次数与延迟的关系。重试2次退避可能让总延迟超过30秒。对实时交互场景建议重试1次就降级。降级模型的取舍。降级模型大幅降低延迟但质量损失明显。需要区分允许降级的场景闲聊、辅助写作和不允许的场景关键业务分析。缓存带来的数据一致性问题。缓存相同prompt可能返回旧结果。对动态内容如实时数据查询设置短TTL或不缓存。成本控制。每次重试都是API成本。建议设置每日预算上限超限后直接降级到缓存或预设回复。一个真实的取舍决策某AI客服产品在高峰期遇到主模型限流。团队面临选择——降级到小模型响应质量下降30%还是等待重试延迟从2秒升到15秒。最终选择降级因为客服场景的响应速度比回答质量更重要。用户能接受简短但及时的回答不能接受详细但要等15秒。场景决定了容错策略的优先级实时交互场景优先保速度分析类场景优先保质量。五、总结LLM推理服务的容错设计三层架构覆盖了大多数故障场景。重试策略保证瞬态故障恢复降级方案应对持续故障Fallback提供底线体验。部署建议先接入至少两个模型提供商配置熔断器和指数退避。在生产环境运行一周后根据失败率数据调优熔断阈值和退避参数。容错是持续优化的过程。
