第一次读聚合站的 SSE 流：data:、[DONE] 和中间那些 chunk

为什么需要读懂原始流

调用 LLM API 时，流式响应（streaming）能让你在模型生成的同时逐步展示内容。但很多开发者直接使用 SDK 封装好的迭代器，遇到超时、截断或格式异常时无从下手。理解原始 SSE 流的结构，是排查这类问题的第一步。

聚合站的流式接口与主流 LLM 供应商（如 OpenAI、Anthropic）的格式兼容，本文以常见的 chat completions 流为例，逐行拆解。

SSE 基础：data: 帧是什么

SSE（Server-Sent Events）是一种基于 HTTP 的文本协议，服务端通过持续发送 data: 开头的行来推送事件。一个典型的流式响应片段如下：

data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4","choices":[{"index":0,"delta":{"content":"你"},"finish_reason":null}]}

data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4","choices":[{"index":0,"delta":{"content":"好"},"finish_reason":null}]}

data: [DONE]

每一行 data: 后面是一个 JSON 对象（或特殊的终止标记 [DONE]），行与行之间用空行分隔。客户端逐行读取，解析 data: 后的内容即可。

注意：data: 和 JSON 之间通常有一个空格，但解析时建议忽略空格，直接取 data: 之后的子串。

单个 chunk 里装了什么

每个 data: 帧的 JSON 结构一般包含以下字段：

• id：本次请求的唯一标识，所有 chunk 共享同一个 id。
• object：固定为 chat.completion.chunk，表明这是流式分块。
• created：时间戳。
• model：实际使用的模型名称。
• choices：一个数组，通常只有一个元素（index 为 0），包含 delta 和 finish_reason。

delta 对象

delta 是增量内容，可能包含：

• role：仅在第一个 chunk 中出现，通常是 assistant。
• content：模型生成的文本片段。可能为空字符串，也可能不存在（例如只包含 role 的首个 chunk）。
• toolcalls 或 functioncall：如果模型决定调用工具，会在这里逐步输出工具调用的参数。

一个只包含 role 的 chunk 示例：

data: {"choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}

后续 chunk 通常只包含 content：

data: {"choices":[{"index":0,"delta":{"content":"Hello"},"finish_reason":null}]}

finish_reason

finishreason 在流结束前会出现，表示停止原因，常见值有 stop（自然结束）、length（达到最大 token 数）、toolcalls（模型调用了工具）。在大多数 chunk 中它为 null。

流的正常结束：[DONE] 与 finish_reason

一个正常的流式响应会以两种方式之一结束：

1. 收到 data: [DONE]：这是 OpenAI 兼容接口的显式终止标记，表示没有更多数据。
2. 收到带有非 null finish_reason 的 chunk，随后连接关闭。

例如最后一个内容 chunk 可能如下：

data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

然后服务端发送 data: [DONE] 并关闭连接。

判断流正常结束的关键：

• 如果你收到了 [DONE]，或者收到了 finish_reason 非 null 的 chunk，并且 HTTP 连接正常关闭，那么流是完整的。
• 如果连接在未收到 [DONE] 且未收到 finish_reason 的情况下断开，则流可能被截断（网络问题、服务端超时等）。

如何判断流是卡住了还是正常

卡住（hang）通常表现为连接保持打开但长时间没有新数据。你可以通过以下方式判断：

• 设置读超时：在读取每个 chunk 时设置一个合理的超时时间（例如 30 秒），超时则主动断开并重试。
• 监控心跳：部分服务端会发送注释行（以 : 开头）作为心跳，但并非所有实现都有。不要依赖心跳判断结束。
• 结合 finishreason：如果已经收到 finishreason 但连接未关闭，可能是服务端未及时发送 [DONE]，可以主动关闭连接。

一个简单的处理逻辑：

import json
import requests

resp = requests.post(url, json=payload, stream=True, timeout=(10, 30))
for line in resp.iter_lines():
    if not line:
        continue
    if line.startswith(b'data: '):
        data = line[6:]
        if data == b'[DONE]':
            break
        chunk = json.loads(data)
        delta = chunk['choices'][0]['delta']
        content = delta.get('content', '')
        if content:
            print(content, end='', flush=True)
        finish_reason = chunk['choices'][0].get('finish_reason')
        if finish_reason:
            # 流即将结束，可以选择继续读取直到 [DONE] 或直接退出
            pass

常见异常帧

• 错误帧：某些服务端在出错时会发送 data: {"error": {...}}，而不是正常的 chunk。需要检查 error 字段。
• 空行：SSE 用空行分隔事件，解析时跳过空行即可。
• 多行 data：标准 SSE 允许一个事件有多行 data:，但 LLM 流式接口通常每行一个完整 JSON，简单按行处理即可。

小结

读懂原始 SSE 流的关键在于：识别 data: 帧、解析 delta 内容、通过 [DONE] 或 finish_reason 判断结束、用超时机制应对卡住。聚合站的流式响应遵循相同模式，掌握这些细节后，你就能更自信地处理流式输出中的各种边界情况。