企业官网建设流程全解析

最早给对话机器人接前端时，我图省事，后端用一次普通调用等大模型把整段回答生成完，再把完整文本包进 JSON 返回给页面。结果用户那边永远是一个转圈的 loading，短则几秒，长则十几秒，脾气急一点的直接关页面，留我在监控面板前干瞪眼。后来我把这块改成流式输出，Token 一个接一个地从接口里蹦出来，首屏几乎瞬间就有反应，用户体验立刻不一样了。

这篇文章要聊的就是 LangChain 里的流式处理能力。我会先把"流式到底在流什么"讲透，然后从最简单的stream()入手，逐步讲到回调机制、事件流，再到 Chain、Agent、RAG 这些复杂链路的打通方案，最后把我在真实项目里反复踩过的流式失效坑全部倒出来。无论你是刚接触 LangChain 的新手，还是已经在生产环境里被流式折腾过的老手，应该都能找到对应的解法。

1. 流式到底在流什么：Token、消息块与事件三层概念

1.1 大模型为什么天生就适合流式输出

先说个很多人容易忽略的事实：大模型本身就是一个 Token 一个 Token 往外蹦的。输入上下文之后，模型先预测第一个 Token，再带着第一个 Token 预测第二个，以此类推，直到遇到结束符。也就是说，完整的回答本来就不是一次性生成的，而是逐步生成的。

那为什么用普通接口调用时，我们感受到的是"等了半天，一次性全出来"？因为服务端在返回 HTTP 响应之前，把生成的整个流程都攒在内存里，等全部 Token 生成完才一起发给客户端。用户体验上就成了"要么没反应，要么全出来"。

流式处理要做的，其实就是把服务端这段等待拆开：模型每生成一个 Token，就立刻通过流式传输送出去。对用户来说，看到的是文字一个字一个字地出现，但这背后是大模型真实的生成节奏。所以流式不是"模拟打字效果"的假把式，而是把生成过程本身暴露给了调用方。

1.2 Runnable 的统一抽象：invoke、stream、batch 的关系

LangChain 把核心抽象统一成了 Runnable 接口。凡是实现了这个接口的组件——模型、提示模板、输出解析器，以及用管道符串起来的链——都支持四类标准调用方式：

这里最关键的认知是：stream()不是invoke()的"加速版"，而是返回一个迭代器，你需要自己去遍历它。每遍历一次，拿到的是生成过程中的一个片段。以聊天模型为例，stream()产生的每个块通常是一个AIMessageChunk对象，里面的.content字段保存的是当前片段新增的那一小段文字。

要注意的是：一个链是否真的能逐 Token 输出，取决于链里每个环节是否都支持并传播流式。这个问题后面在复杂场景部分会详细展开，这里先记住结论——Runnable 层面的抽象保证了"接口长得一样"，但底层到底流不流得动，还要看具体实现。

1.3 三个容易被混为一谈的"流"

在 LangChain 的文档和社区讨论里，"流式"这个词至少指代三种东西，很多人讨论时把这三种混在一起，自然就糊涂了。

第一种是Token 流，也就是模型生成的最原始内容增量。模型每算出一个新 Token，框架就把它封装成一个块交给你。这种流最细、最实时，适合直接推到前端渲染文字。

第二种是消息块流。聊天模型的流式输出通常以AIMessageChunk为单位，里面除了content增量之外，还可能带工具调用参数、用量统计这类附加信息。在 Agent 场景里，模型可能会一边生成普通文字、一边生成工具调用参数，这些信息也会以块的形式流出来。

第三种是事件流。这不再是模型输出层面的事，而是整条链在执行过程中产生的事件序列：链开始、检索器命中、Prompt 组装、模型开始吐字、模型结束……每个事件带上下文信息。事件流适合用来做状态展示、日志追踪和复杂调试。

我的经验是：做文字输出，盯着 Token 流就够；做状态反馈和调试，用事件流；消息块流通常藏在中间层，普通业务很少直接消费。后面讲到的回调机制，本质上就是让这三层信息都有一条通路可以透出来。

2. 第一次接通流式：从 stream() 到异步流

2.1 用 stream() 替换 invoke()，三行代码跑通逐字输出

如果你的业务就是"大模型给我讲个故事"，那接通流式最简单的方式，就是把invoke()换成stream()。先看一下典型写法：

from langchain_openai import ChatOpenAI MODEL_NAME = "你实际使用的模型名称" # 这里以 ChatOpenAI 为例，换成任何实现了流式响应的模型封装类都一样 llm = ChatOpenAI(model=MODEL_NAME, temperature=0.7) full_text = "" for chunk in llm.stream("给我讲一个程序员的幽默故事"): text_piece = chunk.content if isinstance(text_piece, str): print(text_piece, end="", flush=True) full_text += text_piece print("\n--- 完整内容 ---") print(full_text)

这里有几个细节值得说明。

第一，chunk.content在聊天模型里可能直接是字符串，也可能是多模态内容列表。如果模型返回的是纯文本，它就是一个普通字符串；如果模型返回带图片或结构化内容，content可能是一段列表，遍历时要注意做类型判断。写通用代码时，我习惯先判断isinstance(text_piece, str)。

第二，必须自己累加。stream()返回的是增量，不是全量。想要最终完整文本，就在循环里把每个chunk.content拼起来。很多新手在这里踩坑：直接拿最后一个 chunk 当完整结果，发现只有结尾几个字。

第三，flush=True在终端里很重要。Python 的print默认有缓冲，不加flush的话，效果还是"等一会一次性蹦出来"。

2.2 不想手写循环就用 StreamingStdOutCallbackHandler

有些场景你并不想自己遍历迭代器，而是希望框架在内部生成新 Token 时自动"通知"你。这时可以用回调处理器。LangChain 官方提供了一个现成的打印处理器，叫StreamingStdOutCallbackHandler：

from langchain_core.callbacks import StreamingStdOutCallbackHandler streaming_llm = ChatOpenAI( model=MODEL_NAME, streaming=True, # 告诉模型封装层：走流式生成 callbacks=[StreamingStdOutCallbackHandler()], ) result = streaming_llm.invoke("讲一个程序员笑话") print("\n最终结果：", result.content)

注意这里有意思的点：invoke()本来是不流式的，但当你把streaming=True打开，同时挂上回调处理器之后，模型生成过程中每个新 Token 都会触发on_llm_new_token回调，StreamingStdOutCallbackHandler就会把它打印到标准输出。于是你既拿到了invoke()返回的完整结果，又享受了"逐字打印"的视觉效果。

这个"回调 + streaming 开关"的组合，是 LangChain 流式能力里最容易被忽略的一层。后面自定义回调处理器那一节，就是在这个基础上扩展的。

2.3 异步场景：astream() 与事件循环注意事项

如果项目本身是异步框架（比如后端服务），那就要用astream()。用法和stream()几乎一样，只是返回的是异步迭代器：

import asyncio async def generate(): full_text = "" async for chunk in llm.astream("写一段关于异步编程的比喻"): piece = chunk.content if isinstance(piece, str): full_text += piece return full_text result = asyncio.run(generate()) print(result)

异步版本有两个高频坑，我提前说。

一个是在 Jupyter 这类已经有事件循环的环境里直接调用asyncio.run()会报错。Jupyter 本身就跑在一个事件循环上，asyncio.run()要求"没有正在运行的循环"，所以会抛异常。这时候直接在单元格里写await llm.astream(...)即可。

另一个是在生产环境里把异步循环用错。比如在异步回调里同步调用阻塞式 HTTP 请求去转发 Token，这个同步调用会卡住事件循环，导致后续 Token 全部延迟。正确做法是：要么用异步 HTTP 客户端，要么把转发任务丢到线程池里，总之别让同步 IO 阻塞事件循环。这个问题我在踩坑章节还会详细讲。

3. 回调机制：流式项目的"神经系统"

3.1 回调事件的生命周期：从 on_llm_start 到 on_llm_end

LangChain 流式能力真正强大之处，不只是stream()那层迭代器，而是它背后完整的回调机制。你可以把回调理解成"钩子"：框架在执行链的各个阶段时，会抛出对应的事件名，你在处理器里实现同名方法，就能在那一刻插入自己的逻辑。

以大模型执行为例，最核心的几个事件如下：

• on_llm_start：模型开始生成时触发，参数里带着请求的消息列表。
• on_llm_new_token：每生成一个新 Token 时触发，这是流式场景里最常用的钩子。
• on_llm_end：模型完整生成结束时触发，带完整结果。
• on_llm_error：生成过程中抛异常时触发。

除了模型层，链层面、检索器层面、工具层面也都有对应的开始和结束事件。比如on_chain_start/on_chain_end、on_retriever_start/on_retriever_end、on_tool_start/on_tool_end。

理解这套生命周期，对流式项目非常有帮助。因为很多产品的真实需求不是"把字打出来"那么简单，而是：用户问问题之后，界面要先显示"正在检索资料"，然后显示"正在思考"，最后才看到文字逐字蹦出。这些阶段状态，靠的就是监听不同层次的事件。

3.2 自定义回调处理器：把每个 Token 转发给上层

我自己用得最多的场景，是把流式 Token 实时转发给消息通道或流式响应。下面这个处理器，就是把 Token 放进一个队列，由另一个消费线程负责推给前端：

import queue from langchain_core.callbacks import BaseCallbackHandler class TokenForwardHandler(BaseCallbackHandler): """把每个新 Token 放入队列，由外部消费者推给前端。""" def __init__(self, token_queue: queue.Queue): self.token_queue = token_queue def on_llm_new_token(self, token: str, **kwargs) -> None: if token: self.token_queue.put(token) def on_llm_end(self, response, **kwargs) -> None: # 通知消费者生成结束 self.token_queue.put(None)

用的时候，把它挂到模型或链上：

token_queue = queue.Queue() handler = TokenForwardHandler(token_queue) chat_model = ChatOpenAI( model=MODEL_NAME, streaming=True, callbacks=[handler], ) # 在任意线程里执行 invoke，事件回调会自动往 queue 里塞 token chat_model.invoke("给我解释什么是流式输出")

这个模式的好处是：生成的执行线程和消费线程彻底解耦。哪怕你后面把生成逻辑换成异步的、换成 Agent、换成 RAG，只要回调处理器还在，前端消费队列的代码一行都不用改。

在实际项目里，我还会在on_llm_start时向队列推送一条{"type": "status", "content": "thinking"}这类状态消息，让前端能展示"正在思考"的过程状态；在on_retriever_start/end时推送检索状态。这样前端的交互就能完整反映后台每个阶段。

3.3 用 astream_events 直接拿统一事件流

如果你的链比较复杂，不想在多个组件上分别挂回调，LangChain 还提供一个更粗暴但也更灵活的方式：astream_events()。它会把整条链运行过程中的所有事件，按照发生顺序抛出来，让你统一监听。

from langchain_core.prompts import ChatPromptTemplate prompt = ChatPromptTemplate.from_template("用一句话解释：{topic}") chain = prompt | chat_model async for event in chain.astream_events( {"topic": "什么是流式处理"}, version="v1", # 新版必须带 version 参数 ): kind = event["event"] if kind == "on_chat_model_stream": chunk = event["data"].get("chunk") if chunk is not None: content = chunk.content if isinstance(content, str): print(content, end="", flush=True) elif kind == "on_chain_end": # 处理链结束事件，可以在这里做清理或统计 pass

astream_events()的事件对象是个字典，核心字段有event（事件类型）、data（事件数据）、run_id（本次运行的唯一标识）等。事件类型可以粗略分成前缀on_加上组件类型和阶段，比如on_chat_model_start、on_chat_model_stream、on_chat_model_end、on_retriever_end等。

用它的好处是：一整套事件都在一条流里，你可以在一个地方统一处理状态推进、Token 转发、日志记录。代价是事件非常密，如果只关心某几类事件，记得像我上面那样用if过滤，否则光日志就能把你淹没。

4. 复杂链路里的流式实践：Chain、Agent 与 RAG

4.1 链式调用中，流式为什么经常"断"在中间

很多人第一次用链式写法时都会遇到一个怪现象：直接对模型llm.stream()是逐字出的，但一旦写成prompt | llm | parser这种管道链，再对整个链调用stream()，有时候就变成"一把梭"——等半天，哗啦全出来。

原因是流式在链里传播是有条件的。管道中的每个组件，必须支持"把上一级的块流式地往下传"。如果中间某个组件的实现是先把上游全部消费完再加工，那么流式到它这里就断了。

举个最常见的例子：输出解析器。像StrOutputParser这类解析器是支持流式的，因为字符串解析可以逐段处理；但某些需要"完整结果"才能工作的解析器（比如要校验 JSON Schema 或构造结构化对象的解析器），必须在内部缓存所有片段，等全部 Token 到位后再解析，于是下游拿到的就只有一大坨结果。

排查这种问题的思路很简单：逐级测试。先llm.stream()看模型层是否流畅，再(prompt | llm).stream()看加提示模板后是否流畅，最后加解析器再看。哪一级开始不对劲，问题就在哪一级。

如果问题出在自定义组件，LangChain 提供了一个底层方法叫transform()，专门用来让上游的块序列经过处理后继续以块的形式往下游流。你自己写的自定义可运行组件如果需要对流式块做加工，建议用它而不是等上游全部结束再一次性处理。

4.2 Agent 场景：让用户在工具调用期间也能看见动静

Agent（智能体）场景比普通链复杂得多，因为它的执行过程是循环的：模型可能先生成一段思考文字，然后发起工具调用，工具返回结果后，模型再基于结果继续生成，直到给出最终回答。

早年间用基于执行器的 Agent 时，有个让很多人头疼的问题：最终回答的生成阶段拿不到干净的逐 Token 流。你在前端看到的是工具调用结束后，停顿几秒，然后整段最终回答一次性出现。

我当时用的解决办法，是给 Agent 执行器挂上自定义回调处理器：

from langchain.agents import AgentExecutor, create_react_agent token_queue = queue.Queue() handler = TokenForwardHandler(token_queue) # 复用第三节的处理器 agent = create_react_agent(llm=chat_model, tools=tools, prompt=agent_prompt) executor = AgentExecutor( agent=agent, tools=tools, callbacks=[handler], ) executor.invoke({"input": "帮我查一下最近的天气，然后写一首关于雨的诗"})

挂上之后，on_llm_new_token会捕获到模型在整个 Agent 循环里生成的每一个 Token，包括思考过程和工具调用的中间输出。前端拿到这些 Token 后，就能实时展示"智能体正在思考 / 正在调用工具 / 正在写诗"的完整过程。

这里有个附加问题：你会同时收到思考过程和最终回答的 Token，前端如果都渲染出来会很乱。我的做法是给回调处理器增加一个状态标记：在on_tool_start时切换状态为工具调用阶段，在on_llm_start时如果发现后面紧跟工具调用，就把这批 Token 标为思考内容，只在日志里展示；只有工具调用之后的那一批 Token，才作为正式回答推给前端。

如果你已经迁移到基于图执行的新版 Agent 框架，情况会好很多。它支持配置流式模式，可以直接拿到生成消息的增量块，也可以拿到每步的状态更新。我的建议是：新项目用新框架的流式能力，老项目用回调处理器兜底，不要为了一个流式效果反复重写业务逻辑。

4.3 RAG 场景：检索状态先行，生成内容后至

RAG（检索增强生成）的链路通常是：用户提问 → 检索相关文档 → 拼接上下文 → 模型生成回答。很多人一开始只对最后一步做流式，结果用户提完问题后，界面白屏好几秒（检索 + 拼接），然后文字才慢慢出来。

这里的问题不是流式没接通，而是检索阶段没有给用户任何反馈。检索器本身不是一个流式组件，它是一次性的阻塞操作；但这不代表用户应该干等。

我的标准做法是分两段反馈：

• 第一段，检索阶段。监听on_retriever_start，向队列推一条状态消息："正在检索相关文档…"；监听on_retriever_end，推送"已找到若干篇相关文档，正在生成回答…"。
• 第二段，生成阶段。用回调处理器或astream_events监听on_chat_model_stream，把 Token 逐字推给前端。

两段拼起来，用户看到的就是一个连续流畅的过程：提问 → 检索状态 → 生成状态 → 文字逐字蹦出。技术上完全没有难点，就是"状态消息"和"Token 消息"走同一个队列而已。

这段实践也说明一个通用原则：流式不等于"只流 Token"。在复杂链路里，把阶段状态也当成一种"流"，一起推给前端，体验会好非常多。

5. 踩坑实录：流式失效的六个高频原因与排查链路

说完了怎么把流式用起来，这一节专门讲它为什么会失效。下面六个坑都是我或身边同事真实踩过的，每个都按"现象 → 排查 → 根因 → 修复"的思路来讲，方便你直接对照。

5.1 模型服务端根本没有开启流式

现象：stream()能正常迭代出块，但第一个块要等很久才来，而且整个迭代几乎在最后一瞬间一次性吐出所有内容，和普通调用没区别。

排查：先发一个裸请求直接打模型提供商的接口，检查响应是不是流式格式；或者看服务端接口的耗时分布，确认是不是首字节延迟极高。

根因：模型封装类虽然支持流式，但底层 HTTP 请求里的流式开关没打开。不同提供商的参数名不一样，有的叫stream，有的叫streaming，有的要求显式传True才走流式通道。

修复：检查你用的模型封装类的构造函数，确认流式参数已开启。以ChatOpenAI为例，streaming=True会直接影响底层请求；如果调用了stream()方法但没开这个开关，部分版本的表现就会退化。另外，如果你用的是网关、代理这类中间件，还要确认中间件没有把流式响应缓冲成普通 JSON。

5.2 方法用错：invoke() 永远等完整结果

现象：代码里写了llm.invoke(...)，然后来问为什么没有流式效果。

排查：这个其实不用排查，看一眼方法名就知道。invoke()的设计语义就是"返回完整结果"，它不是流式接口。

根因：对 Runnable 体系的接口理解有偏差。invoke()、batch()都是面向"结果"的；stream()、astream()才是面向"过程"的。想让文字逐字出现，必须用流式方法，或者用streaming=True加回调的组合。

修复：二选一。如果你需要自己控制输出节奏，用stream()遍历；如果你不关心输出节奏、只是想让终端或日志里看到生成过程，用invoke()配合StreamingStdOutCallbackHandler。

5.3 回调处理器里做重活，把首字延迟拖垮

现象：Token 确实在流，但第一个 Token 要 3 到 5 秒才出现，后续每个 Token 之间也有明显卡顿。

排查：先去掉自定义回调处理器，单独用StreamingStdOutCallbackHandler测试，看延迟是否恢复正常。如果恢复正常，说明问题出在自定义处理器里。

根因：on_llm_new_token是同步回调，在模型每生成一个 Token 时都会执行一次。如果你在这个方法里做了重活——比如同步发 HTTP、写数据库、写磁盘日志——每次回调都会阻塞生成循环。Token 数一多，累积延迟极其可观。

修复：回调里只做轻量操作，把 Token 放进队列，由独立线程批量消费，攒够一定数量或每隔一段时间再批量转发。或者直接用异步队列配合异步任务处理，避免阻塞事件循环。

5.4 异步环境运行冲突：Jupyter、事件循环与嵌套调用

现象：在 Jupyter 里调用asyncio.run()报错"event loop is already running"；在生产环境里，流式接口偶尔会突然卡住几秒再恢复。

排查：看调用点所在环境。Jupyter 本身就有事件循环，asyncio.run()要求"无正在运行的循环"，直接冲突。生产环境的卡顿，则需要检查代码里是否在异步回调中做了同步阻塞调用。

根因：异步代码的事件循环模型和同步代码不同。asyncio.run()不能在已有循环的环境里嵌套使用；在事件循环里执行同步阻塞调用，会阻断整个循环，导致所有任务排队等待。

修复：Jupyter 里直接用await astream(...)；生产环境里保证事件循环内不做阻塞 IO，同步转发逻辑放进线程池，或用异步客户端。另外注意，异步迭代器要正确使用async for消费，不能简单转换成列表再遍历，否则又回到"等全部生成完"的老路。

5.5 链中步骤提前聚合，流式被缓冲截断

现象：对整个链调用stream()，前几个块出来后突然没动静，最后整个结果一次性输出。

排查：从上到下逐级测试：先单独测llm，再测prompt | llm，再加解析器，找到断点。断点往往出现在某个自定义函数、输出解析器或者用了批量映射的组件上。

根因：链中某个组件没有实现流式传播，而是先聚合上游全部输入，处理完后再一次性丢给下游。对用户来说，这一级之前是流的，到这一级就"堵住"了。

修复：给自定义组件实现transform()或使用支持流式传播的组合方式；对不支持流式的第三方解析器，评估是否值得换一个支持流式的同类实现。如果只是日志或统计需要完整结果，可以像前面说的那样用回调在链尾收集，而不是在中游打断流式。

5.6 流式传输链路问题：超时、代理缓冲与前端断线

现象：后端流式正常，Token 也在生成，但前端页面长时间不更新；或者连接在生成中途断开，Token 丢了后半截。

排查：用命令行工具直接请求流式接口，观察是"没数据"还是"一次性收到全部数据"，或者"生成到一半连接断开"。同时检查链路中是否有 Nginx 这类反向代理。

根因：流式接口的坑往往不在应用层而在传输层。反向代理默认会对响应做缓冲，可能攒够一定字节才往下发；网关超时设置过短，长输出生成到一半就断开；前端如果长时间没收到任何数据，部分浏览器会主动断开连接。

修复：在响应头里显式设置关闭代理缓冲；生成期间定期发送保持连接的心跳行，防止中间层误判超时；前端实现自动重连逻辑，并在断线后带上"已生成内容"的续传参数。实测中这些改动叠加后，长文本生成的稳定性提升非常明显。

6. 什么时候该用流式，什么时候别硬上

6.1 流式的隐性成本：不只是"改一行代码"

流式看起来只是把invoke换成stream，但它带来的系统影响比想象中大。首先是日志和存储：逐 Token 事件会产生大量日志，如果你的监控系统按条计费，成本会肉眼可见地上升。其次是统计口径：原来的"一次调用算一笔耗时"，变成"每个 Token 都要记时"，指标聚合逻辑要跟着改。再次是测试方式：返回完整结果时断言一个字符串即可；流式场景下要模拟客户端逐块接收，测试代码复杂度直接翻倍。

这些成本不是劝退，而是提醒你：流式是给终端用户体验用的，不是给后台任务用的。如果你的消费者不是一个"等待文字出现的人"，而是批处理脚本、定时任务、数据管道，那就没必要上流式，invoke()反而更省心。

6.2 按业务场景选型的经验表

下面是我自己项目里用过的选型方案，供你参考：

6.3 最后分享几条实战心得

写到这里，挑几条自己反复受益的经验作为收尾。

第一，永远先测"模型裸流"。不管你的链路多复杂，第一件事永远是llm.stream("你好")看能不能逐字输出。模型层不通，后面全白搭；模型层通了，每加一层就测一次，问题会快速定位。

第二，状态流和 Token 流分开设计，但走同一个通道。前端的渲染逻辑可以简单地按消息类型分发：状态类型切换状态文案，Token 类型追加文字。这样后端加什么新阶段，前端都不用改协议。

第三，流式接口一定加超时和重连。网络环境没有理想的，流式连接随时可能断。为前端设计"断线重连 + 续传提示"能力，比追求绝对不断连更现实。

关于流式处理，能聊的细节其实远不止这些，比如多轮对话里如何避免重复输出历史、Prompt 缓存对首字延迟的影响等等。但核心思路就一条：把"生成过程"和"消费过程"解耦，让实时性成为架构的默认选项，而不是事后补丁。希望这篇指南能帮你少走一些弯路。