在构建基于大语言模型LLM的应用程序时响应速度和用户体验至关重要。流式输出Streaming技术允许应用程序在 LLM 生成文本的每个 Token 完成后立即将其呈现给用户而不是等待整个响应完成。LangGraph作为构建有状态、多参与者 LLM 应用的强大框架对流式输出提供了全面且深入的支持。本文将详细介绍 LangGraph 的流式输出机制涵盖其核心概念、API 使用方法并深入探讨如“流式冒泡”等高级应用场景。引言为什么流式输出如此重要想象一下你正在使用一个 AI 助手来撰写一封长邮件或分析一份复杂的文档。如果 AI 在几秒钟内“沉默不语”然后一次性吐出整个结果体验会很糟糕。你会怀疑它是否在正常工作并感到长时间的等待。流式输出解决了这个问题。它就像一个打字员在键盘上敲击每个字后立即显示在屏幕上。这带来了以下好处更好的用户体验 (UX)用户可以即时看到 AI 的“思考”和输出过程减少了等待的焦虑感。更快的感知响应速度即使用户获得完整结果的时间没有变化但看到第一个字符的时间大大缩短让系统感觉更“快”。实时反馈与交互流式输出是构建真正交互式应用的基础。例如用户可以在 AI 生成过程中打断它或者根据已生成的内容进行后续操作。LangGraph 的优势在于它不仅能流式传输 LLM 生成的单个 Token还能流式传输整个图的状态更新这对于构建复杂的、多步骤的智能体应用来说是无价的。LangGraph 流式输出的核心概念在深入了解 API 之前我们需要区分 LangGraph 中两种不同的流式输出维度。Token 级流式输出这是最常见也最直观的流式方式。它指的是从 LLM 模型如 OpenAI 的 GPT-4、Anthropic 的 Claude中逐个 Token 地获取输出。特点细粒度逐字符/逐词。适用场景聊天界面、代码生成、文本创作等任何需要实时显示生成内容的场景。LangGraph 中的实现LangGraph 通过astream_eventsAPI可以捕获来自图中任何节点的 LLM 模型生成的on_chat_model_stream事件。状态更新流式输出LangGraph 的核心是一个有向图每个节点执行后会更新图的共享状态。状态更新流式输出是指在图执行的每个步骤或每个节点完成后将状态的变化流式传输出来。特点粗粒度以“步骤”或“节点”为单位。适用场景调试和监控实时查看 Agent 的决策过程、使用了哪个工具、工具的输入输出是什么。复杂工作流在一个多步骤的分析流程中每完成一个分析步骤就向用户报告进度和中间结果。LangGraph 中的实现主要通过astreamAPI 实现它可以配置为流式传输updates增量更新或values当前完整状态。两者的差异与结合特性Token 级流式输出状态更新流式输出数据单元单个 Token节点的输出/状态的变化实时性极高中等取决于节点执行时间信息量仅包含生成的文本内容包含整个图的状态、工具调用结果、决策逻辑等主要用途终端用户体验 (UX)开发者调试、复杂工作流状态展示关键点LangGraph 的强大之处在于你可以同时使用这两种流式输出。例如在一个聊天应用中你可以用astream_events来流式显示 LLM 正在生成的文本为了 UX同时用astream来在后台监控 Agent 的整个执行流程为了调试和日志记录。LangGraph 流式输出 APILangGraph 提供了几个核心方法来实现流式输出其中astream_events是最新、最强大的。astream_events: 最全面的事件流astream_events是 LangGraph 0.2 版本后引入的、推荐的流式事件 API。它提供了一个统一的、基于生成器的接口可以访问图执行过程中发生的所有事件。方法签名async def astream_events(input, config, versionv1, *, interrupt_beforeNone, interrupt_afterNone)核心参数input: 图的初始输入。config: 一个包含配置信息的字典其中最重要的是configurable字段可以用来传递thread_id等信息。version: 协议版本当前推荐使用v1。返回值一个异步生成器它会按顺序产生一系列事件对象。事件结构每个事件都是一个字典包含以下关键字段event: 事件类型例如on_chat_model_start,on_chat_model_stream,on_tool_start,on_tool_end。name: 触发事件的组件名称如节点名称、模型名称。run_id: 当前运行的唯一 ID用于追踪一次完整的图执行。tags: 与事件关联的标签可以用来过滤事件。metadata: 事件的元数据包含诸如langgraph_step当前步骤数、langgraph_node当前节点名等信息。data: 事件的具体数据例如on_chat_model_stream事件中data包含一个chunk对象其中就有生成的 Token。astream: 简化的状态更新流astream是一个更高级的 API专注于流式传输图的状态更新。它比astream_events更简单当你只关心状态如何变化而不关心底层的 Token 或工具调用细节时它非常有用。方法签名async def astream(input, config, *, stream_modeupdates, interrupt_beforeNone, interrupt_afterNone)核心参数stream_mode这是最重要的参数它决定了流式输出的内容。updates(默认)流式传输每个节点执行后产生的状态更新。这是一个字典只包含被节点修改的状态键值对。values流式传输每个节点执行后的完整状态。这是一个包含当前所有状态键值的字典。debug流式传输用于调试的内部信息。返回值一个异步生成器产生(node_name, state_update_or_value)元组。stream:astream的同步版本如果你在同步环境中工作可以使用stream它是astream的同步包装器用法和参数完全一致。代码实战构建一个支持流式输出的图让我们通过一个简单的例子来演示如何使用这些 API。我们将构建一个简单的两节点图一个节点生成一个故事大纲另一个节点根据大纲扩展成一个完整的故事。步骤 1: 定义状态和模型首先我们定义图的共享状态并初始化一个 ChatModel。importoperatorfromtypingimportAnnotated,List,TypedDictfromlangchain_core.messagesimportBaseMessage,HumanMessagefromlangchain_openaiimportChatOpenAIfromlanggraph.graphimportStateGraph,END# 1. 定义状态classAgentState(TypedDict):messages:Annotated[List[BaseMessage],operator.add]# 消息列表使用add操作符合并# 2. 初始化模型llmChatOpenAI(modelgpt-4o-mini,temperature0)步骤 2: 定义节点和边现在我们定义两个节点函数并构建图的结构。# 3. 定义节点函数defoutline_node(state:AgentState):第一个节点生成故事大纲print(---outline_node 正在执行---)messages[HumanMessage(content请为一个关于AI的故事生成一个简短的大纲。)]responsellm.invoke(messages)return{messages:[response]}# 返回一个字典更新状态defstory_node(state:AgentState):第二个节点根据大纲扩展成完整故事print(---story_node 正在执行---)# 获取上一个节点的大纲outlinestate[messages][-1].content messages[HumanMessage(contentf请根据以下大纲写一个完整的AI故事\n\n{outline})]responsellm.invoke(messages)return{messages:[response]}# 4. 构建图workflowStateGraph(AgentState)# 添加节点workflow.add_node(outline_agent,outline_node)workflow.add_node(story_agent,story_node)# 设置边workflow.set_entry_point(outline_agent)workflow.add_edge(outline_agent,story_agent)workflow.add_edge(story_agent,END)# 5. 编译图appworkflow.compile()步骤 3: 编译图并实现流式输出现在我们使用astream和astream_events来流式执行这个图。使用astream流式传输状态更新importasyncioasyncdefmain():print( 使用 astream (stream_modeupdates) )# 使用 updates 模式只流式传输每个节点的状态更新asyncforeventinapp.astream({messages:[]},stream_modeupdates):fornode_name,node_outputinevent.items():print(f\n[节点{node_name}完成执行产生更新])print(f更新内容:{node_output})print(\n 使用 astream (stream_modevalues) )# 使用 values 模式流式传输每个节点执行后的完整状态asyncforeventinapp.astream({messages:[]},stream_modevalues):print(f\n[当前步骤完成完整状态中的消息数量:{len(event[messages])}])# print(event) # 如果取消注释将打印整个状态if__name____main__:asyncio.run(main())预期输出简化版 使用 astream (stream_modeupdates) [节点 outline_agent 完成执行产生更新] 更新内容: {messages: [AIMessage(content一个关于AI的觉醒与自我探索的故事...)]} [节点 story_agent 完成执行产生更新] 更新内容: {messages: [AIMessage(content在未来的某一天一个名为“艾瑞安”的AI...)]} 使用 astream (stream_modevalues) [当前步骤完成完整状态中的消息数量: 1] [当前步骤完成完整状态中的消息数量: 2]使用astream_events流式传输 Token这是实现聊天界面的关键。asyncdefprint_token_stream():print(\n 使用 astream_events 流式传输 Token )# 使用 astream_events 来获取所有事件asyncforeventinapp.astream_events({messages:[]},versionv1):kindevent[event]# 过滤出 chat_model 的流式输出事件ifkindon_chat_model_stream:contentevent[data][chunk].contentifcontent:# 实时打印每个 Token用 end 防止换行print(content,end,flushTrue)# 也可以选择性地打印工具调用等事件elifkindon_tool_start:print(f\n[工具调用开始]:{event[name]})elifkindon_tool_end:print(f\n[工具调用结束]:{event[name]})if__name____main__:asyncio.run(print_token_stream())预期输出模拟 使用 astream_events 流式传输 Token 一 个 关 于 AI 的 觉 醒 ... [这里会逐字打印出完整的故事] ... 。进阶场景流式冒泡机制在一个复杂的图中你可能有多个分支并行执行或者一个节点的输出需要被另一个节点处理。在这些情况下如何优雅地将不同来源的流式输出呈现给用户是一个挑战。这就是“流式冒泡”机制要解决的问题。什么是流式冒泡“流式冒泡”不是一个 LangGraph 的官方术语而是一种设计模式。它指的是将来自图中不同节点、不同层级的流式输出通常是 Token有序地、聚合地“冒泡”到顶层最终呈现给用户。想象一个气泡从水底升起它可能来自不同的源头但最终都会冒出水面。在我们的场景中“水面”就是用户界面。如何实现流式冒泡实现流式冒泡的核心思路是使用astream_events作为唯一数据源它能捕获所有节点的所有事件。在应用层进行聚合和排序你需要编写逻辑来处理这些事件流。使用元数据来区分来源利用事件中的metadata字段如langgraph_node来知道 Token 来自哪个节点。管理输出顺序这是最棘手的部分。如果节点 A 和节点 B 并行执行它们的 Token 会交替到达。你需要决定是按节点分组先显示完节点 A 的所有输出再显示节点 B 的。按时间顺序混合就像两个人同时说话交替显示它们的输出。优先级冒泡某个节点的输出如最终答案优先级更高优先显示。代码示例实现多分支流式冒泡让我们修改之前的图让它并行生成两个不同风格的故事开头然后将这些流式输出“冒泡”到控制台。fromlanggraph.graphimportStateGraph,END# 1. 定义状态 (与之前相同)classAgentState(TypedDict):messages:Annotated[List[BaseMessage],operator.add]llmChatOpenAI(modelgpt-4o-mini,temperature0)# 2. 定义两个并行的故事生成节点defscifi_story_node(state:AgentState):print(---scifi_story_node 正在执行---)messages[HumanMessage(content写一个科幻风格的故事开头大约50字。)]responsellm.invoke(messages)return{messages:[response]}deffairytale_story_node(state:AgentState):print(---fairytale_story_node 正在执行---)messages[HumanMessage(content写一个童话风格的故事开头大约50字。)]responsellm.invoke(messages)return{messages:[response]}# 3. 构建一个并行图workflowStateGraph(AgentState)workflow.add_node(scifi_agent,scifi_story_node)workflow.add_node(fairytale_agent,fairytale_story_node)workflow.set_entry_point(scifi_agent)# 让两个节点并行执行。LangGraph 会从入口点开始同时执行所有没有依赖的节点。# 在这个简单例子中我们需要一个虚拟的入口来触发并行。# 更好的做法是使用 Send API 或者一个条件边。为了简单起见我们用条件边。fromlanggraph.graphimportSTARTdefroute_to_parallel(state):# 这个函数只是简单地触发两个节点return[scifi_agent,fairytale_agent]# 更简单的并行触发方式是直接在 START 后连接两个节点workflow.add_edge(START,scifi_agent)workflow.add_edge(START,fairytale_agent)workflow.add_edge(scifi_agent,END)workflow.add_edge(fairytale_agent,END)appworkflow.compile()asyncdefstream_with_bubbling():print(\n 并行节点的流式冒泡 )# 用于存储每个节点的输出字符串node_outputs{scifi_agent:,fairytale_agent:}asyncforeventinapp.astream_events({messages:[]},versionv1):kindevent[event]ifkindon_chat_model_stream:node_nameevent[metadata][langgraph_node]contentevent[data][chunk].contentifcontentandnode_nameinnode_outputs:# 将 Token 追加到对应节点的字符串中node_outputs[node_name]content# --- 冒泡逻辑 ---# 这里我们选择“按时间顺序混合”的冒泡方式print(f[{node_name}]:{content},end,flushTrue)print(\n\n 最终聚合结果 )fornode,textinnode_outputs.items():print(f{node}的完整输出:{text})if__name____main__:asyncio.run(stream_with_bubbling())预期输出模拟 并行节点的流式冒泡 [scifi_agent]: 在 [fairytale_agent]: 从 [scifi_agent]: 2 [fairytale_agent]: 前 [scifi_agent]: 0 [fairytale_agent]: [scifi_agent]: 4 [fairytale_agent]: 有 [scifi_agent]: 2 [fairytale_agent]: 一 [scifi_agent]: 年 [fairytale_agent]: 个... ... [scifi_agent]: 星际飞船。 [fairytale_agent]: 魔法森林。 最终聚合结果 scifi_agent 的完整输出: 在2042年星际飞船... fairytale_agent 的完整输出: 从前有一个魔法森林...这个例子展示了如何将来自两个并行节点的流式输出交错地“冒泡”到用户界面。你可以根据需要修改冒泡逻辑例如先收集完一个节点的所有输出再显示下一个或者根据用户的选择优先显示某个节点的输出。实践技巧始终使用异步 API (astream,astream_events)流式输出是一个 I/O 密集型操作。在 Python 中使用async/await可以显著提高性能尤其是在处理多个并发流时。理解stream_mode的选择如果你的目标是构建一个聊天 UI请使用astream_events来获取 Token。如果你的目标是构建一个进度展示或调试工具astream配合stream_modevalues是更好的选择。善用tags和metadata在创建图或调用 LLM 时可以添加自定义tags。在astream_events中你可以根据这些tags来过滤事件只处理你关心的部分。处理流式输出的中断LangGraph 支持在图的特定节点之前或之后中断执行 (interrupt_before,interrupt_after)。这对于需要人工介入的流程非常有用。在流式输出时你可以在用户界面上提供一个“停止”按钮当被点击时可以设置一个标志来中止流式生成循环。为不同的事件类型设计 UI不要只把所有 Token 混在一起显示。on_chat_model_stream: 显示为普通文本。on_tool_start: 可以显示为一个“正在使用工具 X…”的加载动画。on_tool_end: 可以显示工具调用的结果例如如果工具是搜索可以展示搜索结果的摘要。在生产环境中使用回调处理器astream_events非常适合调试和原型开发。但在生产环境中你可能会使用 LangChain 的回调处理器 (Callbacks)如StreamingStdOutCallbackHandler或自定义的回调处理器它们可以更紧密地集成到你的日志和监控系统中。LangGraph 正在快速发展其流式输出能力也在不断增强。未来的改进可能包括更精细的事件过滤提供更强大的 API 来订阅特定类型、特定节点或特定标签的事件而不是在客户端手动过滤。原生支持 Server-Sent Events (SSE) 和 WebSocketLangGraph 可能会提供更直接的集成使得将流式输出推送到前端变得更简单。与前端框架的深度集成可能会出现与 React, Vue 等前端框架更紧密绑定的库自动处理流的订阅、聚合和 UI 更新。总结流式输出是构建现代、响应式 LLM 应用的核心技术。LangGraph 通过astream_events和astream这两个强大而灵活的 API为开发者提供了对整个图执行过程的细粒度控制。本文从基础概念入手详细介绍了 Token 级和状态级流式输出的区别并通过代码示例展示了如何使用核心 API。我们还深入探讨了“流式冒泡”这一高级模式展示了如何在复杂的并行图中优雅地处理和呈现流式数据。掌握 LangGraph 的流式输出机制将帮助你打造出不仅智能而且快速、流畅、用户友好的 AI 应用。随着 LangGraph 的不断演进其流式能力必将变得更加强大和易用为下一代 AI 体验奠定基础。
