什么是“流式输出”？ - 代码学习笔记

一、什么是“流式输出”？

在传统的 HTTP 请求响应模型中，客户端发送一个请求，服务器在准备好所有数据后，将完整响应返回给客户端。这就像点一份需要30分钟准备的大餐，你必须等到厨师完全做好，服务员才会把整盘菜一次性端到你面前。

流式输出则完全不同，它更像是在吃一顿精心准备的“omakase”。厨师（AI模型）每完成一道精致的料理（生成一小段文本），服务员（服务器）就立刻端到你面前（推送给客户端）。这种“边生成边推送”的模式，在与大型语言模型（LLM）交互时优势巨大。

为什么需要流式输出？ LLM 生成一个完整回答可能需要几秒到几十秒。如果采用传统模式，用户只能面对一个加载中的圈圈干等，体验非常糟糕。流式输出能让用户立即看到模型的“思考”过程，一个字一个字地输出，大大改善了交互的流畅感和实时感。
它的本质是什么？ 从技术上看，流式输出是一种保持 HTTP 连接持续打开，并多次发送数据片段的技术。服务器不再“憋大招”，而是“现做现发”。

提示：流式输出主要优化的是用户体验（减少感知延迟）和交互效率（无需等待完整响应即可开始处理）。它并不能让模型的实际生成速度变快。

二、揭秘 SSE：Server-Sent Events 协议

要实现上述的“现做现发”，就需要一个规范来告诉客户端：“嘿，连接我保持打开了，等会儿有数据我会一小块一小块地发给你。” 这个规范就是 SSE（Server-Sent Events）。

SSE 是一种基于 HTTP 的、服务端向客户端单向推送事件流的协议。它不是一个全新的、复杂的协议，而是对 HTTP 的巧妙利用。服务器在响应头里声明 Content-Type: text/event-stream，然后持续发送格式化的文本流。客户端可以通过浏览器内置的 EventSource API 或任何支持 HTTP 流的库来轻松接收。

SSE 的每条消息格式非常简单，由字段和值组成，用换行符分隔。最常见的格式如下：

data: 这是第一条消息内容\n
\n
data: 这是第二条消息内容\n
event: 更新\n
data: {"key": "value"}\n
\n

每条消息以一个或多个字段开头，每行一个 field: value。
data: 字段是必须的，它承载消息的主要内容。
连续两个换行符 \n\n 标志着一条消息的结束。

三、MiMo 如何与 SSE 结合工作？

MiMo（小米自研大模型）的 API 提供了流式输出能力，其背后的技术支撑正是 SSE 协议。当你在 API 请求中设置 "stream": true 时，就触发了这个流程。

整个过程可以概括为：

客户端发起请求：带上 "stream": true 参数。
服务器改变响应模式：不再等待模型生成完毕，而是立即以 text/event-stream 头响应该连接，并保持连接打开。
数据推送：MiMo 模型每生成一小段文本（例如，一个词或一个句子），服务器就将其封装成一个 data 事件（可能以 delta 的形式出现），通过 SSE 推送给客户端。
客户端接收并拼接：客户端持续监听流，每当收到一个新的 data 事件，就将其中的文本追加到已显示的内容后面，给用户“打字机”般的输出效果。

关键点：SSE 是单向的（服务端 -> 客户端）。在对话场景中，客户端的每一轮新提问仍然是一个独立的 HTTP 请求，而模型的回答则通过上一次请求的 SSE 流返回。

四、实战代码示例

理论说再多，不如代码一行直观。下面是一个使用 Python requests 库接收 MiMo 流式输出的极简示例。请注意，你需要将 your_api_key 替换为你的真实密钥。

import requests
import json

# 模拟一个 API 端点（实际请使用 MiMo 官方文档提供的地址）
api_url = "https://api.example.com/mimo/chat"
headers = {
    "Authorization": "Bearer your_api_key",
    "Content-Type": "application/json"
}

payload = {
    "prompt": "用简单的话解释什么是量子计算。",
    "stream": True  # 关键：启用流式输出
}

# 使用 requests 发起流式请求
response = requests.post(api_url, headers=headers, json=payload, stream=True)

# 检查响应头是否是 event-stream
if response.headers.get('Content-Type') == 'text/event-stream':
    print("AI回答：", end="", flush=True)
    # 迭代读取流中的每一行
    for line in response.iter_lines():
        if line: # 忽略空行
            decoded_line = line.decode('utf-8')
            # 简单的事件解析（生产环境需更健壮）
            if decoded_line.startswith('data:'):
                data_str = decoded_line[5:].strip() # 取出 “data: ” 后面的JSON字符串
                if data_str != '[DONE]': # 流结束标记
                    try:
                        data_json = json.loads(data_str)
                        # 假设API返回的增量文本在 ['choices'][0]['delta']['content']
                        token = data_json.get('choices', [{}])[0].get('delta', {}).get('content', '')
                        print(token, end="", flush=True) # 逐token打印，不换行
                    except json.JSONDecodeError:
                        # 处理非JSON数据或结束标记
                        pass
    print() # 流结束后换行
else:
    print("请求失败或非流式响应:", response.text)

这个脚本清晰地展示了客户端如何“被动”地接收流：建立连接后，通过循环不断检查是否有新数据到达，并实时处理、输出。

五、使用流式输出的注意事项

在享受流式输出带来的流畅体验时，有几点需要在开发中特别注意：

错误处理变得复杂：网络中断、服务器错误可能在流传输的中途发生，而不是在开始前。客户端需要优雅地处理连接中断，向用户提示“回答中断”。
前端渲染成本：频繁地更新DOM（将新收到的文本拼接到页面上）可能会对性能造成压力，尤其是在移动端。可以考虑使用防抖（debounce）技术或批量更新。
完整性校验：流的最后通常会有一个特殊的结束标记（如 [DONE]）。你需要确保正确检测到这个标记，以确认回答已完整接收，避免 UI 留在“正在输入”的状态。
取消请求：如果用户想停止生成，仅仅关闭页面或标签页并不一定能立即终止服务器端的模型计算。理想的API应提供一个中断流或取消任务的端点。

六、与其他技术的简单对比

除了 SSE，实现流式输出或实时推送还有其它方式，了解其区别有助于做出合适的技术选型：

HTTP 长轮询：客户端定期询问服务器“有新数据吗？”，效率低且不实时，与 SSE 的“推送”模式有本质区别。
WebSocket：提供真正的全双工通信通道，客户端和服务端可以随时互相发送数据。适用于需要复杂、高频双向交互的场景（如实时游戏、协同编辑）。对于 AI 对话这种主要是“服务端推、客户端收”的单向流场景，SSE 更简单、更轻量、更贴合其语义，且天然支持HTTP基础设施（认证、缓存、代理等）。
gRPC Streaming：基于 HTTP/2，性能很高，但客户端库相对复杂，主要用于后端服务间的通信，在浏览器端支持需要额外网关。

对于 MiMo 的对话 API，选择 SSE 是一个在功能、复杂度和兼容性之间取得良好平衡的明智决策。它让我们能用最简单的 HTTP 知识，构建出最流畅的交互体验。