1. 从一条 429 rate_limit_error 说起批量生成 Worker 的并发数从 4 调到 32 的那个下午日志里最先炸掉的不是我们的业务代码而是 HTTP 客户端429 rate_limit_error、529 overloaded_error、httpx.ReadTimeout混在一起刷屏同一批任务里有的 300ms 返回有的挂到 90 秒才超时。更麻烦的是这类失败会污染整个批次的产物重试一次成本翻倍最后算下来单位 Token 成本比低并发时还高。这件事的根因不是额度不够而是我们把并发当成一个可以随手拧大的旋钮却从来没有测过它在当前链路下的真实阈值。这篇内容不做行业评论只解决一个问题在 TaoToken 的 Base URL 下怎么用一份可复跑的压测脚本把 Anthropic API 的并发阈值测出来并把结果固化成生产环境的限流参数。所有配置、脚本都在本地执行读者可以逐段跟做。开始之前先把入口固定下来到 TaoToken 官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentconcurrency_intro 注册账号并拿到 API KeyBase URL 统一使用https://taotoken.net/apiKey 在本文的代码里统一用YOUR_API_KEY占位。Key 只放环境变量或本地配置文件不要硬编码进仓库。之所以现在值得认真做这件事是因为外部环境正在变Anthropic 近期向投资者释放了盈利改善的信号市场也在讨论其上市可能性。对工程团队来说这类消息的实际含义只有一个——推理侧的单位成本会被更严格地审视而并发参数直接决定了你为同样的 Token 量付出多少重试和超时代价。所以压测不是有空再做的优化项而是成本控制的前置动作。2. 压测环境基线Base URL、Key 与三套客户端的落盘配置压测结果不可复现十有八九是因为客户端侧的环境不统一。同一个人用 Claude Code 跑、另一个人用 Codex 跑、第三个人直接 curl三者的超时、重试、Header 全都不一样。所以第一步是把三套常用客户端的配置一次性落盘之后所有压测都基于同一份基线。2.1 Claude Codesettings.json ANTHROPIC_* 环境变量Claude Code 走 Anthropic 原生协议配置写在~/.claude/settings.json项目级则放.claude/settings.json。核心是三个变量ANTHROPIC_BASE_URL指向 TaoToken 的 Base URLANTHROPIC_AUTH_TOKEN放 Key模型 ID 按控制台当前可用的列表填写。{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: claude-sonnet-4-5, ANTHROPIC_SMALL_FAST_MODEL: claude-haiku-4-5, CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: 1 } }ANTHROPIC_MODEL与ANTHROPIC_SMALL_FAST_MODEL的具体 ID 请以模型对话页展示的为准不同账号可见的模型可能不同。写完配置后新开一个终端窗口让 Claude Code 重新读取环境避免旧进程缓存了旧的 Base URL。2.2 Codexconfig.toml不要复用 ANTHROPIC_*Codex 走的是 OpenAI 兼容协议配置文件是~/.codex/config.toml。这里最容易踩的坑是把ANTHROPIC_*系列变量套过来——Codex 不认这些变量。它需要的是model_provider段里声明一个自定义供应商Base URL 在https://taotoken.net/api后面补/v1走 OpenAI 兼容路径。model gpt-5-codex model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api/v1 env_key TAOTOKEN_API_KEY wire_api chat对应的 Key 用TAOTOKEN_API_KEY这个环境变量名注入和 Claude Code 的变量名分开避免两套工具互相覆盖export TAOTOKEN_API_KEYYOUR_API_KEY2.3 CC Switch 三件套配置档案、切换、验证如果团队里同时用 Claude Code 和 Codex手工改配置文件迟早会出错。CC Switch 的作用就是把这些配置做成可切换的档案实际用起来是三件套供应商档案名称 Base URL Key、工具绑定哪个档案给 Claude Code、哪个给 Codex、切换后验证切换完必须重启终端并跑一次最小请求。档案里填的内容就是前面两节里的值Claude Code 侧填https://taotoken.net/apiCodex 侧填https://taotoken.net/api/v1Key 用同一把即可。切换完成后不要凭感觉认为生效了跑一条最小请求确认curl -sS https://taotoken.net/api/v1/messages \ -H x-api-key: $TAOTOKEN_API_KEY \ -H anthropic-version: 2023-06-01 \ -H content-type: application/json \ -d {model:claude-sonnet-4-5,max_tokens:32,messages:[{role:user,content:ping}]} \ | head -c 400返回里有正常的content结构说明 Base URL 和 Key 都对如果是 401先查 Key 有没有多余空格如果是 404先查路径是/v1/messages还是/messages。这一步确认完压测的客户端基线才算建立。配置过程中如果对协议细节有疑问可以对照 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentconfig_baseline 上的接入说明核对。3. 把并发拆成四个可调旋钮很多人说并发调到 32其实说的是四件不同的事。压测前必须把它们拆开否则测出来的数字没法指导生产。第一个旋钮是并发度in-flight也就是同一时刻未完成的请求数。它决定瞬时压力峰值对应客户端里的信号量大小。第二个旋钮是发起速率RPS每秒新起多少个请求。在固定并发度下RPS 由平均延迟反推RPS ≈ 并发 / 平均延迟。如果只控制 RPS 不控制并发度遇到延迟抖动时会瞬间堆积大量未完成请求。第三个旋钮是单请求负载包括输入 Token 数、max_tokens、是否流式。输入从 200 Token 涨到 4000 Token同样的并发度对后端的压力完全不同。压测时必须固定这一项否则梯度表没有可比性。第四个旋钮是重试策略包括最大重试次数、退避基数、是否带 jitter、哪些状态码才重试。重试本质上是在放大并发一个 3 次重试的策略会让实际请求量变成原来的数倍。除了旋钮还要先定义指标。本文统一用四个指标定义为什么重要成功率2xx 响应数 / 总请求数直接决定批次产物是否完整p50 延迟中位端到端耗时反映常态体验p95 延迟95 分位端到端耗时反映尾部抖动超时阈值应由它推等效吞吐成功请求数 / 墙钟时间判断加并发是否还有收益注意一个容易被忽略的污染源提示缓存。如果同一段系统提示反复出现第二次之后可能命中缓存延迟会明显低于冷启动。压测要么固定使用同一段冷提示要么在每档并发开始前先跑一轮预热再计时并在报告里注明。4. 压测脚本可跑的并发梯度 runner下面这份脚本用httpx直连 Anthropic Messages 接口按并发梯度逐档扫描。它做三件事用信号量控制 in-flight 并发、记录每次请求的状态码与延迟、按档输出统计结果并追加写入 CSV。# bench_gradient.py import asyncio import csv import os import random import time import httpx BASE_URL https://taotoken.net/api API_KEY os.environ[TAOTOKEN_API_KEY] PATH /v1/messages MODEL claude-sonnet-4-5 LEVELS [1, 2, 4, 8, 16, 32, 64] # 并发梯度 SAMPLES_PER_LEVEL 60 # 每档样本数p95 才稳定 OUT_CSV concurrency_gradient.csv HEADERS { x-api-key: API_KEY, anthropic-version: 2023-06-01, content-type: application/json, } # 固定负载所有档位使用完全相同的 body BODY { model: MODEL, max_tokens: 256, temperature: 0, messages: [ {role: user, content: 用三句话解释什么是连接池不要使用列表。} ], } def percentile(values, q): if not values: return float(nan) xs sorted(values) k (len(xs) - 1) * q lo int(k) hi min(lo 1, len(xs) - 1) return xs[lo] (xs[hi] - xs[lo]) * (k - lo) async def single_call(client, sem, idx, records): async with sem: t0 time.perf_counter() status -1 err try: resp await client.post(PATH, jsonBODY, headersHEADERS) status resp.status_code except Exception as exc: # 超时、连接重置等 err type(exc).__name__ dt time.perf_counter() - t0 records.append( {idx: idx, status: status, latency: dt, error: err} ) async def run_level(client, level): sem asyncio.Semaphore(level) records [] t0 time.perf_counter() await asyncio.gather( *(single_call(client, sem, i, records) for i in range(SAMPLES_PER_LEVEL)) ) wall time.perf_counter() - t0 ok [r for r in records if 200 r[status] 300] lat [r[latency] for r in ok] return { concurrency: level, total: len(records), ok: len(ok), success_rate: round(len(ok) / len(records), 4), p50: round(percentile(lat, 0.50), 3), p95: round(percentile(lat, 0.95), 3), max_latency: round(max((r[latency] for r in records), default0), 3), throughput: round(len(ok) / wall, 3), wall: round(wall, 3), } async def main(): limits httpx.Limits(max_connections256, max_keepalive_connections128) timeout httpx.Timeout(connect10.0, read180.0, write30.0, pool10.0) async with httpx.AsyncClient( base_urlBASE_URL, headersHEADERS, limitslimits, timeouttimeout ) as client: rows [] for level in LEVELS: row await run_level(client, level) rows.append(row) print( f并发 {row[concurrency]:3} | 成功率 {row[success_rate]:.2%} f| p50 {row[p50]:.2f}s | p95 {row[p95]:.2f}s f| 吞吐 {row[throughput]:.2f} req/s ) await asyncio.sleep(5) # 档位之间留冷却窗口 with open(OUT_CSV, w, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnameslist(rows[0].keys())) writer.writeheader() writer.writerows(rows) print(f\n结果已写入 {OUT_CSV}) if __name__ __main__: asyncio.run(main())几个必须注意的实现细节第一limits里的max_connections要大于最大并发档位否则瓶颈会落在本地连接池而不是服务端梯度表会失真。第二读超时给足示例里是 180 秒但如果你的业务超时是 30 秒就应该按业务值设因为超时本身也是被测量的对象。第三每档之间留冷却窗口避免上一档的残留请求影响下一档的 p95。第四temperature设为 0 并固定max_tokens减少输出长度波动带来的延迟差异。跑之前先做一次连通性检查避免整轮压测全废在 401 上export TAOTOKEN_API_KEYYOUR_API_KEY python -c import os, httpx r httpx.post(https://taotoken.net/api/v1/messages, headers{x-api-key: os.environ[TAOTOKEN_API_KEY], anthropic-version: 2023-06-01, content-type: application/json}, json{model:claude-sonnet-4-5,max_tokens:16, messages:[{role:user,content:ping}]}, timeout60) print(r.status_code) print(r.text[:300]) 5. 并发梯度表怎么读找拐点而不是找最大值脚本跑完你会得到一份 CSV。下表是形态示意真实数字必须以你自己跑出来的结果为准不同模型、不同输入长度、不同时段的结果差异可能很大。并发成功率p50p95等效吞吐观察1100%1.1s1.4s0.9 req/s线性区起点延迟最稳2100%1.2s1.6s1.7 req/s吞吐接近翻倍4100%1.4s2.1s2.9 req/s仍在线性区8100%1.8s3.4s4.4 req/s吞吐增幅开始放缓1699%2.6s6.8s5.9 req/sp95 明显抬头3293%4.1s14.2s6.8 req/s拐点区重试开始出现6478%7.3s31.5s6.4 req/s吞吐不升反降读这张表有四个判据判据一吞吐增量斜率。并发从 1 翻到 2、2 翻到 4吞吐几乎等比上升这是线性区。从 8 到 16吞吐只涨了约三分之一说明已经进入收益递减。判据二成功率跌破 99% 的位置。一旦出现 429 或 529就说明有效并发已经越过了服务端能平稳承接的区间。生产阈值应该设在这个位置之下而不是恰好卡在它上面。判据三p95 与超时阈值的关系。生产环境通常把超时设在 p95 的 1.5 到 2 倍。如果并发 16 时 p95 是 6.8 秒超时就不该设成 8 秒而要设到 12 秒以上否则你会把本来能成功的慢请求主动掐断制造出一批假失败。判据四吞吐是否回落。并发 64 时吞吐低于并发 32这是典型的拥塞信号——多出来的请求没有变成产出只变成了排队和重试。把四个判据合起来本次示例的结论是稳态并发取 8 到 12突发上限取 16超过 16 必须靠队列而不是靠加并发。这个数字要写进配置而不是留在某个人的记忆里。需要额外说明的是梯度表的绝对数值受时段影响。同样的脚本在业务高峰期跑拐点可能提前一到两档。所以建议至少跑两轮一轮在你自己的业务高峰时段一轮在低谷时段取更保守的那组作为生产参数。拿到梯度表之后如果想对照不同模型的延迟表现再决定主力模型可以在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentgradient_table 上用相同 prompt 做一轮小样本对比。6. 从阈值到生产信号量、重试预算与熔断测出阈值只是第一步把它变成生产代码才是真正的收益点。以下三段代码可以直接嵌进批量生成 Worker。第一段信号量 有界队列。不要让任务无限制地gather用固定大小的信号量把 in-flight 钉死在阈值内超出部分排队等待。import asyncio MAX_INFLIGHT 10 # 来自压测结论留出安全余量 MAX_QUEUE 200 # 队列上限防止内存被堆积任务吃满 _sem asyncio.Semaphore(MAX_INFLIGHT) _queue_depth 0 async def guarded_call(coro_factory): global _queue_depth if _queue_depth MAX_QUEUE: raise RuntimeError(队列已满本次批次应降速或分片重跑) _queue_depth 1 try: async with _sem: return await coro_factory() finally: _queue_depth - 1第二段重试预算 指数退避 jitter。关键是区分值得重试和不该重试的状态码429、529、5xx 和网络异常值得重试400、401、403 重试一万次也没用只会浪费预算。import asyncio import random RETRYABLE {429, 500, 502, 503, 504, 529} MAX_ATTEMPTS 3 def backoff_delay(attempt, base0.6, cap12.0): raw min(cap, base * (2 ** attempt)) return raw * (0.5 random.random() * 0.5) # full jitter 的一半区间 async def call_with_budget(fn, *args, **kwargs): last_exc None for attempt in range(MAX_ATTEMPTS): try: resp await fn(*args, **kwargs) if resp.status_code in RETRYABLE: last_exc RuntimeError(fretryable status {resp.status_code}) else: return resp except (asyncio.TimeoutError, OSError) as exc: last_exc exc await asyncio.sleep(backoff_delay(attempt)) raise last_exc注意backoff_delay里的 jitter固定退避会让所有失败的请求在同一毫秒重新发起形成新一轮脉冲式拥塞。加了随机抖动之后重试请求会被摊平到一段时间窗口里这对稳定 p95 帮助很大。第三段滑动窗口熔断。当最近 N 次请求的成功率跌破阈值时主动暂停新请求一段时间给链路一个恢复窗口而不是继续往里灌。import time from collections import deque WINDOW 40 FAIL_RATE_TRIP 0.15 # 失败率超过 15% 触发 COOLDOWN 20.0 # 熔断冷却秒数 _events deque(maxlenWINDOW) _open_until 0.0 def record(success: bool): _events.append(success) def allowed() - bool: if time.time() _open_until: return False if len(_events) WINDOW: return True fail_rate 1 - sum(_events) / len(_events) if fail_rate FAIL_RATE_TRIP: global _open_until _open_until time.time() COOLDOWN _events.clear() return False return True这三段组合起来的逻辑是信号量管住并发上限重试预算管住失败放大熔断管住系统性劣化。三者缺一不可——只加信号量不加熔断遇到持续 429 时会一直卡在限速里慢慢磨只加重试不加信号量等于自己给自己制造并发。7. 报错对照表429 / 529 / 超时 / 401 分别怎么处理压测和线上排障时会遇到的状态码就那么几个但处理方式完全不同。下面这张表建议直接贴到值班手册里。现象典型来源处理方式不要做什么429瞬时请求超过限速降并发、按retry-after退避、启用熔断立刻加并发硬冲529服务端整体负载高拉长冷却窗口缩小批次错峰执行密集重试会加剧拥塞读超时请求已发出但迟迟无响应先确认是否流式、检查超时阈值是否过紧直接把超时砍到 5 秒连接池等待超时本地连接池上限小于并发档位调大max_connections误判为服务端问题401Key 错误或未注入检查环境变量是否被 shell 转义、有无多余空格反复重试403Key 无该模型权限换用当前账号可见的模型 ID改 Base URL 路径试探404路径写错Anthropic 协议用/v1/messagesOpenAI 兼容用/v1/chat/completions两套协议混用502网关层瞬时抖动计入重试预算观察是否成片出现单点失败就全批回滚两个排查习惯值得养成。第一记录状态码分布而不是只记成功率。成功率 95% 看着还行但如果失败全部是 429那就是并发问题如果全部是超时那就是超时阈值问题。第二把重试次数和最终成功数一起打点。重试 3 次才成功的请求成本是 3 倍当这类请求占比上升时说明你的并发已经贴着阈值在跑了应该主动往回收一档。8. 把压测变成例行公事回到最开始那条 429。它真正暴露的问题是我们把并发当成经验值而不是测量值。一旦完成一轮完整的梯度扫描你会发现该设多大这个问题有明确答案并且这个答案会随着模型版本、输入长度、时段变化而漂移所以它需要被周期性地重新测量而不是一次定终身。落地节奏建议这样安排上线前跑一轮全梯度确定稳态并发、突发上限和超时阈值之后每两周或每次切换主力模型时复跑一轮每次线上出现成片 429 或超时时把当时的并发、p95、重试率三个数拉出来对照梯度表判断是参数问题还是外部波动。如果你想直接开始路径很短先用 https://taotoken.net/models/detail/chat?utm_sourcetaotoken_aicg_blog_endutm_contentcta_chat 在模型对话页跑几条真实请求确认延迟量级再按业务用量在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcta_coding_plan 选择合适的套餐避免压测阶段反复触发限速然后在 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcta_api_keys 创建独立用途的 Key把压测 Key 和生产 Key 分开这样即使压测打爆了限速也不会波及线上 Worker最后回到 https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentcta_claude_code_doc 把 Claude Code 的接入配置核对一遍确保ANTHROPIC_BASE_URL、ANTHROPIC_AUTH_TOKEN和模型 ID 三项与压测环境完全一致。再补一句关于成本的提醒。在 Anthropic 盈利改善、市场关注其上市进程的背景下推理侧的成本纪律只会越来越重要。并发参数设得太保守浪费的是时间设得太激进浪费的是重试和超时带来的重复 Token后者往往更贵且更难被发现。压测的价值就在于把这两笔账都算清楚让每一个并发数字都有出处。 并发参数怎么设，Anthropic API 在 TaoToken 下压测取阈值
1. 从一条 429 rate_limit_error 说起批量生成 Worker 的并发数从 4 调到 32 的那个下午日志里最先炸掉的不是我们的业务代码而是 HTTP 客户端429 rate_limit_error、529 overloaded_error、httpx.ReadTimeout混在一起刷屏同一批任务里有的 300ms 返回有的挂到 90 秒才超时。更麻烦的是这类失败会污染整个批次的产物重试一次成本翻倍最后算下来单位 Token 成本比低并发时还高。这件事的根因不是额度不够而是我们把并发当成一个可以随手拧大的旋钮却从来没有测过它在当前链路下的真实阈值。这篇内容不做行业评论只解决一个问题在 TaoToken 的 Base URL 下怎么用一份可复跑的压测脚本把 Anthropic API 的并发阈值测出来并把结果固化成生产环境的限流参数。所有配置、脚本都在本地执行读者可以逐段跟做。开始之前先把入口固定下来到 TaoToken 官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentconcurrency_intro 注册账号并拿到 API KeyBase URL 统一使用https://taotoken.net/apiKey 在本文的代码里统一用YOUR_API_KEY占位。Key 只放环境变量或本地配置文件不要硬编码进仓库。之所以现在值得认真做这件事是因为外部环境正在变Anthropic 近期向投资者释放了盈利改善的信号市场也在讨论其上市可能性。对工程团队来说这类消息的实际含义只有一个——推理侧的单位成本会被更严格地审视而并发参数直接决定了你为同样的 Token 量付出多少重试和超时代价。所以压测不是有空再做的优化项而是成本控制的前置动作。2. 压测环境基线Base URL、Key 与三套客户端的落盘配置压测结果不可复现十有八九是因为客户端侧的环境不统一。同一个人用 Claude Code 跑、另一个人用 Codex 跑、第三个人直接 curl三者的超时、重试、Header 全都不一样。所以第一步是把三套常用客户端的配置一次性落盘之后所有压测都基于同一份基线。2.1 Claude Codesettings.json ANTHROPIC_* 环境变量Claude Code 走 Anthropic 原生协议配置写在~/.claude/settings.json项目级则放.claude/settings.json。核心是三个变量ANTHROPIC_BASE_URL指向 TaoToken 的 Base URLANTHROPIC_AUTH_TOKEN放 Key模型 ID 按控制台当前可用的列表填写。{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: claude-sonnet-4-5, ANTHROPIC_SMALL_FAST_MODEL: claude-haiku-4-5, CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: 1 } }ANTHROPIC_MODEL与ANTHROPIC_SMALL_FAST_MODEL的具体 ID 请以模型对话页展示的为准不同账号可见的模型可能不同。写完配置后新开一个终端窗口让 Claude Code 重新读取环境避免旧进程缓存了旧的 Base URL。2.2 Codexconfig.toml不要复用 ANTHROPIC_*Codex 走的是 OpenAI 兼容协议配置文件是~/.codex/config.toml。这里最容易踩的坑是把ANTHROPIC_*系列变量套过来——Codex 不认这些变量。它需要的是model_provider段里声明一个自定义供应商Base URL 在https://taotoken.net/api后面补/v1走 OpenAI 兼容路径。model gpt-5-codex model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api/v1 env_key TAOTOKEN_API_KEY wire_api chat对应的 Key 用TAOTOKEN_API_KEY这个环境变量名注入和 Claude Code 的变量名分开避免两套工具互相覆盖export TAOTOKEN_API_KEYYOUR_API_KEY2.3 CC Switch 三件套配置档案、切换、验证如果团队里同时用 Claude Code 和 Codex手工改配置文件迟早会出错。CC Switch 的作用就是把这些配置做成可切换的档案实际用起来是三件套供应商档案名称 Base URL Key、工具绑定哪个档案给 Claude Code、哪个给 Codex、切换后验证切换完必须重启终端并跑一次最小请求。档案里填的内容就是前面两节里的值Claude Code 侧填https://taotoken.net/apiCodex 侧填https://taotoken.net/api/v1Key 用同一把即可。切换完成后不要凭感觉认为生效了跑一条最小请求确认curl -sS https://taotoken.net/api/v1/messages \ -H x-api-key: $TAOTOKEN_API_KEY \ -H anthropic-version: 2023-06-01 \ -H content-type: application/json \ -d {model:claude-sonnet-4-5,max_tokens:32,messages:[{role:user,content:ping}]} \ | head -c 400返回里有正常的content结构说明 Base URL 和 Key 都对如果是 401先查 Key 有没有多余空格如果是 404先查路径是/v1/messages还是/messages。这一步确认完压测的客户端基线才算建立。配置过程中如果对协议细节有疑问可以对照 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentconfig_baseline 上的接入说明核对。3. 把并发拆成四个可调旋钮很多人说并发调到 32其实说的是四件不同的事。压测前必须把它们拆开否则测出来的数字没法指导生产。第一个旋钮是并发度in-flight也就是同一时刻未完成的请求数。它决定瞬时压力峰值对应客户端里的信号量大小。第二个旋钮是发起速率RPS每秒新起多少个请求。在固定并发度下RPS 由平均延迟反推RPS ≈ 并发 / 平均延迟。如果只控制 RPS 不控制并发度遇到延迟抖动时会瞬间堆积大量未完成请求。第三个旋钮是单请求负载包括输入 Token 数、max_tokens、是否流式。输入从 200 Token 涨到 4000 Token同样的并发度对后端的压力完全不同。压测时必须固定这一项否则梯度表没有可比性。第四个旋钮是重试策略包括最大重试次数、退避基数、是否带 jitter、哪些状态码才重试。重试本质上是在放大并发一个 3 次重试的策略会让实际请求量变成原来的数倍。除了旋钮还要先定义指标。本文统一用四个指标定义为什么重要成功率2xx 响应数 / 总请求数直接决定批次产物是否完整p50 延迟中位端到端耗时反映常态体验p95 延迟95 分位端到端耗时反映尾部抖动超时阈值应由它推等效吞吐成功请求数 / 墙钟时间判断加并发是否还有收益注意一个容易被忽略的污染源提示缓存。如果同一段系统提示反复出现第二次之后可能命中缓存延迟会明显低于冷启动。压测要么固定使用同一段冷提示要么在每档并发开始前先跑一轮预热再计时并在报告里注明。4. 压测脚本可跑的并发梯度 runner下面这份脚本用httpx直连 Anthropic Messages 接口按并发梯度逐档扫描。它做三件事用信号量控制 in-flight 并发、记录每次请求的状态码与延迟、按档输出统计结果并追加写入 CSV。# bench_gradient.py import asyncio import csv import os import random import time import httpx BASE_URL https://taotoken.net/api API_KEY os.environ[TAOTOKEN_API_KEY] PATH /v1/messages MODEL claude-sonnet-4-5 LEVELS [1, 2, 4, 8, 16, 32, 64] # 并发梯度 SAMPLES_PER_LEVEL 60 # 每档样本数p95 才稳定 OUT_CSV concurrency_gradient.csv HEADERS { x-api-key: API_KEY, anthropic-version: 2023-06-01, content-type: application/json, } # 固定负载所有档位使用完全相同的 body BODY { model: MODEL, max_tokens: 256, temperature: 0, messages: [ {role: user, content: 用三句话解释什么是连接池不要使用列表。} ], } def percentile(values, q): if not values: return float(nan) xs sorted(values) k (len(xs) - 1) * q lo int(k) hi min(lo 1, len(xs) - 1) return xs[lo] (xs[hi] - xs[lo]) * (k - lo) async def single_call(client, sem, idx, records): async with sem: t0 time.perf_counter() status -1 err try: resp await client.post(PATH, jsonBODY, headersHEADERS) status resp.status_code except Exception as exc: # 超时、连接重置等 err type(exc).__name__ dt time.perf_counter() - t0 records.append( {idx: idx, status: status, latency: dt, error: err} ) async def run_level(client, level): sem asyncio.Semaphore(level) records [] t0 time.perf_counter() await asyncio.gather( *(single_call(client, sem, i, records) for i in range(SAMPLES_PER_LEVEL)) ) wall time.perf_counter() - t0 ok [r for r in records if 200 r[status] 300] lat [r[latency] for r in ok] return { concurrency: level, total: len(records), ok: len(ok), success_rate: round(len(ok) / len(records), 4), p50: round(percentile(lat, 0.50), 3), p95: round(percentile(lat, 0.95), 3), max_latency: round(max((r[latency] for r in records), default0), 3), throughput: round(len(ok) / wall, 3), wall: round(wall, 3), } async def main(): limits httpx.Limits(max_connections256, max_keepalive_connections128) timeout httpx.Timeout(connect10.0, read180.0, write30.0, pool10.0) async with httpx.AsyncClient( base_urlBASE_URL, headersHEADERS, limitslimits, timeouttimeout ) as client: rows [] for level in LEVELS: row await run_level(client, level) rows.append(row) print( f并发 {row[concurrency]:3} | 成功率 {row[success_rate]:.2%} f| p50 {row[p50]:.2f}s | p95 {row[p95]:.2f}s f| 吞吐 {row[throughput]:.2f} req/s ) await asyncio.sleep(5) # 档位之间留冷却窗口 with open(OUT_CSV, w, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnameslist(rows[0].keys())) writer.writeheader() writer.writerows(rows) print(f\n结果已写入 {OUT_CSV}) if __name__ __main__: asyncio.run(main())几个必须注意的实现细节第一limits里的max_connections要大于最大并发档位否则瓶颈会落在本地连接池而不是服务端梯度表会失真。第二读超时给足示例里是 180 秒但如果你的业务超时是 30 秒就应该按业务值设因为超时本身也是被测量的对象。第三每档之间留冷却窗口避免上一档的残留请求影响下一档的 p95。第四temperature设为 0 并固定max_tokens减少输出长度波动带来的延迟差异。跑之前先做一次连通性检查避免整轮压测全废在 401 上export TAOTOKEN_API_KEYYOUR_API_KEY python -c import os, httpx r httpx.post(https://taotoken.net/api/v1/messages, headers{x-api-key: os.environ[TAOTOKEN_API_KEY], anthropic-version: 2023-06-01, content-type: application/json}, json{model:claude-sonnet-4-5,max_tokens:16, messages:[{role:user,content:ping}]}, timeout60) print(r.status_code) print(r.text[:300]) 5. 并发梯度表怎么读找拐点而不是找最大值脚本跑完你会得到一份 CSV。下表是形态示意真实数字必须以你自己跑出来的结果为准不同模型、不同输入长度、不同时段的结果差异可能很大。并发成功率p50p95等效吞吐观察1100%1.1s1.4s0.9 req/s线性区起点延迟最稳2100%1.2s1.6s1.7 req/s吞吐接近翻倍4100%1.4s2.1s2.9 req/s仍在线性区8100%1.8s3.4s4.4 req/s吞吐增幅开始放缓1699%2.6s6.8s5.9 req/sp95 明显抬头3293%4.1s14.2s6.8 req/s拐点区重试开始出现6478%7.3s31.5s6.4 req/s吞吐不升反降读这张表有四个判据判据一吞吐增量斜率。并发从 1 翻到 2、2 翻到 4吞吐几乎等比上升这是线性区。从 8 到 16吞吐只涨了约三分之一说明已经进入收益递减。判据二成功率跌破 99% 的位置。一旦出现 429 或 529就说明有效并发已经越过了服务端能平稳承接的区间。生产阈值应该设在这个位置之下而不是恰好卡在它上面。判据三p95 与超时阈值的关系。生产环境通常把超时设在 p95 的 1.5 到 2 倍。如果并发 16 时 p95 是 6.8 秒超时就不该设成 8 秒而要设到 12 秒以上否则你会把本来能成功的慢请求主动掐断制造出一批假失败。判据四吞吐是否回落。并发 64 时吞吐低于并发 32这是典型的拥塞信号——多出来的请求没有变成产出只变成了排队和重试。把四个判据合起来本次示例的结论是稳态并发取 8 到 12突发上限取 16超过 16 必须靠队列而不是靠加并发。这个数字要写进配置而不是留在某个人的记忆里。需要额外说明的是梯度表的绝对数值受时段影响。同样的脚本在业务高峰期跑拐点可能提前一到两档。所以建议至少跑两轮一轮在你自己的业务高峰时段一轮在低谷时段取更保守的那组作为生产参数。拿到梯度表之后如果想对照不同模型的延迟表现再决定主力模型可以在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentgradient_table 上用相同 prompt 做一轮小样本对比。6. 从阈值到生产信号量、重试预算与熔断测出阈值只是第一步把它变成生产代码才是真正的收益点。以下三段代码可以直接嵌进批量生成 Worker。第一段信号量 有界队列。不要让任务无限制地gather用固定大小的信号量把 in-flight 钉死在阈值内超出部分排队等待。import asyncio MAX_INFLIGHT 10 # 来自压测结论留出安全余量 MAX_QUEUE 200 # 队列上限防止内存被堆积任务吃满 _sem asyncio.Semaphore(MAX_INFLIGHT) _queue_depth 0 async def guarded_call(coro_factory): global _queue_depth if _queue_depth MAX_QUEUE: raise RuntimeError(队列已满本次批次应降速或分片重跑) _queue_depth 1 try: async with _sem: return await coro_factory() finally: _queue_depth - 1第二段重试预算 指数退避 jitter。关键是区分值得重试和不该重试的状态码429、529、5xx 和网络异常值得重试400、401、403 重试一万次也没用只会浪费预算。import asyncio import random RETRYABLE {429, 500, 502, 503, 504, 529} MAX_ATTEMPTS 3 def backoff_delay(attempt, base0.6, cap12.0): raw min(cap, base * (2 ** attempt)) return raw * (0.5 random.random() * 0.5) # full jitter 的一半区间 async def call_with_budget(fn, *args, **kwargs): last_exc None for attempt in range(MAX_ATTEMPTS): try: resp await fn(*args, **kwargs) if resp.status_code in RETRYABLE: last_exc RuntimeError(fretryable status {resp.status_code}) else: return resp except (asyncio.TimeoutError, OSError) as exc: last_exc exc await asyncio.sleep(backoff_delay(attempt)) raise last_exc注意backoff_delay里的 jitter固定退避会让所有失败的请求在同一毫秒重新发起形成新一轮脉冲式拥塞。加了随机抖动之后重试请求会被摊平到一段时间窗口里这对稳定 p95 帮助很大。第三段滑动窗口熔断。当最近 N 次请求的成功率跌破阈值时主动暂停新请求一段时间给链路一个恢复窗口而不是继续往里灌。import time from collections import deque WINDOW 40 FAIL_RATE_TRIP 0.15 # 失败率超过 15% 触发 COOLDOWN 20.0 # 熔断冷却秒数 _events deque(maxlenWINDOW) _open_until 0.0 def record(success: bool): _events.append(success) def allowed() - bool: if time.time() _open_until: return False if len(_events) WINDOW: return True fail_rate 1 - sum(_events) / len(_events) if fail_rate FAIL_RATE_TRIP: global _open_until _open_until time.time() COOLDOWN _events.clear() return False return True这三段组合起来的逻辑是信号量管住并发上限重试预算管住失败放大熔断管住系统性劣化。三者缺一不可——只加信号量不加熔断遇到持续 429 时会一直卡在限速里慢慢磨只加重试不加信号量等于自己给自己制造并发。7. 报错对照表429 / 529 / 超时 / 401 分别怎么处理压测和线上排障时会遇到的状态码就那么几个但处理方式完全不同。下面这张表建议直接贴到值班手册里。现象典型来源处理方式不要做什么429瞬时请求超过限速降并发、按retry-after退避、启用熔断立刻加并发硬冲529服务端整体负载高拉长冷却窗口缩小批次错峰执行密集重试会加剧拥塞读超时请求已发出但迟迟无响应先确认是否流式、检查超时阈值是否过紧直接把超时砍到 5 秒连接池等待超时本地连接池上限小于并发档位调大max_connections误判为服务端问题401Key 错误或未注入检查环境变量是否被 shell 转义、有无多余空格反复重试403Key 无该模型权限换用当前账号可见的模型 ID改 Base URL 路径试探404路径写错Anthropic 协议用/v1/messagesOpenAI 兼容用/v1/chat/completions两套协议混用502网关层瞬时抖动计入重试预算观察是否成片出现单点失败就全批回滚两个排查习惯值得养成。第一记录状态码分布而不是只记成功率。成功率 95% 看着还行但如果失败全部是 429那就是并发问题如果全部是超时那就是超时阈值问题。第二把重试次数和最终成功数一起打点。重试 3 次才成功的请求成本是 3 倍当这类请求占比上升时说明你的并发已经贴着阈值在跑了应该主动往回收一档。8. 把压测变成例行公事回到最开始那条 429。它真正暴露的问题是我们把并发当成经验值而不是测量值。一旦完成一轮完整的梯度扫描你会发现该设多大这个问题有明确答案并且这个答案会随着模型版本、输入长度、时段变化而漂移所以它需要被周期性地重新测量而不是一次定终身。落地节奏建议这样安排上线前跑一轮全梯度确定稳态并发、突发上限和超时阈值之后每两周或每次切换主力模型时复跑一轮每次线上出现成片 429 或超时时把当时的并发、p95、重试率三个数拉出来对照梯度表判断是参数问题还是外部波动。如果你想直接开始路径很短先用 https://taotoken.net/models/detail/chat?utm_sourcetaotoken_aicg_blog_endutm_contentcta_chat 在模型对话页跑几条真实请求确认延迟量级再按业务用量在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcta_coding_plan 选择合适的套餐避免压测阶段反复触发限速然后在 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcta_api_keys 创建独立用途的 Key把压测 Key 和生产 Key 分开这样即使压测打爆了限速也不会波及线上 Worker最后回到 https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentcta_claude_code_doc 把 Claude Code 的接入配置核对一遍确保ANTHROPIC_BASE_URL、ANTHROPIC_AUTH_TOKEN和模型 ID 三项与压测环境完全一致。再补一句关于成本的提醒。在 Anthropic 盈利改善、市场关注其上市进程的背景下推理侧的成本纪律只会越来越重要。并发参数设得太保守浪费的是时间设得太激进浪费的是重试和超时带来的重复 Token后者往往更贵且更难被发现。压测的价值就在于把这两笔账都算清楚让每一个并发数字都有出处。