FastAPI实现LLM流式通讯：SSE、WebSocket与生产部署实战

去年下半年我接了一个LLM客服机器人的项目，服务端技术栈选了FastAPI。需求看起来很简单：用户提问，模型流式返回答案渲染到前端聊天框。等真正把服务搭起来，才发现表面上一个“流式返回”背后牵扯着SSE、WebSocket、HTTP连接复用、代理缓冲、心跳保活一整套东西。那段时间我几乎把FastAPI的通讯能力翻了个底朝天，也踩了不少坑。这篇就把我当时从选型到落地全过程的思考、代码和排障记录整理出来，希望能帮到正在做类似LLM应用后端的朋友。

整篇文章会围绕三个核心问题展开：一是LLM应用为什么需要SSE、WebSocket这类长连接协议，普通HTTP到底哪里不够用；二是FastAPI分别怎么实现这三种通讯方式，代码怎么写才不容易出问题；三是生产环境里这些连接怎么共存、怎么部署、怎么调优。都是实操层面的东西，测试环境跑通了Demo可不算完，真正上线那一刻才是坑的开始。

1. 为什么LLM应用绕不开SSE和WebSocket：HTTP协议的边界在哪

1.1 传统的请求响应模型碰上了“慢接口”

先看看最常见的HTTP请求流程：客户端发一个POST请求，服务端处理完返回完整响应，连接关闭。对普通API来说这没问题，一个查询可能几十毫秒就结束了，等就等了。

但LLM推理不是这么回事。一个大模型的生成速度通常在每秒几十到上百个token，一个稍长的回答可能需要几十秒。如果用传统HTTP请求，前端发一个请求后要傻等几十秒什么都拿不到，这期间用户盯着一个空白的加载转圈。更麻烦的是，如果中途网络抖动一下，整个请求就断了，几十秒的生成白费了，用户还得重新提问。

LLM应用必须让用户看到“正在生成”的过程，而不是干等。这样交互体验好是一方面，更重要的是心理上让人更容易接受等待，用户看到字一个个蹦出来，会觉得系统在干活，而不是死了。

1.2 轮询是个妥协方案，但不是正解

有人会说，那用轮询好了。前端每秒钟请求一次“生成完了没”，服务端查一下状态返回“还没好”。这个方案的问题很明显：一是浪费资源，几十秒的生成过程要发几十次HTTP请求，每次都要过一遍路由和鉴权；二是延迟，轮询间隔再短也有空隙，用户体验总感觉一顿一顿的；三是服务端还得维护任务状态存储，平白增加复杂度。

轮询的本质是用频繁的短连接去模拟长连接，属于“用错误工具硬凑”。当你的并发用户一多，轮询对网关、对服务端、对数据库的压力会成倍放大，纯属给自己埋雷。

1.3 SSE和WebSocket各自的定位

SSE，全称Server-Sent Events，服务端推送事件。它是建立在HTTP之上的单向通道——只能服务端往客户端推，客户端想发消息得另开普通HTTP请求。这让它在语义上非常契合“模型生成内容持续往外吐”这个场景：客户端发起提问，服务端按模型输出的节奏，一个个把token通过SSE推给前端。

WebSocket则是全双工的双向通道，客户端和服务端都能随时发消息。握手阶段通过HTTP完成，之后升级到独立的TCP长连接。多轮对话、协作编辑、实时状态同步这类场景，客户端需要随时打断模型推理，或者同时发多个指令，WebSocket更合适。

要理解这两者和普通HTTP的关系，我习惯用打电话和信件来类比：HTTP就像寄信，你寄一封，对方回一封，事毕；SSE像订阅了一份定期报纸，报社天天往你家送，但你不太容易往回写信；WebSocket像通了电话，两边想说话就说。LLM应用里，这三种方式各有各的适用位置，不存在绝对替代的关系。

2. SSE实战：从流式响应到断连排查的完整过程

2.1 用StreamingResponse实现SSE的两种方式

FastAPI实现SSE，核心掌握了两个方案就够用：一种是FastAPI内置的StreamingResponse，另一种是sse-starlette库。我两个都用过，先看StreamingResponse的方式。

from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app = FastAPI() async def token_generator(prompt: str): # 模拟从LLM服务获取流式输出 tokens = ["你好", "，", "我是", "AI", "助手", "。"] for token in tokens: # SSE协议要求以data:开头，最后以空行结束 yield f"data: {json.dumps({'token': token}, ensure_ascii=False)}\n\n" await asyncio.sleep(0.2) @app.post("/chat/stream") async def chat_stream(prompt: str): return StreamingResponse( token_generator(prompt), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # 禁用Nginx缓冲，后面会细说 } )

StreamingResponse接收一个异步生成器，FastAPI会把生成器产出的内容一块块直接写回客户端，而不是等整个生成器跑完。这是实现流式响应的关键。media_type必须设置成"text/event-stream"，这是SSE的MIME类型，浏览器和解析库都认这个。

再看sse-starlette的写法：

from sse_starlette.sse import EventSourceResponse async def token_generator(prompt: str): tokens = ["你好", "，", "我是", "AI", "助手", "。"] for token in tokens: yield { "event": "message", "data": json.dumps({"token": token}, ensure_ascii=False) } await asyncio.sleep(0.2) @app.post("/chat/stream") async def chat_stream(prompt: str): return EventSourceResponse(token_generator(prompt))

sse-starlette帮你处理了SSE协议的格式包装、心跳、断连检测、ping间隔设置等。我自己在不需要对底层格式做定制的话，优先用sse-starlette多一点，省心，它底层处理了一些边界情况比如客户端断开后自动取消生成器。

2.2 SSE消息格式细节与前端解析陷阱

SSE协议本身很轻量，但格式上有个容易踩坑的地方。标准格式是：

data: 第一行数据\n data: 第二行数据\n \n

以data:开头，多个连续data:行会被合并成一个数据块，中间用换行符连接，以空行结束一条完整消息。还有event:字段可以自定义事件名，id:字段用于断连重试时的断点续传。

前端解析上有两种常见做法。第一种是用EventSource，它原生支持SSE，但是只支持GET请求，而且无法自定义请求头（比如Auth Token）。对LLM应用来说，提问内容往往比较长且参数复杂，用GET很别扭。第二种是用fetch加ReadableStream自己解析，这种方式灵活得多，POST、请求头都能自定义，也是我推荐的方案。关键代码长这样：

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(); while (true) { const { value, done } = await reader.read(); if (done) break; const chunk = decoder.decode(value, { stream: true }); // 按SSE格式解析出data:后面的内容 const lines = chunk.split("\n"); for (const line of lines) { if (line.startsWith("data: ")) { const data = JSON.parse(line.slice(6)); console.log(data.token); } } }

很多人会在这一步出问题：直接把response.json()来读。fetch拿到的StreamingResponse如果直接用json()解析，会一直等到整个流结束才能拿到完整数据，流式效果完全消失。必须用getReader()逐步消费。

2.3 “stream disconnected before completion: idle timeout waiting for sse”问题排查

热搜词里有一条非常扎心：“stream disconnected before completion: idle timeout waiting for sse”。我当时在测试环境第一次遇到这个报错，排查了很久。

先解释这个报错的本质：服务端或中间件有一个空闲超时时间，如果在超时时间内没有任何数据包经过，连接就会被强制断开。而LLM模型有时候会在生成一个较长片段前思考一段时间，比如做内部推理、组织答案结构，这段时间通常是几秒到几十秒不等，没有任何token输出。从代理服务器的角度看，这就是一条“空闲”连接，超出阈值就被掐断了。

我当时的排查链路是这样的：

第一步，先确认服务端有没有问题。在FastAPI代码里加了很多日志，发现服务端明明在正常等待模型输出，没有主动断开。排除了应用层问题。

第二步，检查是不是本机直接访问接口就没事、走Nginx才会断。直接在服务器上curl接口，发现确实稳定，但通过域名访问就会出现断连。问题范围缩小到Nginx这一层。

第三步，查Nginx配置。我最初用的配置是：

proxy_pass http://backend_upstream; proxy_http_version 1.1; proxy_set_header Connection "";

少了proxy_read_timeout和proxy_buffering off这两个关键配置。默认的proxy_read_timeout是60秒，模型思考超过60秒没出token就会被断开。改成下面这样就好了：

location /chat/stream { proxy_pass http://backend_upstream; proxy_http_version 1.1; proxy_set_header Connection ""; proxy_set_header Cache-Control no-cache; proxy_set_header X-Accel-Buffering no; proxy_buffering off; proxy_cache off; proxy_read_timeout 3600s; proxy_send_timeout 3600s; proxy_connect_timeout 10s; }

第四步，检查客户端自己有没有超时。有些前端请求库默认超时时间很短，Axios默认甚至没有超时，但如果你显式设了timeout: 30000，那30秒没收到数据就断了。要确保给SSE请求单独设置较长超时，或者干脆把超时默认改大。

还有一层，部分云负载均衡器也会介入长连接的空闲超时，比如阿里云SLB默认空闲超时60秒，AWS ALB默认60秒。要么在控制台把空闲超时调大，要么通过配置心跳维持连接，让代理不判定为空闲。心跳可以发放注释行，也就是定时发送一条以冒号开头的注释行，SSE规范规定注释行会被忽略，但能有效维持连接活性。

async def heartbeat(): while True: # SSE注释行：客户端解析时会忽略，但代理不会认为是空闲连接 yield ": ping\n\n" await asyncio.sleep(15)

这个坑的本质是：你的服务端业务是健康的，但链路中的任何一环都可能“好心办坏事”把你的连接干掉。排查SSE断连，永远要从客户端到服务端逐层检查超时配置，而不是只盯着应用代码。

3. WebSocket与多轮对话：双向通道的正确实现方式

3.1 FastAPI的WebSocket端点怎么搭

SSE解决“单向不停push”的问题，但LLM应用里还有一种需求：用户随时可能打断模型输出，或者服务端检测到用户离线需要主动提示，这些都需要双向通信。聊天机器人多轮对话中用户发一句、模型回一段的场景，用WebSocket更为顺手。

FastAPI对WebSocket的支持相当直接，一个websocket装饰器就能处理：

from fastapi import FastAPI, WebSocket, WebSocketDisconnect app = FastAPI() @app.websocket("/ws/chat") async def chat_endpoint(websocket: WebSocket): await websocket.accept() try: while True: # 接收客户端消息 request = await websocket.receive_text() # 模拟LLM流式生成 tokens = ["正在", "处理", "你的", "问题", "请稍候"] for token in tokens: await websocket.send_json({ "type": "token", "content": token }) await asyncio.sleep(0.2) # 发送流结束标记 await websocket.send_json({ "type": "end", "content": "" }) except WebSocketDisconnect: print("客户端断开连接")

这段代码看着简单，但它演示了WebSocket最核心的循环模式：accept握手 → receive接收请求 → 服务端处理 → send推送结果 → 继续等下一个请求。整个对话过程在一个长连接里完成，没有HTTP连接反复建立的开销。

send_json底层会帮你做JSON序列化，比手动send_text(json.dumps(...))方便一些。需要发二进制内容，比如音频片段、图片，用send_bytes。

3.2 客户端随时打断模型输出和上下文管理

多轮对话最大的特点就是“随时可能翻车”。用户问了A问题，模型刚开始回答，用户发现问错了，紧接着又问B问题。这时候服务端如果还在跑A的推理，就会出现串场——先输出A的答案，再输出B的答案，用户角度看就是消息错乱。

我的做法是引入“最新的请求ID + 生成任务取消”机制。具体说，客户端发送的新请求带一个时间戳或者自增ID，服务端在处理新请求前，取消上一个还在进行的生成任务。

import asyncio from fastapi import WebSocket class ChatSession: def __init__(self): self.current_task = None self.current_request_id = None async def handle_new_request(self, websocket: WebSocket, request_id: str, prompt: str): # 取消上一个还在跑的任务 if self.current_task and not self.current_task.done(): self.current_task.cancel() try: await self.current_task except asyncio.CancelledError: pass # 上一个任务被取消，正常 self.current_request_id = request_id self.current_task = asyncio.create_task( self.generate_and_send(websocket, prompt) )

取消任务这个动作很重要，因为FastAPI的异步生成器在收到CancelledError时，如果里面有finally块，可以执行清理动作，比如释放占用的GPU资源、关闭内部HTTP连接等。如果放任旧任务跑完，不仅浪费算力，还可能往WebSocket里写入不属于当前请求的内容，把前端搞乱。

上下文管理则是另一个容易被忽略的点。多轮对话不是说把历史消息全发给模型就完事，要控制上下文窗口。token数超过阈值时要裁剪历史，保留最近的N轮。像这类逻辑放在WebSocket处理循环的外层，用session维度存储对话历史，避免每次请求都从头传全部消息。

3.3 WebSocket心跳机制的实现与“僵尸连接”治理

热搜词里也有“websocket心跳机制实现”，这个必须单独说一下。WebSocket长连接最大的隐患在于：网络链路是复杂的，任何一端断网都不会主动通知对方。比如用户手机WiFi突然断了，或者中间路由器重启了，TCP连接表面上还活着，实际已经完全不可用。这种连接就是“僵尸连接”，不及时清理会一直占用服务端资源。

服务端怎么发现僵尸连接？最常见的是ping/pong机制。核心逻辑是：服务端定时发ping帧（心跳探测），客户端必须回应pong帧。如果连续多次没收到pong，服务端判定连接死亡，主动关闭。

FastAPI的WebSocket没有暴露原生的ping/pong控制，因为uvicorn和Starlette在底层管理了这些。实践中我一般自己做一套应用层心跳：

async def heartbeat_check(websocket: WebSocket, last_activity: dict): while True: await asyncio.sleep(30) if time.time() - last_activity["last"] > 60: # 超过60秒没有任何活跃消息，主动断开 await websocket.close(code=4001, reason="heartbeat timeout") break

配合接收端的活跃时间更新：

while True: message = await websocket.receive_text() last_activity["last"] = time.time() # 处理业务

这里要理解一个逻辑：如果客户端还活着，它发的任何消息都会刷新活跃时间；如果客户端死了，不会有任何消息，活跃时间会一直停留在最后一条消息的时刻，超过阈值就可以判定死亡。

前端侧的心跳更简单，设置一个setInterval定时发送{"type": "ping"}，服务端收到后回一个{"type": "pong"}即可。如果一定时间内没收到pong，前端就主动重连。

但有一点要提醒：把心跳间隔设置得太频繁反而会浪费资源，网络栈本身有TCP keepalive兜底，应用层心跳只是用来做业务级检测，30秒到60秒的间隔是常见配置。间隔太短会让Nginx、负载均衡看到大量无意义包，白白增加开销。

4. HTTP基础面与连接复用：被忽略的服务端性能关键

4.1 Keep-Alive与连接池：让LLM服务端扛住并发

聊完SSE和WebSocket，回头再看HTTP本身。LLM应用里，普通HTTP仍然是基础：前端获取模型列表、查询任务状态、上传文档做知识库检索，这些都是传统请求-响应模型。但它们同样有讲究——连接复用。

HTTP/1.1默认开启Keep-Alive，同一个TCP连接可以顺序处理多个请求，避免每次请求都重建TCP连接折腾三次握手。问题在于，服务端代码写得不当时，连接复用的收益被白白浪费掉。一个典型的错误是在做LLM服务调用时，每个请求都新建一个httpx或requests的客户端对象：

# 错误示范：每次调用都创建新连接池，连接无法复用 async def call_llm_bad(prompt: str): async with httpx.AsyncClient(timeout=30) as client: resp = await client.post("http://llm-service/generate", json={"prompt": prompt}) return resp.json()

正确做法是复用一个客户端实例，让连接池存活在进程级别：

import httpx # 全局复用连接池 _client = httpx.AsyncClient( base_url="http://llm-service", timeout=httpx.Timeout(30.0, connect=10.0), limits=httpx.Limits(max_keepalive_connections=20, max_connections=100) ) async def call_llm_good(prompt: str): resp = await _client.post("/generate", json={"prompt": prompt}) return resp.json()

max_keepalive_connections控制空闲连接保持多少条，max_connections控制最大并发连接数。连接池的好处不只是省了握手开销，更重要的是在并发场景下避免每次请求都去新建连接导致瞬时连接数暴涨。只要连接还活着，新请求就能复用已建立的TCP通道，延迟和系统开销都能明显降下来。

4.2 流式HTTP响应：让HTTP也能“边生成边下发”

虽然SSE适合专门的流式推送，但如果你的前端只是需要在接口返回后处理一段较长的文本，HTTP流式响应（不带SSE格式）也有它的价值。区别在于，SSE多了一层事件格式，需要用data:前缀和空行分隔；而HTTP流式响应就是原始的字节流。

FastAPI里实现也简单，还是用StreamingResponse，只是media_type改成text/plain：

from fastapi.responses import StreamingResponse async def gen_text(): for char in "这是一段很长的返回内容": yield char await asyncio.sleep(0.05) @app.get("/text-stream") async def text_stream(): return StreamingResponse(gen_text(), media_type="text/plain")

这种接口适合的场景是：客户端拿到响应后要自己处理增量内容，不关心SSE的事件结构。有经验的工程师通常会说：“不要为了流式而流式，先想清楚数据形态。”如果是给大模型服务做内部代理，直接转发字节流就够了；如果是给浏览器前端消费，用SSE更规范。

5. 从单机Demo到生产部署：多协议并存怎么规划才不翻车

5.1 一次真实的接口拆分设计

我做的LLM客服机器人，最后落地时用的是混合架构。具体拆分情况我用一张表列出来，这种拆分也是目前业界比较通用的模式：

这个拆分的内在逻辑是：按交互模式划分，而不是按功能模块划分。同样是聊天能力，单轮问候场景用SSE就够了，客户端发起请求后只是被动接收；多轮自由对话则必须支持用户随时打断、切换话题，这只有WebSocket能自然支持。如果硬要用SSE做多轮对话，用户想打断的时候只能靠中止HTTP请求，服务端压根感知不到，非常别扭。

5.2 多个Worker下的WebSocket状态同步问题

部署时最容易踩的坑是：FastAPI应用用了多进程（uvicorn --workers 4），WebSocket被负载均衡分发到不同进程，进程之间消息不同步。

举个例子：用户A连到了Worker 1，用户B连到了Worker 2。A发消息给B，服务端把消息推送到Worker 1的连接管理器里，B在Worker 2上根本收不到。这在LLM场景很常见：一个用户可能在浏览器和App两个端同时开着会话，服务端要把事件同步推给同一个用户的所有连接。

解决方案无外乎两种。一是引入Redis Pub/Sub做跨进程消息广播：

import redis.asyncio as redis from fastapi import WebSocket redis_client = redis.from_url("redis://localhost:6379") class ConnectionManager: def __init__(self): self.connections: dict[str, WebSocket] = {} async def connect(self, user_id: str, websocket: WebSocket): await websocket.accept() self.connections[user_id] = websocket # 订阅用户专属频道 pubsub = redis_client.pubsub() await pubsub.subscribe(f"user:{user_id}") asyncio.create_task(self.listen_pubsub(pubsub, websocket)) async def publish_to_user(self, user_id: str, message: str): # 把消息发布到Redis频道，所有进程都能收到 await redis_client.publish(f"user:{user_id}", message)

二是把连接信息放Redis而不放内存，每个worker从Redis读取目标连接所在的worker，再通过IPC转发。方案二复杂一些，一般只有需要大规模集群时才用得上。单机多worker部署，Redis Pub/Sub足够。

另外还有个细节：多worker情况下，SSE的请求分布在不同进程里，但SSE是单向的，同一客户端发起的请求固定被负载均衡到某个worker就行，不需要跨进程协调。所以SSE多worker问题不大，真正的痛点在WebSocket这种有状态的双向连接上。

5.3 日志丢失、Gunicorn管理Uvicorn与生产配置参考

热搜词里有“uvicorn fastapi 日志丢失问题”，这个坑我也遇到过。用uvicorn app:app --workers 4直接启动多进程时，默认每个进程各自输出日志到标准输出，如果你用systemd管理，有时日志会混杂甚至看起来像丢失——其实是某个worker的日志写到了别的地方，或者被系统的/dev/null吃掉了。

解决办法是用Gunicorn来管理Uvicorn worker。Gunicorn作为进程管理器，统一接管worker生命周期和日志：

gunicorn app.main:app \ -k uvicorn.workers.UvicornWorker \ -w 4 \ -b 0.0.0.0:8000 \ --timeout 120 \ --graceful-timeout 30 \ --access-logfile - \ --error-logfile -

这个配置下，Gunicorn统一收集所有worker的访问日志和错误日志，不会再出现日志“丢失”的现象。--timeout 120是worker空闲超时，如果SSE长连接超过这个时间没有任务，Gunicorn会杀掉worker。所以如果你走Gunicorn管理SSE接口，要把timeout调大，或者干脆在Nginx层就不经过Gunicorn直接代理。

日志这块还有一个点：多worker下Python的logging模块输出的日志如果不同步，会出现日志顺序错乱。原因是多个进程同时写同一个文件时，操作系统不保证写入顺序一致。解决方案是把日志写到同一个syslog或者用专门的日志采集agent（如Filebeat）去抓取，避免自己多进程直接写文件。

6. FastAPI、Flask与其它框架的对比：为什么我最终选了FastAPI

6.1 异步原生、类型提示和现代API设计

热搜词里有“flask 与 fastapi 比较”，这确实是很多人纠结的点。我的立场很明确：如果你的后端核心是LLM应用这一类IO密集型任务，FastAPI优势非常明显。

这个优势根源在于FastAPI的非阻塞事件循环。LLM应用的主要耗时是等待模型推理结果，CPU本身占用不高，真正卡住的是网络IO、磁盘IO这类等待操作。异步框架能在这类等待时间里切换去处理别的请求，让同一个进程支持高得多的并发。Flask是同步框架，一个请求没处理完，其他请求就要排队；虽然也能靠跑多个进程撑并发，但资源开销大得多。

再有一个就是类型提示。FastAPI基于Python类型注解自动做请求体解析、参数校验和文档生成。这等于把你从大量手写的pydantic模型校验代码中解放出来。LLM应用的请求结构非常复杂，用户消息、历史上下文、参数配置（temperature、top_p等）一坨一坨的，手写校验容易漏，类型注解让边界清晰很多。

6.2 与AIOHTTP、Sanic、Flask异步插件的取舍

除了Flask，我也对比过AIOHTTP和Sanic。AIOHTTP功能全面但框架感更重，很多设计偏底层，开发效率不如FastAPI。Sanic性能曾经很亮眼，但它一直是“社区驱动、商业支持不足”的状态，周边生态比FastAPI差不少。

Flask要做异步，得靠quart或者gevent这类方案打补丁。不是不行，但属于“硬拗”——框架设计之初就没有为异步事件循环考虑过，补丁方案总在某个角落出问题。FastAPI的Starlette底层从一开始就基于asyncio构建，SSE、WebSocket这些长连接场景天然就是它的一等公民，不需要额外插件去绕。

我个人的建议很简单：新项目，直接FastAPI起步；老项目已经在用Flask且没有大的并发诉求，不用急着迁移重写。但如果你确认要做流式输出、WebSocket聊天、高并发连接这类LLM应用架构，从Flask迁移到FastAPI是值得的，一次迁移能把后续维护成本降下来不少。

6.3 性能边界：FastAPI的异步模型在长连接场景下的表现

有人担忧FastAPI单进程性能不够。以我实际压测结果看，纯HTTP短请求接口，单worker的FastAPI大概能稳定支持数千QPS（取决于业务逻辑复杂度）；而SSE和WebSocket这类长连接场景，瓶颈主要在内存和文件描述符，单worker维持几千个并发WebSocket连接没有问题，前提是别在事件循环里跑任何同步阻塞代码。

这一点值得展开说：FastAPI虽然异步，但如果你在端点里写了time.sleep(2)或者直接调用一个同步的requests库请求外部服务，整个事件循环都会被阻塞，这一个worker的所有连接都会卡住。异步编程有它自己的纪律——处处都要用await，任何阻塞操作都要丢给线程池或改用异步库。这也是很多从Flask转FastAPI的人最容易犯的错。

7. 阶段性避坑清单：这些是我实际踩过的真实深坑

最后整理一份“照着做就能避坑”的清单，全部来自我实际项目中的血泪教训。

注意事项

1. SSE响应头务必加Cache-Control: no-cache，否则某些浏览器和中间层会把流式响应缓存起来，结果就是用户拿到一段一次性渲染的完整文本，流式效果完全丢失。

2. 用Nginx代理SSE时，三个配置缺一不可：proxy_buffering off、proxy_read_timeout 3600s、X-Accel-Buffering no。第一个关闭代理缓冲，第二个防止空闲超时断开，第三个是告诉Nginx别缓冲这个响应。缺一个你就等着看“idle timeout”报错吧。

3. WebSocket连接必须做心跳检测，服务端要主动清理僵尸连接，否则连接泄漏会让你在并发上来后频繁遇到“EMFILE: too many open files”。

4. httpx客户端必须全局复用，并且设置合理的Limits(max_keepalive_connections, max_connections)。每请求新建客户端不仅慢，还会把端口和文件描述符耗尽。

5. 多worker部署WebSocket服务时，连接管理器一定要用Redis Pub/Sub或类似外部消息总线做跨进程同步，收到新消息要广播给其它worker的对应连接。

这些坑单独看都不难理解，但零散分布在链路的不同环节里——客户端、Nginx、Gunicorn、应用代码、系统文件描述符。LLM应用后端的特点是链路比普通Web服务长得多，一端连着用户的浏览器，一端连着模型推理服务，中间还隔着知识库、缓存、向量数据库。每一环都可能出现超时、断连、缓冲、资源泄漏这类问题，排查要靠逐层拆解。

我的经验是：不要等到线上出问题再排查，而是在开发阶段就主动用生产环境配置去跑一遍。Nginx代理、多worker、Redis同步这些配置，本地开发环境越接近生产，踩坑的周期就越短。远程模型服务不一定随时可用，可以在本地搭一个模拟延迟的假模型服务，专门用来调试流式断连、超时这类客户端和服务端的交互边界问题。

通讯架构这块，选型只是起点。真正决定项目能不能稳定跑的，是你对SSE格式的理解、对WebSocket心跳的重视、对连接池和缓冲配置的敏感度。这些细碎的工程经验，才是LLM应用后端拉开差距的地方。