大模型 API 错误处理与重试

大模型错误处理与重试

API 调用中的错误类型、处理策略和重试机制。

为什么需要错误处理？

大模型 API 调用面临多种不确定性：

• 网络波动：请求超时、连接中断
• 服务限流：超出速率限制被拒绝
• 服务故障：厂商服务临时不可用
• 参数错误：请求格式不正确

良好的错误处理机制可以提升应用的稳定性和用户体验，避免因单次失败导致整个流程中断。

错误处理的重要性：

在生产环境中，API 调用失败是常态而非例外。根据经验，即使是最稳定的 API 服务，也会有 0.1%-1% 的请求失败率。对于高流量应用，这意味着每天可能有数千次失败。

没有错误处理的后果：

错误处理流程

一个健壮的错误处理流程应该包含：错误分类、重试决策、降级策略、告警通知。

flowchart TD
    A[API 调用] --> B{成功?}
    B -->|是| C[返回结果]
    B -->|否| D{错误类型}
    
    D -->|429 限流| E[指数退避重试]
    D -->|500 服务错误| E
    D -->|401 认证| F[检查 API Key]
    D -->|400 参数| G[检查请求格式]
    D -->|网络错误| H[重试/切换备用]
    
    E --> I{重试次数?}
    I -->|< 3| A
    I -->|>= 3| J[降级/报警]

错误处理原则：

1. 快速失败：对于不可恢复的错误（如认证失败），立即返回，不要重试
2. 优雅降级：当主服务不可用时，切换到备用方案
3. 用户友好：向用户展示有意义的错误信息，而非技术细节
4. 可观测：记录所有错误，便于事后分析

错误码对照表

不同厂商的错误码基本一致，但有细微差异：

错误码详解：

400 Bad Request 请求格式有问题，常见原因：

• JSON 格式错误
• 缺少必需参数（如 Claude 的 max_tokens）
• 参数类型错误
• 模型名称不存在

401 Unauthorized 认证失败，检查：

• API Key 是否正确
• API Key 是否过期
• 环境变量是否设置

429 Too Many Requests 触发速率限制，这是最常见的可重试错误。响应头通常包含 Retry-After 字段，指示应该等待多久。

500/503 Server Error 服务端问题，通常是临时性的。建议等待后重试。

重试策略

并非所有错误都应该重试。一般原则：

• 可重试：429（限流）、500/503（服务错误）、网络超时
• 不可重试：400（参数错误）、401（认证失败）、403（无权限）

重试时应避免”重试风暴”，即大量客户端同时重试导致服务雪崩。

指数退避

指数退避是最常用的重试策略。每次重试的等待时间呈指数增长（1s → 2s → 4s），可以有效分散重试请求，减轻服务器压力。

为什么使用指数退避？

建议加入随机抖动（jitter），避免多个客户端在同一时刻重试。

gantt
    title 指数退避重试时间线
    dateFormat X
    axisFormat %s秒
    
    section 重试
    第1次请求     :0, 1
    等待 1s       :1, 2
    第2次重试     :2, 3
    等待 2s       :3, 5
    第3次重试     :5, 6
    等待 4s       :6, 10
    第4次重试     :10, 11

实现

import time
import random
from openai import RateLimitError

def call_with_retry(func, max_retries=3):
    for i in range(max_retries):
        try:
            return func()
        except RateLimitError:
            wait = (2 ** i) + random.uniform(0, 1)  # 指数退避 + 抖动
            time.sleep(wait)
    raise Exception("重试耗尽")

使用 tenacity 库（推荐）：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import RateLimitError

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=60),
    retry=retry_if_exception_type(RateLimitError)
)
def call_api():
    return client.chat.completions.create(...)

SDK 内置重试

主流 SDK 都内置了重试机制，通常不需要自己实现：

# OpenAI 配置重试
from openai import OpenAI
client = OpenAI(max_retries=5)

# Anthropic 配置重试
from anthropic import Anthropic
client = Anthropic(max_retries=5)

SDK 重试的行为：

• 只对可重试错误（429、500、503）自动重试
• 使用指数退避策略
• 尊重 Retry-After 响应头
• 超过最大次数后抛出异常

异常类型

OpenAI

from openai import (
    APIError,           # 基类
    RateLimitError,     # 429
    APIConnectionError, # 网络
    AuthenticationError # 401
)

完整的异常处理示例：

from openai import OpenAI, APIError, RateLimitError, APIConnectionError, AuthenticationError

client = OpenAI()

try:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello"}]
    )
except AuthenticationError:
    print("API Key 无效，请检查配置")
except RateLimitError as e:
    print(f"触发速率限制，请等待后重试: {e}")
except APIConnectionError:
    print("网络连接失败，请检查网络")
except APIError as e:
    print(f"API 错误: {e.status_code} - {e.message}")

Anthropic

from anthropic import (
    APIError,
    RateLimitError,
    APIConnectionError,
    AuthenticationError
)

熔断降级

当某个服务持续失败时，继续重试只会浪费资源。熔断器模式可以在检测到连续失败后”断开”请求，直接返回降级响应，待服务恢复后再”闭合”。

熔断器有三种状态：

• 关闭：正常请求，记录失败次数
• 打开：拒绝请求，返回降级响应
• 半开：允许少量探测请求，判断是否恢复

stateDiagram-v2
    [*] --> 关闭
    关闭 --> 打开: 连续失败 >= 阈值
    打开 --> 半开: 超时后
    半开 --> 关闭: 探测成功
    半开 --> 打开: 探测失败

简单实现

class CircuitBreaker:
    def __init__(self, threshold=5, timeout=60):
        self.failures = 0
        self.threshold = threshold
        self.timeout = timeout
        self.last_fail = 0
        
    def call(self, func):
        if self.failures >= self.threshold:
            if time.time() - self.last_fail < self.timeout:
                raise Exception("熔断中")
            self.failures = 0  # 半开状态
        
        try:
            result = func()
            self.failures = 0
            return result
        except Exception:
            self.failures += 1
            self.last_fail = time.time()
            raise

多模型降级

生产环境中，建议配置多个模型作为备选。当主模型不可用时，自动切换到备用模型，保证服务可用性。

降级顺序通常按照：能力相近 → 成本可接受 → 响应速度 来排列。例如 GPT-4o 故障时，可以降级到 Claude Sonnet，再降级到 Gemini Pro。

flowchart LR
    A[请求] --> B[GPT-4o]
    B -->|失败| C[Claude]
    C -->|失败| D[Gemini]
    D -->|失败| E[返回错误]
    
    B -->|成功| F[返回]
    C -->|成功| F
    D -->|成功| F

MODELS = ["gpt-4o", "claude-sonnet-4-20250514", "gemini-1.5-flash"]

def call_with_fallback(messages):
    for model in MODELS:
        try:
            return completion(model=model, messages=messages)
        except Exception:
            continue
    raise Exception("所有模型不可用")

最佳实践

超时处理

超时是最常见的网络问题。需要区分不同类型的超时：

超时重试策略

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10)
)
def call_with_timeout(client, messages):
    try:
        return client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            timeout=60.0
        )
    except httpx.TimeoutException:
        raise  # 触发重试

日志记录

良好的日志是排查问题的关键。建议记录以下信息：

请求日志

import logging
import time

logger = logging.getLogger(__name__)

def log_request(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        request_id = generate_id()
        
        logger.info(f"[{request_id}] 开始请求 model={kwargs.get('model')}")
        
        try:
            result = func(*args, **kwargs)
            latency = time.time() - start
            logger.info(f"[{request_id}] 成功 latency={latency:.2f}s tokens={result.usage.total_tokens}")
            return result
        except Exception as e:
            latency = time.time() - start
            logger.error(f"[{request_id}] 失败 latency={latency:.2f}s error={e}")
            raise
    return wrapper

日志字段建议

监控告警

关键指标

监控代码示例

from dataclasses import dataclass
from collections import defaultdict
import time

@dataclass
class Metrics:
    total_requests: int = 0
    failed_requests: int = 0
    retried_requests: int = 0
    total_latency: float = 0
    error_counts: dict = None
    
    def __post_init__(self):
        self.error_counts = defaultdict(int)

metrics = Metrics()

def record_request(success: bool, latency: float, error_type: str = None, retried: bool = False):
    metrics.total_requests += 1
    metrics.total_latency += latency
    
    if not success:
        metrics.failed_requests += 1
        if error_type:
            metrics.error_counts[error_type] += 1
    
    if retried:
        metrics.retried_requests += 1

def get_error_rate():
    if metrics.total_requests == 0:
        return 0
    return metrics.failed_requests / metrics.total_requests

def check_alerts():
    error_rate = get_error_rate()
    if error_rate > 0.05:
        send_alert(f"错误率过高: {error_rate:.2%}")

完整错误处理示例

import time
import random
import logging
from openai import OpenAI, RateLimitError, APIConnectionError, APIError

logger = logging.getLogger(__name__)

class LLMClient:
    def __init__(self, max_retries=3, timeout=60):
        self.client = OpenAI(timeout=timeout)
        self.max_retries = max_retries
        self.fallback_models = ["gpt-4o", "gpt-4o-mini"]
    
    def chat(self, messages, model="gpt-4o"):
        """带完整错误处理的聊天方法"""
        last_error = None
        
        # 尝试主模型和备用模型
        models_to_try = [model] + [m for m in self.fallback_models if m != model]
        
        for current_model in models_to_try:
            for attempt in range(self.max_retries):
                try:
                    start = time.time()
                    response = self.client.chat.completions.create(
                        model=current_model,
                        messages=messages
                    )
                    
                    latency = time.time() - start
                    logger.info(f"成功 model={current_model} latency={latency:.2f}s")
                    return response
                    
                except RateLimitError as e:
                    wait = (2 ** attempt) + random.uniform(0, 1)
                    logger.warning(f"限流，等待 {wait:.1f}s 后重试")
                    time.sleep(wait)
                    last_error = e
                    
                except APIConnectionError as e:
                    logger.warning(f"连接错误，尝试下一个模型")
                    last_error = e
                    break  # 跳到下一个模型
                    
                except APIError as e:
                    if e.status_code >= 500:
                        wait = (2 ** attempt) + random.uniform(0, 1)
                        logger.warning(f"服务器错误，等待 {wait:.1f}s 后重试")
                        time.sleep(wait)
                        last_error = e
                    else:
                        # 4xx 错误不重试
                        raise
        
        raise Exception(f"所有模型和重试都失败: {last_error}")

# 使用
client = LLMClient()
response = client.chat([{"role": "user", "content": "Hello"}])

常见问题

Q: 应该重试多少次？

一般建议 2-3 次。太少可能无法恢复临时故障，太多会增加延迟和成本。

Q: 重试间隔应该多长？

使用指数退避：1s → 2s → 4s，加上随机抖动。最大等待时间建议不超过 60s。

Q: 如何处理部分成功的流式响应？

流式响应中断时，已接收的内容可能是不完整的。建议：

1. 检测中断（没有收到 [DONE]）
2. 重试整个请求
3. 或者保存已接收内容，提示用户继续

Q: 多个服务同时调用 API，如何避免同时触发限流？

1. 使用不同的 API Key
2. 实现全局限流器
3. 错开请求时间（加随机延迟）

相关文档

• 01-大模型API-协议概述
• 08-大模型-SDK使用指南
• 14-大模型-生产环境最佳实践
• 25-大模型-常见问题与排错