子图是一个 ,它被用作另一个图中的一个 节点——这是封装概念在 LangGraph 中的应用。子图允许您构建包含多个组件(这些组件本身也是图)的复杂系统。

Subgraph

使用子图的一些原因是:

  • 构建 多代理系统
  • 当您想在多个图中重用一组节点时
  • 当您希望不同团队独立处理图的不同部分时,您可以将每个部分定义为子图;只要子图接口(输入和输出模式)得到遵守,父图就可以在不了解子图任何细节的情况下构建。

添加子图时的主要问题是父图和子图如何通信,即它们在图执行期间如何相互传递 状态。有两种情况:

  • 父图和子图在它们的状态 模式 中具有 共享状态键。在这种情况下,您可以 将子图作为父图中的一个节点包含

    <span id="__span-0-1">from langgraph.graph import StateGraph, MessagesState, START
    <span id="__span-0-2">
    <span id="__span-0-3"># Subgraph
    <span id="__span-0-4">
    <span id="__span-0-5">def call_model(state: MessagesState):
    <span id="__span-0-6">    response = model.invoke(state["messages"])
    <span id="__span-0-7">    return {"messages": response}
    <span id="__span-0-8">
    <span id="__span-0-9">subgraph_builder = StateGraph(State)
    <span id="__span-0-10">subgraph_builder.add_node(call_model)
    <span id="__span-0-11">...
    <span id="__span-0-12">subgraph = subgraph_builder.compile()
    <span id="__span-0-13">
    <span id="__span-0-14"># Parent graph
    <span id="__span-0-15">
    <span id="__span-0-16">builder = StateGraph(State)
    <span id="__span-0-17">builder.add_node("subgraph_node", subgraph)
    <span id="__span-0-18">builder.add_edge(START, "subgraph_node")
    <span id="__span-0-19">graph = builder.compile()
    <span id="__span-0-20">...
    <span id="__span-0-21">graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})
    
  • 父图和子图具有 不同的模式(它们的状态 模式 中没有共享状态键)。在这种情况下,您必须 从父图中的一个节点内部调用子图:当父图和子图具有不同的状态模式,并且您需要在调用子图之前或之后转换状态时,这会很有用。

    <span id="__span-1-1">from typing_extensions import TypedDict, Annotated
    <span id="__span-1-2">from langchain_core.messages import AnyMessage
    <span id="__span-1-3">from langgraph.graph import StateGraph, MessagesState, START
    <span id="__span-1-4">from langgraph.graph.message import add_messages
    <span id="__span-1-5">
    <span id="__span-1-6">class SubgraphMessagesState(TypedDict):
    <span id="__span-1-7">    subgraph_messages: Annotated[list[AnyMessage], add_messages]
    <span id="__span-1-8">
    <span id="__span-1-9"># Subgraph
    <span id="__span-1-10">
    <span id="__span-1-11">def call_model(state: SubgraphMessagesState):
    <span id="__span-1-12">    response = model.invoke(state["subgraph_messages"])
    <span id="__span-1-13">    return {"subgraph_messages": response}
    <span id="__span-1-14">
    <span id="__span-1-15">subgraph_builder = StateGraph(State)
    <span id="__span-1-16">subgraph_builder.add_node(call_model)
    <span id="__span-1-17">...
    <span id="__span-1-18">subgraph = subgraph_builder.compile()
    <span id="__span-1-19">
    <span id="__span-1-20"># Parent graph
    <span id="__span-1-21">
    <span id="__span-1-22">def call_subgraph(state: MessagesState):
    <span id="__span-1-23">    response = subgraph.invoke({"subgraph_messages": state["messages"]})
    <span id="__span-1-24">    return {"messages": response["subgraph_messages"]}
    <span id="__span-1-25">
    <span id="__span-1-26">builder = StateGraph(State)
    <span id="__span-1-27">builder.add_node("subgraph_node", call_subgraph)
    <span id="__span-1-28">builder.add_edge(START, "subgraph_node")
    <span id="__span-1-29">graph = builder.compile()
    <span id="__span-1-30">...
    <span id="__span-1-31">graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})