Subgraphs 子图
Subgraph（子图）是 LangGraph 中实现模块化和复用的核心机制。你可以在一个图中嵌入另一个图，子图在父图中作为一个节点运行。
为什么需要 Subgraph
- 模块化：将复杂逻辑拆分为独立子图
- 复用：同一子图可在多个父图中使用
- 封装：子图有自己的 state，与父图隔离
- 嵌套调试：子图的执行过程也可以流式查看
基本用法
python
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
# === 子图 ===
class SubState(TypedDict):
sub_input: str
sub_output: str
def sub_node1(state: SubState):
return {"sub_output": f"子图处理: {state['sub_input']}"}
sub_builder = StateGraph(SubState)
sub_builder.add_node("sub_node1", sub_node1)
sub_builder.add_edge(START, "sub_node1")
sub_builder.add_edge("sub_node1", END)
sub_graph = sub_builder.compile()
# === 父图 ===
class ParentState(TypedDict):
input: str
output: str
def parent_node(state: ParentState):
# 调用子图
sub_result = sub_graph.invoke({"sub_input": state["input"]})
return {"output": sub_result["sub_output"]}
builder = StateGraph(ParentState)
builder.add_node("parent_node", parent_node)
builder.add_edge(START, "parent_node")
builder.add_edge("parent_node", END)
graph = builder.compile()
result = graph.invoke({"input": "hello"})
print(result) # {'input': 'hello', 'output': '子图处理: hello'}子图作为节点添加
更优雅的方式：将子图编译后直接作为节点添加到父图：
python
# 子图
sub_builder = StateGraph(SubState)
sub_builder.add_node("process", process_node)
sub_builder.add_edge(START, "process")
sub_builder.add_edge("process", END)
sub_graph = sub_builder.compile()
# 父图 - 子图作为一个节点
builder = StateGraph(ParentState)
builder.add_node("sub_task", sub_graph) # 子图作为节点
builder.add_node("finalize", finalize_node)
builder.add_edge(START, "sub_task")
builder.add_edge("sub_task", "finalize")
builder.add_edge("finalize", END)子图与父图的状态交互
方法 1：传入/传出字典
python
class ParentState(TypedDict):
query: str
documents: list
answer: str
class SubState(TypedDict):
query: str
result: list
def parent_to_sub(parent_state: ParentState) -> SubState:
"""将父状态映射到子状态"""
return {
"query": parent_state["query"],
"result": parent_state.get("documents", []),
}
def sub_to_parent(sub_state: SubState) -> dict:
"""将子状态合并回父状态"""
return {
"documents": sub_state["result"],
}
# 在父图节点中调用
def parent_node(state: ParentState):
sub_input = parent_to_sub(state)
sub_output = sub_graph.invoke(sub_input)
return sub_to_parent(sub_output)方法 2：使用 Command
python
from langgraph.graph import Command
def router_to_sub(state: ParentState):
return Command(
goto="my_subgraph_node",
update={"mapped_input": state["input"]}
)嵌套子图
子图可以包含子图，形成多层嵌套：
python
# 最内层
inner = StateGraph(InnerState).compile()
# 中间层
middle = StateGraph(MidState)
middle.add_node("inner", inner) # 内层子图
# ...
# 最外层
outer = StateGraph(OuterState)
outer.add_node("middle", middle) # 中间层子图
# ...子图的 Streaming
子图的 streaming 信息会包含在父图的 streaming 输出中：
python
for event in parent_graph.stream(
{"input": "hello"},
stream_mode="updates",
subgraphs=True # 同时输出子图事件
):
print(event)
# 可能输出:
# (('sub_task',), {'process': {'sub_output': '子图处理...'}})
# ((), {'finalize': {'output': '最终结果'}})实际案例：RAG 子图
python
class RAGState(TypedDict):
question: str
context: list
answer: str
class RetrievalState(TypedDict):
query: str
docs: list
def retrieve(state: RetrievalState):
docs = vectorstore.similarity_search(state["query"])
return {"docs": docs}
# RAG 子图
rag_builder = StateGraph(RetrievalState)
rag_builder.add_node("retrieve", retrieve)
rag_builder.add_edge(START, "retrieve")
rag_builder.add_edge("retrieve", END)
rag_subgraph = rag_builder.compile()
# 在 Agent 中使用
def agent_with_rag(state: State):
question = state["messages"][-1].content
# 调用 RAG 子图
rag_result = rag_subgraph.invoke({"query": question})
context = "\n".join(rag_result["docs"])
prompt = f"基于以下内容回答：\n{context}\n\n问题：{question}"
response = llm.invoke(prompt)
return {"messages": [response]}最佳实践
- 子图应该有清晰的边界：明确的输入输出接口
- 不要过度嵌套：2-3 层就足够了
- 子图状态独立：避免与父图共享 state 字段名冲突
- 子图也应有 Checkpointer：如果子图需要中断或记忆
- 命名空间：子图的 checkpoint 会带命名空间前缀，便于区分