AI Agent错误处理与优雅重试——生产级实现指南

做AI Agent的朋友肯定遇到过：LLM突然超时、工具调用莫名其妙报错、外部API甩个500回来。你的Agent怎么办？直接崩溃？原地卡死？

生产环境的Agent得接受一个现实——错误不是意外，是常态。好的错误处理不是事后打补丁，是架构的一部分。

这篇文章不讲概念，直接上能跑的代码，从最简单的重试到断路器模式，一步步搭一套生产级的错误处理体系。

---

一、基础：LLM调用的重试策略

绝大多数LLM调用失败都是暂时的（网络抖一下、服务端限了下流），重试就能搞定。

指数退避 + 抖动

import asyncio
import random
from typing import TypeVar, Callable, Awaitable

T = TypeVar('T')

async def retry_with_backoff(
    fn: Callable[[], Awaitable[T]],
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0
) -> T:
    """带指数退避和抖动的异步重试"""
    last_exception = None
    
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except Exception as e:
            last_exception = e
            if attempt >= max_retries:
                raise
            
            # 指数退避: 2^attempt * base_delay
            delay = min(base_delay * (2 ** attempt), max_delay)
            # 抖动: ±25%，防止同时触发的请求一起重试造成雪崩
            jitter = delay * 0.25 * random.uniform(-1, 1)
            actual_delay = delay + jitter
            
            print(f"[重试 {attempt + 1}/{max_retries}] 等 {actual_delay:.1f}s...")
            await asyncio.sleep(actual_delay)
    
    raise last_exception  # type: ignore

分清哪些错值得重试

不是所有错误重试都有用。认证过期、请求体太大这种，重试一万次也是白搭。

class RetryableError(Exception):
    """可以重试的错误"""
    pass

class RateLimitError(RetryableError):
    pass

class ServerOverloadedError(RetryableError):
    pass

class AuthError(Exception):
    """别重试了，检查配置吧"""
    pass

def is_retryable(status_code: int, error_body: str) -> bool:
    if status_code in (429, 503, 504):
        return True
    if status_code in (400, 401, 403, 413):
        return False
    if status_code >= 500:
        return True
    return False

踩坑： 有次OpenAI返回503，重试了三次都失败，排查半天发现是API key过期了——503只是表象。所以不能只看状态码，还得结合错误消息内容来判断。

---

二、进阶：给不同LLM Provider定制策略

各家LLM厂商的限流套路不一样，统一的重试策略效率很低。

| 厂商 | 怎么限流 | 怎么重试 |

|------|---------|---------|

| OpenAI | 告诉你 Retry-After 头 | 照它说的等 |

| Claude | 固定速率（每分钟多少请求） | 算剩余配额，按比率等 |

| 本地vLLM | 排队 | 短间隔快试（排空概率大） |

| 国内厂商 | 随机限 | 抖动范围大一点 |

class LLMProviderHandler:
    """按厂商标配不同的重试等待时间"""
    
    @staticmethod
    def get_retry_delay(
        provider: str, status_code: int,
        response_headers: dict, attempt: int
    ) -> float:
        if provider == "openai" and "Retry-After" in response_headers:
            return float(response_headers["Retry-After"])
        
        if provider == "anthropic":
            base = 2.0 * (2 ** attempt)
            return base + random.uniform(0, 2)
        
        if provider == "local-vllm":
            return 0.5 * (2 ** attempt) + random.uniform(0, 0.5)
        
        return min(2.0 * (2 ** attempt), 60.0) + random.uniform(0, 1)

工具调用失败了怎么办

工具调用（Function Calling）报错跟LLM调用报错不一样。工具出错了，往往需要把错误喂回给LLM让它重新生成参数，而不是直接重试上一次的调用。

async def retry_tool_call(
    agent, tool_name: str, original_prompt: str,
    previous_error: str, max_retries: int = 2
) -> dict:
    """带着报错信息让LLM重新生成工具调用参数"""
    
    for attempt in range(max_retries + 1):
        enhanced_prompt = f"""{original_prompt}

【系统提示】上次调 {tool_name} 报了这个错：
{previous_error}

参数可能有问题，检查一下重新调。"""
        
        response = await agent.llm_call(enhanced_prompt)
        tool_calls = extract_tool_calls(response)
        
        if tool_calls:
            try:
                result = await execute_tool(tool_calls[0])
                return result
            except Exception as e:
                previous_error = str(e)
                continue
        
        # LLM觉得不需要调了，直接返回文本
        return {"type": "text", "content": response}
    
    raise ToolCallError(f"工具 {tool_name} 重试了 {max_retries} 次还是不行")

---

三、高级：断路器（Circuit Breaker）

重试能解决临时故障，但如果下游服务挂了半小时（比如某LLM大面积宕机），一直重试只是浪费资源和时间。这时候需要断路器。

三个状态的切换

正常运行 ──连续失败N次──▶ 断开
     ▲                     │
     │                     │ 等一段时间
     │                     ▼
     │                 半开试探
     │                     │
     └──试探成功───────────┘
         试探失败 ────▶ 重新断开

完整实现

import time
import asyncio
from enum import Enum
from dataclasses import dataclass

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5
    recovery_timeout: float = 30.0
    half_open_max_requests: int = 3
    success_threshold: int = 2

class CircuitBreaker:
    def __init__(self, name: str, config: CircuitBreakerConfig = None):
        self.name = name
        self.config = config or CircuitBreakerConfig()
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0.0
        self.half_open_successes = 0
        self.half_open_requests = 0
    
    async def call(self, fn, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time >= self.config.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_successes = 0
                self.half_open_requests = 0
                print(f"[{self.name}] OPEN → HALF_OPEN")
            else:
                raise CircuitBreakerOpenError(f"断路器 {self.name} 已断开")
        
        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_requests >= self.config.half_open_max_requests:
                raise CircuitBreakerOpenError(f"半开状态已达试探上限")
            self.half_open_requests += 1
        
        try:
            result = await fn(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
    
    def _on_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.half_open_successes += 1
            if self.half_open_successes >= self.config.success_threshold:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                print(f"[{self.name}] HALF_OPEN → CLOSED")
        else:
            self.failure_count = 0
    
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
            print(f"[{self.name}] HALF_OPEN → OPEN（试探失败）")
        elif self.failure_count >= self.config.failure_threshold:
            self.state = CircuitState.OPEN
            print(f"[{self.name}] CLOSED → OPEN（连续失败{self.failure_count}次）")

用起来

# 不同厂商用不同配置
openai_breaker = CircuitBreaker("OpenAI", CircuitBreakerConfig(
    failure_threshold=3,
    recovery_timeout=60.0,
))

claude_breaker = CircuitBreaker("Claude", CircuitBreakerConfig(
    failure_threshold=5,
    recovery_timeout=30.0,
))

async def call_llm_safely(provider: str, prompt: str) -> str:
    breaker = openai_breaker if provider == "openai" else claude_breaker
    
    try:
        return await breaker.call(lambda: llm_client.chat(provider, prompt))
    except CircuitBreakerOpenError:
        return get_fallback_response(provider, prompt)

---

四、完整的错误处理链路

把上面的组件串起来：

用户输入
    │
    ▼
① 输入检查 ─── 格式、长度、敏感词
    │
    ▼
② LLM调用 ─── 断路器检查 → 指数退避重试(最多3次) → 超时控制(30s)
    │
    ▼
③ 工具调用 ─── 参数校验 → 错误反馈重试(带异常信息)
    │
    ▼
④ 输出校验 ─── 格式检查 → 完整性验证
    │
    ▼
   输出

超时控制最容易漏

LLM调用最怕的不是报错，是卡住。不加超时控制，一个慢请求能把整个Agent拖死。

async def call_with_timeout(fn, timeout: float = 30.0):
    try:
        return await asyncio.wait_for(fn(), timeout=timeout)
    except asyncio.TimeoutError:
        logger.warning(f"LLM调用超时 ({timeout}s)")
        raise RetryableError("LLM调用超时")

建议根据prompt长度动态设超时：短prompt给15s，长上下文给60s+。

---

五、记录与告警

每次重试、每次断路器状态变化，记下来。

| 什么情况 | 提醒 |

|---------|------|

| LLM 5分钟内失败率 > 5% | 通知 |

| 连续重试超过3次 | 通知 |

| 断路器断开了 | 立刻通知 |

| 平均响应时间超过15s | 关注 |

---

六、几个容易忽略的点

1. 重试不是万能的——非幂等的操作（比如下单）不能简单重试
2. 断路器恢复时间别太短——LLM大面积故障通常持续几分钟起
3. 超时得按场景来——短查询15s，长上下文60s+
4. 用asyncio，别用线程——Python的并发重试用asyncio走
5. 测试你的错误处理——模拟429、网络断开、超时，看代码扛不扛得住

最后一句： 错误处理的终极目标不是消灭错误（不可能），而是错误来了系统不会崩，用户体验不会断。