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

在构建复杂 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: str
    joke: str
    facts: List[str]  # 子图会写入


class ResearchState(TypedDict, total=False):
    topic: str
    facts: 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)
                continue

            if 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 = payload
                    print(f"[ns={ns_str(ns)}] {mode_name}: {chunk}")
                else:
                    # 单一模式：payload 就是 chunk
                    print(f"[ns={ns_str(ns)}] {payload}")

            elif len(item) == 3:
                # 三元组模式：(ns, mode, chunk)
                ns, mode_name, chunk = item
                if 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 = item
                print(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__":
    # 只看完整 state
    demo_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__":
    # 只看完整 state
    demo_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 系统架构中的观测利器。

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

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` 的行为与取舍

6. 最佳实践与应用场景

6.1 适用场景

6.2 建议

7. 总结

网站公告

今日签到

热门文章

最新发布

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

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 的行为与取舍

6. 最佳实践与应用场景

6.1 适用场景

6.2 建议

7. 总结

网站公告

今日签到

热门文章

最新发布

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

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`

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

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

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

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

5) `invoke` 一次性结果

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