基础错误处理：超时、重试、限流与指数退避

前面几篇我们让模型返回了结构化 JSON、学会了流式输出，代码看上去已经能干活了。但只要程序开始真实、持续地调用大模型 API，就一定会遇到各种”不稳定”：网络抖一下就超时，高峰期被限流，服务商偶尔返回 500。如果对这些错误毫无准备，程序会在第一次网络波动时直接崩溃。本篇讲清楚大模型 API 调用中最常见的几类错误，以及如何用超时、重试和指数退避让程序稳得住。

先认识会出哪些错

调用大模型 API 时，错误大致分三类：

连接类错误：根本没建立连接，或请求发出去了但迟迟没有响应。对应 SDK 的 APIConnectionError 和它的子类 APITimeoutError。常见于网络不稳定、DNS 解析失败、服务商瞬时不可达。
限流错误：请求太频繁或超出配额，服务端返回 HTTP 429。对应 SDK 的 RateLimitError。通常响应头里会带一个 Retry-After，告诉你多久之后再试。
服务端错误：服务商内部出问题，返回 HTTP 500、502、503 等。对应 SDK 的 InternalServerError。这类错误往往是暂时的，等一会儿重试通常就能恢复。

除了这三类”可重试”的错误，还有一些错误重试也没用：BadRequestError（400，请求参数本身有问题）、AuthenticationError（401，密钥错误）、NotFoundError（404，模型名写错）。这些属于”你这边的问题”，重试一百次结果也一样，应当直接抛出让开发者去修。

一条原则：只重试可能恢复的瞬时错误，不要重试逻辑性错误。 对 400 重试只是在浪费时间和 token。

设置超时：给每次请求设一个上限

默认情况下，OpenAI Python SDK 的超时时间是 600 秒（10 分钟）。对于大多数应用来说这太长了——如果一次请求卡住 10 分钟，用户体验会非常差。我们可以在创建客户端时设置更合理的超时：

import os

from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ.get("OPENAI_BASE_URL"),
    timeout=30.0,      # 请求整体超时 30 秒
    max_retries=2,     # SDK 内置重试次数
)

resp = client.chat.completions.create(
    model=os.environ.get("MODEL_NAME", "gpt-4o-mini"),
    messages=[{"role": "user", "content": "用一句话介绍 Python。"}],
)
print(resp.choices[0].message.content)

timeout 是单次请求的等待上限，超时后会抛出 APITimeoutError。max_retries 控制 SDK 在遇到可重试错误时自动重试的次数，默认就是 2。也就是说，SDK 默认已经会帮你重试两次，前提是你没有把它设成 0。

需要注意的是，这里的 timeout 是”单次请求”的超时，不是”整个调用过程”的超时。如果开启了重试，总耗时大约是 timeout × (1 + max_retries) 再加上重试之间的等待时间。

SDK 内置的重试机制

OpenAI Python SDK 自带重试逻辑，不需要你手写。它会自动重试以下情况：

HTTP 408（请求超时）、409（锁冲突）、429（限流）
HTTP 5xx（服务端错误）
连接超时 APITimeoutError

重试之间的等待时间采用指数退避加抖动：第一次重试等约 0.5 秒，第二次等约 1 秒，第三次约 2 秒，最长不超过 8 秒，并且每次会加一点随机抖动，避免大量客户端同时重试造成”惊群”。如果服务端在响应头里给了 Retry-After，SDK 会优先遵守这个值（但不超过 120 秒）。

这套默认机制对大多数场景已经够用。你只需要在创建客户端时把 max_retries 设成一个合理的值（比如 2 到 4），SDK 就会帮你处理瞬时抖动。

什么时候需要手写重试

SDK 内置重试有几个局限：它只处理 HTTP 层面的错误，不处理你业务逻辑里的错误；重试次数是固定的，无法根据错误类型做不同策略；它也不会把”重试了、还是失败”这件事告诉你，方便你记日志或降级。

当内置重试不够用时，就需要在外面再包一层手写重试。一个典型场景是：调用失败后，你想把错误记到日志里，或者在重试若干次仍失败后走降级逻辑（比如换一个更便宜的模型，或返回缓存结果）。

手写指数退避重试

下面是一个完整的手写重试示例，用 tenacity 库实现指数退避。tenacity 是 Python 生态里最常用的重试库之一，API 清晰、功能完备：

import logging
import os

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential_jitter,
    retry_if_exception_type,
    before_sleep_log,
)

from openai import OpenAI, RateLimitError, APIConnectionError, InternalServerError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ.get("OPENAI_BASE_URL"),
    max_retries=0,  # 关掉 SDK 内置重试，完全由 tenacity 接管
)


@retry(
    retry=retry_if_exception_type(
        (RateLimitError, APIConnectionError, InternalServerError)
    ),
    wait=wait_exponential_jitter(initial=1, max=16),
    stop=stop_after_attempt(4),
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def chat(messages: list[dict], model: str | None = None) -> str:
    """调用大模型，失败时按指数退避重试，最多 4 次。"""
    resp = client.chat.completions.create(
        model=model or os.environ.get("MODEL_NAME", "gpt-4o-mini"),
        messages=messages,
    )
    return resp.choices[0].message.content


if __name__ == "__main__":
    answer = chat([{"role": "user", "content": "用一句话介绍 Python。"}])
    print(answer)

几个关键点：

max_retries=0 关掉了 SDK 自己的重试，把重试逻辑完全交给 tenacity，避免两层重试叠加导致重试次数失控。
retry_if_exception_type 指定只重试这三类瞬时错误。BadRequestError、AuthenticationError 这些不在列表里，会直接抛出，不会浪费重试次数。
wait_exponential_jitter(initial=1, max=16) 表示第一次重试等 1 秒，之后指数增长，但单次等待不超过 16 秒，并带随机抖动。
stop_after_attempt(4) 表示最多尝试 4 次（1 次初始 + 3 次重试），超过就放弃并把最后一次的异常抛出。
before_sleep_log 在每次重试前打一条日志，方便排查”为什么慢”。

安装依赖：pip install tenacity openai python-dotenv。

限流的应对策略

遇到 429 限流时，重试只是”治标”。要从根本上减少限流，可以做几件事：

降低并发：如果你在批量调用，控制同时发出的请求数。最简单的办法是用 asyncio.Semaphore 或线程池限制并发数。
遵守 Retry-After：服务端返回 429 时通常会在响应头里给 Retry-After，告诉你多久之后再试。SDK 内置重试会自动遵守这个值；手写重试时也可以从异常的 response.headers 里读取。
错峰调用：如果业务允许，把非实时的批量任务安排在低峰时段。
分级降级：触发限流时临时切换到更便宜、配额更宽裕的模型，保证服务可用性。

下面是一个从 RateLimitError 中读取 Retry-After 的片段：

import time

from openai import RateLimitError

def chat_with_retry(messages, model=None):
    for attempt in range(4):
        try:
            return client.chat.completions.create(
                model=model or os.environ.get("MODEL_NAME", "gpt-4o-mini"),
                messages=messages,
            )
        except RateLimitError as e:
            # SDK 已根据 Retry-After 计算好建议等待时间，也可自行解析响应头
            retry_after = float(
                e.response.headers.get("Retry-After", 5)
                if e.response is not None else 5
            )
            print(f"触发限流，{retry_after:.1f} 秒后重试（第 {attempt + 1} 次）")
            time.sleep(retry_after)
    raise RuntimeError("重试 4 次后仍被限流")

这段代码没有用 tenacity，纯标准库实现，适合不想引入额外依赖的场景。它的逻辑是：捕获 RateLimitError 后，从响应头读 Retry-After，休眠相应时间再重试。

常见问题

重试会不会导致重复扣费？ 不会。chat.completions.create 是非幂等的，但 SDK 只在”请求没成功或没收到响应”时才重试——一旦收到完整响应，哪怕后续处理出错也不会重试。所以正常完成的请求只会计费一次。APITimeoutError 比较特殊：请求可能已经到达服务端，但由于没收到响应而重试，理论上可能产生两次计费。如果对成本敏感，可以把超时设得保守一点，并在日志里监控超时频率。

重试次数设多少合适？ 对交互式应用（比如聊天机器人），2 到 3 次足够，总等待时间控制在十几秒内，否则用户等不及。对后台批处理任务，可以设到 5 次以上，配合更长的退避间隔。关键是设置一个 stop_after_attempt 上限，避免无限重试。

指数退避为什么要加抖动？ 如果不加抖动，所有在某一时刻同时失败的客户端都会在完全相同的时间点重试，再次同时冲击服务端，造成”惊群效应”。加一点随机抖动（比如 0.75 到 1.25 倍的系数）能让重试时间错开，大幅减轻服务端压力。tenacity 的 wait_exponential_jitter 已经内置了抖动。

APIConnectionError 和 APITimeoutError 是什么关系？ APITimeoutError 是 APIConnectionError 的子类。连接超时属于连接错误的一种特殊情况。在 retry_if_exception_type 里写 APIConnectionError 就能同时覆盖连接失败和超时两种情况。

要不要对 BadRequestError 重试？ 不要。400 表示你的请求参数有问题（比如消息格式错误、模型名不存在），重试多少次结果都一样。正确的做法是修复请求参数。对 AuthenticationError（401）也一样——密钥错了，重试没用。

小结

大模型 API 的错误分三类：连接类（超时、断连）、限流（429）、服务端（5xx），前两类和第三类中的瞬时错误适合重试。
OpenAI Python SDK 自带指数退避重试，默认重试 2 次，通过 timeout 和 max_retries 即可调节，大多数场景够用。
需要更精细控制时，用 tenacity 手写重试：按异常类型决定是否重试，用指数退避加抖动控制节奏，设置最大尝试次数兜底。
限流的根本对策是降并发、遵守 Retry-After、错峰调用和分级降级，重试只是临时手段。
下一步我们将学习模型选择与成本基础，理解如何根据任务在能力、延迟和价格之间做权衡。

搜索时光笔记