🧠 图解记忆：为什么 LLM 调用需要重试；点击图片可查看原图。
💡 答案要点
为什么 LLM 调用需要重试？
LLM API 调用失败的两大原因：
- 瞬时错误：网络抖动、API 限流（HTTP 429、500）
- 长尾延迟：模型冷启动、超时
最佳重试策略：指数退避 + 抖动指数退避原理：
| 策略 | 重试间隔 | 优点 | 缺点 |
|---|---|---|---|
| 固定间隔 | 1s, 1s, 1s | 简单 | 可能打爆限流 |
| 线性增长 | 1s, 2s, 3s | 温和 | 收敛慢 |
| 指数退避 | 1s, 2s, 4s, 8s | 收敛快 | 可能过长 |
| 指数退避+抖动 | 1±0.5s, 2±1s... | 平衡 | 最优 |
生产级重试实现：
展开 Python 代码示例（86 行）
python
import asyncio
import random
from typing import TypeVar, Callable, Any
from functools import wraps
import logging
logger = logging.getLogger(__name__)
T = TypeVar('T')
class RetryError(Exception):
"""重试耗尽异常"""
def __init__(self, attempts: int, last_error: Exception):
self.attempts = attempts
self.last_error = last_error
super().__init__(f"重试 {attempts} 次后失败: {last_error}")
def async_retry(
max_attempts: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter: bool = True,
retry_on: tuple = (Exception,),
):
"""
异步重试装饰器（指数退避 + 抖动）
Args:
max_attempts: 最大尝试次数
base_delay: 基础延迟（秒）
max_delay: 最大延迟上限
exponential_base: 指数基数
jitter: 是否加随机抖动
retry_on: 需要重试的异常类型
"""
def decorator(func: Callable[..., T]) -> Callable[..., T]:
@wraps(func)
async def wrapper(*args, **kwargs) -> T:
last_error = None
for attempt in range(1, max_attempts + 1):
try:
return await func(*args, **kwargs)
except retry_on as e:
last_error = e
if attempt == max_attempts:
raise RetryError(max_attempts, last_error) from e
# 计算延迟
delay = min(base_delay * (exponential_base ** (attempt - 1)), max_delay)
# 加抖动（避免多实例同时重试打爆服务）
if jitter:
delay = delay * (0.5 + random.random() * 0.5)
# 根据错误类型调整延迟
if "429" in str(e) or "rate_limit" in str(e).lower():
delay = max(delay, 10) # 限流错误至少等10秒
logger.warning(
f"Attempt {attempt}/{max_attempts} failed for {func.__name__}: {e}. "
f"Retrying in {delay:.2f}s..."
)
await asyncio.sleep(delay)
raise RetryError(max_attempts, last_error)
return wrapper
return decorator
# 使用示例
class LLMClient:
def __init__(self):
self.client = AsyncOpenAI()
@async_retry(max_attempts=4, base_delay=1.5, retry_on=(RateLimitError, TimeoutError, APIError))
async def chat(self, prompt: str) -> str:
response = await self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content带熔断器的重试（防止雪崩）：
展开 Python 代码示例（49 行）
python
import time
from dataclasses import dataclass, field
@dataclass
class CircuitBreaker:
"""熔断器：连续失败 N 次后暂停服务"""
failure_threshold: int = 5
recovery_timeout: float = 60.0
failures: int = field(default=0)
last_failure_time: float = field(default=0)
state: str = "closed" # closed, open, half_open
def record_success(self):
self.failures = 0
self.state = "closed"
def record_failure(self):
self.failures += 1
self.last_failure_time = time.time()
if self.failures >= self.failure_threshold:
self.state = "open"
logger.error(f"Circuit breaker opened after {self.failures} failures")
def can_attempt(self) -> bool:
if self.state == "closed":
return True
if self.state == "open":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "half_open"
return True
return False
return True # half_open
# 组合使用
circuit_breaker = CircuitBreaker()
@async_retry(max_attempts=3)
async def safe_chat(prompt: str) -> str:
if not circuit_breaker.can_attempt():
raise Exception("Circuit breaker is open")
try:
result = await llm_client.chat(prompt)
circuit_breaker.record_success()
return result
except Exception as e:
circuit_breaker.record_failure()
raise面试话术：
"LLM 重试不是失败后无条件重来。先区分可重试错误，例如部分 429、超时和临时 5xx，再使用带抖动的指数退避，避免多实例同步重试；设置总时限和最大次数，并配合并发限制、熔断与可观测性。工具调用等有副作用的操作还要使用幂等键或去重，具体参数应根据供应商响应头和业务 SLO 调整。"