1. 429 限流到底卡在哪从一次批量任务说起大模型 API 返回 429 Too Many Requests本质是服务端在告诉你「这一秒的请求量超过了我给你分配的配额」。它和普通 HTTP 接口的限流不太一样普通接口扩容机器就能扛而大模型每个请求都要占 GPU 显存厂商的卡是有限的所以限流阈值通常压得很低。你如果只是偶尔调一次对话可能几个月都碰不到 429但只要开始跑批量任务、做 Agent 循环、或者多个同事共用一个 Key429 几乎必然出现。我先把常见的限流粒度摆出来你对照自己的场景看卡在哪一层限流维度典型阈值触发场景RPM每分钟请求数60 次/分钟循环调用、Agent 多轮工具调用TPM每分钟 Token 数100K token/分钟长文本摘要、RAG 塞大段上下文并发连接数3-5 并发批量任务同时起飞单日总量100 万 token/天持续跑的数据处理流水线最容易踩的坑是你以为自己 QPS 不高但每个请求的 prompt 有 8000 tokenTPM 先爆了返回的照样是 429。还有一种情况是多个服务共用一个 Key各自看自己流量都不大加起来就超了。这篇要解决的问题很具体在 TaoToken 统一通道下怎么把 429 处理成「可预期的重试」而不是「雪崩」。TaoToken 是一个统一的大模型 API 接入通道你用它一个 Key 就能调多家模型平台侧对上游厂商的限流做了池化调度。但注意平台帮你扛了一部分不代表客户端可以裸调——你自己的重试策略写错了照样会把本地服务拖死。下面从接入配置开始一步步把重试封装做出来。适合谁看正在写批量调用脚本的开发者、做 Agent 应用的工程师、以及被 429 打断过任务流的人。你不需要很深的分布式背景跟着配置和代码走就行。2. TaoToken 前置准备Base URL、Key 与 Model ID 三件套在写重试逻辑之前先把接入信息固定下来。TaoToken 的调用方式和 OpenAI 兼容接口一致所以任何支持自定义 Base URL 的 SDK 或工具都能接。核心就三样东西Base URL、API Key、Model ID。Base URL 用https://taotoken.net/api注意这里不带任何查询参数SDK 里填的就是这个根地址具体路径由 SDK 自己拼/v1/chat/completions。API Key 在控制台的 API Keys 页面创建地址是https://taotoken.net/console/api-keys。Model ID 就是你实际要调的模型名比如deepseek-chat、qwen-plus这类具体以文档里的模型列表为准文档入口在https://taotoken.net/doc。如果你用的是 Claude Code 这类命令行工具配置方式略有不同需要设置ANTHROPIC_BASE_URL和ANTHROPIC_AUTH_TOKEN两个环境变量Base URL 同样指向 TaoToken 的 API 地址。这类工具的接入细节在官方文档里有专门章节建议先照着配一遍再回来加重试逻辑。这里要强调一个工程习惯把 Base URL、Key、Model ID 抽成配置不要硬编码在业务代码里。原因很简单——重试策略要针对不同模型调参数如果模型名散落在十几个文件里你改一次降级链就要全局搜索。我一般用一个config.json或者环境变量组来管理{ base_url: https://taotoken.net/api, api_key: sk-你的Key, primary_model: deepseek-chat, fallback_models: [qwen-plus, glm-4], max_retries: 5, base_delay: 1.0, max_delay: 60.0, max_concurrency: 5 }这个配置文件后面会被重试封装直接读取。fallback_models是降级链当主模型连续 429 或超时就按顺序切到备用模型。max_concurrency是并发上限配合信号量使用。关于 Key 的安全不要把 Key 提交到 Git 仓库用环境变量或者本地配置文件加.gitignore。TaoToken 控制台可以创建多个 Key如果你有多个服务建议一个服务一个 Key这样某个服务跑飞了不会影响其他服务也方便在控制台看用量。配置好之后先用最简单的 curl 验证一下通道是通的别等写完重试逻辑才发现 Key 填错了curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer sk-你的Key \ -H Content-Type: application/json \ -d { model: deepseek-chat, messages: [{role: user, content: 你好}] }返回里有choices数组就说明通了。这一步花两分钟能省掉后面半小时的排查。3. 可复制的重试封装指数退避 抖动 并发控制现在进入核心部分。我把重试封装拆成三层令牌桶限流器事前、信号量并发控制事中、指数退避加抖动事后。三层叠起来才稳只做重试是不够的。先看令牌桶。它的作用是让请求在发出之前就被削峰尽量不触发 429。桶以恒定速率补充令牌请求必须拿到令牌才能发。这样即使业务侧突然来一波流量也会被平滑成厂商能接受的速率import time import threading class TokenBucket: def __init__(self, rate_per_sec: float, capacity: int None): self.rate rate_per_sec self.capacity capacity or max(1, int(rate_per_sec * 2)) self.tokens float(self.capacity) self.last_refill time.time() self.lock threading.Lock() def acquire(self, timeout: float 30.0) - bool: deadline time.time() timeout while time.time() deadline: with self.lock: now time.time() elapsed now - self.last_refill self.tokens min(self.capacity, self.tokens elapsed * self.rate) self.last_refill now if self.tokens 1: self.tokens - 1 return True time.sleep(0.05) return Falserate_per_sec怎么定如果你知道模型的 RPM 是 60那就是每秒 1 个请求填 1.0。但实际建议填得比理论值低一点比如 0.8留出余量给突发。capacity是桶容量决定了能允许多大的瞬时爆发默认给速率的两倍。第二层是并发控制。令牌桶管的是「速率」信号量管的是「同时在飞的请求数」。这两个不一样速率限制每秒发几个并发限制同一时刻有几个没返回。批量任务里如果同时起 50 个协程即使令牌桶限了速率已经发出去的请求也可能把连接池占满。用信号量把并发压到 5 以内import asyncio class ConcurrencyGuard: def __init__(self, max_concurrency: int 5): self.semaphore asyncio.Semaphore(max_concurrency) async def run(self, coro): async with self.semaphore: return await coro第三层是重试本身。关键点有三个指数退避让等待时间翻倍、随机抖动打散不同请求的重试时刻、尊重服务端返回的Retry-After头。抖动是必须的否则 1000 个请求会在同一秒集体醒来形成脉冲把刚缓过来的服务端再打回去import random import asyncio import httpx async def call_with_retry( client: httpx.AsyncClient, config: dict, messages: list, model: str None, ): model model or config[primary_model] max_retries config[max_retries] base_delay config[base_delay] max_delay config[max_delay] for attempt in range(max_retries): try: resp await client.post( f{config[base_url]}/v1/chat/completions, headers{ Authorization: fBearer {config[api_key]}, Content-Type: application/json, }, json{model: model, messages: messages}, timeout120.0, ) if resp.status_code 200: return resp.json() if resp.status_code 429: retry_after resp.headers.get(Retry-After) if retry_after: wait float(retry_after) else: wait base_delay * (2 ** attempt) wait wait random.uniform(0, wait) wait min(wait, max_delay) print(f[429] 第 {attempt1} 次重试等待 {wait:.2f}s) await asyncio.sleep(wait) continue resp.raise_for_status() except (httpx.TimeoutException, httpx.ConnectError) as e: wait min(base_delay * (2 ** attempt) random.uniform(0, 1), max_delay) print(f[{type(e).__name__}] 第 {attempt1} 次重试等待 {wait:.2f}s) await asyncio.sleep(wait) raise RuntimeError(f模型 {model} 重试 {max_retries} 次仍失败)注意wait wait random.uniform(0, wait)这一行这是「全抖动」策略等待时间在[wait, 2*wait]之间随机。也有用「等抖动」的即wait/2 random.uniform(0, wait/2)效果类似。选哪种看团队习惯关键是必须有随机成分。把三层串起来再加一个降级链就是完整的客户端class RobustClient: def __init__(self, config: dict): self.config config self.bucket TokenBucket(rate_per_sec0.8) self.guard ConcurrencyGuard(config[max_concurrency]) self.client httpx.AsyncClient() async def chat(self, messages: list): models [self.config[primary_model]] self.config[fallback_models] last_err None for model in models: try: if not self.bucket.acquire(): await asyncio.sleep(1) return await self.guard.run( call_with_retry(self.client, self.config, messages, model) ) except Exception as e: last_err e print(f模型 {model} 失败尝试降级) raise last_err这段代码可以直接复制去改。fallback_models里放功能等价的模型比如主模型是deepseek-chat备用放qwen-plus。对大多数通用对话场景切换模型用户几乎无感知。4. 验证请求构造 429 观察退避间隔与成功率写完封装不能就算完得验证它真的按预期工作。最直接的办法是构造 429 响应观察退避间隔是否符合指数增长、抖动是否生效、最终成功率是多少。第一种验证方式把令牌桶速率调到极低比如rate_per_sec0.1然后一次性发 20 个请求。理论上大部分请求会排队等待少数会触发 429 并进入重试。你在日志里应该看到等待时间大致是 1s、2s、4s、8s 这样的序列且每次都有随机偏移。如果看到所有请求都在同一秒重试说明抖动没生效回去检查random.uniform那行。第二种方式用 mock 服务模拟 429。起一个本地 HTTP 服务前三次请求返回 429 并带Retry-After: 2第四次返回 200。然后让你的客户端去打这个 mock观察它是否尊重了Retry-After、是否在第四次成功拿到结果from fastapi import FastAPI, Response import uvicorn app FastAPI() counter {n: 0} app.post(/v1/chat/completions) def mock_chat(): counter[n] 1 if counter[n] 3: return Response( content{error:rate limited}, status_code429, headers{Retry-After: 2}, ) return {choices: [{message: {role: assistant, content: ok}}]}跑起来后把客户端的base_url指向http://127.0.0.1:8000发一个请求。预期日志是三次 429、每次等 2 秒、第四次成功。这个测试能验证Retry-After解析、退避逻辑、成功返回三条路径。第三种方式真实压测。用 TaoToken 的通道把并发开到 20跑 200 个请求统计 429 率和最终成功率。我实测下来在令牌桶限速 0.8/秒、并发 5 的配置下200 个请求的 429 触发次数通常在个位数最终成功率 100%。如果你不加令牌桶直接裸调429 率可能到 20% 以上而且重试会放大流量。验证时重点看三个指标429 触发次数、平均重试等待时间、最终成功率。前两个反映退避策略是否合理第三个反映整体是否可用。如果最终成功率不是 100%说明max_retries太小或者降级链没配好。还有一个容易忽略的点验证幂等性。重试的前提是请求可安全重放。对话补全接口本身是幂等的同样的输入得到同样的输出不产生副作用所以重试安全。但如果你调的是带副作用的接口比如创建任务、扣费操作重试前必须确认幂等键。TaoToken 的对话接口不涉及这类问题但你自己封装其他接口时要留意。5. 常见报错排查401、local proxy failed、reading choices、OAuth重试逻辑跑起来后你可能会碰到一些不是 429 但同样卡人的报错。这里按真实遇到的频率排一下。401 Unauthorized最常见的原因是 Key 没填对或者带了多余空格。检查Authorization头是不是Bearer sk-xxx格式中间一个空格。还有一种情况是 Key 被禁用或额度耗尽去控制台 API Keys 页面看状态。如果你用的是环境变量确认变量名没拼错比如ANTHROPIC_AUTH_TOKEN和ANTHROPIC_API_KEY是两个不同的变量Claude Code 用的是前者。local proxy failed / connection refused这个报错通常出现在你本地配了代理但代理没起来或者 Base URL 写错了。先确认base_url是https://taotoken.net/api不要多加/v1也不要少写。然后检查本地网络能不能直接访问这个域名用 curl 试一下。如果公司网络有出口限制联系网络管理员放行。reading choices 相关报错典型信息是KeyError: choices或者list index out of range。这说明返回的 JSON 里没有choices字段通常是上游返回了错误结构但状态码是 200或者你解析的层级不对。打印完整响应体看看正常结构是{choices: [{message: {...}}]}。如果返回的是{error: {...}}说明请求本身有问题比如模型名写错了。OAuth 相关报错如果你用 Claude Code 或类似工具可能会碰到 OAuth token 过期或未授权。这类工具首次使用需要走一次授权流程之后 token 会缓存。如果报 OAuth 错误先检查配置文件里的 token 是否还在有效期必要时重新授权。注意不要把 OAuth token 和 API Key 搞混它们是两套认证体系。模型名不存在报错信息通常是model not found或invalid model。去文档页确认模型 ID 的准确拼写大小写敏感。降级链里的备用模型也要确认都存在否则主模型失败后降级也会失败。排查通用思路先看 HTTP 状态码4xx 是请求问题Key、参数、模型名5xx 是服务端问题重试即可。然后看响应体里的error字段通常有具体原因。最后看你的配置Base URL、Key、Model ID 三件套逐个核对。把日志打全包括请求头脱敏后、请求体、响应状态码、响应体排查效率会高很多。6. 长期跑批量任务把重试策略固化成配置如果你只是偶尔调几次上面的封装够用了。但如果你要长期跑批量任务、做 Agent 应用建议把重试策略固化成配置并且考虑用 Coding Plan 这类长期方案来管理调用配额。配置化的好处是不同任务用不同策略。比如离线批处理可以容忍长等待max_retries设大一点、base_delay设长一点在线对话要求低延迟max_retries设小、快速降级到备用模型。把这些参数放在配置文件里按任务加载不同的 profile。对于 Agent 类应用重试策略还要和工具调用循环配合。Agent 一轮对话可能触发多次模型调用如果每次都独立重试总延迟会累积。更好的做法是在 Agent 层面做整体超时控制单次模型调用失败快速降级不要让一个工具调用卡住整个循环。最后给一个实用建议把 429 率作为监控指标。在客户端记录每次 429 的时间戳和模型名定期统计。如果某个模型的 429 率持续偏高说明你的流量模式需要调整要么降速、要么换模型、要么升级配额。监控比事后排查有用得多。调用地址汇总一下模型对话在https://taotoken.net/apiCoding Plan 适合长期编码和 Agent 场景控制台在https://taotoken.net/console/api-keys管理 Key接入文档在https://taotoken.net/doc查模型列表和参数。把这几个地址存好配置和排查都用得上。