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

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

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

准备工作¶

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

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

为 LangGraph 开发设置 LangSmith

注册 LangSmith，以便快速发现问题并提升你的 LangGraph 项目的性能。LangSmith 允许你使用跟踪数据来调试、测试和监控使用 LangGraph 构建的大语言模型应用程序 — 点击此处了解更多关于如何开始使用的信息。

定义图和子图¶

让我们定义 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()

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!'}

完美！父图正确调用了子图和孙图（我们之所以知道这一点，是因为 “, how are you” 和 “today?” 被添加到了我们原始的 “my_key” 状态值中）。