什么是流式输出 (Streaming)？

一、什么是流式输出 (Streaming)？

在大语言模型（LLM）的上下文中，流式输出是一种响应生成模式。它与我们常见的“一次性返回全部结果”的模式（即非流式）相对立。具体来说，当用户向模型（如 MiMo）发送一个请求后，服务器不会等到完整回答全部生成完毕，而是逐个 token（词元）或一小批 token 地生成，并立刻通过已建立的 HTTP 连接推送回客户端。客户端可以实时接收并逐步显示这些内容。

这种体验就像在即时聊天或打字机上打字：你问一个问题，对方一边思考一边开始回答，你马上就能看到第一个字，然后是第二个字……直到整个回答完成。这极大地优化了用户的等待体验，尤其是对于需要生成长文本的复杂请求。

二、为什么我们需要流式输出？

流式输出的核心价值在于降低感知延迟和提升交互体验。大模型的生成过程在计算上是非常耗时的。如果采用非流式方式，用户必须等待整个回答（可能长达数秒甚至数十秒）都生成并传输完成后才能看到任何内容，期间只能面对一个加载图标。

更好的用户体验：用户几乎立即看到反馈，感觉应用响应迅速、“有生命”。
适用于特定场景：对于需要实时交互的聊天机器人、代码自动补全、文章实时撰写辅助等场景，流式输出是不可或缺的。
资源利用与中断：客户端可以在流式传输过程中根据业务逻辑随时中断请求，避免浪费后端计算资源去生成一个已无用的完整回答。

三、SSE：实现流式的“高速公路”

流式输出需要一个可靠的协议来支持这种“服务器持续向客户端推送数据”的模式。HTTP 协议本身主要是“请求-响应”模式。SSE (Server-Sent Events) 正是为此设计的一项 Web 标准。

你可以把 SSE 看作一个单向、持久化的 HTTP 连接。客户端发起一个普通的 HTTP 请求，但服务器通过特定的响应头（Content-Type: text/event-stream）告知客户端：“接下来我会不断发数据给你，别关连接”。之后，服务器就可以通过这个打开的连接，以特定的文本格式（事件流）不断发送数据块，直到任务完成。

SSE 基于纯 HTTP，无需额外协议（如 WebSocket），因此实现简单、穿透性强，且天然支持自动重连，非常适合用于大模型的文本流式传输。

四、MiMo 的流式输出请求与响应

向 MiMo API 发起流式请求非常简单。你只需要在标准的 API 请求体中，将 stream 参数设置为 true。

import requests
import json

url = "https://api.example.com/v1/chat/completions"
headers = {
    "Authorization": "Bearer YOUR_API_KEY",
    "Content-Type": "application/json"
}
data = {
    "model": "MiMo-8B",
    "messages": [{"role": "user", "content": "用Python写一个快速排序算法"}],
    "stream": True  # 关键参数：开启流式输出
}

response = requests.post(url, headers=headers, json=data, stream=True)

当 stream=True 时，服务器的响应内容类型将变为 text/event-stream。响应体不再是完整的 JSON，而是一个由多个事件（Event） 组成的流。每个事件通常以 data: 前缀开头，其值是一个 JSON 对象，包含了新生成的 token 信息（如 choices[0].delta.content）。最后一个事件的 data 字段通常为 [DONE]，表示流结束。

五、代码实战：处理 SSE 流

获取到流式响应后，我们需要逐行读取并解析。requests 库的 response.iter_lines() 方法非常适合此场景。

full_response = ""
for line in response.iter_lines():
    if line:
        decoded_line = line.decode('utf-8')
        # 忽略以“:”开头的行（SSE注释）和空行
        if decoded_line.startswith('data:'):
            json_str = decoded_line[len('data:'):].strip()
            if json_str == '[DONE]':
                break
            try:
                chunk = json.loads(json_str)
                delta = chunk['choices'][0]['delta']
                # 逐步拼接内容
                if 'content' in delta:
                    content_piece = delta['content']
                    print(content_piece, end='', flush=True) # 实时打印
                    full_response += content_piece
            except json.JSONDecodeError:
                continue

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

提示：在实际生产代码中，你需要更健壮的错误处理，例如处理网络中断、超时、JSON 解析失败等情况。同时，考虑使用专门的 SSE 客户端库（如 sseclient-py）可以简化很多底层细节。

六、调试技巧与注意事项

开发和调试流式 API 时，有一些实用技巧：

使用命令行工具测试：可以先用 curl 快速验证流是否正常工作，例如 curl -N -X POST ... -d '{"stream": true}'。-N 参数禁用缓冲，能立刻看到流式输出。
关注连接超时：流式连接是长连接，需确保客户端和代理（如 Nginx）的超时时间设置足够长。
错误处理：流式传输中，错误信息也可能作为事件流的一部分发送，需要正确识别和解析，而非等到流结束。

重要提示：流式输出并不意味着模型的生成速度变快了，它只是改变了数据“到达”用户的时间顺序。整体任务完成的总耗时可能与非流式相当，甚至因为连接保持略有开销而稍长。它的核心优势在于交互体验。

七、总结

理解 MiMo 的流式输出，关键在于把握三个层次：为什么（降低延迟，优化体验）、是什么（逐 token 发送，SSE 协议）、以及怎么用（请求设置 stream: true，响应解析事件流）。SSE 作为一项成熟、简单的 Web 标准，为构建实时响应的 AI 应用提供了坚实的基础。掌握流式输出的处理，是从开发“能用”的AI应用走向开发“好用”、“体验流畅”的AI应用的重要一步。