从一次“漫长”的等待说起

一、从一次“漫长”的等待说起

当你与一个大型语言模型（LLM）对话时，最直观的体验就是等待。你发送一个问题，然后……等待。服务器需要耗费数秒乃至更长时间生成完整的回答，才能一次性将所有内容返回给你的浏览器或应用。这段时间里，用户面对的是空白的界面或一个加载动画，体验上无疑是“漫长”且不确定的。

流式输出（Streaming） 正是为解决这一痛点而生。它的核心思想是：不等待模型生成完毕，而是一边生成，一边将已生成的部分数据“流”式地推送给客户端。这就像你看视频网站的在线播放，视频数据是持续不断地下载并播放的，而不是等整个文件下载完才能开始看。在大模型领域，这通常表现为回答的文字逐字或逐句地出现在屏幕上，为用户提供了实时的反馈和更佳的交互体验。

二、SSE：流式通信的“高速公路”

要实现服务端向客户端持续推送数据，我们需要一个合适的通信协议。SSE（Server-Sent Events） 就是专为此类场景设计的轻量级协议。它基于标准的 HTTP 协议，是“单向”的：只有服务端能向客户端发送事件流，客户端通过一个持久的 HTTP 连接来接收。

SSE 的优势非常明显：

轻量简单：实现起来比 WebSocket 更简单，因为它就是标准的 HTTP。
自动重连：浏览器原生支持 EventSource API，内置了连接中断后的自动重连机制。
文本友好：数据格式是纯文本，非常适合传递大模型生成的自然语言文本。
兼容性好：几乎所有现代浏览器和 HTTP 服务器都支持。

一个典型的 SSE 响应，其 Content-Type 头被设置为 text/event-stream。服务器发送的数据遵循固定的格式，例如：

data: 这是第一条消息\n\n
data: {"token": "你好", "is_end": false}\n\n
data: 这是最后一条，后面会跟两个换行符表示结束\n\n

客户端（如浏览器中的 EventSource 对象）会持续监听这个连接，每当收到一个以 \n\n 结尾的数据块时，就会触发一个事件。

三、MiMo 的流式输出协议解析

MiMo 作为小米的智能大模型，其 API 自然支持流式输出。它的流式输出正是基于 SSE 协议实现的。当你发起 API 请求并将 stream 参数设置为 true 时，服务器会返回一个 SSE 事件流。

在该事件流中，每一条事件（即每一个 data: 行）通常是一个 JSON 对象，包含了模型当前生成的“一块”内容。关键字段如下：

0：这是最核心的字段，存放着模型在当前这一步新生成的文本片段。它可能是一个词、一个字或一个标点。
0：这是一个可选字段。当模型的生成结束时，最后一个事件中此字段通常会设置为 stop，标识一次完整的回答已传输完毕。

整个交互流程是：客户端发送请求 → 服务器保持连接打开 → 服务器按序发送多个 data 事件 → 客户端依次处理并显示 delta.content → 服务器发送一个表示结束的 finish_reason 事件。

四、客户端实战：用 Python 消费流式数据

理解了协议，我们来看看如何在代码中实际接收和处理流式数据。这里以 Python 的 requests 库为例（使用 stream=True 参数）。

import requests
import json

api_url = "https://api.mimo.example.com/v1/chat/completions"
headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer YOUR_API_KEY"
}
payload = {
    "model": "MiMo-chat",
    "messages": [{"role": "user", "content": "用三句话介绍量子计算"}],
    "stream": True  # 关键参数：开启流式输出
}

response = requests.post(api_url, headers=headers, json=payload, stream=True)

full_content = ""
for line in response.iter_lines():
    if line:
        line = line.decode('utf-8')
        # SSE 数据以 'data: ' 开头
        if line.startswith('data: '):
            data_str = line[6:]  # 去掉 'data: ' 前缀
            if data_str.strip() == '[DONE]':  # 有些API会用[DONE]标记结束
                break
            try:
                data_obj = json.loads(data_str)
                delta = data_obj['choices'][0]['delta']
                if 'content' in delta:
                    token = delta['content']
                    print(token, end='', flush=True)  # 实时打印，flush确保立即输出
                    full_content += token
            except json.JSONDecodeError:
                continue  # 忽略可能的不完整行

print("\n\n--- 完整回答 ---")
print(full_content)

提示：许多云服务 SDK（如 OpenAI 的）已经对流式输出做了高度封装，你可能只需在创建客户端时传入 stream=True，然后使用一个简单的迭代器（如 for chunk in response）即可，无需手动解析 data: 行。上述代码展示了底层原理，有助于理解协议本身。

五、优缺点与适用场景

使用流式输出并非没有代价，它和批量响应是两种不同的设计选择。

流式输出的优点：

用户体验极佳：显著减少用户感知的等待时间（Time To First Token, TTFT）。
实时反馈：适合构建对话、实时协作、内容生成等交互式应用。
资源利用更平滑：服务器无需一次性准备巨量的内存来缓存整个回答。

流式输出的缺点与挑战：

客户端处理复杂：需要持续监听、拼接数据，并处理网络中断等异常。
连接管理：服务端需要维持更多的持久 HTTP 连接，对服务器资源有一定要求。
错误处理：在网络传输中，部分数据已发送，若连接中断，需要决定如何重试或恢复。

因此，流式输出非常适合：聊天机器人、实时翻译、代码补全、搜索实时联想等需要即时反馈的交互场景。而对于后台任务、批量处理或需要完整响应才能进行后续操作的场景，使用一次性响应可能更简洁。

六、进阶思考：错误处理与并发

在实际生产环境中使用流式输出，还需要考虑更多细节。

首先是错误与中断处理。流式连接可能因为网络波动而中断。一个健壮的客户端应该实现重连机制（EventSource 内置了）或能够从最后断点处尝试恢复。同时，要妥善处理服务端返回的错误事件。

其次是并发与背压。如果客户端处理数据的速度慢于服务器发送的速度（例如，前端渲染过慢），会导致数据在缓冲区堆积，这就是“背压”。服务器和客户端需要某种机制（如流量控制）来协调，避免内存溢出。在大模型流式场景下，由于生成速度通常不是极快，这个问题相对不突出，但在设计高吞吐系统时不容忽视。

最后，对于进度感知与中断，流式协议也可以用来承载更丰富的元信息，例如当前生成到哪个段落、预估剩余时间等，这些都可以通过自定义的 SSE 事件来发送，从而让前端界面能提供更友好的进度条或状态提示。