obsidian-notes/InBox/BV1if7E64Ex5-SSE-FastAPI流式响应笔记.md

SSE 到底是什么？用 FastAPI 一次讲透流式响应 | BV1if7E64Ex5

Milky 整理

一、三种通信协议对比

协议	类比	特点
HTTP	懒惰的服务员	你问一次，他答一次，服务器无法主动推送
SSE	勤快的服务员	只需要开口说一次，他会主动汇报最新进度
WebSocket	双方都带着对讲机	任何一方都可以随时主动说话，完全平等的双向通信

二、SSE 协议的适用场景

• HTTP：必须等模型生成完所有内容才能一次性返回，用户等待时间过长
• WebSocket：功能强大但成本高，且大多数 AI 场景中用户不需要主动发消息
• SSE：只发起一次请求，服务器边生成边推送，用户实时看到文字逐字出现

三、SSE 工作原理详解

完整流程

客户端 → 普通 HTTP 请求 → 服务器
客户端 ← Content-Type: text/event-stream ← 服务器
客户端 ← Connection: keep-alive ← 服务器
客户端 ← 持续推送事件数据 ← 服务器
客户端 ← event: done / data: [DONE] ← 服务器

HTTP 响应头

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

SSE 消息格式

字段	作用
data	消息内容
event	事件类型
id	消息编号
retry	断线重连间隔（毫秒）

每条消息之间用空行分隔。

浏览器端接收

const source = new EventSource('/stream');
source.onmessage = (event) => {
    console.log(event.data);
};

四、FastAPI 核心前置知识

• 与 Pydantic 深度绑定，自动请求解析和响应序列化
• 自动生成交互式 API 文档（/docs）
• 原生支持 async/await 异步编程

五、生成器（Generator）与 yield 关键字

yield 函数的四个特点：

1. 用 yield 代替 return
2. 执行到 yield 暂停并返回值
3. 下次调用从 yield 后继续
4. 按需产生数据，适合流式场景

六、完整 FastAPI SSE 实现

环境要求

FastAPI >= 0.135

代码

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

app = FastAPI()

class ChatRequest(BaseModel):
    message: str

@app.post("/stream")
async def stream_chat(request: ChatRequest):
    async for chunk in model_stream(request.message):
        if chunk == "[DONE]":
            return
        yield ServerSentEvent(data=chunk)

核心公式

EventSourceResponse + yield ServerSentEvent

七、三种协议总结对比

协议	实现复杂度	推送方向	适用场景
HTTP	⭐ 最简单	仅服务端响应	简单的一问一答
SSE	⭐⭐ 中等	服务端持续推送	AI 流式输出、实时通知
WebSocket	⭐⭐⭐ 复杂	双向通信	需要客户端主动发消息的场景

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