1. 为什么普通 Locust 脚本压不了大模型流式接口大模型流式输出接口和传统 REST 接口有个本质区别它不会一次性把完整响应体吐给你而是通过 SSEServer-Sent Events把内容切成很多小块边生成边推送。你用 Locust 默认的self.client.post()去压它会在响应完全结束后才返回于是你拿到的response_time是「从发请求到最后一个 token 落地」的总时长中间的首 Token 延迟、Token 间抖动、每秒出多少 token全被抹平成一个数字。更麻烦的是并发统计。流式接口的连接存活时间可能是 3 秒、8 秒甚至 30 秒取决于模型输出长度。Locust 按「请求数 / 时间」算 RPS但流式场景下你真正关心的是「同时有多少条连接在跑」「每条连接每秒吐多少 token」。如果只盯着 RPS你会得到一个漂亮但毫无意义的数字。我试过用默认脚本压一个流式接口报告里平均响应时间 6.2 秒看起来还行但实际上首 Token 要等 2.8 秒后面每个 token 间隔忽快忽慢用户体感早就崩了。所以必须自己解析 SSE 流把 TTFT首 Token 时间、ITLToken 间延迟、TPS每秒 Token 数拆出来单独上报。这篇就围绕 TaoToken 统一 Key 通道下的流式接口给你一套能直接跑的 Locust 脚本骨架包含 SSE 事件解析、超时配置、自定义指标上报以及压测执行后怎么校验结果是否可信。2. TaoToken 前置准备统一 Key 与流式通道确认在写脚本之前先把接入侧的事情理清楚。TaoToken 提供的是统一 API 通道你拿一个 Key 就能调不同模型流式接口走的是标准 SSE 格式stream: true打开后返回data:行序列。你需要先拿到 API Key。打开 https://taotoken.net/api-keys 登录后创建一个 Key复制出来。注意这个 Key 只在创建时完整显示一次丢了就重新建。拿到 Key 后先确认你的流式接口能通。用 curl 快速验证一下别急着上 Locustcurl -N -X POST https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer sk-你的Key \ -H Content-Type: application/json \ -d { model: claude-sonnet-4-20250514, stream: true, messages: [{role: user, content: 用三句话解释什么是SSE}] }-N参数关闭 curl 的缓冲你就能看到 token 一个个往外蹦。如果这一步卡住不动或者报 401先解决鉴权问题再往下走。接口文档在 https://taotoken.net/doc 里面有完整的请求参数和流式响应格式说明。确认能通之后记下你的 base_url 和 model 名称下一步写脚本要用。3. 可复制的 Locust 脚本骨架下面这份脚本可以直接存成locustfile.py。核心思路是用streamTrue拿到原始响应流逐块解码按行切分识别data:前缀解析 JSON区分event类型然后在关键时间点触发自定义事件上报。import json import time from locust import HttpUser, task, between, events class StreamLLMUser(HttpUser): wait_time between(1, 3) host https://taotoken.net def on_start(self): self.api_key sk-你的Key self.headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } task def stream_chat(self): payload { model: claude-sonnet-4-20250514, stream: True, messages: [ {role: user, content: 写一段200字的产品介绍} ], } request_start_time time.time() first_token_received False first_token_time None prev_token_time None token_count 0 valid_data_line_count 0 full_response_text with self.client.post( /api/v1/chat/completions, headersself.headers, jsonpayload, catch_responseTrue, streamTrue, timeout60, ) as response: if response.status_code ! 200: response.failure(fHTTP {response.status_code}) return try: for chunk in response.iter_content(chunk_sizeNone): if not chunk: continue try: decoded_chunk chunk.decode(utf-8) except UnicodeDecodeError: continue for line in decoded_chunk.splitlines(): stripped_line line.strip() if not stripped_line.startswith(data:): continue data_part stripped_line[5:].strip() if not data_part or data_part [DONE]: continue valid_data_line_count 1 try: sse_data_obj json.loads(data_part) except json.JSONDecodeError: continue event_type sse_data_obj.get(event) if event_type ! token: continue current_time time.time() token_count 1 if not first_token_received: first_token_time current_time first_token_received True ttft first_token_time - request_start_time self.environment.events.request.fire( request_typeTIMING, nameTime to First Token, context{}, response_timeint(ttft * 1000), response_lengthlen(chunk), exceptionNone, start_timerequest_start_time, ) if prev_token_time is not None: itl current_time - prev_token_time self.environment.events.request.fire( request_typeTIMING, nameInter-Token Latency, context{}, response_timeint(itl * 1000), response_lengthlen(stripped_line.encode(utf-8)), exceptionNone, start_timeprev_token_time, ) prev_token_time current_time full_response_text data_part \n except Exception as e: response.failure(f读取流式响应时发生未知错误: {e}) return total_response_time time.time() - request_start_time tokens_per_second 0 if total_response_time 0 and valid_data_line_count 0: tokens_per_second valid_data_line_count / total_response_time self.environment.events.request.fire( request_typeMETRIC, nameTotal Tokens, context{}, response_time0, response_lengthvalid_data_line_count, exceptionNone, start_timerequest_start_time, ) self.environment.events.request.fire( request_typeMETRIC, nameTokens Per Second, context{}, response_timetokens_per_second * 1000, response_length0, exceptionNone, start_timerequest_start_time, ) response.success()几个关键点解释一下。timeout60是必须的流式接口如果模型输出很长默认超时可能不够但也不能设太大否则卡死的连接会拖垮压测机。chunk_sizeNone让 requests 按底层 TCP 缓冲区大小返回数据避免人为切碎 SSE 事件。valid_data_line_count统计的是所有非空data:行包括最终校准行。token_count只统计event token的行用于计算 ITL。这两个计数分开是因为有些接口会在最后发一个汇总事件它算内容但不该参与 Token 间延迟计算。4. 验证请求与结果校验脚本写好后先用单用户跑一遍确认指标能正常上报locust -f locustfile.py --headless -u 1 -r 1 -t 30s --host https://taotoken.net跑完后看控制台输出你应该能看到Time to First Token、Inter-Token Latency、Total Tokens、Tokens Per Second这几行。如果某个指标没出现说明对应的事件没触发回去检查 SSE 解析逻辑。确认单用户没问题后再上并发。建议从 10 用户开始逐步加到 50、100locust -f locustfile.py --headless -u 50 -r 5 -t 120s --host https://taotoken.net --csvresult-u 50是并发用户数-r 5是每秒启动 5 个用户-t 120s跑 2 分钟--csvresult把结果导出成 CSV 方便分析。结果校验重点看三个地方。第一Time to First Token的 P95 是否稳定如果并发上去后 TTFT 从 1 秒飙到 5 秒说明服务端排队严重。第二Inter-Token Latency的波动正常应该在几十毫秒级别如果出现几百毫秒的尖刺可能是网络抖动或服务端 GC。第三Tokens Per Second的均值是否随并发下降如果 10 用户时 40 token/s50 用户时掉到 8 token/s说明吞吐已经打满。还有一个容易忽略的点失败率。流式接口在高压下可能出现连接被重置、data:行截断、JSON 解析失败等情况。脚本里用response.failure()标记了这些异常压测报告里会体现。如果失败率超过 1%先别急着加并发回去看日志定位是网络问题还是服务端限流。5. 本篇常见错排查报错一json.JSONDecodeError频繁出现。大概率是 chunk 边界把一行data:切成了两半。SSE 是按行传输的但 TCP 不保证按行到达。解决办法是在解析前维护一个缓冲区把不完整的行拼到下一次 chunk 里。简单做法是用splitlines()后检查最后一行是否完整不完整就暂存。报错二TTFT 为 0 或负数。检查request_start_time是不是在self.client.post()之前记录的。如果放在 with 块里面计时起点就晚了。另外确认first_token_received标志位只触发一次别在循环里重复赋值。报错三Locust 报告里Tokens Per Second显示为 0。因为response_time字段是整数毫秒tokens_per_second * 1000如果小于 1 会被截断成 0。可以在上报前加个判断或者把吞吐率乘以 10000 提高精度看报告时再除以 10。报错四并发上去后大量Connection reset by peer。先确认不是压测机端口耗尽用ulimit -n看文件描述符上限。如果是服务端主动断开可能是触发了限流这时候要看 TaoToken 控制台的用量统计确认是否超出配额。控制台地址是 https://taotoken.net/console 。报错五event字段不存在。不同模型的 SSE 格式可能有差异有些直接返回{choices:[{delta:{content:...}}]}没有顶层event。这时候要改成判断choices[0].delta.content是否存在。建议先 curl 一次看实际返回结构再调整解析逻辑。6. 长期压测与 Coding Plan 接入建议如果你只是偶尔验证一次流式接口性能上面这套脚本够用了。但如果你要把压测纳入 CI或者长期监控不同模型的流式表现手动跑 Locust 会比较累。这时候可以考虑用 TaoToken 的 Coding Plan它更适合把模型调用集成到自动化流程里配合定时任务做周期性压测。具体做法是把 Locust 脚本的参数化部分抽出来模型名称、prompt 长度、并发数做成配置项然后用 Coding Plan 的额度跑定时任务。这样你每周都能拿到一份 TTFT、ITL、TPS 的趋势数据而不是等到用户投诉才发现流式变慢了。接入文档在 https://taotoken.net/doc 里面有 Coding Plan 的调用方式和配额说明。模型对话调试可以用 https://taotoken.net/models 先在页面上确认目标模型的流式行为符合预期再写进压测脚本。最后提醒一句压测流式接口时压测机本身的网络出口带宽和 CPU 解码能力也会成为瓶颈。如果并发加到 200 以上发现指标异常先排除压测机自身问题再怀疑服务端。可以在压测机上开top看 Python 进程的 CPU 占用如果单核跑满说明解码逻辑太重考虑用多进程模式跑 Locust。