LangGraph 子图模式与模块化状态管理

状态爆炸的困境

构建复杂 Agent 时，第一个冲击往往来自状态管理。当流程节点超过二十个，State 定义开始膨胀，边界条件相互纠缠，调试时需要在巨大的对象中追踪字段变化。

这不是代码质量问题，而是架构层面的耦合。

LangGraph 的子图（Subgraph）机制提供了一条出路。它将单一巨型状态图拆分为多个自治单元，每个子图维护自己的状态和转换逻辑，通过定义良好的接口与外部通信。

子图的核心价值

graph TB
    subgraph 父图 ParentGraph
    A[用户查询] --> B[研究子图]
    B --> C[写作子图]
    C --> D[审核子图]
    D --> E[最终输出]
    end
    
    subgraph 研究子图 ResearchSubgraph
    B1[输入映射] --> B2[搜索节点]
    B2 --> B3[分析节点]
    B3 --> B4[综合节点]
    B4 --> B5[输出映射]
    end
    
    subgraph 写作子图 WritingSubgraph
    C1[输入映射] --> C2[大纲生成]
    C2 --> C3[段落撰写]
    C3 --> C4[润色优化]
    C4 --> C5[输出映射]
    end
    
    subgraph 审核子图 ReviewSubgraph
    D1[输入映射] --> D2[内容检查]
    D2 --> D3[人工审核]
    D3 --> D4[修改建议]
    D4 --> D5[输出映射]
    end
    
    B -.->|调用| B1
    C -.->|调用| C1
    D -.->|调用| D1

1. 状态隔离

子图拥有独立的 State 定义。父图只需关心输入输出，不必了解内部字段。这种封装使每个模块的复杂度控制在可控范围内。

# 父图状态 - 极简
class ParentState(TypedDict):
    user_query: str
    research_result: str  # 子图输出
    final_answer: str

# 子图状态 - 内部细节
class ResearchState(TypedDict):
    query: str
    search_results: List[dict]
    synthesized: str
    confidence: float

2. 独立演进

子图的内部实现可以修改而不影响父图。更换搜索策略、添加分析节点、调整评分逻辑，这些变更被限制在子图边界内。

3. 复用性

一个设计良好的子图可以在多个父图中复用。研究子图可以被问答 Agent、报告生成 Agent、数据清洗 Agent 共享。

子图接入的三种模式

graph LR
    subgraph 模式一：函数包装
    A1[父图状态] --> B1[函数包装器]
    B1 --> C1[构建子图]
    C1 --> D1[执行子图]
    D1 --> E1[返回结果]
    E1 --> F1[父图状态更新]
    end
    
    subgraph 模式二：显式编译
    A2[父图状态] --> B2[输入映射]
    B2 --> C2[预编译子图]
    C2 --> D2[执行子图]
    D2 --> E2[输出映射]
    E2 --> F2[父图状态更新]
    end
    
    subgraph 模式三：状态通道
    A3[父图状态] --> B3[批量触发]
    B3 --> C3[Send子图1]
    B3 --> D3[Send子图2]
    B3 --> E3[Send子图3]
    C3 --> F3[结果聚合]
    D3 --> F3
    E3 --> F3
    F3 --> G3[父图状态更新]
    end

模式一：函数包装（简单场景）

from langgraph.graph import StateGraph

def research_subgraph(state: ParentState):
    # 构建并执行子图
    subgraph = StateGraph(ResearchState)
    # ... 添加节点和边
    chain = subgraph.compile()
    
    # 状态转换
    result = chain.invoke({
        "query": state["user_query"]
    })
    
    return {"research_result": result["synthesized"]}

# 在父图中作为普通节点使用
parent.add_node("research", research_subgraph)

适用场景：子图逻辑简单，不需要访问父图的其他字段。

模式二：显式编译（推荐）

# 独立定义子图
research_builder = StateGraph(ResearchState)
research_builder.add_node("search", search_node)
research_builder.add_node("analyze", analyze_node)
research_builder.add_edge("search", "analyze")
research_builder.set_entry_point("search")
research_builder.set_finish_point("analyze")

research_graph = research_builder.compile()

# 在父图中接入
def call_research(state: ParentState):
    # 输入映射
    sub_state = {
        "query": state["user_query"],
        "search_results": [],
        "synthesized": "",
        "confidence": 0.0
    }
    
    # 执行子图
    result = research_graph.invoke(sub_state)
    
    # 输出映射
    return {"research_result": result["synthesized"]}

parent.add_node("research", call_research)

这种模式将子图构建与调用分离，子图可以被独立测试和复用。

模式三：状态通道（复杂交互）

当子图需要与父图进行多轮交互时，使用 Send/Interrupt 机制：

from langgraph.types import Send

# 父图中批量触发子图
def fan_out(state: ParentState):
    queries = decompose_query(state["user_query"])
    return [Send("research_subgraph", {"query": q}) for q in queries]

parent.add_conditional_edges("decomposer", fan_out, ["research_subgraph"])

# 子图作为独立节点接入
parent.add_node("research_subgraph", research_graph)

子图执行完成后，结果会自动聚合回父图状态。

子图设计的边界原则

单一职责

每个子图只做一件事。研究子图只负责信息检索与综合，不处理用户交互，不做格式转换。

最小接口

子图的输入输出字段尽可能少。避免将父图的完整状态传入子图，这会破坏封装。

# 不推荐
sub_result = research_graph.invoke(state)  # 传入整个状态

# 推荐
sub_result = research_graph.invoke({
    "query": state["user_query"]  # 只传入需要的数据
})

错误隔离

子图内部应处理自己的异常，而不是将错误抛给父图。

def robust_research_node(state: ResearchState):
    try:
        results = search(state["query"])
        return {"search_results": results}
    except SearchAPIError:
        # 子图内部降级，不影响父图
        return {"search_results": [], "error": "search_failed"}

递归子图：嵌套的可能性

子图内部可以继续包含子图，形成层级结构：

graph TD
    A[ParentGraph 父图] --> B[ResearchSubgraph]
    A --> C[WritingSubgraph]
    A --> D[ReviewSubgraph]
    
    B --> B1[SearchSubgraph]
    B --> B2[SynthesisNode]
    
    B1 --> B1a[WebSearch]
    B1 --> B1b[VectorSearch]
    B1 --> B1c[DBSearch]
    
    C --> C1[OutlineNode]
    C --> C2[WritingSubgraph]
    C2 --> C2a[Para1]
    C2 --> C2b[Para2]
    C2 --> C2c[Para3]
    
    D --> D1[AutoCheck]
    D --> D2[HumanReview]
    D --> D3[Revision]
    
    style A fill:#e1f5ff
    style B fill:#fff4e1
    style B1 fill:#ffe1e1
    style C fill:#fff4e1
    style C2 fill:#ffe1e1
    style D fill:#fff4e1

层级示意：

ParentGraph
├── ResearchSubgraph
│   ├── SearchSubgraph
│   │   ├── WebSearch
│   │   └── VectorSearch
│   └── SynthesisNode
├── WritingSubgraph
└── ReviewSubgraph

这种嵌套需要谨慎使用。建议限制在三层以内，过深的层级会增加调试难度。

调试策略

1. 分层测试

先独立测试子图，确认其输入输出符合预期，再接入父图。

# 子图单元测试
def test_research_subgraph():
    result = research_graph.invoke({
        "query": "LangGraph subgraph patterns"
    })
    assert "synthesized" in result
    assert result["confidence"] > 0.5

2. 状态检查点

在子图边界设置检查点，观察状态转换：

1
2
3

parent.add_node("research", call_research)
parent.add_node("_inspect", lambda s: print(f"Research output: {s['research_result']}") or s)
parent.add_edge("research", "_inspect")

3. 可视化

LangGraph 的绘图功能对子图同样有效：

# 可视化子图
research_graph.get_graph().draw_png("research_subgraph.png")

# 可视化父图（子图显示为复合节点）
parent.get_graph().draw_png("parent_graph.png")

与 Human-in-the-Loop 的结合

子图特别适合封装需要人工审核的环节：

# 审核子图
review_builder = StateGraph(ReviewState)
review_builder.add_node("present", present_for_review)
review_builder.add_node("await_feedback", interrupt_for_human)
review_builder.add_node("apply_decision", apply_review_decision)

review_graph = review_builder.compile(checkpointer=checkpointer)

# 接入父图
parent.add_node("human_review", review_graph)

审核子图内部处理所有与用户的交互细节，父图只需等待最终结果。

性能考量

状态拷贝成本

子图调用时状态会被序列化和反序列化。对于高频调用的子图，保持状态体积小是关键。

并行执行

使用 Send 机制可以并行执行多个子图实例：

def parallel_research(state: ParentState):
    sub_queries = expand_query(state["query"])
    # 所有子图实例并行执行
    return [Send("research_subgraph", {"query": sq}) for sq in sub_queries]