SubGraph allows you to build complex systems that contain multiple components, and these components themselves can be graphs. A common use example of SubGraph is building a multi-agent system.
When adding SubGraph, the main consideration is how the parent graph and SubGraph communicate, i.e. how they pass the state (State) to each other during graph execution.
There are two scenarios:
Top graph and subgraph Share schema keys If you do. In this case Add nodes to compiled subgraphs can do
Top graph and subgraph Different schemas If you have. In this case Add node function to call subgraph Should do.
This is useful when the parent graph and subgraph have different state schemas and need to convert states before and after calling the subgraph.
Below we will show you how to add a subgraph for each scenario.
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
Case 1: If you share a schema key
Add nodes with compiled SubGraph
This is a common case when parent graphs and subgraphs communicate through the State Key.
For example, Multi agent system In agents are mainly shared messages Communicate via key.
If the subgraph shares the status key with the parent graph, you can add it to the graph by following the steps:
Define the subgraph workflow (of example below subgraph_builder ) Compilation
When defining the parent graph walkflow .add_node Passing the subgraph compiled in the method
So, let's look at the simple example below.
Copy
Visualize the graph.
Copy
Copy
Copy
Copy
Copy
The final output of the parent graph contains the results of the subgraph call.
At this time, when streaming to check the output of the subgraph subgraphs=True Specify.
Copy
Copy
Case 2: If you do not share the schema key
Add node function to call subgraph
For more complex systems, you may need to define a subgraph with a schema that is completely different from the parent graph (if there is no shared state key).
If this is the case, Node function to call subgraph You need to define.
This function needs to convert the parent state to the child state before calling the subgraph, and the result back to the parent state before returning the status update from the node.
Below shows how to modify the original example to call a subgraph inside the node. Reference
Two or more within the same node subgraph Number of calls There is no .
from langgraph.graph import START, END, StateGraph
from typing import TypedDict
# # TypedDict class for defining subgraph state, containing a name key shared with the parent graph and a family_name key exclusive to the subgraph.
class ChildState(TypedDict):
name: str # State keys shared with parent graph
family_name: str
# Set initial value for the first node of the subgraph, family_name key
def subgraph_node_1(state: ChildState):
return {"family_name": "Lee"}
# The second node of the subgraph creates a new state by combining the subgraph-specific family_name key and the shared name key.
def subgraph_node_2(state: ChildState):
# Perform an update using the family_name key and the shared state key name, which are only available inside the subgraph.
return {"name": f'{state["name"]} {state["family_name"]}'}
# Defining the subgraph structure and establishing the connection relationship between nodes.
subgraph_builder = StateGraph(ChildState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()
from langchain_teddynote.graphs import visualize_graph
visualize_graph(subgraph, xray=True)
# TypedDict class for defining the state of the parent graph, containing only the name key.
class ParentState(TypedDict):
name: str
company: str
# Create a new state by modifying the value of the name key in the first node of the parent graph.
def node_1(state: ParentState):
return {"name": f'My name is {state["name"]}'}
# Define the parent graph structure and establish the connection relationship between nodes including subgraphs.
builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
# Add the compiled subgraph as a node in the parent graph.
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)
graph = builder.compile()
from langchain_teddynote.graphs import visualize_graph
visualize_graph(graph, xray=True)
# Process data in chunks in a graph stream and output each chunk.
for chunk in graph.stream({"name": "Teddy"}):
print(chunk)
{'node_1': {'name':'My name is Teddy'}}
{'node_2': {'name':'My name is Teddy Lee'}}
# Sequential output of subgraph data chunks through graph streaming processing
# Stream processing with subgraphs included by setting the subgraphs parameter to True
for chunk in graph.stream({"name": "Teddy"}, subgraphs=True):
print(chunk)
((), {'node_1': {'name':'My name is Teddy'}})
(('node_2:d4bf5dab-e19f-2cd0-669b-63ac15c0ef77',), {'subgraph_node_1': {'family_name':'Lee'}})
(('node_2:d4bf5dab-e19f-2cd0-669b-63ac15c0ef77',), {'subgraph_node_2': {'name':'My name is Teddy Lee'}})
((), {'node_2': {'name':'My name is Teddy Lee'}})
# Define the state type of the subgraph (does not share keys with the parent graph)
class ChildState(TypedDict):
# Keys not shared with parent graph
name: str
# First node of the subgraph: set initial value for name key
def subgraph_node_1(state: ChildState):
return {"name": "Teddy " + state["name"]}
# Second node of the subgraph: return the name value as is
def subgraph_node_2(state: ChildState):
return {"name": f'My name is {state["name"]}'}
# Initializing the subgraph builder and configuring node connections
subgraph_builder = StateGraph(ChildState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()
# Defining the state type of the parent graph
class ParentState(TypedDict):
family_name: str
full_name: str
# First node of parent graph: returns the family_name value as is
def node_1(state: ParentState):
return {"family_name": state["family_name"]}
# Second node of parent graph: subgraph and state transition and result processing
def node_2(state: ParentState):
# Convert parent state to subgraph state
response = subgraph.invoke({"name": state["family_name"]})
# Convert subgraph response to parent state
return {"full_name": response["name"]}
# Initialize the parent graph builder and configure node connections.
builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
# Use node_2 function that calls subgraph instead of compiled subgraph
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)
graph = builder.compile()
from langchain_teddynote.graphs import visualize_graph
visualize_graph(graph, xray=True)
# Sequential output of subgraph data chunks through graph streaming processing
# Process stream data including subgraphs with the subgraphs=True option
for chunk in graph.stream({"family_name": "Lee"}, subgraphs=True):
print(chunk)
((), {'node_1': {'family_name':'Lee'}})
(('node_2:4a8b89a5-6875-5175-95b9-20986c84a7dd',), {'subgraph_node_1': {'name':'Teddy Lee'}})
(('node_2:4a8b89a5-6875-5175-95b9-20986c84a7dd',), {'subgraph_node_2': {'name':'My name is Teddy Lee'}})
((), {'node_2': {'full_name':'My name is Teddy Lee'}})
Copy
Copy
Copy