企业官网建设流程全解析

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，不用动其他配置。这对做多模型对比压测特别省事。

你需要准备三样东西，我把它叫做"三件套"：

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_URL="https://taotoken.net/api" export TAOTOKEN_API_KEY="sk-你的Key" export TAOTOKEN_MODEL="claude-sonnet-4-5"

注意：Base URL 结尾不要带/v1之外的路径，OpenAI 兼容接口的完整路径由 SDK 自动拼接。如果你手动发 HTTP 请求，对话补全的路径是/v1/chat/completions。

环境变量配好后，先跑一次最简单的连通性测试，确认 Key 有效、模型可访问。这一步不做压测，只验证链路。

import os from openai import OpenAI client = OpenAI( base_url=os.environ["TAOTOKEN_BASE_URL"], api_key=os.environ["TAOTOKEN_API_KEY"], ) resp = client.chat.completions.create( model=os.environ["TAOTOKEN_MODEL"], messages=[{"role": "user", "content": "只回复两个字：通了"}], max_tokens=16, ) 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.py：

import asyncio import json import os import time import statistics import httpx with open("config.json", "r", encoding="utf-8") as f: CFG = json.load(f) API_KEY = os.environ["TAOTOKEN_API_KEY"] HEADERS = { "Authorization": f"Bearer {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", headers=HEADERS, json=payload, timeout=CFG["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_connections=CFG["concurrency"] * 2) async with httpx.AsyncClient(limits=limits) 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(f"QPS: {len(ok) / wall:.2f}") print(f"TPM: {total_tokens / wall * 60:.0f}") print(f"TTFT P50/P95/P99: " f"{pct(ttfts,50)*1000:.0f}/" f"{pct(ttfts,95)*1000:.0f}/" f"{pct(ttfts,99)*1000:.0f} ms") print(f"TPOT P50/P95/P99: " f"{pct(tpots,50)*1000:.1f}/" f"{pct(tpots,95)*1000:.1f}/" f"{pct(tpots,99)*1000:.1f} ms") print(f"E2E 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 Token，0.82 + 142 × 0.0183 ≈ 3.42 秒，对得上，说明埋点没问题。如果 E2E 明显大于这个估算值，可能是流式解析漏了 chunk，或者 usage 没取到导致输出 Token 数偏小。

单请求验证通过后，开始做并发梯度测试。这是找容量拐点的关键步骤。把total_requests固定为 64，concurrency依次设为 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_ascii=False, indent=2) EOF echo "===== concurrency=$c =====" python bench.py done

跑完后把结果整理成表格，重点看三个信号：

判读规则很直接：当 TPM 不再随并发增长、而 TTFT P95 开始快速上升时，就是稳定容量的拐点。上表里拐点在并发 8 到 16 之间，生产配置应该取拐点左侧、留出余量的值，比如并发 8。

这里要强调一个常见误区：并发不是越高越好。并发从 16 提到 32，TPM 几乎没变，但 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_connections=concurrency*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 不超过 30ms，E2E P95 不超过 20 秒。

这样的描述才具备可复现性。别人拿到这套 SLO，可以用同样的脚本、同样的输入分布去验证，而不是各说各话。

线上监控时，把压测脚本里的采集逻辑抽成一个轻量探针，定时发一个固定长度的请求，记录 TTFT 和 TPOT，上报到监控系统。这样你能持续看到真实链路的延迟趋势，而不是等用户投诉才发现问题。探针的 prompt 要固定，输入长度要稳定，否则指标没有可比性。

容量规划上，用压测得到的拐点值乘以一个安全系数。比如拐点并发是 8，生产配置取 6，留 25% 余量应对流量波动和长请求。如果业务里有 RAG 这类长上下文场景，要单独压一轮，因为长输入的 Prefill 开销会显著拉高 TTFT，和短请求混在一起测会互相干扰。

最后提醒一点：TPM 配额和并发限制往往是两套独立的约束。你可能并发没到上限，但 TPM 先打满了，表现为请求被限流或降级。压测时同时记录 TPM 和错误率，才能判断到底撞的是哪个墙。如果发现 TPM 先到顶，要么申请更高配额，要么优化 prompt 减少输入 Token，要么把部分流量分流到其他模型。

到这里，从单次请求的 TTFT 打点，到并发梯度的拐点识别，再到线上 SLO 的定义和探针监控，整条指标链路就闭环了。你可以直接拿bench.py和config.json改成自己业务的输入分布，跑一轮属于自己的容量基线。