SSE、WebSocket 与流式响应：先分帧，再解释

流式响应让服务端分段发送内容，客户端不必等整段回答结束才开始显示。一个分段可能是网络传来的字节块，也可能是已经按协议拼好的事件；它们不是同一个单位。

它像听语音：对方还没说完，你已经听到前半句。普通响应像收到一封写完的信；流式响应则边写边送。接收端需要把陆续到达的片段拼起来，再判断一句话或一条数据是否完整。

下面用“你”“好”两个文字事件练习拼接：先看完整结果，再把传输内容切成更小的块，观察拼接过程。这样可以理解聊天界面怎样逐步显示文字，也能看清它何时得到完整数据。

1. 先看两条完整事件

wire 是准备传输的文本。下面每条 data: 行后都有一个空行；空行就是事件边界：

data: {"delta":"你"}

data: {"delta":"好"}

delta 表示这次新增的文字。先分出两条事件，再把两个值相连，结果是 你好。实际网络可能把第一条行拆在中间，接收程序要保存未完整的内容；后面的实验专门观察这一步。

再决定用哪种传输方式

SSE（Server-Sent Events）让服务器沿一个 HTTP 响应持续发送文本事件；浏览器原生 EventSource 接收的是服务端到客户端的单向消息。HTML 标准规定了 text/event-stream、字段行和空行结束事件的格式。WebSocket 建立后可双向发送消息，适合持续交互；但双向能力不是所有聊天页面都需要。MDN WebSocket API介绍了浏览器接口及应用需自行处理的流控问题。

选型时还需考虑代理服务器、认证方式、浏览器和提供方 API 的要求。SSE 与 WebSocket 描述的是传输方式。

2. 从网络块组装事件，再拼接答案

一次 read() 可能只得到 data: {"del，也可能把两个事件一次性带回来。SSE 的空行才标志事件边界；data: 行的内容还可能需要进一步按提供方约定解析。HTML 标准允许多个 data: 行、不同换行形式、event、id 和注释；这里简化程序只实现 LF 且每事件恰好一条 data: ，目的是集中观察分块问题，不是完整 SSE 兼容实现。

Agent 流式界面可把 delta 文本暂时显示出来，但工具调用参数须等完整对象、结束状态和本地校验后才交给工具。用户取消连接也不证明服务端操作已取消；若之前已有写入请求，还要按重试与未知结果的规则核查。若需要断线续传，事件 ID、重连策略和去重是产品协议的一部分，不能假设网络自动提供“恰好一次”。

3. 运行最小分帧实验

把 stream_lab.py、test_stream_lab.py 按下方代码逐个保存到同一个新文件夹。在资源管理器打开文件夹，地址栏输入 powershell，再运行：

macOS：先在 Python 篇的学习文件夹中打开终端，执行 source .venv/bin/activate。再输入 cd ，把当前例子的文件夹拖入终端并回车，运行下面的命令。

保存为 stream_lab.py：

"""把分段收到的文字拼成完整SSE事件，再读取每段回答。"""
import json


def parse_events(chunks):
    """读取每个事件的一行data内容，用两个换行符识别事件结束。"""
    buffer = ""
    events = []
    for chunk in chunks:
        # chunk 是这次收到的一段文字；先累积，等事件完整后再解析。
        buffer += chunk
        while "\n\n" in buffer:
            frame, buffer = buffer.split("\n\n", 1)
            lines = frame.split("\n")
            data = [line[6:] for line in lines if line.startswith("data: ")]
            if len(data) != 1:
                raise ValueError("expected one data line")
            events.append(json.loads(data[0]))
    return events, buffer


def consume(chunks):
    events, pending = parse_events(chunks)
    if pending:
        return {"status": "incomplete", "visible": "", "events": len(events)}
    parts = []
    for event in events:
        if not isinstance(event, dict) or set(event) != {"delta"} or not isinstance(event["delta"], str):
            return {"status": "invalid", "visible": "", "events": len(events)}
        parts.append(event["delta"])
    # 将各事件的delta（新增文字片段）按顺序拼接成显示的回答。
    return {"status": "complete", "visible": "".join(parts), "events": len(events)}


if __name__ == "__main__":
    wire = 'data: {"delta":"你"}\n\ndata: {"delta":"好"}\n\n'
    print(json.dumps({"split_chunks": consume([wire[:8], wire[8:23], wire[23:]]),
                      "truncated": consume([wire[:-1]])}, ensure_ascii=False, indent=2))

# 核对SSE、WebSocket 与流式响应的正常结果和错误分支；断言失败时会显示对应的测试。
import unittest
from stream_lab import consume, parse_events


class StreamLabTests(unittest.TestCase):
    def test_arbitrary_chunk_boundary(self):
        wire = 'data: {"delta":"你"}\n\ndata: {"delta":"好"}\n\n'
        self.assertEqual(consume([wire[:1], wire[1:18], wire[18:]])["visible"], "你好")

    def test_truncated_frame_is_not_success(self):
        self.assertEqual(consume(['data: {"delta":"你"}\n'])["status"], "incomplete")

    def test_invalid_shape(self):
        self.assertEqual(consume(['data: {"tool":"run"}\n\n'])["status"], "invalid")

    def test_json_root_must_be_object(self):
        # JSON 可以合法解析成数字、列表或 null；它们仍不是回答事件对象。
        for value in ('3', '[]', 'null', '"delta"'):
            with self.subTest(value=value):
                self.assertEqual(consume(['data: ' + value + '\n\n'])["status"], "invalid")

    def test_malformed_json(self):
        with self.assertRaises(ValueError):
            parse_events(['data: {oops}\n\n'])

    def test_multiple_data_lines_out_of_subset(self):
        with self.assertRaises(ValueError):
            parse_events(['data: 1\ndata: 2\n\n'])


if __name__ == "__main__":
    unittest.main()

ls
python stream_lab.py
python -m unittest -v test_stream_lab.py

找不到 Python 时先按 Python 环境篇检查安装和完整版本；找不到文件时用 Get-ChildItem 核对目录。预期 split_chunks 为 complete、可见文本 你好、事件数 2；这里的 complete 只表示传入这次教学函数的字符串末尾没有未闭合帧，不能当作真实服务已发结束信号。truncated 为 incomplete，可见文本保持空。六项测试还覆盖任意分块、错误对象形状、合法 JSON 的非对象根值、损坏 JSON 和超出教学子集的多行 data:。

先读 parse_events：每个 chunk 追加到 buffer，遇到两个连续换行符才取出完整帧。把主程序 "split_chunks" 后的 consume([wire[:8], wire[8:23], wire[23:]]) 精确替换为 consume(list(wire))，保持 wire 定义不变。list(wire) 每次提供一个字符，重跑仍应是 complete/你好/2。

随后 consume 检查每个事件只含字符串 delta，最后一次性返回结果；它没有在每个 chunk 到达时向界面返回预览，也没有网络或完成事件。把 wire 里的第一项改成 {"tool":"run"} 会得到 invalid；删除最后一个换行会得到 incomplete，visible 为空。恢复后再跑测试。

从分帧函数扩到聊天界面

刚接触流式输出时，最容易把“用户已经看到文字”和“程序已经得到结果”混成一个 done。可以把状态拆开：

这张表是下一步应用设计，随文函数尚未实现这些状态。尤其是最后一行：关闭浏览器会断开连接，已发出的远端写入仍要核查。对只读聊天可丢弃未完成回答；支付、退款、发邮件须另查执行状态。

沿下方路径读图时，complete 只表示本地分帧函数读到了完整输入；真实服务还必须给出供应商定义的完成信号。随后应用要校验对象并检查授权，不能直接把一个局部 delta 传给写工具。

看两个具体例子。例一：一个 data: 事件被拆成十个网络块，客户端逐块 json.loads 会失败，但正确的分帧器应继续等待空行。例二：流里已经显示 {"recipient":"A"，用户立刻点击取消；此时没有闭合对象，也没有结束信号，不能因为“看起来像收件人 A”就发送邮件。这两个例子分别检验传输边界和副作用边界，不能靠单一的“能流式显示”测试覆盖。

4. 接入网络服务

完整 SSE 客户端还要处理不同换行符、多 data 行、事件名、注释、事件 ID、重连、字节编码和超时。接入网络服务时，使用符合协议的库，并按供应商定义的完成与错误事件更新状态。WebSocket 也要定义消息边界、心跳、断线恢复和接收速度上限；MDN 的 WebSocket 文档提醒经典浏览器 API 本身不提供背压机制。

这张时间线解释分块与完成信号。要比较流式请求的速度，需要在真实网络调用中分别测首次可见时间、完整回答时间、取消后的资源释放、错误恢复和成本，并记录模型与网络环境。

自查与下一站

一次 read() 得到的 chunk 可能只含半个事件；先按空行组出完整 frame，再读取里面的 delta。把事件拆成任意小块，最终仍应得到 你好；删掉末尾空行则应保持 incomplete，等待完整参数并通过校验后再调用工具。

上一站：解析失败、重试与回退。下一站：模型选择与路由。

现有练习

资料

参考资料：WHATWG HTML：Server-sent events、MDN：EventSource、MDN：WebSocket API。