LangGraph Subgraph：子图与模块化设计

随着业务需求增长，LangGraph Agent 的节点数量会不断增加。当一个 Agent 拥有三十多个节点，所有节点堆在同一个 State 里，build_graph() 函数写了两百行，维护就变得很困难——每次改动都需要理解整张图的结构，生怕一条边写错了整个流程就乱了。

这时候需要的不是优化，是重构。而 LangGraph 的子图（Subgraph）机制就是为这种情况准备的。

1.1 为什么需要子图

LangGraph 子图模块化——主图通过 Send 分发任务给子图 A/B/C，子图独立执行后合并

复杂 Agent 面临三个现实问题。

第一是可维护性。三十个节点堆在一张图里，很难一眼看出"这一块是负责搜索的，那一块是负责写作的"。任何修改都需要理解整张图，代价很高。

第二是复用性。搜索逻辑在 A Agent 里用，在 B Agent 里也用。如果搜索流程是一个独立的图，可以把它像一个组件一样插进任何父图，而不是把代码复制过去再改一遍。

第三是独立测试。一个三十节点的图，出问题了不知道是哪段逻辑的问题。如果这三十个节点被拆成三个子图，每个子图可以单独跑单独测，问题隔离就简单多了。

子图的核心思想是：把一组相关节点封装成一个独立的 StateGraph，然后把这个 StateGraph 作为父图的一个节点来使用。

1.2 子图的核心概念

子图是一个完整的 StateGraph，有自己的 State 定义、自己的节点和边，可以单独编译，也可以作为父图的一个节点嵌入使用。

子图有自己独立的 State，这是关键。子图的 State 字段由子图自己管理，父图不需要知道子图内部用了哪些字段。

父图和子图之间通过共享的 State 字段名称通信。如果父图的 State 有一个字段叫 search_results，子图的 State 也有一个字段叫 search_results，那么子图执行完之后，LangGraph 会自动把子图里 search_results 的值同步回父图。

这就是"约定大于配置"——不需要显式地写数据传递逻辑，字段名对上就行。

1.3 两种使用方式

1.3.1 方式一：直接编译后挂载

把子图编译成 CompiledGraph，直接调用 add_node 加进父图。

python

# 子图编译后作为节点
sub_graph = SubGraphBuilder().compile()
parent_graph.add_node("search_subgraph", sub_graph)

这种方式的约束是：子图 State 和父图 State 中，同名字段的类型必须一致。LangGraph 在执行时会自动做字段同步，如果类型不匹配会报错。

这种方式代码最简洁，适合子图和父图 State 字段能对上的情况。

1.3.2 方式二：函数包装

用一个普通函数把子图调用包起来，在函数内部手动做 State 的转换。

python

def search_subgraph_node(state: ParentState) -> dict:
    # 把父图 State 转换成子图需要的 State
    sub_input = {
        "query": state["user_input"],
        "max_results": 5,
    }
    # 调用子图
    sub_result = compiled_sub_graph.invoke(sub_input)
    # 把子图输出映射回父图需要的字段
    return {
        "search_results": sub_result["results"],
        "search_metadata": sub_result["metadata"],
    }

parent_graph.add_node("search", search_subgraph_node)

这种方式更灵活，父图和子图的 State 可以完全不同，字段名不需要匹配，开发者自己控制输入输出的映射关系。代价是多写一些转换代码。

实际项目里，方式二更为常用，因为父子图独立演化时不需要保持字段命名的同步。

1.4 完整示例：研究 Agent

下面是一个研究 Agent，整体结构是：父图协调 → 搜索子图收集资料 → 写作子图生成报告。

python

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


# ============================================================
# 子图 1：搜索子图
# 负责执行多轮搜索，收集足够的资料
# ============================================================

class SearchState(TypedDict):
    query: str                    # 搜索关键词
    results: List[str]            # 搜索结果列表
    search_count: int             # 已搜索次数
    is_sufficient: bool           # 结果是否足够


def search_execute_node(state: SearchState) -> dict:
    """执行一次搜索"""
    query = state["query"]
    count = state["search_count"] + 1

    print(f"  [搜索子图] 第 {count} 次搜索：{query}")

    # 实际项目里换成 Tavily / Serper 等真实搜索 API
    new_results = [
        f"资料{count}-A：关于 {query} 的核心概念",
        f"资料{count}-B：关于 {query} 的实践案例",
    ]

    all_results = state["results"] + new_results
    print(f"  [搜索子图] 累计 {len(all_results)} 条资料")

    return {
        "results": all_results,
        "search_count": count,
    }


def search_evaluate_node(state: SearchState) -> dict:
    """评估搜索结果是否足够"""
    sufficient = len(state["results"]) >= 4
    print(f"  [搜索子图] 资料评估：{'足够' if sufficient else '不足，继续搜索'}")
    return {"is_sufficient": sufficient}


def search_router(state: SearchState) -> str:
    if state["is_sufficient"] or state["search_count"] >= 3:
        return "done"
    return "continue"


def build_search_subgraph():
    graph = StateGraph(SearchState)

    graph.add_node("execute", search_execute_node)
    graph.add_node("evaluate", search_evaluate_node)

    graph.add_edge(START, "execute")
    graph.add_edge("execute", "evaluate")
    graph.add_conditional_edges(
        "evaluate",
        search_router,
        {"continue": "execute", "done": END}
    )

    return graph.compile()


# ============================================================
# 子图 2：写作子图
# 负责根据资料生成结构化报告
# ============================================================

class WritingState(TypedDict):
    topic: str                    # 报告主题
    source_materials: List[str]   # 原始资料
    outline: str                  # 报告大纲
    draft: str                    # 报告草稿
    final_report: str             # 最终报告


def outline_node(state: WritingState) -> dict:
    """生成报告大纲"""
    topic = state["topic"]
    material_count = len(state["source_materials"])
    print(f"  [写作子图] 基于 {material_count} 条资料生成大纲")

    # 实际项目里这里调用 LLM 生成大纲
    outline = f"""# {topic} 研究报告大纲

## 一、背景与概述
## 二、核心发现
## 三、案例分析
## 四、结论与建议"""

    return {"outline": outline}


def draft_node(state: WritingState) -> dict:
    """根据大纲和资料撰写草稿"""
    print(f"  [写作子图] 根据大纲撰写草稿")

    materials_text = "\n".join(
        f"- {m}" for m in state["source_materials"]
    )
    # 实际项目里这里调用 LLM 扩写
    draft = f"{state['outline']}\n\n【参考资料】\n{materials_text}"

    return {"draft": draft}


def polish_node(state: WritingState) -> dict:
    """润色和格式化最终报告"""
    print(f"  [写作子图] 润色报告")

    # 实际项目里这里调用 LLM 润色
    final = f"【最终版本】\n\n{state['draft']}"

    return {"final_report": final}


def build_writing_subgraph():
    graph = StateGraph(WritingState)

    graph.add_node("outline", outline_node)
    graph.add_node("draft", draft_node)
    graph.add_node("polish", polish_node)

    graph.add_edge(START, "outline")
    graph.add_edge("outline", "draft")
    graph.add_edge("draft", "polish")
    graph.add_edge("polish", END)

    return graph.compile()


# ============================================================
# 父图：研究 Agent
# 协调搜索子图和写作子图
# ============================================================

class ResearchState(TypedDict):
    user_question: str            # 用户提问
    search_query: str             # 传给搜索子图的关键词
    raw_materials: List[str]      # 从搜索子图拿回来的资料
    report: str                   # 从写作子图拿回来的最终报告


# 编译子图（全局只需要编译一次）
compiled_search = build_search_subgraph()
compiled_writing = build_writing_subgraph()


def prepare_node(state: ResearchState) -> dict:
    """准备阶段：分析用户问题，提取搜索关键词"""
    question = state["user_question"]
    print(f"\n[父图] 开始研究任务：{question}")

    # 实际项目里用 LLM 分析问题，提取搜索词
    search_query = question.replace("是什么", "").replace("如何", "").strip()
    print(f"[父图] 提取搜索关键词：{search_query}")

    return {"search_query": search_query}


def search_phase_node(state: ResearchState) -> dict:
    """搜索阶段：调用搜索子图"""
    print(f"\n[父图] 进入搜索阶段")

    # 构造子图的输入 State
    search_input = {
        "query": state["search_query"],
        "results": [],
        "search_count": 0,
        "is_sufficient": False,
    }

    # 调用搜索子图
    search_output = compiled_search.invoke(search_input)

    print(f"[父图] 搜索阶段完成，获得 {len(search_output['results'])} 条资料")

    # 把子图输出映射回父图 State
    return {"raw_materials": search_output["results"]}


def writing_phase_node(state: ResearchState) -> dict:
    """写作阶段：调用写作子图"""
    print(f"\n[父图] 进入写作阶段")

    # 构造子图的输入 State
    writing_input = {
        "topic": state["user_question"],
        "source_materials": state["raw_materials"],
        "outline": "",
        "draft": "",
        "final_report": "",
    }

    # 调用写作子图
    writing_output = compiled_writing.invoke(writing_input)

    print(f"[父图] 写作阶段完成")

    # 把子图输出映射回父图 State
    return {"report": writing_output["final_report"]}


def build_research_agent():
    graph = StateGraph(ResearchState)

    graph.add_node("prepare", prepare_node)
    graph.add_node("search_phase", search_phase_node)
    graph.add_node("writing_phase", writing_phase_node)

    graph.add_edge(START, "prepare")
    graph.add_edge("prepare", "search_phase")
    graph.add_edge("search_phase", "writing_phase")
    graph.add_edge("writing_phase", END)

    return graph.compile()


# ============================================================
# 运行
# ============================================================

if __name__ == "__main__":
    agent = build_research_agent()

    result = agent.invoke({
        "user_question": "LangGraph 子图如何实现模块化",
        "search_query": "",
        "raw_materials": [],
        "report": "",
    })

    print("\n" + "=" * 60)
    print("研究报告：")
    print("=" * 60)
    print(result["report"])

运行后输出：

code

[父图] 开始研究任务：LangGraph 子图如何实现模块化
[父图] 提取搜索关键词：LangGraph 子图实现模块化

[父图] 进入搜索阶段
  [搜索子图] 第 1 次搜索：LangGraph 子图实现模块化
  [搜索子图] 累计 2 条资料
  [搜索子图] 资料评估：不足，继续搜索
  [搜索子图] 第 2 次搜索：LangGraph 子图实现模块化
  [搜索子图] 累计 4 条资料
  [搜索子图] 资料评估：足够
[父图] 搜索阶段完成，获得 4 条资料

[父图] 进入写作阶段
  [写作子图] 基于 4 条资料生成大纲
  [写作子图] 根据大纲撰写草稿
  [写作子图] 润色报告
[父图] 写作阶段完成

父图的日志和子图的日志都打印出来了，层次一目了然。

1.5 子图的独立测试

子图可以完全独立于父图进行测试，这是子图最大的工程价值之一。

python

def test_search_subgraph():
    """单独测试搜索子图，不需要父图存在"""
    compiled_search = build_search_subgraph()

    test_input = {
        "query": "Python 异步编程",
        "results": [],
        "search_count": 0,
        "is_sufficient": False,
    }

    result = compiled_search.invoke(test_input)

    # 断言子图的输出符合预期
    assert len(result["results"]) > 0, "搜索子图应该返回至少一条结果"
    assert result["is_sufficient"] is True, "搜索子图应该在结果足够后结束"
    print(f"搜索子图测试通过，最终获得 {len(result['results'])} 条结果")


def test_writing_subgraph():
    """单独测试写作子图"""
    compiled_writing = build_writing_subgraph()

    test_input = {
        "topic": "Python 异步编程",
        "source_materials": [
            "资料1：asyncio 的基本用法",
            "资料2：async/await 语法详解",
            "资料3：异步 IO 的性能优势",
        ],
        "outline": "",
        "draft": "",
        "final_report": "",
    }

    result = compiled_writing.invoke(test_input)

    assert result["final_report"], "写作子图应该生成非空报告"
    print(f"写作子图测试通过，报告长度：{len(result['final_report'])} 字符")


if __name__ == "__main__":
    print("=== 单独测试子图 ===")
    test_search_subgraph()
    test_writing_subgraph()
    print("所有子图测试通过")

这和 Java 里的单元测试思路完全一样：每个模块独立测，然后集成测。区别是 LangGraph 的子图天然有边界，不需要 Mock 父图。

1.6 父子图的层次结构

父图只关心"搜索阶段"和"写作阶段"这两个节点，子图内部有几个节点、有没有循环，父图完全不关心。

1.7 几个实践建议

子图的 State 字段要精简。子图的 State 只应该包含这个子图需要的字段，不要把父图的其他字段也带进去。State 越精简，子图越容易被复用。

共享字段命名要一致。如果使用直接挂载的方式（方式一），父子图之间共享的字段必须同名同类型。建议在项目里定义一份共享字段的常量，避免改了一处忘了另一处。

子图不应该知道父图的存在。这是模块化的基本原则。子图只读写自己 State 里的字段，不持有父图的任何引用。父图负责做输入/输出的映射，子图不参与这个过程。

1.8 小结

一个三十节点的 Agent 拆成三个十节点的子图，每个子图有清晰的职责边界，可以独立开发、独立测试、独立复用——这是能长期维护的结构。

LangGraph 不限制代码组织方式，子图只是它提供的一种拆分手段。用不用，看工程规模。

下一篇讲 Persistence 持久化，解决 Agent 执行到一半进程崩了怎么从断点恢复的问题。