我接触过的几十个LLM落地项目绝大部分问题都不出在模型效果上而出在“数据怎么从服务端流到前端”这个环节。模型吐字快前端展示慢对话轮次多连接直接断工具调用好几步客户端根本不知道当前执行到哪了。这些问题的根源往往就是通讯架构没设计好。这篇文章我会把 FastAPI 在 LLM 应用中的通讯方案完整复盘一遍——SSE 怎么做流式输出、WebSocket 怎么支撑多轮对话、传统 HTTP 还被用在哪些地方以及三者之间怎么协同。不管你是后端开发、AI 应用工程师还是自己动手搭 RAG 项目、Agent 服务这篇文章应该都能给你一份可以直接参照的方案。1. 为什么 LLM 应用需要一套专门的通讯架构先说一个大前提LLM 接口并不是“能调通”就算完事。模型推理本身动辄几秒到几十秒客户端如果采用传统的请求-响应模式用户只能干盯着一个 loading 转圈。更关键的是大模型的输出是逐 token 生成的同一个请求内部存在明确的时间差。通讯架构要解决的本质上就是“如何把这段时间差转化为流式的、实时的、可中断的交互体验”。1.1 从一次“卡顿”说起普通 HTTP 请求的真实瓶颈我见过不少项目第一版后端就是一个/chat接口前端用 axios 发 POST等模型完整吐完结果后再一次性返回 JSON。用户问一个问题页面要转圈 8 到 15 秒然后整段文字“啪”地一下出现。这种体验在 Demo 阶段还能接受真放到产品里就没法用了。造成这个问题的主要有两层原因。第一层是模型推理速度LLM 生成 500 个 token在普通消费级 GPU 上可能就需要十几秒即使调云端 API也要几秒到几十秒不等。第二层是 HTTP 协议的请求-响应模型在响应结束之前客户端拿不到任何中间数据。这两层一叠加用户等待时间就被拉满了。解决思路有两个方向。一是“伪造流式”——模型还是完整返回后端响应后前端模拟打字机效果逐字展示。这种做法对接口要求低但用户感知上的首字延迟没有改善多轮长对话时依然很僵硬。二是“真实流式”——后端把模型逐 token 生成的结果实时推给前端首字能几百毫秒内出来。要实现真流式通讯协议就必须支持“连接保持 分块传输”。1.2 三种协议的本质差异选型前先想明白这三件事聊选型之前先把三个协议的区别用大白话捋清楚。HTTP 是“叫外卖”你下单商家做好一次性送到你手上。整个过程是一来一回中间没有状态。SSE 是“听收音机”你打开频道服务端不停给你播放单项流动你不需要回话。WebSocket 是“打电话”双方都能随时说话你一言我一语连接一旦建立双向实时通行。对应到 LLM 场景协议通讯方向典型场景优势短板HTTP一请求一响应模型列表、鉴权、文件上传、同步查询简单、可靠、生态完善无法实时推送中间结果SSE服务端单向推送流式对话、逐 token 输出、进度通知基于 HTTP实现成本低自动重连机制单向客户端不能通过同一连接发指令WebSocket双向实时多轮对话、工具调用、状态同步、中断控制全双工、低延迟、连接复用实现复杂度高需处理心跳和断线重连选型只需要回答三个问题数据往哪个方向流实时性要求多高是不是只需要一次交互如果只是“模型输出结果给用户看”SSE 是绝对优先的因为它简单、稳定浏览器原生支持连前端解析都有现成的库。如果是“用户可能随时打断模型回答、重新提问、或者让模型执行多步工具调用”那必须上 WebSocket因为这类交互是双向的客户端需要频繁发送指令服务端还要主动推送执行状态。至于普通 HTTP它永远不会被替代模型列表、鉴权、知识库管理这类低频、非流式的接口用 HTTP 反而最合适。2. 三种协议在 FastAPI 中的实现细节这一节进入正题。我会把每个协议在 FastAPI 中的实现要点拆开讲重点说清楚那些“文档里不会写”的细节。2.1 SSE 不是“轮询”是“单向管道”——流式输出的核心实现SSEServer-Sent Events服务端发送事件本质上就是一次 HTTP 长连接服务端不断向客户端推送分块的文本数据。它和普通 HTTP 响应的最大区别在于连接不会因为响应体写完而关闭而是会持续输出直到服务端主动结束。用 FastAPI 实现 SSE核心对象是StreamingResponse。很多人第一次写的时候会直接把它当成普通 JSON 响应来用结果前端收不到流式效果。关键在于两点第一必须设置media_typetext/event-stream第二必须用生成器函数作为响应体不能直接返回一个列表或字符串。一个最小可用示例from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app FastAPI() async def token_generator(message: str): # 模拟大模型逐 token 输出 for token in message: await asyncio.sleep(0.05) yield fdata: {json.dumps({token: token}, ensure_asciiFalse)}\n\n app.post(/chat/stream) async def chat_stream(prompt: str): # 实际项目中这里会调用 LLM 推理 async def generate(): async for token in token_generator(f你问我:{prompt}我是AI助手): yield token return StreamingResponse( generate(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 禁止 Nginx 缓冲 } )这里有个隐藏坑前端的fetchAPI 默认不会处理流式响应必须用ReadableStream自己解析。很多前端会直接用response.json()那肯定拿不到流式效果。正确做法是const response await fetch(/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ prompt: 你好 }) }); const reader response.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { value, done } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n); buffer lines.pop(); // 保留最后不完整的一行 for (const line of lines) { if (line.startsWith(data:)) { const data JSON.parse(line.slice(5).trim()); processToken(data.token); } } }特别提醒SSE 的每条消息必须以data:开头以两个换行符\n\n结束。这是协议规定的格式服务端拼错一个字符前端解析就会紊乱。另外如果数据里有换行符推送前必须处理掉或转义否则会破坏消息边界。2.2 WebSocket 的“全双工”——多轮对话和工具调用的完整支持SSE 解决的是“单向流式输出”但如果用户想随时打断模型回答或者让模型连续执行多步工具调用、期间还要上报执行状态那就需要 WebSocket 了。WebSocket 在 FastAPI 里用起来也不复杂它不走StreamingResponse而是用websocket依赖注入做握手和消息收发。一个典型的多轮对话长连接长这样from fastapi import WebSocket, WebSocketDisconnect class ConnectionManager: def __init__(self): self.active_connections: list[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def send_json(self, websocket: WebSocket, data: dict): await websocket.send_json(data) manager ConnectionManager() app.websocket(/ws/chat) async def websocket_chat_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: # 接收客户端消息 request await websocket.receive_json() # 解析消息类型 if request[type] chat: # 模拟LLM流式推理 async def stream_response(): for token in request[message].split(): await websocket.send_json({ type: token, content: token }) await asyncio.sleep(0.1) await websocket.send_json({ type: done, content: [END] }) await stream_response() elif request[type] cancel: # 实际需要配合任务取消机制不能简单地break await websocket.send_json({type: cancelled}) except WebSocketDisconnect: manager.disconnect(websocket)这里重点是设计一个 JSON 消息协议。我的习惯是每条消息必须有type字段常见类型有chat客户端发消息、token服务端流式返回、status状态更新、done结束、error错误、cancel客户端取消。有了type前端才能精确知道当前消息是正文、状态还是结束信号不至于把工具调用状态和模型回答混在一起。有个细节经常被忽略websocket.send_json()会一次性发送整个 JSON 对象。你可以在应用层用循环持续发 token但这是一个个独立的 WebSocket 消息不代表底层帧也是分块的。WebSocket 本身不关心你发的是一段完整消息还是半截消息你要在应用层自己约定好“这一次发送的内容是一段结束、还是还有后续”。2.3 普通 HTTP 在 LLM 应用里还剩下哪些不可替代的位置很多人觉得有了 SSE 和 WebSocketHTTP 就没用了。事实恰恰相反一个正常的 LLM 应用HTTP 接口数量反而是最多的。具体来说这些场景用 HTTP 最合适模型列表查询GET/models、应用鉴权POST/auth/token、知识库文档上传POST/documents、历史记录查询GET/history/session_id、服务健康检查GET/health。这些接口的共同点是不需要流式、交互次数少、逻辑简单。强行给它们加 SSE 或 WebSocket 只会增加无谓的复杂度。在 FastAPI 中写这类接口很简单但有两点经验值得分享。第一是异步客户端的选择如果在后端需要调用外部 LLM API比如 OpenAI 兼容接口一定要用httpx.AsyncClient并且设置连接复用import httpx client httpx.AsyncClient( timeouthttpx.Timeout(30.0, read120.0), limitshttpx.Limits(max_connections200, max_keepalive_connections50), headers{Authorization: Bearer your-api-key} ) app.get(/models) async def list_models(): resp await client.get(https://api.llm.example/v1/models) return resp.json()第二是 HTTP 连接复用问题。HTTP/1.1 的 keep-alive 可以减少 TCP 握手开销HTTP/2 更是支持真正的多路复用。在自建的 LLM 网关里如果客户端并发高务必让反向代理开启 HTTP/2否则大量连接重开会成为瓶颈。3. 一套完整可落地的项目结构三协议如何协同只讲零散代码不太够我把自己在项目里常用的一个 LLM 通讯架构完整梳理出来你完全可以照着搭。3.1 项目目录结构与模块划分先说目录结构。不要把所有路由全堆在一个main.py里分层是必须的llm-gateway/ ├── app/ │ ├── main.py # FastAPI 实例与根路由 │ ├── api/ │ │ ├── http_routes.py # HTTP 常规接口 │ │ ├── sse_routes.py # SSE 流式接口 │ │ └── ws_routes.py # WebSocket 接口 │ ├── services/ │ │ └── llm_service.py # 统一的 LLM 调用封装 │ ├── schemas/ │ │ └── chat.py # 请求/响应模型 │ └── core/ │ ├── config.py # 配置 │ └── logger.py # 日志 ├── tests/ └── requirements.txt这里的分层逻辑很明确api层只负责协议适配和参数校验不直接调大模型services层统一封装 LLM 调用不管是 OpenAI SDK 还是自部署的 vLLM都收敛到一个llm_service里。这样如果将来要换模型服务商只需要改 service 内部逻辑三个协议层完全不用动。3.2 SSE 与 WebSocket 如何共享同一套 LLM 服务很多项目会出现逻辑重复SSE 里写了一遍调用大模型的代码WebSocket 里又复制了一遍。我的做法是把“生成 token 的异步生成器”抽成公共方法让两个协议层都来消费同一个生成器# services/llm_service.py async def stream_chat(prompt: str, history: list[dict] | None None): 返回一个异步生成器每次 yield 一个 token 字符串。 不管是 SSE 层还是 WebSocket 层都直接消费这个生成器。 messages [] if history: messages.extend(history) messages.append({role: user, content: prompt}) # 这里以 OpenAI 兼容接口为例 async with httpx.AsyncClient() as client: async with client.stream( POST, https://api.llm.example/v1/chat/completions, json{model: your-model, messages: messages, stream: True} ) as response: async for line in response.aiter_lines(): if not line.startswith(data:): continue data json.loads(line[5:].strip()) if data.get(choices) and data[choices][0].get(delta): token data[choices][0][delta].get(content) if token: yield token然后 SSE 路由层只做一件事把llm_service.stream_chat()的 yield 转换成 SSE 格式# api/sse_routes.py app.post(/v1/chat/stream) async def sse_chat(request: ChatRequest): async def event_generator(): async for token in llm_service.stream_chat(request.prompt, request.history): payload json.dumps({token: token}, ensure_asciiFalse) yield fdata: {payload}\n\n yield data: [DONE]\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)WebSocket 路由层则可以做更多编排接收消息、判断消息类型、根据类型决定是流式返回正文还是执行工具调用# api/ws_routes.py app.websocket(/v1/ws/chat) async def ws_chat(websocket: WebSocket): await websocket.accept() try: while True: msg await websocket.receive_json() if msg[type] chat: full_text async for token in llm_service.stream_chat(msg[content]): await websocket.send_json({type: token, content: token}) full_text token await websocket.send_json({type: done, content: full_text}) elif msg[type] ping: await websocket.send_json({type: pong}) except WebSocketDisconnect: pass这套结构最大的好处是业务逻辑服务层与通讯方式协议层完全解耦。想从 SSE 切到 WebSocket只改路由层想换模型服务只改服务层。3.3 连接鉴权与密钥处理两个协议层都绕不开的问题在真实项目中连接鉴权是个容易踩坑的点。HTTP 接口的鉴权很成熟用Authorization: Bearer token头FastAPI 里用Depends依赖做校验即可。但 SSE 和 WebSocket 就有讲究了。SSE 的浏览器 EventSource API 是没法自定义请求头的只能通过 URL 参数或 cookie 携带鉴权信息。常用做法# 用query参数携带token app.get(/v1/chat/stream) async def sse_chat_with_auth(prompt: str, token: str): user verify_token(token) # 校验失败则抛出HTTPException ...注意query 参数里的 token 会记录在服务器访问日志中如果对安全性要求高应该用一次性短时凭证比如先调用 HTTP 接口换取一个 5 分钟有效的临时 token再拼到 SSE URL 中。WebSocket 的情况稍微好一点因为它是手写协议可以在建立连接时要求客户端在 URL 中携带 token。但是不能在websocket.accept()之后才校验否则已经建立了连接再断开就很奇怪。正确顺序是先解析 query、做校验、再 acceptapp.websocket(/v1/ws/chat) async def ws_chat(websocket: WebSocket, token: str): user verify_token(token) if not user: await websocket.close(code4001, reasonunauthorized) return await websocket.accept() ...4. LLM 通讯架构中的关键问题与磨出来的实践心得架构搭得再好看跑起来一定会遇到问题。这节把我踩过的坑、还有别人踩过但我会主动避开的坑一次性写全。4.1 SSE 流式输出就卡顿检查代理缓冲和超时设置使用 SSE 最常见的坑是部署在 Nginx 后面之后流式响应变成“一阵一阵”的甚至完全卡死。原因通常出在反向代理的缓冲机制上。Nginx 默认会缓冲上游响应等全部内容收完再发给客户端。这就把 SSE 的“流式”属性废掉了。解决办法是在 Nginx 配置里为 SSE 路径单独关闭缓冲location /v1/chat/stream { proxy_pass http://127.0.0.1:8000; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; proxy_send_timeout 300s; proxy_http_version 1.1; proxy_set_header Connection ; }除此之外还有一个容易忽略的点proxy_read_timeout。Nginx 的默认值是 60 秒如果模型思考时间超过这个数字连接会被 Nginx 掐断。很多人遇到过stream disconnected before completion: idle timeout waiting for sse这类错误十有八九就是代理层超时设置太短。建议把proxy_read_timeout调整到 300 秒以上同时定期发送心跳注释行: keep-alive\n\n来维持连接活跃。4.2 WebSocket 连接建立但没有数据多半是消息格式或准入门槛WebSocket 的“连接不上”和“连接上但收不到数据”是两个完全不同的问题。如果是“连接上但收不到数据”第一反应应该是服务端send_json是否真的执行了是否被业务代码里的异常吞掉了我在本地调试时就遇到过LLM 服务调用抛异常后服务端代码直接 return 了WebSocket 连接还挂着前端一直没有收到任何消息也没有报错。建议在 WebSocket 消息处理循环里统一包裹 try-except在异常时也往客户端发送一个error类型的消息而不是静默关闭或退出。如果是“前端收不到数据但服务端显示已发送”则要看是不是浏览器对 WebSocket 消息的事件绑定写错了。常见原因是onmessage回调里解析 JSON 失败导致异常中断。调试这类问题最好的办法就是打开浏览器开发者工具的 Network 面板查看 WebSocket 帧的实际内容看服务端发的到底是什么。4.3 心跳机制别让连接死在“无声的等待”里无论是 SSE 还是 WebSocket长时间没有数据流动时连接都可能被中间网络设备或云厂商的网关主动回收。WebSocket 的心跳实现方式很简单客户端每隔 30 秒发一个ping服务端收到后回一个pong。我这里写一个标准代码模式# 客户端心跳 setInterval(() { if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ type: ping })); } }, 30000); ws.onmessage (event) { const msg JSON.parse(event.data); if (msg.type undefined) return; if (msg.type pong) { // 心跳正常不做额外处理 return; } // 其他业务消息 };服务端只要在消息循环里识别ping类型并回复pong即可。SSE 的心跳则是另一种形式因为服务端可以随时向客户端推送数据所以可以在空闲时每隔 15 到 20 秒发送一行注释yield : keep-alive\n\n这是 SSE 协议允许的注释格式浏览器会自动忽略但能有效维持连接不被中间层断开。4.4 日志丢失FastAPI 的异步环境与自定义日志很多人在开发时没有留意日志问题。在 FastAPI Uvicorn 的异步环境下如果直接使用 Python 的print()日志可能不会出现在终端或者打印的 token 顺序错乱。建议使用logging模块显式配置输出格式尤其是要确保StreamHandler的flush及时。另外一个类似的经验如果是用 Uvicorn 启动服务记得关闭access_log或者把它重定向到独立的日志文件否则大量 WebSocket 的心跳 log 会把重要日志淹没在垃圾信息里。4.5 前端两种消费模式的最终选择这里再补一个前端选型问题。很多项目是 Vue 或 React 写的前后端联调时前端工程师往往会问SSE 好还是 WebSocket 好我给出的建议非常明确如果业务里全是不需要打断的单向流式对话前端用 SSE。因为浏览器原生EventSource就支持自动重连你不需要写任何重连逻辑而且fetch加流式读取的代码复杂度远比 WebSocket 低。如果你的业务里有“打断当前生成”“取消并重新提问”“多个并发 Agent 任务的状态同步”这类需求那么就选择 WebSocket因为 SSE 的连接是单向的你没法用同一个连接去发送“停止”指令。如果不确定可以采取混合方案对话模型这类高频实时通道走 WebSocket而 token 级的流式输出也走 WebSocket 的token消息。前面我给的代码示例就是这种混合模式实际项目中非常稳定。5. 常见问题速查表与最后的实操建议为了方便直接查漏我把上面遇到的问题整理成一张速查表。现象可能原因解决方案SSE 响应一次性返回无流式效果StreamingResponse 用了 List/str 而不是生成器或 Nginx 缓冲未关闭确保响应体为 async generatorNginx 设置 proxy_buffering off连接 60 秒后被断开代理层 read timeout 太短调整 proxy_read_timeoutSSE 定期发心跳注释WebSocket 连接成功但收不到消息服务端异常被吞前端事件绑定有误统一 try-except用 DevTools 检查 WebSocket 帧多进程部署时 WebSocket 竞态轮询分发到不同进程状态丢失用 Redis 共享连接状态必要时用 sticky sessionHTTP 接口大量超时未使用连接池默认 httpx 每请求新建连接全局复用 AsyncClient配置 Limitstoken 流中断或乱码编码问题或消息边界被破坏统一 UTF-8SSE 消息以\n\n结尾token 内容转义部署后鉴权失败反向代理丢弃了 Authorization 头或 query 参数检查 Nginx 的 proxy_set_header 配置最后再分享一个小技巧。调试 SSE 或 WebSocket 的时候不要一上来就启动完整的前端项目用命令行直接测才是最直观的。SSE 用一行curl就能看流式效果curl -N http://127.0.0.1:8000/v1/chat/stream?prompt你好WebSocket 可以用websocat工具连上之后手动发 JSONwebsocat ws://127.0.0.1:8000/v1/ws/chat {type:chat,content:你好}这两个工具基本是后端调试的标配能帮你快速确认问题出在服务端还是前端。我个人在实际项目里已经完整跑过好几套这样的架构了最大的体会是通讯协议是 LLM 应用的骨架骨架不对模型效果再好也白搭。这三个协议本身并不是“谁替代谁”的关系而是各管一段协同工作。先把每个协议的边界想清楚再把服务层抽离开来后面的路走起来就顺多了。 
长河君 · 2026/10/10 13:28:55 · 阅读 1.2k FastAPI通讯方案：SSE与WebSocket在LLM应用中的协同
我接触过的几十个LLM落地项目绝大部分问题都不出在模型效果上而出在“数据怎么从服务端流到前端”这个环节。模型吐字快前端展示慢对话轮次多连接直接断工具调用好几步客户端根本不知道当前执行到哪了。这些问题的根源往往就是通讯架构没设计好。这篇文章我会把 FastAPI 在 LLM 应用中的通讯方案完整复盘一遍——SSE 怎么做流式输出、WebSocket 怎么支撑多轮对话、传统 HTTP 还被用在哪些地方以及三者之间怎么协同。不管你是后端开发、AI 应用工程师还是自己动手搭 RAG 项目、Agent 服务这篇文章应该都能给你一份可以直接参照的方案。1. 为什么 LLM 应用需要一套专门的通讯架构先说一个大前提LLM 接口并不是“能调通”就算完事。模型推理本身动辄几秒到几十秒客户端如果采用传统的请求-响应模式用户只能干盯着一个 loading 转圈。更关键的是大模型的输出是逐 token 生成的同一个请求内部存在明确的时间差。通讯架构要解决的本质上就是“如何把这段时间差转化为流式的、实时的、可中断的交互体验”。1.1 从一次“卡顿”说起普通 HTTP 请求的真实瓶颈我见过不少项目第一版后端就是一个/chat接口前端用 axios 发 POST等模型完整吐完结果后再一次性返回 JSON。用户问一个问题页面要转圈 8 到 15 秒然后整段文字“啪”地一下出现。这种体验在 Demo 阶段还能接受真放到产品里就没法用了。造成这个问题的主要有两层原因。第一层是模型推理速度LLM 生成 500 个 token在普通消费级 GPU 上可能就需要十几秒即使调云端 API也要几秒到几十秒不等。第二层是 HTTP 协议的请求-响应模型在响应结束之前客户端拿不到任何中间数据。这两层一叠加用户等待时间就被拉满了。解决思路有两个方向。一是“伪造流式”——模型还是完整返回后端响应后前端模拟打字机效果逐字展示。这种做法对接口要求低但用户感知上的首字延迟没有改善多轮长对话时依然很僵硬。二是“真实流式”——后端把模型逐 token 生成的结果实时推给前端首字能几百毫秒内出来。要实现真流式通讯协议就必须支持“连接保持 分块传输”。1.2 三种协议的本质差异选型前先想明白这三件事聊选型之前先把三个协议的区别用大白话捋清楚。HTTP 是“叫外卖”你下单商家做好一次性送到你手上。整个过程是一来一回中间没有状态。SSE 是“听收音机”你打开频道服务端不停给你播放单项流动你不需要回话。WebSocket 是“打电话”双方都能随时说话你一言我一语连接一旦建立双向实时通行。对应到 LLM 场景协议通讯方向典型场景优势短板HTTP一请求一响应模型列表、鉴权、文件上传、同步查询简单、可靠、生态完善无法实时推送中间结果SSE服务端单向推送流式对话、逐 token 输出、进度通知基于 HTTP实现成本低自动重连机制单向客户端不能通过同一连接发指令WebSocket双向实时多轮对话、工具调用、状态同步、中断控制全双工、低延迟、连接复用实现复杂度高需处理心跳和断线重连选型只需要回答三个问题数据往哪个方向流实时性要求多高是不是只需要一次交互如果只是“模型输出结果给用户看”SSE 是绝对优先的因为它简单、稳定浏览器原生支持连前端解析都有现成的库。如果是“用户可能随时打断模型回答、重新提问、或者让模型执行多步工具调用”那必须上 WebSocket因为这类交互是双向的客户端需要频繁发送指令服务端还要主动推送执行状态。至于普通 HTTP它永远不会被替代模型列表、鉴权、知识库管理这类低频、非流式的接口用 HTTP 反而最合适。2. 三种协议在 FastAPI 中的实现细节这一节进入正题。我会把每个协议在 FastAPI 中的实现要点拆开讲重点说清楚那些“文档里不会写”的细节。2.1 SSE 不是“轮询”是“单向管道”——流式输出的核心实现SSEServer-Sent Events服务端发送事件本质上就是一次 HTTP 长连接服务端不断向客户端推送分块的文本数据。它和普通 HTTP 响应的最大区别在于连接不会因为响应体写完而关闭而是会持续输出直到服务端主动结束。用 FastAPI 实现 SSE核心对象是StreamingResponse。很多人第一次写的时候会直接把它当成普通 JSON 响应来用结果前端收不到流式效果。关键在于两点第一必须设置media_typetext/event-stream第二必须用生成器函数作为响应体不能直接返回一个列表或字符串。一个最小可用示例from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app FastAPI() async def token_generator(message: str): # 模拟大模型逐 token 输出 for token in message: await asyncio.sleep(0.05) yield fdata: {json.dumps({token: token}, ensure_asciiFalse)}\n\n app.post(/chat/stream) async def chat_stream(prompt: str): # 实际项目中这里会调用 LLM 推理 async def generate(): async for token in token_generator(f你问我:{prompt}我是AI助手): yield token return StreamingResponse( generate(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 禁止 Nginx 缓冲 } )这里有个隐藏坑前端的fetchAPI 默认不会处理流式响应必须用ReadableStream自己解析。很多前端会直接用response.json()那肯定拿不到流式效果。正确做法是const response await fetch(/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ prompt: 你好 }) }); const reader response.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { value, done } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n); buffer lines.pop(); // 保留最后不完整的一行 for (const line of lines) { if (line.startsWith(data:)) { const data JSON.parse(line.slice(5).trim()); processToken(data.token); } } }特别提醒SSE 的每条消息必须以data:开头以两个换行符\n\n结束。这是协议规定的格式服务端拼错一个字符前端解析就会紊乱。另外如果数据里有换行符推送前必须处理掉或转义否则会破坏消息边界。2.2 WebSocket 的“全双工”——多轮对话和工具调用的完整支持SSE 解决的是“单向流式输出”但如果用户想随时打断模型回答或者让模型连续执行多步工具调用、期间还要上报执行状态那就需要 WebSocket 了。WebSocket 在 FastAPI 里用起来也不复杂它不走StreamingResponse而是用websocket依赖注入做握手和消息收发。一个典型的多轮对话长连接长这样from fastapi import WebSocket, WebSocketDisconnect class ConnectionManager: def __init__(self): self.active_connections: list[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def send_json(self, websocket: WebSocket, data: dict): await websocket.send_json(data) manager ConnectionManager() app.websocket(/ws/chat) async def websocket_chat_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: # 接收客户端消息 request await websocket.receive_json() # 解析消息类型 if request[type] chat: # 模拟LLM流式推理 async def stream_response(): for token in request[message].split(): await websocket.send_json({ type: token, content: token }) await asyncio.sleep(0.1) await websocket.send_json({ type: done, content: [END] }) await stream_response() elif request[type] cancel: # 实际需要配合任务取消机制不能简单地break await websocket.send_json({type: cancelled}) except WebSocketDisconnect: manager.disconnect(websocket)这里重点是设计一个 JSON 消息协议。我的习惯是每条消息必须有type字段常见类型有chat客户端发消息、token服务端流式返回、status状态更新、done结束、error错误、cancel客户端取消。有了type前端才能精确知道当前消息是正文、状态还是结束信号不至于把工具调用状态和模型回答混在一起。有个细节经常被忽略websocket.send_json()会一次性发送整个 JSON 对象。你可以在应用层用循环持续发 token但这是一个个独立的 WebSocket 消息不代表底层帧也是分块的。WebSocket 本身不关心你发的是一段完整消息还是半截消息你要在应用层自己约定好“这一次发送的内容是一段结束、还是还有后续”。2.3 普通 HTTP 在 LLM 应用里还剩下哪些不可替代的位置很多人觉得有了 SSE 和 WebSocketHTTP 就没用了。事实恰恰相反一个正常的 LLM 应用HTTP 接口数量反而是最多的。具体来说这些场景用 HTTP 最合适模型列表查询GET/models、应用鉴权POST/auth/token、知识库文档上传POST/documents、历史记录查询GET/history/session_id、服务健康检查GET/health。这些接口的共同点是不需要流式、交互次数少、逻辑简单。强行给它们加 SSE 或 WebSocket 只会增加无谓的复杂度。在 FastAPI 中写这类接口很简单但有两点经验值得分享。第一是异步客户端的选择如果在后端需要调用外部 LLM API比如 OpenAI 兼容接口一定要用httpx.AsyncClient并且设置连接复用import httpx client httpx.AsyncClient( timeouthttpx.Timeout(30.0, read120.0), limitshttpx.Limits(max_connections200, max_keepalive_connections50), headers{Authorization: Bearer your-api-key} ) app.get(/models) async def list_models(): resp await client.get(https://api.llm.example/v1/models) return resp.json()第二是 HTTP 连接复用问题。HTTP/1.1 的 keep-alive 可以减少 TCP 握手开销HTTP/2 更是支持真正的多路复用。在自建的 LLM 网关里如果客户端并发高务必让反向代理开启 HTTP/2否则大量连接重开会成为瓶颈。3. 一套完整可落地的项目结构三协议如何协同只讲零散代码不太够我把自己在项目里常用的一个 LLM 通讯架构完整梳理出来你完全可以照着搭。3.1 项目目录结构与模块划分先说目录结构。不要把所有路由全堆在一个main.py里分层是必须的llm-gateway/ ├── app/ │ ├── main.py # FastAPI 实例与根路由 │ ├── api/ │ │ ├── http_routes.py # HTTP 常规接口 │ │ ├── sse_routes.py # SSE 流式接口 │ │ └── ws_routes.py # WebSocket 接口 │ ├── services/ │ │ └── llm_service.py # 统一的 LLM 调用封装 │ ├── schemas/ │ │ └── chat.py # 请求/响应模型 │ └── core/ │ ├── config.py # 配置 │ └── logger.py # 日志 ├── tests/ └── requirements.txt这里的分层逻辑很明确api层只负责协议适配和参数校验不直接调大模型services层统一封装 LLM 调用不管是 OpenAI SDK 还是自部署的 vLLM都收敛到一个llm_service里。这样如果将来要换模型服务商只需要改 service 内部逻辑三个协议层完全不用动。3.2 SSE 与 WebSocket 如何共享同一套 LLM 服务很多项目会出现逻辑重复SSE 里写了一遍调用大模型的代码WebSocket 里又复制了一遍。我的做法是把“生成 token 的异步生成器”抽成公共方法让两个协议层都来消费同一个生成器# services/llm_service.py async def stream_chat(prompt: str, history: list[dict] | None None): 返回一个异步生成器每次 yield 一个 token 字符串。 不管是 SSE 层还是 WebSocket 层都直接消费这个生成器。 messages [] if history: messages.extend(history) messages.append({role: user, content: prompt}) # 这里以 OpenAI 兼容接口为例 async with httpx.AsyncClient() as client: async with client.stream( POST, https://api.llm.example/v1/chat/completions, json{model: your-model, messages: messages, stream: True} ) as response: async for line in response.aiter_lines(): if not line.startswith(data:): continue data json.loads(line[5:].strip()) if data.get(choices) and data[choices][0].get(delta): token data[choices][0][delta].get(content) if token: yield token然后 SSE 路由层只做一件事把llm_service.stream_chat()的 yield 转换成 SSE 格式# api/sse_routes.py app.post(/v1/chat/stream) async def sse_chat(request: ChatRequest): async def event_generator(): async for token in llm_service.stream_chat(request.prompt, request.history): payload json.dumps({token: token}, ensure_asciiFalse) yield fdata: {payload}\n\n yield data: [DONE]\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)WebSocket 路由层则可以做更多编排接收消息、判断消息类型、根据类型决定是流式返回正文还是执行工具调用# api/ws_routes.py app.websocket(/v1/ws/chat) async def ws_chat(websocket: WebSocket): await websocket.accept() try: while True: msg await websocket.receive_json() if msg[type] chat: full_text async for token in llm_service.stream_chat(msg[content]): await websocket.send_json({type: token, content: token}) full_text token await websocket.send_json({type: done, content: full_text}) elif msg[type] ping: await websocket.send_json({type: pong}) except WebSocketDisconnect: pass这套结构最大的好处是业务逻辑服务层与通讯方式协议层完全解耦。想从 SSE 切到 WebSocket只改路由层想换模型服务只改服务层。3.3 连接鉴权与密钥处理两个协议层都绕不开的问题在真实项目中连接鉴权是个容易踩坑的点。HTTP 接口的鉴权很成熟用Authorization: Bearer token头FastAPI 里用Depends依赖做校验即可。但 SSE 和 WebSocket 就有讲究了。SSE 的浏览器 EventSource API 是没法自定义请求头的只能通过 URL 参数或 cookie 携带鉴权信息。常用做法# 用query参数携带token app.get(/v1/chat/stream) async def sse_chat_with_auth(prompt: str, token: str): user verify_token(token) # 校验失败则抛出HTTPException ...注意query 参数里的 token 会记录在服务器访问日志中如果对安全性要求高应该用一次性短时凭证比如先调用 HTTP 接口换取一个 5 分钟有效的临时 token再拼到 SSE URL 中。WebSocket 的情况稍微好一点因为它是手写协议可以在建立连接时要求客户端在 URL 中携带 token。但是不能在websocket.accept()之后才校验否则已经建立了连接再断开就很奇怪。正确顺序是先解析 query、做校验、再 acceptapp.websocket(/v1/ws/chat) async def ws_chat(websocket: WebSocket, token: str): user verify_token(token) if not user: await websocket.close(code4001, reasonunauthorized) return await websocket.accept() ...4. LLM 通讯架构中的关键问题与磨出来的实践心得架构搭得再好看跑起来一定会遇到问题。这节把我踩过的坑、还有别人踩过但我会主动避开的坑一次性写全。4.1 SSE 流式输出就卡顿检查代理缓冲和超时设置使用 SSE 最常见的坑是部署在 Nginx 后面之后流式响应变成“一阵一阵”的甚至完全卡死。原因通常出在反向代理的缓冲机制上。Nginx 默认会缓冲上游响应等全部内容收完再发给客户端。这就把 SSE 的“流式”属性废掉了。解决办法是在 Nginx 配置里为 SSE 路径单独关闭缓冲location /v1/chat/stream { proxy_pass http://127.0.0.1:8000; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; proxy_send_timeout 300s; proxy_http_version 1.1; proxy_set_header Connection ; }除此之外还有一个容易忽略的点proxy_read_timeout。Nginx 的默认值是 60 秒如果模型思考时间超过这个数字连接会被 Nginx 掐断。很多人遇到过stream disconnected before completion: idle timeout waiting for sse这类错误十有八九就是代理层超时设置太短。建议把proxy_read_timeout调整到 300 秒以上同时定期发送心跳注释行: keep-alive\n\n来维持连接活跃。4.2 WebSocket 连接建立但没有数据多半是消息格式或准入门槛WebSocket 的“连接不上”和“连接上但收不到数据”是两个完全不同的问题。如果是“连接上但收不到数据”第一反应应该是服务端send_json是否真的执行了是否被业务代码里的异常吞掉了我在本地调试时就遇到过LLM 服务调用抛异常后服务端代码直接 return 了WebSocket 连接还挂着前端一直没有收到任何消息也没有报错。建议在 WebSocket 消息处理循环里统一包裹 try-except在异常时也往客户端发送一个error类型的消息而不是静默关闭或退出。如果是“前端收不到数据但服务端显示已发送”则要看是不是浏览器对 WebSocket 消息的事件绑定写错了。常见原因是onmessage回调里解析 JSON 失败导致异常中断。调试这类问题最好的办法就是打开浏览器开发者工具的 Network 面板查看 WebSocket 帧的实际内容看服务端发的到底是什么。4.3 心跳机制别让连接死在“无声的等待”里无论是 SSE 还是 WebSocket长时间没有数据流动时连接都可能被中间网络设备或云厂商的网关主动回收。WebSocket 的心跳实现方式很简单客户端每隔 30 秒发一个ping服务端收到后回一个pong。我这里写一个标准代码模式# 客户端心跳 setInterval(() { if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ type: ping })); } }, 30000); ws.onmessage (event) { const msg JSON.parse(event.data); if (msg.type undefined) return; if (msg.type pong) { // 心跳正常不做额外处理 return; } // 其他业务消息 };服务端只要在消息循环里识别ping类型并回复pong即可。SSE 的心跳则是另一种形式因为服务端可以随时向客户端推送数据所以可以在空闲时每隔 15 到 20 秒发送一行注释yield : keep-alive\n\n这是 SSE 协议允许的注释格式浏览器会自动忽略但能有效维持连接不被中间层断开。4.4 日志丢失FastAPI 的异步环境与自定义日志很多人在开发时没有留意日志问题。在 FastAPI Uvicorn 的异步环境下如果直接使用 Python 的print()日志可能不会出现在终端或者打印的 token 顺序错乱。建议使用logging模块显式配置输出格式尤其是要确保StreamHandler的flush及时。另外一个类似的经验如果是用 Uvicorn 启动服务记得关闭access_log或者把它重定向到独立的日志文件否则大量 WebSocket 的心跳 log 会把重要日志淹没在垃圾信息里。4.5 前端两种消费模式的最终选择这里再补一个前端选型问题。很多项目是 Vue 或 React 写的前后端联调时前端工程师往往会问SSE 好还是 WebSocket 好我给出的建议非常明确如果业务里全是不需要打断的单向流式对话前端用 SSE。因为浏览器原生EventSource就支持自动重连你不需要写任何重连逻辑而且fetch加流式读取的代码复杂度远比 WebSocket 低。如果你的业务里有“打断当前生成”“取消并重新提问”“多个并发 Agent 任务的状态同步”这类需求那么就选择 WebSocket因为 SSE 的连接是单向的你没法用同一个连接去发送“停止”指令。如果不确定可以采取混合方案对话模型这类高频实时通道走 WebSocket而 token 级的流式输出也走 WebSocket 的token消息。前面我给的代码示例就是这种混合模式实际项目中非常稳定。5. 常见问题速查表与最后的实操建议为了方便直接查漏我把上面遇到的问题整理成一张速查表。现象可能原因解决方案SSE 响应一次性返回无流式效果StreamingResponse 用了 List/str 而不是生成器或 Nginx 缓冲未关闭确保响应体为 async generatorNginx 设置 proxy_buffering off连接 60 秒后被断开代理层 read timeout 太短调整 proxy_read_timeoutSSE 定期发心跳注释WebSocket 连接成功但收不到消息服务端异常被吞前端事件绑定有误统一 try-except用 DevTools 检查 WebSocket 帧多进程部署时 WebSocket 竞态轮询分发到不同进程状态丢失用 Redis 共享连接状态必要时用 sticky sessionHTTP 接口大量超时未使用连接池默认 httpx 每请求新建连接全局复用 AsyncClient配置 Limitstoken 流中断或乱码编码问题或消息边界被破坏统一 UTF-8SSE 消息以\n\n结尾token 内容转义部署后鉴权失败反向代理丢弃了 Authorization 头或 query 参数检查 Nginx 的 proxy_set_header 配置最后再分享一个小技巧。调试 SSE 或 WebSocket 的时候不要一上来就启动完整的前端项目用命令行直接测才是最直观的。SSE 用一行curl就能看流式效果curl -N http://127.0.0.1:8000/v1/chat/stream?prompt你好WebSocket 可以用websocat工具连上之后手动发 JSONwebsocat ws://127.0.0.1:8000/v1/ws/chat {type:chat,content:你好}这两个工具基本是后端调试的标配能帮你快速确认问题出在服务端还是前端。我个人在实际项目里已经完整跑过好几套这样的架构了最大的体会是通讯协议是 LLM 应用的骨架骨架不对模型效果再好也白搭。这三个协议本身并不是“谁替代谁”的关系而是各管一段协同工作。先把每个协议的边界想清楚再把服务层抽离开来后面的路走起来就顺多了。