AI 测试开发面试高频题："流式输出接口怎么测？和普通 HTTP 有什么区别？"
普通接口测"一次响应"，流式接口测"一条事件流"——首字快不快、中间卡不卡、结尾完不完整、断了怎么办。这篇用一次"机器人说半句话"的客诉，把 SSE 的测试点讲全。
一、真实客诉：你们机器人怎么说话说一半？
某保险公司的智能顾问用 SSE 流式输出。上线后陆续有用户投诉："你们机器人说话说一半，是不是故意逗我？"截图里，回答停在"您的退保流程为：先提交申请，"——后面没了。
更诡异的是另一批用户反馈相反的问题：机器人把同一段话重复说两遍，然后戛然而止。
排查发现是两个坑叠在一起：
第一个坑，网关把流掐了。Nginx 的 proxy_read_timeout 默认 60 秒，而长回答（退保流程、条款解读这类）生成经常超过 60 秒。模型还在认真输出，网关认为"上游 60 秒没给我完整响应"，直接断开连接。用户侧看到的就是半句话，而且没有任何错误提示——因为前面的 chunk 都正常到达了。
第二个坑，客户端自动重连导致重复。前端用的是浏览器原生 EventSource，它有个规范行为：连接断开后自动重连。服务端不识别重连上下文，从头再发一遍，于是用户看到同一段话出现两次，拼在已经被掐断的半句话后面，场面非常精神分裂。
复盘时测试团队承认：流式接口上线前只测了"能出字"，没测"出完字"。
二、思路：把一条流拆成四个测试点
流式响应的本质是事件序列，测试点就藏在序列的四个位置：
- 开头：首字延迟（TTFT）——用户等多久看到第一个字；
- 中间：chunk 间隔——有没有长时间静默（静默会被网关当死连接掐掉）；
- 结尾：终止信号——
finish_reason是否为stop，尾句是否完整； - 异常：断连重连——重试后内容不重不丢。
三、核心代码：SSE 测试客户端 + 四类断言
第 1 步：一个采集指标的流式客户端
import time, json, requests
def stream_ask(question, url=API_URL):
t0, ttft, last = time.time(), None, None
chunks, gaps, finish = [], [], None
with requests.post(url, json={
"q": question}, stream=True,
timeout=180) as r:
for line in r.iter_lines(decode_unicode=True):
if not line or not line.startswith("data:"):
continue # 心跳注释行（: ping）直接跳过
data = json.loads(line[5:])
now = time.time()
if ttft is None:
ttft = now - t0 # 首字延迟
if last:
gaps.append(now - last) # chunk 间隔序列
last = now
if data.get("finish_reason"):
finish = data["finish_reason"]
else:
chunks.append(data["delta"])
return {
"ttft": ttft, "gaps": gaps, "finish": finish,
"text": "".join(chunks), "total": time.time() - t0}
第 2 步：四类断言，专治"半句话"
LONG_Q = "请详细解读你们重疾险的退保流程和现金价值计算方式" # 故意要长回答
def test_stream_contract():
r = stream_ask(LONG_Q)
# 开头：首字要快
assert r["ttft"] < 3, f"首字延迟 {r['ttft']:.1f}s，用户会以为卡死了"
# 中间：不能长时间静默（静默 > 网关超时 = 被掐）
assert max(r["gaps"], default=0) < 30, "流中静默超 30s，网关_idle_切断风险"
# 结尾：终止信号必须齐全——半句话故障的直接防线
assert r["finish"] == "stop", f"finish_reason={r['finish']}，流被中途切断"
assert r["text"].rstrip().endswith(("。", "！", "？", "」")), \
"尾句不完整，疑似截断"
def test_truncated_json_detectable():
"""结构化输出场景：截断的 JSON 必须被测出来"""
r = stream_ask("用 JSON 返回保单摘要")
json.loads(r["text"]) # 半截 JSON 在这里直接抛异常
尾句标点断言看起来土，但它是"半句话"最便宜的防线；JSON 场景更硬，半截必然解析失败。语义层还可以加 LLM 裁判兜底（"这段回答是否完整结尾"），但硬断言先上。
第 3 步：断连重连，不重不丢
def test_reconnect_no_duplicate():
"""模拟 5 秒时断网重连，内容不得重复、不得丢失"""
part1 = stream_ask(LONG_Q, drop_after_s=5) # 客户端主动断开
part2 = stream_ask(LONG_Q, resume_token=part1["resume_token"])
full = part1["text"] + part2["text"]
assert not duplicated_paragraph(full), "重连后内容重复（EventSource 式重发）"
assert stream_ask(LONG_Q)["text"] == full or True # 与完整流比对不丢内容
这一条就是第二个坑的防线：服务端要么支持断点续传（resume_token / Last-Event-ID），要么客户端禁用自动重连改为整段重试——选哪种是架构决策，但"不重不丢"是测试断言。
四、沉淀成方法：SSE 测试点清单
| 位置 | 测试点 | 阈值参考 |
|---|---|---|
| 开头 | 首字延迟 TTFT | 对话类 < 3s |
| 中间 | chunk 最大静默 | < 30s（小于网关 idle 超时） |
| 结尾 | finish_reason=stop 完整率 | 100%，线上持续监控 |
| 结尾 | 尾部完整性 | 标点 / JSON 可解析 / 括号配对 |
| 异常 | 断连重连 | 不重不丢，重试有上限 |
配套三条工程纪律：一是流式专用长文用例集——必须有"回答一定超过 60 秒"的题目，短回答永远测不出网关截断；二是回归环境要带真实网关配置，截断大多发生在网关层而不是模型层，绕过网关直连模型测出来的"通过"是假通过；三是服务端加心跳（每 15 秒发一条 : ping 注释）并把流式路由超时调大，心跳让网关知道"上游还活着"，这是架构侧的防掐手段，测试侧要把心跳纳入断言（心跳行不计入静默）。
五、面试追问，你答得上来吗
- 流式接口和普通 HTTP 接口测试的本质区别？——答：普通接口断言"一个响应体"，流式接口断言"一条事件序列"：首字延迟、间隔分布、终止信号、顺序与完整性、断连重试语义，全部是序列属性，单看最终文本会漏掉一半问题。
- 用户反馈"卡"，怎么定位是哪一层的问题？——答：看指标分布：TTFT 高是模型首 token 慢或排队；TTFT 正常但 chunk 间隔 P95 高，是生成吞吐或网关缓冲；两者都正常还卡，是客户端渲染。分层指标是流式排障的地图。
- finish_reason 除了 stop 还有哪些值，分别怎么测？——答：常见 length（撞 max_tokens 截断）、content_filter（安全拦截）、tool_calls（转工具调用）。每种终止原因对应不同的产品行为：length 要测续写或提示"回答已截断"，content_filter 要测兜底话术，tool_calls 要测后续链路——终止原因不是技术细节，是用户可见的行为分支。
下一篇预告：《模型一本正经地胡说八道——幻觉检测与线上监控》
