1. 项目概述为什么流式输出是LangGraph的灵魂如果你用过LangChain大概率体验过那种“等待-等待-砰”的完整响应返回模式。在构建复杂的AI工作流时这种同步阻塞的体验对于终端用户和开发者调试来说都是一种煎熬。而LangGraph的Stream流式输出功能正是为了解决这个核心痛点而生。它不仅仅是把一个大块的文本拆成小段吐出来那么简单而是将整个工作流的执行过程和中间状态实时地、透明地暴露给你。想象一下你构建了一个包含“检索-分析-生成-审核”多步骤的智能客服流程。没有流式输出时用户面对的是一个沉默的输入框直到所有步骤跑完答案才一次性出现。这不仅让用户感到焦虑“它到底有没有在干活”一旦出错你也很难定位问题到底卡在了“检索”还是“生成”环节。而有了流式输出你可以看到“正在检索相关知识库...”、“找到了3条相关文档”、“开始组织回答...”、“正在生成最终回复...”最后才是完整的答案。这种可观测性和即时反馈是构建可靠、可信AI应用的关键。在LangGraph中stream不仅仅是一个输出模式它更是一种调试工具和交互设计范式。通过流式接口我们可以实时捕获工作流中每个节点的输入输出、状态变更甚至是分支决策的逻辑。这对于理解复杂图Graph的执行路径、优化性能瓶颈、以及向最终用户提供渐进式体验都具有不可替代的价值。本次笔记我们就深入LangGraph的流式世界从基础用法到高级监控彻底掌握这一核心特性。2. 核心概念与流式输出原理拆解在深入代码之前我们必须厘清几个关键概念这能帮你理解流式输出背后的设计哲学而不仅仅是调用一个API。2.1 LangGraph中的“流”是什么这里的“流”Stream是一个广义概念它包含两个层面数据流Data Streaming这是最常见的形式即LLM大语言模型生成文本时以Token词元为单位的逐词输出。这依赖于底层LLM如OpenAI的ChatCompletion接口本身支持的流式响应。事件流Event Streaming这是LangGraph流式输出的精髓。它流式传输的是工作流执行过程中的各种事件。这些事件让你能像看电影一样观察整个有状态图StateGraph的运行帧。LangGraph的stream方法返回的是一个异步迭代器Async Iterator每次迭代产出的不是一个简单的字符串而是一个包含丰富信息的事件对象。这是它与LangChain普通流式回调最根本的区别。2.2 关键事件类型解析当你调用graph.stream(input)时你会接收到不同类型的事件。理解每种事件的含义是有效利用流式输出的前提。主要事件类型包括on_chain_start / on_chain_end标志一个“链”可以是单个LLM调用也可以是一个复杂的RunnableSequence的开始和结束。on_chain_end事件中会包含该链的输出结果。这是获取每个节点产出的主要途径。on_tool_start / on_tool_end标志一个工具Tool调用的开始和结束。on_tool_end中包含工具执行后的返回结果。例如你调用了一个网络搜索工具这里就能实时看到搜索到的内容。on_llm_start / on_llm_stream / on_llm_end专门针对LLM调用的事件。on_llm_stream是真正的Token级流式输出事件它携带了LLM实时生成的每个Delta增量。on_llm_end则包含了LLM调用的完整响应和Token用量等信息。on_graph_start / on_graph_end标志整个图工作流的开始和结束。注意在LangGraph中一个节点Node可以是一个简单的函数也可以是一个复杂的LangChain Runnable如一个链。当节点是Runnable时其内部执行会触发上述on_chain_start/end等更细粒度的事件。这形成了层次化的监控体系。2.3 流式输出与普通invoke的本质区别很多人初学会混淆graph.invoke()和graph.stream()。invoke是同步阻塞调用它一次性输入等待所有计算完成然后一次性返回最终的状态State。你丢失了所有中间过程。而stream是异步非阻塞的。它立即返回一个异步生成器你可以一边消费事件一边图还在继续执行后续节点。你得到的是过程Events而最终的State通常包含在最后一个on_graph_end事件或通过其他方式获取。这种设计使得实时交互前端可以随着LLM的思考Reasoning或工具调用结果逐步更新UI。过程调试你可以精确看到错误发生在哪个节点的哪个环节而不是得到一个笼统的异常。资源优化对于长流程可以在生成部分结果后提前进行一些预处理或验证。3. 基础到进阶四种流式输出实战理论说得再多不如一行代码。我们从一个最简单的图开始逐步演示不同颗粒度的流式输出方法。3.1 搭建一个基础示例图我们先构建一个包含两个节点的简单工作流一个节点生成诗歌主题另一个节点根据主题写诗。from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI import operator # 1. 定义状态 class PoemState(TypedDict): topic: str poem: Annotated[list, operator.add] # 用于累积消息 final_poem: str # 2. 初始化LLM llm ChatOpenAI(modelgpt-4o-mini, streamingTrue) # 注意此处streamingTrue启用了LLM层的流式 # 3. 定义节点函数 def generate_topic(state: PoemState): 生成诗歌主题 message llm.invoke(请随机生成一个中文诗歌主题例如‘秋夜’、‘远山’。只返回主题词。) return {topic: message.content} def write_poem(state: PoemState): 根据主题写诗 prompt f以‘{state[topic]}’为主题创作一首七言绝句。 # 这里我们调用stream以便在节点内部也观察LLM流式生成 full_response for chunk in llm.stream(prompt): if chunk.content is not None: full_response chunk.content # 在实际应用中这里可以将chunk.content发送给前端 print(f[LLM正在生成]: {chunk.content}, end, flushTrue) print() # 换行 return {poem: [HumanMessage(contentfull_response)], final_poem: full_response} # 4. 构建图 workflow StateGraph(PoemState) workflow.add_node(generate_topic, generate_topic) workflow.add_node(write_poem, write_poem) workflow.set_entry_point(generate_topic) workflow.add_edge(generate_topic, write_poem) workflow.add_edge(write_poem, END) app workflow.compile()3.2 方法一消费原始事件流最详细这是最底层、信息最全的方式。你直接遍历stream方法返回的异步生成器处理每一个事件对象。async def stream_events_example(): inputs {topic: , poem: [], final_poem: } async for event in app.astream(inputs, stream_modevalues): # event 是一个元组 (node_name, event_data) node_name, event_data event print(f\n--- 节点事件: {node_name} ---) print(f事件类型: {event_data.get(event)}) if data in event_data: data event_data[data] # 根据不同事件类型处理data if event_data[event] on_chain_end: print(f输出: {data.get(output)}) elif event_data[event] on_llm_stream: # 这里是真正的Token流 if data.get(chunk): content data[chunk].content if content: print(fToken: {content}, end, flushTrue) print(- * 30) # 运行 import asyncio asyncio.run(stream_events_example())输出示例--- 节点事件: generate_topic --- 事件类型: on_llm_start ------------------------------ --- 节点事件: generate_topic --- 事件类型: on_llm_stream Token: 孤 Token: 舟 Token: 蓑 Token: 笠 ------------------------------ --- 节点事件: generate_topic --- 事件类型: on_chain_end 输出: {topic: 孤舟蓑笠} ------------------------------ --- 节点事件: write_poem --- 事件类型: on_llm_start ------------------------------ ...实操心得stream_modevalues参数是关键。它还有updates只流式状态更新和messages专用于消息数组等选项。对于调试values模式最全面。但在生产环境面向用户时你可能更关心updates或直接处理on_llm_stream来推送文字。3.3 方法二聚焦状态更新流如果你只关心图状态State的变化可以使用stream_modeupdates。这过滤掉了大量的中间事件只在你定义的节点函数返回时推送状态发生了哪些改变。async def stream_updates_example(): inputs {topic: , poem: [], final_poem: } async for chunk in app.astream(inputs, stream_modeupdates): # chunk 直接就是状态更新字典 print(f\n状态更新: {chunk}) # 例如第一次更新可能是 {topic: 孤舟蓑笠} # 第二次更新可能是 {poem: [HumanMessage(...)], final_poem: ...} asyncio.run(stream_updates_example())这种方法输出更简洁直接对应你return的内容非常适合用来驱动前端状态同步。3.4 方法三使用stream_events方法获取结构化事件LangGraph提供了一个更便捷的stream_events方法它返回的事件对象结构更统一易于解析。这是目前官方更推荐的方式。inputs {topic: , poem: [], final_poem: } for event in app.stream_events(inputs, versionv1): kind event[event] # 事件类型如 on_chain_start, on_llm_stream name event.get(name) # 节点或Runnable的名称 data event.get(data, {}) if kind on_llm_stream: chunk data.get(chunk) if chunk and chunk.content: print(chunk.content, end, flushTrue) # 实时输出LLM生成内容 elif kind on_tool_end: print(f\n[工具 {name} 执行完毕]: 输出 - {data.get(output)}) elif kind on_chain_end: # 一个节点链运行结束 print(f\n[节点 {name} 完成])stream_events方法的事件结构非常清晰并且versionv1参数保证了API的稳定性。它是我在开发和调试中最常使用的工具。3.5 方法四在普通函数中消费流非异步如果你的环境不支持顶层async/await可以使用stream方法注意不是astream在同步上下文中迭代。inputs {topic: , poem: [], final_poem: } for event in app.stream(inputs, stream_modevalues): node_name, event_data event # 处理逻辑与异步版本类似但注意内部如果是异步LLM调用可能仍有阻塞。 if event_data.get(event) on_llm_stream: chunk event_data.get(data, {}).get(chunk) if chunk and chunk.content: print(chunk.content, end, flushTrue)重要提示在同步stream中虽然外层是迭代但图节点的执行仍然是顺序且同步的。它不会像真正的异步那样实现并发。对于复杂的、包含多个可并行节点的图应优先使用异步astream。4. 高级应用流式输出与复杂控制流、自定义回调掌握了基础用法我们来看看如何将流式输出应用到更复杂的场景中。4.1 在条件边Conditional Edge和分支中追踪路径当你的图包含conditional_edge或tools分支时流式输出能让你清晰地看到执行路径的选择。from langgraph.graph import StateGraph, END from langgraph.checkpoint import MemorySaver from typing import Literal class RouterState(TypedDict): question: str answer: str route: Literal[physics, math, general] def route_question(state: RouterState): 路由问题到不同领域 question state[question] if 力 in question or 运动 in question: return {route: physics} elif 方程 in question or 几何 in question: return {route: math} else: return {route: general} def physics_expert(state: RouterState): return {answer: 这是一个物理问题涉及牛顿力学。} def math_expert(state: RouterState): return {answer: 这是一个数学问题需要解方程。} def general_ai(state: RouterState): return {answer: 这是一个通用问题我来尝试回答。} # 构建带条件边的图 workflow StateGraph(RouterState) workflow.add_node(router, route_question) workflow.add_node(physics, physics_expert) workflow.add_node(math, math_expert) workflow.add_node(general, general_ai) workflow.set_entry_point(router) # 根据route字段的值决定下一个节点 workflow.add_conditional_edges( router, lambda state: state[route], { physics: physics, math: math, general: general } ) workflow.add_edge(physics, END) workflow.add_edge(math, END) workflow.add_edge(general, END) app workflow.compile() # 流式执行并观察路径 inputs {question: 如何计算物体在斜面上的加速度, answer: , route: } print(开始流式执行观察路由选择) for event in app.stream_events(inputs, versionv1): if event[event] on_chain_end: name event.get(name) data event.get(data, {}) output data.get(output, {}) if name router: print(f\n路由节点决策结果: {output}) # 会输出 {route: physics} elif name in [physics, math, general]: print(f专家节点 {name} 产生答案: {output})通过流式事件你可以明确看到router节点输出了{route: physics}然后图自动进入了physics节点。这对于调试复杂业务逻辑至关重要。4.2 集成自定义回调与外部监控你可以将LangGraph的流式事件接入你现有的监控系统如Logging, OpenTelemetry, 或前端WebSocket。import json import logging class CustomStreamingHandler: 自定义流式处理器用于日志和推送 def __init__(self, websocketNone): self.websocket websocket self.logger logging.getLogger(__name__) async def handle_event(self, event): kind event[event] # 1. 结构化日志 self.logger.info(fLangGraph Event - {kind}, extra{event_data: event}) # 2. 向前端推送特定事件例如LLM生成内容 if kind on_llm_stream: chunk event.get(data, {}).get(chunk) if chunk and chunk.content and self.websocket: message json.dumps({type: token, data: chunk.content}) await self.websocket.send_text(message) elif kind on_tool_end: # 推送工具执行结果 tool_name event.get(name) output event.get(data, {}).get(output) if self.websocket: message json.dumps({type: tool_result, tool: tool_name, data: str(output)}) await self.websocket.send_text(message) # 使用示例 async def run_graph_with_custom_handler(inputs): handler CustomStreamingHandler() # 可以传入真实的WebSocket对象 events app.stream_events(inputs, versionv1) async for event in events: # 注意stream_events 在最新版本中也可能有异步版本 # 这里为了演示我们模拟异步迭代。实际需根据LangGraph版本调整。 # 核心思想在消费事件的循环中调用自定义处理器。 await handler.handle_event(event)通过这种方式你将LangGraph的内部执行流转化为了可观测、可交互的业务事件流。4.3 处理流式中断与超时在生产环境中网络可能不稳定或者用户可能提前关闭连接。你需要优雅地处理流式中断。import asyncio from contextlib import AsyncExitStack async def stream_with_timeout_and_cancel(app, inputs, timeout30, cancel_eventNone): 带超时和取消机制的流式调用 Args: cancel_event: 一个asyncio.Event用于从外部如HTTP请求取消通知中断。 try: # 使用wait_for设置总超时 async for event in asyncio.timeout(timeout, app.astream(inputs, stream_modeupdates)): # 检查外部取消信号 if cancel_event and cancel_event.is_set(): print(流被外部取消) break # 正常处理事件 print(f收到更新: {event}) # 这里可以yield给调用方 yield event except asyncio.TimeoutError: print(f流式执行超过 {timeout} 秒已超时终止) # 这里应该触发一些清理逻辑 except Exception as e: print(f流式执行发生错误: {e}) raise这个模式在部署为Web API如FastAPI、Django时非常有用你需要将请求的disconnect事件映射到cancel_event上。5. 常见问题、性能调优与避坑指南在实际使用中我踩过不少坑也总结了一些优化经验。5.1 流式输出“不流”了检查这三点LLM客户端配置确保你的ChatOpenAI、ChatAnthropic等客户端初始化时传入了streamingTrue。这是Token级流式的基础。# 正确 llm ChatOpenAI(modelgpt-4, streamingTrue) # 错误即使外层用了streamLLM内部也不会流式生成Token。 llm ChatOpenAI(modelgpt-4)节点函数内部是否阻塞如果你的节点函数内部是调用llm.invoke()同步阻塞那么即使外层用app.stream你也只能收到节点开始和结束的事件看不到LLM生成Token的过程。应改用llm.stream()或在异步上下文中使用llm.ainvoke()。stream_mode选择如果你用了stream_modeupdates那么你只会收到状态更新事件而不会收到on_llm_stream这种细粒度事件。调试时建议先用values或stream_events。5.2 性能瓶颈分析与优化流式输出本身开销很小但不当使用会影响整体吞吐。瓶颈一过多的细粒度事件。在包含数十个节点的复杂图中每秒可能产生上百个事件。如果你的自定义处理器如写入数据库、复杂转换很重会成为瓶颈。优化在处理器中过滤事件只处理你关心的如仅on_llm_stream和on_tool_end。或者使用异步队列将事件处理与消费解耦。瓶颈二同步I/O阻塞事件循环。如果你在异步的流式循环中执行了同步的磁盘写入、网络请求未使用异步库会阻塞整个事件循环导致流式卡顿。优化将所有的I/O操作异步化。使用aiofiles代替open使用aiohttp或httpx代替requests。瓶颈三状态过大。每次状态更新整个State对象都会被序列化并在流中传递。如果State中存储了巨大的列表或文档会显著增加网络和内存开销。优化精简State。只存放必要的数据。对于大文档考虑存放引用如ID或路径在节点需要时再加载。5.3 错误处理与状态回滚流式执行中某个节点可能抛出异常。你需要决定是让整个流停止还是尝试恢复。from langgraph.graph import StateGraph, END from langgraph.checkpoint import MemorySaver workflow StateGraph(...) # ... 添加节点和边 ... memory MemorySaver() app workflow.compile(checkpointermemory) config {configurable: {thread_id: user_123}} try: async for event in app.astream(input, configconfig): # 处理事件 pass except Exception as e: print(f工作流执行失败: {e}) # 利用检查点可以获取失败时的状态用于分析或恢复 checkpoint await memory.aget(config) print(f失败时的状态快照: {checkpoint[channel_values]})结合检查点Checkpoint你不仅能实现“断点续跑”还能在流式执行出错时精准定位到出错前的状态极大方便了调试和错误恢复。5.4 前端对接实战要点将LangGraph的流式输出对接前端如WebSocket时需要注意数据格式。定义清晰的事件协议不要简单地把LangGraph的原始事件对象扔给前端。前端只关心业务状态。建议定义如下的简单协议{type: status, data: 正在检索...} {type: token, data: 今天} {type: tool_result, tool: search, data: 找到了10条结果} {type: final_answer, data: 综上所述...}处理连接中断如前所述前端页面关闭或刷新时后端要及时捕获asyncio.CancelledError并停止流式生成避免资源浪费。错误信息流式传递如果某个节点执行失败不要等到整个流结束才报错。可以通过自定义事件立即向前端推送一个{type: error, data: ...}的消息。流式输出不是一种炫技而是构建现代AI应用的基础设施。它连接了后端的复杂逻辑与前端的即时体验也架起了开发者与黑盒工作流之间的调试桥梁。从最初只是简单打印Token到如今能精细控制整个图的执行脉络LangGraph在可观测性上确实向前迈了一大步。我个人的体会是在项目早期就接入流式输出虽然增加了一些开发复杂度但它在调试和用户体验上带来的收益远超这点投入。下次当你构建一个LangGraph应用时不妨先从stream_events开始让它成为你理解自己作品运行过程的“第三只眼”。
