跳到内容

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

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

在这种情况下，您需要在调用子图之前转换其输入，然后在返回之前转换其输出。本指南展示了如何做到这一点。

安装

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

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

设置 LangSmith 以进行 LangGraph 开发

注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序 — 在此处阅读有关如何开始使用的更多信息 here。

定义图和子图

让我们定义 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 参考: 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”状态值中）。

评论