AI流式输出实战:原理、前后端实现与踩坑指南
2026/9/19 1:05:09 网站建设 项目流程

不知道你有没有遇到过这种场景:在对话框里问 AI 一个问题,它要一次性把整段回答全部生成完,才肯把结果一股脑展示给你。问题短的话还好,一旦让它写篇文章、生成一段代码、整理一份长报告,你就得对着屏幕干等十几秒甚至几十秒,中间什么都看不到,想早点发现答偏了也只能等它说完。

这就是大模型应用里最常见的体验痛点之一:AI 流式输出没有做,或者做得不够好。

所谓流式输出,简单说就是把 AI 生成的内容拆成一个个小片段,像水管里流动的水一样,边生成边送到用户眼前。效果就是你发出问题之后,AI 开始“边说边听”,第一个字可能一两秒内就出现,剩下的内容像打字机一样逐步展示,而不是憋一个大招再放出来。

我最近正好把一个对话类产品的输出方式从“等它说完”改成了“边说边听”,前后端的改动量并不大,但体感提升非常明显。这篇文章就把这套流式输出方案从头到尾拆开讲清楚,包括底层原理、技术选型、前后端实现、常见坑点,以及我在实际改造里积累的一些经验。适合正在做对话产品、AI 应用、或者想优化大模型交互体验的朋友,不管你是后端、前端还是全栈,读完都能直接上手。

1. 为什么 AI 需要流式输出?先看看“等它说完”有多难受

1.1 传统完整输出的体验问题

大模型的生成方式本质上是逐 token 预测,也就是一个词一个词地蹦出来。如果不做流式输出,服务端就必须等整段生成完毕,再把完整文本一次性返回给前端。这里有个很直接的问题:大模型生成最后一部分内容时,早先生成好的内容其实已经躺在缓冲区里了,但没有被传输出去,用户只能干等。

我实测过一个中等规模的模型,生成 1000 字左右的内容大约需要 5 到 10 秒,如果生成一篇完整的技术方案甚至要 20 秒往上。假设网络传输再占一点时间,用户可能在发出请求后的十几秒里一直面对一个空白对话框或转圈图标。长时间没有反馈,用户会下意识认为程序卡死了,要么刷新页面,要么直接关掉。

更麻烦的是,如果模型在生成过程中理解出现偏差,用户也是等全部内容出来之后才能发现。你让 AI 写一份周报,它写到一半开始扯无关内容,你等它全部写完才发现整段跑题,然后不得不重新组织语言再问一次。这个流程既浪费用户时间,也在消耗用户对产品的信任。

1.2 流式输出改写了什么

流式输出把“生成”和“展示”解耦了。后端不再等整段生成完再响应,而是每当模型产出一个小小的文本片段,就立刻通过长连接通道推给前端。前端收到一个片段就渲染一个片段,于是用户看到的效果就是文字一行一行冒出来。

这种体验更接近人和人之间的实时对话。对方说完前半句,你就能判断出大概要表达什么方向,如果你觉得不对可以随时打断,而不是等他全部说完才后悔。放在 AI 应用里也一样,用户看到开头就觉得不对,可以直接停止、重新提问,省掉等待剩余部分生成的时间。

从产品角度看,流式输出还顺带解决了一个隐性需求:反馈即时性。心理学上有一个“系统响应时间”的概念,超过 2 秒没有反馈,用户就会开始焦虑,超过 10 秒就会显著不耐烦。流式输出把首字返回时间压缩到 1 到 3 秒内,后面每秒钟都会有新内容出现,用户的等待焦虑会明显缓解。我之前在一个测试群里做过一次简单的内部对比,使用流式输出后,用户完成一次提问到收到完整回答的“主观等待时间”感知明显变短了,很多人甚至没意识到长文本生成竟然用了十几秒。

1.3 哪些场景特别依赖流式输出

不是所有 AI 应用都需要“边说边听”,但下面几类场景属于刚需:

  • 对话助手 / 客服机器人:一句话或几句话的回复,用户对响应速度极其敏感。流式输出是标配。
  • 代码生成工具:AI 生成的代码往往很长,而且用户希望看到生成过程,及时判断是否偏题,随时中断。
  • 写作 / 摘要 / 翻译工具:长文本生成场景,流式输出能显著降低等待焦虑。
  • Agent 任务执行面板:Agent 在执行多步骤任务时,需要实时回传状态和中间结果。流式输出不仅是给用户展示,也是在提供任务执行的“进度条”。
  • AI 短剧 / AI 漫剧的旁白与字幕生成:这类内容生产场景里,用户希望内容是逐句出现的,而不是整块砸过来。

我在做流程改造时,就是把一个内部“方案助手”的完整输出改成了流式。改动范围覆盖了后端接口、前端调用、异常处理三个部分,整体代码量不算大,但体验升级明显。接下来我会把这套方案的核心细节逐一展开。

2. 流式输出背后的技术原理:模型、连接与数据格式

2.1 模型的“逐 token”生成决定了流式的可行性

先说清楚一件事:大模型本身天然就是流式生成的。模型在推断阶段,每一步只预测下一个 token 的概率分布,把它作为输入继续生成下一个 token,循环往复。你可以把它理解成一个普通话痨,脑子里其实一句话还没想清楚,嘴巴已经在蹦单词了,是边想边说的。

所以从原理上讲,让模型把中间结果逐步吐出来,再合理不过。相反,如果后端把模型生成的所有内容先攒起来再返回,反而是在人为制造延迟。这和去餐厅吃饭一样,厨师做一道菜要 15 分钟,但每做好一个菜就端上来一个菜,你从下单到吃到第一口只要 3 分钟,体验当然比 15 分钟后一次性上满一桌要舒服。

实现流式输出的关键是:如果你用的是某个大模型 API,需要开启对应的流式参数,例如stream: true;如果是自己部署大模型,推理框架通常也支持 hf 或 OpenAI 兼容的流式接口,返回的是“逐个 token 的事件流”。数据格式一般类似这样:

data: {"choices":[{"delta":{"content":"你"}}]} data: {"choices":[{"delta":{"content":"好"}}]} data: [DONE]

每一行就是一个“事件”,前端解析到data:之后的内容,拼接到界面上就行。

2.2 长连接通道选型:SSE 还是 WebSocket

拿到模型逐步返回的内容后,后端要把这些片段实时传给浏览器或客户端。这里有两个主流方案,我直接放一张对比表,方便你快速判断。

对比维度SSEWebSocket
方向服务器单向推送双向实时通信
协议基于 HTTP独立的 WebSocket 协议
简单程度很低,随便一个 HTTP 框架都能支持较高,需要维护连接状态
自动重连原生支持需要自己实现
消息格式纯文本 / 自定义事件文本或二进制帧
适用场景AI 输出推送、通知类聊天、游戏、协作白板等双向场景

就 AI 流式输出这个场景,我强烈建议优先选SSE。原因很简单:流式输出的数据方向是单向的,就是从服务器流向浏览器,中途用户基本不需要再往服务器发指令。SSE 天然就是为“服务器推数据”设计的,实现简单,用一个 HTTP 连接就能搞定,而且支持断线自动重连,省掉不少代码。

WebSocket 当然也能做,但代价是你要处理连接握手、心跳、重连、消息格式约定等一系列额外逻辑。除非你的场景同时需要频繁的前后端双向通信,否则没必要用它。

2.3 SSE 的传输格式与前端接入方式

SSE 的底层其实不复杂,它就是服务器返回一个Content-Type: text/event-stream的响应,然后持续向客户端写入多行文本。每段消息以data:开头,消息之间用空行分隔。前端浏览器原生支持EventSource对象,可以直接接收这种类型的事件流。

但这里有一个常见的坑:EventSource只支持 GET 请求,不能自定义请求头。很多大模型接口需要 POST 携带认证信息或参数,这时原生 EventSource 就不太够了。我在实践中更推荐用fetch+ReadableStream的方式,由前端手动读取流数据并逐段解析。这样既能保持一个普通 POST 请求的样子,又能拿到实时增量,代码虽然多几行,但可控性高得多。

数据格式大概是这样组织的:

data: 第一段内容 data: 第二段内容 data: [DONE]

前端需要按行读取,凡是遇到以data:开头的行,就取冒号后面的内容,直到遇到[DONE]表示输出结束。

3. 实操:从后端到前端,一套可直接抄的流式输出方案

3.1 后端:用 FastAPI 快速实现流式接口

我平时项目里常用 Python FastAPI,这里就以它为例。如果你用 Java 或 Node.js,思路完全一致,只是框架语法不同。

先看最简单的流式接口:

from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app = FastAPI() async def generate_reply(prompt: str): # 这里模拟模型逐步生成内容 # 实际项目中替换为大模型调用即可 chunks = ["你好", ",", "我", "是", "AI", "助手", "。"] for chunk in chunks: yield f"data: {json.dumps({'content': chunk}, ensure_ascii=False)}\n\n" await asyncio.sleep(0.1) yield "data: [DONE]\n\n" @app.post("/chat") async def chat(prompt: str): return StreamingResponse( generate_reply(prompt), media_type="text/event-stream", headers={"Cache-Control": "no-cache"} )

几个值得注意的点:

  • media_type一定要设置成text/event-stream,否则前端不认识这是 SSE 流。
  • Cache-Control: no-cache是为了防止中间层缓存响应。如果响应被缓存,内容会攒到结束才返回到浏览器,流式效果就没了。
  • 实际项目中,你会把generate_reply里的模拟数据换成真正的模型 API 或本地推理结果。用 OpenAI 兼容接口时,开启流式参数后,遍历返回的流式对象,把每个内容片段yield出去就行。

如果你用的是 Java 生态,最常见的做法是ResponseBodyEmitter或 Spring 的SseEmitter,同样可以做到逐步推送。

3.2 前端:用 fetch 解析流式增量

前端的核心逻辑是:发起一个普通 POST 请求,但不用response.json()去拿完整结果,而是通过response.body.getReader()读取字节流,再用TextDecoder解码成文本。

下面是一段可以直接放进 Vue 或 React 项目里的核心代码:

async function streamChat(prompt) { const response = await fetch('/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ prompt }) }); if (!response.ok) { throw new Error(`请求失败: ${response.status}`); } const reader = response.body.getReader(); const decoder = new TextDecoder('utf-8'); let buffer = ''; while (true) { const { done, value } = 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) { const trimmed = line.trim(); if (!trimmed.startsWith('data:')) continue; const data = trimmed.slice(5).trim(); if (data === '[DONE]') { // 输出结束 return; } try { const parsed = JSON.parse(data); const content = parsed.content || ''; // 把 content 追加到界面上 appendMessage(content); } catch (e) { // 个别片段可能不是完整 JSON,跳过不处理 console.warn('解析失败:', data, e); } } } }

这里有三个很容易踩的细节,我说一下:

第一,解码要用{ stream: true }。因为value是一段二进制流,一个中文字符可能被拆成两个字节跨在两个value里,如果不用流式解码,就会出现中文乱码。

第二,拆行处理时,最后一行不能直接处理,因为它可能是一个还没传完整的事件。要把它留在下一次循环里补全,否则会出现“内容只显示一半”或“JSON 解析失败”的问题。

第三,reader.read()返回的donetrue只代表响应流关闭,不代表大模型生成的业务内容正常结束。有时候后端因为异常中断连接,前端也能正常读到done,所以最好结合业务标志(比如[DONE])来判定是否真正结束。

3.3 服务端的 flush 与缓冲区问题

后端的流式接口写好后,本地一测,效果很正常,但一部署到服务器上就发现不对劲:前端还是得等很久才看到内容。这个问题十有八九出在缓冲区上。

具体表现是:后端明明已经在 for 循环里逐步yield数据了,但用户端还是隔很久才看到批量内容。原因是中间某些环节在帮忙攒数据,比如:

  • Web 框架默认开了响应缓冲;
  • Nginx 或网关开启了响应缓冲;
  • 某些服务端语言在写 data 时先写入缓冲区,没有及时 flush。

解决办法分几步:

  • Nginx 层:如果在你自己的可控范围内,关闭或调小某个位置的缓冲,例如把相关缓冲设置为关闭,并调低超时时间。注意,很多云厂商的网关或负载均衡也可能有类似缓冲行为,必要时在响应头里加X-Accel-Buffering: no,这个头对部分中间层是有效的。
  • 应用层:在响应对象上主动调用某种“手动输出”操作。比如在 FastAPI 里yield本身就是逐步输出,一般不用手动 flush;但在 Java Servlet 里,你要主动调用response.flushBuffer(),否则数据会憋在缓冲区里。
  • 测试方法:用一个命令行工具直接请求流式接口,观察返回内容是否逐段出现。例如curl -N,这里-N表示关闭缓冲,逐行显示返回结果。如果 curl 能看到逐步输出,前端却看不到,问题大概率出在前端或浏览器层;如果 curl 也是攒着输出,问题就在服务端或中间层。

我在一次部署里就踩过这个坑,本地跑得顺滑无比,上到服务器之后首字延迟变成了 8 秒。排查了半天,最后发现是网关层默认开了缓冲,把内容的一段段拼接起来了。解决方式是调整网关设置,并在响应头里加上Cache-Control: no-cacheX-Accel-Buffering: no。改完之后首字返回时间直接降回 1 秒左右。

3.4 中断、超时与重试机制

流式输出还有一个经常被忽略的问题:长连接的超时设置

普通 HTTP 接口可能 30 秒就超时了,但流式接口在生成长文本时,连接可能要持续一分多钟。如果你没有调大超时,用户看到的内容会在中途被截断。我见过一个案例,前端一直无响应,后端连接早就被底层断开,两边都不报错,看起来就是“AI 说到一半突然闭嘴了”。

处理思路是这样:

  • 客户端/服务端都要设置合理的读超时和空闲超时,至少大于大模型最长生成时间。比如你预计一条回复最多 120 秒,超时就应该设到 150 秒以上;
  • 前端要增加自动重试或手动重试按钮。如果流中途断开,保留已显示的内容,提示用户“生成中断”,并提供“继续”或“重新生成”操作;
  • 后端要做好取消处理。如果用户中断了前端请求,后端应当收到信号并停止模型生成,否则模型还会傻傻地把整段内容生成完,浪费算力。

关于取消,前端可以在 fetch 时传入AbortController,用户点击“停止生成”时调用controller.abort(),同时给后端发送取消信号。后端收到取消信号后要停止迭代模型结果,并及时释放连接。

有一个产品细节我觉得很值得做:断点续传式的重试。前端把已经收到的内容保留住,重新发起请求时告诉后端“我已有这些内容,只续写后面部分”。现在很多模型 API 本身支持历史上下文,你把已有内容作为历史消息传回去,让它接着生成,就能实现“从断点继续写”的体验。当然,这个方案需要模型本身支持续写,不是所有场景都能直接套用,但做出来之后用户好感度会提升不少。

4. 踩坑实录:流式输出最常见的几个翻车现场

4.1 “标签返回未完整”的处理方案

做 AI 对话产品时,经常遇到一个头疼问题:前端采用流式渲染富文本或 Markdown,但 AI 输出的内容是逐步到达的,某个时刻你拿到的可能是半个 HTML 标签或半段 Markdown 语法。比如<b>标签只传来了<b,下一段才传来>和内容。如果你直接用v-htmldangerouslySetInnerHTML渲染,页面会闪烁、错位,甚至暂时显示成纯文本,非常难看。

这类问题的本质是:渲染时机与数据完整性有冲突。流式输出要求实时渲染,但语法标签需要完整才能正确渲染。

我建议这几种处理方式:

  • 在流式渲染时,只把内容文本显示出来,不即时解析 Markdown 或 HTML 标签。等待整个流结束后,再统一做一次完整渲染。这样牺牲了一点实时性,但不会出现标签错乱。
  • 侧边或底部展示“原始输出”或“预览模式”。预览模式用实时纯文本,完整模式在结束后渲染富文本。
  • 如果你用的是成熟的 Markdown 渲染库,很多库已经内置了流式兼容处理,它们会等待一个代码块或列表结构完整后再渲染。实测这类库的效果好很多,推荐优先尝试。

这里要特别提醒一下:不要想在流式过程中做到 100% 的语法渲染正确,那是一个性价比很低的方向。用户在看流式输出时,重点是阅读和理解,不是欣赏排版。把最终结果渲染好就够了。

4.2 内容截断与乱码

按我的经验,流式输出内容截断和乱码,绝大多数不是大模型本身的问题,而是传输和解析链路上的问题。

截断常见原因包括:

  • 超时时间设短了,连接被中断;
  • 中间层缓冲数据还没 flush,前端就判断流结束;
  • 部分字符被拆到两个 chunk 里,前端没做好拼接;
  • 后端异常崩溃,异常没被捕获,连接被迫断开。

乱码常见原因包括:

  • 前端使用了单个二进制片段直接解码,没有处理跨片段的中文字符;
  • 解码字节流时没有用 UTF-8;
  • 服务端输出时没有标注正确的字符编码。

排查建议:先看原始响应,用命令行工具直接请求接口,看是否有逐步输出、是否中文正常。如果原始响应正常,问题就出在前端解析;如果原始响应本身乱码,问题出在后端编码。

我自己的经验是:前端解码一定用TextDecoder('utf-8'),并且传入{ stream: true }。跨 chunk 的字符拼接则统一用 buffer 变量先暂存不完整的行,等补齐后再处理。这两个小细节能解决绝大多数乱码和截断问题。

4.3 连接一直挂着不结束

还有一个常见故障:界面上的文字一直不显示,连接状态显示“等待中”,过了一分钟也没变化。这种往往不是完全没数据,而是数据到了某个环节但没被推给用户。

排查思路要分几步:

  1. 先确认后端是否真的在输出。看日志,看有没有 yield 的记录;
  2. 再确认中间层是否缓冲。用命令行工具按住-N请求同一个接口,观察输出节奏;
  3. 再确认前端是否正确处理了拆包。打印收到的字节流内容,看有没有“半个 JSON”被丢掉;
  4. 最后确认是不是响应没有及时 flush。

很多调整过 Kafka、消息队列的同学应该对这种“看起来连接没问题但数据没流动”的情况不陌生,本质上是“数据管道中间某个环节把消息吞了”或“缓冲区没释放”。流式输出也是同理,链路越长,越需要用最小化实验逐步定位问题。

5. 流式输出不止于“打字机效果”,还能怎么用

5.1 从“边生成边展示”到“边生成边理解”

流式输出最直接的效果是打字机般的实时反馈,但它的价值不止于此。内容生成过程中,用户其实已经在阅读、思考,甚至不需要等待全部完成,就可以准备下一轮问题。

这种“边说边听”的交互模式,给了 AI 应用一个重要的产品机会:更早地让用户参与进来。用户发现 AI 理解错误时,可以立即打断纠正,而不是等生成完再纠正,这能显著缩短多轮对话的周期。

我做过的另一个尝试,是在 Agent 类型应用里,把任务执行中的中间状态通过流式接口回传。比如一个数据分析 Agent,执行步骤是“读取数据 → 清洗 → 建模 → 输出结论”,后端每完成一步就向前端推送一条状态。用户能看到任务正在推进,知道卡在哪一步,比面对一个“正在处理”的转圈图标安心很多。

这背后的技术和 AI 流式输出完全一样,只是推送内容从“文字片段”变成了“结构化状态”。你可以把状态封装成 JSON 事件,前端根据事件类型做不同渲染。比如:

{"type": "status", "data": "正在读取数据"} {"type": "content", "data": "{"summary": "..."}"}

5.2 与前端框架结合时的几个细节

现在前端主流框架是 Vue 或 React,流式输出也经常涉及状态管理。我的建议是:把流式解析逻辑放到一个独立的工具函数或 Hook 里,不要在组件里写一大坨流式解析代码,否则后期维护非常痛苦。

以 React 为例,你可以封装一个useChatStream的 Hook,内部管理 messages、isStreaming、error、stop 等状态,对外只暴露sendMessagestop方法。组件层面只需要关心界面渲染,不用关心流式解析的细节。

Vue 里思路类似,可以用一个 composable 函数来管理。

另外,界面渲染时要注意:更新消息内容不能使用全量替换。比如你用一个字符串保存回复内容,每次收到新 chunk 就重新setState一次,这本身没毛病,但如果同时渲染长文本,可能会有性能问题。更稳妥的做法是维护一个消息数组,在流式更新时只更新最后一条消息对象,而不是重建整个数组。

5.3 流式输出的技术债务与架构展望

流式输出虽然能提升体验,但它也会给系统引入一些新的复杂度,尤其在高并发场景下。想象一下:传统接口是“用户请求进来,服务端算完,返回结果,连接释放”;流式接口则会把一个连接长期占用,并发一高,对连接数、负载、网关压力都会明显上升。

所以做流式输出时,有几个架构层面的问题需要提前想清楚:

  • 连接数管理:SSE 长连接一般会一直挂着,连接数会堆得比普通接口高很多。服务端要设置合理的最大连接数,超过时给出明确提示。
  • 横向扩展与负载均衡:如果你的服务有多台机器,长连接会黏在其中一台机器上。中间层要配置好保持连接相关的机制,否则前端每次请求都可能被分发到不同机器,导致流式中断。
  • 多副本部署时的消息广播或定向推送,要考虑“请求落在哪个实例”的问题。通常最好在网关或负载层开启粘性会话,或者在应用层引入共享状态来协同。

这些架构层面的细节,比“怎么写出流式接口”要复杂得多。这篇先不展开细说,如果后面有时间,我想专门写一篇高并发的流式推送架构笔记,把网关、连接池、背压、限流这些话题一次性聊透。

6. 流式输出相关的另一种形态:不只是聊天

聊到这里,可能有人会问:流式输出是不是只适用于大模型对话场景?其实不是。凡是“数据量很大、可以边取边用”的场景,都能借鉴流式的思路。

举一个和数据库相关的例子:一般来说,执行一条查询,应用会等数据库把查询结果全部返回,再一次性处理。但如果查询结果有几十万行甚至上百万行,一次性加载到内存很容易把内存打爆。流式查询的思路是:应用发起查询后,数据库不断产生数据,应用按行或按批次读取处理。常见做法是给 JDBC 驱动配置游标抓取参数,例如useCursorFetch=true并设置一个合适的fetchSize,这样查询结果会像流水一样一条条“流”到应用端,而不是整包倒进来。

再比如 AI 编程场景里的代码补全工具,模型在生成代码补全建议时也是流式输出的,编辑器里看到的代码是一行一行蹦出来的。还有 AI 短剧、AI 漫剧的字幕生成,字幕会随着剧情逐句出现,这背后同样是流式输出技术在支撑。

所以你会发现,“流式”本质上是数据处理的一种通用思维:不要等全部就绪再交付,而是边生产、边传输、边消费。这种思维在大模型时代被推到了前台,但它其实在很多传统技术领域早就存在了。

理解了这个底层逻辑之后,你再去看各种流式输出技术,会发现万变不离其宗:生产者逐步产出数据,传输管道保持开放,消费者边收边处理。只要这三步协调好,几乎所有“数据大、耗时长”的场景都能用流式来优化。

我自己的体会是,流式输出最难的不是写代码,而是建立“边生成边交付”的思维方式。刚开始做 AI 应用时,我也习惯把所有内容攒起来,一次性返回给前端。那时觉得这样逻辑简单、不容易出错。但用过流式之后,再回头去看那些“等它说完”的产品,基本上已经不能忍了。特别是当你面对一个闲聊机器人问“给我讲个故事”,它要是干巴巴停 5 秒才开口,你会立刻觉得这个机器人不太聪明;相反,哪怕它 1 秒内能蹦出“从前有个山”,你都觉得它特别机灵。

最后再分享一个小技巧:如果你的前端用的是 Vue 或 React,调试流式输出时不要过度依赖浏览器开发者工具的网络面板。开发者工具会默认对 SSE 响应做缓存,看到的效果可能不是实时的。更好的做法是在代码里加一个临时日志面板,把每次收到的 chunk 原样打印出来。我调试流式接口时就是这么干的,能看到每一个小片段到达的瞬间,排查问题效率高很多。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询