Skip to main content
子图是一个,它被用作另一个图中的节点 —— 这是将封装概念应用于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!"}]})