subgraphstate 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
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
Visualize the graph.
Copy
Copy
Copy
child Justice
Copy
Copy
Copy
Copy
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
Visualize the graph.
Copy
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.
# 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()
from langchain_teddynote.graphs import visualize_graph
visualize_graph(grandchild_graph, xray=True)
# Graph Call
for chunk in grandchild_graph.stream(
{"my_grandchild_key": "Hi, Teddy!"}, subgraphs=True
):
print(chunk)
# 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()
visualize_graph(child_graph, xray=True)
# Calling the child_graph graph
for chunk in child_graph.stream({"my_child_key": "Hi, Teddy!"}, subgraphs=True):
print(chunk)
# 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_graph(parent_graph, xray=True)
# 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
Copy