写这篇东西的起因是前两天有个朋友在群里问为什么现在这些AI大模型应用回答问题时都是一个字一个字往外蹦的像极了老式打字机就不能一次性把整段答案甩过来吗我当时第一反应就是这背后就是SSEServer-Sent Events在做推送。说真的SSE这个技术不算新HTTP/1.1时代就有雏形了但这两年因为AI大模型爆发式落地它一下子从后端小透明变成了流式输出的主力方案。你要是准备做AI应用开发、搞智能客服或者只是想弄明白大模型回答为什么是打字机式的这篇东西应该能帮你把来龙去脉捋清楚。1. 大模型回答为什么是一个字一个字蹦出来的1.1 模型推理的token机制它不是等待而是生成很多人有个误解觉得大模型是在一个巨大的知识库里查答案查到了就一次性返回。实际完全不是这么回事。大模型的工作方式是逐tokentoken可以理解为词元一个英文单词可能对应1到2个token一个中文汉字大概对应1到2个token地生成内容。每生成一个token它都要基于前面已经生成的所有token重新做一次概率计算挑出概率最高的下一个token。我举个直观点的类比你让AI写一段年终总结它其实是先预测今年这个token然后基于今年再预测工作或者业绩每一步都要跑一遍完整的神经网络前向计算。所以从模型本身的角度看回答是天然串行生成的它没办法在生成完第一个token之前就告诉你后面的内容。这个过程耗时往往在几秒到几十秒如果等到全部生成完再统一返回用户在页面上就干瞪眼等着体验极差。这里就会牵扯出一个关键问题既然结果是逐步生成的那服务端能不能边生成边把token送给浏览器这里就需要一种能让服务端主动推送的技术。很多人第一反应是WebSocket这个思路对但有点杀鸡用牛刀。AI场景下大模型的token流是典型的单向传输——服务端生成什么客户端就接收什么客户端基本不需要反向给服务端发指令最多发一个停止生成。这种单向、持续的流式推送SSE反而是更轻量、更合适的选择。1.2 用户体验和网络交互流式返回的价值再往深了说流式返回不只是让用户早点看到字它还有两个实打实的影响。第一个是感知速度。心理学上有个概念叫系统响应时间感知当等待时间超过1秒用户就会明显感到卡顿超过5秒用户大概率会关掉页面或重新提问。SSE把首字返回时间压缩到了几百毫秒用户看到字在动会下意识觉得它在认真思考等待焦虑被显著化解。这一点在商业场景客服机器人、智能写作助手里直接关系到转化率和留存。第二个是错误成本。如果非流式返回用户等30秒后收到一个跑偏的或者涉及敏感内容被后端拦截的完整回答这30秒就彻底浪费了。而流式输出允许我们边生成边做内容审核发现苗头不对可以立即中断连接止损成本低得多。也有不少团队利用SSE做逐段翻译、流式语音合成本质都是利用了这个边生成边传输的特性。2. SSE技术拆解一条HTTP连接如何实现持续推送2.1 SSE的协议本质text/event-streamSSE的全称是Server-Sent Events服务端推送事件。从协议层面看它仍然走的是普通HTTP只不过服务端在响应头里明确告诉客户端我要用text/event-stream这个MIME类型给你返回内容而且这段内容不会马上结束连接会保持打开持续推送数据块。用代码来理解更直观。一个最基本的SSE响应后端返回的数据格式约定如下data: 你好欢迎使用AI助手 data: 今天天气不错 data: [DONE]每一行以data:开头后面跟实际内容。两个数据之间用空行分隔。客户端这边浏览器原生提供了一个叫EventSource的API专门用来处理这种格式。它收到服务端的数据后会自动触发onmessage回调把data:后面的内容取出来。我们可以把SSE理解成一根水管HTTP请求是水管的入口服务端是水源水源端不断往管子里灌水客户端在水管出口拿杯子接。水管本身不需要反复拆装接一次就能一直用。这里面和普通HTTP最本质的区别在于普通HTTP是请求-响应一次闭环SSE则是响应头返回后连接不关闭把响应体当成一条持续的数据通道。2.2 三种推送方案对比轮询、WebSocket、SSE的取舍我在做AI应用架构时经常被问到为什么不用WebSocket反而用SSE这里我详细对比一下。其实没有绝对的谁好谁坏关键是看场景。对比项短轮询WebSocketSSE通信方向客户端主动请求服务端被动响应全双工双向通信服务端单向推送协议基础HTTP独立的WS协议需握手升级HTTP底层就是普通TCP连接连接开销极低每次请求完就释放高需要状态维护和心跳保活低HTTP长连接断线可自动重连重连机制无需业务层处理需自己实现浏览器EventSource内置自动重连实现复杂度低但实时性差高需要处理二进制帧、粘包、拥塞控制低纯文本协议后端只需往响应流里写字符串适用场景低频状态查询聊天、游戏对战、协同编辑AI流式输出、股票行情、日志推送WebSocket确实是全双工的服务器也能主动给客户端发消息但从AI问答的实际需求看这个全字对应的复杂度是多余的。我见过有团队用WebSocket做AI流式返回结果要处理连接鉴权、心跳保活、消息帧解析、断线重试状态机前后端联调还经常因为帧格式不统一产生bug。而SSE用EventSource接数据代码量少一半还不止。说个使用体验上的差异SSE断线后浏览器会自动重连并带上上次收到的Last-Event-ID字段服务端可以根据这个字段续传这个特性在弱网环境太实用了。WebSocket断线后业务侧得自己设计重连和续传逻辑稍不留神就会丢消息。2.3 选型建议什么场景必须用SSE什么场景别硬上基于上面这些对比我提炼几条选型建议纯服务端到客户端的信息推送尤其是AI生成、实时日志、行情推送首选SSE。它足够简单不需要引入额外的协议依赖。需要双向实时交互的比如在线白板协同、多人在线游戏必须用WebSocket。SSE是单向的客户端没法通过同一连接向服务端发送业务数据。实时性要求不高只要几秒钟刷一次短轮询或长轮询就够了没必要上SSE。你如果要做移动端原生APP的流式对话得注意一下iOS的NSURLSession和Android的OkHttp对SSE的支持都不错但部分老版本HTTP状态码处理有坑后面我会专门讲排障。有些后端同学会担心SSE占用连接时间太长导致服务器连接数膨胀。这个问题确实存在但它和WebSocket占用的连接资源差不了太多只要做好超时控制和连接数监控问题不大。而且SSE底层是普通HTTP完全可以通过Nginx、CDN做负载均衡和缓存比起WebSocket需要专门处理跨域、协议升级SSE在基础设施层面友好得多。3. 手把手实现一个AI流式问答系统3.1 后端FastAPI实现SSE接口我用Python的FastAPI做后端示例因为目前主流的大模型推理框架不管是接OpenAI风格API还是本地跑transformers几乎都有Python SDKFastAPI的异步特性也天然适合做SSE。先看核心代码from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app FastAPI() async def mock_ai_response(prompt: str): 模拟大模型逐token生成过程。 实际项目中这里会调用真实的大模型推理接口 比如openai.ChatCompletion.create(streamTrue)。 words list(你好我是一个AI助手很高兴为你服务。) for word in words: # 这里模拟大模型生成一个token的耗时 await asyncio.sleep(0.1) # SSE格式要求每条消息以data:开头空行结尾 yield fdata: {json.dumps({content: word}, ensure_asciiFalse)}\n\n # 流式结束标记约定俗成用[DONE] yield data: [DONE]\n\n app.post(/chat/stream) async def chat_stream(payload: dict): prompt payload.get(prompt, ) return StreamingResponse( mock_ai_response(prompt), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, # 这个后面排障会细说 } )这里有个容易踩的坑StreamingResponse的media_type必须写text/event-stream否则浏览器EventSource会拒绝解析。另外每条SSE消息必须以\n\n结尾这是协议规定的分隔符漏了客户端就会把所有data:行拼接在一起。实际项目中你的mock_ai_response函数要替换成真实的大模型调用。如果是OpenAI风格的API代码大概是这样的from openai import AsyncOpenAI client AsyncOpenAI( api_keyyour_api_key, base_urlyour_model_endpoint ) async def real_ai_response(prompt: str): stream await client.chat.completions.create( modelyour-model-name, messages[{role: user, content: prompt}], streamTrue, # 开启流式这点不能忘 ) async for chunk in stream: delta chunk.choices[0].delta if delta and delta.content: yield fdata: {json.dumps({content: delta.content}, ensure_asciiFalse)}\n\n yield data: [DONE]\n\n注意设置streamTrue这一步很多新手漏传这个参数结果拿到的是完整文本SSE流自然也就变成一次性吐出一坨了。3.2 前端EventSource的接收与解析前端反而是最省事的浏览器原生API直接上// 建立连接 const eventSource new EventSource(/chat/stream, { // 注意EventSource默认是不支持自定义Header的后面排障会专门说 }); let currentText ; // 收到默认消息事件 eventSource.onmessage function(event) { if (event.data [DONE]) { eventSource.close(); console.log(流式输出结束); return; } try { const payload JSON.parse(event.data); currentText payload.content; // 把currentText实时渲染到页面上 document.getElementById(answer).textContent currentText; } catch (e) { console.error(解析SSE消息失败, e); } }; // 连接异常EventSource内部会自动重连 eventSource.onerror function(event) { console.warn(SSE连接异常浏览器正在自动重连); }; // 可选手动中断 function stopGenerate() { eventSource.close(); // 同时最好调一个后端接口通知模型停止生成省算力 fetch(/chat/stop, { method: POST, body: JSON.stringify({ taskId: currentTaskId }) }); }CSS方面我建议在流式输出时给文本容器加一个光标闪烁效果视觉上更像打字机这个小细节对用户体验提升很明显代码就不贴了大家可以自行发挥。EventSource的onmessage默认监听的是没有event:字段的消息。如果后端想发送不同类型的消息比如event: error可以用addEventListener(error, handler)来监听。我在实际项目中用这种方式来区分正常token和错误提示比把所有类型混在onmessage里友好得多。3.3 处理流式数据token块拼接与消息结束标志有个细节非常重要SSE的data:行里传输的数据并不一定是一个完整的词。大模型返回的chunk可能是你也可能是你好的一部分最稳妥的做法是把每一个data:都视为一段独立的增量前端只做currentText chunk的拼接不要尝试在中间状态里做语义完整判断。我来解释一下为什么。以中文场景为例大模型的tokenizer切词方式并不完全按字来一个词可能被切成大和模型两个token但返回顺序依然是线性的。如果你在拼接过程中做中文分词、关键词提取、或者语法标点的二次加工很容易因为上下文不完整而误判。我见过有人在SSE回调里对每一小段做敏感词过滤结果过滤和策略拆成了两个token中间态有滤策这种奇怪文本过滤正则直接命中误报。正确做法是先拼接完流式内容再做整句级别的后处理。消息结束标志目前行业里流行data: [DONE]这是OpenAI带起来的约定很多自建推理框架也照搬了。严格说这不是SSE协议的一部分属于应用层约定所以前后端必须对齐。前端判断[DONE]后要立即close()连接避免服务端已经断开但客户端还挂着空连接。4. 真实场景对接大模型API时的SSE协议细节4.1 OpenAI风格流式协议data和[DONE]的江湖规矩现在市面上几乎所有大模型API包括各类开源模型的商业化托管在流式模式下都遵循OpenAI定下的格式每个chunk是一条SSE消息data:后面跟一个JSON从choices[0].delta.content里取增量文本最后以data: [DONE]收尾。这个江湖规矩的好处是兼容性好一套解析代码可以通吃主流的模型平台。坏处是各家会在JSON里塞一些额外的字段比如usage统计、logprobs、自定义元数据解析时要注意容错。我在对接时写过这么一段通用解析逻辑async def parse_llm_chunk(raw_text: str): 解析一行SSE数据返回增量内容和原始chunk if raw_text.startswith(data:): payload raw_text[5:].strip() if payload [DONE]: return None, None, True # 表示结束 ...有一类特殊情况有些API在流式过程中还会发ping或keepalive消息data内容是空字符串或者注释行以:开头。如果你的解析器直接把空字符串当成完整内容回传给前端页面上会出现空行闪烁。正确姿势是忽略所有没有实际content增量的chunk。4.2 增量与全量为什么有的API要传streamTrue才有流式效果这里要展开说一下增量输出和全量输出的区别。当你不传streamTrue时服务端会等模型把整个回答生成完毕后一次性返回完整JSON。这种方式用SSE包装其实毫无意义——你的后端拿到的已经是一整段文本了再分段推给前端纯属脱裤子放屁。当你传了streamTrue后服务端的行为变成模型每生成一个token立即通过HTTP连接推送一个chunk。这时候你的后端代码才能做到边收边推。我遇到过不少朋友在业务代码里忘了传这个参数然后跑来问为什么我的SSE接口还是等半天才出数据。所有大模型推理框架包括OpenAI官方接口、Anthropic、Ollama的/api/chat流式开关都是显式参数不传就等于关掉了。顺带提一句如果你想给用户展示生成耗时统计不要在SSE连接开始时记时间、结束后算差值这样误差很大。正确做法是在每个chunk里记录服务端时间戳前端展示最后一条消息解析时的时间差这样统计到的是真实的端到端首字耗时和总耗时。4.3 实测踩坑EventSource的跨域与携带Token问题在讲第5个大节前我先插一段实测踩坑记录。因为我发现这个坑几乎每个人都会遇到。事情是这样的前端用EventSource连接后端/chat/stream这个接口需要鉴权前端把token放在Authorization: Bearer xxx头里。结果跑起来发现EventSource这个API根本没有办法自定义请求头它只支持同源URL、Cookie传递鉴权。这个问题是浏览器对EventSource的硬性限制不是后端能解决的。我当时给了两个解决方案。方案一把token放在URL的query参数里比如/chat/stream?tokenxxx后端从query里取。简单直接但token会出现在日志里安全上要自己做防护最好给这个token设置短时效且只允许单次使用。方案二先请求一个普通接口获取一次性会话ID然后用这个ID去建立SSE连接连接建立后做一次绑定校验。流程稍微复杂但安全性高很多。另外在跨域场景下要特别注意CORS配置。大多数后端框架对SSE的CORS支持有坑你必须显式允许text/event-stream这个Content-Type并且Access-Control-Allow-Origin不能设为通配符*因为EventSource不允许credentials模式配通配符。我把这个坑写在前面后面排障大节里就不重复说了。5. 常见问题与排查技巧实录5.1 前端一切正常但就是收不到推送后端日志显示在发这个问题在本地开发时最常出现。浏览器收到SSE流的前几个字符后就再也没动静了。查后端日志数据明明在持续写入。最典型的元凶是Nginx缓冲或者是企业网关的缓冲。Nginx默认对SSE这类响应会做了缓冲它会把后端吐出的数据攒到一定大小或者等连接结束才一次性发给客户端流式效果直接失效。解决办法是在Nginx配置里关掉该路径的缓冲location /chat/stream { proxy_buffering off; proxy_cache off; chunked_transfer_encoding on; proxy_http_version 1.1; proxy_set_header Connection ; }同时后端响应头加X-Accel-Buffering: no这是给Nginx看的专用头告诉它别缓存这个响应。如果你们前面还有CDN或企业级WAF同样要检查它们是否对text/event-stream有特殊的缓冲策略。当然还有一个常见低级问题后端用了WSGI服务器比如Flask自带的开发服务器这类服务器对长连接支持不稳定写SSE建议用uvicorn或gunicorn(gevent worker)这类ASGI服务器。5.2 断线重连EventSource自动重连为什么会带来重复消息EventSource自带断线重连默认重试间隔大概几秒3秒左右。这个机制是好的但如果服务端已经向前端推送了部分内容断线重连后前端从头接收就会出现重复token现象——你看到页面上同一句话被拼了两遍。规避办法是合理利用SSE协议的id:字段。后端在每条消息里带上序号id: 1 data: {content: 你} id: 2 data: {content: 好}浏览器断线重连时会自动带上Last-Event-ID: 1的请求头服务端读取到这个头后从第2条消息开始推而不是从头推。这个机制是SSE协议内置的实现成本很低强烈建议用。注意如果你的应用是无状态多副本部署Last-Event-ID的续传逻辑要放到Redis这类共享存储里否则用户请求打到另一个Pod上它根本不知道之前推到哪里了。另外还有一种策略如果业务上能接受丢增量、不能接受重复那么重连后前端可以清空文本重新生成。这个方案实现简单但对用户不友好。我做过一个折中方案重连后前端先把已渲染文本暂存如果新流从头推对比发现头部字串一致就去掉重复叠加的那段再继续接入。效果还行就是代码复杂一点。5.3 输出到中文字符乱码问题这个坑非常隐蔽。SSE默认是UTF-8文本流但如果你的后端项目没有统一设置编码某些中间件会自作主张按Latin-1或GBK解码后再转发前端拿到的就是æ¥å¥½这种乱码。排查步骤很简单用curl直接打后端SSE接口看原始字节流curl -N -H Accept: text/event-stream http://localhost:8000/chat/stream如果curl看到的是正常中文说明后端没问题问题出在中间转发层网关、负载均衡、浏览器的预解析。如果curl直接就是乱码那就是后端响应流编码不对检查框架的默认字符集配置。FastAPI/Starlette默认就是UTF-8一般不会翻车容易翻车的是Java系的框架和某些老旧的Tomcat配置。我额外加一个建议所有SSE消息里的文本传输统一推荐用JSON字符串包裹如data: {content:你好}不要直接裸传data: 你好。因为JSON字符串里的中文会被转成\u4f60\u597d这种ASCII形式的Unicode转义传输过程中无论如何编码都不会乱还能天然规避数据里的换行和特殊字符把SSE协议搞坏。这是从实际项目里总结出的最稳方案。5.4 连接耗尽长连接多了以后服务器端口不够用SSE的本质是HTTP长连接每个连接会长时间占用一个socket。当并发用户量上去后服务器可能出现端口耗尽、文件描述符耗尽。这里分享几个实践经验操作系统层面调大net.core.somaxconn、ulimit -n已经是老生常谈不再赘述。应用层一定要设置空闲超时和心跳机制。即使SSE也不能让连接永远挂着。我一般在服务端每隔15秒发一条: heartbeat注释行以冒号开头的行是SSE协议规定的注释消息客户端不会触发onmessage但能维持TCP活跃客户端30秒内没收到任何消息就主动重连。能直连就不要过太多层转发。每经过一层代理连接的保活和超时配置都要对齐稍有不慎就会造成客户端连接已断但服务端不知道的半开连接堆积起来迟早出事。上Kubernetes后SSE和负载均衡的兼容性要想清楚。Nginx ingress默认的proxy-buffering要关掉而且长连接会导致Pod副本负载不均——因为同一个连接始终粘在同一个Pod上你需要根据实际QPS评估副本数不要被连接数看起来很高迷惑。5.5 多客户端广播与控制不只是一对一发消息最后一个场景是如果我有多个端网页、App、小程序同时登录用户在一端发起AI提问另一端要不要也收到流式消息这个在业务产品里很常见但SSE本身是不支持广播的。通常做法是把SSE连接交给一个消息中心管理。每个连接注册一个独立的队列发布时通过用户ID找到所有设备对应的连接把增量消息各推一份。这里有一个很容易踩的雷不要用Redis的Pub/Sub直接做广播因为订阅者要从同一份流里取消息各设备的消费进度不一样有的设备断线重连后需要从Last-Event-ID续传用共享队列 游标管理更稳妥。我当时是引入了一个轻量的内存队列做连接注册生产环境直接换成了Redis Streams每条消息带全局递增ID各设备记录自己的消费游标。这种设计下「停止生成」也要同步到消息中心让所有同用户的连接都能收到终止指令否则会出现手机上点了停止网页端还在继续吐字的怪现象。6. 我的一些个人体会和补充做AI应用这一年多我切实感受到SSE已经从小众技术变成了AI应用的基建之一。它没有WebSocket那么重的协议负担也没有轮询那种浪费资源的毛病。尤其是配合大模型的token流式输出SSE的简单、可靠、易调试帮我省下了大量联调时间。每次前端说又粘包了我只要把SSE流用curl拉一遍看到data:分隔清晰基本就能定位问题在前端解析逻辑。最后我再补充一个小技巧为了方便排查线上问题后端给每条SSE消息都打上序号和时间戳你会在调试消息乱序和重连重复时感激自己当初这个决定。接口上线后维护一份SSE协议的内部约定文档别只靠代码注释——前端、后端、测试各看各的代码协议细节很容易在迭代中走样。记住一点AI应用里流式体验就是产品的一半SSE是实现这种体验最经济实惠的路子值得花点心思打磨。