更新于

AI Agent 降级与容错策略：生产级可靠性设计

AI Agent容错设计工程实践可靠性

AI Agent 进入生产环境后，你很快会发现：它的故障模式比传统 API 复杂得多。模型响应超时、工具调用循环、推理路径偏离、rate limit 触发……每一种都可能让整个 Agent 进入卡死或无效状态。

本文总结了一套生产级 AI Agent 容错与降级框架，覆盖从单次工具调用到多 Agent 编排的全链路，提供可直接落地的代码模式。

一、AI Agent 的典型故障分类

在设计容错之前，先把故障分清楚。Agent 系统的失败远不止”接口 500”：

故障类型具体表现发生频率
模型超时响应时间超过设定阈值中
Rate Limit429 限流，短时大量请求触发高
工具调用失败外部 API 返回错误，工具抛出异常中
推理循环Agent 反复调用同一工具无法推进低但致命
输出格式错误模型没有按预期结构输出，解析失败中
上下文溢出多轮对话超过模型 context 上限中
幻觉工具调用调用了不存在的工具或参数格式错误低
成本失控Agent 进入无限循环，token 消耗爆炸低但危险

这 8 种故障对应不同的处理策略，不能用一个”重试”解决所有问题。

二、基础层：超时 + 指数退避重试

最基础的容错是对每次 LLM 调用设置超时和重试：

import anthropic
import time
from typing import Optional

def call_with_retry(
    client: anthropic.Anthropic,
    model: str,
    messages: list,
    max_tokens: int = 2048,
    timeout: float = 30.0,
    max_retries: int = 3,
) -> Optional[anthropic.types.Message]:
    """带超时和指数退避的 LLM 调用"""
    for attempt in range(max_retries):
        try:
            return client.messages.create(
                model=model,
                max_tokens=max_tokens,
                messages=messages,
                timeout=timeout,
            )
        except anthropic.APITimeoutError:
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # 1s, 2s, 4s
                print(f"超时，{wait}s 后重试 ({attempt+1}/{max_retries})")
                time.sleep(wait)
            else:
                return None
        except anthropic.RateLimitError:
            wait = 5 * (attempt + 1)  # 更长的等待
            print(f"限流，{wait}s 后重试")
            time.sleep(wait)
        except anthropic.APIStatusError as e:
            if e.status_code >= 500:  # 服务端错误才重试
                time.sleep(2 ** attempt)
            else:
                raise  # 4xx 不重试，直接抛出
    return None

关键原则：4xx 不重试，5xx 才重试。Rate limit（429）是特殊的 4xx，需要等待后重试，但不能像处理 5xx 一样无脑重发。

更完整的 LLM 重试策略设计见 LLM 错误重试策略详解。

三、熔断器模式：防止级联失败

当下游模型服务持续故障时，持续重试会让调用方积压大量等待请求，最终拖垮整个系统。**熔断器（Circuit Breaker）**模式可以在故障率超阈值时”断开”，快速失败，给服务恢复时间。

import time
from enum import Enum
from dataclasses import dataclass, field

class CircuitState(Enum):
    CLOSED = "closed"      # 正常
    OPEN = "open"          # 熔断中
    HALF_OPEN = "half_open"  # 探测恢复中

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5      # 连续失败 N 次触发熔断
    recovery_timeout: float = 60.0  # 熔断后等待 N 秒再探测
    
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    last_failure_time: float = field(default_factory=lambda: 0.0)
    
    def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise RuntimeError("Circuit breaker OPEN: 模型服务熔断中，快速失败")
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
    
    def _on_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED
    
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            print(f"⚡ 熔断器开启，{self.recovery_timeout}s 后自动探测恢复")

四、模型降级：从强模型切到弱模型

Agent 系统中最实用的降级策略是模型降级：主模型（如 claude-opus-4-8）出现问题时，自动切换到更快更便宜的模型（如 claude-haiku-4-5-20251001）继续服务。

MODEL_FALLBACK_CHAIN = [
    "claude-opus-4-8",
    "claude-sonnet-4-6",
    "claude-haiku-4-5-20251001",
]

def call_with_model_fallback(client, messages: list, max_tokens=1024) -> dict:
    """依次尝试降级模型链"""
    last_error = None
    for model in MODEL_FALLBACK_CHAIN:
        try:
            resp = client.messages.create(
                model=model,
                max_tokens=max_tokens,
                messages=messages,
                timeout=20.0,
            )
            if model != MODEL_FALLBACK_CHAIN[0]:
                print(f"⚠️ 已降级使用模型: {model}")
            return {"model": model, "response": resp}
        except (anthropic.APITimeoutError, anthropic.APIStatusError) as e:
            print(f"模型 {model} 失败: {e}，尝试下一个")
            last_error = e
    
    raise RuntimeError(f"所有模型均失败: {last_error}")

模型降级时需要注意：弱模型可能无法完成强模型的复杂推理任务。建议对降级后的输出做额外验证，或在任务分配层就区分”必须强模型”和”可降级”的任务类型。

五、工具调用层容错

Agent 的工具调用是最容易出问题的地方。推荐用装饰器统一处理：

import functools
from typing import Callable, Any

def tool_with_fallback(fallback_value: Any = None, max_retries: int = 2):
    """工具调用容错装饰器"""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    result = func(*args, **kwargs)
                    return {"success": True, "result": result}
                except TimeoutError:
                    if attempt < max_retries:
                        time.sleep(1)
                    else:
                        return {
                            "success": False,
                            "error": "工具调用超时",
                            "fallback": fallback_value,
                        }
                except Exception as e:
                    return {
                        "success": False,
                        "error": str(e),
                        "fallback": fallback_value,
                    }
        return wrapper
    return decorator

# 使用示例
@tool_with_fallback(fallback_value={"status": "unavailable"})
def search_database(query: str) -> dict:
    # 实际数据库查询
    ...

工具调用失败时，Agent 应该收到明确的错误信息，而不是静默失败。在 tool result 中返回结构化的错误描述，让模型能做出合理的”我没有这个信息”回复，而非幻觉一个答案。

六、推理循环检测与强制终止

Agent 进入推理循环（同一工具反复调用，推理没有进展）是生产环境中的低频但高危故障：

from collections import Counter

class LoopDetector:
    def __init__(self, max_same_calls: int = 3, max_total_steps: int = 20):
        self.max_same_calls = max_same_calls
        self.max_total_steps = max_total_steps
        self.tool_calls: list[str] = []
    
    def record_tool_call(self, tool_name: str, tool_input: str) -> None:
        call_sig = f"{tool_name}:{hash(str(tool_input))}"
        self.tool_calls.append(call_sig)
    
    def is_looping(self) -> tuple[bool, str]:
        if len(self.tool_calls) > self.max_total_steps:
            return True, f"步骤数超限 ({len(self.tool_calls)} > {self.max_total_steps})"
        
        counts = Counter(self.tool_calls[-self.max_same_calls * 2:])
        for call, count in counts.items():
            if count >= self.max_same_calls:
                tool_name = call.split(":")[0]
                return True, f"工具 {tool_name} 被重复调用 {count} 次，疑似循环"
        
        return False, ""

def run_agent_with_loop_guard(client, system_prompt: str, user_message: str):
    detector = LoopDetector()
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        resp = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=system_prompt,
            messages=messages,
            tools=[...],  # 你的工具列表
        )
        
        if resp.stop_reason == "end_turn":
            return resp.content[-1].text
        
        # 处理工具调用
        tool_results = []
        for block in resp.content:
            if block.type == "tool_use":
                detector.record_tool_call(block.name, block.input)
                looping, reason = detector.is_looping()
                if looping:
                    return f"[Agent 强制终止] {reason}，请拆分任务重新描述需求。"
                
                # 执行工具
                result = execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result),
                })
        
        messages.append({"role": "assistant", "content": resp.content})
        messages.append({"role": "user", "content": tool_results})

七、上下文溢出的分块策略

多轮 Agent 对话中，历史消息会不断累积。当接近 context 上限时，不应该直接截断（会丢失关键信息），而应该压缩或摘要旧消息：

def compress_history(client, messages: list, keep_recent: int = 6) -> list:
    """保留最近 N 条消息，将更早的消息摘要为一条 system 消息"""
    if len(messages) <= keep_recent + 2:  # +2 for first user + system
        return messages
    
    old_messages = messages[:-keep_recent]
    recent_messages = messages[-keep_recent:]
    
    # 用 haiku 做摘要（便宜且快）
    summary_resp = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=512,
        messages=[
            {
                "role": "user",
                "content": f"请用 200 字以内概括以下对话历史的关键信息和决策：\n\n{str(old_messages)}",
            }
        ],
    )
    summary = summary_resp.content[0].text
    
    compressed = [
        {
            "role": "user",
            "content": f"[对话历史摘要] {summary}",
        },
        {
            "role": "assistant",
            "content": "已了解历史背景，继续处理当前任务。",
        },
    ]
    return compressed + recent_messages

八、成本护栏：防止 Agent 失控消费

没有成本上限的 Agent 在循环或误操作时可能在几分钟内烧掉大量 token。建议在每个 Agent 任务上设置预算上限：

class BudgetGuard:
    def __init__(self, max_tokens: int = 100_000, max_cost_usd: float = 1.0):
        self.max_tokens = max_tokens
        self.max_cost_usd = max_cost_usd
        self.total_input_tokens = 0
        self.total_output_tokens = 0
        
        # claude-sonnet-4-6 价格（仅作示例，实际以官网为准）
        self.input_price_per_1m = 3.0
        self.output_price_per_1m = 15.0
    
    def record(self, response: anthropic.types.Message):
        self.total_input_tokens += response.usage.input_tokens
        self.total_output_tokens += response.usage.output_tokens
    
    def check_budget(self) -> tuple[bool, str]:
        total_tokens = self.total_input_tokens + self.total_output_tokens
        if total_tokens > self.max_tokens:
            return False, f"token 用量超限: {total_tokens} > {self.max_tokens}"
        
        cost = (
            self.total_input_tokens / 1_000_000 * self.input_price_per_1m
            + self.total_output_tokens / 1_000_000 * self.output_price_per_1m
        )
        if cost > self.max_cost_usd:
            return False, f"预算超限: ${cost:.4f} > ${self.max_cost_usd}"
        
        return True, ""

关于 Agent 成本控制的更多实践，可参考 AI Coding Agent 成本控制。

九、降级策略的层次结构

一个完整的生产级 Agent 容错体系，应该形成如下的层次：

用户请求
  └─ 任务调度层
       ├─ 预算护栏 (BudgetGuard)
       └─ Agent 执行层
            ├─ 熔断器 (CircuitBreaker)
            ├─ 模型降级链 (MODEL_FALLBACK_CHAIN)
            ├─ 循环检测 (LoopDetector)
            └─ 工具调用层
                 ├─ 超时 + 指数退避
                 ├─ 工具级容错装饰器
                 └─ 上下文压缩

每一层独立处理自己的故障模式，不依赖上层来兜底，也不会向上传播可以自己处理的异常。

十、监控与可观测性

容错是被动防御，监控是主动发现。建议埋入以下指标：

  • agent.step_count：每次任务的步骤数，超过阈值报警
  • agent.tool_failure_rate：工具调用失败率
  • agent.model_fallback_count：模型降级次数（持续升高说明主模型有问题）
  • agent.budget_exceeded_count：预算超限次数
  • agent.loop_detected_count：循环检测触发次数

配合 Langfuse 或 OpenTelemetry 做链路追踪，可以在问题发生时快速定位是哪一层的故障。

十一、相关阅读

生产级 AI Agent 的稳定运行，离不开可靠的 API 基础设施。YoTradeApi 提供高可用的 Claude、GPT 等模型中转，内置速率均衡与自动切换，帮助你的 Agent 在国内网络环境下保持稳定。

相关阅读