LangGraph Subgraphs
Compose modular, reusable workflow components with nested graphs.
Two Primary Patterns
Pattern 1: Invoke from Node (Different Schemas)
Use when subgraph needs completely isolated state.
from langgraph.graph import StateGraph, START, END
Parent state
class ParentState(TypedDict): query: str analysis_result: dict
Subgraph state (completely different)
class AnalysisState(TypedDict): input_text: str findings: list[str] score: float
Build subgraph
analysis_builder = StateGraph(AnalysisState) analysis_builder.add_node("analyze", analyze_node) analysis_builder.add_node("score", score_node) analysis_builder.add_edge(START, "analyze") analysis_builder.add_edge("analyze", "score") analysis_builder.add_edge("score", END) analysis_subgraph = analysis_builder.compile()
Parent node that invokes subgraph
def call_analysis(state: ParentState) -> dict: """Transform state at boundaries.""" # Map parent → subgraph state subgraph_input = {"input_text": state["query"], "findings": [], "score": 0.0}
# Invoke subgraph
subgraph_output = analysis_subgraph.invoke(subgraph_input)
# Map subgraph → parent state
return {
"analysis_result": {
"findings": subgraph_output["findings"],
"score": subgraph_output["score"]
}
}
Add to parent graph
parent_builder = StateGraph(ParentState) parent_builder.add_node("analysis", call_analysis)
Pattern 2: Add as Node (Shared State)
Use when parent and subgraph share state keys.
from langgraph.graph.message import add_messages
Shared state with messages channel
class SharedState(TypedDict): messages: Annotated[list, add_messages] context: dict
Subgraph uses same state
agent_builder = StateGraph(SharedState) agent_builder.add_node("think", think_node) agent_builder.add_node("act", act_node) agent_builder.add_edge(START, "think") agent_builder.add_edge("think", "act") agent_builder.add_edge("act", END) agent_subgraph = agent_builder.compile()
Add compiled subgraph directly as node
parent_builder = StateGraph(SharedState) parent_builder.add_node("agent_team", agent_subgraph) # Direct embedding parent_builder.add_edge(START, "agent_team") parent_builder.add_edge("agent_team", END)
When to Use Each Pattern
Pattern Use When
Invoke Different schemas, private message histories, multi-level nesting
Add as Node Shared state keys, agent coordination, message passing
Multi-Level Nesting
Grandchild subgraph
grandchild = grandchild_builder.compile()
Child subgraph (contains grandchild)
def call_grandchild(state: ChildState): result = grandchild.invoke({"data": state["input"]}) return {"processed": result["output"]}
child_builder.add_node("processor", call_grandchild) child = child_builder.compile()
Parent (contains child)
def call_child(state: ParentState): result = child.invoke({"input": state["query"]}) return {"result": result["processed"]}
parent_builder.add_node("child_workflow", call_child)
Checkpointing Strategies
Parent-Only (Recommended)
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)
Checkpointer propagates to all subgraphs automatically
parent = parent_builder.compile(checkpointer=checkpointer)
Independent Subgraph Memory
Subgraph maintains its own checkpoint history
Useful for agent message histories that should persist independently
agent_subgraph = agent_builder.compile(checkpointer=True)
Parent with its own checkpointer
parent = parent_builder.compile(checkpointer=PostgresSaver(...))
State Mapping Best Practices
def call_subgraph_with_mapping(state: ParentState) -> dict: """Explicit state transformation at boundaries.""" # 1. Extract relevant data from parent subgraph_input = { "query": state["user_query"], "context": state.get("context", {}), "history": [] # Subgraph has own history }
# 2. Invoke with config propagation
config = get_runnable_config()
result = subgraph.invoke(subgraph_input, config)
# 3. Transform output back to parent schema
return {
"subgraph_result": result["output"],
"metadata": {
"subgraph": "analysis",
"steps": result.get("step_count", 0)
}
}
Streaming & Inspection
Stream with subgraph visibility
for namespace, chunk in graph.stream(inputs, subgraphs=True, stream_mode="updates"): depth = len(namespace) prefix = " " * depth print(f"{prefix}[{'/'.join(namespace) or 'root'}] {chunk}")
Inspect subgraph state (only works when interrupted)
config = {"configurable": {"thread_id": "thread-1"}} state = graph.get_state(config, subgraphs=True)
Access nested state
for subgraph_state in state.tasks: print(f"Subgraph: {subgraph_state.name}") print(f"State: {subgraph_state.state}")
Key Decisions
Decision Recommendation
Schema design Shared for coordination, isolated for encapsulation
Checkpointing Parent-only unless agents need independent history
State mapping Explicit transforms at boundaries for clarity
Team development Each team owns their subgraph with defined interface
Common Mistakes
-
Not transforming state at boundaries (schema mismatch errors)
-
Forgetting to propagate config for tracing/checkpointing
-
Using shared state when isolation is needed
-
Missing subgraphs=True when streaming nested graphs
Evaluations
See references/evaluations.md for test cases.
Related Skills
-
langgraph-streaming
-
Stream updates from subgraphs
-
langgraph-supervisor
-
Subgraphs as workers in supervisor patterns
-
langgraph-checkpoints
-
Cross-subgraph checkpointing strategies
-
langgraph-state
-
State schema mapping between graphs
-
langgraph-parallel
-
Parallel subgraph execution
-
langgraph-functional
-
Subgraphs with Functional API
Capability Details
invoke-pattern
Keywords: invoke, different schema, isolated state, transform Solves:
-
Embed graphs with different state schemas
-
Isolate subgraph state from parent
-
Transform data at graph boundaries
add-as-node-pattern
Keywords: add_node, shared state, messages, coordination Solves:
-
Embed graphs with shared state
-
Coordinate agents via message passing
-
Build multi-agent systems
nested-graphs
Keywords: nested, multi-level, parent, child, grandchild Solves:
-
Build deeply nested graph hierarchies
-
Compose complex workflows from simple parts
-
Implement recursive graph patterns
subgraph-checkpointing
Keywords: checkpoint, memory, independent, propagate Solves:
-
Configure checkpointing for subgraphs
-
Maintain independent agent histories
-
Handle persistence in nested structures