一、 什么是流式输出？为什么大模型需要它？
当我们与像 MiMo 这样的大型语言模型（LLM）交互时，最直观的体验就是“等待”。传统模式下，我们需要耐心地等待模型完成所有文字的生成，然后一次性看到完整的回复。这种模式在生成长文本时，用户界面会显得“假死”，体验非常糟糕。流式输出（Streaming） 则是解决这一痛点的核心技术。
它的原理是：模型每生成一小段文本（通常是一个 token 或几个 token），就立即将这部分结果“推送”给客户端，客户端可以边接收边渲染。这就像看一场现场直播，你无需等待整场活动录完再看回放，而是实时地获取内容。对于大模型应用而言，流式输出能极大提升用户感知的响应速度，让交互过程更加流畅、自然。
二、 核心驱动力：SSE 协议详解
实现流式输出，在 HTTP 协议层面有多种选择，如 WebSocket、长轮询等。而对于服务器向客户端单向、实时地推送文本数据，SSE（Server-Sent Events） 是一项非常优雅且被广泛支持的标准。
SSE 是 HTML5 规范的一部分，它基于一个持久的 HTTP 连接。服务器可以通过这个连接，多次、分批地将数据以简单的文本格式发送给客户端，而客户端通过原生的 EventSource API 或一些库来轻松接收。它的优势在于：协议简单、基于 HTTP（无需特殊端口或协议升级）、浏览器原生支持、自动重连机制。对于 AI 模型的流式输出，这个场景完美契合——服务器（模型后端）作为数据源，持续不断地生成并“推送” token 流。
关键区别：与 WebSocket 的全双工通信不同，SSE 是单向的（仅服务器到客户端）。对于用户提问（客户端到服务器）这个动作，我们依然使用常规的 HTTP POST 请求。SSE 通道仅用于接收模型的流式回复。
三、 剖析一个 MiMo 流式 API 调用流程
一个典型的结合了普通请求和 SSE 流式响应的交互流程如下：
- 客户端发起请求：前端通过一个
POST请求（如/v1/chat/completions）将用户的聊天历史等参数发送给后端服务。 - 后端处理与流式转发：后端服务验证参数后，调用 MiMo 模型的流式推理接口。此接口会立即返回一个可迭代的文本流。
- 建立 SSE 连接：后端在接收到模型的第一个数据片段时，开始向客户端返回一个特定的 HTTP 响应。这个响应头中包含
Content-Type: text/event-stream，表明这是一个 SSE 流。 - 数据分帧推送：后端将模型生成的每个 token，按照 SSE 规范格式化后（如
data: {"content":"你"}\n\n），写入响应体并立即刷新，客户端的浏览器或请求库便会实时收到。 - 完成与关闭：当模型生成结束（输出
[DONE]标记或流自然结束）后，后端关闭连接。
四、 动手实践：用 Python 消费 MiMo 的 SSE 流
理解了原理，让我们看看代码层面如何实现。这里以 Python 为例，使用 requests 库来展示如何发起一个流式请求并处理响应。
import requests
import json
# 假设的 MiMo 流式 API 端点
api_url = "https://api.example.com/v1/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": "Bearer YOUR_API_KEY" # 认证令牌
}
# 请求体，注意 stream 参数设置为 True
payload = {
"model": "MiMo-chat",
"messages": [
{"role": "user", "content": "用生动的比喻解释量子纠缠是什么？"}
],
"stream": True # 这是关键！告诉服务器我们需要流式响应
}
# 发起流式请求（stream=True）
response = requests.post(api_url, headers=headers, json=payload, stream=True)
# 检查响应状态
if response.status_code == 200:
full_response = []
# iter_lines 会按行（line）迭代响应内容，完美匹配 SSE 的格式
for line in response.iter_lines():
if line:
# 移除 SSE 数据的 “data: ” 前缀
decoded_line = line.decode('utf-8').strip()
if decoded_line.startswith('data: '):
data_str = decoded_line[6:] # 获取 “data: ” 后面的JSON部分
if data_str == "[DONE]":
print("\n[流式传输结束]")
break
try:
data = json.loads(data_str)
content = data['choices'][0]['delta'].get('content', '')
if content:
# 实时打印模型生成的每个token，实现“打字机”效果
print(content, end='', flush=True)
full_response.append(content)
except json.JSONDecodeError:
# 可能遇到 [DONE] 之外的非JSON行（如注释），简单忽略
pass
final_answer = ''.join(full_response)
print(f"\n\n完整回答：{final_answer}")
else:
print(f"请求失败，状态码：{response.status_code}")五、 前端如何监听：一个简单的 JavaScript 例子
前端部分使用浏览器原生的 EventSource API（对于 SSE 流）或 fetch API（支持流式读取）即可。下面是一个用 fetch 和 ReadableStream 的现代写法示例。
async function streamChat(prompt) {
const response = await fetch('/v1/chat/completions', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
messages: [{ role: 'user', content: prompt }],
stream: true // 启用流
})
});
// 获取可读流（ReadableStream）的读取器
const reader = response.body.getReader();
const decoder = new TextDecoder();
let result = '';
while (true) {
// 读取流中的下一个数据块
const { done, value } = await reader.read();
if (done) break;
// 解码二进制数据为文本
const chunk = decoder.decode(value, { stream: true });
// 这里需要解析SSE格式的文本块，提取 “data: ” 后面的JSON
const lines = chunk.split('\n');
lines.forEach(line => {
if (line.startsWith('data: ') && line !== 'data: [DONE]') {
try {
const data = JSON.parse(line.substring(6));
const content = data.choices[0]?.delta?.content || '';
// 将内容拼接或直接追加到DOM元素中
result += content;
// 例如：document.getElementById('output').textContent += content;
console.log(content); // 模拟实时输出
} catch (e) { /* 解析错误则忽略 */ }
}
});
}
return result;
}六、 关键要点与最佳实践总结
- 0 标志位：在请求体中显式设置
stream: True是触发流式输出的开关。没有它，模型将按同步模式返回。 - 0 ：服务器返回的
Content-Type必须是这个值，客户端才能正确地以 SSE 流来处理。 - 数据格式：SSE 事件通常以
data:开头，后跟一个 JSON 字符串。JSON 中常包含delta字段，里面是增量生成的content。 - 结束标记：流结束时，服务器通常会发送一个特殊的
data: [DONE]事件，客户端应据此关闭连接并处理最终结果。 - 用户体验：流式输出不仅是技术优化，更是产品设计。它让AI的“思考过程”可见，减少了用户的焦虑感，是构建拟人化、高响应性AI应用的基石。
提示：在生产环境中，务必处理好网络中断、服务端错误等情况。SSE 协议本身有内置的重连机制，但在客户端也需要考虑适当的超时设置和错误恢复逻辑，以确保应用的健壮性。