流式输出这件事我在用LangChain做对话系统时第一反应是“不就是加个stream参数嘛”但真正落到Web服务里才发现事情远没那么简单。LLM的生成过程天生是词元级别的增量行为可框架层、传输层、前端消费层每一环都有把“流”悄悄变成“全量”的可能。这篇内容我梳理了LangChain中与之相关的完整链路从底层生成器原理到上层SSE推送再到Agent场景下的事件流方案和常见的缓冲陷阱希望对正在做类似开发的你有点实际帮助。1. 流式处理的原理认知先搞懂框架在背后做了什么1.1 为什么大模型响应天生就该“流”着看大语言模型的推理过程本质上是逐步预测下一个词元模型每生成一个token都会把它加入到上下文里再继续预测下一个。这个机制决定了“完整回答”其实是由几千个“小碎片”依次产生的而不是一次性拼好的。这就带来一个非常直观的体验问题如果等服务端把整段文字全部生成完再返回用户看到的就是一个漫长且无反馈的等待过程。尤其当回答较长时几秒甚至十几秒的空白会让用户怀疑服务是不是挂了。而如果每生成一个token就立刻推送出去用户几乎能在模型“思考”的同时看到文字逐渐浮现首字延迟TTFT即从发起请求到收到第一个字的时间可以压缩到几百毫秒级别。拿下载文件来类比就很好理解老办法是等整个文件下载完才能打开流式则是边下载边播放。LLM流式输出的价值不是炫技而是把“长时间的静默等待”转化为“持续的内容反馈”这在交互质量和产品体验上是有本质差异的。同时流式还能配合用户中断操作让无效生成成本尽量降低。1.2 LangChain流式处理的三个层次LangChain对流式的支持从底往上分成三个层次理解这三个层次能帮你判断代码到底写在哪一层。最底层是模型调用层。你使用的模型服务商如果原生支持流式接口LangChain通过streamingTrue这类参数就能拿到模型侧逐词元返回的数据。这层是最硬的“真流式”因为数据确实是从模型推理过程中逐步到达的。中间层是链Chain层。基于模型层LangChain在Runnable协议里统一实现了stream和astream方法对语音、文本、JSON等不同的输出类型做了增量处理。这一层保证了你在组装Prompt、输出解析器之后依然能拿到流式结果而不只是模型层的裸token。最上层是事件层。LangChain提供astream_events机制把链在执行过程中发生的各种事件以流式方式暴露出来比如“模型开始生成新token”“某个工具调用开始了”“Agent拿到了最终结果”等。在Agent这种多步骤场景里这层才是实现“真流式”的关键。现实里很多“伪流式”方案只在最外层做了文章后台等完整结果生成完前端再用打字机效果逐字渲染。这样做确实能看到逐字出现的动画但TTFT没有半点改善用户在后台仍然要傻等完整结果。我见过不少团队把这种方案当成了流式交付上线后用户该觉得卡还是觉得卡就是这个原因。1.3 同步与异步Python生成器协议带来的两种姿势LangChain的流式依赖 Python 的生成器协议。stream返回的是一个迭代器底层通过yield逐个产出内容块astream则返回异步生成器。理解这一点很重要因为生成器是惰性求值的数据只在被迭代时才产生所以你的应用可以边取边处理不需要等整段数据完整落地。因此选择同步还是异步不只是语法偏好的问题它直接关系到你的服务能支撑什么样的并发模型。如果你的应用是简单的脚本、命令行工具或低并发的内部服务同步stream就够用。但如果你在做Web服务尤其是面对大量同时在线请求时异步astream才是正确姿势。以FastAPI为例同步生成器会阻塞事件循环一个慢请求会导致后续所有请求排队换成异步生成器线程不会被占用处理高并发时资源利用率会提升一大截。我写流式接口时默认都用异步不是装样子是实测对比过两者的吞吐差距。2. 四组核心API实操stream、astream、invoke、batch这样用才对2.1 同步流式stream() 的正确打开方式先从一个最简单的例子看。假如我们要构建一个问答链用LangChain的ChatPromptTemplate组装提示词再通过StrOutputParser解析输出为纯文本。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 这里换成你自己的模型服务地址和模型ID llm ChatOpenAI( modelyour-model-id, api_keyyour-api-key, base_urlhttp://your-model-endpoint, temperature0.7, streamingTrue ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个编程助手回答问题尽量简洁。), (human, {question}) ]) chain prompt | llm | StrOutputParser() for chunk in chain.stream({question: 请用200字解释闭包是什么}): print(chunk, end, flushTrue)注意两个关键点。第一streamingTrue必须开启。这是模型侧流式输出的开关有些模型服务即使你调用chain.stream但没开这个参数LangChain也只会拿到一次性完整输出。第二StrOutputParser很聪明它在流式模式下会感知到上游在逐块产出数据因此会逐块解析并继续向下游输出而你拿到的chunk是解析后的字符串片段。把end和flushTrue一起用是让结果在控制台里呈现出“打字机”效果的关键否则缓冲会让你感觉跟没流式一样。实际命令行验证流式输出时我很推荐用这个写法。2.2 异步流式astream() 与Web服务的正确姿势换到异步场景代码结构会变为async for循环import asyncio async def main(): async for chunk in chain.astream({question: 请用200字解释闭包是什么}): print(chunk, end, flushTrue) asyncio.run(main())如果你在FastAPI里写流式接口不要在异步函数里又嵌套asyncio.run去调同步的chain.stream。这样等于把异步调用强行变成同步阻塞事件循环照样被卡住流式接口的并发优势就全没了。正确的方式是在异步函数里直接用async for消费astream这样生成器全程跑在同一个事件循环中配合FastAPI原生的异步支持才能发挥效果。如果你确实只能在同步环境里调用LangChain但又要为异步Web框架提供数据常见的做法是使用asyncio.to_thread把同步迭代放到独立线程中执行再通过异步队列把结果传回主循环。这是同步框架和异步框架之间做桥接的实操做法但我建议你有条件还是直接用异步API代码会更清晰。2.3 什么时候别用流式invoke 与 batch 的适用场景不是所有场景都需要流式输出。invoke是同步拿到完整结果batch是批量处理一组输入两者内部其实也具备一定并行能力但它们都不以流式为目标。需要做离线批处理时比如给一批文档生成摘要、给一批问题生成答案你会希望同时丢给服务端并行处理然后统一收集结果这种场景用chain.batch([...])更合理。再比如需要把完整结果用于下一步逻辑判断时如果下一步必须拿到全部文本才能工作中间流式反而没有意义直接invoke更直接。我用一个简单对比表来梳理它API返回方式适用场景注意点invoke完整结果需要全文做后续处理等待时间较长batch一批完整结果离线批量任务一次性提交全部输入stream同步迭代器命令行/脚本/低并发Web需要开streamingTrueastream异步迭代器高并发Web服务配合异步框架使用这个选择过程可以用一句口诀概括给用户看的用流式给机器用的用完整结果。用户关注的是等待体验机器关注的是数据完整性。2.4 一个隐蔽细节模型API的流式开关和LangChain的关系很多人踩过这个坑明明调用了chain.stream控制台或者接口却一口气输出整段结果完全看不出流式的痕迹。我排查过几次发现原因通常是模型服务本身没有开启流式。LangChain的stream方法虽然按迭代器协议执行但底层跟模型服务通信时需要模型侧也以增量方式返回数据。如果模型侧被配置成“非流式”那么LangChain拿到的就是一个完整响应体自然只能一次性产出。也就是说ChatOpenAI的streamingTrue参数和调用链上的stream缺一不可。前者控制模型接口的传输方式后者控制LangChain层面的迭代方式。很多国产模型通过OpenAI兼容协议接入时默认不一定开启流式我在项目中就遇到过一个模型网关把流式参数悄悄剥离掉的情况导致前端始终等不到增量数据。碰到这种问题不要急着怀疑LangChain代码先用最原始的HTTP请求直接验证模型接口是否能流式返回几十秒就能定位到是模型侧还是框架侧的问题。3. 实战接入把LangChain流式输出完整推到Web前端3.1 传输协议选型为什么SSE比WebSocket更省事把LangChain的异步生成器接到Web前端传输环节有两条路SSEServer-Sent Events和WebSocket。这两者在LLM流式场景里都有人用但我的默认选型是SSE。SSE是建立在HTTP之上的单向推送协议浏览器原生支持EventSource接口服务端只需要按text/event-stream格式写出数据块即可。它的实现成本低天然适配文本流这种单向数据。WebSocket则是全双工协议适合双向频繁通信的场景比如在线游戏、实时协作编辑但实现复杂度高还需要处理心跳、重连、消息帧格式等细节。LLM流式输出本质上是服务端到客户端的单向数据流用WebSocket有点大材小用。我做了一个对比表格供参考能力SSEWebSocket传输方向服务端到客户端单向全双工双向浏览器原生支持支持EventSource支持WebSocket自定义请求头不支持EventSource使用GET支持断线重连原生支持需要自己实现实现复杂度低较高唯一要注意的是EventSource只能发GET请求无法自定义请求头。如果你需要POST携带大量参数或走鉴权可以选择用fetch接口配合ReadableStream消费SSE流这样既保留POST的灵活性又能用上SSE的帧格式。下面我会给出两种写法。3.2 FastAPI服务端完整实现用StreamingResponse推流在FastAPI里最直接的实现是StreamingResponse接上异步生成器。关键是要把LangChain的异步流和SSE帧格式缝合起来from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): question: str async def sse_generator(question: str): # 这里的 chain 就是上一节构建的 LCEL 链 async for chunk in chain.astream({question: question}): if chunk: # SSE 帧要求 data: 内容以换行结尾消息之间用空行分隔 cleaned chunk.replace(\n, \\n) yield fdata: {cleaned}\n\n app.post(/chat) async def chat(req: ChatRequest): return StreamingResponse( sse_generator(req.question), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, } )这里有几个细节值得展开说。第一SSE的data字段单行内不能包含裸换行。LangChain流式返回的内容可能会包含换行符如果直接塞进data:后面SSE解析器会把多行内容拆成多条消息前端收到的就是碎片化的错乱文本。所以我在写入前把\n替换成字面量\\n前端拿到后再做反转义这个处理在含代码块和列表的回答中尤为重要。第二Cache-Control: no-cache是告诉各级缓存不要缓存流式响应否则下游CDN或浏览器缓存会把累积的内容一次性回放流式就失效了。X-Accel-Buffering: no是给nginx看的指示头用于关闭nginx侧的缓冲。3.3 前端消费fetch ReadableStream 解析SSE帧前端消费流式响应最常见的方案是用fetch接口读取响应体。原理是response.body是一个ReadableStream可以持续读取二进制块再交给TextDecoder解码为文本最后按SSE帧格式切片解析。async function fetchSSE(question) { const response await fetch(/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question }) }); if (!response.ok || !response.body) { throw new Error(请求失败); } 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 表示一条消息结束 let sepIndex; while ((sepIndex buffer.indexOf(\n\n)) ! -1) { const frame buffer.slice(0, sepIndex); buffer buffer.slice(sepIndex 2); for (const line of frame.split(\n)) { if (line.startsWith(data: )) { const data line.slice(6).replace(/\\n/g, \n); // 这里把 data 交给界面渲染 renderChunk(data); } } } } }这里要特别解释一下“缓冲”的必要性。网络传输中一个SSE帧可能被拆成多个TCP包到达也就是说reader.read()拿到的块并不恰好等于一条完整消息。如果直接按块解析你会遇到“消息被切成两半”的边界问题。上面代码用buffer把未处理完的数据暂存起来每次只取到完整分隔符\n\n为止剩下的留到下一轮拼接这样才能保证帧内容完整。这也是我在实践中被乱码教育过之后总结出来的经验。如果你能接受GET请求且不需要携带自定义鉴权头也可以直接用EventSource它会自动处理重连和帧解析代码会短很多const es new EventSource(/chat?question什么是闭包); es.onmessage (event) { renderChunk(event.data); }; es.onerror () es.close();两种方案我都用过。内部系统或原型验证我会倾向EventSource生产环境要带鉴权、要POST大参数则用fetch ReadableStream更稳妥。3.4 服务端容易被忽略的问题反代缓冲、压缩与超时流式接口开发完在测试环境一切正常一上生产就变成一次性输出这种情况我遇到过不止一次。问题十有八九出在反向代理层的缓冲。nginx是默认开缓冲的它会等上游应用把整个响应全部写完再一次性发给客户端。这样即使FastAPI和LangChain那边是真正的流式到用户浏览器那边也全被积压成了整段内容。解决方式是在nginx的location配置中关闭缓冲location /chat { proxy_pass http://backend_service; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; }注意proxy_read_timeout 300s也很重要。流式接口的特点是连接时间很长但每个小块之间的间隔可能比较短。如果超时时间配置过短模型思考时间稍长代理层就可能主动断开连接前端会莫名其妙地收到一个中断错误。这个参数跑长文本生成任务时尤其需要调大。另一个坑是GZip压缩。普通接口启用GZip确实能大幅减少传输体积但流式场景下压缩会引入不必要的CPU开销并且为了让压缩效率更高中间层往往会缓冲一段数据后再压缩这一缓冲就把流式特性给抵消了。所以对SSE路径建议关闭动态压缩只在静态资源上启用。我记得有一次排查线上问题前后端都确认是流式就是中间加了层网关开启压缩结果所有响应都攒到一起才吐出来排查了很久才定位到。4. Agent与工具调用的“真流式”astream_events事件方案4.1 为什么Agent场景流式这么难普通问答链的流式很简单一条链跑到底中间没有分支模型输出即最终输出。但是Agent场景就复杂得多Agent需要决定调用哪个工具工具返回结果后还要再把结果交给模型继续推理可能循环好几轮才结束。这里有个体验痛点如果Agent在调用工具时前端一直空白用户会以为服务出了故障。部分实现甚至要等整个Agent跑完整轮才返回文本那种等待对用户来说几乎不可接受。实现Agent流式的难点在于LangChain的agent_executor.stream()在早期版本中粒度是“链节点级”的不是“token级”的也就是说你可能拿到的是一整段推理结果而不是逐步生成的文字。解决思路是转换事件粒度。LangChain从0.1版本开始提供的astream_events机制可以从Agent执行的内部把更细粒度的事件捞出来比如模型生成token的事件、工具调用的事件、最终结果的事件。基于这个机制我们才可能在Agent链路里实现真正的token级推送。4.2 astream_events事件机制核心概念与事件名速查astream_events是LangChain在Runnable上提供的异步事件流接口它把链执行过程中产生的事件依次暴露出来。典型的事件对象长这样{ event: on_chat_model_stream, run_id: xxxx, name: ChatOpenAI, data: {chunk: ...} }event字段标识事件类型data.chunk携带当前产生的数据块。不同事件类型在不同阶段触发常用事件名如下事件名触发时机携带内容on_chat_model_start模型开始调用输入messageon_chat_model_stream模型流式产出每个token当前chunkon_chat_model_end模型调用结束完整输出on_tool_start工具开始执行工具输入on_tool_end工具执行结束工具输出on_chain_end某个链节点结束该节点输出从LangChain 0.2.x开始事件名推荐使用v2命名空间也就是上面这种on_chat_model_stream形态。更老的版本里可能看到on_llm_new_token如果你项目中历史代码用了这种事件名需要确认版本兼容性。我建议新项目一律在astream_events(..., versionv2)里显式声明。苹果和桔子的区别在于on_chat_model_stream是按token粒度产生的而on_llm_new_token是旧版为了兼容记忆体里的一种命名。实际开发中我写了一个过滤函数只处理这两类事件其他事件都放行或者打日志这样既能准确拿到增量token又不至于被无关事件干扰。4.3 Agent场景流式实现把事件拼装成前端可见的内容下面是一个基于create_react_agent的完整示例展示如何把Agent执行过程的各个关键事件实时推给前端from langchain.agents import create_react_agent, AgentExecutor from langchain_core.tools import tool from langchain_core.prompts import ChatPromptTemplate tool def add(a: int, b: int) - int: 计算两个数字的和。 return a b tools [add] prompt ChatPromptTemplate.from_messages([ (system, 你是一个能调用工具的助手请基于工具结果回答问题。), (human, {input}), (placeholder, {agent_scratchpad}), ]) agent create_react_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, handle_parsing_errorsTrue) async def agent_sse_generator(user_input: str): async for event in executor.astream_events( {input: user_input}, versionv2 ): event_name event[event] # 模型每个新token这就是最关键的增量内容 if event_name on_chat_model_stream: chunk event[data][chunk] content chunk.content if isinstance(chunk.content, str) else str(chunk.content) if content: yield fdata: {content}\n\n # 工具开始执行推送状态信息 elif event_name on_tool_start: tool_input event[data].get(input) yield fdata: [正在调用工具: {tool_input}]\n\n # 工具执行结束推送结果摘要 elif event_name on_tool_end: output event[data].get(output) yield fdata: [工具返回: {output}]\n\n # Agent最终完成可以推送结束标记 elif event_name on_chain_end: if event.get(name) AgentExecutor: yield fdata: [DONE]\n\n这段代码的核心在于事件过滤on_chat_model_stream负责把模型生成的文本增量实时推送on_tool_start和on_tool_end则把Agent的“思考过程”状态透出给用户让前端显示“正在调用工具……”这类反馈。需要注意一个Agent循环里模型可能被调用多轮推理、读工具结果后再次推理on_chat_model_stream会把这些轮的token全部产生出来。如果你不想把中间推理内容透出给用户那就需要追踪run_id只对最终那一轮的token做推送。这个细节在复杂Agent里很关键否则用户会看到模型在“自言自语”体验很怪异。4.4 事件流方案的其他注意点去重与推理链用astream_events时还会遇到一个重复推送问题。如果Agent内部同时对同一个模型调用了astream和事件流监听同一个token可能会被on_chat_model_stream捕获两次。解决办法是在消费层保存最近处理过的run_id对同一个run_id只推送一次或者直接在事件过滤器中排除掉来自内部监听渠道的事件。另外LLM在Agent中的推理链要不要暴露给用户这是一个产品决策不是纯技术问题。一些场景下展示“Agent正在思考”反而能提升用户对复杂任务的信任度但也有场景下用户只关心最终答案中间工具调用过程完全没必要展示。我的经验是要么完全屏蔽中间过程要么只展示“正在调用工具”这类轻量状态不要展示模型内部的完整思考文本那会喧宾夺主。5. 高频问题排查与调优实录5.1 前端永远等不到第一个字从链路逐层排查前端一直空白、转圈半天才一次性出结果的问题按这个顺序排查速度最快。第一层确认模型侧是否真流式。直接写一个最小脚本调用模型接口打印流式返回的每个chunk。如果模型侧都不流式那就不用查后面的链路了。有些模型服务需要通过特定参数才能开启增量有些网关会吞掉流式标记这层花不了两分钟。第二层确认LangChain链层流式。写好chain.stream的最小验证在终端打印。如果终端能逐字输出说明框架侧没问题。第三层确认Web服务层输出。用curl -N直接请求你的流式接口观察数据是否是逐步到达的。如果curl能看到逐块输出那问题大概率在中间代理或前端解析层如果curl也只能一次性拿到全部则问题在FastAPI或代理缓冲设置上。这个逐层排查法几乎能定位所有流式失效问题。我每次接到这类问题都是按这个顺序排查基本不会跑偏。5.2 流式Token流速太慢缓冲批量推送策略流式虽然好但也有副作用高频的小块推送会带来更多的I/O和网络开销。如果模型的推理速度很快但每个token都要立刻通过网络帧推出去吞吐反而可能被传输限制住。在FastAPI实践中我常用一个“累积缓冲”技巧不每token推送而是攒到一定字数或固定时间间隔后合并推送。比如每收到200个字符或每200毫秒就把这段时间累积的内容作为一条SSE帧推出去。这样前端仍然感觉到平滑输出传输和渲染开销却能大幅下降。import asyncio async def buffered_sse_generator(question: str): buffer [] last_flush asyncio.get_event_loop().time() async for chunk in chain.astream({question: question}): if chunk: buffer.append(chunk) now asyncio.get_event_loop().time() if len(buffer) 20 or now - last_flush 0.2: yield fdata: {.join(buffer)}\n\n buffer.clear() last_flush now if buffer: yield fdata: {.join(buffer)}\n\n这种做法本质上是在“延迟”和“吞吐”之间找一个平衡点。缓冲太小I/O开销高缓冲太大体验又退回一次性输出。我实测下来200毫秒间隔或20个chunk的组合在大多数场景效果不错。如果你做的是低延迟要求极高的交互可以相应缩小阈值但不要降到每个token都推一次除非你的并发量真的很低。5.3 客户端断开连接后的处理流式接口最大的特征就是长连接客户端随时可能在中途关闭页面或断网。如果你不及时处理服务端可能会继续生成无效token既浪费计算资源又可能把数据写到已经断开的通道上触发异常。FastAPI的StreamingResponse在客户端断开时会尝试取消正在运行的异步生成器。你需要在自己的生成器里捕获asyncio.CancelledError及时清理中间资源比如释放锁、关闭会话、终止子任务。有一种印象很深的场景我在一个生产项目中遇到过连接断开后模型调用仍然继续跑了一段时间排查发现是生成器里的任务被另外的线程池持有CancelledError没能及时传递进去。后来改动是统一启动独立Task管理模型调用并将取消信号显式传给对应的Task问题才解决。async def safe_generator(question: str): try: async for chunk in chain.astream({question: question}): yield fdata: {chunk}\n\n except asyncio.CancelledError: # 客户端断开做清理工作 print(连接断开清理资源) raise5.4 常见问题速查表现象可能原因解决方案前端一次性收到全部内容模型/网关未开流式、代理缓冲开启、接口未用StreamingResponse从模型侧到代理侧逐层验证关闭缓冲流式过程中有长暂停模型TTFT高、Agent工具执行阶段无token输出推送“正在思考”状态或对工具调用阶段做状态提示SSE帧内容乱码、多行错乱data字段含裸换行未转义写入前将\n替换为字面量\\n连接频繁中断代理read超时时间过短调大proxy_read_timeout建议300秒以上生产环境偶发一次性输出网关GZip压缩对动态流启用对SSE路径关闭压缩静态资源才启用GZipAgent场景只输出最终内容只在AgentExecutor层stream改用astream_events监听on_chat_model_stream5.5 一个一直沿用的验证技巧最后分享一个我一直在用的小技巧。每次写完流式接口不用急着写前端先用curl验证curl -N -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d {question: 请用100字介绍LangChain}-N参数告诉curl不要缓冲输出而是立即显示收到的数据。如果curl能看到一字一字地出现说明服务端流式是通的后面前端怎么接都不会有大问题。如果curl也是一次性输出继续往前排查大概率是模型服务或代理缓冲那一层的问题。这个习惯帮我省了很多联调时间。流式这种东西测前端之前一定要先确认“源头是活的”否则前端写得再对接口不流式也白搭。如果你跟我一样踩过那种“前端调了半天发现是后端问题”的坑就知道先把源头验证清楚有多重要。 
长河君 · 2026/10/23 5:39:40 · 阅读 1.2k LangChain流式输出实战：从API原理到SSE推送与Agent事件流
流式输出这件事我在用LangChain做对话系统时第一反应是“不就是加个stream参数嘛”但真正落到Web服务里才发现事情远没那么简单。LLM的生成过程天生是词元级别的增量行为可框架层、传输层、前端消费层每一环都有把“流”悄悄变成“全量”的可能。这篇内容我梳理了LangChain中与之相关的完整链路从底层生成器原理到上层SSE推送再到Agent场景下的事件流方案和常见的缓冲陷阱希望对正在做类似开发的你有点实际帮助。1. 流式处理的原理认知先搞懂框架在背后做了什么1.1 为什么大模型响应天生就该“流”着看大语言模型的推理过程本质上是逐步预测下一个词元模型每生成一个token都会把它加入到上下文里再继续预测下一个。这个机制决定了“完整回答”其实是由几千个“小碎片”依次产生的而不是一次性拼好的。这就带来一个非常直观的体验问题如果等服务端把整段文字全部生成完再返回用户看到的就是一个漫长且无反馈的等待过程。尤其当回答较长时几秒甚至十几秒的空白会让用户怀疑服务是不是挂了。而如果每生成一个token就立刻推送出去用户几乎能在模型“思考”的同时看到文字逐渐浮现首字延迟TTFT即从发起请求到收到第一个字的时间可以压缩到几百毫秒级别。拿下载文件来类比就很好理解老办法是等整个文件下载完才能打开流式则是边下载边播放。LLM流式输出的价值不是炫技而是把“长时间的静默等待”转化为“持续的内容反馈”这在交互质量和产品体验上是有本质差异的。同时流式还能配合用户中断操作让无效生成成本尽量降低。1.2 LangChain流式处理的三个层次LangChain对流式的支持从底往上分成三个层次理解这三个层次能帮你判断代码到底写在哪一层。最底层是模型调用层。你使用的模型服务商如果原生支持流式接口LangChain通过streamingTrue这类参数就能拿到模型侧逐词元返回的数据。这层是最硬的“真流式”因为数据确实是从模型推理过程中逐步到达的。中间层是链Chain层。基于模型层LangChain在Runnable协议里统一实现了stream和astream方法对语音、文本、JSON等不同的输出类型做了增量处理。这一层保证了你在组装Prompt、输出解析器之后依然能拿到流式结果而不只是模型层的裸token。最上层是事件层。LangChain提供astream_events机制把链在执行过程中发生的各种事件以流式方式暴露出来比如“模型开始生成新token”“某个工具调用开始了”“Agent拿到了最终结果”等。在Agent这种多步骤场景里这层才是实现“真流式”的关键。现实里很多“伪流式”方案只在最外层做了文章后台等完整结果生成完前端再用打字机效果逐字渲染。这样做确实能看到逐字出现的动画但TTFT没有半点改善用户在后台仍然要傻等完整结果。我见过不少团队把这种方案当成了流式交付上线后用户该觉得卡还是觉得卡就是这个原因。1.3 同步与异步Python生成器协议带来的两种姿势LangChain的流式依赖 Python 的生成器协议。stream返回的是一个迭代器底层通过yield逐个产出内容块astream则返回异步生成器。理解这一点很重要因为生成器是惰性求值的数据只在被迭代时才产生所以你的应用可以边取边处理不需要等整段数据完整落地。因此选择同步还是异步不只是语法偏好的问题它直接关系到你的服务能支撑什么样的并发模型。如果你的应用是简单的脚本、命令行工具或低并发的内部服务同步stream就够用。但如果你在做Web服务尤其是面对大量同时在线请求时异步astream才是正确姿势。以FastAPI为例同步生成器会阻塞事件循环一个慢请求会导致后续所有请求排队换成异步生成器线程不会被占用处理高并发时资源利用率会提升一大截。我写流式接口时默认都用异步不是装样子是实测对比过两者的吞吐差距。2. 四组核心API实操stream、astream、invoke、batch这样用才对2.1 同步流式stream() 的正确打开方式先从一个最简单的例子看。假如我们要构建一个问答链用LangChain的ChatPromptTemplate组装提示词再通过StrOutputParser解析输出为纯文本。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 这里换成你自己的模型服务地址和模型ID llm ChatOpenAI( modelyour-model-id, api_keyyour-api-key, base_urlhttp://your-model-endpoint, temperature0.7, streamingTrue ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个编程助手回答问题尽量简洁。), (human, {question}) ]) chain prompt | llm | StrOutputParser() for chunk in chain.stream({question: 请用200字解释闭包是什么}): print(chunk, end, flushTrue)注意两个关键点。第一streamingTrue必须开启。这是模型侧流式输出的开关有些模型服务即使你调用chain.stream但没开这个参数LangChain也只会拿到一次性完整输出。第二StrOutputParser很聪明它在流式模式下会感知到上游在逐块产出数据因此会逐块解析并继续向下游输出而你拿到的chunk是解析后的字符串片段。把end和flushTrue一起用是让结果在控制台里呈现出“打字机”效果的关键否则缓冲会让你感觉跟没流式一样。实际命令行验证流式输出时我很推荐用这个写法。2.2 异步流式astream() 与Web服务的正确姿势换到异步场景代码结构会变为async for循环import asyncio async def main(): async for chunk in chain.astream({question: 请用200字解释闭包是什么}): print(chunk, end, flushTrue) asyncio.run(main())如果你在FastAPI里写流式接口不要在异步函数里又嵌套asyncio.run去调同步的chain.stream。这样等于把异步调用强行变成同步阻塞事件循环照样被卡住流式接口的并发优势就全没了。正确的方式是在异步函数里直接用async for消费astream这样生成器全程跑在同一个事件循环中配合FastAPI原生的异步支持才能发挥效果。如果你确实只能在同步环境里调用LangChain但又要为异步Web框架提供数据常见的做法是使用asyncio.to_thread把同步迭代放到独立线程中执行再通过异步队列把结果传回主循环。这是同步框架和异步框架之间做桥接的实操做法但我建议你有条件还是直接用异步API代码会更清晰。2.3 什么时候别用流式invoke 与 batch 的适用场景不是所有场景都需要流式输出。invoke是同步拿到完整结果batch是批量处理一组输入两者内部其实也具备一定并行能力但它们都不以流式为目标。需要做离线批处理时比如给一批文档生成摘要、给一批问题生成答案你会希望同时丢给服务端并行处理然后统一收集结果这种场景用chain.batch([...])更合理。再比如需要把完整结果用于下一步逻辑判断时如果下一步必须拿到全部文本才能工作中间流式反而没有意义直接invoke更直接。我用一个简单对比表来梳理它API返回方式适用场景注意点invoke完整结果需要全文做后续处理等待时间较长batch一批完整结果离线批量任务一次性提交全部输入stream同步迭代器命令行/脚本/低并发Web需要开streamingTrueastream异步迭代器高并发Web服务配合异步框架使用这个选择过程可以用一句口诀概括给用户看的用流式给机器用的用完整结果。用户关注的是等待体验机器关注的是数据完整性。2.4 一个隐蔽细节模型API的流式开关和LangChain的关系很多人踩过这个坑明明调用了chain.stream控制台或者接口却一口气输出整段结果完全看不出流式的痕迹。我排查过几次发现原因通常是模型服务本身没有开启流式。LangChain的stream方法虽然按迭代器协议执行但底层跟模型服务通信时需要模型侧也以增量方式返回数据。如果模型侧被配置成“非流式”那么LangChain拿到的就是一个完整响应体自然只能一次性产出。也就是说ChatOpenAI的streamingTrue参数和调用链上的stream缺一不可。前者控制模型接口的传输方式后者控制LangChain层面的迭代方式。很多国产模型通过OpenAI兼容协议接入时默认不一定开启流式我在项目中就遇到过一个模型网关把流式参数悄悄剥离掉的情况导致前端始终等不到增量数据。碰到这种问题不要急着怀疑LangChain代码先用最原始的HTTP请求直接验证模型接口是否能流式返回几十秒就能定位到是模型侧还是框架侧的问题。3. 实战接入把LangChain流式输出完整推到Web前端3.1 传输协议选型为什么SSE比WebSocket更省事把LangChain的异步生成器接到Web前端传输环节有两条路SSEServer-Sent Events和WebSocket。这两者在LLM流式场景里都有人用但我的默认选型是SSE。SSE是建立在HTTP之上的单向推送协议浏览器原生支持EventSource接口服务端只需要按text/event-stream格式写出数据块即可。它的实现成本低天然适配文本流这种单向数据。WebSocket则是全双工协议适合双向频繁通信的场景比如在线游戏、实时协作编辑但实现复杂度高还需要处理心跳、重连、消息帧格式等细节。LLM流式输出本质上是服务端到客户端的单向数据流用WebSocket有点大材小用。我做了一个对比表格供参考能力SSEWebSocket传输方向服务端到客户端单向全双工双向浏览器原生支持支持EventSource支持WebSocket自定义请求头不支持EventSource使用GET支持断线重连原生支持需要自己实现实现复杂度低较高唯一要注意的是EventSource只能发GET请求无法自定义请求头。如果你需要POST携带大量参数或走鉴权可以选择用fetch接口配合ReadableStream消费SSE流这样既保留POST的灵活性又能用上SSE的帧格式。下面我会给出两种写法。3.2 FastAPI服务端完整实现用StreamingResponse推流在FastAPI里最直接的实现是StreamingResponse接上异步生成器。关键是要把LangChain的异步流和SSE帧格式缝合起来from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): question: str async def sse_generator(question: str): # 这里的 chain 就是上一节构建的 LCEL 链 async for chunk in chain.astream({question: question}): if chunk: # SSE 帧要求 data: 内容以换行结尾消息之间用空行分隔 cleaned chunk.replace(\n, \\n) yield fdata: {cleaned}\n\n app.post(/chat) async def chat(req: ChatRequest): return StreamingResponse( sse_generator(req.question), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, } )这里有几个细节值得展开说。第一SSE的data字段单行内不能包含裸换行。LangChain流式返回的内容可能会包含换行符如果直接塞进data:后面SSE解析器会把多行内容拆成多条消息前端收到的就是碎片化的错乱文本。所以我在写入前把\n替换成字面量\\n前端拿到后再做反转义这个处理在含代码块和列表的回答中尤为重要。第二Cache-Control: no-cache是告诉各级缓存不要缓存流式响应否则下游CDN或浏览器缓存会把累积的内容一次性回放流式就失效了。X-Accel-Buffering: no是给nginx看的指示头用于关闭nginx侧的缓冲。3.3 前端消费fetch ReadableStream 解析SSE帧前端消费流式响应最常见的方案是用fetch接口读取响应体。原理是response.body是一个ReadableStream可以持续读取二进制块再交给TextDecoder解码为文本最后按SSE帧格式切片解析。async function fetchSSE(question) { const response await fetch(/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question }) }); if (!response.ok || !response.body) { throw new Error(请求失败); } 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 表示一条消息结束 let sepIndex; while ((sepIndex buffer.indexOf(\n\n)) ! -1) { const frame buffer.slice(0, sepIndex); buffer buffer.slice(sepIndex 2); for (const line of frame.split(\n)) { if (line.startsWith(data: )) { const data line.slice(6).replace(/\\n/g, \n); // 这里把 data 交给界面渲染 renderChunk(data); } } } } }这里要特别解释一下“缓冲”的必要性。网络传输中一个SSE帧可能被拆成多个TCP包到达也就是说reader.read()拿到的块并不恰好等于一条完整消息。如果直接按块解析你会遇到“消息被切成两半”的边界问题。上面代码用buffer把未处理完的数据暂存起来每次只取到完整分隔符\n\n为止剩下的留到下一轮拼接这样才能保证帧内容完整。这也是我在实践中被乱码教育过之后总结出来的经验。如果你能接受GET请求且不需要携带自定义鉴权头也可以直接用EventSource它会自动处理重连和帧解析代码会短很多const es new EventSource(/chat?question什么是闭包); es.onmessage (event) { renderChunk(event.data); }; es.onerror () es.close();两种方案我都用过。内部系统或原型验证我会倾向EventSource生产环境要带鉴权、要POST大参数则用fetch ReadableStream更稳妥。3.4 服务端容易被忽略的问题反代缓冲、压缩与超时流式接口开发完在测试环境一切正常一上生产就变成一次性输出这种情况我遇到过不止一次。问题十有八九出在反向代理层的缓冲。nginx是默认开缓冲的它会等上游应用把整个响应全部写完再一次性发给客户端。这样即使FastAPI和LangChain那边是真正的流式到用户浏览器那边也全被积压成了整段内容。解决方式是在nginx的location配置中关闭缓冲location /chat { proxy_pass http://backend_service; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; }注意proxy_read_timeout 300s也很重要。流式接口的特点是连接时间很长但每个小块之间的间隔可能比较短。如果超时时间配置过短模型思考时间稍长代理层就可能主动断开连接前端会莫名其妙地收到一个中断错误。这个参数跑长文本生成任务时尤其需要调大。另一个坑是GZip压缩。普通接口启用GZip确实能大幅减少传输体积但流式场景下压缩会引入不必要的CPU开销并且为了让压缩效率更高中间层往往会缓冲一段数据后再压缩这一缓冲就把流式特性给抵消了。所以对SSE路径建议关闭动态压缩只在静态资源上启用。我记得有一次排查线上问题前后端都确认是流式就是中间加了层网关开启压缩结果所有响应都攒到一起才吐出来排查了很久才定位到。4. Agent与工具调用的“真流式”astream_events事件方案4.1 为什么Agent场景流式这么难普通问答链的流式很简单一条链跑到底中间没有分支模型输出即最终输出。但是Agent场景就复杂得多Agent需要决定调用哪个工具工具返回结果后还要再把结果交给模型继续推理可能循环好几轮才结束。这里有个体验痛点如果Agent在调用工具时前端一直空白用户会以为服务出了故障。部分实现甚至要等整个Agent跑完整轮才返回文本那种等待对用户来说几乎不可接受。实现Agent流式的难点在于LangChain的agent_executor.stream()在早期版本中粒度是“链节点级”的不是“token级”的也就是说你可能拿到的是一整段推理结果而不是逐步生成的文字。解决思路是转换事件粒度。LangChain从0.1版本开始提供的astream_events机制可以从Agent执行的内部把更细粒度的事件捞出来比如模型生成token的事件、工具调用的事件、最终结果的事件。基于这个机制我们才可能在Agent链路里实现真正的token级推送。4.2 astream_events事件机制核心概念与事件名速查astream_events是LangChain在Runnable上提供的异步事件流接口它把链执行过程中产生的事件依次暴露出来。典型的事件对象长这样{ event: on_chat_model_stream, run_id: xxxx, name: ChatOpenAI, data: {chunk: ...} }event字段标识事件类型data.chunk携带当前产生的数据块。不同事件类型在不同阶段触发常用事件名如下事件名触发时机携带内容on_chat_model_start模型开始调用输入messageon_chat_model_stream模型流式产出每个token当前chunkon_chat_model_end模型调用结束完整输出on_tool_start工具开始执行工具输入on_tool_end工具执行结束工具输出on_chain_end某个链节点结束该节点输出从LangChain 0.2.x开始事件名推荐使用v2命名空间也就是上面这种on_chat_model_stream形态。更老的版本里可能看到on_llm_new_token如果你项目中历史代码用了这种事件名需要确认版本兼容性。我建议新项目一律在astream_events(..., versionv2)里显式声明。苹果和桔子的区别在于on_chat_model_stream是按token粒度产生的而on_llm_new_token是旧版为了兼容记忆体里的一种命名。实际开发中我写了一个过滤函数只处理这两类事件其他事件都放行或者打日志这样既能准确拿到增量token又不至于被无关事件干扰。4.3 Agent场景流式实现把事件拼装成前端可见的内容下面是一个基于create_react_agent的完整示例展示如何把Agent执行过程的各个关键事件实时推给前端from langchain.agents import create_react_agent, AgentExecutor from langchain_core.tools import tool from langchain_core.prompts import ChatPromptTemplate tool def add(a: int, b: int) - int: 计算两个数字的和。 return a b tools [add] prompt ChatPromptTemplate.from_messages([ (system, 你是一个能调用工具的助手请基于工具结果回答问题。), (human, {input}), (placeholder, {agent_scratchpad}), ]) agent create_react_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, handle_parsing_errorsTrue) async def agent_sse_generator(user_input: str): async for event in executor.astream_events( {input: user_input}, versionv2 ): event_name event[event] # 模型每个新token这就是最关键的增量内容 if event_name on_chat_model_stream: chunk event[data][chunk] content chunk.content if isinstance(chunk.content, str) else str(chunk.content) if content: yield fdata: {content}\n\n # 工具开始执行推送状态信息 elif event_name on_tool_start: tool_input event[data].get(input) yield fdata: [正在调用工具: {tool_input}]\n\n # 工具执行结束推送结果摘要 elif event_name on_tool_end: output event[data].get(output) yield fdata: [工具返回: {output}]\n\n # Agent最终完成可以推送结束标记 elif event_name on_chain_end: if event.get(name) AgentExecutor: yield fdata: [DONE]\n\n这段代码的核心在于事件过滤on_chat_model_stream负责把模型生成的文本增量实时推送on_tool_start和on_tool_end则把Agent的“思考过程”状态透出给用户让前端显示“正在调用工具……”这类反馈。需要注意一个Agent循环里模型可能被调用多轮推理、读工具结果后再次推理on_chat_model_stream会把这些轮的token全部产生出来。如果你不想把中间推理内容透出给用户那就需要追踪run_id只对最终那一轮的token做推送。这个细节在复杂Agent里很关键否则用户会看到模型在“自言自语”体验很怪异。4.4 事件流方案的其他注意点去重与推理链用astream_events时还会遇到一个重复推送问题。如果Agent内部同时对同一个模型调用了astream和事件流监听同一个token可能会被on_chat_model_stream捕获两次。解决办法是在消费层保存最近处理过的run_id对同一个run_id只推送一次或者直接在事件过滤器中排除掉来自内部监听渠道的事件。另外LLM在Agent中的推理链要不要暴露给用户这是一个产品决策不是纯技术问题。一些场景下展示“Agent正在思考”反而能提升用户对复杂任务的信任度但也有场景下用户只关心最终答案中间工具调用过程完全没必要展示。我的经验是要么完全屏蔽中间过程要么只展示“正在调用工具”这类轻量状态不要展示模型内部的完整思考文本那会喧宾夺主。5. 高频问题排查与调优实录5.1 前端永远等不到第一个字从链路逐层排查前端一直空白、转圈半天才一次性出结果的问题按这个顺序排查速度最快。第一层确认模型侧是否真流式。直接写一个最小脚本调用模型接口打印流式返回的每个chunk。如果模型侧都不流式那就不用查后面的链路了。有些模型服务需要通过特定参数才能开启增量有些网关会吞掉流式标记这层花不了两分钟。第二层确认LangChain链层流式。写好chain.stream的最小验证在终端打印。如果终端能逐字输出说明框架侧没问题。第三层确认Web服务层输出。用curl -N直接请求你的流式接口观察数据是否是逐步到达的。如果curl能看到逐块输出那问题大概率在中间代理或前端解析层如果curl也只能一次性拿到全部则问题在FastAPI或代理缓冲设置上。这个逐层排查法几乎能定位所有流式失效问题。我每次接到这类问题都是按这个顺序排查基本不会跑偏。5.2 流式Token流速太慢缓冲批量推送策略流式虽然好但也有副作用高频的小块推送会带来更多的I/O和网络开销。如果模型的推理速度很快但每个token都要立刻通过网络帧推出去吞吐反而可能被传输限制住。在FastAPI实践中我常用一个“累积缓冲”技巧不每token推送而是攒到一定字数或固定时间间隔后合并推送。比如每收到200个字符或每200毫秒就把这段时间累积的内容作为一条SSE帧推出去。这样前端仍然感觉到平滑输出传输和渲染开销却能大幅下降。import asyncio async def buffered_sse_generator(question: str): buffer [] last_flush asyncio.get_event_loop().time() async for chunk in chain.astream({question: question}): if chunk: buffer.append(chunk) now asyncio.get_event_loop().time() if len(buffer) 20 or now - last_flush 0.2: yield fdata: {.join(buffer)}\n\n buffer.clear() last_flush now if buffer: yield fdata: {.join(buffer)}\n\n这种做法本质上是在“延迟”和“吞吐”之间找一个平衡点。缓冲太小I/O开销高缓冲太大体验又退回一次性输出。我实测下来200毫秒间隔或20个chunk的组合在大多数场景效果不错。如果你做的是低延迟要求极高的交互可以相应缩小阈值但不要降到每个token都推一次除非你的并发量真的很低。5.3 客户端断开连接后的处理流式接口最大的特征就是长连接客户端随时可能在中途关闭页面或断网。如果你不及时处理服务端可能会继续生成无效token既浪费计算资源又可能把数据写到已经断开的通道上触发异常。FastAPI的StreamingResponse在客户端断开时会尝试取消正在运行的异步生成器。你需要在自己的生成器里捕获asyncio.CancelledError及时清理中间资源比如释放锁、关闭会话、终止子任务。有一种印象很深的场景我在一个生产项目中遇到过连接断开后模型调用仍然继续跑了一段时间排查发现是生成器里的任务被另外的线程池持有CancelledError没能及时传递进去。后来改动是统一启动独立Task管理模型调用并将取消信号显式传给对应的Task问题才解决。async def safe_generator(question: str): try: async for chunk in chain.astream({question: question}): yield fdata: {chunk}\n\n except asyncio.CancelledError: # 客户端断开做清理工作 print(连接断开清理资源) raise5.4 常见问题速查表现象可能原因解决方案前端一次性收到全部内容模型/网关未开流式、代理缓冲开启、接口未用StreamingResponse从模型侧到代理侧逐层验证关闭缓冲流式过程中有长暂停模型TTFT高、Agent工具执行阶段无token输出推送“正在思考”状态或对工具调用阶段做状态提示SSE帧内容乱码、多行错乱data字段含裸换行未转义写入前将\n替换为字面量\\n连接频繁中断代理read超时时间过短调大proxy_read_timeout建议300秒以上生产环境偶发一次性输出网关GZip压缩对动态流启用对SSE路径关闭压缩静态资源才启用GZipAgent场景只输出最终内容只在AgentExecutor层stream改用astream_events监听on_chat_model_stream5.5 一个一直沿用的验证技巧最后分享一个我一直在用的小技巧。每次写完流式接口不用急着写前端先用curl验证curl -N -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d {question: 请用100字介绍LangChain}-N参数告诉curl不要缓冲输出而是立即显示收到的数据。如果curl能看到一字一字地出现说明服务端流式是通的后面前端怎么接都不会有大问题。如果curl也是一次性输出继续往前排查大概率是模型服务或代理缓冲那一层的问题。这个习惯帮我省了很多联调时间。流式这种东西测前端之前一定要先确认“源头是活的”否则前端写得再对接口不流式也白搭。如果你跟我一样踩过那种“前端调了半天发现是后端问题”的坑就知道先把源头验证清楚有多重要。