第 18 期 | Sub-graphs：将底层 Graph 浓缩为一个普通节点

各位 AI 架构师们，欢迎回到我们的《LangGraph 多智能体专家课》。我是你们的老朋友。

在前面的 17 期里，我们的「AI 万能内容创作机构 (AI Content Agency)」已经初具规模。我们有了运筹帷幄的 Planner（策划），有了苦哈哈查资料的 Researcher（研究员），还有了奋笔疾书的 Writer（主笔）和吹毛求疵的 Editor（编辑）。

但是，随着业务的深入，你有没有发现一个让人头皮发麻的问题？ 我们的主流程图（Main Graph）变得越来越像一坨意大利面了！

回想一下，上期我们为了让 Researcher 能够“搜索 -> 抓取网页 -> 总结 -> 评估信息质量（不行再搜）”，给它加了一大堆节点和条件边。结果呢？主图里密密麻麻全是 Researcher 的内部逻辑。Planner 和 Writer 躲在角落里瑟瑟发抖，整个架构的可读性和可维护性直线下降。

这就像是你作为一个公司的 CEO（主 Graph），你不需要、也不应该去管调研部门（Researcher）内部是怎么开会、怎么查资料、怎么吵架的。你只需要把“调研需求”扔给调研主管，然后等他把“调研报告”交给你就行了。

这就是我们今天要讲的终极杀器——Sub-graphs（子图）。

今天，我们要给系统做一次“外科手术级别的重构”：把 Researcher 复杂的内部工作流，折叠、浓缩成一个 Sub-graph，然后像调用一个普通 Node 一样，把它挂载到我们的主 Graph 上。

准备好你的咖啡，我们发车！

🎯 本期学习目标

1. 掌握 Sub-graph 的核心哲学：理解“图嵌套图”的底层逻辑，实现系统复杂度的降维打击。
2. 打通父子图的状态传递（State Mapping）：搞清楚 Main Graph 和 Sub-graph 之间的数据是怎么进去、又是怎么出来的（这是最容易翻车的地方）。
3. 完成 Agency 项目架构重构：将 Researcher 的“搜索-抓取-总结”循环剥离为独立的子图，让主流程回归“Planner -> Researcher -> Writer”的极简形态。
4. 掌握子图的独立调试技巧：学会如何在不启动整个系统的情况下，单独对子图进行单元测试。

📖 原理解析

在 LangGraph 中，任何一个被 compile() 编译过的 StateGraph，都可以直接作为另一个 StateGraph 的 Node 来使用。

这句话听起来简单，但威力无穷。它意味着你可以无限嵌套：公司包含部门，部门包含小组，小组包含个人。这种分形（Fractal）架构，是构建企业级复杂 Multi-Agent 系统的唯一解。

我们来看看重构前后的架构对比。

重构前的“面条图”架构（反面教材）

如果把所有逻辑都塞在一个图里，它看起来是这样的：Planner 规划完后，进入 Search 节点，然后判断是否需要 Scrape，再进入 Summarize，再判断资料够不够……整个主线被严重割裂。

重构后的“模块化”架构（本期目标）

我们把 Researcher 的内部逻辑封装成一个黑盒（Sub-graph）。

来看看我们今天的架构设计图：

graph TD
    %% 主图节点定义
    Start((START))
    Planner[Planner Node\n生成大纲与调研需求]
    Writer[Writer Node\n根据资料写初稿]
    Editor[Editor Node\n审核与润色]
    End((END))

    %% 子图定义（折叠在 Researcher 节点内）
    subgraph Researcher_SubGraph ["🔍 Researcher Node (Sub-graph 内部逻辑)"]
        direction TB
        R_Start((R_Start))
        Search[Web Search\n搜索关键词]
        Scrape[Web Scrape\n抓取网页内容]
        Summarize[Summarize\n提炼核心信息]
        Eval{足够写稿了吗?}
        R_End((R_End))

        R_Start --> Search
        Search --> Scrape
        Scrape --> Summarize
        Summarize --> Eval
        Eval -- "No (补充搜索)" --> Search
        Eval -- "Yes (完成调研)" --> R_End
    end

    %% 主图流转逻辑
    Start --> Planner
    Planner -- "传递调研需求" --> Researcher_SubGraph
    Researcher_SubGraph -- "返回调研报告" --> Writer
    Writer --> Editor
    Editor --> End

    %% 样式美化
    classDef mainNode fill:#2d3436,stroke:#74b9ff,stroke-width:2px,color:#fff;
    classDef subNode fill:#0984e3,stroke:#00cec9,stroke-width:2px,color:#fff;
    classDef subGraphBox fill:#dfe6e9,stroke:#b2bec3,stroke-width:2px,stroke-dasharray: 5 5,color:#2d3436;

    class Planner,Writer,Editor mainNode;
    class Search,Scrape,Summarize,Eval subNode;
    class Researcher_SubGraph subGraphBox;

核心难点：状态（State）的隔离与合并

当你把一个图作为节点放入另一个图时，最关键的问题是：数据怎么交互？

在 LangGraph 中，有两种常见的处理方式：

1. 共享状态（Shared State）：父图和子图使用完全一样的 TypedDict 或 Pydantic 模型。子图直接读写父图的状态。这种做法简单，但破坏了封装性。子图不应该知道父图里还有 draft_content（初稿内容）这种跟它无关的数据。
2. 状态映射（State Mapping / Wrapper）：父图和子图有各自独立的 State。我们在父图调用子图时，只把子图需要的数据传进去，子图运行完后，再把结果映射回父图的 State 中。（这是高级架构师的标配，也是我们今天实战要用的方法）

💻 实战代码演练

下面我们直接进入代码环节。为了让你能直接跑通，我使用了 Mock（模拟）函数来代替真实的 LLM 调用，重点展示 Graph 的路由和嵌套逻辑。

请仔细阅读代码中的双语注释，这是我 10 年踩坑换来的精华。

import operator
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, START, END

# =====================================================================
# 第一部分：定义状态 (States)
# =====================================================================

# 1. 子图状态 (Researcher State)
# 研究员只需要关心：要查什么（query）、查到了什么（raw_docs）、以及最后的总结（summary）
class ResearcherState(TypedDict):
    research_query: str
    raw_docs: Annotated[List[str], operator.add] # 使用 operator.add 自动追加列表
    research_summary: str
    iteration_count: int # 记录查了多少次，防止死循环

# 2. 主图状态 (Agency State)
# 主图是全局视角，包含策划案、调研报告、初稿等一切信息
class AgencyState(TypedDict):
    topic: str
    planner_outline: str
    # 注意：主图不需要 raw_docs，主图只想要研究员最终的 summary
    final_research_report: str 
    draft: str

# =====================================================================
# 第二部分：构建 Sub-graph (Researcher 的内部工作流)
# =====================================================================

def search_node(state: ResearcherState):
    print(f"  [Researcher-Search] 正在搜索引擎中检索: {state['research_query']}")
    # 模拟搜索结果
    new_doc = f"关于 {state['research_query']} 的网页资料片段..."
    return {"raw_docs": [new_doc], "iteration_count": state.get("iteration_count", 0) + 1}

def summarize_node(state: ResearcherState):
    print(f"  [Researcher-Summarize] 正在提炼 {len(state['raw_docs'])} 份资料...")
    # 模拟总结逻辑
    summary = f"【深度调研报告】基于 {len(state['raw_docs'])} 篇资料，核心结论如下..."
    return {"research_summary": summary}

def check_enough_data(state: ResearcherState):
    # 模拟判断逻辑：假设查了2次就觉得资料够了
    if state["iteration_count"] >= 2:
        print("  [Researcher-Eval] 资料已充足，结束调研。")
        return "sufficient"
    else:
        print("  [Researcher-Eval] 资料不够，继续深挖！")
        return "insufficient"

# 组装 Researcher 子图
researcher_builder = StateGraph(ResearcherState)
researcher_builder.add_node("search", search_node)
researcher_builder.add_node("summarize", summarize_node)

researcher_builder.add_edge(START, "search")
researcher_builder.add_edge("search", "summarize")
researcher_builder.add_conditional_edges(
    "summarize",
    check_enough_data,
    {
        "sufficient": END,
        "insufficient": "search" # 循环回去继续搜
    }
)

# 编译子图！现在它变成了一个可以被调用的 Runnable 对象
researcher_graph = researcher_builder.compile()

# =====================================================================
# 第三部分：构建 Main Graph (AI Content Agency 主流程)
# =====================================================================

def planner_node(state: AgencyState):
    print(f"\n[Planner] 收到主题: {state['topic']}。正在生成大纲和调研需求...")
    outline = f"《{state['topic']}》的大纲：1.背景 2.现状 3.趋势"
    return {"planner_outline": outline}

# 💡 核心魔法：Wrapper 函数 (状态映射器)
# 因为 AgencyState 和 ResearcherState 不一样，我们需要一个“翻译官”
def researcher_wrapper_node(state: AgencyState):
    print(f"\n[Main Graph] 唤醒 Researcher 部门，移交调研任务...")
    
    # 1. 从主状态提取信息，构造子图所需的初始状态
    initial_researcher_state = ResearcherState(
        research_query=f"深入挖掘: {state['topic']} ({state['planner_outline']})",
        raw_docs=[],
        research_summary="",
        iteration_count=0
    )
    
    # 2. 调用子图 (就像调用一个普通的 LLM 或 Tool 一样)
    # invoke() 会同步执行完子图的所有逻辑，直到子图到达 END
    final_researcher_state = researcher_graph.invoke(initial_researcher_state)
    
    print(f"[Main Graph] Researcher 部门工作完成，接收报告。")
    
    # 3. 将子图的产出，映射回主状态
    return {"final_research_report": final_researcher_state["research_summary"]}

def writer_node(state: AgencyState):
    print(f"\n[Writer] 根据调研报告开始码字...\n参考资料: {state['final_research_report']}")
    draft = f"这是一篇关于 {state['topic']} 的爆款文章初稿！"
    return {"draft": draft}

# 组装 Main Graph
agency_builder = StateGraph(AgencyState)
agency_builder.add_node("planner", planner_node)
agency_builder.add_node("researcher_dept", researcher_wrapper_node) # 挂载子图 Wrapper
agency_builder.add_node("writer", writer_node)

agency_builder.add_edge(START, "planner")
agency_builder.add_edge("planner", "researcher_dept")
agency_builder.add_edge("researcher_dept", "writer")
agency_builder.add_edge("writer", END)

# 编译主图
agency_graph = agency_builder.compile()

# =====================================================================
# 第四部分：运行测试
# =====================================================================

if __name__ == "__main__":
    print("🚀 启动 AI Content Agency 工作流...\n" + "="*40)
    
    initial_state = AgencyState(
        topic="2024年人形机器人行业发展趋势",
        planner_outline="",
        final_research_report="",
        draft=""
    )
    
    # 执行主图
    final_state = agency_graph.invoke(initial_state)
    
    print("\n" + "="*40 + "\n🎉 最终产出结果：")
    print(final_state["draft"])

运行结果模拟输出

当你运行这段代码时，你会看到极其清晰的层级结构：主图稳步推进，子图在内部疯狂内卷（循环），但主图对此毫不关心，只等结果。

🚀 启动 AI Content Agency 工作流...
========================================

[Planner] 收到主题: 2024年人形机器人行业发展趋势。正在生成大纲和调研需求...

[Main Graph] 唤醒 Researcher 部门，移交调研任务...
  [Researcher-Search] 正在搜索引擎中检索: 深入挖掘: 2024年人形机器人行业发展趋势 (《2024年人形机器人行业发展趋势》的大纲：1.背景 2.现状 3.趋势)
  [Researcher-Summarize] 正在提炼 1 份资料...
  [Researcher-Eval] 资料不够，继续深挖！
  [Researcher-Search] 正在搜索引擎中检索: 深入挖掘: 2024年人形机器人行业发展趋势 (《2024年人形机器人行业发展趋势》的大纲：1.背景 2.现状 3.趋势)
  [Researcher-Summarize] 正在提炼 2 份资料...
  [Researcher-Eval] 资料已充足，结束调研。
[Main Graph] Researcher 部门工作完成，接收报告。

[Writer] 根据调研报告开始码字...
参考资料: 【深度调研报告】基于 2 篇资料，核心结论如下...

========================================
🎉 最终产出结果：
这是一篇关于 2024年人形机器人行业发展趋势 的爆款文章初稿！

坑与避坑指南

在实战中引入 Sub-graphs，往往会让初学者痛不欲生。作为你的导师，我已经替你把雷排好了，注意以下三点：

💣 坑一：父子状态混用导致数据污染

现象：图省事，直接把 AgencyState 传给 researcher_graph。结果 Researcher 内部的一个 Bug，把 Writer 的 draft 字段给覆盖为空了。 避坑：永远、永远、永远使用 Wrapper 函数来做状态隔离（就像我们在代码中写的 researcher_wrapper_node）。遵守迪米特法则（最少知识原则），子图只需要知道它该知道的数据，返回它该返回的数据。

💣 坑二：子图陷入死循环，主图直接卡死

现象：由于大模型幻觉，Researcher 一直觉得“资料不够”，无限次调用 Search，导致整个 Agency 系统假死，API 账单爆炸。 避坑：在子图的 State 中务必加入 iteration_count（迭代次数）字段。在子图的条件判断（Conditional Edge）中，强制设置一个最大循环次数（如 if count >= 3: return END）。不要完全信任 LLM 的判断逻辑。

💣 坑三：流式输出（Streaming）的层级丢失

现象：使用 .stream() 监听主图时，发现子图内部的节点执行信息全丢了，只能看到 researcher_dept 开始和结束。 避坑：在 LangGraph 中，如果你想要在使用 Sub-graph 时依然能追踪底层节点的流式输出，在调用主图的 .stream() 时，需要传入 subgraphs=True 参数。例如：agency_graph.stream(initial_state, subgraphs=True)。这样 LangGraph 就会把嵌套层级也一起 yield 出来。

📝 本期小结

今天我们完成了一次架构上的升华。

通过 **Sub-graphs（子图）**技术，我们把原本臃肿不堪的系统，重构成了高内聚、低耦合的模块化架构。

• 对于 Main Graph 来说，Researcher 只是一个输入“主题”、输出“报告”的普通节点。
• 对于 Researcher 来说，它拥有自己独立的状态机、独立的循环逻辑、独立的重试机制。

这种设计，让你在未来可以随时把 Researcher 团队替换掉、升级掉，甚至把 Researcher 这个 Sub-graph 单独打包成一个微服务发布出去，而主 Graph 一行代码都不用改！这就是架构设计的魅力。

剧透一下： 现在我们的 Agency 运转得很丝滑，但如果 Writer 写出的文章，Editor 觉得不行，甚至 Planner 觉得偏题了怎么办？ 下一期（第 19 期），我们将引入 LangGraph 的高阶特性：Human-in-the-loop（人类在环）与断点机制（Breakpoints）。我们将让系统在关键节点停下来，等你（老板）审批通过后，再继续执行。

保持热爱，我们下期见！代码敲起来！