一、从一次对话的“等待”说起

你是否曾在与 AI 对话时，盯着屏幕等待那个加载圆圈转了好久，然后“唰”地一下，一大段文字一次性蹦出来？这就是传统的“请求-响应”模式。用户发送一个完整的问题，后端模型进行推理，然后一次性返回整个答案。响应延迟（Time To First Token, TTFT） 会比较长，用户必须耐心等待。

流式输出（Streaming） 则提供了截然不同的体验。它模拟了人与人之间实时交流的“打字机”效果。用户发出问题后，后端模型一边推理，一边将已生成的部分答案（Token）持续、分块地发送给前端。前端收到第一个 Token 就立即渲染，用户几乎能立刻看到反馈。这极大地提升了交互的实时感和用户体验，尤其适合需要长时间思考的大语言模型。

核心优势：将用户可感知的响应时间，从模型生成完整答案的耗时，缩短到生成第一个 Token 的耗时。让用户“感觉”模型在即时回复，而不是在黑箱中思考。

二、SSE：实现流式输出的轻量级协议

要实现流式输出，我们需要一个在 HTTP 长连接上，由服务器向客户端单向、持续推送数据的机制。Server-Sent Events (SSE) 正是为解决这个问题而生的 W3C 标准协议。

它基于 HTTP 协议，本质上是一个不会主动关闭的、状态为 Content-Type: text/event-stream 的 HTTP 响应。服务器通过这个连接，持续发送一种特定格式的文本数据块（称为“事件”）。浏览器内置的 EventSource API 可以轻松地监听这些事件。与 WebSocket 这种全双工协议相比，SSE 更简单、更轻量，且天然支持自动重连，完美契合了 AI 对话这种以服务器推送为主的场景。

SSE 消息的基本格式很简单，每条消息以 data: 开头，以两个换行符（\n\n）结尾。例如：

data: 你好

data: ，我

data: 是

data: AI助手。

三、为何是 SSE，而不是 WebSocket？

在需要实时通信的场景中，我们常会纠结用 SSE 还是 WebSocket。对于 MiMo 这类 AI 对话的流式输出，SSE 通常是更优选择，原因如下：

四、前端实践：使用 EventSource 接收流

在前端，使用浏览器原生提供的 EventSource API 来订阅 SSE 流非常简单。以下是一个基本的 JavaScript 示例：

// 建立与服务器 SSE 端点的连接
const eventSource = new EventSource('/api/chat/stream?query=你好MiMo');

// 监听服务器发送的普通消息（默认消息类型）
eventSource.onmessage = function(event) {
    // event.data 是服务器推来的一个数据块（例如一个词或一个标点）
    const token = event.data;
    
    // 将 token 追加到页面上的显示区域，模拟打字机效果
    document.getElementById('response-container').innerHTML += token;
    
    // 可选：自动滚动到底部，确保新内容可见
    window.scrollTo(0, document.body.scrollHeight);
};

// 监听连接打开
eventSource.onopen = function() {
    console.log('SSE 连接已建立。');
};

// 监听错误（包括连接中断）
eventSource.onerror = function(event) {
    console.error('SSE 连接错误:', event);
    // 根据规范，EventSource 会自动尝试重连
    // 你可以在这里更新 UI 状态，如显示“连接中...”
    if (eventSource.readyState === EventSource.CLOSED) {
        console.log('连接已关闭');
    }
};

// 当对话完成或需要主动关闭连接时（例如用户切换了聊天窗口）
// eventSource.close();
关键提示：EventSource 仅支持 GET 请求。如果你的流式对话接口需要发送复杂的上下文或历史记录（通常用 JSON），可以考虑用 fetch API 配合 ReadableStream 来手动实现流式请求，但这会失去 EventSource 的自动重连便利。

五、后端实践：生成符合 SSE 格式的流

后端服务器的任务是：接收请求，调用模型进行流式推理，并将每个生成的 Token 以正确的 SSE 格式发送给客户端。下面以 Python 的 FastAPI 框架为例：

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
from typing import AsyncGenerator

app = FastAPI()

# 模拟一个生成流式文本的异步生成器（实际中会调用你的模型）
async def generate_mock_stream(query: str) -> AsyncGenerator[str, None]:
    response_text = f"你好！你刚才对我说了：“{query}”。流式输出的感觉很酷吧！"
    for char in response_text:
        # 模拟每个字符生成的耗时
        await asyncio.sleep(0.05)
        # 按照 SSE 格式输出：每个 data 字段后跟两个换行符
        yield f"data: {char}\n\n"

@app.get("/api/chat/stream")
async def chat_stream(request: Request, query: str):
    """
    SSE 流式接口
    """
    # 返回一个 StreamingResponse，媒体类型为 text/event-stream
    return StreamingResponse(
        content=generate_mock_stream(query),
        media_type="text/event-stream"
    )

在这个例子中，generate_mock_stream 函数是一个异步生成器。它像一个“生产者”，每次 yield 一个格式为 data: xxx\n\n 的字符串。FastAPI 的 StreamingResponse 则像一个“运输工”，将生成器产出的每个数据块依次发送给客户端。客户端的 EventSource 就能一行一行地接收到这些字符。

六、进阶：事件类型与元数据控制

SSE 协议允许你定义不同的事件类型，而不仅仅局限于默认的 data 字段。这在实际应用中非常有用。

# 后端生成包含不同事件类型的流
async def advanced_stream():
    # 发送元数据事件
    yield "event: metadata\ndata: {\"total_tokens\": 100, \"model\": \"MiMo-7B\"}\n\n"
    
    # 发送普通数据流
    for i in range(10):
        await asyncio.sleep(0.1)
        yield f"data: 这是第{i}个词。\n\n"
    
    # 发送结束事件
    yield "event: finish\ndata: {\"status\": \"success\"}\n\n"
// 前端监听特定事件
const es = new EventSource('/stream');

// 监听普通数据
es.addEventListener('metadata', function(e) {
    const meta = JSON.parse(e.data);
    console.log('元数据:', meta);
});

es.addEventListener('finish', function(e) {
    console.log('流式传输完成。');
    es.close(); // 主动关闭连接
});

// onmessage 只接收没有 event 类型的消息
es.onmessage = function(e) {
    // 处理普通数据
};

合理利用事件类型，可以让你的流式 API 更加规范和强大，能够轻松区分数据流、状态消息和元数据，便于前端进行精细化的渲染和处理。