一. Stream流式输出
流式输出可以说是LangGraph最具特色的核心功能之一。它让图的执行过程不再是一个“黑盒”——每一步的状态变化都能被实时捕捉和观察。那么,LangGraph具体提供了哪几种流式输出的方式?今天我们就逐一拆解分析。
1. updates 和 values 流式输出
先从最直观的两种模式说起:updates 和 values。
可以这样理解:updates 模式更像是“增量推送”,它只告诉你每一步结束后哪些状态字段发生了变化。比如,节点 refine_topic 运行完,只推送它更新过的 topic 字段;节点 generate_joke 运行完,则只推送 joke 字段。这种模式数据量小,非常适合只关心变化点的场景。
而 values 模式则不同,它会在每一步结束后,将当前整个状态的全部值都推送出来。因此你会看到,从只包含 topic,到 topic 被改写,再到最后 topic 和 joke 双双到位,完整的过程一目了然。
来看代码示例:
执行结果:
updates 模式输出:
{'refine_topic': {'topic': 'ice cream and cats'}}
{'generate_joke': {'joke': 'This is a joke about ice cream and cats'}}
values 模式输出:
{'topic': 'ice cream'}
{'topic': 'ice cream and cats'}
{'topic': 'ice cream and cats', 'joke': 'This is a joke about ice cream and cats'}
2. updates 和 values,以及 ["values", "updates"] 混合输出、debug 模式
单一模式能满足大部分需求,但有时我们希望在同一时间拿到不同维度的信息。LangGraph 也考虑到了这一点。
可以通过将 stream_mode 参数设置为一个列表,比如 ["values", "updates"],来同时获取两种模式的输出。这时,每次流出的数据会变成一个元组 (模式名称, 数据块)。对于调试阶段而言,这简直是利器——既能看清整体状态变化,又能追踪每一次具体更新的源头。
更进一步,还可以尝试 debug 模式。它会输出更加详细的调试信息,包括节点执行顺序、输入输出等。当然,这个模式可能需要一些额外配置,并非所有环境都直接可用。
代码示例中,通过一个“思考-回应-反思”的三节点图,展示了这些模式的实际用法:
执行结果:
3. messages 模式的流式输出:一个元组 (message_chunk, metadata)
前面聊的都是状态级别的流式输出,那如果要逐字流式输出LLM生成的内容呢?这时就要用到 messages 模式了。
它的工作原理是:图中任何一个调用了大语言模型(LLM)的节点,其输出的每一个令牌(token)都会被逐字捕获,并形成一个元组。这个元组包含两部分:
- message_chunk:LLM输出的令牌或消息片段,本质上就是模型生成的一个个字、一个个词。
- metadata:一个字典,包含了该令牌来自哪个图节点、是哪一次LLM调用等信息。这在大图中定位问题非常有用。
可以想象一下典型的“客服问答”场景:用户输入问题后,系统将查询传递给大模型,然后逐字返回答案。在 messages 模式下,这些返回的每个字都会被包装成元组,开发者可以逐帧捕获,实现类似“打字机”的流式效果。
代码示例中,通过一个简单的单节点图,直接调用大模型(如通义千问),并逐字打印输出:
执行结果:
4. 自定义(custom)流式输出:组合多种模式
有时候,我们希望从图节点内部发送一些自定义数据,比如进度更新、中间状态、某个关键变量的值等。这时候就需要用到 custom 模式了。
实现起来其实很简单:
- 在节点内部,通过
get_stream_writer()获取一个流写入器。 - 然后调用
writer(data),将任意结构的数据推送出去。数据可以是字典、字符串,甚至更复杂的结构。 - 在调用图的
.stream()时,将stream_mode设置为"custom"即可。当然,也可以与其他模式组合使用,比如["updates", "custom"]。但注意,组合中至少要包含"custom"才会触发自定义数据流。
例如,在一个处理查询的节点中,可以先推送“开始处理”,再推送“步骤1:分析查询内容”,再推送“步骤2:生成结果”,最后推送“完成”。整个流程的进度就能被前端实时捕获,带来极佳的交互体验。
代码示例中,通过一个带自定义流的节点,演示了如何发送进度更新和自定义键值对:
执行结果:
