1. 项目概述：为什么流式输出是LangGraph的灵魂

如果你用过LangChain，大概率体验过那种“等待-等待-砰！”的完整响应返回模式。在构建复杂的AI工作流时，这种同步阻塞的体验，对于终端用户和开发者调试来说，都是一种煎熬。而LangGraph的Stream（流式输出）功能，正是为了解决这个核心痛点而生。它不仅仅是把一个大块的文本拆成小段吐出来那么简单，而是将整个工作流的执行过程和中间状态实时地、透明地暴露给你。

想象一下，你构建了一个包含“检索-分析-生成-审核”多步骤的智能客服流程。没有流式输出时，用户面对的是一个沉默的输入框，直到所有步骤跑完，答案才一次性出现。这不仅让用户感到焦虑（“它到底有没有在干活？”），一旦出错，你也很难定位问题到底卡在了“检索”还是“生成”环节。而有了流式输出，你可以看到：“正在检索相关知识库...”、“找到了3条相关文档”、“开始组织回答...”、“正在生成最终回复...”，最后才是完整的答案。这种可观测性和即时反馈，是构建可靠、可信AI应用的关键。

在LangGraph中，stream不仅仅是一个输出模式，它更是一种调试工具和交互设计范式。通过流式接口，我们可以实时捕获工作流中每个节点的输入输出、状态变更，甚至是分支决策的逻辑。这对于理解复杂图（Graph）的执行路径、优化性能瓶颈、以及向最终用户提供渐进式体验，都具有不可替代的价值。本次笔记，我们就深入LangGraph的流式世界，从基础用法到高级监控，彻底掌握这一核心特性。

2. 核心概念与流式输出原理拆解

在深入代码之前，我们必须厘清几个关键概念，这能帮你理解流式输出背后的设计哲学，而不仅仅是调用一个API。

2.1 LangGraph中的“流”是什么？

这里的“流”（Stream）是一个广义概念，它包含两个层面：

1. 数据流（Data Streaming）：这是最常见的形式，即LLM（大语言模型）生成文本时，以Token（词元）为单位的逐词输出。这依赖于底层LLM（如OpenAI的ChatCompletion接口）本身支持的流式响应。
2. 事件流（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(model="gpt-4o-mini", streaming=True) # 注意：此处streaming=True启用了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="", flush=True) print() # 换行 return {"poem": [HumanMessage(content=full_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_mode="values"): # 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(f"Token: {content}", end="", flush=True) 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_mode="values"参数是关键。它还有"updates"（只流式状态更新）和"messages"（专用于消息数组）等选项。对于调试，"values"模式最全面。但在生产环境面向用户时，你可能更关心"updates"或直接处理on_llm_stream来推送文字。

3.3 方法二：聚焦状态更新流

如果你只关心图状态（State）的变化，可以使用stream_mode="updates"。这过滤掉了大量的中间事件，只在你定义的节点函数返回时，推送状态发生了哪些改变。

async def stream_updates_example(): inputs = {"topic": "", "poem": [], "final_poem": ""} async for chunk in app.astream(inputs, stream_mode="updates"): # 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, version="v1"): 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="", flush=True) # 实时输出LLM生成内容 elif kind == "on_tool_end": print(f"\n[工具 {name} 执行完毕]: 输出 -> {data.get('output')}") elif kind == "on_chain_end": # 一个节点（链）运行结束 print(f"\n[节点 {name} 完成]")

stream_events方法的事件结构非常清晰，并且version="v1"参数保证了API的稳定性。它是我在开发和调试中最常使用的工具。

3.5 方法四：在普通函数中消费流（非异步）

如果你的环境不支持顶层async/await，可以使用stream方法（注意不是astream）在同步上下文中迭代。

inputs = {"topic": "", "poem": [], "final_poem": ""} for event in app.stream(inputs, stream_mode="values"): 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="", flush=True)

重要提示：在同步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, version="v1"): 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, websocket=None): self.websocket = websocket self.logger = logging.getLogger(__name__) async def handle_event(self, event): kind = event["event"] # 1. 结构化日志 self.logger.info(f"LangGraph 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, version="v1") 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, timeout=30, cancel_event=None): """ 带超时和取消机制的流式调用 Args: cancel_event: 一个asyncio.Event，用于从外部（如HTTP请求取消）通知中断。 """ try: # 使用wait_for设置总超时 async for event in asyncio.timeout(timeout, app.astream(inputs, stream_mode="updates")): # 检查外部取消信号 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 流式输出“不流”了？检查这三点

1. LLM客户端配置：确保你的ChatOpenAI、ChatAnthropic等客户端初始化时传入了streaming=True。这是Token级流式的基础。

   # 正确 llm = ChatOpenAI(model="gpt-4", streaming=True) # 错误：即使外层用了stream，LLM内部也不会流式生成Token。 llm = ChatOpenAI(model="gpt-4")

2. 节点函数内部是否阻塞：如果你的节点函数内部是调用llm.invoke()（同步阻塞），那么即使外层用app.stream，你也只能收到节点开始和结束的事件，看不到LLM生成Token的过程。应改用llm.stream()或在异步上下文中使用llm.ainvoke()。

3. stream_mode选择：如果你用了stream_mode="updates"，那么你只会收到状态更新事件，而不会收到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(checkpointer=memory) config = {"configurable": {"thread_id": "user_123"}} try: async for event in app.astream(input, config=config): # 处理事件 pass except Exception as e: print(f"工作流执行失败: {e}") # 利用检查点，可以获取失败时的状态，用于分析或恢复 checkpoint = await memory.aget(config) print(f"失败时的状态快照: {checkpoint['channel_values']}")

结合检查点（Checkpoint），你不仅能实现“断点续跑”，还能在流式执行出错时，精准定位到出错前的状态，极大方便了调试和错误恢复。

5.4 前端对接实战要点

将LangGraph的流式输出对接前端（如WebSocket）时，需要注意数据格式。

1. 定义清晰的事件协议：不要简单地把LangGraph的原始事件对象扔给前端。前端只关心业务状态。建议定义如下的简单协议：

   {"type": "status", "data": "正在检索..."} {"type": "token", "data": "今天"} {"type": "tool_result", "tool": "search", "data": "找到了10条结果"} {"type": "final_answer", "data": "综上所述，..."}

2. 处理连接中断：如前所述，前端页面关闭或刷新时，后端要及时捕获asyncio.CancelledError并停止流式生成，避免资源浪费。
3. 错误信息流式传递：如果某个节点执行失败，不要等到整个流结束才报错。可以通过自定义事件，立即向前端推送一个{"type": "error", "data": "..."}的消息。

流式输出不是一种炫技，而是构建现代AI应用的基础设施。它连接了后端的复杂逻辑与前端的即时体验，也架起了开发者与黑盒工作流之间的调试桥梁。从最初只是简单打印Token，到如今能精细控制整个图的执行脉络，LangGraph在可观测性上确实向前迈了一大步。我个人的体会是，在项目早期就接入流式输出，虽然增加了一些开发复杂度，但它在调试和用户体验上带来的收益，远超这点投入。下次当你构建一个LangGraph应用时，不妨先从stream_events开始，让它成为你理解自己作品运行过程的“第三只眼”。