obsidian-notes/InBox/BV15F7J6dEdm-SSE-AI流式输出笔记.md

别再只会 HTTP 了，SSE 才是 AI 流式输出的答案 | BV15F7J6dEdm

Milky 整理

协议对比：三种通信模式的本质差异

HTTP 协议：懒惰的服务员

客户端发起请求 → 服务器响应 → 连接关闭 → 客户端再次发起请求 → ...

• 客户端主动发起请求，服务端被动响应
• 一问一答模式，服务端不能主动推送消息
• 无状态、单次请求
• 连接在每次响应后立即关闭

适用场景：普通接口查询、网页加载、表单提交、简单数据获取

轮询问题：网络开销大、延迟、实时性差

SSE 协议：勤快的服务员

客户端发起一次请求 → 服务端持续主动推送 → 直到传输结束

• 客户端只需发送一次请求
• 服务端主动、持续推送消息
• 单向通信（服务端 → 客户端）
• 长连接保持

适用场景：AI 大模型流式输出、外卖/订单进度推送、股票 K 线图、监控仪表盘、任何需要服务端主动推送文本数据的场景

WebSocket 协议：随身对讲机

• 双向通信
• 客户端和服务端角色对等
• 实现复杂度最高

适用场景：实时聊天室、语音问答/交互、在线课堂、需要双向实时交互的场景

三种协议对比

特性	HTTP	SSE	WebSocket
通信方向	单向（客户端→服务端）	单向（服务端→客户端）	双向
服务端主动推送	❌	✅	✅
连接类型	短连接	长连接	长连接
实现难度	⭐ 最简单	⭐⭐ 中等	⭐⭐⭐ 复杂
数据格式	任意	文本数据	二进制/文本均可
客户端发送消息	需新建请求	需新建请求	随时可发

为什么 AI 流式输出首选 SSE？

• HTTP 的问题：大模型自回归生成，用户等待 1-2 分钟才能看到完整回复，无法流式体验
• WebSocket 的问题：实现复杂成本高，AI 对话场景不需要双向通信
• SSE 的优势：完美匹配 AI 流式输出，一次请求服务端持续推送增量结果，实现简单

SSE 协议详解

典型交互过程

客户端 → HTTP GET 请求 → 服务器
客户端 ← 响应头 text/event-stream ← 服务器
客户端 ← 持续推送事件流 ← 服务器
客户端 ← 推送结束 ← 服务器

响应头

字段	值	说明
Content-Type	text/event-stream	标识事件流
Cache-Control	no-cache	不缓存
Connection	keep-alive	保持连接活跃

协议格式（四个字段）

field: value\n\n

字段	说明	示例
data	消息数据	data: {"content": "你好"}
event	事件类型（可自定义）	event: message
id	消息ID（断线重连）	id: 1
retry	重连间隔（毫秒）	retry: 5000

结束传输的四种方式

1. 强制断开连接
2. 自定义结束事件：event: done
3. 自定义结束标识：data: [DONE]
4. 业务字段标识：data: {"content": "...", "finish": true}

FastAPI SSE 实战

前置知识

• Pydantic 数据模型 —— FastAPI 自动 JSON ↔ Python 对象转换
• EventSource API —— 浏览器原生 SSE 客户端
• Python yield 关键字 —— 生成器，逐次生成数据流

代码实现

FastAPI 0.135+ 原生支持 SSE，使用 EventSourceResponse + AsyncGenerator[ServerSentEvent]

from fastapi import FastAPI
from fastapi.responses import EventSourceResponse
from sse.starlette.sse import ServerSentEvent
from pydantic import BaseModel
from typing import AsyncGenerator

app = FastAPI()

class ChatRequest(BaseModel):
    message: str

async def event_generator(req: ChatRequest) -> AsyncGenerator[ServerSentEvent, None]:
    response_text = f"你问的是: {req.message}\nSSE 是一种服务端推送协议..."
    words = response_text.split()
    for i, word in enumerate(words):
        is_final = (i == len(words) - 1)
        yield ServerSentEvent(
            event="message",
            data=f'{{"content": "{word} ", "isfinal": {is_final}}}'
        )

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    return EventSourceResponse(event_generator(req))

启动：uvicorn main:app --reload

客户端调用

• JavaScript fetch + ReadableStream：逐块读取并解析 SSE 数据
• EventSource API：浏览器原生，监听 message / 自定义事件

总结

概念	说明
SSE	Server-Sent Events，服务端推送事件协议
核心优势	单次请求 + 服务端持续推送 + 实现简单
最佳场景	AI 大模型流式输出、实时进度推送
协议格式	data / event / id / retry 四个字段
结束方式	断开连接 / 自定义事件 / 特殊标识 / 业务字段
FastAPI 集成	0.135+ 原生支持，使用 EventSourceResponse

────────────────────────────── Generated by MilkyAi@Bilibili