什么是流式输出与 SSE？ - 代码学习笔记

一、什么是流式输出与 SSE？

想象一下，你在和一个 AI 助手聊天。如果你问一个复杂的问题，传统的“一次性”响应方式会让你盯着空白的界面等待，直到整个答案生成完毕，这种体验并不好。流式输出（Streaming）就是为了解决这个问题而生的：它让 AI 的回复像打字一样，一个词、一个词地实时“流”到你的屏幕上。这极大地提升了用户体验的感知速度，用户能立刻看到反馈，即使完整回答还需要几秒钟。

实现这种“打字机”效果，底层需要一种机制来让服务器持续地向客户端推送数据。这就是 SSE（Server-Sent Events）协议的用武之地。SSE 是一种基于 HTTP 的标准协议，允许服务器通过一个长连接，主动、单向地向客户端发送一系列事件。它不是 WebSocket 那种双向通信，对于 AI 回答这种“服务器推送给客户端”的单向场景，SSE 更简单、更轻量，且天然支持自动重连。

二、MiMo 为什么选择流式与 SSE？

对于大语言模型（如 MiMo）的应用来说，流式输出几乎是标配。这不仅是为了用户体验的“快”，更是为了交互设计的“流畅”。当模型在推理、生成长文本时，流式输出能让用户立即开始阅读已生成的部分，减少焦虑感。同时，客户端可以提前开始处理部分内容（如渲染 Markdown 格式），实现无缝的视觉体验。

SSE 协议在这里扮演了最佳传输层角色。与传统的轮询（Polling）相比，SSE 避免了频繁建立和关闭连接的开销，效率更高。与 WebSocket 相比，SSE 实现起来更简单，只需标准的 HTTP 服务器支持，并且它专注于“服务器到客户端”的单向通信，完美匹配了 AI 生成文本的场景需求。MiMo 的 API 选择 SSE，是轻量、高效、标准的工程实践体现。

三、SSE 协议的工作原理简述

理解 SSE 的原理，有助于我们更好地调试和处理流式数据。一个 SSE 连接始于客户端发起的一个普通的 HTTP GET 请求，并在请求头中指定 Accept: text/event-stream。服务器会保持这个连接不关闭，并以特定的文本格式持续发送数据流。

SSE 事件流的每一则消息都由若干行组成，以 \n\n 结尾。每一行可以是一个字段，最常见的是：

data:：携带事件的实际数据内容，可以跨多行。
event:：指定事件的类型名，客户端可以据此监听特定类型的事件。
id:：为事件设置一个 ID，用于客户端的断线重连和事件跟踪。
retry:：建议客户端下次重连的等待时间（毫秒）。

例如，一个 MiMo 的流式响应片段可能看起来像这样：

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"你"},"index":0}]}

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"好"},"index":0}]}

data: [DONE]

其中，data: [DONE] 是一个特殊标记，表示整个流式响应结束。

四、如何用 Python 代码处理 MiMo 的流式响应？

下面我们来看一个具体的代码示例，展示如何使用 requests 库来接收并处理 MiMo 的流式 SSE 响应。关键点在于设置 stream=True 并迭代读取响应内容。

import requests
import json

def stream_mimo_response(prompt, api_url, api_key):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}",
        "Accept": "text/event-stream"  # 关键：声明接受SSE
    }
    payload = {
        "model": "mimo-chat",
        "messages": [{"role": "user", "content": prompt}],
        "stream": True  # 关键：开启流式模式
    }

    # 发起请求，stream=True 表示不立即下载全部响应
    with requests.post(api_url, headers=headers, json=payload, stream=True) as response:
        response.raise_for_status()
        
        # 逐行读取SSE事件流
        full_response = ""
        for line in response.iter_lines():
            if line:
                # 解码并移除行首的 `data: ` 前缀
                decoded_line = line.decode('utf-8')
                if decoded_line.startswith('data: '):
                    event_data = decoded_line[6:]  # 去掉前缀
                    if event_data.strip() == '[DONE]':
                        break  # 收到结束信号，跳出循环
                    
                    try:
                        chunk = json.loads(event_data)
                        # 从OpenAI格式兼容的数据中提取文本片段
                        content = chunk['choices'][0]['delta'].get('content', '')
                        if content:
                            full_response += content
                            print(content, end='', flush=True)  # 实时打印，模拟打字机效果
                    except json.JSONDecodeError:
                        continue  # 忽略无法解析的行（如空行或注释）

        print()  # 打印完成后换行
        return full_response

# 使用示例
# response = stream_mimo_response("给我讲一个关于AI的笑话", "https://api.example.com/v1/chat/completions", "your_api_key")

关键点解析：代码中的 response.iter_lines() 会一行一行地产生来自服务器的数据，直到连接关闭。我们需要自己解析 data: 前缀，并处理可能的空格和结束标记 [DONE]。

五、流式处理中的错误与连接管理

在实际应用中，流式连接并非永远稳定。网络波动、服务器重启都可能导致连接中断。一个健壮的流式客户端需要优雅地处理错误并实现重连。SSE 协议本身内置了 retry 字段来建议客户端的重连等待时间。

当使用 requests 等库时，捕获 ConnectionError 或在循环中检测到意外的数据中断（例如，长时间没有收到新数据），都应该触发重连逻辑。一个简单的策略是使用指数退避（Exponential Backoff）进行重试，避免频繁重试给服务器带来压力。

此外，在客户端（如浏览器）中，使用原生的 EventSource API 可以自动处理 SSE 的很多细节，包括自动重连。这也是为什么如果你在前端开发，直接使用 EventSource 会比自己手动管理 fetch 请求更方便可靠。

六、总结：流式体验背后的技术选型

回顾一下，MiMo 的流式输出核心在于：利用 SSE 协议，在单个 HTTP 长连接上，实现服务器向客户端的持续、低延迟数据推送。选择 SSE 是基于其标准化、简单高效的特性，完美契合了大模型文本生成的单向、顺序输出场景。

对于开发者而言，集成流式功能主要涉及两件事：

请求时声明：确保 API 请求中包含 stream: true 参数和对应的 Accept 头。
响应时解析：在客户端，需要对返回的文本事件流进行实时解析、拼接和渲染，并妥善处理连接生命周期事件。

掌握流式与 SSE，不仅是调用一个 API，更是理解现代、高交互性应用如何构建实时通信体验的关键一步。当你看到 AI 像思考一样逐字输出答案时，背后正是这些简洁而强大的协议在支撑。