当前位置：首页 > news >正文

浅谈 LangGraph 子图流式执行（subgraphs=True/False）模式

news 2025/8/12 17:12:11

在构建复杂 AI 工作流时，单一的扁平化图结构往往无法优雅地组织业务逻辑。子图（Subgraph） 提供了一种将复杂任务模块化的方式，让我们可以在父图中封装一个“子流程”，并且通过 subgraphs=True 在流式输出中保留命名空间信息，清晰地追踪每个步骤的执行过程。

本文将通过一个完整的可运行示例，演示：

如何构建父图与子图
如何在不同 stream_mode 下结合 subgraphs=True 获取输出
如何在实际项目中应用这一模式

1. 为什么要用 `subgraphs=True`？

在复杂 AI 应用（如 ReAct 推理 + 工具调用 + 多智能体协作）中，经常需要嵌套执行多个独立的流程，例如：

主流程负责总体调度
子流程负责特定任务（如信息检索、数据分析）
子流程内部可能还有多步执行

如果不使用 subgraphs=True，流式输出中会丢失子流程的上下文来源，很难定位是哪个子图节点产生的输出。而启用 subgraphs=True 后，每条流式数据都会带上命名空间（namespace），让调试与可视化更直观。

2. 示例代码

下面的例子构建了这样一个流程：

父图：refine_topic → research(子图) → generate_joke子图 research：research_topic → strengthen_topic

运行时，我们会对比不同 stream_mode 下的输出效果。

# -*- coding: utf-8 -*-
"""
LangGraph 子图(stream with subgraphs=True) 演示
- 父图节点：refine_topic -> research(subgraph) -> generate_joke
- 子图(research)内部：research_topic -> strengthen_topic
- 对比不同 stream_mode 下，带 subgraphs=True 的输出（含命名空间）
"""from typing import TypedDict, List
from langgraph.graph import StateGraph, START, END# ---------- 1) 定义父图 / 子图的 State ----------
class ParentState(TypedDict, total=False):topic: strjoke: strfacts: List[str]  # 子图会写入class ResearchState(TypedDict, total=False):topic: strfacts: List[str]# ---------- 2) 子图：research ----------
def research_topic(state: ResearchState):t = state.get("topic") or ""# 这里模拟“检索结果”return {"facts": [f"{t} fact A", f"{t} fact B"]}def strengthen_topic(state: ResearchState):# 做个简单“加工”以便父图后续可见return {"topic": (state.get("topic") or "") + " (researched)"}sub_builder = StateGraph(ResearchState)
sub_builder.add_node("research_topic", research_topic)
sub_builder.add_node("strengthen_topic", strengthen_topic)
sub_builder.add_edge(START, "research_topic")
sub_builder.add_edge("research_topic", "strengthen_topic")
sub_builder.add_edge("strengthen_topic", END)
subgraph = sub_builder.compile()  # ← 编译子图（稍后作为父图的一个节点）# ---------- 3) 父图 ----------
def refine_topic(state: ParentState):# 初步润色主题return {"topic": (state.get("topic") or "") + " and cats"}def generate_joke(state: ParentState):facts = state.get("facts") or []facts_part = ", ".join(facts[:2]) if facts else "no facts"return {"joke": f"This is a joke about {state.get('topic', '')} (with {facts_part})."}parent_builder = StateGraph(ParentState)
parent_builder.add_node("refine_topic", refine_topic)
parent_builder.add_node("research", subgraph)  # ← 直接把“已编译子图”当作一个节点
parent_builder.add_node("generate_joke", generate_joke)
parent_builder.add_edge(START, "refine_topic")
parent_builder.add_edge("refine_topic", "research")
parent_builder.add_edge("research", "generate_joke")
parent_builder.add_edge("generate_joke", END)
graph = parent_builder.compile()# ---------- 4) 打印辅助 ----------
def ns_str(ns):# 命名空间格式化try:return " / ".join(ns) if ns else "<root>"except Exception:return str(ns)def demo_stream(mode, subgraphs=True):print(f"\n=== stream_mode={mode!r}, subgraphs={subgraphs} ===")stream_input = {"topic": "ice cream", "joke": "", "facts": []}  # 可根据需求输入it = graph.stream(stream_input, {}, stream_mode=mode, subgraphs=subgraphs)for item in it:# print(item)if subgraphs:# 可能是 (ns, chunk) / (ns, (mode, chunk)) / (ns, mode, chunk)if not isinstance(item, tuple):print("[WARN] Unexpected item:", item)continueif len(item) == 2:ns, payload = item# 多模式的嵌套二元组：payload = (mode, chunk)if isinstance(payload, tuple) and len(payload) == 2 and payload[0] in {"updates", "values", "messages"}:mode_name, chunk = payloadprint(f"[ns={ns_str(ns)}] {mode_name}: {chunk}")else:# 单一模式：payload 就是 chunkprint(f"[ns={ns_str(ns)}] {payload}")elif len(item) == 3:# 三元组模式：(ns, mode, chunk)ns, mode_name, chunk = itemif mode_name in {"updates", "values", "messages"}:print(f"[ns={ns_str(ns)}] {mode_name}: {chunk}")else:print(f"[ns={ns_str(ns)}] ??? {mode_name}: {chunk}")else:print("[WARN] Unexpected tuple shape:", item)else:# subgraphs=False 时# 可能是 chunk（单模式）或 (mode, chunk)（多模式）if isinstance(item, tuple) and len(item) == 2 and item[0] in {"updates", "values", "messages"}:mode_name, chunk = itemprint(f"{mode_name}: {chunk}")else:print(item)def demo_invoke():print("\n=== invoke 最终结果 ===")res = graph.invoke({"topic": "ice cream", "joke": "", "facts": []}, {})print(res)if __name__ == "__main__":# 只看完整 statedemo_stream("values", subgraphs=True)# 只看增量（每个节点返回值），会带上是哪个节点demo_stream("updates", subgraphs=True)# 同时订阅 updates + values（返回 (mode, chunk)），也带命名空间demo_stream(["updates", "values"], subgraphs=True)# 对照：不带子图命名空间（可选）demo_stream("updates", subgraphs=False)# 最终一次性跑完整图demo_invoke()

3. 输出结果

=== stream_mode='values', subgraphs=True ===
[ns=<root>] {'topic': 'ice cream', 'joke': '', 'facts': []}
[ns=<root>] {'topic': 'ice cream and cats', 'joke': '', 'facts': []}
[ns=research:bc142b2d-56d9-a880-7fef-d634881b6f26] {'topic': 'ice cream and cats', 'facts': []}
[ns=research:bc142b2d-56d9-a880-7fef-d634881b6f26] {'topic': 'ice cream and cats', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}
[ns=research:bc142b2d-56d9-a880-7fef-d634881b6f26] {'topic': 'ice cream and cats (researched)', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}
[ns=<root>] {'topic': 'ice cream and cats (researched)', 'joke': '', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}
[ns=<root>] {'topic': 'ice cream and cats (researched)', 'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}=== stream_mode='updates', subgraphs=True ===
[ns=<root>] {'refine_topic': {'topic': 'ice cream and cats'}}
[ns=research:b04ed7c3-8c84-5be2-6a2c-167d725940cd] {'research_topic': {'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}}
[ns=research:b04ed7c3-8c84-5be2-6a2c-167d725940cd] {'strengthen_topic': {'topic': 'ice cream and cats (researched)'}}
[ns=<root>] {'research': {'topic': 'ice cream and cats (researched)', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}}
[ns=<root>] {'generate_joke': {'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).'}}=== stream_mode=['updates', 'values'], subgraphs=True ===
[ns=<root>] values: {'topic': 'ice cream', 'joke': '', 'facts': []}
[ns=<root>] updates: {'refine_topic': {'topic': 'ice cream and cats'}}
[ns=<root>] values: {'topic': 'ice cream and cats', 'joke': '', 'facts': []}
[ns=research:a7d3b2f3-2f72-acca-c640-30fc13a8c1a0] values: {'topic': 'ice cream and cats', 'facts': []}
[ns=research:a7d3b2f3-2f72-acca-c640-30fc13a8c1a0] updates: {'research_topic': {'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}}
[ns=research:a7d3b2f3-2f72-acca-c640-30fc13a8c1a0] values: {'topic': 'ice cream and cats', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}
[ns=research:a7d3b2f3-2f72-acca-c640-30fc13a8c1a0] updates: {'strengthen_topic': {'topic': 'ice cream and cats (researched)'}}
[ns=research:a7d3b2f3-2f72-acca-c640-30fc13a8c1a0] values: {'topic': 'ice cream and cats (researched)', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}
[ns=<root>] updates: {'research': {'topic': 'ice cream and cats (researched)', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}}
[ns=<root>] values: {'topic': 'ice cream and cats (researched)', 'joke': '', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}
[ns=<root>] updates: {'generate_joke': {'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).'}}
[ns=<root>] values: {'topic': 'ice cream and cats (researched)', 'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}=== stream_mode='updates', subgraphs=False ===
{'refine_topic': {'topic': 'ice cream and cats'}}
{'research': {'topic': 'ice cream and cats (researched)', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}}
{'generate_joke': {'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).'}}=== invoke 最终结果 ===
{'topic': 'ice cream and cats (researched)', 'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}

4. 输出对比

以下面的输入为例：

{"topic": "ice cream", "joke": "", "facts": []}

4.1 `stream_mode="values", subgraphs=True`

输出完整 State，带命名空间：

[ns=<root>] {'topic': 'ice cream', 'joke': '', 'facts': []}
[ns=<root>] {'topic': 'ice cream and cats', 'joke': '', 'facts': []}
[ns=research:UUID] {'topic': 'ice cream and cats', 'facts': []}
[ns=research:UUID] {'topic': 'ice cream and cats', 'facts': ['fact A', 'fact B']}
...

4.2 `stream_mode="updates", subgraphs=True`

只输出节点增量结果，带命名空间：

[ns=<root>] {'refine_topic': {'topic': 'ice cream and cats'}}
[ns=research:UUID] {'research_topic': {'facts': ['fact A', 'fact B']}}
[ns=research:UUID] {'strengthen_topic': {'topic': '...'}}
...

4.3 `stream_mode=["updates", "values"], subgraphs=True`

同时输出增量 + 全量，方便调试：

[ns=<root>] values: {...}
[ns=<root>] updates: {...}
[ns=research:UUID] values: {...}
[ns=research:UUID] updates: {...}
...

4.4 `stream_mode="updates", subgraphs=False`

关闭子图命名空间，输出更简洁，但丢失上下文信息：

{'refine_topic': {...}}
{'research': {...}}
{'generate_joke': {...}}

5、把subgraphs全设置为False

if __name__ == "__main__":# 只看完整 statedemo_stream("values", subgraphs=False)# 只看增量（每个节点返回值），会带上是哪个节点demo_stream("updates", subgraphs=False)# 同时订阅 updates + values（返回 (mode, chunk)），也带命名空间demo_stream(["updates", "values"], subgraphs=False)# 对照：不带子图命名空间（可选）demo_stream("updates", subgraphs=False)# 最终一次性跑完整图demo_invoke()

运行结果如下：

=== stream_mode='values', subgraphs=False ===
{'topic': 'ice cream', 'joke': '', 'facts': []}
{'topic': 'ice cream and cats', 'joke': '', 'facts': []}
{'topic': 'ice cream and cats (researched)', 'joke': '', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}
{'topic': 'ice cream and cats (researched)', 'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}=== stream_mode='updates', subgraphs=False ===
{'refine_topic': {'topic': 'ice cream and cats'}}
{'research': {'topic': 'ice cream and cats (researched)', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}}
{'generate_joke': {'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).'}}=== stream_mode=['updates', 'values'], subgraphs=False ===
values: {'topic': 'ice cream', 'joke': '', 'facts': []}
updates: {'refine_topic': {'topic': 'ice cream and cats'}}
values: {'topic': 'ice cream and cats', 'joke': '', 'facts': []}
updates: {'research': {'topic': 'ice cream and cats (researched)', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}}
values: {'topic': 'ice cream and cats (researched)', 'joke': '', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}
updates: {'generate_joke': {'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).'}}
values: {'topic': 'ice cream and cats (researched)', 'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}=== stream_mode='updates', subgraphs=False ===
{'refine_topic': {'topic': 'ice cream and cats'}}
{'research': {'topic': 'ice cream and cats (researched)', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}}
{'generate_joke': {'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).'}}=== invoke 最终结果 ===
{'topic': 'ice cream and cats (researched)', 'joke': 'This is a joke about ice cream and cats (researched) (with ice cream and cats fact A, ice cream and cats fact B).', 'facts': ['ice cream and cats fact A', 'ice cream and cats fact B']}

当把 subgraphs 统一设为 False 时，LangGraph 的流式输出“折叠”子图事件：所有来自子图的中间事件不再带命名空间逐条上浮，而是被汇总为父图节点的单次更新。

1) `stream_mode="values"`：完整状态但无命名空间

{'topic': 'ice cream', 'joke': '', 'facts': []}                          # 初始 state
{'topic': 'ice cream and cats', 'joke': '', 'facts': []}                 # refine_topic 后
{'topic': 'ice cream and cats (researched)', 'joke': '', 'facts': [...]} # research 子图结束后把结果合并进父 state
{'topic': '... (researched)', 'joke': 'This is a joke ...', 'facts': [...]} # generate_joke 后（最终）

我们仍能看到每一步父图层面的完整 state 快照；
子图内部（research_topic、strengthen_topic）的中间状态不会单独出现，只在“research 子图执行完毕”那一刻，以父 state 的整体变化体现（topic 被加后缀、facts 被填充）。

2) `stream_mode="updates"`：只见父图节点的增量

{'refine_topic': {'topic': 'ice cream and cats'}}
{'research': {'topic': '... (researched)', 'facts': ['...', '...']}}
{'generate_joke': {'joke': 'This is a joke ...'}}

只输出父图节点的增量：refine_topic、research、generate_joke；
整个 research 子图被视为一个“黑盒节点”，其内部两个子节点的增量不会单独上报；
这意味着：研究子图内部进度不可见，只在它把结果“写回父图”时一次性体现。

3) `stream_mode=["updates","values"]`：交替出现的父层增量与父层全量

values: {...初始全量...}
updates: {'refine_topic': {...}}
values: {...refine_topic 后全量...}
updates: {'research': {...}}             # 子图结果一次性上浮
values: {...research 后全量...}
updates: {'generate_joke': {...}}
values: {...最终全量...}

仍然看不到子图内部的逐步事件（没有命名空间，自然也没有内部节点名）；
能对齐每一步：先看到父层增量，再看到对应的全量快照，便于调试 UI 刷新逻辑。

4) 再次调用 `updates` 的重复块

代码里又调用了一次 demo_stream("updates", subgraphs=False)，所以末尾出现了同样的三条 updates 输出，这不是 LangGraph 的重复发送，而仅仅是再次跑了一次相同的演示函数。

5) `invoke` 一次性结果

{'topic': '... (researched)', 'joke': 'This is a joke ...', 'facts': [...]}

与流式一致，最终 state 会包含子图加工后的 topic 和 facts。

6) `subgraphs=False` 的行为与取舍

子图的内部事件被折叠：对外表现为父图中该子图节点的一次更新（research: {...}）。
values 模式只展示父层 state 的连续快照；updates 模式只展示父层节点的增量；混合模式交替展示二者。
看不到子图内部执行细节、中间产物与时序。

优点

输出更简洁、带宽/日志量更小；
对调用方更“稳态”：只关心父层抽象，不被子图内部实现细节干扰。

限制 / 何时不适合

若需要调试子图内部、做实时可视化或对长耗时子图观察中途进度，subgraphs=False 信息不够；
建议在开发/排障阶段使用 subgraphs=True 获取命名空间与内部节点级事件。

折中方案

生产环境保持 subgraphs=False（干净、轻量）；
关键子图内部在必要步骤主动把关键中间结果上浮到父 state（例如在子图中显式写入父 state 某个字段），以便仍能看到有限的过程信息；
或者仅在调试开关开启时切换为 subgraphs=True。

总之，把 subgraphs=False 看成“子图当成一个黑盒节点”：它只在入口和出口影响父图，内部怎么走不外露；想看内部执行，就用 subgraphs=True。

6. 最佳实践与应用场景

6.1 适用场景

多智能体调度：父图调度，子图执行具体 Agent 的任务
模块化工作流：将复杂任务分解为可维护的子流程
可视化与调试：跟踪每个节点的执行情况
流式响应：实时将父图与子图的中间结果推送给客户端

6.2 建议

子图封装明确的职责，不要让一个子图做多种类型的任务
在调试阶段开启 subgraphs=True，方便排查问题
在生产环境根据需求选择：
- 保留命名空间（方便日志与监控）
- 去掉命名空间（减小消息体积）

7. 总结

subgraphs=True 是 LangGraph 在复杂工作流场景下的一个重要特性。它不仅能保持嵌套执行的上下文可追踪，还能与 stream_mode 灵活配合，实现实时可观测的执行流。

在结合 ReAct 推理、MCP 工具调用 或 A2A 多智能体协议 时，这种模式可以显著提升可维护性与可调试性，它也逐渐成为了复杂 AI 系统架构中的观测利器。

查看全文

http://www.dtcms.com/a/326850.html

[鹧鸪云]光伏AI设计平台解锁电站开发新范式

Kubernetes生产环境健康检查自动化指南

Centos8系统在安装Git包时，报错：“没有任何匹配: git”

【ros-humble】4.C++写法巡场海龟（服务通讯）

搭建纯竞拍商城的核心技术要点与实施指南

4-下一代防火墙组网方案

[Element-plus]动态设置组件的语言

GPT-oss：OpenAI再次开源新模型，技术报告解读

【无标题】matplotlib与seaborn数据库

基于FPGA的热电偶测温数据采集系统，替代NI的产品（二）总体设计方案

嵌入式硬件中AI硬件设计方法与技巧

java内部类-匿名内部类

编程技术杂谈4.0

Dify入门指南（2）：5 分钟部署 Dify：云服务 vs 本地 Docker

做调度作业提交过程简单介绍一下

第二十九天（文件io）

Android视频编辑方案测评：轻量化剪辑工具的性能表现

基于51单片机红外遥控定时开关智能家电插座设计

golang 基础案例_02

算法知识笔记

学习日志31 python

【C++】STL——priority_queue的使用与底层模拟实现

查看 php 可用版本

Nestjs框架: RBAC基于角色的权限控制模型初探

STM32TIM定时器

请求报文和响应报文（详细讲解）

Wed前端第二次作业

C语言增删查改实战：高效管理顺序表

docker安装searxng

monorepo架构设计方案

1. 为什么要用 subgraphs=True？

2. 示例代码

3. 输出结果

4. 输出对比

4.1 stream_mode="values", subgraphs=True

4.2 stream_mode="updates", subgraphs=True

4.3 stream_mode=["updates", "values"], subgraphs=True

4.4 stream_mode="updates", subgraphs=False

5、把subgraphs全设置为False

1) stream_mode="values"：完整状态但无命名空间

2) stream_mode="updates"：只见父图节点的增量

3) stream_mode=["updates","values"]：交替出现的父层增量与父层全量

4) 再次调用 updates 的重复块

5) invoke 一次性结果

6) subgraphs=False 的行为与取舍