流式输出只显示半句话——SSE 流式接口的测试点

AI 测试开发面试高频题：“流式输出接口怎么测？和普通 HTTP 有什么区别？”
普通接口测"一次响应"，流式接口测"一条事件流"——首字快不快、中间卡不卡、结尾完不完整、断了怎么办。这篇用一次"机器人说半句话"的客诉，把 SSE 的测试点讲全。

一、真实客诉：你们机器人怎么说话说一半？

某保险公司的智能顾问用 SSE 流式输出。上线后陆续有用户投诉："你们机器人说话说一半，是不是故意逗我？“截图里，回答停在"您的退保流程为：先提交申请，”——后面没了。

更诡异的是另一批用户反馈相反的问题：机器人把同一段话重复说两遍，然后戛然而止。

排查发现是两个坑叠在一起：

第一个坑，网关把流掐了。Nginx 的 proxy_read_timeout 默认 60 秒，而长回答（退保流程、条款解读这类）生成经常超过 60 秒。模型还在认真输出，网关认为"上游 60 秒没给我完整响应"，直接断开连接。用户侧看到的就是半句话，而且没有任何错误提示——因为前面的 chunk 都正常到达了。

第二个坑，客户端自动重连导致重复。前端用的是浏览器原生 EventSource，它有个规范行为：连接断开后自动重连。服务端不识别重连上下文，从头再发一遍，于是用户看到同一段话出现两次，拼在已经被掐断的半句话后面，场面非常精神分裂。

复盘时测试团队承认：流式接口上线前只测了"能出字"，没测"出完字"。

二、思路：把一条流拆成四个测试点

流式响应的本质是事件序列，测试点就藏在序列的四个位置：

1. 开头：首字延迟（TTFT）——用户等多久看到第一个字；
2. 中间：chunk 间隔——有没有长时间静默（静默会被网关当死连接掐掉）；
3. 结尾：终止信号——finish_reason 是否为 stop，尾句是否完整；
4. 异常：断连重连——重试后内容不重不丢。

三、核心代码：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 测试点清单

配套三条工程纪律：一是流式专用长文用例集——必须有"回答一定超过 60 秒"的题目，短回答永远测不出网关截断；二是回归环境要带真实网关配置，截断大多发生在网关层而不是模型层，绕过网关直连模型测出来的"通过"是假通过；三是服务端加心跳（每 15 秒发一条 : ping 注释）并把流式路由超时调大，心跳让网关知道"上游还活着"，这是架构侧的防掐手段，测试侧要把心跳纳入断言（心跳行不计入静默）。

五、面试追问，你答得上来吗

1. 流式接口和普通 HTTP 接口测试的本质区别？——答：普通接口断言"一个响应体"，流式接口断言"一条事件序列"：首字延迟、间隔分布、终止信号、顺序与完整性、断连重试语义，全部是序列属性，单看最终文本会漏掉一半问题。
2. 用户反馈"卡"，怎么定位是哪一层的问题？——答：看指标分布：TTFT 高是模型首 token 慢或排队；TTFT 正常但 chunk 间隔 P95 高，是生成吞吐或网关缓冲；两者都正常还卡，是客户端渲染。分层指标是流式排障的地图。
3. finish_reason 除了 stop 还有哪些值，分别怎么测？——答：常见 length（撞 max_tokens 截断）、content_filter（安全拦截）、tool_calls（转工具调用）。每种终止原因对应不同的产品行为：length 要测续写或提示"回答已截断"，content_filter 要测兜底话术，tool_calls 要测后续链路——终止原因不是技术细节，是用户可见的行为分支。

下一篇预告：《模型一本正经地胡说八道——幻觉检测与线上监控》