Skip to content

子图

子图是一个作为另一个图中节点使用的 —— 这是封装概念在LangGraph中的应用。子图允许你构建由多个自身为图的组件组成的复杂系统。

Subgraph

使用子图的一些原因包括:

  • 构建多智能体系统
  • 当你希望在多个图中重用一组节点时
  • 当你希望不同的团队可以独立地对图的不同部分进行工作时,你可以将每个部分定义为一个子图,只要子图接口(输入和输出模式)被遵守,父图可以在不知道子图任何细节的情况下构建

添加子图时的主要问题是父图和子图如何通信,即它们在图执行过程中如何传递状态。有两种场景:

  • 父图和子图在其状态schema中具有**共享的状态键**。在这种情况下,你可以将子图作为父图中的一个节点包含进来

    from langgraph.graph import StateGraph, MessagesState, START
    
    # Subgraph
    
    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()
    
    # Parent graph
    
    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!"}]})
    
  • 父图和子图具有**不同的 schema**(它们的状态schema中没有共享的状态键)。在这种情况下,你必须从父图中的一个节点内部调用子图:当父图和子图有不同状态 schema,并且你需要在调用子图之前或之后转换状态时,这很有用。

    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]
    
    # Subgraph
    
    def call_model(state: SubgraphMessagesState):
        response = model.invoke(state["subgraph_messages"])
        return {"subgraph_messages": response}
    
    subgraph_builder = StateGraph(State)
    subgraph_builder.add_node(call_model)
    ...
    subgraph = subgraph_builder.compile()
    
    # Parent graph
    
    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!"}]})