如何转换子图的输入和输出¶

有可能你的子图状态与父图状态完全独立，即两者之间没有重叠的通道（键）。例如，你可能有一个监督代理，它需要借助多个ReAct代理来生成报告。ReAct代理子图可能跟踪消息列表，而监督代理只需要用户输入和最终报告，并不需要跟踪消息。

在这种情况下，你需要在调用子图之前转换其输入，并在返回之前转换其输出。本指南将展示如何进行这些转换。

环境搭建¶

首先，让我们安装所需的包

%%capture --no-stderr
%pip install -U langgraph

为LangGraph开发设置LangSmith

注册LangSmith，可以快速发现并解决您的LangGraph项目中的问题，提高项目性能。LangSmith允许您使用跟踪数据来调试、测试和监控使用LangGraph构建的LLM应用程序——了解如何开始使用请参阅此处。

定义图和子图¶

让我们定义3个图： - 一个父图 - 一个由父图调用的子图 - 一个由子图调用的孙图

定义孙元素¶

from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START, END


class GrandChildState(TypedDict):
    my_grandchild_key: str


def grandchild_1(state: GrandChildState) -> GrandChildState:
    # NOTE: child or parent keys will not be accessible here
    return {"my_grandchild_key": state["my_grandchild_key"] + ", how are you"}


grandchild = StateGraph(GrandChildState)
grandchild.add_node("grandchild_1", grandchild_1)

grandchild.add_edge(START, "grandchild_1")
grandchild.add_edge("grandchild_1", END)

grandchild_graph = grandchild.compile()

API Reference: StateGraph

grandchild_graph.invoke({"my_grandchild_key": "hi Bob"})

{'my_grandchild_key': 'hi Bob, how are you'}

定义子元素¶

class ChildState(TypedDict):
    my_child_key: str


def call_grandchild_graph(state: ChildState) -> ChildState:
    # NOTE: parent or grandchild keys won't be accessible here
    # we're transforming the state from the child state channels (`my_child_key`)
    # to the child state channels (`my_grandchild_key`)
    grandchild_graph_input = {"my_grandchild_key": state["my_child_key"]}
    # we're transforming the state from the grandchild state channels (`my_grandchild_key`)
    # back to the child state channels (`my_child_key`)
    grandchild_graph_output = grandchild_graph.invoke(grandchild_graph_input)
    return {"my_child_key": grandchild_graph_output["my_grandchild_key"] + " today?"}


child = StateGraph(ChildState)
# NOTE: we're passing a function here instead of just compiled graph (`child_graph`)
child.add_node("child_1", call_grandchild_graph)
child.add_edge(START, "child_1")
child.add_edge("child_1", END)
child_graph = child.compile()

child_graph.invoke({"my_child_key": "hi Bob"})

{'my_child_key': 'hi Bob, how are you today?'}

注意

我们将grandchild_graph调用包装在一个单独的函数（call_grandchild_graph）中，该函数在调用孙子图之前转换输入状态，并在调用孙子图之后将输出转换回子图状态。如果你直接将grandchild_graph传递给.add_node而不进行转换，LangGraph将抛出一个错误，因为子图状态和孙子图状态之间没有共享状态通道（键）。

需要注意的是，子图和孙图子图具有自己的、**独立**的状态，该状态不会与父图共享。

定义父元素¶

class ParentState(TypedDict):
    my_key: str


def parent_1(state: ParentState) -> ParentState:
    # NOTE: child or grandchild keys won't be accessible here
    return {"my_key": "hi " + state["my_key"]}


def parent_2(state: ParentState) -> ParentState:
    return {"my_key": state["my_key"] + " bye!"}


def call_child_graph(state: ParentState) -> ParentState:
    # we're transforming the state from the parent state channels (`my_key`)
    # to the child state channels (`my_child_key`)
    child_graph_input = {"my_child_key": state["my_key"]}
    # we're transforming the state from the child state channels (`my_child_key`)
    # back to the parent state channels (`my_key`)
    child_graph_output = child_graph.invoke(child_graph_input)
    return {"my_key": child_graph_output["my_child_key"]}


parent = StateGraph(ParentState)
parent.add_node("parent_1", parent_1)
# NOTE: we're passing a function here instead of just a compiled graph (`<code>child_graph</code>`)
parent.add_node("child", call_child_graph)
parent.add_node("parent_2", parent_2)

parent.add_edge(START, "parent_1")
parent.add_edge("parent_1", "child")
parent.add_edge("child", "parent_2")
parent.add_edge("parent_2", END)

parent_graph = parent.compile()

注意

我们将child_graph的调用包装在一个单独的函数call_child_graph中，该函数在调用子图之前转换输入状态，并在调用子图后将子图的输出转换回父图状态。如果你直接将child_graph传递给.add_node而不进行转换，LangGraph将抛出一个错误，因为父图和子图之间没有共享状态通道（键）。

让我们运行父图，并确保它正确调用了子图和孙图子图：

parent_graph.invoke({"my_key": "Bob"})

{'my_key': 'hi Bob, how are you today? bye!'}

完美！父图正确地调用了子图和孙图（我们可以通过原始“my_key”状态值中添加了“，你好吗”和“今天？”来确认这一点）。