1. 从一次请求说起为什么单看 QPS 会误判大模型服务很多人第一次给大模型接口做容量评估时习惯性套用普通 Web 服务的思路压到 100 QPS 不报错就认为这套服务能扛 100 QPS。我试过在一个 RAG 问答项目里按这个逻辑申请配额结果上线第二天就出现大面积超时——请求量其实只有 8 QPS但每个请求平均塞进去 1.2 万 Token 的检索上下文实际 Token 吞吐早就把配额打满了。大模型请求和普通接口最大的区别在于一次请求的计算量不是固定的。查询一个用户资料和查询另一个用户资料消耗的资源几乎一样但一次 200 Token 的闲聊和一次 5 万 Token 的文档分析在推理系统里完全是两个量级的任务。QPS 只统计次数不统计重量所以它单独出现时几乎没有指导意义。一条请求进入推理服务后大致会经历网关鉴权 → 限流 → 排队 → Tokenize → Prefill一次性处理全部输入 Token生成 KV Cache→ 生成第一个输出 Token → Decode逐个生成后续 Token→ 结束。这条链路里Prefill 决定用户等多久看到第一个字Decode 决定回答开始后输出是否流畅排队时间则取决于系统当前有多拥挤。围绕这条链路我们需要五个核心指标来描述服务状态QPS/RPM描述请求速率TPM描述 Token 流量并发描述系统内同时存在的请求数TTFT描述首字延迟TPOT描述单 Token 生成速度。它们不是五个独立数字而是同一条链路上不同位置的观测点。这篇内容的目标很明确用 TaoToken 的统一 Key 和 API 通道从单次请求的指标采集开始一路做到小规模并发压测把 QPS、TPM、并发、TTFT、TPOT 的采集口径和校验方法全部跑通。适合正在做模型容量评估、SLO 定义或者推理性能排障的开发和运维同学。下面所有脚本和配置都可以直接复制运行你只需要替换成自己的 Key。2. TaoToken 统一 Key 前置准备Base URL、Key 与模型 ID 三件套在开始压测之前先把调用通道固定下来。TaoToken 提供统一的 API 入口好处是压测脚本只需要维护一套 Base URL 和 Key切换模型时只改 Model ID不用动其他配置。这对做多模型对比压测特别省事。你需要准备三样东西我把它叫做三件套配置项值说明Base URLhttps://taotoken.net/api所有请求的统一入口不要加 UTM 参数API Key在控制台创建形如sk-xxxx压测建议单独建一个 Key 便于归因Model ID按需选择例如claude-sonnet-4-5、gpt-4o等以控制台模型列表为准Key 的创建入口在控制台的 API Keys 页面建议给压测单独建一个 Key这样在账单和限流统计里能和其他业务流量区分开。创建后立刻复制保存页面刷新后不再完整显示。如果你用的是 Claude Code 这类编码工具配置方式是把 Base URL 和 Key 写进环境变量或配置文件如果是自己写脚本直接用 HTTP 请求即可。下面给一个最小可运行的 Python 环境准备先确认通道能通再谈压测。# 建议用虚拟环境避免污染全局依赖 python3 -m venv venv source venv/bin/activate # 安装压测和指标采集需要的库 pip install httpx openai numpy# 把三件套写进环境变量后续脚本直接读取 export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_MODELclaude-sonnet-4-5注意Base URL 结尾不要带/v1之外的路径OpenAI 兼容接口的完整路径由 SDK 自动拼接。如果你手动发 HTTP 请求对话补全的路径是/v1/chat/completions。环境变量配好后先跑一次最简单的连通性测试确认 Key 有效、模型可访问。这一步不做压测只验证链路。import os from openai import OpenAI client OpenAI( base_urlos.environ[TAOTOKEN_BASE_URL], api_keyos.environ[TAOTOKEN_API_KEY], ) resp client.chat.completions.create( modelos.environ[TAOTOKEN_MODEL], messages[{role: user, content: 只回复两个字通了}], max_tokens16, ) print(resp.choices[0].message.content) print(usage:, resp.usage)跑通后你会看到返回内容和 usage 字段。usage 里的prompt_tokens、completion_tokens、total_tokens是后面计算 TPM 的基础务必确认它们有值。如果这里就报 401先检查 Key 是否复制完整、是否有多余空格如果报模型不存在去控制台核对 Model ID 拼写。这一步的意义在于把变量收敛到最小。压测出问题时如果连通性测试都没过就不用怀疑是并发或脚本的问题直接查 Key 和模型配置即可。3. 可复制的压测配置与指标埋点脚本这一节是全文的核心。我会给出一份完整的压测脚本它同时完成三件事按指定并发发请求、采集每个请求的 TTFT 和 TPOT、汇总 QPS/TPM/并发/延迟分位数。脚本用httpx的异步客户端实现因为流式响应才能准确测出 TTFT。先说清楚采集口径避免你拿到数字后不知道怎么解读TTFT从发出请求到收到第一个流式 chunk 的时间用time.perf_counter()打点。TPOT从第一个 chunk 到最后一个 chunk 的时间除以输出 Token 数 - 1。输出 Token 数优先用 usage 里的completion_tokens没有就按 chunk 数估算。QPS成功请求数 ÷ 压测总时长。TPM总 Token 数输入输出÷ 压测总时长分钟。并发脚本里用信号量控制同时在飞的请求数这就是运行并发。下面是可以直接复制的配置文件和脚本。先写一个config.json把压测参数外置方便反复调整{ base_url: https://taotoken.net/api, model: claude-sonnet-4-5, concurrency: 8, total_requests: 64, max_tokens: 256, prompt: 请用三句话解释什么是连续批处理要求语言简洁。, timeout: 120 }然后是主压测脚本bench.pyimport asyncio import json import os import time import statistics import httpx with open(config.json, r, encodingutf-8) as f: CFG json.load(f) API_KEY os.environ[TAOTOKEN_API_KEY] HEADERS { Authorization: fBearer {API_KEY}, Content-Type: application/json, } async def one_request(client, sem, results, idx): async with sem: payload { model: CFG[model], messages: [{role: user, content: CFG[prompt]}], max_tokens: CFG[max_tokens], stream: True, stream_options: {include_usage: True}, } start time.perf_counter() ttft None last None out_tokens 0 usage None try: async with client.stream( POST, f{CFG[base_url]}/v1/chat/completions, headersHEADERS, jsonpayload, timeoutCFG[timeout], ) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): if not line or not line.startswith(data:): continue data line[5:].strip() if data [DONE]: break chunk json.loads(data) if chunk.get(usage): usage chunk[usage] choices chunk.get(choices) or [] if choices and choices[0].get(delta, {}).get(content): now time.perf_counter() if ttft is None: ttft now - start last now out_tokens 1 end time.perf_counter() if usage and usage.get(completion_tokens): out_tokens usage[completion_tokens] tpot None if ttft is not None and last is not None and out_tokens 1: tpot (last - start - ttft) / (out_tokens - 1) results.append({ ok: True, ttft: ttft, tpot: tpot, e2e: end - start, in_tokens: (usage or {}).get(prompt_tokens, 0), out_tokens: out_tokens, }) except Exception as e: results.append({ok: False, err: str(e), idx: idx}) async def main(): sem asyncio.Semaphore(CFG[concurrency]) results [] limits httpx.Limits(max_connectionsCFG[concurrency] * 2) async with httpx.AsyncClient(limitslimits) as client: wall_start time.perf_counter() tasks [ one_request(client, sem, results, i) for i in range(CFG[total_requests]) ] await asyncio.gather(*tasks) wall time.perf_counter() - wall_start ok [r for r in results if r[ok]] fail [r for r in results if not r[ok]] if not ok: print(全部失败先查 Key 和网络) for r in fail[:3]: print(r[err]) return def pct(vals, p): vals sorted(v for v in vals if v is not None) if not vals: return None k int(round((p / 100) * (len(vals) - 1))) return vals[k] ttfts [r[ttft] for r in ok] tpots [r[tpot] for r in ok] e2es [r[e2e] for r in ok] total_tokens sum(r[in_tokens] r[out_tokens] for r in ok) print(f成功 {len(ok)} / 失败 {len(fail)}) print(f墙钟时长: {wall:.2f}s) print(fQPS: {len(ok) / wall:.2f}) print(fTPM: {total_tokens / wall * 60:.0f}) print(fTTFT P50/P95/P99: f{pct(ttfts,50)*1000:.0f}/ f{pct(ttfts,95)*1000:.0f}/ f{pct(ttfts,99)*1000:.0f} ms) print(fTPOT P50/P95/P99: f{pct(tpots,50)*1000:.1f}/ f{pct(tpots,95)*1000:.1f}/ f{pct(tpots,99)*1000:.1f} ms) print(fE2E P50/P95/P99: f{pct(e2es,50):.2f}/ f{pct(e2es,95):.2f}/ f{pct(e2es,99):.2f} s) if __name__ __main__: asyncio.run(main())这份脚本有几个设计点值得说明。第一用stream_options.include_usage让服务端在流式响应末尾返回 usage这样输出 Token 数是准确的而不是靠数 chunk。第二TTFT 和 TPOT 都用perf_counter打点避免系统时间跳变影响。第三并发用信号量控制total_requests大于concurrency时请求会分批进入更接近真实流量。跑之前确认config.json里的base_url和脚本里拼接的路径一致。脚本用的是{base_url}/v1/chat/completions所以 config 里填https://taotoken.net/api即可不要重复带/v1。4. 验证请求与结果校验从单请求到并发梯度先做单请求验证把concurrency设为 1、total_requests设为 1跑一次看输出是否合理。这一步的目的是确认指标采集口径正确而不是看性能。python bench.py正常输出类似成功 1 / 失败 0 墙钟时长: 3.42s QPS: 0.29 TPM: 4210 TTFT P50/P95/P99: 820/820/820 ms TPOT P50/P95/P99: 18.3/18.3/18.3 ms E2E P50/P95/P99: 3.42/3.42/3.42 s校验逻辑E2E 应该约等于 TTFT (输出 Token 数 - 1) × TPOT。以上面为例输出约 143 Token0.82 142 × 0.0183 ≈ 3.42 秒对得上说明埋点没问题。如果 E2E 明显大于这个估算值可能是流式解析漏了 chunk或者 usage 没取到导致输出 Token 数偏小。单请求验证通过后开始做并发梯度测试。这是找容量拐点的关键步骤。把total_requests固定为 64concurrency依次设为 1、4、8、16、32每次跑完记录结果。建议写个小脚本自动跑梯度for c in 1 4 8 16 32; do python - EOF import json cfg json.load(open(config.json)) cfg[concurrency] $c json.dump(cfg, open(config.json, w), ensure_asciiFalse, indent2) EOF echo concurrency$c python bench.py done跑完后把结果整理成表格重点看三个信号并发QPSTPMTTFT P95TPOT P95判断10.294210820ms18.3ms基线41.0515200950ms21.0ms吞吐上升延迟平稳81.92278001300ms26.5ms接近饱和162.10301003400ms41.2ms吞吐停滞TTFT 恶化322.12305008900ms68.7ms过载队列堆积判读规则很直接当 TPM 不再随并发增长、而 TTFT P95 开始快速上升时就是稳定容量的拐点。上表里拐点在并发 8 到 16 之间生产配置应该取拐点左侧、留出余量的值比如并发 8。这里要强调一个常见误区并发不是越高越好。并发从 16 提到 32TPM 几乎没变但 TTFT P95 从 3.4 秒涨到 8.9 秒用户体验直接崩掉。多出来的并发只是让请求在队列里排队没有增加任何有效吞吐。如果你要对比不同模型或不同输入长度下的表现只需要改config.json里的model和prompt其他不动。这也是用统一 Key 的好处——切换模型不用改脚本指标口径完全一致横向对比才有意义。5. 本篇常见报错排查401、local proxy failed、reading choices、OAuth压测过程中最容易撞上的几类报错我按出现频率排一下并给出定位方法。401 Unauthorized。最常见的原因是 Key 没读到或格式不对。先确认环境变量真的注入了echo $TAOTOKEN_API_KEY | head -c 8如果输出为空说明当前 shell 没加载。注意export只在当前会话有效换终端要重新执行。如果 Key 前几位正常检查请求头是不是Bearer加空格再加 Key少空格会直接 401。另外压测用的 Key 如果被单独限流或禁用也会返回 401 或 403去控制台确认 Key 状态。local proxy failed / connection error。这类报错通常出现在httpx或openai客户端层提示连接不上。先排除本机网络和 DNS用 curl 直接打一次curl -sS -o /dev/null -w %{http_code}\n \ -X POST $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\:\hi\}],\max_tokens\:8}返回 200 说明通道没问题问题在脚本的客户端配置。常见坑是httpx的连接池上限设得太小高并发时连接被耗尽表现为大量 connection error。脚本里已经用httpx.Limits(max_connectionsconcurrency*2)处理了如果你自己改脚本记得同步调整。reading choices / KeyError choices。这个报错说明脚本在解析响应时某个 chunk 里没有choices字段。流式响应里最后一个带 usage 的 chunk 通常choices为空数组如果代码直接写chunk[choices][0]就会崩。正确写法是先判断choices chunk.get(choices) or [] if choices and choices[0].get(delta, {}).get(content): ...另外如果服务端返回的是错误 JSON比如限流提示结构里也没有choices同样会触发这个错。建议在解析前先检查chunk.get(error)有错误就打印出来。OAuth / 认证方式不匹配。如果你用的是 Claude Code 或某些 CLI 工具它们可能默认走 OAuth 或特定的认证流程而不是简单的 Bearer Token。这时候需要把配置改成 API Key 模式明确指定 Base URL、Key 和 Model ID 三件套。以 Claude Code 为例配置里要写清楚{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: sk-你的Key, ANTHROPIC_MODEL: claude-sonnet-4-5 } }三件套缺一不可Base URL 决定请求发到哪Key 决定身份Model ID 决定用哪个模型。只配了 Key 没配 Base URL工具会往默认地址发自然认证失败。如果你在 Cline、Cursor 这类工具里接入同样检查这三项是否齐全尤其是 Model ID 要和控制台列表完全一致大小写和连字符都不能错。还有一个隐蔽的坑超时设置过短。压测高并发时单个请求的 E2E 可能到几十秒如果客户端 timeout 设成 10 秒会大量报超时看起来像服务端问题其实是客户端主动断开。脚本里timeout设的是 120 秒你可以按业务最长响应时间调整。6. 把指标链路用起来从压测结论到线上 SLO跑完上面的梯度测试你手里应该有一张并发-吞吐-延迟对照表。这张表的价值不只是选一个并发数字而是帮你定义一套可验证的 SLO。一个完整的 SLO 不能只写支持 8 QPS而要同时约束输入输出规模和延迟目标。比如基于前面的测试结果可以这样定义在输入 Token P95 不超过 8000、输出 Token P95 不超过 500 的前提下稳定 2 QPS 时成功率不低于 99.5%TTFT P95 不超过 1.5 秒TPOT P95 不超过 30msE2E P95 不超过 20 秒。这样的描述才具备可复现性。别人拿到这套 SLO可以用同样的脚本、同样的输入分布去验证而不是各说各话。线上监控时把压测脚本里的采集逻辑抽成一个轻量探针定时发一个固定长度的请求记录 TTFT 和 TPOT上报到监控系统。这样你能持续看到真实链路的延迟趋势而不是等用户投诉才发现问题。探针的 prompt 要固定输入长度要稳定否则指标没有可比性。容量规划上用压测得到的拐点值乘以一个安全系数。比如拐点并发是 8生产配置取 6留 25% 余量应对流量波动和长请求。如果业务里有 RAG 这类长上下文场景要单独压一轮因为长输入的 Prefill 开销会显著拉高 TTFT和短请求混在一起测会互相干扰。最后提醒一点TPM 配额和并发限制往往是两套独立的约束。你可能并发没到上限但 TPM 先打满了表现为请求被限流或降级。压测时同时记录 TPM 和错误率才能判断到底撞的是哪个墙。如果发现 TPM 先到顶要么申请更高配额要么优化 prompt 减少输入 Token要么把部分流量分流到其他模型。到这里从单次请求的 TTFT 打点到并发梯度的拐点识别再到线上 SLO 的定义和探针监控整条指标链路就闭环了。你可以直接拿bench.py和config.json改成自己业务的输入分布跑一轮属于自己的容量基线。 
长河君 · 2026/10/18 1:02:33 · 阅读 1.2k 从一次请求到整套推理集群：用 TaoToken 统一 Key 实测 QPS、TPM、并发与 TTFT/TPOT 性能指标
1. 从一次请求说起为什么单看 QPS 会误判大模型服务很多人第一次给大模型接口做容量评估时习惯性套用普通 Web 服务的思路压到 100 QPS 不报错就认为这套服务能扛 100 QPS。我试过在一个 RAG 问答项目里按这个逻辑申请配额结果上线第二天就出现大面积超时——请求量其实只有 8 QPS但每个请求平均塞进去 1.2 万 Token 的检索上下文实际 Token 吞吐早就把配额打满了。大模型请求和普通接口最大的区别在于一次请求的计算量不是固定的。查询一个用户资料和查询另一个用户资料消耗的资源几乎一样但一次 200 Token 的闲聊和一次 5 万 Token 的文档分析在推理系统里完全是两个量级的任务。QPS 只统计次数不统计重量所以它单独出现时几乎没有指导意义。一条请求进入推理服务后大致会经历网关鉴权 → 限流 → 排队 → Tokenize → Prefill一次性处理全部输入 Token生成 KV Cache→ 生成第一个输出 Token → Decode逐个生成后续 Token→ 结束。这条链路里Prefill 决定用户等多久看到第一个字Decode 决定回答开始后输出是否流畅排队时间则取决于系统当前有多拥挤。围绕这条链路我们需要五个核心指标来描述服务状态QPS/RPM描述请求速率TPM描述 Token 流量并发描述系统内同时存在的请求数TTFT描述首字延迟TPOT描述单 Token 生成速度。它们不是五个独立数字而是同一条链路上不同位置的观测点。这篇内容的目标很明确用 TaoToken 的统一 Key 和 API 通道从单次请求的指标采集开始一路做到小规模并发压测把 QPS、TPM、并发、TTFT、TPOT 的采集口径和校验方法全部跑通。适合正在做模型容量评估、SLO 定义或者推理性能排障的开发和运维同学。下面所有脚本和配置都可以直接复制运行你只需要替换成自己的 Key。2. TaoToken 统一 Key 前置准备Base URL、Key 与模型 ID 三件套在开始压测之前先把调用通道固定下来。TaoToken 提供统一的 API 入口好处是压测脚本只需要维护一套 Base URL 和 Key切换模型时只改 Model ID不用动其他配置。这对做多模型对比压测特别省事。你需要准备三样东西我把它叫做三件套配置项值说明Base URLhttps://taotoken.net/api所有请求的统一入口不要加 UTM 参数API Key在控制台创建形如sk-xxxx压测建议单独建一个 Key 便于归因Model ID按需选择例如claude-sonnet-4-5、gpt-4o等以控制台模型列表为准Key 的创建入口在控制台的 API Keys 页面建议给压测单独建一个 Key这样在账单和限流统计里能和其他业务流量区分开。创建后立刻复制保存页面刷新后不再完整显示。如果你用的是 Claude Code 这类编码工具配置方式是把 Base URL 和 Key 写进环境变量或配置文件如果是自己写脚本直接用 HTTP 请求即可。下面给一个最小可运行的 Python 环境准备先确认通道能通再谈压测。# 建议用虚拟环境避免污染全局依赖 python3 -m venv venv source venv/bin/activate # 安装压测和指标采集需要的库 pip install httpx openai numpy# 把三件套写进环境变量后续脚本直接读取 export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_MODELclaude-sonnet-4-5注意Base URL 结尾不要带/v1之外的路径OpenAI 兼容接口的完整路径由 SDK 自动拼接。如果你手动发 HTTP 请求对话补全的路径是/v1/chat/completions。环境变量配好后先跑一次最简单的连通性测试确认 Key 有效、模型可访问。这一步不做压测只验证链路。import os from openai import OpenAI client OpenAI( base_urlos.environ[TAOTOKEN_BASE_URL], api_keyos.environ[TAOTOKEN_API_KEY], ) resp client.chat.completions.create( modelos.environ[TAOTOKEN_MODEL], messages[{role: user, content: 只回复两个字通了}], max_tokens16, ) print(resp.choices[0].message.content) print(usage:, resp.usage)跑通后你会看到返回内容和 usage 字段。usage 里的prompt_tokens、completion_tokens、total_tokens是后面计算 TPM 的基础务必确认它们有值。如果这里就报 401先检查 Key 是否复制完整、是否有多余空格如果报模型不存在去控制台核对 Model ID 拼写。这一步的意义在于把变量收敛到最小。压测出问题时如果连通性测试都没过就不用怀疑是并发或脚本的问题直接查 Key 和模型配置即可。3. 可复制的压测配置与指标埋点脚本这一节是全文的核心。我会给出一份完整的压测脚本它同时完成三件事按指定并发发请求、采集每个请求的 TTFT 和 TPOT、汇总 QPS/TPM/并发/延迟分位数。脚本用httpx的异步客户端实现因为流式响应才能准确测出 TTFT。先说清楚采集口径避免你拿到数字后不知道怎么解读TTFT从发出请求到收到第一个流式 chunk 的时间用time.perf_counter()打点。TPOT从第一个 chunk 到最后一个 chunk 的时间除以输出 Token 数 - 1。输出 Token 数优先用 usage 里的completion_tokens没有就按 chunk 数估算。QPS成功请求数 ÷ 压测总时长。TPM总 Token 数输入输出÷ 压测总时长分钟。并发脚本里用信号量控制同时在飞的请求数这就是运行并发。下面是可以直接复制的配置文件和脚本。先写一个config.json把压测参数外置方便反复调整{ base_url: https://taotoken.net/api, model: claude-sonnet-4-5, concurrency: 8, total_requests: 64, max_tokens: 256, prompt: 请用三句话解释什么是连续批处理要求语言简洁。, timeout: 120 }然后是主压测脚本bench.pyimport asyncio import json import os import time import statistics import httpx with open(config.json, r, encodingutf-8) as f: CFG json.load(f) API_KEY os.environ[TAOTOKEN_API_KEY] HEADERS { Authorization: fBearer {API_KEY}, Content-Type: application/json, } async def one_request(client, sem, results, idx): async with sem: payload { model: CFG[model], messages: [{role: user, content: CFG[prompt]}], max_tokens: CFG[max_tokens], stream: True, stream_options: {include_usage: True}, } start time.perf_counter() ttft None last None out_tokens 0 usage None try: async with client.stream( POST, f{CFG[base_url]}/v1/chat/completions, headersHEADERS, jsonpayload, timeoutCFG[timeout], ) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): if not line or not line.startswith(data:): continue data line[5:].strip() if data [DONE]: break chunk json.loads(data) if chunk.get(usage): usage chunk[usage] choices chunk.get(choices) or [] if choices and choices[0].get(delta, {}).get(content): now time.perf_counter() if ttft is None: ttft now - start last now out_tokens 1 end time.perf_counter() if usage and usage.get(completion_tokens): out_tokens usage[completion_tokens] tpot None if ttft is not None and last is not None and out_tokens 1: tpot (last - start - ttft) / (out_tokens - 1) results.append({ ok: True, ttft: ttft, tpot: tpot, e2e: end - start, in_tokens: (usage or {}).get(prompt_tokens, 0), out_tokens: out_tokens, }) except Exception as e: results.append({ok: False, err: str(e), idx: idx}) async def main(): sem asyncio.Semaphore(CFG[concurrency]) results [] limits httpx.Limits(max_connectionsCFG[concurrency] * 2) async with httpx.AsyncClient(limitslimits) as client: wall_start time.perf_counter() tasks [ one_request(client, sem, results, i) for i in range(CFG[total_requests]) ] await asyncio.gather(*tasks) wall time.perf_counter() - wall_start ok [r for r in results if r[ok]] fail [r for r in results if not r[ok]] if not ok: print(全部失败先查 Key 和网络) for r in fail[:3]: print(r[err]) return def pct(vals, p): vals sorted(v for v in vals if v is not None) if not vals: return None k int(round((p / 100) * (len(vals) - 1))) return vals[k] ttfts [r[ttft] for r in ok] tpots [r[tpot] for r in ok] e2es [r[e2e] for r in ok] total_tokens sum(r[in_tokens] r[out_tokens] for r in ok) print(f成功 {len(ok)} / 失败 {len(fail)}) print(f墙钟时长: {wall:.2f}s) print(fQPS: {len(ok) / wall:.2f}) print(fTPM: {total_tokens / wall * 60:.0f}) print(fTTFT P50/P95/P99: f{pct(ttfts,50)*1000:.0f}/ f{pct(ttfts,95)*1000:.0f}/ f{pct(ttfts,99)*1000:.0f} ms) print(fTPOT P50/P95/P99: f{pct(tpots,50)*1000:.1f}/ f{pct(tpots,95)*1000:.1f}/ f{pct(tpots,99)*1000:.1f} ms) print(fE2E P50/P95/P99: f{pct(e2es,50):.2f}/ f{pct(e2es,95):.2f}/ f{pct(e2es,99):.2f} s) if __name__ __main__: asyncio.run(main())这份脚本有几个设计点值得说明。第一用stream_options.include_usage让服务端在流式响应末尾返回 usage这样输出 Token 数是准确的而不是靠数 chunk。第二TTFT 和 TPOT 都用perf_counter打点避免系统时间跳变影响。第三并发用信号量控制total_requests大于concurrency时请求会分批进入更接近真实流量。跑之前确认config.json里的base_url和脚本里拼接的路径一致。脚本用的是{base_url}/v1/chat/completions所以 config 里填https://taotoken.net/api即可不要重复带/v1。4. 验证请求与结果校验从单请求到并发梯度先做单请求验证把concurrency设为 1、total_requests设为 1跑一次看输出是否合理。这一步的目的是确认指标采集口径正确而不是看性能。python bench.py正常输出类似成功 1 / 失败 0 墙钟时长: 3.42s QPS: 0.29 TPM: 4210 TTFT P50/P95/P99: 820/820/820 ms TPOT P50/P95/P99: 18.3/18.3/18.3 ms E2E P50/P95/P99: 3.42/3.42/3.42 s校验逻辑E2E 应该约等于 TTFT (输出 Token 数 - 1) × TPOT。以上面为例输出约 143 Token0.82 142 × 0.0183 ≈ 3.42 秒对得上说明埋点没问题。如果 E2E 明显大于这个估算值可能是流式解析漏了 chunk或者 usage 没取到导致输出 Token 数偏小。单请求验证通过后开始做并发梯度测试。这是找容量拐点的关键步骤。把total_requests固定为 64concurrency依次设为 1、4、8、16、32每次跑完记录结果。建议写个小脚本自动跑梯度for c in 1 4 8 16 32; do python - EOF import json cfg json.load(open(config.json)) cfg[concurrency] $c json.dump(cfg, open(config.json, w), ensure_asciiFalse, indent2) EOF echo concurrency$c python bench.py done跑完后把结果整理成表格重点看三个信号并发QPSTPMTTFT P95TPOT P95判断10.294210820ms18.3ms基线41.0515200950ms21.0ms吞吐上升延迟平稳81.92278001300ms26.5ms接近饱和162.10301003400ms41.2ms吞吐停滞TTFT 恶化322.12305008900ms68.7ms过载队列堆积判读规则很直接当 TPM 不再随并发增长、而 TTFT P95 开始快速上升时就是稳定容量的拐点。上表里拐点在并发 8 到 16 之间生产配置应该取拐点左侧、留出余量的值比如并发 8。这里要强调一个常见误区并发不是越高越好。并发从 16 提到 32TPM 几乎没变但 TTFT P95 从 3.4 秒涨到 8.9 秒用户体验直接崩掉。多出来的并发只是让请求在队列里排队没有增加任何有效吞吐。如果你要对比不同模型或不同输入长度下的表现只需要改config.json里的model和prompt其他不动。这也是用统一 Key 的好处——切换模型不用改脚本指标口径完全一致横向对比才有意义。5. 本篇常见报错排查401、local proxy failed、reading choices、OAuth压测过程中最容易撞上的几类报错我按出现频率排一下并给出定位方法。401 Unauthorized。最常见的原因是 Key 没读到或格式不对。先确认环境变量真的注入了echo $TAOTOKEN_API_KEY | head -c 8如果输出为空说明当前 shell 没加载。注意export只在当前会话有效换终端要重新执行。如果 Key 前几位正常检查请求头是不是Bearer加空格再加 Key少空格会直接 401。另外压测用的 Key 如果被单独限流或禁用也会返回 401 或 403去控制台确认 Key 状态。local proxy failed / connection error。这类报错通常出现在httpx或openai客户端层提示连接不上。先排除本机网络和 DNS用 curl 直接打一次curl -sS -o /dev/null -w %{http_code}\n \ -X POST $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\:\hi\}],\max_tokens\:8}返回 200 说明通道没问题问题在脚本的客户端配置。常见坑是httpx的连接池上限设得太小高并发时连接被耗尽表现为大量 connection error。脚本里已经用httpx.Limits(max_connectionsconcurrency*2)处理了如果你自己改脚本记得同步调整。reading choices / KeyError choices。这个报错说明脚本在解析响应时某个 chunk 里没有choices字段。流式响应里最后一个带 usage 的 chunk 通常choices为空数组如果代码直接写chunk[choices][0]就会崩。正确写法是先判断choices chunk.get(choices) or [] if choices and choices[0].get(delta, {}).get(content): ...另外如果服务端返回的是错误 JSON比如限流提示结构里也没有choices同样会触发这个错。建议在解析前先检查chunk.get(error)有错误就打印出来。OAuth / 认证方式不匹配。如果你用的是 Claude Code 或某些 CLI 工具它们可能默认走 OAuth 或特定的认证流程而不是简单的 Bearer Token。这时候需要把配置改成 API Key 模式明确指定 Base URL、Key 和 Model ID 三件套。以 Claude Code 为例配置里要写清楚{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: sk-你的Key, ANTHROPIC_MODEL: claude-sonnet-4-5 } }三件套缺一不可Base URL 决定请求发到哪Key 决定身份Model ID 决定用哪个模型。只配了 Key 没配 Base URL工具会往默认地址发自然认证失败。如果你在 Cline、Cursor 这类工具里接入同样检查这三项是否齐全尤其是 Model ID 要和控制台列表完全一致大小写和连字符都不能错。还有一个隐蔽的坑超时设置过短。压测高并发时单个请求的 E2E 可能到几十秒如果客户端 timeout 设成 10 秒会大量报超时看起来像服务端问题其实是客户端主动断开。脚本里timeout设的是 120 秒你可以按业务最长响应时间调整。6. 把指标链路用起来从压测结论到线上 SLO跑完上面的梯度测试你手里应该有一张并发-吞吐-延迟对照表。这张表的价值不只是选一个并发数字而是帮你定义一套可验证的 SLO。一个完整的 SLO 不能只写支持 8 QPS而要同时约束输入输出规模和延迟目标。比如基于前面的测试结果可以这样定义在输入 Token P95 不超过 8000、输出 Token P95 不超过 500 的前提下稳定 2 QPS 时成功率不低于 99.5%TTFT P95 不超过 1.5 秒TPOT P95 不超过 30msE2E P95 不超过 20 秒。这样的描述才具备可复现性。别人拿到这套 SLO可以用同样的脚本、同样的输入分布去验证而不是各说各话。线上监控时把压测脚本里的采集逻辑抽成一个轻量探针定时发一个固定长度的请求记录 TTFT 和 TPOT上报到监控系统。这样你能持续看到真实链路的延迟趋势而不是等用户投诉才发现问题。探针的 prompt 要固定输入长度要稳定否则指标没有可比性。容量规划上用压测得到的拐点值乘以一个安全系数。比如拐点并发是 8生产配置取 6留 25% 余量应对流量波动和长请求。如果业务里有 RAG 这类长上下文场景要单独压一轮因为长输入的 Prefill 开销会显著拉高 TTFT和短请求混在一起测会互相干扰。最后提醒一点TPM 配额和并发限制往往是两套独立的约束。你可能并发没到上限但 TPM 先打满了表现为请求被限流或降级。压测时同时记录 TPM 和错误率才能判断到底撞的是哪个墙。如果发现 TPM 先到顶要么申请更高配额要么优化 prompt 减少输入 Token要么把部分流量分流到其他模型。到这里从单次请求的 TTFT 打点到并发梯度的拐点识别再到线上 SLO 的定义和探针监控整条指标链路就闭环了。你可以直接拿bench.py和config.json改成自己业务的输入分布跑一轮属于自己的容量基线。