拓冰网站定制 
当一批脚本在几秒内把大量格式异常的文件推给上游接口时网关看到的不是正常业务流量而是典型的探测流量请求数密集、payload 小而畸形、失败后立刻重试最终把并发压力全部转嫁到限流层。这类场景最近被反复讨论某智能体批量向 Hugging Face 账户投递格式异常文件的公开报道本质上就是自动化脚本 高并发 无退避重试的放大版。如果你也在维护类似的探测、巡检或回放任务真正要解决的不是要不要并发而是并发上去之后 Token 怎么记账、限流怎么接住。TaoTokenhttps://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency 在这一层的价值很直接它把请求统一收口到 OpenAI 兼容的 Base URL让并发控制、重试退避和用量统计都能落在你自己的代码里而不是散落在十几个不同的供应商 SDK 中。下面这篇内容不做行业评论只讲可复制的接入与排障并发配置片段怎么写、限流日志长什么样、用量对照表怎么回填以及 Claude Code、Codex、CC Switch 三件套如何改到同一条链路上。1. 探测式并发的三个特征决定了它一定会先撞限流先把问题定义清楚。探测类任务的流量特征和普通聊天应用完全不同第一请求密度高但单请求极短。一个文件探测脚本可能在 10 秒内发出几百个请求每个请求的 prompt 只有几十个 token输出也只要几个 token。单请求成本低但请求数RPM会瞬间打满。第二失败重试是同步放大。很多脚本在except里直接continue或者无条件重试一旦上游返回 429重试线程和原始线程同时存在实际并发是名义并发的 2 到 3 倍。第三payload 形态异常。格式异常的文件、超长文件名、非 UTF-8 内容这些在网关侧看起来高度可疑容易触发风控而不是单纯的配额限流。风控限流通常不给Retry-After直接掐连接这比 429 更难排查。所以并发控制必须在客户端做两层一层是并发闸门同时最多几个在飞另一层是速率闸门每分钟最多几个请求、每分钟最多消耗多少 token。只做第一层遇到慢响应仍然会堆积只做第二层遇到突发仍然会瞬间超速。在动手改脚本之前先确认你的 Key 和 Base URL 是同一套。从 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency 的控制台拿到 Key 后把请求地址统一指向https://taotoken.net/api后续所有并发参数、重试策略、日志字段都在这个统一入口上生效不需要为不同模型改不同的 SDK 初始化逻辑。2. 最小可跑通Key、Base URL 与一次请求验证在改并发之前先用一条命令确认链路是通的。不要跳过这一步很多限流其实是 Key 写错或者路径拼错导致的 401/404被误判成 429。# 环境变量先固定下来脚本和 CLI 工具共用同一套 export TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODEL你控制台里可用的模型 ID # 非流式最小请求用来确认鉴权和路径 curl -sS $TAOTOKEN_BASE_URL/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: $TAOTOKEN_MODEL, messages: [{role: user, content: 只回复 ok}], max_tokens: 8, stream: false }返回体里会带usage字段包含prompt_tokens、completion_tokens、total_tokens。这三个数字是你后面做用量对照的唯一依据务必在脚本里把它们记下来而不是只看请求是否成功。流式请求单独测一次因为流式场景下限流表现不一样连接建立成功后才开始计费中途被限流时你可能已经拿到部分输出。可以用curl -N -sS $TAOTOKEN_BASE_URL/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: $TAOTOKEN_MODEL, messages: [{role: user, content: 数到三}], max_tokens: 32, stream: true }如果这一步报 401检查 Key 是否带上了多余的引号或换行报 404检查 Base URL 后面有没有多写/v1之外的路径。确认无误后再进入并发改造。3. 并发配置片段信号量 速率闸门 退避重试下面是可直接粘贴运行的异步脚本骨架。它做了四件事限制同时在飞的请求数、限制单位时间内的请求数、按Retry-After退避、把每次请求的 token 用量打成结构化日志。import asyncio import json import os import random import time from collections import deque import httpx BASE_URL os.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api) API_KEY os.environ[TAOTOKEN_API_KEY] MODEL os.environ.get(TAOTOKEN_MODEL, 你的模型 ID) # ---- 并发与限流参数按实测调整 ---- MAX_CONCURRENCY 8 # 同时在飞的请求数上限 MAX_RPM 120 # 每分钟请求数上限先设保守值 MAX_RETRIES 5 # 429/5xx 最大重试次数 BACKOFF_BASE 1.5 # 退避基数秒 BACKOFF_CAP 30.0 # 退避上限秒 REQUEST_TIMEOUT 60.0 # 单请求超时秒 _sem asyncio.Semaphore(MAX_CONCURRENCY) _call_times deque() # 记录请求时间戳用于 RPM 闸门 def log_event(event: str, **fields) - None: 统一的结构化日志方便后面用 jq 聚合。 record {ts: round(time.time(), 3), event: event, **fields} print(json.dumps(record, ensure_asciiFalse), flushTrue) async def rpm_gate() - None: 滚动窗口限速保证最近 60 秒内的请求数不超过 MAX_RPM。 while True: now time.monotonic() while _call_times and now - _call_times[0] 60: _call_times.popleft() if len(_call_times) MAX_RPM: _call_times.append(now) return sleep_for 60 - (now - _call_times[0]) 0.05 await asyncio.sleep(max(sleep_for, 0.05)) def parse_retry_after(resp: httpx.Response) - float: 优先读 Retry-After没有就按退避基数指数增长并加抖动。 raw resp.headers.get(retry-after) if raw: try: return min(float(raw), BACKOFF_CAP) except ValueError: pass return 0.0 async def call_model(client: httpx.AsyncClient, trace_id: str, prompt: str) - dict: payload { model: MODEL, messages: [{role: user, content: prompt}], max_tokens: 64, stream: False, } headers { Authorization: fBearer {API_KEY}, Content-Type: application/json, } for attempt in range(1, MAX_RETRIES 1): await rpm_gate() async with _sem: started time.perf_counter() try: resp await client.post( /v1/chat/completions, jsonpayload, headersheaders, timeoutREQUEST_TIMEOUT, ) latency_ms round((time.perf_counter() - started) * 1000, 1) if resp.status_code 200: body resp.json() usage body.get(usage, {}) or {} log_event( request_done, trace_idtrace_id, attemptattempt, status200, latency_mslatency_ms, prompt_tokensusage.get(prompt_tokens), completion_tokensusage.get(completion_tokens), total_tokensusage.get(total_tokens), concurrencyMAX_CONCURRENCY, ) return body if resp.status_code in (429, 500, 502, 503, 504): wait parse_retry_after(resp) if wait 0: wait min(BACKOFF_BASE ** attempt, BACKOFF_CAP) wait random.uniform(0, 0.4) # 抖动避免重试风暴同频 log_event( retry_scheduled, trace_idtrace_id, attemptattempt, statusresp.status_code, wait_sround(wait, 2), concurrencyMAX_CONCURRENCY, ) await asyncio.sleep(wait) continue # 4xx非 429通常是请求本身有问题重试无意义 log_event( request_failed, trace_idtrace_id, attemptattempt, statusresp.status_code, bodyresp.text[:300], ) return {error: resp.status_code} except (httpx.TimeoutException, httpx.TransportError) as exc: wait min(BACKOFF_BASE ** attempt, BACKOFF_CAP) random.uniform(0, 0.4) log_event( transport_error, trace_idtrace_id, attemptattempt, errortype(exc).__name__, wait_sround(wait, 2), ) await asyncio.sleep(wait) log_event(give_up, trace_idtrace_id, attemptsMAX_RETRIES) return {error: exhausted} async def main() - None: prompts [f探测样本 {i}请只回复 ok for i in range(200)] limits httpx.Limits(max_connectionsMAX_CONCURRENCY * 2, max_keepalive_connectionsMAX_CONCURRENCY) async with httpx.AsyncClient(base_urlBASE_URL, limitslimits) as client: tasks [ call_model(client, trace_idfprobe-{i:04d}, promptp) for i, p in enumerate(prompts) ] await asyncio.gather(*tasks) if __name__ __main__: asyncio.run(main())几个参数的意义必须搞清楚否则调参会变成玄学MAX_CONCURRENCY控制同时在飞直接决定峰值压力。先设 4 到 8观察日志里的 429 数量再往上抬。MAX_RPM控制每分钟请求数和并发数不是一回事。低并发 高 RPM 依然会触发请求数限流。MAX_RETRIES配合退避本质上是在补给速度和压力放大之间取平衡。重试次数设太高遇到持续限流时会把并发放大数倍。抖动的存在是为了防止所有协程在同一毫秒重试形成新的尖峰。另外要注意httpx.Limits里的连接池上限要和信号量匹配。如果连接池只有 8 个连接而信号量是 16多出来的协程会在连接池排队你的限流日志里会看到大量latency_ms异常高的成功请求容易被误判成上游变慢。4. 限流日志字段设计、采集与判定并发改造做完之后真正的产出物是日志。没有日志你无法回答到底是 RPM 超了还是 TPM 超了重试放大了几倍这两个问题。建议的日志字段固定为下面这一组不要随意加自由文本{ts:1730000000.123,event:request_done,trace_id:probe-0007,attempt:1,status:200,latency_ms:812.4,prompt_tokens:128,completion_tokens:64,total_tokens:192,concurrency:8} {ts:1730000001.451,event:retry_scheduled,trace_id:probe-0011,attempt:2,status:429,wait_s:3.12,concurrency:8} {ts:1730000002.907,event:retry_scheduled,trace_id:probe-0019,attempt:1,status:503,wait_s:1.94,concurrency:8} {ts:1730000004.220,event:request_failed,trace_id:probe-0023,attempt:1,status:400,body:invalid request payload}判定逻辑要写死status429且带retry-after属于配额类限流降并发或降 RPM 后通常能恢复。status429但不带retry-after更可能是风控类拦截此时降并发没用要检查 payload 形态和请求特征是否过于异常。status503/504上游过载退避重试有效但要在日志里和 429 分开统计。status400请求本身有问题重试是浪费配额直接修 payload。跑一轮之后把日志导出成本地文件用命令聚合。这些命令都在你本地执行不涉及任何远端数据库# 按事件类型统计数量 jq -r .event run.log | sort | uniq -c | sort -rn # 统计每个并发档位下的总 token 和请求数 jq -r select(.eventrequest_done) | [.concurrency, .total_tokens] | tsv run.log \ | awk {sum[$1]$2; cnt[$1]} END {for (c in sum) printf 并发%s 请求%d 总tokens%d 均值%.1f\n, c, cnt[c], sum[c], sum[c]/cnt[c]} # 统计 429 占比 jq -r select(.eventretry_scheduled and .status429) | .trace_id run.log | wc -l # 提取所有退避等待时长判断是否出现了长尾 jq -r select(.eventretry_scheduled) | .wait_s run.log | sort -n | tail -5如果你在脚本里同时记录了latency_ms还可以算出 P95jq -r select(.eventrequest_done) | .latency_ms run.log \ | sort -n \ | awk {a[NR]$1} END {printf P50%.0fms P95%.0fms 样本%d\n, a[int(NR*0.5)], a[int(NR*0.95)], NR}有了这些聚合结果限流日志就不是一堆散乱的行而是一份能直接支撑调参的证据。5. 用量对照把并发档位跑成一张可回填的表用量对照的关键不是去追某个最优并发数而是找到你自己任务形态下的拐点从哪一档开始429 开始明显上升而吞吐不再增长。测试方法要固定三件事同一份 payload 集合、同一模型、同一时间段长度。然后逐档提升MAX_CONCURRENCY例如 4、8、16、32每档跑 5 到 10 分钟把日志分别落盘。下表是字段模板数值请用你自己的日志回填不要照抄任何外部数字并发档位成功请求429 次数429 占比prompt tokenscompletion tokens总 tokensP95 延迟(ms)有效吞吐(请求/分钟)结论4待回填待回填待回填待回填待回填待回填待回填待回填基准档8待回填待回填待回填待回填待回填待回填待回填待回填观察 429 是否抬头16待回填待回填待回填待回填待回填待回填待回填待回填大概率出现拐点32待回填待回填待回填待回填待回填待回填待回填待回填验证是否已无收益几个读表要点看有效吞吐而不是看名义并发。如果并发从 16 提到 32成功请求数没变但 429 翻倍说明已经进入重试吃掉配额的阶段此时应该降回 16而不是继续加机器。看 token 均值而不是总量。如果某一档的总tokens/成功请求数明显变大说明输出长度在膨胀可能是重试请求被计费、或者模型返回变长这时候成本上升和并发无关。看 P95 而不是平均值。平均值掩盖长尾。限流一旦发生P95 会先动平均值可能还很好看。区分 prompt 与 completion。prompt tokens 主要来自你的输入探测脚本里的文件内容、元数据completion tokens 来自模型输出。如果 prompt 占比极高优化方向是压缩输入而不是降并发。把这四档跑完你就能给出一个明确结论在这个任务形态下安全并发档位是多少对应的每分钟 token 预算是多少。这个结论比任何经验值都可靠。6. 客户端侧三件套Claude Code、Codex 与 CC Switch 的配置除了自己写脚本很多人的探测任务是通过 Claude Code 或 Codex 这类 CLI 触发的。如果你希望这些工具也走同一条入口需要分别改配置注意两边的环境变量体系完全不同不要混用。Claude Code改~/.claude/settings.json使用ANTHROPIC_*前缀。{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: 你的模型 ID, ANTHROPIC_SMALL_FAST_MODEL: 你的小模型 ID, CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: 1 } }写完保存重开终端让配置生效。如果仍然走旧链路检查是否有 shell 里 export 的ANTHROPIC_BASE_URL覆盖了文件配置——环境变量优先级通常高于配置文件。Codex改~/.codex/config.toml用model_providers段不要用ANTHROPIC_*。model 你的模型 ID model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chat对应的环境变量是TAOTOKEN_API_KEY值填YOUR_API_KEY。Codex 读取的是env_key指定的那个变量名把ANTHROPIC_AUTH_TOKEN塞进来是不会生效的这是最容易踩的坑。CC Switch 三件套保持三处一致只切不混。三件套指的是Claude Code 配置文件、Codex 配置文件、以及 CC Switch 自身维护的供应商条目。切换时最容易出问题的是三处没对齐——Claude Code 已经从旧供应商切到新入口Codex 那侧还留着旧配置结果同一个终端窗口里两个工具打到两个不同的后端日志对不上账。一个稳妥的做法是在切换完成后立刻做一次一致性校验# 本地校验确认 CLI 实际使用的入口与 Key 是否指向同一套 grep -R base_url\|BASE_URL ~/.claude/settings.json ~/.codex/config.toml 2/dev/null echo env: ${TAOTOKEN_API_KEY:0:6}... ${ANTHROPIC_BASE_URL:-未设置}只要输出的入口都是https://taotoken.net/api且 Key 前缀一致就说明三件套已经对齐。关于工具链的完整配置说明可以参考 Claude Code 文档https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency 里面有环境变量和配置文件的对应关系比反复试错省时间。7. 常见报错与排查路径对照把排障路径固化下来能省掉大量来回猜测。现象可能原因处理方式401 UnauthorizedKey 带引号、换行或用了别家的 Key重新从控制台复制确认Authorization: Bearer格式正确404 Not FoundBase URL 拼错或路径重复/v1确认 Base 为https://taotoken.net/api请求路径为/v1/chat/completions429 且带 Retry-After请求配额类限流降低MAX_RPM或按 Retry-After 退避不要硬重试429 且不带 Retry-After请求特征异常触发风控检查 payload 是否畸形、是否高频重放同一内容大量成功但延迟 P95 飙升连接池上限低于信号量把httpx.Limits.max_connections提到并发数的 2 倍流式中途断开超时设置过短或上游抖动提高REQUEST_TIMEOUT并对流式单独做重试分支token 用量与请求数不成比例重试请求也被计费在日志里按trace_id聚合区分首次请求与重试请求Codex 配置不生效误用了ANTHROPIC_*改回env_key指定的TAOTOKEN_API_KEY排查顺序建议固定为先确认鉴权与路径401/404再确认限流类型429 有无 Retry-After最后才动并发参数。反过来做的话你会在一个本来就配错的链路上反复调参越调越乱。8. 把并发控制变成可复用的工程习惯回到最初那个场景格式异常的文件被批量并发投递暴露出来的从来不是模型能力问题而是工程侧的三个缺口——没有并发闸门、没有退避策略、没有用量记账。补上这三个缺口之后同一份探测脚本的稳定性会有明显变化。落地顺序建议是这样先把 Key 和 Base URL 统一到https://taotoken.net/api用一次 curl 确认链路通。把脚本里的重试逻辑改成读 Retry-After 指数退避 抖动去掉无条件重试。加上结构化日志保证每次请求都能落到trace_id、status、usage三个字段。跑 4/8/16/32 四档并发回填上面的对照表找到自己的拐点。把 CLI 工具Claude Code、Codex、CC Switch也对齐到同一条链路避免多套配置互相污染。做完这五步你手里就有了一份可复现的并发配置、一份可聚合的限流日志和一张有数据的用量对照表。以后再遇到类似的批量任务直接复用这套骨架即可。如果你还没开始建议先从最小验证走起在模型对话页发一条请求确认 Key 可用再看 Coding Plan 了解配额与并发档位的匹配关系接着在控制台创建独立的 API Key 用于脚本不要和交互式工具共用一个 Key方便分开记账最后按 Claude Code 文档把 CLI 侧配置补齐。模型对话体验https://taotoken.net/models/detail/chat?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrencyCoding Plan 说明https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency创建 API Keyhttps://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrencyClaude Code 配置文档https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency官方入口与最新说明以 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency 为准Base URL 固定为https://taotoken.net/apiKey 统一用YOUR_API_KEY占位替换即可。 当 Hugging Face 探测任务并发时，TaoToken 做限流
当一批脚本在几秒内把大量格式异常的文件推给上游接口时网关看到的不是正常业务流量而是典型的探测流量请求数密集、payload 小而畸形、失败后立刻重试最终把并发压力全部转嫁到限流层。这类场景最近被反复讨论某智能体批量向 Hugging Face 账户投递格式异常文件的公开报道本质上就是自动化脚本 高并发 无退避重试的放大版。如果你也在维护类似的探测、巡检或回放任务真正要解决的不是要不要并发而是并发上去之后 Token 怎么记账、限流怎么接住。TaoTokenhttps://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency 在这一层的价值很直接它把请求统一收口到 OpenAI 兼容的 Base URL让并发控制、重试退避和用量统计都能落在你自己的代码里而不是散落在十几个不同的供应商 SDK 中。下面这篇内容不做行业评论只讲可复制的接入与排障并发配置片段怎么写、限流日志长什么样、用量对照表怎么回填以及 Claude Code、Codex、CC Switch 三件套如何改到同一条链路上。1. 探测式并发的三个特征决定了它一定会先撞限流先把问题定义清楚。探测类任务的流量特征和普通聊天应用完全不同第一请求密度高但单请求极短。一个文件探测脚本可能在 10 秒内发出几百个请求每个请求的 prompt 只有几十个 token输出也只要几个 token。单请求成本低但请求数RPM会瞬间打满。第二失败重试是同步放大。很多脚本在except里直接continue或者无条件重试一旦上游返回 429重试线程和原始线程同时存在实际并发是名义并发的 2 到 3 倍。第三payload 形态异常。格式异常的文件、超长文件名、非 UTF-8 内容这些在网关侧看起来高度可疑容易触发风控而不是单纯的配额限流。风控限流通常不给Retry-After直接掐连接这比 429 更难排查。所以并发控制必须在客户端做两层一层是并发闸门同时最多几个在飞另一层是速率闸门每分钟最多几个请求、每分钟最多消耗多少 token。只做第一层遇到慢响应仍然会堆积只做第二层遇到突发仍然会瞬间超速。在动手改脚本之前先确认你的 Key 和 Base URL 是同一套。从 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency 的控制台拿到 Key 后把请求地址统一指向https://taotoken.net/api后续所有并发参数、重试策略、日志字段都在这个统一入口上生效不需要为不同模型改不同的 SDK 初始化逻辑。2. 最小可跑通Key、Base URL 与一次请求验证在改并发之前先用一条命令确认链路是通的。不要跳过这一步很多限流其实是 Key 写错或者路径拼错导致的 401/404被误判成 429。# 环境变量先固定下来脚本和 CLI 工具共用同一套 export TAOTOKEN_API_KEYYOUR_API_KEY export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODEL你控制台里可用的模型 ID # 非流式最小请求用来确认鉴权和路径 curl -sS $TAOTOKEN_BASE_URL/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: $TAOTOKEN_MODEL, messages: [{role: user, content: 只回复 ok}], max_tokens: 8, stream: false }返回体里会带usage字段包含prompt_tokens、completion_tokens、total_tokens。这三个数字是你后面做用量对照的唯一依据务必在脚本里把它们记下来而不是只看请求是否成功。流式请求单独测一次因为流式场景下限流表现不一样连接建立成功后才开始计费中途被限流时你可能已经拿到部分输出。可以用curl -N -sS $TAOTOKEN_BASE_URL/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: $TAOTOKEN_MODEL, messages: [{role: user, content: 数到三}], max_tokens: 32, stream: true }如果这一步报 401检查 Key 是否带上了多余的引号或换行报 404检查 Base URL 后面有没有多写/v1之外的路径。确认无误后再进入并发改造。3. 并发配置片段信号量 速率闸门 退避重试下面是可直接粘贴运行的异步脚本骨架。它做了四件事限制同时在飞的请求数、限制单位时间内的请求数、按Retry-After退避、把每次请求的 token 用量打成结构化日志。import asyncio import json import os import random import time from collections import deque import httpx BASE_URL os.environ.get(TAOTOKEN_BASE_URL, https://taotoken.net/api) API_KEY os.environ[TAOTOKEN_API_KEY] MODEL os.environ.get(TAOTOKEN_MODEL, 你的模型 ID) # ---- 并发与限流参数按实测调整 ---- MAX_CONCURRENCY 8 # 同时在飞的请求数上限 MAX_RPM 120 # 每分钟请求数上限先设保守值 MAX_RETRIES 5 # 429/5xx 最大重试次数 BACKOFF_BASE 1.5 # 退避基数秒 BACKOFF_CAP 30.0 # 退避上限秒 REQUEST_TIMEOUT 60.0 # 单请求超时秒 _sem asyncio.Semaphore(MAX_CONCURRENCY) _call_times deque() # 记录请求时间戳用于 RPM 闸门 def log_event(event: str, **fields) - None: 统一的结构化日志方便后面用 jq 聚合。 record {ts: round(time.time(), 3), event: event, **fields} print(json.dumps(record, ensure_asciiFalse), flushTrue) async def rpm_gate() - None: 滚动窗口限速保证最近 60 秒内的请求数不超过 MAX_RPM。 while True: now time.monotonic() while _call_times and now - _call_times[0] 60: _call_times.popleft() if len(_call_times) MAX_RPM: _call_times.append(now) return sleep_for 60 - (now - _call_times[0]) 0.05 await asyncio.sleep(max(sleep_for, 0.05)) def parse_retry_after(resp: httpx.Response) - float: 优先读 Retry-After没有就按退避基数指数增长并加抖动。 raw resp.headers.get(retry-after) if raw: try: return min(float(raw), BACKOFF_CAP) except ValueError: pass return 0.0 async def call_model(client: httpx.AsyncClient, trace_id: str, prompt: str) - dict: payload { model: MODEL, messages: [{role: user, content: prompt}], max_tokens: 64, stream: False, } headers { Authorization: fBearer {API_KEY}, Content-Type: application/json, } for attempt in range(1, MAX_RETRIES 1): await rpm_gate() async with _sem: started time.perf_counter() try: resp await client.post( /v1/chat/completions, jsonpayload, headersheaders, timeoutREQUEST_TIMEOUT, ) latency_ms round((time.perf_counter() - started) * 1000, 1) if resp.status_code 200: body resp.json() usage body.get(usage, {}) or {} log_event( request_done, trace_idtrace_id, attemptattempt, status200, latency_mslatency_ms, prompt_tokensusage.get(prompt_tokens), completion_tokensusage.get(completion_tokens), total_tokensusage.get(total_tokens), concurrencyMAX_CONCURRENCY, ) return body if resp.status_code in (429, 500, 502, 503, 504): wait parse_retry_after(resp) if wait 0: wait min(BACKOFF_BASE ** attempt, BACKOFF_CAP) wait random.uniform(0, 0.4) # 抖动避免重试风暴同频 log_event( retry_scheduled, trace_idtrace_id, attemptattempt, statusresp.status_code, wait_sround(wait, 2), concurrencyMAX_CONCURRENCY, ) await asyncio.sleep(wait) continue # 4xx非 429通常是请求本身有问题重试无意义 log_event( request_failed, trace_idtrace_id, attemptattempt, statusresp.status_code, bodyresp.text[:300], ) return {error: resp.status_code} except (httpx.TimeoutException, httpx.TransportError) as exc: wait min(BACKOFF_BASE ** attempt, BACKOFF_CAP) random.uniform(0, 0.4) log_event( transport_error, trace_idtrace_id, attemptattempt, errortype(exc).__name__, wait_sround(wait, 2), ) await asyncio.sleep(wait) log_event(give_up, trace_idtrace_id, attemptsMAX_RETRIES) return {error: exhausted} async def main() - None: prompts [f探测样本 {i}请只回复 ok for i in range(200)] limits httpx.Limits(max_connectionsMAX_CONCURRENCY * 2, max_keepalive_connectionsMAX_CONCURRENCY) async with httpx.AsyncClient(base_urlBASE_URL, limitslimits) as client: tasks [ call_model(client, trace_idfprobe-{i:04d}, promptp) for i, p in enumerate(prompts) ] await asyncio.gather(*tasks) if __name__ __main__: asyncio.run(main())几个参数的意义必须搞清楚否则调参会变成玄学MAX_CONCURRENCY控制同时在飞直接决定峰值压力。先设 4 到 8观察日志里的 429 数量再往上抬。MAX_RPM控制每分钟请求数和并发数不是一回事。低并发 高 RPM 依然会触发请求数限流。MAX_RETRIES配合退避本质上是在补给速度和压力放大之间取平衡。重试次数设太高遇到持续限流时会把并发放大数倍。抖动的存在是为了防止所有协程在同一毫秒重试形成新的尖峰。另外要注意httpx.Limits里的连接池上限要和信号量匹配。如果连接池只有 8 个连接而信号量是 16多出来的协程会在连接池排队你的限流日志里会看到大量latency_ms异常高的成功请求容易被误判成上游变慢。4. 限流日志字段设计、采集与判定并发改造做完之后真正的产出物是日志。没有日志你无法回答到底是 RPM 超了还是 TPM 超了重试放大了几倍这两个问题。建议的日志字段固定为下面这一组不要随意加自由文本{ts:1730000000.123,event:request_done,trace_id:probe-0007,attempt:1,status:200,latency_ms:812.4,prompt_tokens:128,completion_tokens:64,total_tokens:192,concurrency:8} {ts:1730000001.451,event:retry_scheduled,trace_id:probe-0011,attempt:2,status:429,wait_s:3.12,concurrency:8} {ts:1730000002.907,event:retry_scheduled,trace_id:probe-0019,attempt:1,status:503,wait_s:1.94,concurrency:8} {ts:1730000004.220,event:request_failed,trace_id:probe-0023,attempt:1,status:400,body:invalid request payload}判定逻辑要写死status429且带retry-after属于配额类限流降并发或降 RPM 后通常能恢复。status429但不带retry-after更可能是风控类拦截此时降并发没用要检查 payload 形态和请求特征是否过于异常。status503/504上游过载退避重试有效但要在日志里和 429 分开统计。status400请求本身有问题重试是浪费配额直接修 payload。跑一轮之后把日志导出成本地文件用命令聚合。这些命令都在你本地执行不涉及任何远端数据库# 按事件类型统计数量 jq -r .event run.log | sort | uniq -c | sort -rn # 统计每个并发档位下的总 token 和请求数 jq -r select(.eventrequest_done) | [.concurrency, .total_tokens] | tsv run.log \ | awk {sum[$1]$2; cnt[$1]} END {for (c in sum) printf 并发%s 请求%d 总tokens%d 均值%.1f\n, c, cnt[c], sum[c], sum[c]/cnt[c]} # 统计 429 占比 jq -r select(.eventretry_scheduled and .status429) | .trace_id run.log | wc -l # 提取所有退避等待时长判断是否出现了长尾 jq -r select(.eventretry_scheduled) | .wait_s run.log | sort -n | tail -5如果你在脚本里同时记录了latency_ms还可以算出 P95jq -r select(.eventrequest_done) | .latency_ms run.log \ | sort -n \ | awk {a[NR]$1} END {printf P50%.0fms P95%.0fms 样本%d\n, a[int(NR*0.5)], a[int(NR*0.95)], NR}有了这些聚合结果限流日志就不是一堆散乱的行而是一份能直接支撑调参的证据。5. 用量对照把并发档位跑成一张可回填的表用量对照的关键不是去追某个最优并发数而是找到你自己任务形态下的拐点从哪一档开始429 开始明显上升而吞吐不再增长。测试方法要固定三件事同一份 payload 集合、同一模型、同一时间段长度。然后逐档提升MAX_CONCURRENCY例如 4、8、16、32每档跑 5 到 10 分钟把日志分别落盘。下表是字段模板数值请用你自己的日志回填不要照抄任何外部数字并发档位成功请求429 次数429 占比prompt tokenscompletion tokens总 tokensP95 延迟(ms)有效吞吐(请求/分钟)结论4待回填待回填待回填待回填待回填待回填待回填待回填基准档8待回填待回填待回填待回填待回填待回填待回填待回填观察 429 是否抬头16待回填待回填待回填待回填待回填待回填待回填待回填大概率出现拐点32待回填待回填待回填待回填待回填待回填待回填待回填验证是否已无收益几个读表要点看有效吞吐而不是看名义并发。如果并发从 16 提到 32成功请求数没变但 429 翻倍说明已经进入重试吃掉配额的阶段此时应该降回 16而不是继续加机器。看 token 均值而不是总量。如果某一档的总tokens/成功请求数明显变大说明输出长度在膨胀可能是重试请求被计费、或者模型返回变长这时候成本上升和并发无关。看 P95 而不是平均值。平均值掩盖长尾。限流一旦发生P95 会先动平均值可能还很好看。区分 prompt 与 completion。prompt tokens 主要来自你的输入探测脚本里的文件内容、元数据completion tokens 来自模型输出。如果 prompt 占比极高优化方向是压缩输入而不是降并发。把这四档跑完你就能给出一个明确结论在这个任务形态下安全并发档位是多少对应的每分钟 token 预算是多少。这个结论比任何经验值都可靠。6. 客户端侧三件套Claude Code、Codex 与 CC Switch 的配置除了自己写脚本很多人的探测任务是通过 Claude Code 或 Codex 这类 CLI 触发的。如果你希望这些工具也走同一条入口需要分别改配置注意两边的环境变量体系完全不同不要混用。Claude Code改~/.claude/settings.json使用ANTHROPIC_*前缀。{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: 你的模型 ID, ANTHROPIC_SMALL_FAST_MODEL: 你的小模型 ID, CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: 1 } }写完保存重开终端让配置生效。如果仍然走旧链路检查是否有 shell 里 export 的ANTHROPIC_BASE_URL覆盖了文件配置——环境变量优先级通常高于配置文件。Codex改~/.codex/config.toml用model_providers段不要用ANTHROPIC_*。model 你的模型 ID model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chat对应的环境变量是TAOTOKEN_API_KEY值填YOUR_API_KEY。Codex 读取的是env_key指定的那个变量名把ANTHROPIC_AUTH_TOKEN塞进来是不会生效的这是最容易踩的坑。CC Switch 三件套保持三处一致只切不混。三件套指的是Claude Code 配置文件、Codex 配置文件、以及 CC Switch 自身维护的供应商条目。切换时最容易出问题的是三处没对齐——Claude Code 已经从旧供应商切到新入口Codex 那侧还留着旧配置结果同一个终端窗口里两个工具打到两个不同的后端日志对不上账。一个稳妥的做法是在切换完成后立刻做一次一致性校验# 本地校验确认 CLI 实际使用的入口与 Key 是否指向同一套 grep -R base_url\|BASE_URL ~/.claude/settings.json ~/.codex/config.toml 2/dev/null echo env: ${TAOTOKEN_API_KEY:0:6}... ${ANTHROPIC_BASE_URL:-未设置}只要输出的入口都是https://taotoken.net/api且 Key 前缀一致就说明三件套已经对齐。关于工具链的完整配置说明可以参考 Claude Code 文档https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency 里面有环境变量和配置文件的对应关系比反复试错省时间。7. 常见报错与排查路径对照把排障路径固化下来能省掉大量来回猜测。现象可能原因处理方式401 UnauthorizedKey 带引号、换行或用了别家的 Key重新从控制台复制确认Authorization: Bearer格式正确404 Not FoundBase URL 拼错或路径重复/v1确认 Base 为https://taotoken.net/api请求路径为/v1/chat/completions429 且带 Retry-After请求配额类限流降低MAX_RPM或按 Retry-After 退避不要硬重试429 且不带 Retry-After请求特征异常触发风控检查 payload 是否畸形、是否高频重放同一内容大量成功但延迟 P95 飙升连接池上限低于信号量把httpx.Limits.max_connections提到并发数的 2 倍流式中途断开超时设置过短或上游抖动提高REQUEST_TIMEOUT并对流式单独做重试分支token 用量与请求数不成比例重试请求也被计费在日志里按trace_id聚合区分首次请求与重试请求Codex 配置不生效误用了ANTHROPIC_*改回env_key指定的TAOTOKEN_API_KEY排查顺序建议固定为先确认鉴权与路径401/404再确认限流类型429 有无 Retry-After最后才动并发参数。反过来做的话你会在一个本来就配错的链路上反复调参越调越乱。8. 把并发控制变成可复用的工程习惯回到最初那个场景格式异常的文件被批量并发投递暴露出来的从来不是模型能力问题而是工程侧的三个缺口——没有并发闸门、没有退避策略、没有用量记账。补上这三个缺口之后同一份探测脚本的稳定性会有明显变化。落地顺序建议是这样先把 Key 和 Base URL 统一到https://taotoken.net/api用一次 curl 确认链路通。把脚本里的重试逻辑改成读 Retry-After 指数退避 抖动去掉无条件重试。加上结构化日志保证每次请求都能落到trace_id、status、usage三个字段。跑 4/8/16/32 四档并发回填上面的对照表找到自己的拐点。把 CLI 工具Claude Code、Codex、CC Switch也对齐到同一条链路避免多套配置互相污染。做完这五步你手里就有了一份可复现的并发配置、一份可聚合的限流日志和一张有数据的用量对照表。以后再遇到类似的批量任务直接复用这套骨架即可。如果你还没开始建议先从最小验证走起在模型对话页发一条请求确认 Key 可用再看 Coding Plan 了解配额与并发档位的匹配关系接着在控制台创建独立的 API Key 用于脚本不要和交互式工具共用一个 Key方便分开记账最后按 Claude Code 文档把 CLI 侧配置补齐。模型对话体验https://taotoken.net/models/detail/chat?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrencyCoding Plan 说明https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency创建 API Keyhttps://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrencyClaude Code 配置文档https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency官方入口与最新说明以 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contenthf_probe_concurrency 为准Base URL 固定为https://taotoken.net/apiKey 统一用YOUR_API_KEY占位替换即可。