出错、重试、限流和花钱

在自己电脑上跑例子时，API 调用几乎从不出错。上线之后就不一样了：高峰期服务器返回 429 说你请求太多，某次请求卡了一分钟没反应，偶尔还会有 500 错误。如果代码里没有准备，一个用户就会看到一大段报错，或者页面一直转圈。

还有钱。一个写错的循环，一个没设上限的重试，可能一晚上烧掉你一个月的预算。

这一课先把常见的错误都复现一遍，然后写一个能直接放进项目里的调用函数。

常见的错误

code/03-llm-apps/errors.py 故意制造了三种错误：

print("== 1. 密钥错误")
try:
    OpenAI(api_key="sk-wrong", base_url=BASE_URL).chat.completions.create(model=MODEL, messages=HELLO)
except openai.AuthenticationError as e:
    print(f"{type(e).__name__}，状态码 {e.status_code}")

print("== 2. 模型名写错")
try:
    OpenAI(api_key=os.environ["LLM_API_KEY"], base_url=BASE_URL).chat.completions.create(model="deepseek-flsh", messages=HELLO)
except openai.APIStatusError as e:
    print(f"{type(e).__name__}，状态码 {e.status_code}，{e.message[:100]}")

print("== 3. 超时（故意把超时设成 0.5 秒，并关掉 SDK 自带的重试）")
start = time.time()
try:
    OpenAI(api_key=os.environ["LLM_API_KEY"], base_url=BASE_URL, timeout=0.5, max_retries=0).chat.completions.create(
        model=MODEL, messages=[{"role": "user", "content": "写一篇 800 字的文章"}], extra_body=NO_THINKING)
except openai.APITimeoutError as e:
    print(f"{type(e).__name__}，用了 {time.time() - start:.1f} 秒")

运行结果：

== 1. 密钥错误
AuthenticationError，状态码 401
== 2. 模型名写错
BadRequestError，状态码 400，Error code: 400 - {'error': {'message': 'The supported API model names are deepseek-flash, deepseek-
== 3. 超时（故意把超时设成 0.5 秒，并关掉 SDK 自带的重试）
APITimeoutError，用了 0.7 秒

openai SDK 把不同的错误变成了不同的异常类，你可以按类型分别处理。常见的有这些：

规律很简单：错在你的，重试没用；错在对方或者网络的，可以重试。模型名写错的报错信息很友好，直接列出了支持的模型名，照着改就行。

另外，余额不足时，DeepSeek 会返回 402 状态码，在 SDK 里是一个通用的 APIStatusError。这种错误也不该重试，应该通知你去充值。

SDK 已经帮你做了什么

很多人不知道，openai SDK 默认就会自动重试。我查了当前版本（3.14）的源码：

• 默认 max_retries=2，也就是失败后最多再试 2 次。
• 会重试的情况：超时、连接失败，以及状态码 408、409、429 和所有 500 以上的错误。服务器在响应头里明确要求重试或者不重试时，也会照办。
• 两次重试之间会等待，从 0.5 秒开始逐次加倍，最长 8 秒。
• 默认超时是 600 秒，其中建立连接最多等 5 秒。

也就是说，就算你什么都不写，偶尔的 429 和 500 也会被 SDK 默默地重试掉。这是好事，但有两个问题需要注意。

600 秒的超时太长了。用户不会在网页上等 10 分钟。一般的对话场景，建议把超时设成 30～60 秒；开着思考做复杂任务时可以长一点。在创建客户端时传 timeout=60 就行。

你看不到重试发生了。SDK 的重试是静默的，你不知道一次调用其实试了三次、等了十几秒。排查"为什么这么慢"时，这个信息很重要。

一个能放进项目里的调用函数

所以我通常关掉 SDK 自带的重试，自己写一个调用函数，把重试、限流、记账放在一起：

client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=BASE_URL,
    timeout=60,  # 单次请求最多等 60 秒
    max_retries=0,  # 关掉 SDK 自带的重试，由下面的函数统一处理，方便记录
)
RETRYABLE = (openai.RateLimitError, openai.APITimeoutError, openai.APIConnectionError, openai.InternalServerError)
limiter = threading.Semaphore(5)  # 同一时刻最多 5 个请求在路上
log_lock = threading.Lock()
LOG = Path("calls.jsonl")


def call_llm(messages, max_attempts=4, **kwargs):
    for attempt in range(1, max_attempts + 1):
        start = time.time()
        try:
            with limiter:
                response = client.chat.completions.create(model=MODEL, messages=messages, **kwargs)
        except RETRYABLE as e:
            if attempt == max_attempts:
                raise
            # 指数退避：1 秒、2 秒、4 秒……再加一点随机，避免大家同时重试
            wait = 2 ** (attempt - 1) + random.random()
            print(f"  第 {attempt} 次失败（{type(e).__name__}），{wait:.1f} 秒后重试")
            time.sleep(wait)
            continue
        record = {
            "time": time.strftime("%Y-%m-%d %H:%M:%S"),
            "model": response.model,
            "seconds": round(time.time() - start, 2),
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
            "cost_usd": round(cost_usd(response.usage, MODEL), 6),
            "attempts": attempt,
        }
        with log_lock:
            with LOG.open("a") as f:
                f.write(json.dumps(record, ensure_ascii=False) + "\n")
        return response

逐块解释。

只重试该重试的错误。RETRYABLE 列出了四种可以重试的异常。401、400 这类错误不在里面，会直接抛出去，让你第一时间发现。

指数退避加随机抖动。第一次失败等 1 秒多，第二次 2 秒多，第三次 4 秒多。等待时间逐次加倍，是为了给服务器恢复的时间：如果对方正在过载，你每秒重试一次只会让情况更糟。加上 random.random() 的随机部分，是为了避免很多请求在同一时刻一起失败、又在同一时刻一起重试，形成一波又一波的冲击。

设重试次数的上限。4 次都失败就放弃，把异常抛给调用方。永远不要写无限重试。

限制并发。threading.Semaphore(5) 保证同一时刻最多有 5 个请求在进行。服务商对并发数和每分钟请求数都有限制（截至 2026 年 9 月，DeepSeek 文档写的是 deepseek-flash 并发上限 2500，deepseek-v4-pro 是 500），自己先控制住，比等着被 429 更好。更重要的是，它能防止代码里一个 bug 瞬间发出几千个请求。

每次调用记一笔账。时间、模型、耗时、输入输出词元、费用、试了几次，写成一行 JSON 追加到 calls.jsonl。费用用的是第 01 模块第 4 课的 cost_usd。有了这个日志，就能回答"这个功能一天花多少钱"、"哪些请求特别慢"、"重试有多频繁"这些问题。06 模块第 3 课会在此基础上做完整的日志和监控。

试一下：并发 20 个请求

questions = [f"用一句话解释 HTTP 状态码 {code} 的含义。" for code in [200, 201, 204, 301, 302, 304, 400, 401, 403, 404,
                                                                  405, 408, 409, 418, 429, 500, 502, 503, 504, 505]]
with ThreadPoolExecutor(20) as pool:
    answers = list(pool.map(lambda q: call_llm([{"role": "user", "content": q}], extra_body=NO_THINKING), questions))

20 个线程同时调用，但 limiter 只允许 5 个同时进行：

== 4. 并发 20 个请求，最多同时 5 个，每次调用记账
20 个请求用了 3.5 秒
第一条回答： HTTP 状态码 200 表示服务器成功处理了请求，并正常返回了所请求的资源。
日志共 20 条，总花费 0.000655 美元，第一条：{'time': '2026-09-14 21:35:48', 'model': 'deepseek-flash', 'seconds': 0.62, 'prompt_tokens': 16, 'completion_tokens': 19, 'cost_usd': 2.8e-05, 'attempts': 1}

20 个请求用了 3.5 秒，一个一个串行调用大约要 12 秒。这次运行没有遇到需要重试的错误，所以每条记录的 attempts 都是 1。

花钱的几道保险

重试和并发控制防的是意外，下面这些防的是账单：

• 给每次调用设 max_tokens。防止模型没完没了地输出。开着思考时要给得宽裕一些，第 00 模块第 3 课讲过思考会占用这个额度。
• 给循环设上限。工具调用的循环、重试的循环，都要有最大次数。第 3 课的工具调用循环最多 5 轮，就是这个道理。
• 在平台上设余额提醒。DeepSeek 是预付费的，余额用完就停，天然有一个上限。但最好还是只充够一段时间用的钱，并留意余额。
• 看日志。每天看一眼 calls.jsonl 的总花费，异常的增长一眼就能看出来。

练习

1. 把 limiter 的并发数改成 1 和 20，分别跑一次，比较 20 个请求的总耗时。
2. 写一个模拟的故障：定义一个函数，前两次调用时抛出 openai.APITimeoutError，第三次才真正调用模型。把它换进 call_llm 里，看看重试和等待的输出，以及日志里的 attempts。（提示：openai.APITimeoutError(request=...) 需要一个 request 参数，可以用 httpx.Request("POST", "https://example.com")。）
3. 写一个小脚本，读取 calls.jsonl，打印总调用次数、总花费、平均耗时、最慢的 3 次调用。

自测

1. 遇到 401 错误，要不要重试？遇到 429 呢？

401 是密钥错误，重试多少次结果都一样，应该直接报错，让人去检查密钥。429 是请求太频繁被限流，是暂时的，应该等一会儿再重试，并且等待时间要逐次加长。

2. 为什么重试之间的等待时间要逐次加倍，还要加一点随机数？

加倍是为了给服务器恢复的时间，对方过载时频繁重试只会让情况更糟。随机数是为了让很多同时失败的请求错开重试的时间，避免它们在同一时刻再次涌向服务器。

3. 什么都不设置时，openai SDK 会自动重试吗？

会。当前版本默认 max_retries=2，遇到超时、连接失败，以及 408、409、429 和 500 以上的状态码时，会等待一小段时间后自动重试，最多 2 次。默认超时是 600 秒，对大多数交互场景来说太长，建议自己设置 timeout。