从“转圈圈”到“逐字输出”一次流式接口改造的真实体验我最早接触 OpenAI 接口流式输出时很多人还在做点击按钮等两三秒一次性把整段结果打印到页面这种交互。结果用户等得心慌又不知道系统到底有没有在跑动不动就重复点击把请求打重了。后来我把接口改成流式打印生成结果体验立刻不一样了——用户能看到文字像打字机一样逐字蹦出来心里踏实反馈也完全变了。这篇内容我会围绕 openai 接口流式返回的完整链路来写从底层机制、Python 后端实现、前端流式接收到实际项目里踩过的坑和排查思路都展开讲。适合正在做 AI 应用、想让交互更贴近 ChatGPT 原版体验的开发者也适合想搞懂流式原理、避免走弯路的新手。1. 从转圈圈到逐字输出流式接口解决的到底是什么问题1.1 用户等待的心理模型为什么一边生成一边显示比结束后统一显示重要一个很反直觉的事实是同样一段文本如果让用户等待 3 秒后一次性看到完整结果用户会感觉等了非常久但如果让用户看到文字实时生成哪怕生成时间拉长到 4 秒用户也觉得响应很快。这不是玄学而是人对不确定性的天然焦虑——无进度地等待是最难熬的。以前我们调 openai 接口的时候用的是最传统的非流式请求response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 介绍一下流式接口}], ) print(response.choices[0].message.content)这种方式很直白请求发出去模型把整段内容都生成完SDK 再把完整的 message 内容返回给你。问题是大模型生成一段长文可能需要好几秒甚至几十秒这段时间请求一直挂着用户在界面上看到的就是一个转圈动画。转圈超过 2 秒用户就开始怀疑是不是出了 bug超过 5 秒相当一部分人已经点了刷新。但如果你把streamTrue打开模型每生成一小块 token 就通过 SSEServer-Sent Events服务端发送事件推给客户端前端拿到一个增量就渲染一个字或一个词整个交互就从等待结果变成了看着 AI 思考并书写。用户不再焦虑甚至会产生一种这模型好聪明的积极感受。做 AI 应用的人应该都体会过这种差异它直接决定了产品的第一印象。1.2 从后端耗时到前端体验流式不改变生成总时长但改变了感知时长有一点需要先说明清楚免得大家产生误解流式接口并不会让模型生成得更快。模型该生成 500 个 token还是得一个 token 一个 token 地算总耗时基本没有变化。流式真正改变的是第一个可见输出的到达时间。在非流式请求里你必须等全部内容生成完毕才收到完整响应。一个 800 token 的回答可能需要 10 秒那用户前 9.9 秒看到的都是空白和转圈。而在流式请求里第一个 token 可能 0.5 秒就到了后面的 token 陆续到达到了就渲染。用户第 1 秒看到前几个字第 3 秒看到一小段第 10 秒看到结尾——虽然最终完成时间没变但已经出现内容的感觉会极大缓解等待焦虑。这个特性特别重要我后来做企业级 AI 聊天助手时深有体会。用户提问后如果界面上一直没有任何反馈在线客服系统里的用户往往就会再发一句在吗或者直接关掉页面。而流式打印模式下用户看到第一个字出来就知道系统已经接管问题并开始处理了这对留存率的影响非常直观。1.3 流式还顺带解决了长响应容易超时的问题除了体验层面流式还有一个很多人忽略的技术价值避免 HTTP 长连接超时。调用非流式接口时如果模型生成时间超过网关或负载均衡器的超时限制比如 60 秒中间层可能直接断开连接客户端收到一个 timeout 错误。尤其在做一些复杂推理、代码生成、长文本总结时生成耗时很容易飙到几十秒非流式请求在这种场景下特别脆弱。而流式请求会持续地往客户端推数据。只要服务器还在不断产生新 token连接就是活跃的网关那边就不会判定为闲置连接而掐断。哪怕这次生成总共花了 90 秒连接也一直是有数据流动的状态这从工程层面绕开了中间设备的超时限制。我当时在生产环境遇到过线上服务频繁报 504 的问题换成流式之后这类报错几乎消失了。2. 流式返回的底层机制stream 参数、SSE 协议与 delta 增量2.1 一次完整的流式响应长什么样想真正掌握 openai 流式打印生成结果不能只停留在加个 streamTrue 就跑通了的程度得知道线上传回来的到底是什么。我用一个最朴素的 HTTP 层面的工具curl来演示这样大家能看清底层长什么样curl https://api.openai.com/v1/chat/completions \ -H Authorization: Bearer YOUR_API_KEY \ -H Content-Type: application/json \ -d { model: gpt-4o-mini, messages: [{role: user, content: 你好}], stream: true }响应不是传统的完整 JSON而是这样一串 SSE 格式的数据data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1710000000,model:gpt-4o-mini,choices:[{index:0,delta:{role:assistant,content:},finish_reason:null}]} data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1710000000,model:gpt-4o-mini,choices:[{index:0,delta:{content:你},finish_reason:null}]} data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1710000000,model:gpt-4o-mini,choices:[{index:0,delta:{content:好},finish_reason:null}]} data: [DONE]每一行data:后面都是一个独立的 JSON 对象这个对象在官方文档里叫chat.completion.chunk也就是一个分片。每个分片里的choices[0].delta是本次推送的增量内容。注意看第一个 chunk 里的delta它带的是role: assistant后面再来的 chunk 才带真正的content文本。等模型生成完了最后一个 chunk 的finish_reason会变成stop或者length、tool_calls等最后再发一个data: [DONE]表示结束。有个值得注意的细节第二个 chunk 里面delta只有content:你并没有重复携带之前的文本每一个 chunk 都是增量不是全量。如果你的代码把每个 chunk 直接赋值给变量而不是做拼接那最后页面上只会显示一个好字这是新手最容易犯的错。后面我会再用代码演示正确的拼接方式。2.2 SSE 协议的三个关键特征SSE 全称是 Server-Sent Events它和普通 HTTP 响应最大的区别是连接一旦建立服务端可以不断往同一个响应里写入新数据客户端则不断读取这些新数据直到连接关闭。这里我要强调三个容易被忽略的特征理解了它们你在排查流式问题时会顺很多Content-Type 必须是text/event-stream。如果你用底层工具直接发请求要留意响应头。如果看到响应头是application/json说明服务端没有按 SSE 方式返回这时候要么是请求参数没带对要么是你用了某个网关把流式缓存成了整体 JSON。每条消息以data:开头消息之间用空行分隔。你可以在服务端自定义event: xxx和id: xxx等字段但 OpenAI 官方接口目前就返回data字段。解析时按空行拆开每一块再按冒号拆字段名和值。连接结束的标志是data: [DONE]。注意[DONE]不是一个 JSON它是 SSE 里的一个特殊结束标记。如果你用 JSON.parse 去解析这行会直接报错所以解析逻辑里必须对data [DONE]做特殊判断。很多人在项目里手写过 SSE 解析工具网上能找到各种版本的代码。但说实话做 OpenAI 应用时你通常不需要自己手写这些直接用官方 SDK 就行。真正需要手写解析的场景是你自己封装一个中间服务把 OpenAI 的流式响应转给你的前端。到那种场景下上面这几个细节就非常关键了。2.3 官方 SDK 是流式友好的别自己乱造轮子官方 Python SDK 对流式的支持已经很完善了你只需要传入streamTrue然后迭代返回对象。一个最基础的流式打印例子长这样from openai import OpenAI client OpenAI(api_keyYOUR_API_KEY) stream client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 写一首关于春天的短诗}], streamTrue, ) for chunk in stream: if chunk.choices[0].delta.content is not None: print(chunk.choices[0].delta.content, end, flushTrue)这个例子虽然短但有几个重要的点chunk.choices[0].delta.content可能为None。比如第一个 chunk 里只有role没有content或者某些特殊情况下delta里根本没有content字段。如果你不做is not None判断None被打印出来就会变成一行None。end是为了把每次输出后的默认换行去掉实现同一行续写flushTrue是为了立刻把缓冲区的内容刷出来否则 Python 的标准输出可能会积攒到一定量才打印你就看不到逐字输出的效果了。迭代器会在收到[DONE]后自动终止你不需要自己在循环里判断结束。这个例子可以直接跑但离真正能用到项目里还差几步比如处理 usage 信息、处理工具调用、处理多个 choice 等。这些我在后面专门讲踩坑的章节里再展开。3. 用 OpenAI SDK 实现流式打印从一段可跑通的代码说起3.1 环境准备版本是一件不该被忽略的事我见过太多人在环境这一步出问题。openai 的 Python SDK 更新速度非常快旧版本和新版本的 API 差别很大。早期版本甚至没有OpenAI这个类用的是openai.ChatCompletion.create新版本统一成了OpenAI()客户端模块结构完全变了。我建议的安装命令pip install -U openai装完以后可以快速验证一下版本python -c import openai; print(openai.__version__)如果你看到的是0.x开头说明版本非常旧建议更新到1.x以上。我的生产环境目前用的是 1.x 系列代码风格和本文保持一致。另外需要提一句现在很多国内开发者是通过 Azure OpenAI 或第三方中转服务来调用 ChatGPT 接口的这种情况下需要额外配置base_url。比如你把 openai SDK 指向 Azure OpenAI 端点时要这么写client OpenAI( azure_endpointhttps://your-resource.openai.azure.com/, api_keyYOUR_AZURE_API_KEY, api_version2024-02-15-preview, )但你说到底还是用官方接口还是 Azure 接口取决于你的业务部署环境。代码结构基本是兼容的只是client的初始化方式不同。我这里不会去细讲某种特定转发的配置细节因为不同的平台差异太大了你按自己服务商的文档来就好。不过请务必记住无论走哪种通道流式返回的 SSE 机制和 chunk 结构基本一致后面的代码思路是通用的。3.2 完整代码一个带状态管理的流式打印函数下面这段代码是我在实际项目中用过的一个相对完整的版本它除了把流式内容打印出来还会自动拼接成完整文本、避免None值污染并在结束后打印本轮 token 消耗import os from openai import OpenAI client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) def stream_chat(user_prompt: str, system_prompt: str ): messages [] if system_prompt: messages.append({role: system, content: system_prompt}) messages.append({role: user, content: user_prompt}) stream client.chat.completions.create( modelgpt-4o-mini, messagesmessages, streamTrue, stream_options{include_usage: True}, # 让最后一个chunk携带usage temperature0.7, ) full_response [] print(Assistant: , end, flushTrue) for chunk in stream: delta chunk.choices[0].delta # 有些 chunk 只有 role没有 content必须跳过 if delta.content is not None: print(delta.content, end, flushTrue) full_response.append(delta.content) # 检查结束原因 if chunk.choices[0].finish_reason is not None: print() # 模型输出结束补一个换行 print(f[finish_reason] {chunk.choices[0].finish_reason}) # usage 信息需要在请求里显式加 stream_options if hasattr(chunk, usage) and chunk.usage is not None: print(f\n[usage] prompt_tokens{chunk.usage.prompt_tokens}, fcompletion_tokens{chunk.usage.completion_tokens}, ftotal_tokens{chunk.usage.total_tokens}) return .join(full_response) if __name__ __main__: result stream_chat(用三句话解释什么是流式接口) print(\n\n完整拼接结果) print(result)说几个我在实际使用中的经验stream_options{include_usage: True}是在 openai 1.x 后期版本里才支持的功能。不加的话chunk.usage一直为None拿不到 token 用量。这个字段加不加影响不大但如果你做商业化产品需要对用户做计费统计这一块就必须搞清楚。finish_reason的判断放在循环里而不是放在循环外。因为最后一个 chunk 会同时带有delta.content内容为空和finish_reasonstop如果你在循环外判断也可以但会丢掉最后一轮 chunk是最接近完成状态的时机这个信息。放在循环里你打印出来的效果更自然。我把print()和返回值分开处理。打印是指实时输出给控制台或日志返回值是完整的拼接文本可以存库、做后续处理。这两个场景是流式接口最常见的两种消费方式兼容并包最省心。3.3 为什么这样设计状态拼装、中断处理与可观测性很多人写第一版流式代码时只会在循环里打印不拼接导致后面的逻辑没法用这个结果。我的经验是流式消费的本质是把增量变成全量。这个全量一方面可以实时打印出去另一方面要存在一个累积变量里。以后你想把对话记录入库、把生成结果发送给另一个 API直接拿这个累积后的字符串就行不用重新调用一次接口。还有一个实战中容易踩的坑是用户中途打断退出。比如你用 Gradio 或 FastAPI 做 Web 服务前端 WebSocket 断开时post 请求的响应通道可能已经没了但你还在for chunk in stream循环里。这时候如果不对异常做处理服务端进程会继续把文本生成完白白消耗 token。解决思路是给请求添加取消逻辑最简单的方式是用asyncio配合httpx的异步客户端或者在销毁时主动调用stream.close()。我在项目里一般会做一个try/finally的保护try: stream client.chat.completions.create(...) for chunk in stream: process(chunk) finally: # 确保连接被释放避免资源泄漏 if stream is not None: stream.close()虽然官方 SDK 的迭代器用完后会自动断开但万一你在循环内部抛异常退出连接未必会立刻释放。养成try/finally的习惯能少很多诡异的连接数上涨问题。另外我强烈建议在流式打印时加上时间戳。生产环境排查问题时如果只看到内容文本你很难判断哪一段花了多少时间。给每一批 chunk 打印一个毫秒级时间戳或者至少记录收到第一个 chunk 的时间和收到[DONE]的时间对分析接口响应慢这类问题特别有用。4. 前端接过流式数据fetch 流式读取的完整链路4.1 为什么不能直接用 axios 处理流式响应后端拿到 OpenAI 的流式结果后通常有两种做法一种是直接把 OpenAI 的 SSE 流转发给浏览器另一种是后端自己把内容聚合后再通过 WebSocket 或 SSE 推给前端。这里我先讲前端怎么消费 SSE因为这是大部分 AI 对话应用最关心的一环。很多前端同学习惯用 axios但 axios 默认并不会给你流式读取的能力。axios 会把整个响应体下载完再交给 then 回调也就是说即使你用了stream: true前端看到的还是一个整体字符串完全失去实时效果。所以这里我推荐用浏览器原生fetch通过ReadableStream逐块读取。下面是一段可以直接放到 Vue 或 React 项目里的前端代码以 fetch 为例async function fetchStream(messages) { const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json, }, body: JSON.stringify({ messages }), }); if (!response.ok) { throw new Error(HTTP error! status: ${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 }); // SSE 消息之间以空行分隔这里按 \n\n 拆出完整消息 const lines buffer.split(\n\n); buffer lines.pop(); for (const line of lines) { if (!line.startsWith(data:)) continue; const data line.replace(/^data:\s*/, ); if (data [DONE]) return; try { const json JSON.parse(data); const content json.choices?.[0]?.delta?.content; if (content) { // 这里追加到你的页面显示变量 console.log(增量:, content); } } catch (err) { console.warn(解析 chunk 失败, data, err); } } } }我来说明几个关键点response.body.getReader()返回的是一个流读取器reader.read()每次返回{ done, value }。done为 true 表示流结束。TextDecoder(utf-8, { stream: true })里的stream: true很重要。因为浏览器收到的二进制块可能把一个 UTF-8 字符截断成两半如果你不告诉解码器这是一个流那它就只能按独立的块解码可能出现乱码。buffer变量用来缓冲不完整的 SSE 消息。因为reader.read()返回的块大小不固定可能一条data:消息被拆成两次读取也可能一次读取包含多条消息。buffer.split(\n\n)以后最后一个元素要放回 buffer等下一轮数据到齐再处理。4.2 后端转发时的两个关键设计流式透传与自定义事件格式很多情况下你不会直接把 OpenAI 的api.openai.com暴露给浏览器一方面有安全风险另一方面你需要控制权限、限流、计费。所以你会在自己的后端做一个转发接口这个接口同时扮演中间人的角色。用 Node.js 写一个转发接口其实是比较自然的// Express 或 Koa 场景下 app.post(/api/chat, async (req, res) { res.setHeader(Content-Type, text/event-stream); res.setHeader(Cache-Control, no-cache); res.setHeader(Connection, keep-alive); const upstream await fetch(https://api.openai.com/v1/chat/completions, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${process.env.OPENAI_API_KEY}, }, body: JSON.stringify({ model: gpt-4o-mini, messages: req.body.messages, stream: true, }), }); const reader upstream.body.getReader(); const decoder new TextDecoder(); try { while (true) { const { done, value } await reader.read(); if (done) break; // 直接按原样转发给前端 res.write(decoder.decode(value, { stream: true })); } } finally { reader.releaseLock(); res.end(); } });这套做法的核心好处是前端和后端都用同一套 SSE 解析逻辑格式统一不容易出错。你不用在后端做任何内容解析和拼接OpenAI 返回什么你转发什么。前端直接解析 OpenAI 的 chunk 结构即可。当然也有一些场景下你更希望在服务端把文本聚合好了再用自定义的 SSE 格式推给前端比如你要在流式响应中附带额外的元数据像本轮消耗的 token、命中的知识库文档 ID、搜索结果的来源链接等。这时候你可以定义自己的事件格式比如data: {type: start, conversation_id: xxx} data: {type: delta, content: 你好} data: {type: search_result, results: [...]} data: {type: done, usage: {total_tokens: 123}}前端拿到后按type字段分流处理比如delta类型就追加文本search_result类型就展示引用卡片。这样比直接透传多了一层控制力代价是后端需要自己解析和拼装 OpenAI 的 chunk逻辑复杂度会上升。我的建议是简单场景直接透传复杂场景后端自定义事件流不要两头都做解析否则维护成本翻倍。4.3 连接安全与中断恢复前端如何处理用户离开页面和网络波动流式接口在生产环境一定会遇到两类问题用户中途离开、网络抖动导致连接中断。第一次踩到这类问题是在一个文档写作产品里。用户让 AI 生成一篇 2000 字的长文生成到一半用户切换到别的页面或者直接关掉了浏览器标签页。后端如果还在继续调用 OpenAI 接口已经生成的内容就被白白浪费了。后来我在前端加了一个AbortController用户离开页面或者点击停止生成按钮时主动中止 fetch 请求const controller new AbortController(); // 用户点击停止时调用 controller.abort(); await fetchStream(messages, controller.signal);在后端当客户端连接断开时转发循环会抛出ERR_STREAM_PREMATURE_CLOSE之类的错误。如果你用的是 Node.js fetchreader.read()会抛异常你需要在catch里调用upstream.cancel()来中断上游 OpenAI 请求这样就不会继续消耗 token 了。这个链路如果不打通后果就是用户明明已经停止生成了你的成本还在涨。网络波动则是另一回事。如果用户的网络在中断恢复后重新连上但你的应用没有重连机制那用户看到的就是AI 说到一半突然不动了。比较好的处理方式是前端捕获流读取异常后记录已经收到的内容然后提示用户连接已断开你可以点击重试重试时把已收到的内容作为上下文一起发给模型让 AI 接着生成。这样虽然会浪费一点 token但用户感知上比内容丢了重新生成好得多。5. 流式场景下的常见翻车现场与排查思路5.1 问题一第一个 chunk 迟迟不到或收到的是完整 JSON 而不是 SSE这是一个特别典型的坑。你明明在代码里设置了streamTrue但服务端返回的却是完整 JSON前端解析不到增量数据。这类问题 90% 出在请求没有真正把streamTrue传到上游。我在排查一个同事的代码时发现他是用requests库直接请求接口的resp requests.post(url, headersheaders, jsondata) for line in resp.iter_lines(): print(line)问题在于他构造data的时候把stream参数写成了字符串True而不是布尔值True。OpenAI 接口收到stream: True后可能不会识别为流式直接返回完整 JSON。这种问题从代码上看几乎一模一样但行为完全不同。另外如果你在公司或云服务商那里接了一层 API 网关网关默认会做响应缓冲把上游的 SSE 攒成一个完整响应体再返回给客户端。这时你也会看到完整 JSON。判断方法很简单自己用 curl 直接打一次上游接口看响应头Content-Type是不是text/event-stream。如果 curl 直接打是流式打你的网关就不是流式那基本确认是网关在中间做了缓冲需要去网关上关掉响应缓冲或开启流式透传。5.2 问题二前端解析出现data: [DONE]报错或渲染出null字符串我在前面的代码里专门处理了[DONE]但仍有不少同学会拿一个通用的 SSE 解析库直接对接 OpenAI 流解析库把所有data:后面的内容都当成 JSON 去解析于是[DONE]就炸了。还有一个常见问题如果直接把delta.content输出到页面而不判断None页面上会出现很多null文字。因为某些 chunk 里delta.content确实就是null它不是内容而是某种标记。比如第一个 chunk 里delta.role是assistant但delta.content是null再比如触发内容审核拦截的时候delta里可能没有content字段。我在项目里为了防止这类问题前端添加了一个渲染钩子function renderDelta(delta) { const text delta?.content || ; if (!text) return; // 分情况处理如果有引用块、markdown 代码块可以做按格式追加 appendToPanel(text); }同步处理里也尽量保证没有内容就不渲染避免页面上出现一行行 null。5.3 问题三流式打印乱码、缺字尤其是中文场景中文环境下做流式输出最经典的问题是半个字。因为 UTF-8 编码下一个汉字占 3 个字节网络包不可能每次恰好拆在字符边界上。如果你在前端直接new TextDecoder().decode(value)可能会拿到一个残缺的字节序列于是页面上一会儿出现一个 一会儿出现半个字。解决这个问题的方法我刚才已经写了用TextDecoder(utf-8, { stream: true })来解码并且用一个buffer变量拼接不完整的字符串。其实原理很简单——TextDecoder在stream: true模式下会把当前块内无法解析的尾部字节暂存在内部等下一块字节到达时再组合成完整的字符。你不需要自己手动存字节只要别把解码器的状态重建就行。后端如果也是 Python 的requests循环里逐行读取一般不容易出现半个字的问题因为 SSE 是按行发送的一行里不会拆半个字符。但如果你用 Node.js 的getReader()原生二进制块去读上游响应然后又转给自己的前端就必须在前端做同样的缓冲处理。5.4 问题四usage 字段取不到计费系统只能凭估算非流式接口的响应里会有usage字段记录 prompt_tokens、completion_tokens 等数据。但流式接口里默认的每个 chunk 都不带usage因为这个字段如果放在每个 chunk 里那数据量就爆炸了。官方设计是只有在开启stream_options{include_usage: True}时最后一个 chunk 才会携带 usage。有同事跟我反馈说自己明明设置了stream_options但返回的 chunk 里还是没有 usage。我让他打印一下所有 chunk 的末尾尤其是收到[DONE]前的那一个才发现 usage 不在choices[0]里而在 chunk 的最外层{ id: chatcmpl-xxx, object: chat.completion.chunk, created: 1710000000, model: gpt-4o-mini, choices: [], usage: { prompt_tokens: 12, completion_tokens: 168, total_tokens: 180 } }注意这个 chunk 的choices数组是空的所以如果你用chunk.choices[0]去取直接会报 IndexError。我当时看到这段代码第一反应就是要把usage的判断放在choices判断之外先判断if chunk.usage is not None再做其他逻辑。5.5 问题五工具调用function calling模式下流式内容打印出来是碎片化的现在的 GPT 模型支持 function calling。当你让模型调用某个工具时它不是直接输出文本而是输出一个结构化的函数调用参数。在流式模式下函数名和参数也是分片到达的而且参数内容不是给用户看的自然语言而是一段 JSON。我在做一个智能体项目时前端只能展示模型说的一段话却忽略了工具调用的增量结果对话流程走到了工具调用分支页面却什么都没有显示用户以为是 bug。后来我把工具调用和普通文本输出分开处理普通文本追加到聊天气泡工具调用片段则先拼接成一个完整的 JSON等finish_reasontool_calls出现时再整体执行工具。核心判断逻辑是看delta.tool_callsfor chunk in stream: if chunk.choices[0].delta.tool_calls: tool_calls chunk.choices[0].delta.tool_calls for tool_call in tool_calls: if tool_call.id: current_tool_id tool_call.id if tool_call.function.name: current_tool_name tool_call.function.name if tool_call.function.arguments: current_tool_args tool_call.function.arguments参数大都是 JSON 字符串的增量比如{city: 北、京}必须在一个循环里累积拼接直到流程走到finish_reason为tool_calls时你才能去json.loads完整参数。如果不理解这个机制很容易在参数还没传完时就尝试解析得到一堆 JSONDecodeError。6. 让流式输出更好用打印格式化与服务化改造的细节6.1 控制台流式打印的体验优化实时刷新 vs 逐行换行在写 CLI 工具时很多人以为流式打印就是print(chunk, end)实际上这个做法在 IDE 里常常不生效因为 PyCharm 的终端会对标准输出做缓冲。之前我在 PyCharm 里调试代码里加了flushTrue还是一段一段地蹦最后发现是 IDE 的模拟终端模式导致的行为差异。换到系统原生终端跑效果就正常了。如果你希望控制台输出更接近 ChatGPT 的视觉效果可以这样做单词级别的流式输出用end实现同一行连续打印每个 token 之间不换行直到一段完整的话结束才换行。段落级别输出如果内容是 Markdown你想保留段落结构可以在遇到\n\n时单独处理换行正常 token 则原地追加。我还经常给流式输出加一个简单的光标闪烁动画比如在等待第一个 token 到达时先打印一个转圈符号等 token 到了再覆盖掉。这个在小工具里非常加分用户一眼就知道程序在运行而不是卡死了。6.2 从打印到控制台升级为打印到 Web 页面如果你做的是 Web 应用那么打印就不只是控制台日志而是把增量内容实时渲染到页面上。这里有一个我反复权衡过的架构选择到底是直接用 SSE 推文本增量还是用 WebSocket 推整个 chunk我的结论是纯文本聊天场景用 SSE 就够了不需要 WebSocket。SSE 是基于 HTTP 的天然支持断线重连、自动重试、事件 ID 管理而且实现成本低。而 WebSocket 是全双工通信适合需要双向高频交互的场景比如多人协作编辑器、实时白板、游戏。如果你只是想展示 OpenAI 的流式输出SSE 是更轻、更不容易出问题的方式。不过在实际开发中我用了很多次 SSE发现有一个容易踩的坑浏览器对同域名下 SSE 连接数有限制。HTTP/1.1 下Chrome 对同一个域名的最大并发连接数是 6如果你的页面同时打开了多个 SSE 连接后面的连接会被挂起。对于普通聊天应用一个用户同时只开一个 SSE 连接问题不大但如果你做了多标签页聊天、多并发生成就要留意这个限制。解决方案要么是改用 WebSocket要么在后端把多个生成任务的 SSE 合并到一个连接里用事件 ID 区分。6.3 流式输出的 Markdown 渲染先分段、再渲染避免整体卡顿很多 AI 应用界面上都会展示 Markdown。但如果你把流式输出一段一段地直接塞给 Markdown 渲染器性能会很差而且会频繁闪烁。我的经验是做一个双缓冲正在接收的增量文本先追加到一个原始字符串渲染的时候把原始字符串最后一段不完整的内容临时去掉比如最后一个未闭合的代码块标记只渲染已完成的部分。这样做的原因很好理解Markdown 解析器遇到一个没有闭合的代码块时会把后面的整个文本都当成代码块内容导致前面已经渲染完的列表、标题突然全部消失一下非常影响体验。具体实现思路function renderMarkdownStream(fullText) { // 如果 fullText 末尾有未闭合的代码块标记先截掉 let safeText fullText; const fenceCount (safeText.match(//g) || []).length; if (fenceCount % 2 1) { safeText safeText.slice(0, safeText.lastIndexOf()); } // 渲染 safeText renderer.setMarkdown(safeText); }这段逻辑在前后端都适用。前后端交互时你只需要保证每次都快照渲染就不会出现 Markdown 不断抖动的问题。6.4 日志打印的坑别把流式增量全量打进日志文件最后说一个大家容易忽略但影响很大的点日志文件里大量写入流式增量会导致磁盘 IO 压力陡增。我有一个项目上线后遇到过一个诡异的问题打印机物理打印机的打印作业变得特别慢同时服务端日志文件飞快膨胀。后来一排查发现是我们在流式接口里每收到一个 chunk 就写一条日志服务调用一多日志直接爆炸连打印服务所在的同一块磁盘都被撑满了。这个案例虽然不是 OpenAI 接口本身的问题但充分说明了流式打印里的打印如果落到日志文件里需要格外谨慎。我的建议是把流式增量打印到控制台没问题但打日志时只记录元信息比如收到第一个 chunk 的时间、chunk 数量、finish_reason、usage、总耗时等。完整文本等拼接完成后存数据库或者按需归档。这样既保证可观测性又不会被日志量拖垮。如果你确实需要实时查看流式文本内容可以设置一个日志级别开关只在排查问题时打开DEBUG级别生产环境默认INFO把增量文本debug级别输出。这样既能保留排障能力又不会影响正常运行。写在最后的实践心得这篇文章在写的过程中我脑子里一直回放着自己第一次把streamTrue加进去的场景。当时只改了一行代码整个应用的交互质感立刻提升了一个档次那种从转圈到打字机的变化是所有做 AI 应用的人都值得亲自体验一次的。流式输出不是一个高深的概念但它牵扯到 HTTP 协议、前后端协作、异常处理和用户心理学实际做起来细节远比想象中多。如果只能从这篇文章里带走三句话我希望能是这三句流式只改变感知速度不改变真实速度所以别指望它能提高模型生成效率但一定要用它来提升用户体验做流式最好的方式是前端直接解析 OpenAI 的 chunk 格式能透传就透传别在中间层做多余的转码遇到流式问题先抓原始响应用 curl 看一眼是不是真的text/event-stream别依赖任何框架的封装去猜。最后再分享一个小技巧代码里所有打印流式内容的地方都记得加上flushTrue尤其是在 Docker 容器里运行服务时不加这个参数你看到的日志永远是延迟一截的。 
长河君 · 2026/9/11 13:03:23 · 阅读 1.2k 从转圈到逐字输出：OpenAI流式接口改造实战
从“转圈圈”到“逐字输出”一次流式接口改造的真实体验我最早接触 OpenAI 接口流式输出时很多人还在做点击按钮等两三秒一次性把整段结果打印到页面这种交互。结果用户等得心慌又不知道系统到底有没有在跑动不动就重复点击把请求打重了。后来我把接口改成流式打印生成结果体验立刻不一样了——用户能看到文字像打字机一样逐字蹦出来心里踏实反馈也完全变了。这篇内容我会围绕 openai 接口流式返回的完整链路来写从底层机制、Python 后端实现、前端流式接收到实际项目里踩过的坑和排查思路都展开讲。适合正在做 AI 应用、想让交互更贴近 ChatGPT 原版体验的开发者也适合想搞懂流式原理、避免走弯路的新手。1. 从转圈圈到逐字输出流式接口解决的到底是什么问题1.1 用户等待的心理模型为什么一边生成一边显示比结束后统一显示重要一个很反直觉的事实是同样一段文本如果让用户等待 3 秒后一次性看到完整结果用户会感觉等了非常久但如果让用户看到文字实时生成哪怕生成时间拉长到 4 秒用户也觉得响应很快。这不是玄学而是人对不确定性的天然焦虑——无进度地等待是最难熬的。以前我们调 openai 接口的时候用的是最传统的非流式请求response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 介绍一下流式接口}], ) print(response.choices[0].message.content)这种方式很直白请求发出去模型把整段内容都生成完SDK 再把完整的 message 内容返回给你。问题是大模型生成一段长文可能需要好几秒甚至几十秒这段时间请求一直挂着用户在界面上看到的就是一个转圈动画。转圈超过 2 秒用户就开始怀疑是不是出了 bug超过 5 秒相当一部分人已经点了刷新。但如果你把streamTrue打开模型每生成一小块 token 就通过 SSEServer-Sent Events服务端发送事件推给客户端前端拿到一个增量就渲染一个字或一个词整个交互就从等待结果变成了看着 AI 思考并书写。用户不再焦虑甚至会产生一种这模型好聪明的积极感受。做 AI 应用的人应该都体会过这种差异它直接决定了产品的第一印象。1.2 从后端耗时到前端体验流式不改变生成总时长但改变了感知时长有一点需要先说明清楚免得大家产生误解流式接口并不会让模型生成得更快。模型该生成 500 个 token还是得一个 token 一个 token 地算总耗时基本没有变化。流式真正改变的是第一个可见输出的到达时间。在非流式请求里你必须等全部内容生成完毕才收到完整响应。一个 800 token 的回答可能需要 10 秒那用户前 9.9 秒看到的都是空白和转圈。而在流式请求里第一个 token 可能 0.5 秒就到了后面的 token 陆续到达到了就渲染。用户第 1 秒看到前几个字第 3 秒看到一小段第 10 秒看到结尾——虽然最终完成时间没变但已经出现内容的感觉会极大缓解等待焦虑。这个特性特别重要我后来做企业级 AI 聊天助手时深有体会。用户提问后如果界面上一直没有任何反馈在线客服系统里的用户往往就会再发一句在吗或者直接关掉页面。而流式打印模式下用户看到第一个字出来就知道系统已经接管问题并开始处理了这对留存率的影响非常直观。1.3 流式还顺带解决了长响应容易超时的问题除了体验层面流式还有一个很多人忽略的技术价值避免 HTTP 长连接超时。调用非流式接口时如果模型生成时间超过网关或负载均衡器的超时限制比如 60 秒中间层可能直接断开连接客户端收到一个 timeout 错误。尤其在做一些复杂推理、代码生成、长文本总结时生成耗时很容易飙到几十秒非流式请求在这种场景下特别脆弱。而流式请求会持续地往客户端推数据。只要服务器还在不断产生新 token连接就是活跃的网关那边就不会判定为闲置连接而掐断。哪怕这次生成总共花了 90 秒连接也一直是有数据流动的状态这从工程层面绕开了中间设备的超时限制。我当时在生产环境遇到过线上服务频繁报 504 的问题换成流式之后这类报错几乎消失了。2. 流式返回的底层机制stream 参数、SSE 协议与 delta 增量2.1 一次完整的流式响应长什么样想真正掌握 openai 流式打印生成结果不能只停留在加个 streamTrue 就跑通了的程度得知道线上传回来的到底是什么。我用一个最朴素的 HTTP 层面的工具curl来演示这样大家能看清底层长什么样curl https://api.openai.com/v1/chat/completions \ -H Authorization: Bearer YOUR_API_KEY \ -H Content-Type: application/json \ -d { model: gpt-4o-mini, messages: [{role: user, content: 你好}], stream: true }响应不是传统的完整 JSON而是这样一串 SSE 格式的数据data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1710000000,model:gpt-4o-mini,choices:[{index:0,delta:{role:assistant,content:},finish_reason:null}]} data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1710000000,model:gpt-4o-mini,choices:[{index:0,delta:{content:你},finish_reason:null}]} data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1710000000,model:gpt-4o-mini,choices:[{index:0,delta:{content:好},finish_reason:null}]} data: [DONE]每一行data:后面都是一个独立的 JSON 对象这个对象在官方文档里叫chat.completion.chunk也就是一个分片。每个分片里的choices[0].delta是本次推送的增量内容。注意看第一个 chunk 里的delta它带的是role: assistant后面再来的 chunk 才带真正的content文本。等模型生成完了最后一个 chunk 的finish_reason会变成stop或者length、tool_calls等最后再发一个data: [DONE]表示结束。有个值得注意的细节第二个 chunk 里面delta只有content:你并没有重复携带之前的文本每一个 chunk 都是增量不是全量。如果你的代码把每个 chunk 直接赋值给变量而不是做拼接那最后页面上只会显示一个好字这是新手最容易犯的错。后面我会再用代码演示正确的拼接方式。2.2 SSE 协议的三个关键特征SSE 全称是 Server-Sent Events它和普通 HTTP 响应最大的区别是连接一旦建立服务端可以不断往同一个响应里写入新数据客户端则不断读取这些新数据直到连接关闭。这里我要强调三个容易被忽略的特征理解了它们你在排查流式问题时会顺很多Content-Type 必须是text/event-stream。如果你用底层工具直接发请求要留意响应头。如果看到响应头是application/json说明服务端没有按 SSE 方式返回这时候要么是请求参数没带对要么是你用了某个网关把流式缓存成了整体 JSON。每条消息以data:开头消息之间用空行分隔。你可以在服务端自定义event: xxx和id: xxx等字段但 OpenAI 官方接口目前就返回data字段。解析时按空行拆开每一块再按冒号拆字段名和值。连接结束的标志是data: [DONE]。注意[DONE]不是一个 JSON它是 SSE 里的一个特殊结束标记。如果你用 JSON.parse 去解析这行会直接报错所以解析逻辑里必须对data [DONE]做特殊判断。很多人在项目里手写过 SSE 解析工具网上能找到各种版本的代码。但说实话做 OpenAI 应用时你通常不需要自己手写这些直接用官方 SDK 就行。真正需要手写解析的场景是你自己封装一个中间服务把 OpenAI 的流式响应转给你的前端。到那种场景下上面这几个细节就非常关键了。2.3 官方 SDK 是流式友好的别自己乱造轮子官方 Python SDK 对流式的支持已经很完善了你只需要传入streamTrue然后迭代返回对象。一个最基础的流式打印例子长这样from openai import OpenAI client OpenAI(api_keyYOUR_API_KEY) stream client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 写一首关于春天的短诗}], streamTrue, ) for chunk in stream: if chunk.choices[0].delta.content is not None: print(chunk.choices[0].delta.content, end, flushTrue)这个例子虽然短但有几个重要的点chunk.choices[0].delta.content可能为None。比如第一个 chunk 里只有role没有content或者某些特殊情况下delta里根本没有content字段。如果你不做is not None判断None被打印出来就会变成一行None。end是为了把每次输出后的默认换行去掉实现同一行续写flushTrue是为了立刻把缓冲区的内容刷出来否则 Python 的标准输出可能会积攒到一定量才打印你就看不到逐字输出的效果了。迭代器会在收到[DONE]后自动终止你不需要自己在循环里判断结束。这个例子可以直接跑但离真正能用到项目里还差几步比如处理 usage 信息、处理工具调用、处理多个 choice 等。这些我在后面专门讲踩坑的章节里再展开。3. 用 OpenAI SDK 实现流式打印从一段可跑通的代码说起3.1 环境准备版本是一件不该被忽略的事我见过太多人在环境这一步出问题。openai 的 Python SDK 更新速度非常快旧版本和新版本的 API 差别很大。早期版本甚至没有OpenAI这个类用的是openai.ChatCompletion.create新版本统一成了OpenAI()客户端模块结构完全变了。我建议的安装命令pip install -U openai装完以后可以快速验证一下版本python -c import openai; print(openai.__version__)如果你看到的是0.x开头说明版本非常旧建议更新到1.x以上。我的生产环境目前用的是 1.x 系列代码风格和本文保持一致。另外需要提一句现在很多国内开发者是通过 Azure OpenAI 或第三方中转服务来调用 ChatGPT 接口的这种情况下需要额外配置base_url。比如你把 openai SDK 指向 Azure OpenAI 端点时要这么写client OpenAI( azure_endpointhttps://your-resource.openai.azure.com/, api_keyYOUR_AZURE_API_KEY, api_version2024-02-15-preview, )但你说到底还是用官方接口还是 Azure 接口取决于你的业务部署环境。代码结构基本是兼容的只是client的初始化方式不同。我这里不会去细讲某种特定转发的配置细节因为不同的平台差异太大了你按自己服务商的文档来就好。不过请务必记住无论走哪种通道流式返回的 SSE 机制和 chunk 结构基本一致后面的代码思路是通用的。3.2 完整代码一个带状态管理的流式打印函数下面这段代码是我在实际项目中用过的一个相对完整的版本它除了把流式内容打印出来还会自动拼接成完整文本、避免None值污染并在结束后打印本轮 token 消耗import os from openai import OpenAI client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) def stream_chat(user_prompt: str, system_prompt: str ): messages [] if system_prompt: messages.append({role: system, content: system_prompt}) messages.append({role: user, content: user_prompt}) stream client.chat.completions.create( modelgpt-4o-mini, messagesmessages, streamTrue, stream_options{include_usage: True}, # 让最后一个chunk携带usage temperature0.7, ) full_response [] print(Assistant: , end, flushTrue) for chunk in stream: delta chunk.choices[0].delta # 有些 chunk 只有 role没有 content必须跳过 if delta.content is not None: print(delta.content, end, flushTrue) full_response.append(delta.content) # 检查结束原因 if chunk.choices[0].finish_reason is not None: print() # 模型输出结束补一个换行 print(f[finish_reason] {chunk.choices[0].finish_reason}) # usage 信息需要在请求里显式加 stream_options if hasattr(chunk, usage) and chunk.usage is not None: print(f\n[usage] prompt_tokens{chunk.usage.prompt_tokens}, fcompletion_tokens{chunk.usage.completion_tokens}, ftotal_tokens{chunk.usage.total_tokens}) return .join(full_response) if __name__ __main__: result stream_chat(用三句话解释什么是流式接口) print(\n\n完整拼接结果) print(result)说几个我在实际使用中的经验stream_options{include_usage: True}是在 openai 1.x 后期版本里才支持的功能。不加的话chunk.usage一直为None拿不到 token 用量。这个字段加不加影响不大但如果你做商业化产品需要对用户做计费统计这一块就必须搞清楚。finish_reason的判断放在循环里而不是放在循环外。因为最后一个 chunk 会同时带有delta.content内容为空和finish_reasonstop如果你在循环外判断也可以但会丢掉最后一轮 chunk是最接近完成状态的时机这个信息。放在循环里你打印出来的效果更自然。我把print()和返回值分开处理。打印是指实时输出给控制台或日志返回值是完整的拼接文本可以存库、做后续处理。这两个场景是流式接口最常见的两种消费方式兼容并包最省心。3.3 为什么这样设计状态拼装、中断处理与可观测性很多人写第一版流式代码时只会在循环里打印不拼接导致后面的逻辑没法用这个结果。我的经验是流式消费的本质是把增量变成全量。这个全量一方面可以实时打印出去另一方面要存在一个累积变量里。以后你想把对话记录入库、把生成结果发送给另一个 API直接拿这个累积后的字符串就行不用重新调用一次接口。还有一个实战中容易踩的坑是用户中途打断退出。比如你用 Gradio 或 FastAPI 做 Web 服务前端 WebSocket 断开时post 请求的响应通道可能已经没了但你还在for chunk in stream循环里。这时候如果不对异常做处理服务端进程会继续把文本生成完白白消耗 token。解决思路是给请求添加取消逻辑最简单的方式是用asyncio配合httpx的异步客户端或者在销毁时主动调用stream.close()。我在项目里一般会做一个try/finally的保护try: stream client.chat.completions.create(...) for chunk in stream: process(chunk) finally: # 确保连接被释放避免资源泄漏 if stream is not None: stream.close()虽然官方 SDK 的迭代器用完后会自动断开但万一你在循环内部抛异常退出连接未必会立刻释放。养成try/finally的习惯能少很多诡异的连接数上涨问题。另外我强烈建议在流式打印时加上时间戳。生产环境排查问题时如果只看到内容文本你很难判断哪一段花了多少时间。给每一批 chunk 打印一个毫秒级时间戳或者至少记录收到第一个 chunk 的时间和收到[DONE]的时间对分析接口响应慢这类问题特别有用。4. 前端接过流式数据fetch 流式读取的完整链路4.1 为什么不能直接用 axios 处理流式响应后端拿到 OpenAI 的流式结果后通常有两种做法一种是直接把 OpenAI 的 SSE 流转发给浏览器另一种是后端自己把内容聚合后再通过 WebSocket 或 SSE 推给前端。这里我先讲前端怎么消费 SSE因为这是大部分 AI 对话应用最关心的一环。很多前端同学习惯用 axios但 axios 默认并不会给你流式读取的能力。axios 会把整个响应体下载完再交给 then 回调也就是说即使你用了stream: true前端看到的还是一个整体字符串完全失去实时效果。所以这里我推荐用浏览器原生fetch通过ReadableStream逐块读取。下面是一段可以直接放到 Vue 或 React 项目里的前端代码以 fetch 为例async function fetchStream(messages) { const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json, }, body: JSON.stringify({ messages }), }); if (!response.ok) { throw new Error(HTTP error! status: ${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 }); // SSE 消息之间以空行分隔这里按 \n\n 拆出完整消息 const lines buffer.split(\n\n); buffer lines.pop(); for (const line of lines) { if (!line.startsWith(data:)) continue; const data line.replace(/^data:\s*/, ); if (data [DONE]) return; try { const json JSON.parse(data); const content json.choices?.[0]?.delta?.content; if (content) { // 这里追加到你的页面显示变量 console.log(增量:, content); } } catch (err) { console.warn(解析 chunk 失败, data, err); } } } }我来说明几个关键点response.body.getReader()返回的是一个流读取器reader.read()每次返回{ done, value }。done为 true 表示流结束。TextDecoder(utf-8, { stream: true })里的stream: true很重要。因为浏览器收到的二进制块可能把一个 UTF-8 字符截断成两半如果你不告诉解码器这是一个流那它就只能按独立的块解码可能出现乱码。buffer变量用来缓冲不完整的 SSE 消息。因为reader.read()返回的块大小不固定可能一条data:消息被拆成两次读取也可能一次读取包含多条消息。buffer.split(\n\n)以后最后一个元素要放回 buffer等下一轮数据到齐再处理。4.2 后端转发时的两个关键设计流式透传与自定义事件格式很多情况下你不会直接把 OpenAI 的api.openai.com暴露给浏览器一方面有安全风险另一方面你需要控制权限、限流、计费。所以你会在自己的后端做一个转发接口这个接口同时扮演中间人的角色。用 Node.js 写一个转发接口其实是比较自然的// Express 或 Koa 场景下 app.post(/api/chat, async (req, res) { res.setHeader(Content-Type, text/event-stream); res.setHeader(Cache-Control, no-cache); res.setHeader(Connection, keep-alive); const upstream await fetch(https://api.openai.com/v1/chat/completions, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${process.env.OPENAI_API_KEY}, }, body: JSON.stringify({ model: gpt-4o-mini, messages: req.body.messages, stream: true, }), }); const reader upstream.body.getReader(); const decoder new TextDecoder(); try { while (true) { const { done, value } await reader.read(); if (done) break; // 直接按原样转发给前端 res.write(decoder.decode(value, { stream: true })); } } finally { reader.releaseLock(); res.end(); } });这套做法的核心好处是前端和后端都用同一套 SSE 解析逻辑格式统一不容易出错。你不用在后端做任何内容解析和拼接OpenAI 返回什么你转发什么。前端直接解析 OpenAI 的 chunk 结构即可。当然也有一些场景下你更希望在服务端把文本聚合好了再用自定义的 SSE 格式推给前端比如你要在流式响应中附带额外的元数据像本轮消耗的 token、命中的知识库文档 ID、搜索结果的来源链接等。这时候你可以定义自己的事件格式比如data: {type: start, conversation_id: xxx} data: {type: delta, content: 你好} data: {type: search_result, results: [...]} data: {type: done, usage: {total_tokens: 123}}前端拿到后按type字段分流处理比如delta类型就追加文本search_result类型就展示引用卡片。这样比直接透传多了一层控制力代价是后端需要自己解析和拼装 OpenAI 的 chunk逻辑复杂度会上升。我的建议是简单场景直接透传复杂场景后端自定义事件流不要两头都做解析否则维护成本翻倍。4.3 连接安全与中断恢复前端如何处理用户离开页面和网络波动流式接口在生产环境一定会遇到两类问题用户中途离开、网络抖动导致连接中断。第一次踩到这类问题是在一个文档写作产品里。用户让 AI 生成一篇 2000 字的长文生成到一半用户切换到别的页面或者直接关掉了浏览器标签页。后端如果还在继续调用 OpenAI 接口已经生成的内容就被白白浪费了。后来我在前端加了一个AbortController用户离开页面或者点击停止生成按钮时主动中止 fetch 请求const controller new AbortController(); // 用户点击停止时调用 controller.abort(); await fetchStream(messages, controller.signal);在后端当客户端连接断开时转发循环会抛出ERR_STREAM_PREMATURE_CLOSE之类的错误。如果你用的是 Node.js fetchreader.read()会抛异常你需要在catch里调用upstream.cancel()来中断上游 OpenAI 请求这样就不会继续消耗 token 了。这个链路如果不打通后果就是用户明明已经停止生成了你的成本还在涨。网络波动则是另一回事。如果用户的网络在中断恢复后重新连上但你的应用没有重连机制那用户看到的就是AI 说到一半突然不动了。比较好的处理方式是前端捕获流读取异常后记录已经收到的内容然后提示用户连接已断开你可以点击重试重试时把已收到的内容作为上下文一起发给模型让 AI 接着生成。这样虽然会浪费一点 token但用户感知上比内容丢了重新生成好得多。5. 流式场景下的常见翻车现场与排查思路5.1 问题一第一个 chunk 迟迟不到或收到的是完整 JSON 而不是 SSE这是一个特别典型的坑。你明明在代码里设置了streamTrue但服务端返回的却是完整 JSON前端解析不到增量数据。这类问题 90% 出在请求没有真正把streamTrue传到上游。我在排查一个同事的代码时发现他是用requests库直接请求接口的resp requests.post(url, headersheaders, jsondata) for line in resp.iter_lines(): print(line)问题在于他构造data的时候把stream参数写成了字符串True而不是布尔值True。OpenAI 接口收到stream: True后可能不会识别为流式直接返回完整 JSON。这种问题从代码上看几乎一模一样但行为完全不同。另外如果你在公司或云服务商那里接了一层 API 网关网关默认会做响应缓冲把上游的 SSE 攒成一个完整响应体再返回给客户端。这时你也会看到完整 JSON。判断方法很简单自己用 curl 直接打一次上游接口看响应头Content-Type是不是text/event-stream。如果 curl 直接打是流式打你的网关就不是流式那基本确认是网关在中间做了缓冲需要去网关上关掉响应缓冲或开启流式透传。5.2 问题二前端解析出现data: [DONE]报错或渲染出null字符串我在前面的代码里专门处理了[DONE]但仍有不少同学会拿一个通用的 SSE 解析库直接对接 OpenAI 流解析库把所有data:后面的内容都当成 JSON 去解析于是[DONE]就炸了。还有一个常见问题如果直接把delta.content输出到页面而不判断None页面上会出现很多null文字。因为某些 chunk 里delta.content确实就是null它不是内容而是某种标记。比如第一个 chunk 里delta.role是assistant但delta.content是null再比如触发内容审核拦截的时候delta里可能没有content字段。我在项目里为了防止这类问题前端添加了一个渲染钩子function renderDelta(delta) { const text delta?.content || ; if (!text) return; // 分情况处理如果有引用块、markdown 代码块可以做按格式追加 appendToPanel(text); }同步处理里也尽量保证没有内容就不渲染避免页面上出现一行行 null。5.3 问题三流式打印乱码、缺字尤其是中文场景中文环境下做流式输出最经典的问题是半个字。因为 UTF-8 编码下一个汉字占 3 个字节网络包不可能每次恰好拆在字符边界上。如果你在前端直接new TextDecoder().decode(value)可能会拿到一个残缺的字节序列于是页面上一会儿出现一个 一会儿出现半个字。解决这个问题的方法我刚才已经写了用TextDecoder(utf-8, { stream: true })来解码并且用一个buffer变量拼接不完整的字符串。其实原理很简单——TextDecoder在stream: true模式下会把当前块内无法解析的尾部字节暂存在内部等下一块字节到达时再组合成完整的字符。你不需要自己手动存字节只要别把解码器的状态重建就行。后端如果也是 Python 的requests循环里逐行读取一般不容易出现半个字的问题因为 SSE 是按行发送的一行里不会拆半个字符。但如果你用 Node.js 的getReader()原生二进制块去读上游响应然后又转给自己的前端就必须在前端做同样的缓冲处理。5.4 问题四usage 字段取不到计费系统只能凭估算非流式接口的响应里会有usage字段记录 prompt_tokens、completion_tokens 等数据。但流式接口里默认的每个 chunk 都不带usage因为这个字段如果放在每个 chunk 里那数据量就爆炸了。官方设计是只有在开启stream_options{include_usage: True}时最后一个 chunk 才会携带 usage。有同事跟我反馈说自己明明设置了stream_options但返回的 chunk 里还是没有 usage。我让他打印一下所有 chunk 的末尾尤其是收到[DONE]前的那一个才发现 usage 不在choices[0]里而在 chunk 的最外层{ id: chatcmpl-xxx, object: chat.completion.chunk, created: 1710000000, model: gpt-4o-mini, choices: [], usage: { prompt_tokens: 12, completion_tokens: 168, total_tokens: 180 } }注意这个 chunk 的choices数组是空的所以如果你用chunk.choices[0]去取直接会报 IndexError。我当时看到这段代码第一反应就是要把usage的判断放在choices判断之外先判断if chunk.usage is not None再做其他逻辑。5.5 问题五工具调用function calling模式下流式内容打印出来是碎片化的现在的 GPT 模型支持 function calling。当你让模型调用某个工具时它不是直接输出文本而是输出一个结构化的函数调用参数。在流式模式下函数名和参数也是分片到达的而且参数内容不是给用户看的自然语言而是一段 JSON。我在做一个智能体项目时前端只能展示模型说的一段话却忽略了工具调用的增量结果对话流程走到了工具调用分支页面却什么都没有显示用户以为是 bug。后来我把工具调用和普通文本输出分开处理普通文本追加到聊天气泡工具调用片段则先拼接成一个完整的 JSON等finish_reasontool_calls出现时再整体执行工具。核心判断逻辑是看delta.tool_callsfor chunk in stream: if chunk.choices[0].delta.tool_calls: tool_calls chunk.choices[0].delta.tool_calls for tool_call in tool_calls: if tool_call.id: current_tool_id tool_call.id if tool_call.function.name: current_tool_name tool_call.function.name if tool_call.function.arguments: current_tool_args tool_call.function.arguments参数大都是 JSON 字符串的增量比如{city: 北、京}必须在一个循环里累积拼接直到流程走到finish_reason为tool_calls时你才能去json.loads完整参数。如果不理解这个机制很容易在参数还没传完时就尝试解析得到一堆 JSONDecodeError。6. 让流式输出更好用打印格式化与服务化改造的细节6.1 控制台流式打印的体验优化实时刷新 vs 逐行换行在写 CLI 工具时很多人以为流式打印就是print(chunk, end)实际上这个做法在 IDE 里常常不生效因为 PyCharm 的终端会对标准输出做缓冲。之前我在 PyCharm 里调试代码里加了flushTrue还是一段一段地蹦最后发现是 IDE 的模拟终端模式导致的行为差异。换到系统原生终端跑效果就正常了。如果你希望控制台输出更接近 ChatGPT 的视觉效果可以这样做单词级别的流式输出用end实现同一行连续打印每个 token 之间不换行直到一段完整的话结束才换行。段落级别输出如果内容是 Markdown你想保留段落结构可以在遇到\n\n时单独处理换行正常 token 则原地追加。我还经常给流式输出加一个简单的光标闪烁动画比如在等待第一个 token 到达时先打印一个转圈符号等 token 到了再覆盖掉。这个在小工具里非常加分用户一眼就知道程序在运行而不是卡死了。6.2 从打印到控制台升级为打印到 Web 页面如果你做的是 Web 应用那么打印就不只是控制台日志而是把增量内容实时渲染到页面上。这里有一个我反复权衡过的架构选择到底是直接用 SSE 推文本增量还是用 WebSocket 推整个 chunk我的结论是纯文本聊天场景用 SSE 就够了不需要 WebSocket。SSE 是基于 HTTP 的天然支持断线重连、自动重试、事件 ID 管理而且实现成本低。而 WebSocket 是全双工通信适合需要双向高频交互的场景比如多人协作编辑器、实时白板、游戏。如果你只是想展示 OpenAI 的流式输出SSE 是更轻、更不容易出问题的方式。不过在实际开发中我用了很多次 SSE发现有一个容易踩的坑浏览器对同域名下 SSE 连接数有限制。HTTP/1.1 下Chrome 对同一个域名的最大并发连接数是 6如果你的页面同时打开了多个 SSE 连接后面的连接会被挂起。对于普通聊天应用一个用户同时只开一个 SSE 连接问题不大但如果你做了多标签页聊天、多并发生成就要留意这个限制。解决方案要么是改用 WebSocket要么在后端把多个生成任务的 SSE 合并到一个连接里用事件 ID 区分。6.3 流式输出的 Markdown 渲染先分段、再渲染避免整体卡顿很多 AI 应用界面上都会展示 Markdown。但如果你把流式输出一段一段地直接塞给 Markdown 渲染器性能会很差而且会频繁闪烁。我的经验是做一个双缓冲正在接收的增量文本先追加到一个原始字符串渲染的时候把原始字符串最后一段不完整的内容临时去掉比如最后一个未闭合的代码块标记只渲染已完成的部分。这样做的原因很好理解Markdown 解析器遇到一个没有闭合的代码块时会把后面的整个文本都当成代码块内容导致前面已经渲染完的列表、标题突然全部消失一下非常影响体验。具体实现思路function renderMarkdownStream(fullText) { // 如果 fullText 末尾有未闭合的代码块标记先截掉 let safeText fullText; const fenceCount (safeText.match(//g) || []).length; if (fenceCount % 2 1) { safeText safeText.slice(0, safeText.lastIndexOf()); } // 渲染 safeText renderer.setMarkdown(safeText); }这段逻辑在前后端都适用。前后端交互时你只需要保证每次都快照渲染就不会出现 Markdown 不断抖动的问题。6.4 日志打印的坑别把流式增量全量打进日志文件最后说一个大家容易忽略但影响很大的点日志文件里大量写入流式增量会导致磁盘 IO 压力陡增。我有一个项目上线后遇到过一个诡异的问题打印机物理打印机的打印作业变得特别慢同时服务端日志文件飞快膨胀。后来一排查发现是我们在流式接口里每收到一个 chunk 就写一条日志服务调用一多日志直接爆炸连打印服务所在的同一块磁盘都被撑满了。这个案例虽然不是 OpenAI 接口本身的问题但充分说明了流式打印里的打印如果落到日志文件里需要格外谨慎。我的建议是把流式增量打印到控制台没问题但打日志时只记录元信息比如收到第一个 chunk 的时间、chunk 数量、finish_reason、usage、总耗时等。完整文本等拼接完成后存数据库或者按需归档。这样既保证可观测性又不会被日志量拖垮。如果你确实需要实时查看流式文本内容可以设置一个日志级别开关只在排查问题时打开DEBUG级别生产环境默认INFO把增量文本debug级别输出。这样既能保留排障能力又不会影响正常运行。写在最后的实践心得这篇文章在写的过程中我脑子里一直回放着自己第一次把streamTrue加进去的场景。当时只改了一行代码整个应用的交互质感立刻提升了一个档次那种从转圈到打字机的变化是所有做 AI 应用的人都值得亲自体验一次的。流式输出不是一个高深的概念但它牵扯到 HTTP 协议、前后端协作、异常处理和用户心理学实际做起来细节远比想象中多。如果只能从这篇文章里带走三句话我希望能是这三句流式只改变感知速度不改变真实速度所以别指望它能提高模型生成效率但一定要用它来提升用户体验做流式最好的方式是前端直接解析 OpenAI 的 chunk 格式能透传就透传别在中间层做多余的转码遇到流式问题先抓原始响应用 curl 看一眼是不是真的text/event-stream别依赖任何框架的封装去猜。最后再分享一个小技巧代码里所有打印流式内容的地方都记得加上flushTrue尤其是在 Docker 容器里运行服务时不加这个参数你看到的日志永远是延迟一截的。