从手写SSE帧到生产级流式问答：FastAPI + sse-starlette 实战踩坑记

最近用 FastAPI 做了一个 RAG 流式问答系统支持上传 PDF、多轮对话、打字机效果输出。本文记录从手写 SSE 协议帧到引入 sse-starlette 的全过程包括两个真实踩过的坑依赖版本冲突、断连语义和一个提前预研的问题放到 Nginx 后面会怎样。如果你也在做类似的东西本文能帮你少走弯路。一、为什么需要流式输出普通接口的做法等大模型把答案全部生成完再一次性返回 JSON。问题是用户要白屏等 5~10 秒体验很差。流式输出的目标是 打字机效果—— 大模型吐一个字前端就显示一个字。技术方案对比方案方向复杂度适合场景普通 JSON 响应请求 / 响应低短回答SSEServer-Sent Events服务器 → 客户端单向中等AI 流式输出本文用这个WebSocket双向高实时聊天、协作编辑SSE 本质上就是一个普通 HTTP 长连接服务器可以持续往里面写数据浏览器自动逐条接收。二、SSE 帧到底长什么样SSE 的数据格式非常简单每一帧长这样data: {type:delta,content:中国}注意三个关键点data:是协议规定的前缀不能改后面是一个完整的 JSON 字符串我们自定义的 type、content 都打包在这个 JSON 里最后必须有一个空行\n\n而且这个空行在 JSON 引号外面—— 它是帧与帧之间的分隔符。空行位置写错浏览器永远收不到事件。三、第一版手写 SSE 帧最开始不引入任何第三方库自己拼帧。FastAPI 里核心代码长这样import json from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel app FastAPI() class AskRequest(BaseModel): question: str def _sse_event(data: dict) - str: 把事件 dict 序列化为 SSE 帧 return fdata: {json.dumps(data, ensure_asciiFalse)}\n\n app.post(/ask/stream) async def ask_stream(req: AskRequest): question req.question.strip() if not question: raise HTTPException(status_code400, detailquestion 不能为空) def generate(): full_answer for event in rag.ask_stream(question): if event[type] delta: full_answer event[content] yield _sse_event({type: delta, content: event[content]}) elif event[type] sources: yield _sse_event({type: sources, sources: event[sources]}) yield _sse_event({type: done}) return StreamingResponse(generate(), media_typetext/event-stream)这版能跑但手写协议有三个坑坑 1\n\n漏一个字符前端就收不到事件。SSE 规定空行才是事件结束少一个 \n浏览器会一直等下一帧。坑 2中文必须写ensure_asciiFalse。不写的话中国 会被序列化成 \u4e2d\u56fd帧体积变大调试时也看不懂。坑 3StreamingResponse要手动设media_typetext/event-stream。不设的话浏览器不知道这是 SSE 流会当作普通 JSON 响应等全部下载完。手写帧的好处是能看清协议本质但生产环境不该自己维护这些细节 —— 容易错还少了心跳、断线重连这些能力。四、引入 sse-starlette结果依赖冲突标准做法是用 sse-starlette 这个库它封装了 EventSourceResponse 和 ServerSentEventpip install sse-starlette然后改造代码from sse_starlette.sse import EventSourceResponse, ServerSentEvent app.post(/ask/stream) async def ask_stream(req: AskRequest): question req.question.strip() if not question: raise HTTPException(status_code400, detailquestion 不能为空) def generate(): full_answer for event in rag.ask_stream(question): if event[type] delta: full_answer event[content] yield ServerSentEvent(data{type: delta, content: event[content]}) elif event[type] sources: yield ServerSentEvent(data{type: sources, sources: event[sources]}) yield ServerSentEvent(data{type: done}) return EventSourceResponse(generate())改动点_sse_event 辅助函数整个删掉ServerSentEvent(data...) 自动帮你做序列化和拼帧StreamingResponse(..., media_type...) 换成 EventSourceResponse(generate())MIME 类型自动设置。踩坑版本冲突新装完启动直接报错fastapi 0.115.0 requires starlette0.39.0,0.37.2, but you have starlette 1.7.0 which is incompatible.原因最新版 sse-starlette 3.x 要求 starlette ≥ 0.49把 starlette 拉到了 1.7.0而我的 FastAPI 0.115.0 锁死 starlette 0.39。两个要求完全不重叠。解决方法是降版本pip install sse-starlette1.8.2pip 会自动把 starlette 降回 0.38.6和 FastAPI 匹配。这个版本的 API 和 3.x 用法完全一样代码不用改。五、断连后发生了什么有个问题我一开始想当然了用户回答到一半关掉浏览器数据库会留下什么我的第一反应是 大模型继续生成完然后存库。因为实际生活里的大模型是这样但是我们目前做出来的demo还不是。真实流程是用户关浏览器 → TCP 连接断开Starlette 检测到客户端断开关闭生成器抛 GeneratorExitfor event in rag.ask_stream(...) 循环被打断循环后面的 db.save_message(...)根本不会执行full_answer 拼到一半就随生成器销毁。也就是说断连后历史表里不会留下半截回答。这其实是好事 —— 历史表里不会出现 用户问了一半、回答了两个字 的脏数据。正常流程请求 → 读历史 → 流式生成 → 存用户问题 → 存完整回答 → done 断连流程请求 → 读历史 → 流式生成一半 → 连接断 → 循环中断 → 什么都不存如果产品要求 哪怕断连也要存完整回答就需要把生成任务放到后台协程里跑让它独立于客户端连接。这个改动大概 40 行代码demo 阶段可以zanshi需要做。六、提前想一下放到 Nginx 后面会遇到什么坑本地跑通后我在想SSE 这种长时间保持连接的接口放到反向代理后面会不会有问题查了一下 Nginx 默认配置果然有个 proxy_read_timeoutproxy_read_timeout 60s;它的意思是Nginx 等后端数据超过 60 秒还没收到就主动断开连接。那问题就来了大模型在 思考检索资料、推理这段时间没有 token 输出SSE 连接上 60 秒没有任何数据流过Nginx 判定超时断开连接前端表现为 卡住然后失败。这也是为什么 EventSourceResponse 自带心跳每隔几秒自动发一个 SSE 注释帧 : ping\n\n即使大模型没产出新 token连接上也一直有数据流过Nginx 就不会误判超时。这也是为什么手写 StreamingResponse 虽然能跑但生产环境更推荐 sse-starlette—— 心跳这种细节库已经帮你处理好了自己写容易漏。EventSourceResponse 自带心跳每隔几秒自动发一个 SSE 注释帧 : ping\n\n即使大模型没产出新 token连接上也一直有数据流过Nginx 就不会误判超时。如果坚持用 StreamingResponse需要自己在生成器里加心跳任务定期往连接里写注释帧。七、上传 PDF 的三道防线顺便记录上传接口的安全设计这部分和流式无关但做 RAG 都要用到ALLOWED_EXTENSIONS {.pdf} MAX_UPLOAD_SIZE 50 * 1024 * 1024 # 50MB app.post(/upload) async def upload_pdf(file: UploadFile File(...)): original_name file.filename or upload.pdf ext pathlib.Path(original_name).suffix.lower() if ext not in ALLOWED_EXTENSIONS: raise HTTPException(status_code400, detail仅支持 PDF 文件) content await file.read() if len(content) MAX_UPLOAD_SIZE: raise HTTPException(status_code413, detail文件超过 50MB 限制) # 防路径穿越存储名用 uuid原文件名只做展示 stored_name f{uuid.uuid4().hex}{ext} pdf_path UPLOAD_DIR / stored_name pdf_path.write_bytes(content) ...三道防线扩展名白名单只允许 PDF大小限制50MB防止超大文件撑爆内存UUID 重命名用户文件名可能是 ../../etc/passwd直接拼路径会写穿目录。用随机名存储原文件名只作为元数据。八、最终项目结构KubeRAG/ ├── app/ │ ├── main.py # FastAPI 路由层 │ ├── rag.py # PDF 切片、向量化、Chroma 检索 │ ├── db.py # SQLite 会话历史 │ └── agent.py # 工具调用 Agent ├── data/ │ ├── uploads/ # PDF 原文 │ └── chat_history.db ├── static/ │ └── index.html # 前端页面 └── requirements.txt完整代码已上传 GitHubBowliceDXY/KubeRAG (github.com)九、面试可能会追问的问题写这篇文章的过程中我自己整理了几个面试官大概率会问的问题供参考SSE 和 WebSocket 怎么选—— 单向推送选 SSE双向交互选 WebSocket\n\n为什么在 JSON 外面—— 它是帧分隔符不是数据内容用户中途关页面历史会存半截吗—— 不会生成器被关闭save_message 不执行Nginx 超时断连怎么解决—— 心跳 ping保持连接活跃sse-starlette 和手写 StreamingResponse 区别—— 封装了序列化、MIME、心跳少写协议细节。总结做流式输出本身不难难的是处理那些教程文章里不太会写的边缘情况依赖版本冲突、断连后发生什么、代理超时怎么办。写这篇文章的初衷不是当教程而是把自己这几天学习过程中踩过的坑完整记录下来。如果其中某一段刚好帮到正在做同样事情的同学那就真的太好了。我也是边学边做文章里的理解不一定全对。如果你发现哪里有问题、或者有更优雅的实现方式欢迎评论区指出来咱们一起讨论。