流式输出

一、什么是流式输出？

流式输出是一种边生成、边传输、边展示的数据处理模式，核心是将完整内容拆分为小数据块，生成一块推送一块，无需等待全部内容就绪，最典型的场景就是AI对话的逐字输出效果。它与批量输出（全部生成完一次性返回）的核心区别的是，流式输出首字节返回快、无需缓存全文、可随时中断，而批量输出延迟高、需缓存完整内容、无法中途终止。

类比来说，流式输出类似自来水边流边用，批量输出则类似等待餐品全部做好后再取用。

二、SSE规范是什么？

SSE（服务端发送事件）是实现流式输出最常用、最适合入门的标准化方案，基于HTTP协议实现，无需复杂握手流程。其核心规范包括两部分：

1.，必须的响应头，需包含

Content-Type: text/event-stream; charset=utf-8
Cache-Control: no-cache  //禁止缓存
Connection: keep-alive  //保持长连接

数据格式

每条消息由一行或多行字段组成，以空行 \n\n 结束，标识单条消息结束

4 个标准字段（仅这 4 个有效）

1）data: 内容（必选，最常用）

承载消息正文，UTF-8 文本，每行都以 data: 开头
```
data: 第一行
data: 第二行
\n
```
2）event: 事件名（可选）

自定义事件类型，前端 addEventListener 监听，不写默认是 message 事件
```
event: update
data: {"status":"ok"}
\n
```
3）id: 字符串（可选）

消息唯一 ID，用于断点续传，客户端重连时自动带请求头：Last-Event-ID: 上次的id
```
id: 123
data: hello
\n
```
4）retry: 毫秒（可选）

客户端断线后自动重连间隔，默认：浏览器自定（通常 3–5 秒）
```
retry: 3000
data: reconnect in 3s
\n
```

三、实操

前后端如何实现SSE流式连接？完整流程分为四步：

前端发起连接，通过浏览器原生EventSource对象，指定服务端接口地址（如示例地址http://127.0.0.1:5000/stream），自动发起HTTP请求并监听连接；
```
// 前端
const es = new EventSource("http://127.0.0.1:5000/stream");
```
后端响应并维持长连接，通过Flask框架编写接口，返回SSE标准响应头，利用yield关键字逐段生成并推送数据（yield与return的区别是，yield可暂停推送并保持连接，return会一次性返回并关闭连接）；

后端返回必须带这三个头，标志这是 SSE 长连接：

   Content-Type: text/event-stream
   Cache-Control: no-cache
   Connection: keep-alive

浏览器看到这个头，就知道：不要关闭连接，持续监听后端发的数据

后端保持连接不断开，循环生成数据

Python 后端用 yield 逐段产出：

   def sse_gen():
       yield "data: 第一段内容\n\n"
       time.sleep(1)
       yield "data: 第二段内容\n\n"

后端按 SSE 规范发数据块

每条消息格式固定，必须以两个换行 \n\n 结尾：

   data: 你好呀\n\n

浏览器收到 \n\n 就认为一条消息结束，触发前端事件。

数据推送与渲染，后端按SSE规范推送数据，前端通过监听EventSource的onmessage事件，实时接收并渲染数据，实现流式展示；4. 连接保活与重连，后端定时发送心跳信息，网络中断时浏览器会按retry配置自动重连，并携带上一次接收的消息ID实现断点续传。

   es.onmessage = function(e) {
       // e.data 就是后端发的 data 内容
       document.getElementById("box").innerText += e.data;
   };

后端定时发 : heartbeat\n\n 心跳，防止网关断开长连接；如果网络断了，浏览器自动重连，不用前端写重连逻辑。

SSE 本质就是：前端 EventSource 发起普通 HTTP 请求→后端返回 SSE 专属头，保持长连接不关闭→后端按 data:xxx\n\n 格式逐段推数据→前端监听 onmessage，收到一段渲染一段

四、补充说明

流式输出的核心优势的是低延迟、省内存、体验好、抗超时，适用于AI对话、实时日志、大文件下载等场景；其与WebSocket的区别是，SSE是单向推送、基于HTTP、自动重连、操作简单，WebSocket是双向通信、基于独立协议、需手动重连、复杂度高。此外，提供了两种Python流式输出代码，分别是控制台本地模拟流式打字效果，以及Flask+SSE接口流式返回（含前端测试代码），可根据需求选择使用。

本地控制台流式输出

原理：逐个字符输出，加延时，不一次性打印全部

import time
import sys

def stream_output(text, delay=0.05):
    """逐字符流式输出"""
    for char in text:
        sys.stdout.write(char)   # 逐个写入
        sys.stdout.flush()       # 强制刷新缓冲区
        time.sleep(delay)
    print()  # 换行结束

if __name__ == "__main__":
    content = "你好，这是Python实现的流式输出效果，像AI逐字打字一样～"
    stream_output(content)

Flask + SSE 接口流式输出

服务端代码 server.py

from flask import Flask, Response
import time

app = Flask(__name__)

# 模拟大模型逐段生成文本
def generate_stream():
    sentences = [
        "哈喽～",
        "我是流式输出接口",
        "采用SSE协议逐块推送数据",
        "不需要等待全部生成完"
    ]
    for sent in sentences:
        # SSE 固定格式：data:xxx\n\n
        yield f"data: {sent}\n\n"
        time.sleep(0.8)

@app.route('/stream')
def stream():
    return Response(
        generate_stream(),
        mimetype="text/event-stream",  # 标识SSE流式
        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}
    )

if __name__ == "__main__":
    app.run(debug=True)

前端测试 html

<!DOCTYPE html>
<html>
<body>
<div id="content"></div>

<script>
const evtSource = new EventSource("http://127.0.0.1:5000/stream");
evtSource.onmessage = function(e) {
    document.getElementById("content").innerText += e.data;
};
</script>
</body>
</html>

PREVIOUS把个人博客改造成共创博客：一次 Jekyll/TeXt 实战记录

NEXTGitHub上利用CI/CD自动化部署上线静态网页