在使用 LangGraph 开发 Agent 应用时，应用刚开始往往很简单：几个 Node、一份 State，再用 Edge 把执行顺序串起来即可实现预期任务。

随着功能增加，Graph 很快就会变复杂。比如，检索流程里增加查询改写、召回、过滤和重排；订单流程加入校验和人工确认；不同 Agent 又各自需要消息记录和中间状态。所有节点继续共用一份 State，此时，代码虽然还能运行，但模块之间的关系却会越来越难理清。

针对这种复杂的场景，LangGraph 提供了子图（Subgraph），它本身也是一张完整的 Graph，只是被放到另一张 Graph 中执行。

实际使用 Subgraph 时，需要注意三点：哪些流程应该独立出来，父子图之间如何传递状态，以及子图内部状态要保存多久。

一、复杂图的边界

假设我们正在实现一个企业知识助手，最初只有几个步骤：

代码片段Text
用户问题
   ↓
问题分析
   ↓
知识检索
   ↓
生成回答

随着检索能力增强，其中的“知识检索”可能继续扩展：

代码片段Text
查询改写
   ↓
向量检索
   ↓
结果过滤
   ↓
结果重排
   ↓
整理上下文

系统随后又加入订单查询、数据库查询和人工审批，整张图很快就可能增加到二三十个节点。

节点数量多一些通常还不是问题。麻烦往往来自 State。

最初只有几个字段：

代码片段Text
messages
query
answer

业务增加后，State 可能逐渐变成：

代码片段Text
messages
query
rewritten_query
documents
reranked_documents
order_id
order_status
approval_result
sql
sql_result
final_answer
...

其中很多数据只属于某一段流程。

例如，重排节点只需要文档列表，订单流程只关心订单状态，SQL 流程则需要 SQL 和查询结果。把这些字段全部放到同一份 State 中，会让全局状态不断膨胀，也很难判断一个节点究竟依赖哪些数据。

Multi-Agent 场景下这个问题更加明显。比如，Research Agent 可能有自己的检索记录和消息历史，SQL Agent 也会产生 SQL、执行结果和修正过程。如果所有 Agent 共用同一个 messages，各种中间消息最终都会混在一起。

更合理的结构是把相对完整的局部流程独立出来：

代码片段Text
                    ┌─ Research Subgraph
用户请求 → Router ──┼─ Order Subgraph
                    └─ SQL Subgraph
                           ↓
                       Final Answer

父图负责整体流程编排，各个子图负责自己的任务和状态。

LangGraph 中的 Subgraph 与普通 Graph 使用同一套机制，它仍然可以有自己的 State、Node、Edge、条件路由、Interrupt 和 Checkpoint。

因此，判断一段流程要不要做成 Subgraph，主要看它有没有形成相对独立的工作流。

二、Subgraph 的结构

Subgraph 没有单独的一套构建方式，仍然使用 StateGraph。

例如：

代码片段Python
class SearchState(TypedDict):
    query: str
    result: str

TypedDict 用来描述 State 中有哪些字段，以及每个字段的数据类型。

然后创建 StateGraph：

代码片段Python
builder = StateGraph(SearchState)

这里传入的 SearchState，就是这张 Graph 使用的状态结构。

接下来添加节点和执行关系：

代码片段Python
builder.add_node("search", search)
builder.add_edge(START, "search")
builder.add_edge("search", END)

几个常用 API 的作用分别是：

add_node() 用来注册节点。节点可以是一个普通 Python 函数，也可以是一张已经编译好的 Subgraph。

add_edge() 用来指定节点之间的执行顺序。START 和 END 分别表示 Graph 的入口和出口。

所有节点和 Edge 定义完成后，再调用：

代码片段Python
graph = builder.compile()

compile() 会生成一张可以实际执行的 Graph。

运行时调用：

代码片段Python
result = graph.invoke(input_state)

invoke() 接收初始状态，执行 Graph，并返回运行结束后的状态。

因此，Subgraph 在父图中的结构通常是这样的：

代码片段Text
Parent Graph

START
  ↓
Router
  ↓
Search Subgraph
  │
  ├─ Rewrite
  ├─ Search
  └─ Rerank
  ↓
Answer
  ↓
END

对于父图来说，整个 Search Subgraph 就是一个节点。查询改写、检索和重排这些细节，都留在子图内部。

只要父图和子图之间约定的数据没有变化，子图内部增加节点、调整执行顺序，通常不会影响父图。

父子图组合时，主要有两种写法。

如果它们共享一部分 State 字段，可以直接把编译后的子图加入父图。

如果两边使用不同的 State，则需要增加一个普通 Node，在这个 Node 中完成输入和输出转换。

三、共享状态模式

我们先从结构比较简单的情况开始。

假设父图使用：

代码片段Text
text
result

子图使用：

代码片段Text
text
result
normalized

其中 text 和 result 同时出现在父图和子图中，可以直接用于传递数据。

normalized 只在子图内部使用。

完整代码如下：

代码片段Python
from typing_extensions import NotRequired, TypedDict
from langgraph.graph import END, START, StateGraph
 
 
class ParentState(TypedDict):
    text: str
    result: str
 
 
class TextSubgraphState(TypedDict):
    text: str
    result: str
    normalized: NotRequired[str]
 
 
def normalize_text(state: TextSubgraphState):
    return {
        "normalized": state["text"].strip().lower()
    }
 
 
def build_result(state: TextSubgraphState):
    return {
        "result": state["normalized"]
    }
 
 
subgraph_builder = StateGraph(TextSubgraphState)
 
subgraph_builder.add_node("normalize", normalize_text)
subgraph_builder.add_node("build_result", build_result)
 
subgraph_builder.add_edge(START, "normalize")
subgraph_builder.add_edge("normalize", "build_result")
subgraph_builder.add_edge("build_result", END)
 
text_subgraph = subgraph_builder.compile()
 
 
parent_builder = StateGraph(ParentState)
 
parent_builder.add_node("text_pipeline", text_subgraph)
 
parent_builder.add_edge(START, "text_pipeline")
parent_builder.add_edge("text_pipeline", END)
 
graph = parent_builder.compile()
 
result = graph.invoke({
    "text": "  LANGGRAPH SUBGRAPH  ",
    "result": ""
})
 
print(result)

代码中出现了：

代码片段Python
normalized: NotRequired[str]

NotRequired 表示 normalized 属于 TextSubgraphState，但调用 Graph 时不要求提供这个字段。

这种写法很适合描述运行过程中才会产生的数据。

调用子图时，我们只传入 text 和 result。执行 normalize_text 后，State 中才出现 normalized。

运行结果如下：

代码片段Text
{
    'text': '  LANGGRAPH SUBGRAPH  ',
    'result': 'langgraph subgraph'
}

执行过程中，子图先从共享字段 text 中读取原始内容，然后生成内部字段 normalized，最后把处理结果写入共享字段 result。

子图结束后，父图可以拿到更新后的 result。

normalized 仍然只属于子图，不需要加入 ParentState。

父图中有一行与普通 Node 不太一样：

代码片段Python
parent_builder.add_node("text_pipeline", text_subgraph)

这里传给 add_node() 的不是函数，而是 compile() 之后得到的 Subgraph。

只要父图和子图之间存在可以直接传递的共享字段，就可以这样组合，不需要再写一层调用函数。

例如：

代码片段Text
ParentState
├─ messages
├─ user_id
└─ result

ResearchState
├─ messages
├─ result
├─ documents
└─ research_notes

messages 和 result 负责父子图之间的数据交换，documents 和 research_notes 则留在 Research Subgraph 内部。

这种方式适合父子图的数据结构本来就比较接近的场景。

四、独立状态模式

另一种情况更常见：父图和子图处理的数据本来就不同。

例如父图只有：

代码片段Text
query
answer

而 Search Subgraph 内部使用：

代码片段Text
raw_query
normalized_query
documents

如果为了直接挂载 Subgraph，把这些检索字段全部加进父图，State 很快又会回到不断膨胀的状态。

这时可以增加一个普通 Node，专门负责父图和子图之间的数据转换。

代码片段Python
from typing_extensions import NotRequired, TypedDict
from langgraph.graph import END, START, StateGraph
 
 
class SearchState(TypedDict):
    raw_query: str
    normalized_query: NotRequired[str]
 
 
def normalize_query(state: SearchState):
    return {
        "normalized_query": state["raw_query"].strip().lower()
    }
 
 
search_builder = StateGraph(SearchState)
search_builder.add_node("normalize", normalize_query)
search_builder.add_edge(START, "normalize")
search_builder.add_edge("normalize", END)
 
search_graph = search_builder.compile()
 
 
class ParentState(TypedDict):
    query: str
    answer: str
 
 
def call_search(state: ParentState):
    child_input = {
        "raw_query": state["query"]
    }
 
    child_output = search_graph.invoke(child_input)
 
    return {
        "answer": child_output["normalized_query"]
    }
 
 
parent_builder = StateGraph(ParentState)
parent_builder.add_node("search", call_search)
 
parent_builder.add_edge(START, "search")
parent_builder.add_edge("search", END)
 
graph = parent_builder.compile()
 
result = graph.invoke({
    "query": "  LangGraph Context  ",
    "answer": ""
})
 
print(result)

运行结果如下：

代码片段Text
{
    'query': '  LangGraph Context  ',
    'answer': 'langgraph context'
}

这段代码中，call_search() 仍然是父图里的普通 Node。

它先取出父图中的：

代码片段Python
state["query"]

再构造成 Search Subgraph 需要的输入：

代码片段Python
{
    "raw_query": state["query"]
}

然后调用：

代码片段Python
child_output = search_graph.invoke(child_input)

子图执行完成后，再把：

代码片段Python
child_output["normalized_query"]

写回父图的 answer。

整个过程可以表示为：

代码片段Text
ParentState.query
       ↓
SearchState.raw_query
       ↓
Search Subgraph
       ↓
SearchState.normalized_query
       ↓
ParentState.answer

这样，父图只保存整个业务流程真正需要的数据，检索过程中的字段全部留在 Search Subgraph 中。

Multi-Agent 项目中也经常使用这种方式。

例如父 Agent 把一个任务交给 Research Agent 时，只传当前任务和必要背景；Research Agent 自己维护搜索记录、工具结果和消息历史；任务完成后，只把研究结果返回父 Agent。

这样更容易控制每个 Agent 实际拿到的上下文。

五、上下文隔离

Subgraph 在 Agent 系统中的一个重要用途，就是缩小上下文范围。

假设我们有一个客服系统：

代码片段Text
                 ┌─ Account Agent
Request → Router ├─ Billing Agent
                 └─ Technical Agent

父图可以只保留：

代码片段Text
messages
user_id
intent
final_answer

Billing Subgraph 使用自己的状态：

代码片段Text
invoice_id
payment_status
refund_policy
approval_result
billing_messages

Technical Subgraph 则维护：

代码片段Text
product
version
error_code
documents
diagnosis
technical_messages

账单处理和技术支持使用的是两套完全不同的数据。

如果这些字段全部塞进父图，State 会越来越难管理。更麻烦的是，如果几个 Agent 再共用一份 messages，技术排查过程中产生的大量工具返回值，也可能被带入之后的账单处理。

使用独立 State 后，可以在调用子图时只传它真正需要的内容。

例如父图当前拥有：

代码片段Text
历史消息
用户信息
产品信息
当前问题

调用 Technical Subgraph 时，只传：

代码片段Text
当前问题
产品版本
错误信息

子图处理完成后，只返回：

代码片段Text
诊断结果

最终是否把这个结果加入 messages，由父图决定。

Subgraph 本身不会自动帮我们裁剪上下文。真正控制上下文的是 State 的设计，以及父子图之间明确的输入输出。

这也是前面两种组合方式的主要区别。

共享 State 写起来更省事，适合父子图确实需要共同使用的数据。

如果希望严格控制子图能够获得哪些内容，独立 State 配合显式转换会更清晰。

六、子图的持久化

State 决定数据放在哪里，Persistence 还要解决另一个问题：这些数据需要保留多久。

LangGraph 可以通过 compile() 的 checkpointer 参数控制子图的持久化方式：

代码片段Python
subgraph = builder.compile(checkpointer=...)

常见的三种情况如下：

模式checkpointer状态保存范围常见场景
Per-invocationNone单次调用普通独立任务
Per-threadTrue同一 Thread需要连续上下文的子 Agent
StatelessFalse不保存无需恢复的简单流程

Per-invocation

默认情况下，可以直接：

代码片段Python
subgraph = builder.compile()

此时 checkpointer 为 None。

每次调用子图时，内部状态都会重新开始。

例如：

代码片段Text
第一次调用
查询订单 A

第二次调用
查询订单 B

第二次查询不需要继承订单 A 的内部处理状态，因此这种模式就足够了。

如果父图本身配置了 Checkpointer，子图在单次调用过程中仍然可以使用相应的 Checkpoint 能力，因此 Interrupt 和 Durable Execution 仍然可以工作。

对于大多数普通 Subgraph，默认模式已经够用。

Per-thread

如果希望子图在同一个 Thread 中保留之前的状态，可以使用：

代码片段Python
subgraph = builder.compile(checkpointer=True)

例如 Research Agent 连续执行：

代码片段Text
第 1 次
研究 PostgreSQL Checkpointer

第 2 次
继续比较 SQLite

第 3 次
根据前面的结果整理方案

如果这几次调用属于同一个 Thread，Research Subgraph 就可以继续使用此前保存的状态。

这里的 True 并不会创建新的 Checkpointer。

它表示这个子图要持续保存状态，实际仍然使用父图配置的 Checkpointer。因此父图需要先具备持久化能力。

这种模式还涉及一个限制：同一个需要持续保存状态的子图，不适合在同一位置同时发起多次调用，否则可能写入相同的 Checkpoint 空间。

如果没有明确的跨调用状态需求，不必急着使用 Per-thread。

Stateless

还可以显式关闭子图的 Checkpoint：

代码片段Python
subgraph = builder.compile(checkpointer=False)

这种模式不会保存子图内部运行状态。

它适合执行时间很短、失败后重新运行成本也很低的局部流程。

关闭 Checkpoint 后，子图自然也无法依靠内部 Checkpoint 完成暂停恢复。如果执行过程中进程退出，这部分流程需要重新执行。

因此，是否保存子图状态，应根据恢复需求来决定，而不是所有 Subgraph 都使用同一种配置。

七、实际项目的划分

项目变大以后，很容易走到另一个极端：每看到几个 Node，就想单独做成一个 Subgraph。

这样同样不利于维护。

一个局部流程是否值得独立出来，可以根据以下几个方面进行考量：

  • 几个节点共同完成一个相对完整的任务；
  • 存在较多只在这段流程中使用的 State；
  • 内部流程以后还会继续变化；
  • 需要自己的消息上下文；
  • 有单独的 Interrupt 或持久化需求；
  • 需要被其他 Graph 复用。

例如：

代码片段Text
Rewrite → Search → Rerank

这三个节点共同完成一次检索，而且内部还有查询和文档状态，做成 SearchSubgraph 比较合适。

另一段流程如果只是：

代码片段Text
Validate → Format

两个节点都直接使用父图 State，也没有单独扩展的需求，继续留在父图里反而更清楚。

确定要使用 Subgraph 之后，再判断父子图之间的 State 关系。

如果双方确实共享一部分业务数据：

代码片段Text
ParentState
    messages
    result

SubgraphState
    messages
    result
    local_data

可以直接加入：

代码片段Python
parent.add_node("worker", subgraph)

如果两边的数据结构差异比较大：

代码片段Text
ParentState
    query
    answer

SubgraphState
    task
    documents
    summary

更适合写一个普通 Node，明确完成输入输出转换：

代码片段Python
def call_subgraph(state):
    output = subgraph.invoke(...)
    return ...

前一种方式代码更短，后一种方式边界更明确。选择哪一种，主要取决于父子图之间究竟需要共享多少状态。

回到文章开头那张不断扩大的 Graph，真正需要确定的其实是几个很实际的问题：

代码片段Text
哪些节点属于同一个局部流程
哪些数据只在这段流程中使用
父图需要给子图什么
子图最终需要返回什么
内部状态要不要跨调用保留

这些问题明确以后，Subgraph 的划分通常也就清楚了。

父图负责整体流程，各个子图管理自己的局部状态和执行过程，再通过少量明确的数据互相连接。对于规模逐渐增大的 LangGraph 应用，这种结构比持续扩充一份全局 State 更容易维护。

总结

本文围绕 LangGraph 的 Subgraph 机制，说明它如何把复杂、耦合的 Graph 拆分为可独立维护的子图模块。

在主图中以节点形式调用子图，既能复用已有流程，也能借助独立的 State 与消息通道实现上下文隔离，避免父子图之间的状态互相污染，让复杂 Agent 的结构更清晰。