批量调用大模型API时，遇到HTTP 429限流和500服务端错误，导致采集任务中断。本文设计了一套异常分类、指数退避重试、熔断和降级方案，适用于Python异步环境。读者可参考实现自己的重试逻辑，但需根据实际API限流策略调整参数。本文不涉及流式响应部分失败的处理。

环境说明

• 编程语言：Python 3.10+
• 大模型API：兼容OpenAI接口格式（如GPT系列、Claude、本地部署模型）
• 依赖库：httpx、asyncio
• 运行环境：Linux服务器，Python异步框架

异常分类

首先需要区分哪些异常值得重试。我根据HTTP状态码和异常类型做了分类：

异常类型	示例	是否可重试
网络超时	httpx.TimeoutException	是
服务端错误	HTTP 500/502/503	是
限流	HTTP 429	是（需等待）
认证失败	HTTP 401/403	否
请求参数错误	HTTP 400	否
模型不可用	模型不存在、上下文超长	否

不可重试的异常直接抛出，避免浪费重试次数。

重试策略：指数退避 + 抖动

最初我尝试了固定间隔重试，但发现多个请求同时重试时容易加剧限流。改用指数退避后，退避时间随尝试次数指数增长，再加入随机抖动分散重试时间。

import random
import time

def compute_backoff(attempt: int, base_delay: float = 1.0, max_delay: float = 60.0, jitter: float = 0.5) -> float:
    delay = min(base_delay * (2 ** attempt), max_delay)
    jitter_value = random.uniform(0, jitter * delay)
    return delay + jitter_value

参数说明：

• base_delay：基础延迟，默认1秒。
• max_delay：最大延迟，防止无限增长。
• jitter：抖动系数，默认0.5，即抖动范围为0~50%的延迟。

例如，attempt=0时，延迟约为1.0_{1.5秒；attempt=1时，约为2.0}3.0秒；attempt=2时，约为4.0~6.0秒。

熔断器

当连续失败达到阈值时，熔断器打开，直接拒绝请求，让下游服务恢复。我实现了简单的状态机：

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 30.0):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"

    def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "HALF_OPEN"
            else:
                raise Exception("Circuit breaker is OPEN")
        try:
            result = func(*args, **kwargs)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
            raise e

熔断器状态：

• CLOSED：正常请求，记录失败次数。
• OPEN：失败达到阈值，直接拒绝。
• HALF_OPEN：超时后允许一个试探请求，成功则关闭，失败则继续打开。

降级策略

当重试耗尽或熔断打开时，需要降级：

• 返回默认值（如空字符串、缓存结果）
• 记录失败任务到死信队列，后续人工处理
• 切换备用模型或API端点

核心实现

以下是一个完整的异步调用函数，整合了异常分类、重试、熔断和降级：

import asyncio
import logging
from typing import Callable, Any

logger = logging.getLogger(__name__)

async def call_with_retry(
    api_func: Callable,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    circuit_breaker: CircuitBreaker = None,
    fallback: Any = None,
) -> Any:
    last_exception = None
    for attempt in range(max_retries + 1):
        try:
            if circuit_breaker:
                return circuit_breaker.call(api_func)
            else:
                return await api_func()
        except Exception as e:
            last_exception = e
            if not _is_retryable(e):
                logger.warning(f"Non-retryable exception: {e}")
                break
            if attempt < max_retries:
                delay = compute_backoff(attempt, base_delay, max_delay)
                logger.info(f"Retry {attempt+1}/{max_retries} after {delay:.2f}s due to {e}")
                await asyncio.sleep(delay)
            else:
                logger.error(f"All retries exhausted: {e}")
    if fallback is not None:
        return fallback
    raise last_exception

def _is_retryable(e: Exception) -> bool:
    if isinstance(e, (TimeoutError, ConnectionError)):
        return True
    if hasattr(e, 'status_code'):
        return e.status_code in (429, 500, 502, 503)
    return False

验证结果

在测试环境中模拟不同异常，正常情况下应当看到以下现象：

• 网络超时：重试3次后成功，日志显示退避时间依次增加（例如约1.2s、2.5s、5.1s，具体取决于抖动）。
• 连续5次500错误：熔断器打开，后续请求直接返回降级值，30秒后半开试探成功。
• 认证失败：立即抛出异常，不重试。

踩坑与避坑

1. 重试次数与超时：总重试时间可能很长，需设置整体超时。例如单次调用超时10s，重试3次，总超时应设为10 + 2 + 4 + 8 = 24s以上。
2. 抖动范围：抖动不宜过大，否则退避时间不可控。建议抖动系数为0.5。
3. 熔断恢复：半开状态只放行一个请求，避免瞬间压垮服务。
4. 日志记录：记录每次重试的原因、退避时间、最终结果，便于排查。
5. 幂等性：确保API调用是幂等的，否则重试可能导致重复写入。

总结

本文针对AI回答采集中的异常场景，设计了一套包含异常分类、指数退避重试、熔断和降级的工程方案。核心在于区分可重试与不可重试异常，并用熔断器保护下游服务。该方案适用于大多数大模型API调用场景，但需根据实际API的限流策略和业务容忍度调整参数。当前限制：未处理流式响应的部分失败，以及长任务的中断恢复。

参考资料

• OpenAI API Error Codes: https://platform.openai.com/docs/guides/error-codes
• Circuit Breaker Pattern: https://martinfowler.com/bliki/CircuitBreaker.html