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

langgraph-subgraphs