文章简介：在AI回答采集系统中，调用大模型API时常遇到限流（429）、超时或服务端错误（5xx）。本文从实际工程出发，设计并实现指数退避重试、熔断和降级策略，给出Python代码、参数说明和验证方法，帮助开发者构建稳定的API调用层。

一、业务背景与问题

假设有一个AI回答采集系统，每天需要调用大模型API数万次，采集特定领域的问题回答。系统使用Python 3.10，通过httpx库调用OpenAI兼容接口（示例使用gpt-3.5-turbo），部署在Ubuntu 22.04上。

运行一段时间后，发现以下问题：

• 高峰期频繁出现429限流，导致采集任务中断。
• 偶尔出现502/503服务端错误，重试后恢复。
• 网络波动导致连接超时，任务失败。

如果不处理这些异常，采集任务会中断、数据丢失，甚至触发上游系统的连锁故障。

二、异常类型与处理策略

首先梳理可能遇到的异常类型，区分可重试和不可重试：

异常类型	HTTP状态码	典型原因	是否可重试
限流	429	请求频率超过API配额	是
服务端错误	500/502/503	API服务临时故障	是
超时	无（连接超时/读取超时）	网络波动或API响应慢	是
认证错误	401	API Key无效或过期	否
参数错误	400	请求参数格式错误	否

对可重试的异常，采用指数退避重试；对连续失败，引入熔断保护；当重试和熔断都失效时，执行降级。

三、指数退避重试

核心思路：每次重试的等待时间按指数增长，并加入随机抖动，避免对API造成二次压力。

import time
import random
from typing import Callable, Any

def retry_with_exponential_backoff(
    func: Callable,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    retryable_exceptions: tuple = (TimeoutError, ConnectionError),
    retryable_status_codes: tuple = (429, 500, 502, 503),
) -> Callable:
    def wrapper(*args, **kwargs):
        last_exception = None
        for attempt in range(max_retries + 1):
            try:
                response = func(*args, **kwargs)
                if response.status_code in retryable_status_codes:
                    if attempt == max_retries:
                        response.raise_for_status()
                    delay = min(base_delay * (2 ** attempt) + random.uniform(0, 0.5), max_delay)
                    time.sleep(delay)
                    continue
                return response
            except retryable_exceptions as e:
                last_exception = e
                if attempt == max_retries:
                    raise
                delay = min(base_delay * (2 ** attempt) + random.uniform(0, 0.5), max_delay)
                time.sleep(delay)
        raise last_exception
    return wrapper

参数说明：

• max_retries=5：重试5次，加上首次请求共6次。
• base_delay=1.0：首次重试等待1秒，第二次2秒，第三次4秒，以此类推。
• max_delay=60.0：最大等待时间60秒，防止无限增长。
• 随机抖动random.uniform(0, 0.5)：避免多个客户端同时重试造成惊群效应。

踩坑点：

• 对于429限流，优先读取响应头中的Retry-After字段。如果存在，应使用该值作为等待时间，而不是固定指数退避。
• 不要对401和400错误重试，重试无意义且可能触发账号封禁。

四、熔断机制

当错误率达到阈值时，暂时切断请求，让系统恢复。

import time
from enum import Enum

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

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 30.0,
        half_open_max_requests: int = 1,
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_requests = half_open_max_requests
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0.0
        self.half_open_requests = 0

    def call(self, func: Callable, *args, **kwargs) -> Any:
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_requests = 0
            else:
                raise CircuitBreakerOpenError("Circuit breaker is open")

        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_requests >= self.half_open_max_requests:
                raise CircuitBreakerOpenError("Half-open max requests reached")
            self.half_open_requests += 1

        try:
            result = func(*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.state = CircuitState.CLOSED
            self.failure_count = 0
            self.half_open_requests = 0
        elif self.state == CircuitState.CLOSED:
            self.failure_count = 0

    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

class CircuitBreakerOpenError(Exception):
    pass

使用场景：当API连续失败超过5次，熔断器打开，后续请求直接快速失败，避免浪费资源。30秒后进入半开状态，允许1个请求探测API是否恢复。如果成功，熔断器关闭；如果失败，重新打开。

参数取舍：

• failure_threshold：根据API的SLA调整。如果API偶尔抖动，阈值可设大些（如10次）；如果经常失败，设小些（如3次）。
• recovery_timeout：太短可能导致频繁开关，太长影响可用性。建议根据API平均恢复时间设置。

五、降级策略

当重试和熔断都无法恢复时，提供降级方案：

1. 返回缓存数据：如果之前有成功采集的结果，直接返回缓存。
2. 返回默认值：例如空字符串或预设的占位文本。
3. 跳过当前任务：记录失败日志，继续处理后续任务。
4. 切换备用API：如果有多个API供应商，自动切换到备用。

from functools import lru_cache

@lru_cache(maxsize=100)
def get_cached_response(prompt: str) -> str:
    """从缓存获取历史回答，缓存键为prompt。实际实现中应使用Redis等外部缓存。"""
    return None

def fetch_with_fallback(prompt: str, api_func: Callable) -> str:
    try:
        return api_func(prompt)
    except Exception:
        cached = get_cached_response(prompt)
        if cached:
            return cached
        return ""  # 返回默认值

注意：缓存数据应设置过期时间，避免返回过时信息。

六、整体实现

结合以上策略，使用tenacity库简化重试逻辑，实现完整采集函数：

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=60),
    retry=retry_if_exception_type((httpx.TimeoutException, httpx.HTTPStatusError)),
)
def call_llm_api(prompt: str, api_key: str, base_url: str) -> str:
    with httpx.Client(timeout=30.0) as client:
        response = client.post(
            f"{base_url}/chat/completions",
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "model": "gpt-3.5-turbo",
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.7,
            },
        )
        response.raise_for_status()
        return response.json()["choices"][0]["message"]["content"]

def collect_answer(prompt: str, circuit_breaker: CircuitBreaker) -> str:
    try:
        return circuit_breaker.call(call_llm_api, prompt, api_key="sk-xxx", base_url="https://api.openai.com/v1")
    except CircuitBreakerOpenError:
        return fetch_with_fallback(prompt, call_llm_api)

七、验证结果

在测试环境中使用Mock服务器模拟异常：

1. 模拟429限流：返回429状态码，观察重试行为。正常情况下，重试间隔依次为1秒、2秒、4秒、8秒、16秒左右，5次重试后如果仍失败则抛出异常。
2. 模拟连续5次500错误：熔断器在连续5次失败后打开，后续请求直接抛出CircuitBreakerOpenError，触发降级返回空字符串。
3. 检查日志：记录每次重试的原因、等待时间、熔断状态切换和降级路径。

预期日志输出示例：

2026-07-27 10:00:01 WARNING  Attempt 1 failed with 429, retrying in 1.2s
2026-07-27 10:00:02 WARNING  Attempt 2 failed with 429, retrying in 2.3s
2026-07-27 10:00:04 WARNING  Attempt 3 failed with 429, retrying in 4.1s
2026-07-27 10:00:08 WARNING  Attempt 4 failed with 429, retrying in 8.5s
2026-07-27 10:00:16 WARNING  Attempt 5 failed with 429, retrying in 16.3s
2026-07-27 10:00:32 ERROR    All retries exhausted, raising exception

八、常见问题与避坑

1. 不要对所有异常都重试：401认证错误、400参数错误重试无意义，应直接失败并告警。
2. 重试次数不宜过多：5次左右足够，过多会延长任务总时间，且可能被API封禁。
3. 熔断恢复时间要合理：太短可能导致频繁开关，太长影响可用性。建议根据API的SLA调整。
4. 降级数据要有时效性：缓存数据应设置过期时间，避免返回过时信息。
5. 日志记录要详细：记录每次重试的原因、等待时间、最终结果，便于排查。

九、总结

本文针对AI回答采集系统中调用大模型API的异常，实现了指数退避重试、熔断和降级策略。指数退避配合随机抖动有效缓解了限流问题；熔断机制保护了系统资源；降级策略保证了极端情况下的可用性。该方案适用于大多数API调用场景，但具体参数（重试次数、熔断阈值、恢复时间）需要根据实际业务和API特性调整。

参考资料

• OpenAI API Error Handling Guide
• tenacity documentation