Alpha Notice: These docs cover the v1-alpha release. Content is incomplete and subject to change.For the latest stable version, see the current LangGraph Python or LangGraph JavaScript docs.
This guide explains the mechanics of using subgraphs. A common application of subgraphs is to build multi-agent systems.When adding subgraphs, you need to define how the parent graph and the subgraph communicate:
Set up LangSmith for LangGraph development
Sign up for LangSmith to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM apps built with LangGraph — read more about how to get started here.
A common case is for the parent graph and subgraph to communicate over a shared state key (channel) in the schema. For example, in multi-agent systems, the agents often communicate over a shared messages key.If your subgraph shares state keys with the parent graph, you can follow these steps to add it to your graph:
Define the subgraph workflow (subgraph_builder in the example below) and compile it
Pass compiled subgraph to the .add_node method when defining the parent graph workflow
from typing_extensions import TypedDictfrom langgraph.graph.state import StateGraph, START# Define subgraphclass SubgraphState(TypedDict): foo: str # (1)! bar: str # (2)!def subgraph_node_1(state: SubgraphState): return {"bar": "bar"}def subgraph_node_2(state: SubgraphState): # note that this node is using a state key ('bar') that is only available in the subgraph # and is sending update on the shared state key ('foo') return {"foo": state["foo"] + state["bar"]}subgraph_builder = StateGraph(SubgraphState)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()# Define parent graphclass ParentState(TypedDict): foo: strdef node_1(state: ParentState): return {"foo": "hi! " + state["foo"]}builder = StateGraph(ParentState)builder.add_node("node_1", node_1)builder.add_node("node_2", subgraph)builder.add_edge(START, "node_1")builder.add_edge("node_1", "node_2")graph = builder.compile()for chunk in graph.stream({"foo": "foo"}): print(chunk)
This key is shared with the parent graph state
This key is private to the SubgraphState and is not visible to the parent graph
For more complex systems you might want to define subgraphs that have a completely different schema from the parent graph (no shared keys). For example, you might want to keep a private message history for each of the agents in a multi-agent system.If that’s the case for your application, you need to define a node function that invokes the subgraph. This function needs to transform the input (parent) state to the subgraph state before invoking the subgraph, and transform the results back to the parent state before returning the state update from the node.
We’re transforming the state from the child state channels (my_child_key) to the child state channels (my_grandchild_key)
We’re transforming the state from the grandchild state channels (my_grandchild_key) back to the child state channels (my_child_key)
We’re passing a function here instead of just compiled graph (grandchild_graph)
We’re transforming the state from the parent state channels (my_key) to the child state channels (my_child_key)
We’re transforming the state from the child state channels (my_child_key) back to the parent state channels (my_key)
We’re passing a function here instead of just a compiled graph (child_graph)
Copy
((), {'parent_1': {'my_key': 'hi Bob'}})(('child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b', 'child_1:781bb3b1-3971-84ce-810b-acf819a03f9c'), {'grandchild_1': {'my_grandchild_key': 'hi Bob, how are you'}})(('child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b',), {'child_1': {'my_child_key': 'hi Bob, how are you today?'}})((), {'child': {'my_key': 'hi Bob, how are you today?'}})((), {'parent_2': {'my_key': 'hi Bob, how are you today? bye!'}})
You only need to provide the checkpointer when compiling the parent graph. LangGraph will automatically propagate the checkpointer to the child subgraphs.
If you want the subgraph to have its own memory, you can compile it with the appropriate checkpointer option. This is useful in multi-agent systems, if you want agents to keep track of their internal message histories:
When you enable persistence, you can inspect the graph state (checkpoint) via the appropriate method. To view the subgraph state, you can use the subgraphs option.You can inspect the graph state via graph.get_state(config). To view the subgraph state, you can use graph.get_state(config, subgraphs=True).
Available only when interrupted
Subgraph state can only be viewed when the subgraph is interrupted. Once you resume the graph, you won’t be able to access the subgraph state.
To include outputs from subgraphs in the streamed outputs, you can set the subgraphs option in the stream method of the parent graph. This will stream outputs from both the parent graph and any subgraphs.
Copy
for chunk in graph.stream( {"foo": "foo"}, subgraphs=True, # (1)! stream_mode="updates",): print(chunk)
Set subgraphs=True to stream outputs from subgraphs.
Stream from subgraphs
Copy
from typing_extensions import TypedDictfrom langgraph.graph.state import StateGraph, START# Define subgraphclass SubgraphState(TypedDict): foo: str bar: strdef subgraph_node_1(state: SubgraphState): return {"bar": "bar"}def subgraph_node_2(state: SubgraphState): # note that this node is using a state key ('bar') that is only available in the subgraph # and is sending update on the shared state key ('foo') return {"foo": state["foo"] + state["bar"]}subgraph_builder = StateGraph(SubgraphState)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()# Define parent graphclass ParentState(TypedDict): foo: strdef node_1(state: ParentState): return {"foo": "hi! " + state["foo"]}builder = StateGraph(ParentState)builder.add_node("node_1", node_1)builder.add_node("node_2", subgraph)builder.add_edge(START, "node_1")builder.add_edge("node_1", "node_2")graph = builder.compile()for chunk in graph.stream( {"foo": "foo"}, stream_mode="updates", subgraphs=True, # (1)!): print(chunk)
Set subgraphs=True to stream outputs from subgraphs.