AI 工程系列：SSE 大模型流式输出基础原理和实践

前言

 本文主要介绍大模型对话场景下 SSE 的入门基础知识，并通过 Python 完成 SSE 前后端开发实践，实现页面打字机演示案例。一句话总结 SSE 在该场景下的角色与工作方式：大模型串行推理输出增量 token；vLLM 等推理后端在上层服务借助生成器，将增量内容逐段封装为 SSE 事件，依托 HTTP 长连接推送；客户端调用该 SSE 接口持续消费事件流，读取增量文本并实时渲染，实现打字机效果。

SSE简介

 大模型对话场景下，生成文本是逐 token 串行推理，需要逐个计算输出，如果等待全部文本生成完毕再一次性返回，用户会感受到长时间空白等待。流式输出可以在模型产出一小段 token 后立刻下发到前端，页面逐段渲染文字，模拟打字机效果，大幅降低用户感知延迟。SSE是实现流式输出的一种方式。
 SSE：Server-Sent Events，服务端发送事件，也叫事件流。它是一套建立在标准 HTTP 协议之上的单向消息推送规范，并非全新传输协议。客户端发起一次 HTTP 请求后，服务端在响应头标记Content-Type: text/event-stream，随后在同一个 HTTP 事务内持续不断输出事件数据，保持长连接，由客户端持续接收服务端推送的消息。SSE 仅支持服务端向客户端单向推送，客户端无法复用这条长连接回传数据。
 和一般的 HTTP REST 请求相比，REST 请求依靠 Content-Length 机制，服务端生成完整结果后一次性返回 body 并结束本次 HTTP 事务，采用一问一答交互，通常返回完整 JSON；SSE 采用 Transfer-Encoding: chunked 分块传输，无需预先知道数据总长度，一次请求即可持续推送多条事件，body 长时间持续输出，拥有独立事件流语法，依靠data:、event:、id:等字段与连续双换行\n\n划分事件块。

SSE数据格式规范

 SSE数据流由事件块组成，一个事件块由若干行key:[可选空格]value组成，可包含的key如下

  • data：必需字段，表示事件数据，支持多行（每行同样以data: 开头）
  • event：可选字段，指定事件类型
  • id：可选字段，设置消息ID，用于断线重连时恢复数据
  • retry：可选字段，指定重连间隔，单位毫秒

 每个字段间用一个换行符分隔，事件块以两个换行符结束，注释行以冒号开头，客户端会忽略这些行。例如符合SSE格式数据如下

data: 模型输出第一段文字：Hello


id: 1
event: message
data: 模型输出第二段
data: 支持多行data，这是第二行内容


retry: 5000
data: 这一条事件指定重连间隔


: ping


id: 4
data: [DONE]

OpenAI 协议流式 SSE 报文解析

 OpenAI 兼容接口，开启 stream=True 流式输出时，默认都是 SSE报文。调用示例如下

from openai import OpenAI

client = OpenAI(
    api_key="sk-xxx",
    base_url="http://10.2.13.11:3000/v1"
)

prompt = """你是一位专业的中文技术文档撰写专家，
请将以下文本进行简洁，准确的摘要，
要求语言通顺，保留技术要点，控制在20字以内:
原文内容:
{}"""

document = """营收48.93亿元、同比增长3.60%，经营活动现金流量净额却是负10.17亿元。沈鼓集团9月14日披露的招股说明书，给出了2026年上半年的这组反差：一家还在赚钱的高端装备企业，为什么经营账上的现金流出了10亿元？答案不在“卖不动”三个字里，而在大设备从接单、备料、交付到客户付款之间的时间差。"""

final_prompt = prompt.format(document)

response = client.chat.completions.create(
    model="gpt-6-luna",
    messages=[{"role": "user", "content": final_prompt}],
    temperature=0.7,
    stream=True
)

full_text = ""

# 遍历SSE流里的每一个chunk
for chunk in response:
    # 取出增量内容 delta.content
    delta = chunk.choices[0].delta
    content = delta.content
    print(chunk)
    if content is not None:
        full_text += content

print("【完整结果】", full_text)

 for循环中每一个chunk是一个事件块，其中一条chunk如下

ChatCompletionChunk(id='resp_0d931806dacce2e1016ac75a37abc887d0b4097b1c4db8a945', choices=[Choice(delta=ChoiceDelta(content='沈', function_call=None, refusal=None, role='assistant', tool_calls=None), finish_reason=None, index=0, logprobs=None, native_finish_reason=None)], created=1791449655, model='gpt-6-luna', object='chat.completion.chunk', moderation=None, service_tier='auto', system_fingerprint=None, usage=None)

 OpenAI包的client.chat.completions.create会对字符串的SSE报文结果封装成ChatCompletionChunk对象，因此看不出SSE格式，原始的SSE报文仅含有data key，且数据为JSON字符串或者[DONE]结束标记，用requests请求如下

import requests
import json

api_key = "sk-xxx"
base_url = "http://10.2.13.11:3000/v1"

prompt = """你是一位专业的中文技术文档撰写专家，
请将以下文本进行简洁，准确的摘要，
要求语言通顺，保留技术要点，控制在20字以内:
原文内容: {}"""

document = """营收48.93亿元、同比增长3.60%，经营活动现金流量净额却是负10.17亿元。沈鼓集团9月14日披露的招股说明书，给出了2026年上半年的这组反差：一家还在赚钱的高端装备企业，为什么经营账上的现金流出了10亿元？答案不在“卖不动”三个字里，而在大设备从接单、备料、交付到客户付款之间的时间差。"""

final_prompt = prompt.format(document)

# 请求参数
url = f"{base_url}/chat/completions"
headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}
payload = {
    "model": "gpt-6-luna",
    "messages": [{"role": "user", "content": final_prompt}],
    "temperature": 0.7,
    "stream": True
}

full_text = ""

# 发起流式请求
resp = requests.post(url, headers=headers, json=payload, stream=True)
resp.raise_for_status()

for line in resp.iter_lines():
    if not line:
        continue
    line = line.decode("utf-8")
    print(line)

    if line.startswith("data: "):
        data_str = line[6:]
        if data_str == "[DONE]":
            break
        try:
            chunk = json.loads(data_str)
            delta = chunk["choices"][0]["delta"]
            content = delta.get("content")
            if content is not None:
                full_text += content
        except Exception as e:
            print("解析异常", e)

print("【完整结果】", full_text)

 部分输出如下

data: {"id":"resp_027751648a1657d6016ac84c44405c87d08f7e93dc08ca4531","object":"chat.completion.chunk","created":1791511620,"model":"gpt-6-luna","choices":[{"index":0,"delta":{"role":"assistant","content":"沈"},"finish_reason":null,"native_finish_reason":null}],"service_tier":"auto"}
data: {"id":"resp_027751648a1657d6016ac84c44405c87d08f7e93dc08ca4531","object":"chat.completion.chunk","created":1791511620,"model":"gpt-6-luna","choices":[{"index":0,"delta":{"role":"assistant","content":"鼓"},"finish_reason":null,"native_finish_reason":null}],"service_tier":"auto"}
data: {"id":"resp_027751648a1657d6016ac84c44405c87d08f7e93dc08ca4531","object":"chat.completion.chunk","created":1791511620,"model":"gpt-6-luna","choices":[{"index":0,"delta":{"role":"assistant","content":"营"},"finish_reason":null,"native_finish_reason":null}],"service_tier":"auto"}
...

 使用requests请求，iter_lines是 requests 流式请求配套的方法，专门用来按行读取响应流，必须配合设置stream=True，iter_lines() 以换行符\n作为行结束标记，它会自动拼接 TCP 网络碎片，等攒到完整一行之后，才把这一行交给上层代码，OpenAI 风格的 SSE 刚好设计成：每一条data:单独占一行，data 内部不包含换行符，事件之间用空行\n\n隔开。因此使用iter_lines()可以很方便地逐行拿到每一条data:。
 拿到每个事件块后，需要自行解析SSE报文，例如解析data以及判断[DONE]结束状态，每个事件块中的choices[0].delta.content对应增量的内容，将所有choices[0].delta.content拼接即可拿到所有输出。本质上OpenAI包的返回是对SSE的JSON输出的对象封装。

Python实现SSE流式输出服务后端

 使用 FastAPI 编写 SSE 后端，使用StreamingResponse返回流式响应；StreamingResponse支持传入可迭代对象。为了边生成边推送、不一次性预加载全部数据，需要传入生成器，因此需要用yield产出分段数据。后端代码如下

from fastapi import FastAPI
from fastapi.responses import StreamingResponse, HTMLResponse
import time

app = FastAPI()


def stream_answer():
    full_text = "你好！欢迎来到Python的世界"
    for char in full_text:
        yield f"data: {char}\n\n"
        time.sleep(0.08)
    yield "data: [DONE]\n\n"


@app.post("/stream")
def sse_stream():
    return StreamingResponse(
        stream_answer(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive"
        }
    )


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("fastapi_test:app", host="127.0.0.1", port=8000)

 在代码中约定通过yield生成器按需分段产出 SSE 报文，传递给StreamingResponse，指定media_type="text/event-stream"告诉客户端这个响应是 SSE 事件流，不是普通文本 /json，指定"Cache-Control": "no-cache"，禁止浏览器缓存这条 SSE 流，不能缓存，因为是实时增量更新数据的，指定"Connection": "keep-alive"SSE 本身需要长连接持续推送。
 客户端请求如下

import requests

url = "http://127.0.0.1:8000/stream"
payload = {}
full_text = ""

resp = requests.post(url, json=payload, stream=True)
resp.raise_for_status()

for line in resp.iter_lines(decode_unicode=True):
    if not line:
        continue

    if line.startswith("data: "):
        data_str = line[6:]
        if data_str == "[DONE]":
            break
        content = data_str
        print(content)
        full_text += content

print("\n【完整结果】", full_text)

运行结果如下

你
好
！
欢
迎
来
到
P
y
t
h
o
n
的
世
界

【完整结果】 你好！欢迎来到Python的世界

前端解析SSE实现打字机效果

 一个基于使用fetch + ReadableStream实现的前端demo代码如下

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>SSE 打字机演示</title>
    <style>
        #box {
            width:600px;
            margin:30px auto;
            font-size:20px;
            line-height:1.6;
        }
    </style>
</head>
<body>
    <div id="box"></div>

<script>
const box = document.getElementById('box');

async function startStream() {
    const res = await fetch("http://127.0.0.1:8000/stream", {
        method: "POST",
        headers: {
            "Content-Type": "application/json"
        },
        body: JSON.stringify({prompt: ""})
    });

    // 获取reader，读取二进制流
    const reader = res.body.getReader();
    const decoder = new TextDecoder();
    let buffer = "";

    while(true) {
        const {done, value} = await reader.read();
        if(done) break;

        // 转字符串，追加到缓冲区
        buffer += decoder.decode(value, {stream:true});

        // 按SSE分隔符 \n\n 切割事件
        while(buffer.includes("\n\n")) {
            const idx = buffer.indexOf("\n\n");
            const eventStr = buffer.slice(0, idx);
            buffer = buffer.slice(idx + 2);

            console.log("eventStr:", eventStr);

            // 提取 data: 后面内容
            if(eventStr.startsWith("data: ")) {
                const data = eventStr.replace("data: ", "");
                console.log("data:", data);
                if(data === "[DONE]") {
                    console.log("流式结束");
                    return;
                }
                box.innerText += data;
            }
        }
    }
}

startStream();
</script>
</body>
</html>

 在服务端加上html引用

@app.get("/", response_class=HTMLResponse)
def index():
    with open("index.html", "r", encoding="utf-8") as f:
        return f.read()

打开浏览器如下


前端demo
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成，浏览时请结合常识与多方信息审慎甄别。
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

友情链接更多精彩内容