这个系列已经写到第三期。前面我们把RAG的召回链路、Agent的基础循环都摸了一遍这周单独把“流式输出”拎出来聊。因为不管你做的是知识库问答还是带工具调用的Agent模型返回的方式会直接影响用户体感和系统稳定性。Or我可以换个说法不是技术的炫技而是工程交付里绕不开的交互细节。不管你是刚入门Python API调用还是已经在生产环境对接大模型这篇都值得花几分钟看看。1. 为什么流式输出比你想象的更重要1.1 用户等待的不是模型而是你的架构我第一次做RAG问答Demo的时候为了图省事直接用了非流式接口。前端点了“提问”后端就傻乎乎地等模型把整段答案生成完再一次性返回给浏览器。结果是什么一段500字的回答模型要跑20秒左右用户盯着空白页面发呆最坏的情况下会以为服务挂了疯狂重复点击。后来我在日志里看到有一半的请求都是用户等不及之后手动刷新造成的重复请求。那一刻我才意识到瓶颈从来不是模型算得快不快而是你愿不愿意把“正在生成”这种状态及时暴露给用户。流式输出的第一个价值就是让用户看到内容像打字机一样逐字蹦出来。这样做不仅让等待变得可感知还能带来一个隐藏优势用户更愿意继续看下去而不是中途关掉。尤其在RAG场景里用户发起问题后你还需要先做检索再送进大模型这中间是有真实延时的。如果不做流式用户会把这个延时全部怪到“系统卡了”上如果你把检索状态、模型生成状态一步步推给前端用户就会觉得“这个系统在进行多步思考”信任度完全不一样。1.2 从“等结果”到“看过程”SSE真正解决了什么流式输出现在的标配是SSE全称Server-Sent Events。简单理解就是服务端保持一条HTTP长连接然后分块把数据推到客户端而不是等所有数据都攒齐再一股脑发出去。关键点是“分块”而且分块的频率很高。在OpenAI官方库的接口里只要你设置了streamTrue返回的就是一个可以迭代的流式对象。每次迭代会得到一个chunkchunk里只有极少量增量内容。客户端可以实时接收并刷新界面。这里我常拿吃饭举例子。非流式就像是去餐厅点餐你在座位干等了40分钟然后后厨一次性把你点的菜全端上来。流式则是厨师做一道菜服务员端一道菜虽然同样等了40分钟但你从第5分钟就开始吃上了。看过这个对比你就能理解为什么现在的对话式应用几乎都在用流式。尤其是RAG场景检索可能需要2到5秒模型生成又需要5到20秒把这两段耗时揉在一起不做流式体验基本就是灾难。我在很多项目里见过“RAG一顿操作猛如虎最后用户看到个白屏就跑了”多半就是没做流式导致的。2. 基础用法streamTrue和增量解析2.1 最小可用的流式对话代码先看一段最基础的新版SDK写法。新版指的是OpenAI官方库v1.x接口命名和0.x时代完全是两回事。from openai import OpenAI client OpenAI(api_keyyour-api-key, base_urlyour-endpoint) response client.chat.completions.create( modelgpt-4o-mini, messages[ {role: user, content: 介绍一下RAG的核心流程} ], streamTrue ) for chunk in response: delta chunk.choices[0].delta content getattr(delta, content, None) if content: print(content, end, flushTrue) print()这里面最关键的两个东西streamTrue告诉服务端返回流式对象。循环迭代每轮拿到的chunk都只是一个非常小的数据切片。注意getattr(delta, content, None)这个写法官方响应对象里有时content字段是None如果你直接做字符串拼接会报错。稳妥的处理方式是先判断再取内容。2.2 逐行拆解chunk里的内容增量第一次打印流式返回的chunk你会被它的结构吓到因为实在是太碎了。我给你看一下一个chunk通常包含什么{ id: chatcmpl-xxx, object: chat.completion.chunk, created: 1690000000, model: gpt-4o-mini, choices: [ { index: 0, delta: {content: RAG}, finish_reason: null } ] }也就是说模型每生成一小段文本SDK就会包装成这样一个chunk推给你。你在循环里做的操作其实很简单不断从chunk.choices[0].delta.content里取那段增量然后追加到完整回答里。很多新手会踩这个误区以为流式返回的第一个chunk包含了完整消息的开头后面的chunk是完整消息的后续。实际上每个chunk都是独立的增量不是累积量。如果你想把完整回答存下来必须自己维护一个变量full_answer for chunk in response: delta chunk.choices[0].delta if delta.content: full_answer delta.content print(delta.content, end, flushTrue)另外finish_reason字段值得关注。当它不再是None而是stop时说明生成结束了。如果中途遇到length说明因为token限制被截断。你可以用这个标志去判断是否继续等下一轮。2.3 版本差异新老SDK写法迁移如果你看过更早的教程会发现当时流行的写法是import openai response openai.ChatCompletion.create( modelgpt-3.5-turbo, messages[], streamTrue )这是老版0.x的接口。现在官方库已经升级到v1.x原来那种直接挂在openai模块下的调用方式被废弃了。今天的推荐做法是实例化一个OpenAI客户端再调用client.chat.completions.create。如果你手里还有老代码迁移时需要注意这几个点openai.ChatCompletion.create改成client.chat.completions.create原来的response.choices[0].message变成了response.choices[0].delta流式时max_tokens建议改成max_tokens或max_completion_tokens不同模型有差异这个坑我在帮某个朋友迁移代码时踩过一遍。他照着两年前的教程写结果一跑就报AttributeError: module openai has no attribute ChatCompletion其实就是SDK版本问题。我的建议是遇到类似报错先看官方库的changelog再顺手把依赖版本固定下来不要用裸的或latest。3. RAG场景落地检索→补全→流式返回3.1 设计一个RAG问答接口的请求/响应结构到了RAG场景流式输出不只是把模型返回的文本转发给前端更要把整个问答链路的状态都暴露出去。所以我在实际接口设计里会把响应格式拆成几种事件类型status表示当前正在执行什么步骤比如“正在检索文档”retrieval_result检索完成后把命中的文档标题或片段摘要推给前端让用户知道系统确实找到了相关资料answer模型开始流式生成回答推送增量内容done生成结束附带完整的统计信息比如耗时、命中文件数这种设计其实是对SSE消息事件的一种自定义。OpenAI库本身只负责流式输出文本你要做的是在这个基础上包一层你自己的事件协议。前端拿到的数据看起来应该长这样data: {event: status, message: 正在检索...} data: {event: retrieval_result, count: 3} data: {event: answer, content: RAG的核心流程包括} data: {event: answer, content: 检索、重排、注入和生成。} data: {event: done, duration: 5.2}如果你用的是Spring Boot后端的SseEmitter或者FastAPI的StreamingResponse都可以按照这个思路实现。3.2 完整的RAG流式实现代码下面这段是我常用的最小可复现版本。我用一个文档列表模拟知识库用简单的余弦相似度做召回重点展示流式输出怎么跟RAG结合。import os import time import numpy as np from openai import OpenAI from fastapi import FastAPI from fastapi.responses import StreamingResponse client OpenAI(api_keyos.environ.get(OPENAI_API_KEY)) # 模拟文档库 documents [ RAG通过检索外部知识库来增强模型生成能力。, Agent需要感知环境、做规划并逐步执行工具调用。, 流式输出可以显著提升对话体验减少用户的等待焦虑。, 向量检索的核心是文本嵌入和高维空间相似度计算。 ] # 简单获取embedding def embed_text(text: str) - list[float]: resp client.embeddings.create( modeltext-embedding-3-small, inputtext ) return resp.data[0].embedding # 简单召回计算余弦相似度返回TopK def retrieve(query: str, top_k: int 2) - list[str]: q_vec embed_text(query) scored [] for doc in documents: doc_vec embed_text(doc) score np.dot(q_vec, doc_vec) / (np.linalg.norm(q_vec) * np.linalg.norm(doc_vec)) scored.append((score, doc)) scored.sort(reverseTrue) return [doc for _, doc in scored[:top_k]] # 核心生成器把检索状态和模型流式输出一起发出去 def rag_stream(question: str): # 1. 发送状态事件 yield data: {event: status, message: 正在检索相关文档}\n\n time.sleep(1) # 模拟检索耗时 # 2. 执行检索 hits retrieve(question) yield fdata: {{event: retrieval_result, count: {len(hits)}}}\n\n # 3. 构造带上下文的prompt context \n.join(hits) messages [ {role: system, content: 你是知识库问答助手请严格基于参考内容回答。}, {role: user, content: f参考内容\n{context}\n\n问题{question}} ] # 4. 流式调用模型 stream client.chat.completions.create( modelgpt-4o-mini, messagesmessages, streamTrue ) for chunk in stream: delta chunk.choices[0].delta if delta.content: content delta.content.replace(\n, \\n) yield fdata: {{event: answer, content: {content}}}\n\n yield data: {event: done, duration: 1}\n\n app FastAPI() app.post(/rag/chat) async def chat(question: str): return StreamingResponse(rag_stream(question), media_typetext/event-stream)这里有几个细节值得说明第一不要指望你的RAG检索逻辑一定能在几毫秒内跑完所以第一个status事件很重要。它让用户知道“系统动了”。第二把文档命中数量发出去是一种性价比很高的体验优化。用户看到count: 2就知道这个回答不是瞎编的是有材料支撑的。这比生硬地说“本回答基于知识库生成”有效得多。第三SSE格式里的事件之间要用\n\n分隔字段内部不要出现裸换行。如果你在content里返回了真实的换行符前端解析会有问题所以我在示例中把\n替换成了\\n前端再反转义。这个细节我踩过坑。3.3 状态提示让用户知道系统正在干什么你在做RAG的时候会不会遇到这种情况用户问了一个问题系统检索到的文档确实包含相关内容但模型生成的第一句话是“根据资料显示……”然后开始了漫长的输出。这个时候用户可能已经在心里打了低分。我的常用策略是在前端把状态提示和时间线展示出来。检索前显示“正在检索”检索完显示“已找到相关内容正在组织回答”生成过程中以网格动画或打字机效果展示。即使模型生成的内容中途偶尔停顿用户也会认为这是正常节奏而不是卡死。为了让这些状态前端能识别后端的事件协议最好从一开始就设计成结构化字段而不是直接推纯文本。我建议至少包含event和data两个层级事件类型要稳定别今天叫status明天叫progress。4. Agent场景落地把工具调用的“思考过程”也流式出来4.1 流式中function_call增量如何拼接Agent和普通对话最大的区别在于模型不仅要生成自然语言还可能中途发出“要调用某个工具”的请求。如果你的Agent不流式用户就会看到转圈圈转很久然后突然蹦出来一个回答。这勉强能接受但一旦遇到多工具调用一轮对话需要30秒甚至更久体验会非常糟。流式场景下模型每生成到一个决策节点会把delta里的function_call字段填充进来。打开流式数据看你会发现这一类chunk的delta大致是{ delta: { function_call: { name: search_documents, arguments: } } }然后就出现一串只带arguments增量的chunk{ delta: { function_call: { arguments: {\query\: \ } } }这里最容易踩坑的地方在于name和arguments是分开推送的而且arguments往往是一点一点过来的。你需要累加这些碎片直到某个chunk的finish_reason为function_call才能认为完整参数已经到位。我记得第一次跑通Agent流式时最惊讶的就是原来工具调用的JSON参数也是像聊天文本一样“吐”出来的。当时我直接用json.loads去解析累积参数结果第一次就报了json.decoder.JSONDecodeError因为中途连接断了参数不完整。所以请你务必要做异常捕获参数不完整时就要触发重试或提醒用户。4.2 一个带工具调用的流式对话循环我整理了一个简化版的Agent流式循环。核心思想是用messages保存对话历史。首次调用模型时设置tools和一个工具映射。循环里区分三种情况正常文本增量、函数调用增量、生成结束。生成结束时如果带有function_call就执行工具把工具结果作为新的系统消息加入messages然后再次发起流式调用。如果没有函数调用就结束整个循环。from openai import OpenAI import json client OpenAI(api_keyyour-api-key) def search_documents(query): return RAG的检索模块负责从知识库中找出相关片段。 tools [ { type: function, function: { name: search_documents, description: 搜索知识库文档, parameters: { type: object, properties: { query: {type: string, description: 搜索关键词} }, required: [query] } } } ] tool_map { search_documents: search_documents } def run_agent_stream(user_input): messages [{role: user, content: user_input}] while True: stream client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, streamTrue ) content_buffer function_name arguments_buffer for chunk in stream: delta chunk.choices[0].delta finish_reason chunk.choices[0].finish_reason if delta.content: content_buffer delta.content yield ftext|{content_buffer} if delta.function_call: if delta.function_call.name: function_name delta.function_call.name yield ffunction|{function_name} if delta.function_call.arguments: arguments_buffer delta.function_call.arguments if finish_reason function_call: arguments json.loads(arguments_buffer) messages.append({ role: assistant, content: None, function_call: { name: function_name, arguments: arguments_buffer } }) tool_result tool_map[function_name](**arguments) messages.append({ role: tool, content: tool_result }) break if finish_reason stop: return这段代码为了演示把函数调用和文本输出混在一个generator里用text|或function|前缀区分。实际生产里建议把事件类型拆成结构化对象前端解析更容易。有两点值得补充content为None的assistant消息是Agent对话里的标准写法。如果你漏了这步直接把工具结果append进去模型会不知道这段工具结果对应哪个调用某些接口会报错。arguments_buffer里的JSON字符串在传输中途可能会被截断。如果在生产环境我通常会先做一次轻量校验比如判断是否以正常花括号结尾再尝试解析。4.3 前端如何区分文本流和函数调用流Agent流式输出到了前端如果处理不好用户会看到很奇怪的内容比如突然翻出一段工具名称。所以前端也需要跟着做事件分流。我的经验是让后端把事件类型作为event字段发出来不要只发裸数据。前端拿到SSE事件后先判断eventevent: text追加到消息气泡event: function在UI上显示“正在调用工具search_documents”event: status更新系统状态标签event: done结束加载态展示完整信息如果后端没有做事件封装前端就只能在字符串里查找特殊前缀这种方式非常脆弱。我强烈建议后端统一标准事件格式这样后续无论是Web端还是App端解析逻辑都只有一套。5. 避坑记录与排查清单5.1 我在生产环境碰到的5个经典问题第一流式响应超时。默认的HTTP客户端超时设置可能只覆盖连接阶段但流式响应需要持续等待模型生成如果你设置的超时时间太短服务端还没吐完客户端就断开了。我习惯把timeout设置为一个可以覆盖整次生成时长的值比如30秒甚至更长。另外要区分连接超时和读取超时流式场景下读取超时更关键。第二输出内容被截断成半个字符。做过中文流式的人大概率遇到这个问题。模型按token输出某些语言模型的tokenizer会把一个汉字拆成多个token你在拼接字符串时可能先拿到半个字符导致前端渲染乱码。解决思路是后端不要用裸字符流直接转发给前端而是对增量内容做缓冲。比如检测到不完整的UTF-8字符时等下一个chunk补齐后再发送。第三SDK版本差异导致接口不兼容。这一点在基础部分提到过。实际项目里如果你同时装了多套SDK依赖很容易出现调用链用了老接口新接口又混进去的情况。我建议在项目根目录固定依赖版本并做一次接口回归测试。第四重复请求导致重复流式输出。很多用户看到页面没反应会反复点击如果你没有做前端防抖后端就可能同时发起多个流式请求。这不仅浪费时间还可能重复消耗接口额度。我用得最多的方案是前端在收到第一个chunk前禁用发送按钮后端再加一层幂等键校验双保险。第五Agent工具调用流式返回的arguments不完整。这种情况除了网络中断还有可能是模型生成到一半被内容过滤或长度限制打断了。我的应对方法是在流式循环中判断finish_reason如果为length或异常中断就把当前累积的arguments作为无效调用丢弃重新发起一次带更充分上下文的重试。5.2 一张表速查流式输出常见问题问题现象可能原因解决方向前端白屏很久才有内容没有开启stream或后端未及时flush设置streamTrue使用SSE发送增量输出断在中间读取超时或网络中断调大timeout增加重试机制中文显示乱码或半个字tokenizer切割导致字符不完整对增量做UTF-8缓冲补齐后再发送接口报AttributeErrorSDK版本过旧或过新查看官方文档更新到v1.x并固定版本参数解析JSONDecodeErrorfunction_call arguments不完整累积参数并校验截断时重试用户重复请求前端未禁用点击在收到首块前禁用按钮后端加幂等第一次请求很慢后续很快首次加载embedding或连接池冷启动预热资源复用embedding模型或连接池Agent工具调用结束后没有继续生成messages缺少assistant function_call按标准格式把function_call和tool消息写入历史这里面有些问题不是一次就能遇到但如果你做过多个项目大概率会撞上其中七八个。5.3 关于重试与网络中断的经验流式输出的重试策略和非流式不太一样。非流式接口返回一个完整结果失败了直接重新请求就行。流式接口的问题是你可能已经给前端推送了一部分内容重试又得从第一个字开始会让用户看到内容回退或歪歪扭扭的重复。我的做法是重试尽量只在“还没有推送任何内容”或者“刚推了第一帧就失败”的阶段进行。一旦已经连续推送了较多内容就不要自动重试而是给用户展示一个可点击的“重新生成”按钮。具体实现里我习惯把流式生成器包一层状态判断例如stream_started False for chunk in stream: if delta.content: if not stream_started: stream_started True yield ...如果stream_started为False时就异常中断可以安全重试如果为True则只记录错误不自动重发。此外SSE连接本身有默认的keep-alive机制。有些代理服务器如果长时间收不到数据会主动断开连接。如果模型生成慢你可以在后端每隔15秒发一个heartbeat注释行或空事件保持连接活跃。这个技巧在跨网络环境时特别管用。这个系列进行到现在我最大的感受是流式输出不是把streamTrue加上就完事它更像是一套前后端配合的交互协议。刚开始你可能只追求“内容会动”做到后面你会发现事件结构、异常处理、重试策略、前端渲染这些部分才真正决定了一个对话产品能不能落地。按我现在的习惯接到一个新项目时我会先把消息协议定义清楚再谈模型选型和RAG召回。协议定好了后面的事情会顺很多。希望这篇能让你少踩几个我踩过的坑。