14. How to convert input and output of subgraph

subgraph state end parent graph It can be completely independent of the state.

In other words, there may not be a state keys that overlap between the two graphs.

In this case subgraph You need to convert the input before calling, and the output before returning.

Preferences

Copy

# Configuration file for managing API keys as environment variables
from dotenv import load_dotenv

# Load API key information
load_dotenv()

Copy

 True 

Copy

# Set up LangSmith tracking. https://smith.langchain.com
# !pip install -qU langchain-teddynote
from langchain_teddynote import logging

# Enter a project name.
logging.langsmith("CH17-LangGraph-Modules")

Copy

 Start tracking LangSmith. 
[Project name] 
CH17-LangGraph-Modules 

graph Wow subgraph Justice

Three as follows graph Let me define.

• parent graph

• parent graph To be called by child subgraph

• child graph To be called by grandchild subgraph

grandchild Justice

Copy

# For state management TypedDict와 StateGraph 관련 모듈 임포트
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START, END


# TypedDict class defining the state of the grandchild node, containing my_grandchild_key of type string
class GrandChildState(TypedDict):
    my_grandchild_key: str


# Function to handle the state of the grandchild node, adding a greeting to the input string
def grandchild_1(state: GrandChildState) -> GrandChildState:
    # Child or parent keys are not accessible here
    return {"my_grandchild_key": f'([GrandChild] {state["my_grandchild_key"]})'}


# Initialize the state graph of the grandchild node
grandchild = StateGraph(GrandChildState)

# Adding a grandchild node to the state graph
grandchild.add_node("grandchild_1", grandchild_1)

# Edge connection from start node to grandchild node
grandchild.add_edge(START, "grandchild_1")

# Edge connection from grandchild node to end node
grandchild.add_edge("grandchild_1", END)

# Compile the defined state graph and generate an executable graph
grandchild_graph = grandchild.compile()

Visualize the graph.

Copy

from langchain_teddynote.graphs import visualize_graph

visualize_graph(grandchild_graph, xray=True)

Copy

# Graph Call
for chunk in grandchild_graph.stream(
    {"my_grandchild_key": "Hi, Teddy!"}, subgraphs=True
):
    print(chunk)

Copy

 ((), {'grandchild_1': {'my_grandchild_key':'([GrandChild] Hi, Teddy!)'}})

child Justice

Copy

# TypedDict class for defining child state types
class ChildState(TypedDict):
    my_child_key: str


# Call the grandchild graph and state transformation function, take child state as input and return transformed child state
def call_grandchild_graph(state: ChildState) -> ChildState:
    # NOTE: Parent or grandchild keys are not accessible here
    # Transitioning state from child state channel to grandchild state channel
    grandchild_graph_input = {"my_grandchild_key": state["my_child_key"]}
    # Return the result after changing the state from the grandchild state channel to the child state channel.
    grandchild_graph_output = grandchild_graph.invoke(grandchild_graph_input)
    return {"my_child_key": f'([Child] {grandchild_graph_output["my_grandchild_key"]})'}


# Initialize child state graph
child = StateGraph(ChildState)
# Note: Passing a function instead of a compiled graph
# Add nodes to child graphs and connect start-end edges
child.add_node("child_1", call_grandchild_graph)
child.add_edge(START, "child_1")
child.add_edge("child_1", END)
# Compile child graph
child_graph = child.compile()

Copy

visualize_graph(child_graph, xray=True)

Copy

# Calling the child_graph graph
for chunk in child_graph.stream({"my_child_key": "Hi, Teddy!"}, subgraphs=True):
    print(chunk)

Copy

 (('child_1:f1df731d-8121-23c6-3999-f47dc5dab794',), {'grandchild_1': {'my_grandchild_key':'([GrandChild] Hi, Teddy!)'}} 
((), {'child_1': {'my_child_key':'([Child] ([GrandChild] Hi, Teddy!))'}}) 

grandchild_graph Separate function of call call_grandchild_graph ) Is wrapped.

This function converts the input state before calling the grandchild graph, and converts the output of the grandchild graph back to the child graph state.

If there is no such conversion grandchild_graph directly .add_node If you pass it to, LangGraph will get an error because there is no state key shared between the child and grandchild states.

important

child subgraph Wow grandchild subgraph has parent graph Your own not shared with Independent state Please note that you have.

parent Justice

Copy

# TypedDict class for defining parent state types
class ParentState(TypedDict):
    my_parent_key: str


# A transformation function that adds the string '[Parent1]' to the my_parent_key value in the parent state.
def parent_1(state: ParentState) -> ParentState:
    # NOTE: Child or grandchild keys are not accessible here.
    return {"my_parent_key": f'([Parent1] {state["my_parent_key"]})'}


# A transformation function that adds the string '[Parent2]' to the my_parent_key value in the parent state.
def parent_2(state: ParentState) -> ParentState:
    return {"my_parent_key": f'([Parent2] {state["my_parent_key"]})'}


# Transforming data between parent and child states and handling child graph calls
def call_child_graph(state: ParentState) -> ParentState:
    # Transform state from parent state channel (my_parent_key) to child state channel (my_child_key)
    child_graph_input = {"my_child_key": state["my_parent_key"]}
    # Transform state from child state channel (my_child_key) to parent state channel (my_parent_key)
    child_graph_output = child_graph.invoke(child_graph_input)
    return {"my_parent_key": child_graph_output["my_child_key"]}


# Initialize parent state graph and configure nodes
parent = StateGraph(ParentState)
parent.add_node("parent_1", parent_1)

# Note: Passing a function rather than a compiled graph
parent.add_node("child", call_child_graph)
parent.add_node("parent_2", parent_2)

# Edge configuration that defines the execution flow of the state graph
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()

Visualize the graph.

Copy

visualize_graph(parent_graph, xray=True)

child_graph Separate function of call call_child_graph Wrapped in, this function converts the input state before calling the child graph and converts the output of the child graph back to the parent graph state.

Without conversion child_graph directly .add_node If you pass it to, LangGraph will get an error because there is no State Key shared between parent and child status. So, I'll run the parent graph to see if the child and grandson subgraphs are called correctly.

Copy

# Run the graph and pass the value "Hi, Teddy!" via the "my_parent_key" parameter for chunk in parent_graph.stream({"my_parent_key": "Hi, Teddy!"}, subgraphs=True): print(chunk)

Copy

 ((), {'parent_1': {'my_parent_key':'([Parent1] Hi, Teddy!)'}}) 
('child:0d27b603-0493-c89d-01e3-391d125005cf','child_1:46cc6195-7591-1c11-974a-f69882aad590'), {'grandchild_ 
(('child:0d27b603-0493-c89d-01e3-391d125005cf',), {'child_1': {'my_child_key':'([Child] ([GrandChild] ([Parent1] Hi, Ted> 
((), {'child': {'my_parent_key':'([Child] ([GrandChild] ([Parent1] Hi, Teddy!)))'}}) 
((), {'parent_2': {'my_parent_key':'([Parent2] ([Child] ([GrandChild] ([Parent1] Hi, Teddy!)))'}})