| name | langgraph-subgraphs |
| description | Build, compose, and configure LangGraph subgraphs as nodes in parent graphs. Use when embedding one graph inside another, managing subgraph persistence (per-invocation, per-thread, stateless), streaming subgraph outputs, or inspecting subgraph state. Covers state schema patterns, checkpointer modes, namespace isolation, and interrupt support. |
Documentation Index
Fetch the complete documentation index at: https://docs.langchain.com/llms.txt
Use this file to discover all available pages before exploring further.
LangGraph Subgraphs
A subgraph is a compiled StateGraph used as a node in a parent graph. Use for multi-agent systems, reusable node sets, or distributing development across teams.
Two Communication Patterns
| Pattern | When | How |
|---|
| Call inside a node | Different state schemas (no shared keys) | Write wrapper fn that maps parent↔subgraph state |
| Add as a node | Shared state keys | Pass compiled subgraph directly to add_node |
Call inside a node (different schemas)
from langgraph.graph.state import StateGraph, START
from typing_extensions import TypedDict
class SubgraphState(TypedDict):
bar: str
def subgraph_node(state: SubgraphState):
return {"bar": "hi! " + state["bar"]}
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node)
subgraph_builder.add_edge(START, "subgraph_node")
subgraph = subgraph_builder.compile()
class ParentState(TypedDict):
foo: str
def call_subgraph(state: ParentState):
output = subgraph.invoke({"bar": state["foo"]})
return {"foo": output["bar"]}
builder = StateGraph(ParentState)
builder.add_node("node_1", call_subgraph)
builder.add_edge(START, "node_1")
graph = builder.compile()
Add as a node (shared schema)
subgraph_builder = StateGraph(State)
subgraph = subgraph_builder.compile()
builder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")
graph = builder.compile()
Subgraph Persistence
The checkpointer parameter on .compile() controls how subgraph state is retained.
| Mode | checkpointer= | Interrupts | Multi-turn memory | Parallel calls |
|---|
| Per-invocation (default) | None | ✅ | ❌ | ✅ |
| Per-thread | True | ✅ | ✅ | ⚠️ no parallel same-subgraph |
| Stateless | False | ❌ | ❌ | ✅ |
Parent must also have a checkpointer for any stateful subgraph features to work.
Per-invocation (default — use for most multi-agent systems)
Each call starts fresh. Inherits parent checkpointer. Supports interrupt().
fruit_agent = create_agent(model="...", tools=[...], prompt="...")
@tool
def ask_fruit_expert(question: str) -> str:
result = fruit_agent.invoke({"messages": [{"role": "user", "content": question}]})
return result["messages"][-1].content
agent = create_agent(model="...", tools=[ask_fruit_expert], checkpointer=MemorySaver())
Per-thread (subagent remembers past calls)
State accumulates across invocations on the same thread.
fruit_agent = create_agent(
model="...", tools=[...], prompt="...",
checkpointer=True,
)
Warning: Per-thread subgraphs cannot be called in parallel (same subgraph, same thread). Use ToolCallLimitMiddleware to prevent the LLM from trying:
from langchain.agents.middleware import ToolCallLimitMiddleware
agent = create_agent(
model="...",
tools=[ask_fruit_expert],
middleware=[ToolCallLimitMiddleware(tool_name="ask_fruit_expert", run_limit=1)],
checkpointer=MemorySaver(),
)
Namespace isolation for multiple per-thread subgraphs:
When calling multiple different per-thread subgraphs from a node, wrap each in its own named StateGraph to ensure stable checkpoint namespaces:
from langgraph.graph import MessagesState, StateGraph
def create_sub_agent(model, *, name, **kwargs):
agent = create_agent(model=model, name=name, **kwargs)
return (
StateGraph(MessagesState)
.add_node(name, agent)
.add_edge("__start__", name)
.compile()
)
fruit_agent = create_sub_agent("...", name="fruit_agent", tools=[...], checkpointer=True)
veggie_agent = create_sub_agent("...", name="veggie_agent", tools=[...], checkpointer=True)
Subgraphs added as nodes (add_node) get name-based namespaces automatically — no wrapper needed.
Stateless
No checkpointing overhead. No interrupts or durable execution.
subgraph = builder.compile(checkpointer=False)
Viewing Subgraph State
Only works when the subgraph is statically discoverable (added as a node or called inside a node). Does not work when called inside a tool function (e.g., multi-agent subagents pattern).
subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state
subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state
Streaming Subgraph Outputs
for chunk in graph.stream(
{"foo": "foo"},
subgraphs=True,
stream_mode="updates",
version="v2",
):
print(chunk["ns"])
print(chunk["data"])
Key Rules
- Shared keys →
add_node directly; different schemas → wrap in a node function
- Default (
checkpointer=None) is right for most cases — fresh per call, supports interrupts
- Per-thread requires parent checkpointer and prevents parallel same-subgraph calls
- Use
ToolCallLimitMiddleware when exposing per-thread subagents as tools
- Namespace isolation — wrap per-thread subagents in named
StateGraph nodes when multiple are called from the same parent node
get_state(subgraphs=True) won't see tool-invoked subgraphs — only statically-added nodes are discoverable; interrupts still propagate regardless
Related Resources
- Multi-agent patterns:
/oss/python/langchain/multi-agent
- Persistence & checkpointers:
/oss/python/langgraph/persistence
- Interrupts / HITL:
/oss/python/langgraph/interrupts
- Graph API reference:
/oss/python/langgraph/graph-api