SSE流式传输实战：大模型逐token输出与生产环境调优

1. 流式传输与SSE协议到底在解决什么问题第一次接触流式传输这个概念很多人脑子里冒出来的画面是水管——数据像水一样哗啦啦地流过来。这个直觉其实相当准确。传统HTTP请求的模型是“一问一答”客户端发一个请求服务端把完整结果算好一次性打包返回连接关闭。这个模式在大多数场景下够用比如你打开一个网页、拉取一份用户列表。但有一类场景它扛不住服务端需要持续不断地往客户端推送数据而且这些数据是逐步产生的不是一次性算完的。最典型的例子就是大模型对话。你问一个问题模型不是瞬间把整段回答生成完再发给你而是一个token一个token地往外吐。如果等全部生成完再返回用户盯着空白屏幕可能要等十几秒甚至更久。流式传输要解决的就是这个问题让数据边产生边传输客户端边接收边渲染用户能实时看到内容逐渐出现。SSE全称Server-Sent Events是HTML5规范里定义的一种技术专门用于服务端向客户端单向推送文本数据。它建立在HTTP协议之上用起来比WebSocket轻得多。WebSocket是全双工通信客户端和服务端可以互相发消息适合聊天室、协同编辑这类双向交互场景。SSE只做一件事服务端推客户端收。这个单向特性看起来是限制但在大模型对话、实时日志推送、股票行情更新这类场景里恰恰是最合适的——因为客户端本来就不需要往服务端发东西发请求用普通HTTP就够了。我刚开始做大模型应用的时候第一版用的是轮询前端每隔一秒发一次请求问“生成完了吗”。这个方案能用但问题很明显——大量无效请求服务端压力大而且实时性受轮询间隔限制。后来换成WebSocket功能上没问题但引入了一套完整的长连接管理逻辑心跳、重连、消息格式定义代码复杂度上去了。直到用上SSE才发现对于“服务端单向推送文本”这个需求它是最简洁的解法基于HTTP不需要额外协议升级浏览器原生支持自动重连服务端实现也简单。这篇文章适合谁看如果你正在做大模型应用的前端或后端开发需要实现“打字机效果”的实时输出如果你在做实时监控面板需要服务端主动推送状态更新如果你只是好奇“为什么ChatGPT的回答能一个字一个字蹦出来”想搞清楚背后的技术原理——那这篇内容就是为你准备的。我会从协议原理讲到代码实现从参数调优讲到踩坑经验尽量把每个环节都讲透。2. SSE协议的核心机制与工作原理拆解2.1 SSE的通信模型一条特殊的HTTP长连接SSE的本质是一条特殊的HTTP连接。说它特殊是因为它虽然走HTTP协议但行为模式和普通请求完全不同。普通HTTP请求是“请求-响应-关闭”的短连接模型而SSE是“请求-保持连接-持续推送”的长连接模型。具体来说客户端发起一个普通的HTTP GET请求但在请求头里带上Accept: text/event-stream。服务端收到这个请求后如果支持SSE会在响应头里返回Content-Type: text/event-stream然后保持这条连接不关闭持续往客户端写数据。客户端收到响应后也不会关闭连接而是持续读取服务端推送过来的数据流。这里有个关键点SSE连接是单向的。客户端发起连接后就只能接收数据不能再通过这条连接往服务端发消息。如果需要双向通信得另外开HTTP请求或者直接用WebSocket。这个设计取舍很明确——SSE把“服务端推送”这件事做到了最简不引入额外的协议复杂度。从网络分层来看SSE完全跑在HTTP/1.1或HTTP/2之上。这意味着它天然穿透代理、防火墙、CDN不需要像WebSocket那样做协议升级握手。很多企业内网环境对WebSocket有限制但SSE因为就是普通HTTP基本不会被拦。这是它在实际部署中的一大优势。2.2 数据格式event、data、id、retry四个字段SSE的数据格式非常简洁用纯文本传输每条消息由若干字段组成字段之间用换行分隔消息之间用空行分隔。规范定义了四个字段data消息内容这是唯一必须的字段。可以有多行每行都以data:开头。event事件类型客户端可以根据不同类型做不同处理。不指定时默认是message类型。id消息ID用于断线重连时告诉服务端“我上次收到的是哪条”方便服务端从断点继续推送。retry重连时间间隔单位毫秒。服务端通过这个字段告诉客户端“如果断了等这么久再重连”。一个典型的消息长这样event: token data: {content: 你} event: token data: {content: 好} event: done data: {finish_reason: stop}注意每条消息末尾必须有一个空行这是消息分隔符。没有空行客户端会把多条消息当成一条。这个细节在手动拼接SSE响应时特别容易出错我后面会专门讲。2.3 自动重连机制浏览器帮你兜底SSE最被低估的特性是浏览器原生支持的自动重连。当连接意外断开时浏览器会自动重新发起连接不需要前端写任何重连逻辑。重连时浏览器会把最后收到的消息ID通过Last-Event-ID请求头发给服务端服务端可以根据这个ID决定从哪继续推送。这个机制在实际生产环境里价值巨大。移动网络切换、WiFi信号波动、服务端重启这些都会导致连接断开。如果用WebSocket你得自己实现心跳检测、重连退避、消息补偿代码量不小。SSE把这些都内置了前端只需要监听onerror事件做点日志记录就行。不过自动重连也有坑。默认重连间隔是3秒左右如果服务端挂了客户端会每3秒重试一次可能造成大量无效请求。服务端可以通过retry字段调整这个间隔比如设成10000毫秒降低重试频率。另外如果服务端返回了非200状态码比如401未授权浏览器会停止重连这时候需要前端手动处理。2.4 与WebSocket、轮询的对比选型选型这件事没有绝对的对错关键看场景。我把三种方案的核心差异整理成表方便对照维度SSEWebSocket轮询通信方向服务端到客户端单向全双工双向客户端主动拉取协议基础HTTP独立协议需升级握手HTTP浏览器支持原生EventSource原生WebSocket原生fetch/XHR自动重连原生支持需自行实现不适用数据格式文本文本二进制任意代理穿透好可能被限制好实现复杂度低中低但请求量大适用场景大模型输出、日志推送、行情更新聊天室、协同编辑、游戏低频状态检查大模型对话场景选SSE核心原因是需求本身就是单向的服务端推token客户端不需要在生成过程中发消息而且输出是纯文本。用WebSocket属于杀鸡用牛刀多出来的双向能力用不上反而增加了连接管理的复杂度。轮询则完全不适合因为token生成是连续的轮询间隔无论设多小都会有延迟设太小又浪费资源。3. 大模型场景下SSE流式输出的完整实现3.1 服务端从模型输出到SSE事件流服务端的核心任务是把模型逐token生成的输出转换成SSE格式的事件流推给客户端。以Python生态为例如果用FastAPI可以这样实现from fastapi import FastAPI from fastapi.responses import StreamingResponse import json import asyncio app FastAPI() async def generate_tokens(prompt: str): # 模拟模型逐token生成 tokens [流, 式, 传, 输, 很, 好, 用] for i, token in enumerate(tokens): yield { event: token, id: str(i), data: json.dumps({content: token}, ensure_asciiFalse) } await asyncio.sleep(0.1) # 模拟生成延迟 yield { event: done, data: json.dumps({finish_reason: stop}) } def format_sse(event: dict) - str: lines [] if event in event: lines.append(fevent: {event[event]}) if id in event: lines.append(fid: {event[id]}) lines.append(fdata: {event[data]}) return \n.join(lines) \n\n app.get(/chat/stream) async def chat_stream(prompt: str): async def event_generator(): async for event in generate_tokens(prompt): yield format_sse(event) return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no } )这段代码有几个关键点值得展开说。StreamingResponse是FastAPI提供的流式响应类它接收一个异步生成器每产出一段内容就立即发给客户端不等待全部生成完。这是实现流式的核心。format_sse函数负责把事件字典转成SSE规范格式。注意末尾的\n\n——这是消息分隔符少了它客户端会把多条消息粘在一起。我见过不少新手在这里翻车前端收到的数据一直累积不触发onmessage排查半天发现是分隔符问题。响应头里的X-Accel-Buffering: no是给Nginx看的。Nginx默认会缓冲响应内容等攒够一定大小才发给客户端这会破坏流式效果。加上这个头告诉Nginx不要缓冲。如果你用的不是Nginx而是其他反向代理需要查对应配置。Cache-Control: no-cache防止中间层缓存SSE响应。SSE是实时数据缓存了就没意义了。3.2 客户端EventSource的使用与边界浏览器端用EventSource接收SSE流代码很简洁const eventSource new EventSource(/chat/stream?prompt你好); eventSource.addEventListener(token, (e) { const data JSON.parse(e.data); appendToChat(data.content); }); eventSource.addEventListener(done, (e) { const data JSON.parse(e.data); console.log(生成完成原因, data.finish_reason); eventSource.close(); }); eventSource.onerror (err) { console.error(SSE连接出错, err); // 浏览器会自动重连这里只做日志 };EventSource的API设计很直观addEventListener监听特定事件类型onerror处理错误close主动关闭连接。注意done事件里要调用close()否则连接会一直保持浪费资源。但EventSource有几个硬限制实际项目中经常需要绕过第一它只支持GET请求不能发POST。大模型对话的prompt可能很长放URL里不合适。解决方案是用fetch配合ReadableStream手动解析SSE流这样可以用POST也能自定义请求头。第二它不能自定义请求头。如果需要传Authorization tokenEventSource做不到。同样需要用fetch方案。第三它不能中途取消。虽然可以close()但那是关闭整个连接不能只取消当前这次生成。大模型场景里用户经常想“停止生成”需要更细粒度的控制。3.3 用fetch替代EventSource更灵活的方案fetch方案的核心是读取response.body这个ReadableStream手动按SSE格式解析async function streamChat(prompt, onToken, onDone, signal) { const response await fetch(/chat/stream, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer token }, body: JSON.stringify({ prompt }), signal // AbortController的signal用于取消 }); const reader response.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const messages buffer.split(\n\n); buffer messages.pop(); // 最后一段可能不完整留到下次 for (const msg of messages) { const lines msg.split(\n); let eventType message; let data ; for (const line of lines) { if (line.startsWith(event:)) { eventType line.slice(6).trim(); } else if (line.startsWith(data:)) { data line.slice(5).trim(); } } if (eventType token) { onToken(JSON.parse(data).content); } else if (eventType done) { onDone(JSON.parse(data)); } } } }这段代码里有个容易忽略的细节buffer的处理。网络传输是分片的一次reader.read()拿到的数据可能刚好在一条消息中间断开。所以要用buffer累积按\n\n切分最后一段不完整的留在buffer里等下次。如果不做这个处理偶尔会出现JSON解析失败而且这种bug很难复现因为取决于网络分片时机。AbortController是实现“停止生成”的关键。调用controller.abort()会中断fetch请求服务端检测到连接断开后停止生成。服务端那边需要监听连接断开事件及时释放模型资源。以FastAPI为例可以在生成器里捕获asyncio.CancelledError来做清理。3.4 服务端如何感知客户端断开这一点很多人会忽略。用户点了“停止生成”前端abort了请求但服务端如果不知道还在傻傻地跑模型白白消耗算力。服务端需要检测连接断开并停止生成。在FastAPI里可以通过request.is_disconnected()轮询检测from fastapi import Request app.post(/chat/stream) async def chat_stream(request: Request): body await request.json() async def event_generator(): async for token in model.generate(body[prompt]): if await request.is_disconnected(): # 客户端断开了停止生成 break yield format_sse({event: token, data: json.dumps({content: token})}) return StreamingResponse(event_generator(), media_typetext/event-stream)request.is_disconnected()是异步方法每次调用会检查底层连接状态。在生成循环里定期检查一旦发现断开就跳出循环。检查频率不用太高每个token检查一次就够了开销很小。如果用其他框架原理类似监听连接关闭事件设置一个标志位生成循环里检查这个标志位。Node.js的Express里可以用req.on(close, ...)Go的net/http里可以用r.Context().Done()。4. 生产环境部署的坑与调优经验4.1 反向代理配置Nginx的缓冲陷阱SSE在生产环境最常见的坑就是反向代理缓冲。Nginx默认开启proxy_buffering会把上游响应攒在缓冲区里等攒够一定大小或连接关闭才发给客户端。这对普通请求是优化对SSE是灾难——客户端会一直收不到数据直到缓冲区满或超时。解决方案是在Nginx配置里针对SSE路径关闭缓冲location /chat/stream { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_set_header Connection ; proxy_http_version 1.1; chunked_transfer_encoding off; proxy_read_timeout 3600s; }proxy_buffering off关闭缓冲proxy_cache off关闭缓存proxy_http_version 1.1确保用HTTP/1.1支持chunked传输proxy_read_timeout设大一点防止长连接被超时切断。Connection 是清除默认的Connection: close头保持长连接。如果用的是云厂商的负载均衡也要检查是否有类似的缓冲配置。有些云LB默认会缓冲响应需要在控制台里关掉。这个坑我踩过本地测试一切正常部署到线上就变成“一次性返回”排查了半天才发现是LB的锅。4.2 心跳与超时保持连接存活SSE连接是长连接中间任何一环客户端、代理、服务端都可能因为超时把连接断掉。为了保持连接存活需要定期发送心跳。心跳的实现很简单服务端每隔一段时间比如15秒发一条注释消息。SSE规范里以冒号开头的行是注释客户端会忽略但能起到保活作用async def event_generator(): last_heartbeat time.time() async for token in model.generate(prompt): if time.time() - last_heartbeat 15: yield : heartbeat\n\n last_heartbeat time.time() yield format_sse({event: token, data: json.dumps({content: token})})心跳间隔要小于所有中间环节的超时时间。Nginx的proxy_read_timeout默认60秒云LB通常也是60秒左右所以15到30秒的心跳间隔比较安全。太频繁浪费带宽太稀疏起不到保活作用。4.3 并发连接数SSE的资源消耗评估SSE连接是长连接每个连接占用一个服务端线程或协程。如果服务端是同步阻塞模型比如传统的FlaskWSGI每个SSE连接会占住一个worker并发数上不去。假设你有4个worker那最多只能支撑4个并发SSE连接第5个用户就得等。所以SSE服务端必须用异步模型。FastAPIuvicorn、Node.js、Go的net/http都是异步的单机支撑几千个并发连接没问题。如果用的是Flask建议换成FastAPI或QuartFlask的异步版本。另外要注意文件描述符限制。每个SSE连接占用一个fdLinux默认单进程fd上限是1024高并发场景需要调大ulimit -n 65535这个值可以在启动脚本里设置或者改/etc/security/limits.conf永久生效。4.4 断线重连的数据补偿SSE的自动重连会带上Last-Event-ID服务端可以根据这个ID决定从哪继续。但大模型场景有个特殊性token是实时生成的服务端不一定缓存了历史token。如果客户端断线重连服务端可能没法从断点继续。实际项目里通常有两种处理方式。一种是服务端缓存最近生成的token序列重连时根据Last-Event-ID从缓存里取。这种方式适合短对话缓存开销可控。另一种是重连后重新生成前端把已收到的内容清掉重新渲染。这种方式简单但用户体验差适合对实时性要求不高的场景。我一般用第一种在服务端用Redis存最近几分钟的token序列key是会话IDvalue是token列表。重连时根据Last-Event-ID从列表里取后续token。缓存设置过期时间比如5分钟过期后重连就重新生成。5. 常见问题排查与避坑速查5.1 前端收不到数据或数据一次性到达这是最高频的问题九成以上是缓冲导致的。排查顺序如下排查点检查方法解决方案Nginx缓冲看响应头是否有X-Accel-Buffering加X-Accel-Buffering: no或改Nginx配置云LB缓冲本地直连正常走LB异常控制台关闭LB缓冲服务端缓冲检查是否用了StreamingResponse确保用流式响应不是普通返回客户端解析看onmessage是否触发检查消息分隔符\n\n是否正确Content-Type看响应头必须是text/event-stream我遇到过一次特别隐蔽的服务端用了gzip压缩中间件把SSE响应压缩了导致客户端收到的是压缩后的数据解析失败。SSE响应不应该压缩需要在中间件里排除text/event-stream类型。5.2 连接频繁断开重连如果日志里看到大量重连记录通常是心跳没做好或超时设置太短。检查服务端心跳间隔是否小于代理超时时间检查proxy_read_timeout是否设得够大。另外如果服务端生成速度很慢比如模型推理卡住长时间没有数据输出也会触发超时。这种情况下心跳机制就特别重要即使没有token输出也要定期发心跳。5.3 中文乱码问题SSE默认用UTF-8编码但有些服务端框架默认用其他编码导致中文乱码。确保响应头里Content-Type: text/event-stream; charsetutf-8服务端输出时用UTF-8编码。Python里json.dumps要加ensure_asciiFalse否则中文会被转成\uXXXX形式虽然客户端能解析但可读性差。5.4 内存泄漏排查SSE长连接如果管理不当容易造成内存泄漏。常见原因有两个一是连接关闭后没有清理相关资源比如生成器没退出、缓存没释放二是EventSource对象没有正确close导致连接堆积。排查方法在服务端记录当前活跃连接数观察是否持续增长不下降。在客户端确保每个EventSource在不需要时调用close()React组件里要在useEffect的清理函数里关闭连接。5.5 实操心得汇总做了几个大模型流式项目后我总结了这几条经验都是文档里不会写的提示SSE的data字段如果包含换行需要拆成多个data:行客户端会自动用换行拼接。直接在一个data:里塞带换行的字符串会导致解析错误。注意EventSource的自动重连在收到非200响应时会停止。如果服务端返回401客户端不会重连需要前端捕获onerror后手动处理token刷新再重建连接。提示开发阶段可以在Chrome DevTools的Network面板里看SSE流但注意DevTools本身可能缓冲导致看起来不是流式的。用curl -N命令测试更准确。注意如果服务端用Python的print输出SSE数据记得加flushTrue否则数据会攒在缓冲区里不发出。用StreamingResponse则不需要担心这个。提示大模型场景下建议在SSE流里加一个usage事件在生成结束时推送token消耗统计。前端可以展示给用户也方便做计费。6. 从SSE延伸出去的几个实用方向SSE的适用范围远不止大模型对话。任何“服务端持续产生数据、客户端实时展示”的场景SSE都是比WebSocket更轻的选择。实时日志推送是一个典型场景。CI/CD流水线运行时构建日志是持续产生的用SSE推给前端用户能实时看到构建进度。实现方式和大模型对话几乎一样只是数据源从模型换成日志文件。股票行情更新也适合SSE。行情数据是服务端主动推送的客户端只需要展示不需要往服务端发消息。用SSE比WebSocket简单而且天然支持断线重连网络波动后自动恢复。IoT设备状态监控同样适用。设备上报的状态变化通过SSE推给监控面板前端实时更新。如果设备数量多、状态变化频繁SSE的单向特性正好匹配——服务端推客户端展示不需要双向通信。如果后续要支持双向通信比如用户在生成过程中想插话调整方向那就需要升级到WebSocket或者用“SSE推 普通POST发”的组合方案。后者在不少大模型应用里已经在用了生成过程用SSE推流用户发送新消息用普通POST请求。这样既保留了SSE的简洁又实现了双向交互。我在实际项目里还遇到过一个需求多个用户同时观看同一个AI生成任务。这种场景下服务端可以把同一个SSE流广播给多个客户端每个客户端独立连接服务端维护一个订阅列表。实现上可以用发布订阅模式模型生成一个token就推给所有订阅者。这个模式在协作场景里很有用比如多人同时看AI写代码。最后分享一个小技巧调试SSE的时候用curl -N -H Accept: text/event-stream http://localhost:8000/chat/stream命令-N参数关闭curl自己的缓冲能实时看到服务端推送的每一条消息。比在浏览器里调试直观得多也不受DevTools缓冲的干扰。