Last Updated: 3/11/2026
Subgraphs
Use subgraphs to build multi-agent systems and reuse workflow components.
What Are Subgraphs?
A subgraph is a LangGraph graph used as a node inside another graph. Subgraphs enable:
- Multi-agent systems: Each agent is a subgraph with its own logic
- Component reuse: Define once, use in multiple graphs
- Team collaboration: Different teams build different subgraphs independently
Two Integration Patterns
Pattern 1: Call Inside a Node
Use when parent and subgraph have different state schemas.
Python:
from langgraph.graph import StateGraph, START
from typing_extensions import TypedDict
# Subgraph with its own state
class SubgraphState(TypedDict):
bar: str
def subgraph_node(state: SubgraphState):
return {"bar": "processed: " + state["bar"]}
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node("process", subgraph_node)
subgraph_builder.add_edge(START, "process")
subgraph = subgraph_builder.compile()
# Parent graph with different state
class ParentState(TypedDict):
foo: str
def call_subgraph(state: ParentState):
# Transform parent state → subgraph input
result = subgraph.invoke({"bar": state["foo"]})
# Transform subgraph output → parent state
return {"foo": result["bar"]}
parent = StateGraph(ParentState)
parent.add_node("delegate", call_subgraph)
parent.add_edge(START, "delegate")
app = parent.compile()JavaScript:
import { StateGraph, StateSchema, START } from "@langchain/langgraph";
import { z } from "zod";
const SubgraphState = new StateSchema({ bar: z.string() });
const subgraph = new StateGraph(SubgraphState)
.addNode("process", (state) => ({ bar: "processed: " + state.bar }))
.addEdge(START, "process")
.compile();
const ParentState = new StateSchema({ foo: z.string() });
const app = new StateGraph(ParentState)
.addNode("delegate", async (state) => {
const result = await subgraph.invoke({ bar: state.foo });
return { foo: result.bar };
})
.addEdge(START, "delegate")
.compile();Pattern 2: Add as a Node
Use when parent and subgraph share state keys.
Python:
class SharedState(TypedDict):
messages: list[str]
# Subgraph reads/writes messages
subgraph_builder = StateGraph(SharedState)
subgraph_builder.add_node("agent", agent_node)
subgraph_builder.add_edge(START, "agent")
subgraph = subgraph_builder.compile()
# Parent directly adds compiled subgraph
parent = StateGraph(SharedState)
parent.add_node("specialist", subgraph) # No wrapper needed
parent.add_edge(START, "specialist")
app = parent.compile()Subgraph Persistence
Control how subgraph state persists between calls:
| Mode | Setting | Behavior |
|---|---|---|
| Per-invocation | checkpointer=None (default) | Each call starts fresh, inherits parent checkpointer |
| Per-thread | checkpointer=True | State accumulates across calls on same thread |
| Stateless | checkpointer=False | No checkpointing, runs like plain function |
Per-Invocation (Default)
Best for most multi-agent systems. Each invocation is independent:
# Subagent - no checkpointer specified
fruit_agent = create_agent(
model="gpt-4o-mini",
tools=[fruit_info],
prompt="You are a fruit expert."
)
# Parent with checkpointer
agent = create_agent(
model="gpt-4o-mini",
tools=[ask_fruit_expert],
checkpointer=MemorySaver()
)
config = {"configurable": {"thread_id": "1"}}
# Each call to fruit_agent starts fresh
agent.invoke({"messages": [{"role": "user", "content": "Tell me about apples"}]}, config)
agent.invoke({"messages": [{"role": "user", "content": "Now bananas"}]}, config)
# Fruit agent doesn't remember applesPer-Thread
Use when subagent needs conversation memory:
# Subagent with its own persistent state
research_agent = create_agent(
model="gpt-4o-mini",
tools=[search_tool],
checkpointer=True # Remembers across calls
)
config = {"configurable": {"thread_id": "1"}}
# First call
agent.invoke({"messages": [{"role": "user", "content": "Research topic A"}]}, config)
# Second call - research_agent remembers topic A
agent.invoke({"messages": [{"role": "user", "content": "Continue with topic B"}]}, config)<Warning> Per-thread subgraphs don’t support parallel tool calls. Use middleware to prevent parallel calls to the same per-thread subagent. </Warning>
View Subgraph State
Inspect subgraph state with get_state(subgraphs=True):
Python:
state = app.get_state(config, subgraphs=True)
subgraph_state = state.tasks[0].state
print(subgraph_state.values)JavaScript:
const state = await app.getState(config, { subgraphs: true });
const subgraphState = state.tasks[0].state;
console.log(subgraphState.values);Stream from Subgraphs
Include subgraph outputs in streaming:
Python:
for chunk in app.stream(
inputs,
stream_mode="updates",
subgraphs=True,
version="v2"
):
if chunk["ns"] == ():
print(f"Parent: {chunk['data']}")
else:
print(f"Subgraph {chunk['ns']}: {chunk['data']}")JavaScript:
for await (const [ns, data] of await app.stream(
inputs,
{ streamMode: "updates", subgraphs: true }
)) {
if (ns.length === 0) {
console.log(`Parent: ${data}`);
} else {
console.log(`Subgraph ${ns}: ${data}`);
}
}Best Practices
- Use per-invocation for stateless agents: Most multi-agent patterns
- Use per-thread for conversational agents: When agent needs memory
- Share keys when possible: Reduces transformation code
- Namespace isolation: Wrap per-thread subagents in unique nodes
- Test independently: Each subgraph should work standalone
Next Steps
- Time Travel: Debug and replay subgraph executions
- Multi-Agent Systems: Build complex agent collaborations
- Graph API Reference: Complete subgraph API details