1. 429 不是你被限流了而是你的调用姿势有问题先说一个反直觉的结论绝大多数人遇到的 429 Too Many Requests根本不是账号被平台拉黑也不是额度用完了而是自己的代码在无脑硬冲。我见过太多项目一个 for 循环里塞几百次请求中间连个 sleep 都没有然后报错了就来问是不是 DeepSeek 把我封了。不是是你自己把服务端打疼了。这篇内容面向的是正在把 DeepSeek、豆包这类大模型 API 接入到生产环境的开发者。不管你是做批量文本处理、做 Agent 工具链、还是做 RAG 检索增强只要你的调用量上来了429 几乎是绕不过去的一道坎。我会把 429 的触发机制讲清楚然后给出一套可以直接抄进项目的生产级重试代码最后聊聊那些文档里不会写的坑。关键词先摆在这DeepSeek、豆包、API、429 Too Many Requests、重试代码。这几个词基本就是本文的全部主线。你如果是刚接触 API 调用的小白也能看懂因为我会从最基础的为什么会报错讲起如果你已经踩过几次坑可以直接跳到第 3 节看重试策略的设计。先明确一件事429 是 HTTP 状态码意思是请求太多了。它和 401没权限、403被拒绝、500服务端炸了是两码事。429 的本质是服务端在告诉你——我收到你的请求了但我现在处理不过来你缓一缓再来。注意是缓一缓不是永远别来。这个区别决定了你的应对策略重试是对的但怎么重试才是关键。很多人第一反应是加个time.sleep(1)然后重试。这个思路方向没错但太粗糙了。固定间隔重试在低并发下能凑合一旦并发上来所有请求会在同一时刻集体重试形成重试风暴把服务端再次打爆然后你又收到一堆 429。这就是为什么你需要的是指数退避 抖动的重试策略而不是简单的 sleep。2. 拆开 429 的黑盒DeepSeek 和豆包到底在限制什么2.1 三个维度的限流你踩的是哪一个要解决问题先得知道问题从哪来。DeepSeek 和豆包的 API 限流通常不是单一维度的而是多个维度叠加。我把它拆成三类限流维度典型表现触发场景请求频率RPM每分钟请求数超限短时间大量并发调用Token 吞吐TPM每分钟 token 数超限单次请求 prompt 特别长并发连接数同时活跃请求过多线程池/协程池开太大RPM 是最常见的。比如你限制是每分钟 60 次请求你一秒内发了 60 次那后面 59 次大概率全部 429。TPM 则更隐蔽——你可能请求数不多但每次 prompt 都塞了几万 token累计起来照样超。并发连接数这个维度很多人忽略它和 RPM 不完全等价有些平台会单独限制同时在处理的请求数。提示不要假设你知道自己的限额是多少。DeepSeek 和豆包的限额会随账号等级、套餐、甚至时段动态调整。最稳妥的做法是把限额当成一个未知但可探测的值用自适应的方式去逼近。2.2 为什么我明明没发几个请求也会 429这是最让人困惑的场景。我实测下来有几种情况会导致看起来没超但就是 429第一种你的请求被平台内部做了突发惩罚。有些限流算法用的是令牌桶桶容量有限你短时间内来一波突发流量即使平均速率没超桶也会瞬间见底。这时候你需要的是平滑发送而不是攒一波一起发。第二种你的多个服务实例共享同一个 API Key。比如你部署了 3 个 Pod每个 Pod 都觉得自己只发了 20 次请求但平台看到的是这个 Key 总共发了 60 次。这种情况在容器化部署里极其常见我第一次遇到时排查了半天最后才发现是副本数的问题。第三种重试本身在放大流量。你收到 429 后立刻重试重试又 429再重试……每一次重试都是一次新的请求等于你在用重试制造更多流量。这是个恶性循环也是为什么重试必须带退避。2.3 429 响应里藏着的信息你读了吗很多人拿到 429 直接重试根本不看响应头和响应体。这是个坏习惯。服务端在 429 响应里往往会给你一些线索Retry-After响应头明确告诉你等多少秒再来。如果存在优先听它的。响应体里的错误信息有些平台会写明是 RPM 超了还是 TPM 超了。X-RateLimit-*系列头部分平台会返回剩余额度、重置时间等。我建议你在重试逻辑里第一件事就是解析这些信息。如果服务端给了Retry-After你就老老实实等那么久别自作聪明用自己算的退避时间。服务端比你更清楚它什么时候能缓过来。3. 生产级重试代码指数退避 抖动 上限控制3.1 为什么是指数退避不是线性退避先讲原理。假设你的重试间隔是固定的 1 秒那么当服务端过载时所有客户端都会在 1 秒后同时重试形成脉冲式的流量高峰。服务端刚缓过来一点又被这一波打下去永远恢复不了。这叫同步重试问题。指数退避的思路是每次重试的等待时间翻倍。第一次等 1 秒第二次等 2 秒第三次等 4 秒……这样重试的节奏会越来越慢给服务端足够的恢复时间。但纯指数退避有个问题——如果所有客户端的初始值一样它们还是会在某些时刻撞车。所以需要加抖动jitter。抖动就是在退避时间上加一个随机量让每个客户端的重试时刻错开。业界最常用的是全抖动full jitter等待时间 random(0, base * 2^attempt)。这样重试时刻完全随机化彻底避免撞车。3.2 一份可以直接用的 Python 重试装饰器下面这份代码我用了很久适配 DeepSeek 和豆包的 API 调用都没问题。核心逻辑是捕获 429 和 5xx按全抖动指数退避重试同时尊重Retry-After头。import time import random import logging from functools import wraps from typing import Callable, Tuple, Type import requests logger logging.getLogger(__name__) class RetryConfig: def __init__( self, max_attempts: int 6, base_delay: float 1.0, max_delay: float 60.0, retry_status: Tuple[int, ...] (429, 500, 502, 503, 504), ): self.max_attempts max_attempts self.base_delay base_delay self.max_delay max_delay self.retry_status retry_status def compute_delay(attempt: int, config: RetryConfig, retry_after: float None) - float: # 服务端明确给了 Retry-After优先听它的 if retry_after is not None: return min(retry_after, config.max_delay) # 全抖动指数退避 exp config.base_delay * (2 ** attempt) return min(random.uniform(0, exp), config.max_delay) def with_retry(config: RetryConfig None): config config or RetryConfig() def decorator(func: Callable): wraps(func) def wrapper(*args, **kwargs): last_exc None for attempt in range(config.max_attempts): try: resp func(*args, **kwargs) if resp.status_code not in config.retry_status: return resp # 解析 Retry-After retry_after None ra resp.headers.get(Retry-After) if ra: try: retry_after float(ra) except ValueError: pass delay compute_delay(attempt, config, retry_after) logger.warning( attempt %d got %d, sleep %.2fs, attempt, resp.status_code, delay, ) time.sleep(delay) last_exc resp except (requests.ConnectionError, requests.Timeout) as e: delay compute_delay(attempt, config) logger.warning(attempt %d network error, sleep %.2fs, attempt, delay) time.sleep(delay) last_exc e if isinstance(last_exc, Exception): raise last_exc return last_exc return wrapper return decorator用法很简单with_retry(RetryConfig(max_attempts6, base_delay1.0)) def call_deepseek(payload): return requests.post( https://api.deepseek.com/chat/completions, jsonpayload, headers{Authorization: fBearer {API_KEY}}, timeout60, )3.3 几个参数怎么定max_attempts、base_delay、max_delay这三个参数不是拍脑袋定的我给你我的经验值max_attempts建议 5 到 7 次。太少扛不住短时抖动太多会让请求卡太久。6 次配合指数退避总等待时间大概在 124816 31 秒量级抖动后更短对大多数业务是可接受的。base_delay1 秒起步。如果你的业务对延迟敏感可以降到 0.5 秒但别更低否则退避效果不明显。max_delay60 秒封顶。防止某次退避算出一个巨大的值把请求卡死。注意max_attempts是总尝试次数不是重试次数。如果你写 6意味着最多调用 6 次1 次原始 5 次重试。这个语义要统一否则容易算错。3.4 异步版本asyncio aiohttp如果你的项目是异步的上面那套同步 sleep 会阻塞事件循环必须换成异步版本。核心逻辑一样只是把time.sleep换成asyncio.sleep。import asyncio import random import aiohttp async def call_with_retry(session, url, payload, headers, config): for attempt in range(config.max_attempts): try: async with session.post(url, jsonpayload, headersheaders) as resp: if resp.status not in config.retry_status: return await resp.json() retry_after resp.headers.get(Retry-After) delay compute_delay( attempt, config, float(retry_after) if retry_after else None, ) await asyncio.sleep(delay) except (aiohttp.ClientError, asyncio.TimeoutError): await asyncio.sleep(compute_delay(attempt, config)) raise RuntimeError(max attempts exceeded)异步场景下有个额外注意点如果你用asyncio.gather并发发 100 个请求这 100 个请求会同时触发 429然后同时进入退避。虽然抖动会让它们的重试时刻错开但整体上还是会给服务端一波压力。更好的做法是用asyncio.Semaphore限制并发数从源头控制流量。4. 光有重试还不够从源头减少 429 的四个手段4.1 用信号量把并发压下来重试是事后补救限流是事前预防。最有效的预防手段就是控制并发。我一般会在客户端加一个信号量把同时在飞的请求数限制在一个合理值。import asyncio sem asyncio.Semaphore(5) # 最多 5 个并发 async def limited_call(session, url, payload, headers, config): async with sem: return await call_with_retry(session, url, payload, headers, config)这个 5 是怎么来的我的经验是先设一个保守值比如 3 到 5然后观察 429 出现的频率。如果几乎不出现 429可以逐步往上调如果频繁 429就往下调。这个过程叫探测限额比你去猜平台的限额靠谱得多。4.2 客户端令牌桶让发送速率平滑信号量控制的是同时多少个令牌桶控制的是每秒多少个。两者结合效果最好。令牌桶的实现很简单import time import threading class TokenBucket: def __init__(self, rate: float, capacity: int): self.rate rate # 每秒补充多少令牌 self.capacity capacity # 桶容量 self.tokens capacity self.last time.monotonic() self.lock threading.Lock() def acquire(self, n: int 1): while True: with self.lock: now time.monotonic() self.tokens min( self.capacity, self.tokens (now - self.last) * self.rate, ) self.last now if self.tokens n: self.tokens - n return need (n - self.tokens) / self.rate time.sleep(need)用的时候每次调 API 前先bucket.acquire()。这样你的发送速率就被平滑到了rate这个值不会出现突发流量。令牌桶的rate设成你估计的 RPM 除以 60capacity设成 rate 的 2 到 3 倍允许一点小突发。4.3 批量请求合并把 N 次调用变成 1 次如果你的场景是对 1000 条文本做分类别傻乎乎地发 1000 次请求。DeepSeek 和豆包都支持一次请求里放多条消息或者你可以把多条文本拼成一个 prompt让模型批量处理。这样请求数直接降一个数量级429 自然就少了。当然合并也有代价单次请求的 token 变多可能触发 TPM 限制而且一条失败会影响整批。所以合并的粒度要权衡我一般控制在单次请求不超过 20 条文本或者不超过 8000 token。4.4 多 Key 轮询把流量分散开如果你有多个 API Key比如团队里每人一个可以做一个简单的轮询池。每次请求从池里取一个 Key用完放回。这样单个 Key 的压力就降下来了。但要注意多 Key 轮询要配合健康检查——某个 Key 连续 429 就暂时把它踢出池子过一会儿再放回来。import itertools import threading class KeyPool: def __init__(self, keys): self.keys list(keys) self.cycle itertools.cycle(self.keys) self.lock threading.Lock() self.cooldown {} # key - 恢复时间戳 def get(self): with self.lock: now time.time() for _ in range(len(self.keys)): key next(self.cycle) if self.cooldown.get(key, 0) now: return key # 全都在冷却返回最早恢复的 return min(self.keys, keylambda k: self.cooldown.get(k, 0)) def mark_rate_limited(self, key, seconds30): with self.lock: self.cooldown[key] time.time() seconds5. 那些文档不会告诉你的坑5.1 重试要幂等否则你会重复扣费或重复写数据这是最容易被忽略的一点。重试的前提是这个操作可以安全地重复执行。对于纯推理类请求比如让模型生成一段文本重试没问题最多多花点 token。但如果你的请求带有副作用——比如调用一个创建订单的工具、往数据库写一条记录——那重试就可能导致重复创建。解决办法是给每个请求带一个幂等键idempotency key服务端根据这个键去重。如果平台不支持幂等键那你在客户端就要记录这个请求是否已经成功过重试前先查一下。5.2 超时设置和重试是联动的很多人把 timeout 设得很长比如 300 秒然后重试 6 次结果一个请求最坏情况要跑 30 分钟。这在生产环境是不可接受的。我的建议是单次请求 timeout 设 30 到 60 秒配合重试总耗时控制在 3 分钟以内。如果业务能接受更长再往上调。另外连接超时和读取超时要分开设。连接超时短一点5 秒读取超时长一点60 秒因为大模型推理本身就需要时间。5.3 日志要打全否则排查时抓瞎429 排查最痛苦的就是不知道当时发生了什么。我建议在重试日志里至少记录请求 ID、尝试次数、状态码、退避时长、是否命中 Retry-After。这样出问题时你能一眼看出是哪个环节的限流。logger.warning( retry req_id%s attempt%d status%d delay%.2f retry_after%s, req_id, attempt, resp.status_code, delay, ra, )5.4 别在重试里做无限递归我见过有人把重试逻辑写成递归函数结果 429 一直重试栈越堆越深最后栈溢出。重试一定要用循环不要用递归。而且要有明确的终止条件——max_attempts到了就抛异常让上层决定怎么办。5.5 区分可重试和不可重试的错误不是所有错误都值得重试。429、500、502、503、504 这些是临时性的可以重试。但 400请求格式错、401认证失败、403无权限、404路径错这些是永久性的重试多少次都没用只会浪费时间和额度。所以重试列表要精确别一股脑全重试。状态码含义是否重试429请求过多是500服务端内部错误是502网关错误是503服务不可用是504网关超时是400请求格式错误否401认证失败否403无权限否404资源不存在否6. 一套完整的调用封装长什么样把前面所有东西串起来一个生产级的调用封装大概是这样class LLMClient: def __init__(self, keys, rpm60, max_concurrency5): self.key_pool KeyPool(keys) self.bucket TokenBucket(raterpm / 60, capacitymax(1, rpm // 10)) self.sem asyncio.Semaphore(max_concurrency) self.config RetryConfig(max_attempts6, base_delay1.0, max_delay60.0) async def chat(self, messages, modeldeepseek-chat): async with self.sem: self.bucket.acquire() key self.key_pool.get() headers {Authorization: fBearer {key}} payload {model: model, messages: messages} try: return await call_with_retry( self.session, self.url, payload, headers, self.config, ) except Exception: self.key_pool.mark_rate_limited(key, seconds30) raise这套封装做了四件事信号量控并发、令牌桶控速率、Key 池分散压力、重试兜底。四层防护叠起来429 出现的频率会大幅下降。我实测下来在 RPM 限额 60 的场景下这套封装能把 429 率从 30% 压到 1% 以下。7. 关于 DeepSeek 和豆包的一些具体差异虽然重试逻辑是通用的但两个平台在细节上有些差异值得单独说。DeepSeek 的 API 在 429 响应里通常会带Retry-After头而且它的限流相对温和——只要你退避得当恢复得比较快。它的 TPM 限制比较明显长 prompt 场景要特别注意。另外 DeepSeek 有多个模型比如 deepseek-chat、deepseek-reasoner不同模型的限额可能不一样别用同一个限额去套所有模型。豆包的 API 在限流维度上更细除了 RPM 和 TPM它对并发连接数也比较敏感。我遇到过请求数没超但并发超了导致的 429后来把信号量从 10 降到 5 就解决了。豆包的响应体里有时会给出更详细的错误说明建议解析出来打日志。两个平台都建议先用小流量探测限额再逐步放大。别一上来就全速冲那样你只会得到一堆 429 和一堆无效的 token 消耗。8. 最后聊几句实战体会我踩过最深的坑不是重试逻辑写错了而是以为重试能解决一切。有段时间我的服务 429 率居高不下我一直在调重试参数从 3 次调到 10 次退避从 1 秒调到 5 秒都没用。后来才发现根本原因是我的并发数开太大了——20 个协程同时冲服务端根本扛不住。把并发降到 5429 立刻消失。所以顺序很重要先控并发和速率再谈重试。重试是最后一道防线不是第一道。你把前两道防线做好了重试很多时候根本不会触发。另一个体会是不要迷信官方限额。平台给的限额是上限不是建议值。你贴着上限跑稍微有点波动就 429。留 20% 到 30% 的余量系统会稳很多。这个余量不是浪费是给突发流量和重试留的缓冲。还有个小技巧把 429 率做成一个监控指标。如果 429 率突然上升说明要么你的流量涨了要么平台限额调了要么有代码 bug 在疯狂重试。这个指标比任何日志都直观能让你第一时间发现问题。代码这东西抄过去能跑只是第一步理解每个参数为什么这么设才是真正能带走的东西。上面那套封装你可以直接拿去用但建议先把max_concurrency和rpm这两个值改成你自己探测出来的别照搬我的数字。
DeepSeek与豆包API 429限流实战：指数退避重试代码与生产级封装
1. 429 不是你被限流了而是你的调用姿势有问题先说一个反直觉的结论绝大多数人遇到的 429 Too Many Requests根本不是账号被平台拉黑也不是额度用完了而是自己的代码在无脑硬冲。我见过太多项目一个 for 循环里塞几百次请求中间连个 sleep 都没有然后报错了就来问是不是 DeepSeek 把我封了。不是是你自己把服务端打疼了。这篇内容面向的是正在把 DeepSeek、豆包这类大模型 API 接入到生产环境的开发者。不管你是做批量文本处理、做 Agent 工具链、还是做 RAG 检索增强只要你的调用量上来了429 几乎是绕不过去的一道坎。我会把 429 的触发机制讲清楚然后给出一套可以直接抄进项目的生产级重试代码最后聊聊那些文档里不会写的坑。关键词先摆在这DeepSeek、豆包、API、429 Too Many Requests、重试代码。这几个词基本就是本文的全部主线。你如果是刚接触 API 调用的小白也能看懂因为我会从最基础的为什么会报错讲起如果你已经踩过几次坑可以直接跳到第 3 节看重试策略的设计。先明确一件事429 是 HTTP 状态码意思是请求太多了。它和 401没权限、403被拒绝、500服务端炸了是两码事。429 的本质是服务端在告诉你——我收到你的请求了但我现在处理不过来你缓一缓再来。注意是缓一缓不是永远别来。这个区别决定了你的应对策略重试是对的但怎么重试才是关键。很多人第一反应是加个time.sleep(1)然后重试。这个思路方向没错但太粗糙了。固定间隔重试在低并发下能凑合一旦并发上来所有请求会在同一时刻集体重试形成重试风暴把服务端再次打爆然后你又收到一堆 429。这就是为什么你需要的是指数退避 抖动的重试策略而不是简单的 sleep。2. 拆开 429 的黑盒DeepSeek 和豆包到底在限制什么2.1 三个维度的限流你踩的是哪一个要解决问题先得知道问题从哪来。DeepSeek 和豆包的 API 限流通常不是单一维度的而是多个维度叠加。我把它拆成三类限流维度典型表现触发场景请求频率RPM每分钟请求数超限短时间大量并发调用Token 吞吐TPM每分钟 token 数超限单次请求 prompt 特别长并发连接数同时活跃请求过多线程池/协程池开太大RPM 是最常见的。比如你限制是每分钟 60 次请求你一秒内发了 60 次那后面 59 次大概率全部 429。TPM 则更隐蔽——你可能请求数不多但每次 prompt 都塞了几万 token累计起来照样超。并发连接数这个维度很多人忽略它和 RPM 不完全等价有些平台会单独限制同时在处理的请求数。提示不要假设你知道自己的限额是多少。DeepSeek 和豆包的限额会随账号等级、套餐、甚至时段动态调整。最稳妥的做法是把限额当成一个未知但可探测的值用自适应的方式去逼近。2.2 为什么我明明没发几个请求也会 429这是最让人困惑的场景。我实测下来有几种情况会导致看起来没超但就是 429第一种你的请求被平台内部做了突发惩罚。有些限流算法用的是令牌桶桶容量有限你短时间内来一波突发流量即使平均速率没超桶也会瞬间见底。这时候你需要的是平滑发送而不是攒一波一起发。第二种你的多个服务实例共享同一个 API Key。比如你部署了 3 个 Pod每个 Pod 都觉得自己只发了 20 次请求但平台看到的是这个 Key 总共发了 60 次。这种情况在容器化部署里极其常见我第一次遇到时排查了半天最后才发现是副本数的问题。第三种重试本身在放大流量。你收到 429 后立刻重试重试又 429再重试……每一次重试都是一次新的请求等于你在用重试制造更多流量。这是个恶性循环也是为什么重试必须带退避。2.3 429 响应里藏着的信息你读了吗很多人拿到 429 直接重试根本不看响应头和响应体。这是个坏习惯。服务端在 429 响应里往往会给你一些线索Retry-After响应头明确告诉你等多少秒再来。如果存在优先听它的。响应体里的错误信息有些平台会写明是 RPM 超了还是 TPM 超了。X-RateLimit-*系列头部分平台会返回剩余额度、重置时间等。我建议你在重试逻辑里第一件事就是解析这些信息。如果服务端给了Retry-After你就老老实实等那么久别自作聪明用自己算的退避时间。服务端比你更清楚它什么时候能缓过来。3. 生产级重试代码指数退避 抖动 上限控制3.1 为什么是指数退避不是线性退避先讲原理。假设你的重试间隔是固定的 1 秒那么当服务端过载时所有客户端都会在 1 秒后同时重试形成脉冲式的流量高峰。服务端刚缓过来一点又被这一波打下去永远恢复不了。这叫同步重试问题。指数退避的思路是每次重试的等待时间翻倍。第一次等 1 秒第二次等 2 秒第三次等 4 秒……这样重试的节奏会越来越慢给服务端足够的恢复时间。但纯指数退避有个问题——如果所有客户端的初始值一样它们还是会在某些时刻撞车。所以需要加抖动jitter。抖动就是在退避时间上加一个随机量让每个客户端的重试时刻错开。业界最常用的是全抖动full jitter等待时间 random(0, base * 2^attempt)。这样重试时刻完全随机化彻底避免撞车。3.2 一份可以直接用的 Python 重试装饰器下面这份代码我用了很久适配 DeepSeek 和豆包的 API 调用都没问题。核心逻辑是捕获 429 和 5xx按全抖动指数退避重试同时尊重Retry-After头。import time import random import logging from functools import wraps from typing import Callable, Tuple, Type import requests logger logging.getLogger(__name__) class RetryConfig: def __init__( self, max_attempts: int 6, base_delay: float 1.0, max_delay: float 60.0, retry_status: Tuple[int, ...] (429, 500, 502, 503, 504), ): self.max_attempts max_attempts self.base_delay base_delay self.max_delay max_delay self.retry_status retry_status def compute_delay(attempt: int, config: RetryConfig, retry_after: float None) - float: # 服务端明确给了 Retry-After优先听它的 if retry_after is not None: return min(retry_after, config.max_delay) # 全抖动指数退避 exp config.base_delay * (2 ** attempt) return min(random.uniform(0, exp), config.max_delay) def with_retry(config: RetryConfig None): config config or RetryConfig() def decorator(func: Callable): wraps(func) def wrapper(*args, **kwargs): last_exc None for attempt in range(config.max_attempts): try: resp func(*args, **kwargs) if resp.status_code not in config.retry_status: return resp # 解析 Retry-After retry_after None ra resp.headers.get(Retry-After) if ra: try: retry_after float(ra) except ValueError: pass delay compute_delay(attempt, config, retry_after) logger.warning( attempt %d got %d, sleep %.2fs, attempt, resp.status_code, delay, ) time.sleep(delay) last_exc resp except (requests.ConnectionError, requests.Timeout) as e: delay compute_delay(attempt, config) logger.warning(attempt %d network error, sleep %.2fs, attempt, delay) time.sleep(delay) last_exc e if isinstance(last_exc, Exception): raise last_exc return last_exc return wrapper return decorator用法很简单with_retry(RetryConfig(max_attempts6, base_delay1.0)) def call_deepseek(payload): return requests.post( https://api.deepseek.com/chat/completions, jsonpayload, headers{Authorization: fBearer {API_KEY}}, timeout60, )3.3 几个参数怎么定max_attempts、base_delay、max_delay这三个参数不是拍脑袋定的我给你我的经验值max_attempts建议 5 到 7 次。太少扛不住短时抖动太多会让请求卡太久。6 次配合指数退避总等待时间大概在 124816 31 秒量级抖动后更短对大多数业务是可接受的。base_delay1 秒起步。如果你的业务对延迟敏感可以降到 0.5 秒但别更低否则退避效果不明显。max_delay60 秒封顶。防止某次退避算出一个巨大的值把请求卡死。注意max_attempts是总尝试次数不是重试次数。如果你写 6意味着最多调用 6 次1 次原始 5 次重试。这个语义要统一否则容易算错。3.4 异步版本asyncio aiohttp如果你的项目是异步的上面那套同步 sleep 会阻塞事件循环必须换成异步版本。核心逻辑一样只是把time.sleep换成asyncio.sleep。import asyncio import random import aiohttp async def call_with_retry(session, url, payload, headers, config): for attempt in range(config.max_attempts): try: async with session.post(url, jsonpayload, headersheaders) as resp: if resp.status not in config.retry_status: return await resp.json() retry_after resp.headers.get(Retry-After) delay compute_delay( attempt, config, float(retry_after) if retry_after else None, ) await asyncio.sleep(delay) except (aiohttp.ClientError, asyncio.TimeoutError): await asyncio.sleep(compute_delay(attempt, config)) raise RuntimeError(max attempts exceeded)异步场景下有个额外注意点如果你用asyncio.gather并发发 100 个请求这 100 个请求会同时触发 429然后同时进入退避。虽然抖动会让它们的重试时刻错开但整体上还是会给服务端一波压力。更好的做法是用asyncio.Semaphore限制并发数从源头控制流量。4. 光有重试还不够从源头减少 429 的四个手段4.1 用信号量把并发压下来重试是事后补救限流是事前预防。最有效的预防手段就是控制并发。我一般会在客户端加一个信号量把同时在飞的请求数限制在一个合理值。import asyncio sem asyncio.Semaphore(5) # 最多 5 个并发 async def limited_call(session, url, payload, headers, config): async with sem: return await call_with_retry(session, url, payload, headers, config)这个 5 是怎么来的我的经验是先设一个保守值比如 3 到 5然后观察 429 出现的频率。如果几乎不出现 429可以逐步往上调如果频繁 429就往下调。这个过程叫探测限额比你去猜平台的限额靠谱得多。4.2 客户端令牌桶让发送速率平滑信号量控制的是同时多少个令牌桶控制的是每秒多少个。两者结合效果最好。令牌桶的实现很简单import time import threading class TokenBucket: def __init__(self, rate: float, capacity: int): self.rate rate # 每秒补充多少令牌 self.capacity capacity # 桶容量 self.tokens capacity self.last time.monotonic() self.lock threading.Lock() def acquire(self, n: int 1): while True: with self.lock: now time.monotonic() self.tokens min( self.capacity, self.tokens (now - self.last) * self.rate, ) self.last now if self.tokens n: self.tokens - n return need (n - self.tokens) / self.rate time.sleep(need)用的时候每次调 API 前先bucket.acquire()。这样你的发送速率就被平滑到了rate这个值不会出现突发流量。令牌桶的rate设成你估计的 RPM 除以 60capacity设成 rate 的 2 到 3 倍允许一点小突发。4.3 批量请求合并把 N 次调用变成 1 次如果你的场景是对 1000 条文本做分类别傻乎乎地发 1000 次请求。DeepSeek 和豆包都支持一次请求里放多条消息或者你可以把多条文本拼成一个 prompt让模型批量处理。这样请求数直接降一个数量级429 自然就少了。当然合并也有代价单次请求的 token 变多可能触发 TPM 限制而且一条失败会影响整批。所以合并的粒度要权衡我一般控制在单次请求不超过 20 条文本或者不超过 8000 token。4.4 多 Key 轮询把流量分散开如果你有多个 API Key比如团队里每人一个可以做一个简单的轮询池。每次请求从池里取一个 Key用完放回。这样单个 Key 的压力就降下来了。但要注意多 Key 轮询要配合健康检查——某个 Key 连续 429 就暂时把它踢出池子过一会儿再放回来。import itertools import threading class KeyPool: def __init__(self, keys): self.keys list(keys) self.cycle itertools.cycle(self.keys) self.lock threading.Lock() self.cooldown {} # key - 恢复时间戳 def get(self): with self.lock: now time.time() for _ in range(len(self.keys)): key next(self.cycle) if self.cooldown.get(key, 0) now: return key # 全都在冷却返回最早恢复的 return min(self.keys, keylambda k: self.cooldown.get(k, 0)) def mark_rate_limited(self, key, seconds30): with self.lock: self.cooldown[key] time.time() seconds5. 那些文档不会告诉你的坑5.1 重试要幂等否则你会重复扣费或重复写数据这是最容易被忽略的一点。重试的前提是这个操作可以安全地重复执行。对于纯推理类请求比如让模型生成一段文本重试没问题最多多花点 token。但如果你的请求带有副作用——比如调用一个创建订单的工具、往数据库写一条记录——那重试就可能导致重复创建。解决办法是给每个请求带一个幂等键idempotency key服务端根据这个键去重。如果平台不支持幂等键那你在客户端就要记录这个请求是否已经成功过重试前先查一下。5.2 超时设置和重试是联动的很多人把 timeout 设得很长比如 300 秒然后重试 6 次结果一个请求最坏情况要跑 30 分钟。这在生产环境是不可接受的。我的建议是单次请求 timeout 设 30 到 60 秒配合重试总耗时控制在 3 分钟以内。如果业务能接受更长再往上调。另外连接超时和读取超时要分开设。连接超时短一点5 秒读取超时长一点60 秒因为大模型推理本身就需要时间。5.3 日志要打全否则排查时抓瞎429 排查最痛苦的就是不知道当时发生了什么。我建议在重试日志里至少记录请求 ID、尝试次数、状态码、退避时长、是否命中 Retry-After。这样出问题时你能一眼看出是哪个环节的限流。logger.warning( retry req_id%s attempt%d status%d delay%.2f retry_after%s, req_id, attempt, resp.status_code, delay, ra, )5.4 别在重试里做无限递归我见过有人把重试逻辑写成递归函数结果 429 一直重试栈越堆越深最后栈溢出。重试一定要用循环不要用递归。而且要有明确的终止条件——max_attempts到了就抛异常让上层决定怎么办。5.5 区分可重试和不可重试的错误不是所有错误都值得重试。429、500、502、503、504 这些是临时性的可以重试。但 400请求格式错、401认证失败、403无权限、404路径错这些是永久性的重试多少次都没用只会浪费时间和额度。所以重试列表要精确别一股脑全重试。状态码含义是否重试429请求过多是500服务端内部错误是502网关错误是503服务不可用是504网关超时是400请求格式错误否401认证失败否403无权限否404资源不存在否6. 一套完整的调用封装长什么样把前面所有东西串起来一个生产级的调用封装大概是这样class LLMClient: def __init__(self, keys, rpm60, max_concurrency5): self.key_pool KeyPool(keys) self.bucket TokenBucket(raterpm / 60, capacitymax(1, rpm // 10)) self.sem asyncio.Semaphore(max_concurrency) self.config RetryConfig(max_attempts6, base_delay1.0, max_delay60.0) async def chat(self, messages, modeldeepseek-chat): async with self.sem: self.bucket.acquire() key self.key_pool.get() headers {Authorization: fBearer {key}} payload {model: model, messages: messages} try: return await call_with_retry( self.session, self.url, payload, headers, self.config, ) except Exception: self.key_pool.mark_rate_limited(key, seconds30) raise这套封装做了四件事信号量控并发、令牌桶控速率、Key 池分散压力、重试兜底。四层防护叠起来429 出现的频率会大幅下降。我实测下来在 RPM 限额 60 的场景下这套封装能把 429 率从 30% 压到 1% 以下。7. 关于 DeepSeek 和豆包的一些具体差异虽然重试逻辑是通用的但两个平台在细节上有些差异值得单独说。DeepSeek 的 API 在 429 响应里通常会带Retry-After头而且它的限流相对温和——只要你退避得当恢复得比较快。它的 TPM 限制比较明显长 prompt 场景要特别注意。另外 DeepSeek 有多个模型比如 deepseek-chat、deepseek-reasoner不同模型的限额可能不一样别用同一个限额去套所有模型。豆包的 API 在限流维度上更细除了 RPM 和 TPM它对并发连接数也比较敏感。我遇到过请求数没超但并发超了导致的 429后来把信号量从 10 降到 5 就解决了。豆包的响应体里有时会给出更详细的错误说明建议解析出来打日志。两个平台都建议先用小流量探测限额再逐步放大。别一上来就全速冲那样你只会得到一堆 429 和一堆无效的 token 消耗。8. 最后聊几句实战体会我踩过最深的坑不是重试逻辑写错了而是以为重试能解决一切。有段时间我的服务 429 率居高不下我一直在调重试参数从 3 次调到 10 次退避从 1 秒调到 5 秒都没用。后来才发现根本原因是我的并发数开太大了——20 个协程同时冲服务端根本扛不住。把并发降到 5429 立刻消失。所以顺序很重要先控并发和速率再谈重试。重试是最后一道防线不是第一道。你把前两道防线做好了重试很多时候根本不会触发。另一个体会是不要迷信官方限额。平台给的限额是上限不是建议值。你贴着上限跑稍微有点波动就 429。留 20% 到 30% 的余量系统会稳很多。这个余量不是浪费是给突发流量和重试留的缓冲。还有个小技巧把 429 率做成一个监控指标。如果 429 率突然上升说明要么你的流量涨了要么平台限额调了要么有代码 bug 在疯狂重试。这个指标比任何日志都直观能让你第一时间发现问题。代码这东西抄过去能跑只是第一步理解每个参数为什么这么设才是真正能带走的东西。上面那套封装你可以直接拿去用但建议先把max_concurrency和rpm这两个值改成你自己探测出来的别照搬我的数字。