AI 回答像打字一样逐段出现,通常涉及“流式输出”。它能让读者更早看到内容,但页面动起来,不代表模型已经生成完,也不代表最后一句不会丢失。做聊天页面、接入模型接口或排查输出中断时,先分清网络分块、协议事件、文本片段和最终答案,比不断提高超时时间更有效。

本文用 DeepSeek 官方聊天接口的流式说明解释 SSE,再做一个完全离线的 Python 练习:把同一份虚构响应拆成七字节的小包,恢复“你好,世界”,同时识别缺少结束标记的情况。示例不使用密钥、不访问模型、不产生模型调用费;文中的响应由本站手工构造,离线练习已执行,不冒充真实 API 请求结果。

一、流式输出到底改变了什么

非流式调用通常等到完整响应返回,再一次展示结果。流式调用让客户端在生成过程中接收增量内容。它改善的是“多久开始看到答案”的体验,不承诺总耗时更短,也不自动提高答案质量。

SSE 是一种基于 HTTP 的服务器向客户端发送事件的文本格式。MDN 文档要求事件流使用 UTF-8,不同消息由空行分隔;消息可以带 data、事件名称和其他字段。它不是任意一段 TCP 数据,也不是每来一个包就能直接解析的完整 JSON。

一个网络包不等于一个字,更不等于一个事件。 一个中文字可能被分到两个字节块里,一个 JSON 事件也可能横跨多个读取结果。反过来,一次读取可能包含多个事件。把“收到一次数据”直接当作“收到一条答案”,就是乱码和解析错误的常见来源。

二、不要把结束标记当成通用标准

DeepSeek 当前聊天接口文档说明,设置 stream: true 后返回增量消息,使用 data-only SSE,并以 data: [DONE] 结束。这里的 [DONE] 是该接口的约定,不是所有 SSE 服务都必须使用的语法。

聊天流里的 delta.content 是增量文本,不是每次返回完整答案。某些事件只有角色信息或结束原因,未必有可见文字。若把前一个答案替换成当前片段,页面会只剩最后几个字;若把整个历史答案当成增量再追加,又会出现重复。

结束原因也要看:正常结束与输出长度受限不是同一种结果。收到 finish_reason 并不能证明答案事实正确;没有网络报错,也不能单凭页面已有一段文字认定完成。不同厂商的字段、工具调用事件和用量统计方式应分别按当前文档处理,不要混用其他接口的示例。

网络字节包、SSE事件、文本增量和界面答案四个层次的示意

三、先准备一个离线练习环境

准备 Python 和一个空文件夹,执行 python --version 确认命令可用。下面只用标准库 codecs 与 json,不需要安装模型 SDK,也不需要把任何账号密钥写进代码。

本练习刻意限制格式:每条事件只有一行 JSON data:,行尾统一为 LF,只有一个候选答案,没有工具调用与命名事件。它用来理解边界,不是生产级 SSE 解析器。通用解析还要处理多行 data、不同换行形式、事件类型、错误事件、重连等问题,正式接入优先采用接口推荐的 SDK 或成熟实现。

把下面内容保存为 stream_demo.py:

import codecs
import json

def recover(wire):
    decoder = codecs.getincrementaldecoder("utf-8")()
    buffer, pieces, reason, done = "", [], None, False
    for offset in range(0, len(wire), 7):
        buffer += decoder.decode(wire[offset:offset + 7])
        while "\n\n" in buffer:
            event, buffer = buffer.split("\n\n", 1)
            if event.startswith(":"):
                continue
            if not event.startswith("data: "):
                raise ValueError("unsupported demonstration event")
            data = event[6:]
            if data == "[DONE]":
                done = True
                continue
            choice = json.loads(data)["choices"][0]
            text = choice.get("delta", {}).get("content")
            if text is not None:
                pieces.append(text)
            if choice.get("finish_reason") is not None:
                reason = choice["finish_reason"]
    buffer += decoder.decode(b"", final=True)
    if buffer:
        raise ValueError("incomplete final event")
    return "".join(pieces), done, reason

def event(text=None, finish=None):
    delta = {} if text is None else {"content": text}
    obj = {"choices": [{"delta": delta, "finish_reason": finish}]}
    return "data: " + json.dumps(obj, ensure_ascii=False) + "\n\n"

wire = (event("你好,") + event("世界") +
        event(finish="stop") + "data: [DONE]\n\n").encode("utf-8")
assert recover(wire) == ("你好,世界", True, "stop")
cut = wire.removesuffix(b"data: [DONE]\n\n")
assert recover(cut) == ("你好,世界", False, "stop")
print(recover(wire))
print(recover(cut))

代码使用 bytes.removesuffix,因此需要 Python 3.9 或更新版本。在文件所在目录执行 python stream_demo.py。本站离线运行得到两行结果:完整输入恢复“你好,世界”、结束标志为真;删去结束事件后,文字相同,但结束标志为假。

这说明文字看起来完整,协议状态仍可能不完整。示例为了观察这种差异保留了第二种结果;实际产品应把它显示为中断或待确认,而不是悄悄标成成功。

四、逐层理解这段程序

第一层是增量 UTF-8 解码。解码器会保留尚未构成完整字符的字节,等后续字节到齐再输出。若每个包都单独调用普通解码并忽略错误,缺掉的字节可能永远回不来。

第二层是事件缓冲。只有找到空行分隔符,才取出一条完整事件;一个包没有分隔符时继续积累,一个包包含多条事件时循环处理。不要在 JSON 尚未到齐时解析,也不要在换行符跨包时丢掉缓冲。

第三层读取业务字段。空的 delta 不应被当成异常字符,结束原因不应直接拼进用户答案,注释也不应该出现在正文。示例遇到不支持的事件会明确报错,而不是假装懂得所有流格式。

第四层维护完成状态。答案文本、结束原因与结束标记分别保存。真实项目还应记录请求是否被取消、是否收到服务端错误、是否触发超时,避免用一个“加载中”布尔值表达所有情况。

五、接到真实页面时怎样做

  1. 在服务端保管模型凭据,浏览器通过自己有访问控制的服务请求,不把密钥塞进前端源码。
  2. 先验证普通请求能返回,再按该接口说明打开流式参数,确认响应类型和事件内容。
  3. 分别保存增量文本和运行状态,界面按增量追加;对空内容、结束事件和错误分支单独处理。
  4. 点击取消时同时处理界面状态与上游请求,不只把按钮隐藏。
  5. 完成后保存最终答案与必要状态,记录真实用量,不拿网页刷新次数猜测 token 消耗。

这里的服务端流式请求不一定使用浏览器原生 EventSource。很多模型接口需要带认证的 POST 请求,传输 API 的选择要符合实际请求方式,不能只因为叫 SSE 就照搬一个 GET 示例。

若接入后整段内容最后才一起出现,先检查服务端框架、中间代理与浏览器渲染是否缓冲。不要未经排查就全站关闭缓存;静态图片缓存和模型事件流的处理方式不是一回事。

六、中断、重试和重复回答怎么区分

断线后直接再发同一个问题,通常是一次新生成,不保证继续原回答。SSE 格式里存在事件 ID 与重连机制,也不代表具体模型接口一定支持从中断位置续传。只有官方明确提供相应协议,才能依赖它。

如果重新生成,应在页面上把旧的部分答案标为已中断,再以新的一次生成处理,避免把两次回答无提示地拼起来。若涉及工具调用或写操作,还要另外处理重复执行问题,不能认为“网络重试”天然安全。

排查时记录发生时间、状态码、错误类型、是否收到结束事件及脱敏后的请求标识即可。完整对话、凭据和客户资料不应直接贴到公开日志或求助帖里。服务器日志没有报错但前端显示异常时,继续比较服务端收到的事件和浏览器实际渲染,不只看其中一侧。

同一段可见答案在完成、中断和长度受限时应显示不同状态的示意

七、用三个小实验验收

先保持完整输入,检查输出文字与完成标记;再删去 [DONE],检查是否变成未完成;最后把结束原因改成 length,检查产品是否提示输出受限,而不是继续显示“完整答案”。最后一个是读者可做的扩展练习,本文不把它描述为模型实际输出。

有基础后,再加入正常空片段、多条事件同包、事件跨包与非法 UTF-8 的测试。每个实验只改变一种条件,更容易区分解码问题、解析问题和业务状态问题。离线样例通过以后,还必须在真正的接口、代理与浏览器组合里复测,才能判断整个链路。

八、速度体验和完成率分别记录

比较流式与非流式时,可以记录发起请求、收到首个可见文字、收到结束事件和页面完成渲染的时间。首个事件可能只有角色信息,不能把它直接当作读者已经看见答案。服务器时间与浏览器时间也不要未经处理混在一起计算。

在同一组任务、同一模型与相近输出范围下多做几次观察,分别记录首字等待、完整等待和中断次数,不拿一次特别快的结果代表所有用户。若界面每来几个字都重新排版整篇长文,卡顿可能来自前端渲染,而不是模型速度;可以适度合并更新,再检查阅读体验。

运行结束后,还要让人工检查是否答到了问题、是否遗漏条件。一个更早显示的错误答案没有因为“流式”而更可靠,正文逐段出现也不能替代来源核对。对长报告,可以清楚展示“生成中”“已中断”“生成完成”等状态,让用户知道当前能否把它作为完整文档使用。

流式输出的落地标准是:文字逐段可读,状态真实,中断可识别,错误可解释。提前看到几个字是体验优化,知道答案何时真正结束才是可靠交付。

官方资料