这个系列已经写到第三期。前面我们把RAG的召回链路、Agent的基础循环都摸了一遍，这周单独把“流式输出”拎出来聊。因为不管你做的是知识库问答，还是带工具调用的Agent，模型返回的方式会直接影响用户体感和系统稳定性。Or我可以换个说法：不是技术的炫技，而是工程交付里绕不开的交互细节。不管你是刚入门Python API调用，还是已经在生产环境对接大模型，这篇都值得花几分钟看看。

1. 为什么流式输出比你想象的更重要

1.1 用户等待的不是模型，而是你的架构

我第一次做RAG问答Demo的时候，为了图省事，直接用了非流式接口。前端点了“提问”，后端就傻乎乎地等模型把整段答案生成完，再一次性返回给浏览器。

结果是什么？

一段500字的回答，模型要跑20秒左右，用户盯着空白页面发呆，最坏的情况下会以为服务挂了，疯狂重复点击。后来我在日志里看到，有一半的请求都是用户等不及之后手动刷新造成的重复请求。那一刻我才意识到：瓶颈从来不是模型算得快不快，而是你愿不愿意把“正在生成”这种状态及时暴露给用户。

流式输出的第一个价值，就是让用户看到内容像打字机一样逐字蹦出来。这样做不仅让等待变得可感知，还能带来一个隐藏优势：用户更愿意继续看下去，而不是中途关掉。尤其在RAG场景里，用户发起问题后，你还需要先做检索再送进大模型，这中间是有真实延时的。如果不做流式，用户会把这个延时全部怪到“系统卡了”上；如果你把检索状态、模型生成状态一步步推给前端，用户就会觉得“这个系统在进行多步思考”，信任度完全不一样。

1.2 从“等结果”到“看过程”：SSE真正解决了什么

流式输出现在的标配是SSE，全称Server-Sent Events。简单理解就是：服务端保持一条HTTP长连接，然后分块把数据推到客户端，而不是等所有数据都攒齐再一股脑发出去。

关键点是“分块”，而且分块的频率很高。在OpenAI官方库的接口里，只要你设置了stream=True，返回的就是一个可以迭代的流式对象。每次迭代会得到一个chunk，chunk里只有极少量增量内容。客户端可以实时接收并刷新界面。

这里我常拿吃饭举例子。非流式就像是去餐厅点餐，你在座位干等了40分钟，然后后厨一次性把你点的菜全端上来。流式则是厨师做一道菜，服务员端一道菜，虽然同样等了40分钟，但你从第5分钟就开始吃上了。

看过这个对比，你就能理解为什么现在的对话式应用几乎都在用流式。尤其是RAG场景，检索可能需要2到5秒，模型生成又需要5到20秒，把这两段耗时揉在一起不做流式，体验基本就是灾难。我在很多项目里见过“RAG一顿操作猛如虎，最后用户看到个白屏就跑了”，多半就是没做流式导致的。

2. 基础用法：stream=True和增量解析

2.1 最小可用的流式对话代码

先看一段最基础的新版SDK写法。新版指的是OpenAI官方库v1.x，接口命名和0.x时代完全是两回事。

from openai import OpenAI client = OpenAI(api_key="your-api-key", base_url="your-endpoint") response = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "user", "content": "介绍一下RAG的核心流程"} ], stream=True ) for chunk in response: delta = chunk.choices[0].delta content = getattr(delta, "content", None) if content: print(content, end="", flush=True) print()

这里面最关键的两个东西：

• stream=True：告诉服务端返回流式对象。
• 循环迭代：每轮拿到的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="", flush=True)

另外，finish_reason字段值得关注。当它不再是None而是"stop"时，说明生成结束了。如果中途遇到"length"，说明因为token限制被截断。你可以用这个标志去判断是否继续等下一轮。

2.3 版本差异：新老SDK写法迁移

如果你看过更早的教程，会发现当时流行的写法是：

import openai response = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[], stream=True )

这是老版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_key=os.environ.get("OPENAI_API_KEY")) # 模拟文档库 documents = [ "RAG通过检索外部知识库来增强模型生成能力。", "Agent需要感知环境、做规划并逐步执行工具调用。", "流式输出可以显著提升对话体验，减少用户的等待焦虑。", "向量检索的核心是文本嵌入和高维空间相似度计算。" ] # 简单获取embedding def embed_text(text: str) -> list[float]: resp = client.embeddings.create( model="text-embedding-3-small", input=text ) 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(reverse=True) 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 f'data: {{"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( model="gpt-4o-mini", messages=messages, stream=True ) for chunk in stream: delta = chunk.choices[0].delta if delta.content: content = delta.content.replace("\n", "\\n") yield f'data: {{"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_type="text/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_key="your-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( model="gpt-4o-mini", messages=messages, tools=tools, stream=True ) 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 f"text|{content_buffer}" if delta.function_call: if delta.function_call.name: function_name = delta.function_call.name yield f"function|{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事件后，先判断event：

• event: 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 一张表速查流式输出常见问题

这里面有些问题不是一次就能遇到，但如果你做过多个项目，大概率会撞上其中七八个。

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注释行或空事件，保持连接活跃。这个技巧在跨网络环境时特别管用。

这个系列进行到现在，我最大的感受是：流式输出不是把stream=True加上就完事，它更像是一套前后端配合的交互协议。刚开始你可能只追求“内容会动”，做到后面你会发现，事件结构、异常处理、重试策略、前端渲染这些部分才真正决定了一个对话产品能不能落地。按我现在的习惯，接到一个新项目时，我会先把消息协议定义清楚，再谈模型选型和RAG召回。协议定好了，后面的事情会顺很多。希望这篇能让你少踩几个我踩过的坑。