子图¶
子图是在另一个图中作为节点使用的一个图——这是将封装(encapsulation)概念应用到 LangGraph 的方式。子图使你能够构建由多个组件组成的复杂系统,而这些组件本身也是图。
使用子图的一些理由包括:
- 构建多智能体系统
- 当你希望在多个图中复用一组节点时
- 当你希望不同团队独立负责图的不同部分时,你可以将每一部分定义为一个子图;只要遵循子图的接口(输入和输出模式),父图就无需了解子图的任何细节即可构建。
在添加子图时,核心问题是父图与子图如何通信,即在图执行过程中彼此如何传递状态。通常有两种情形:
- 父图与子图在各自的状态模式中具有**共享的状态键**。在这种情况下,你可以将子图作为父图中的一个节点直接包含
from langgraph.graph import StateGraph, MessagesState, START
# 子图
def call_model(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": response}
subgraph_builder = StateGraph(State)
subgraph_builder.add_node(call_model)
...
subgraph = subgraph_builder.compile()
# 父图
builder = StateGraph(State)
builder.add_node("subgraph_node", subgraph)
builder.add_edge(START, "subgraph_node")
graph = builder.compile()
...
graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})
- 父图与子图具有**不同的模式**(它们的状态模式中没有共享的状态键)。在这种情况下,你需要在父图的某个节点内部调用子图:当父图与子图的状态模式不同,且你需要在调用子图之前或之后对状态进行转换时,这种方式非常有用。
from typing_extensions import TypedDict, Annotated
from langchain_core.messages import AnyMessage
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.graph.message import add_messages
class SubgraphMessagesState(TypedDict):
subgraph_messages: Annotated[list[AnyMessage], add_messages]
# 子图
def call_model(state: SubgraphMessagesState):
response = model.invoke(state["subgraph_messages"])
return {"subgraph_messages": response}
subgraph_builder = StateGraph(SubgraphMessagesState)
subgraph_builder.add_node("call_model_from_subgraph", call_model)
subgraph_builder.add_edge(START, "call_model_from_subgraph")
...
subgraph = subgraph_builder.compile()
# 父图
def call_subgraph(state: MessagesState):
response = subgraph.invoke({"subgraph_messages": state["messages"]})
return {"messages": response["subgraph_messages"]}
builder = StateGraph(State)
builder.add_node("subgraph_node", call_subgraph)
builder.add_edge(START, "subgraph_node")
graph = builder.compile()
...
graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})