痛点
你的 AI Agent 上线了，Demo 阶段一切顺利。但当日均请求量破万后，问题接踵而至：
- OpenAI API 突然 429 限流，Agent 整条链路卡死
- 上游模型偶发 500 错误，用户侧返回空白响应
- 某个复杂 Agent 任务 token 消耗失控，一晚烧掉 $200
- 模型响应延迟从 2s 飙升到 30s，下游超时级联故障
本质上，LLM API 是你系统中最不稳定的外部依赖。传统 SRE 的可靠性模式——重试、降级、熔断——在 Agent 场景下需要针对性适配。
方案
构建 4 层防御体系：智能重试 → 模型降级 → 熔断保护 → 成本熔断，让 Agent 在生产环境具备自愈能力。
实操步骤
第 1 步：智能重试 — 区分可重试与不可重试错误
LLM API 的错误并非都适合重试。429（限流）和 503（过载）可重试，400（参数错误）重试无意义。
import time
import random
import httpx
RETRYABLE_STATUS = {429, 500, 502, 503}
MAX_RETRIES = 3
def call_llm_with_retry(payload: dict, base_delay: float = 1.0) -> dict:
"""指数退避 + 抖动的智能重试"""
for attempt in range(MAX_RETRIES + 1):
try:
resp = httpx.post(
"https://api.openai.com/v1/chat/completions",
json=payload,
headers={"Authorization": f"Bearer {API_KEY}"},
timeout=30.0,
)
if resp.status_code == 200:
return resp.json()
if resp.status_code not in RETRYABLE_STATUS:
raise NonRetryableError(f"HTTP {resp.status_code}: {resp.text}")
# 可重试错误，读取 Retry-After
retry_after = float(resp.headers.get("Retry-After", base_delay))
delay = retry_after * (2 ** attempt) + random.uniform(0, 1)
except httpx.TimeoutException:
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
if attempt < MAX_RETRIES:
time.sleep(min(delay, 60)) # 上限 60s
raise MaxRetriesExceeded("LLM API retries exhausted")
关键点：必须读 Retry-After 头，盲目重试会加剧限流。
第 2 步：模型降级 — 多模型 Fallback 链
当主力模型不可用时，自动切换到备选模型，保证服务不中断：
MODEL_CHAIN = [
{"model": "gpt-4o", "timeout": 30, "max_tokens": 4096},
{"model": "gpt-4o-mini", "timeout": 15, "max_tokens": 2048},
{"model": "claude-3-5-haiku-20241022", "timeout": 20, "max_tokens": 2048},
]
async def call_with_fallback(messages: list) -> dict:
"""按优先级尝试多模型，逐级降级"""
last_error = None
for config in MODEL_CHAIN:
try:
result = await call_llm(
model=config["model"],
messages=messages,
timeout=config["timeout"],
max_tokens=config["max_tokens"],
)
if config != MODEL_CHAIN[0]:
logger.warning(f"Degraded to model: {config['model']}")
metrics.increment("agent.model_fallback", tags=[f"model:{config['model']}"])
return result
except (RateLimitError, ServiceUnavailableError, TimeoutError) as e:\n last_error = e\n continue\n raise AllModelsFailed(f"All models exhausted: {last_error}")
降级时同步缩减 max_tokens，避免小模型处理超长上下文出错。
第 3 步：熔断保护 — 防止故障级联
当 LLM API 连续失败率超过阈值，触发熔断，快速失败而非堆积请求：
from datetime import datetime, timedelta
from dataclasses import dataclass, field
@dataclass
class CircuitBreaker:
failure_threshold: int = 5 # 连续失败 5 次触发
recovery_timeout: int = 30 # 熔断 30s 后尝试半开
half_open_max: int = 2 # 半开状态最多放行 2 个请求
state: str = "closed" # closed / open / half_open
failure_count: int = 0
last_failure_time: datetime = field(default_factory=datetime.now)
half_open_calls: int = 0
def can_execute(self) -> bool:
if self.state == "closed":
return True
if self.state == "open":
if datetime.now() - self.last_failure_time > timedelta(seconds=self.recovery_timeout):
self.state = "half_open"
self.half_open_calls = 0
return True
return False # 快速失败
# half_open
return self.half_open_calls < self.half_open_max
def record_success(self):
if self.state == "half_open":
self.half_open_calls += 1
if self.half_open_calls >= self.half_open_max:
self.state = "closed"
self.failure_count = 0
self.failure_count = 0
def record_failure(self):
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.failure_count >= self.failure_threshold:
self.state = "open"
logger.error("Circuit breaker OPEN - LLM API failures exceeded threshold")
metrics.increment("agent.circuit_breaker.open")
# 每个模型独立熔断器
breakers = {model: CircuitBreaker() for model in ["gpt-4o", "gpt-4o-mini", "claude-3-5-haiku"]}
第 4 步：成本熔断 — Token 用量实时管控
Agent 的不确定性意味着 token 消耗不可预测。必须设置硬性预算：
import threading
class TokenBudgetGuard:
"""按小时/天滚动窗口控制 token 开销"""
def __init__(self, hourly_limit: int = 500_000, daily_limit: int = 5_000_000):
self.hourly_limit = hourly_limit
self.daily_limit = daily_limit
self.hourly_usage = 0
self.daily_usage = 0
self._lock = threading.Lock()
def check_budget(self, estimated_tokens: int) -> bool:
with self._lock:
if self.hourly_usage + estimated_tokens > self.hourly_limit:
logger.warning(f"Hourly token budget exhausted: {self.hourly_usage}/{self.hourly_limit}")
return False
if self.daily_usage + estimated_tokens > self.daily_limit:
logger.warning(f"Daily token budget exhausted: {self.daily_usage}/{self.daily_limit}")
return False
return True
def record_usage(self, tokens_used: int):
with self._lock:
self.hourly_usage += tokens_used
self.daily_usage += tokens_used
# Prometheus 指标暴露
# agent_token_usage_total{model="gpt-4o", window="hourly"}
# agent_token_budget_remaining{window="daily"}
配合 Prometheus + Alertmanager，当日预算消耗 80% 时告警：
# alertmanager rule
- alert: AgentTokenBudgetHigh
expr: agent_token_daily_usage / agent_token_daily_limit > 0.8
for: 5m
labels:
severity: warning
annotations:
summary: "Agent token budget at {{ $value | humanizePercentage }}"
避坑
-
重试风暴（Retry Storm）：多个 Agent 实例同时重试会加剧 API 限流。解法：加随机抖动（jitter），分布式场景用令牌桶做全局限流，而非每个实例独立重试。
-
降级后语义漂移：从 GPT-4o 降级到 mini 模型后，复杂推理任务的输出质量骤降，下游逻辑出错。解法：降级时同步简化 Prompt（去掉 few-shot 示例、缩短 system prompt），并对关键任务标记
degraded=true让业务层感知。 -
熔断恢复太激进：半开状态一次性放行大量请求，导致 API 再次过载。解法：半开阶段用渐进式恢复（2→4→8 逐步放量），而非直接切回 closed。同时监控半开阶段的成功率，低于 80% 就重新 open。
总结
AI Agent 的可靠性工程本质是把传统微服务的弹性模式适配到 LLM 调用链：
| 策略 | 解决的问题 | 核心参数 |
|---|---|---|
| 智能重试 | 瞬时错误恢复 | 指数退避 + Retry-After + 最大 3 次 |
| 模型降级 | 主模型不可用 | 优先级链 + 同步缩减上下文 |
| 熔断保护 | 故障级联 | 5 次失败触发，30s 恢复窗口 |
| 成本熔断 | Token 预算失控 | 小时/天双窗口 + 80% 告警 |
生产建议：这 4 层防御用一个统一的 middleware 封装，所有 Agent 的 LLM 调用都经过它，配合 Prometheus 指标和 Grafana 大盘做实时可观测。先加监控，再加防御，最后调参数 — 这是可靠性工程的正确顺序。