Composing Subgraphs in LangGraph

LangGraph subgraphs are graphs that you compile on their own and then plug into a parent graph as a single node. You reach for them the moment one graph starts doing too much: a research step, a tool-calling loop, and a summarization pass all crammed into one giant StateGraph with dozens of conditional edges. Splitting that into subgraphs gives you units you can test, reuse, and reason about independently, and it is the pattern LangGraph itself recommends once a graph crosses roughly eight to ten nodes. This article walks through building subgraphs, sharing state between parent and child, handling mismatched schemas, streaming output across graph boundaries, and debugging the whole thing when it does not behave.

What a Subgraph Actually Is

A subgraph in LangGraph is nothing exotic. It is a normal StateGraph that you call .compile() on, which turns it into a CompiledGraph. That compiled graph is a runnable, and any runnable can be added as a node to another graph with add_node. So the "subgraph" concept is really just composition: graphs are nodes, nodes can be graphs.

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class ChildState(TypedDict):
    topic: str
    draft: str

def write_draft(state: ChildState) -> ChildState:
    return {"draft": f"Draft about {state['topic']}"}

child_builder = StateGraph(ChildState)
child_builder.add_node("write_draft", write_draft)
child_builder.add_edge(START, "write_draft")
child_builder.add_edge("write_draft", END)
child_graph = child_builder.compile()

child_graph is now a fully functional graph you can invoke on its own with child_graph.invoke({"topic": "rate limiting"}). The interesting part starts when you wire it into something bigger.

Why Compose Subgraphs Instead of One Big Graph

Three reasons come up in practice, and all three matter more as a project grows past a prototype.

Isolation of state. A subgraph can have its own state schema that is completely separate from the parent's. A research subgraph might track search_results and citations internally, none of which the parent graph needs to know about. Only the fields that matter to the parent cross the boundary.

Reuse across graphs. If you build a "verify with a critic model" subgraph once, you can drop it into a writing pipeline, a coding agent, and a customer support bot without rewriting the logic. Each parent graph just needs to map its state onto the subgraph's expected input.

Independent testing. You can unit test a subgraph by calling .invoke() on it directly with a minimal input dict, without spinning up the rest of the system. This is the single biggest win for teams working on the same LangGraph app in parallel: the research team ships and tests their subgraph, the orchestration team ships and tests the parent, and they only need to agree on the shape of a few shared keys.

There is also a multi-agent framing worth naming directly: in LangGraph, subgraphs are the standard way to build hierarchical multi-agent systems, where a supervisor graph routes to specialized agent subgraphs (a coder, a researcher, a reviewer) that each encapsulate their own tool loop.

Building Your First Subgraph and Adding It to a Parent

The simplest case is when the parent and child graphs share state keys. In that case you can add the compiled subgraph directly as a node, and LangGraph handles passing state in and merging it back out.

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class SharedState(TypedDict):
    topic: str
    draft: str
    final: str

def write_draft(state: SharedState) -> SharedState:
    return {"draft": f"Draft about {state['topic']}"}

subgraph_builder = StateGraph(SharedState)
subgraph_builder.add_node("write_draft", write_draft)
subgraph_builder.add_edge(START, "write_draft")
subgraph_builder.add_edge("write_draft", END)
subgraph = subgraph_builder.compile()

def polish(state: SharedState) -> SharedState:
    return {"final": state["draft"].upper()}

parent_builder = StateGraph(SharedState)
parent_builder.add_node("drafting", subgraph)
parent_builder.add_node("polish", polish)
parent_builder.add_edge(START, "drafting")
parent_builder.add_edge("drafting", "polish")
parent_builder.add_edge("polish", END)
parent_graph = parent_builder.compile()

result = parent_graph.invoke({"topic": "vector databases", "draft": "", "final": ""})
print(result["final"])

Because SharedState is used by both graphs, LangGraph passes the parent's state straight into the subgraph and merges whatever the subgraph returns back into the parent's state. No adapter code needed. This is the pattern to reach for whenever the child logic genuinely operates on the same fields as the parent.

When Parent and Subgraph Have Different State Schemas

Sharing one big TypedDict across every subgraph defeats the purpose of splitting the graph in the first place. The more common (and more useful) case is a subgraph with its own private schema that only overlaps the parent on a few keys.

When schemas differ, you cannot add the compiled subgraph directly as a node, because LangGraph needs to know how to translate between the two shapes. The fix is to wrap the subgraph invocation in a regular Python function that does the translation explicitly.

from typing import TypedDict

class ParentState(TypedDict):
    user_question: str
    answer: str

class ResearchState(TypedDict):
    query: str
    search_results: list[str]
    summary: str

def search(state: ResearchState) -> ResearchState:
    results = [f"result for {state['query']}"]
    return {"search_results": results}

def summarize(state: ResearchState) -> ResearchState:
    joined = "; ".join(state["search_results"])
    return {"summary": f"Summary: {joined}"}

research_builder = StateGraph(ResearchState)
research_builder.add_node("search", search)
research_builder.add_node("summarize", summarize)
research_builder.add_edge(START, "search")
research_builder.add_edge("search", "summarize")
research_builder.add_edge("summarize", END)
research_subgraph = research_builder.compile()

def run_research(state: ParentState) -> ParentState:
    sub_input = {"query": state["user_question"], "search_results": [], "summary": ""}
    sub_output = research_subgraph.invoke(sub_input)
    return {"answer": sub_output["summary"]}

parent_builder = StateGraph(ParentState)
parent_builder.add_node("research", run_research)
parent_builder.add_edge(START, "research")
parent_builder.add_edge("research", END)
parent_graph = parent_builder.compile()

print(parent_graph.invoke({"user_question": "what is a HNSW index", "answer": ""}))

The run_research function is the adapter. It builds the subgraph's expected input from the parent's state, calls .invoke(), and maps the result back to whatever key the parent cares about. This is more code than the shared-schema version, but it buys you real decoupling: the research subgraph's internal fields (search_results, query) never leak into the parent's schema, and you can change the subgraph's internals freely as long as the adapter function still produces an answer.

A rule of thumb: use the shared-schema, direct-node approach for tightly coupled steps within the same feature, and use the adapter-function approach anywhere you are composing genuinely separate subsystems, especially ones owned by different people or reused across multiple parent graphs.

Passing Extra Context Down Into a Subgraph

Sometimes a subgraph needs configuration that is not really "state" but more like runtime context: a user ID, a tenant name, a feature flag. Rather than stuffing that into the state schema, pass it through LangGraph's config parameter, which flows into every node and every subgraph automatically.

def search(state: ResearchState, config: dict) -> ResearchState:
    tenant = config["configurable"].get("tenant_id", "default")
    results = [f"[{tenant}] result for {state['query']}"]
    return {"search_results": results}

result = research_subgraph.invoke(
    {"query": "pricing", "search_results": [], "summary": ""},
    config={"configurable": {"tenant_id": "acme-corp"}},
)

Config propagates down through however many levels of subgraph nesting you have, so a tenant_id set at the top of a three-level-deep graph is visible at the bottom without any node needing to explicitly forward it.

Streaming Output Across Subgraph Boundaries

If you stream a parent graph's execution with .stream(), by default you only see updates at the parent's granularity: each subgraph shows up as one opaque step. To see updates from inside a subgraph as they happen, pass subgraphs=True.

for chunk in parent_graph.stream(
    {"user_question": "what is cosine similarity", "answer": ""},
    subgraphs=True,
):
    print(chunk)

With subgraphs=True, each streamed chunk is a tuple of (namespace, data), where namespace tells you which subgraph (and how deeply nested) the update came from. This matters a lot for anything user-facing: if your parent graph is a supervisor routing between a coder subgraph and a researcher subgraph, subgraphs=True is how you show the user live progress from whichever subgraph is currently running, instead of a single spinner that only resolves when the whole thing finishes.

For token-level streaming (watching an LLM call inside a subgraph token by token), combine subgraphs=True with stream_mode="messages":

for namespace, chunk in parent_graph.stream(
    {"user_question": "explain retrieval augmented generation", "answer": ""},
    subgraphs=True,
    stream_mode="messages",
):
    if hasattr(chunk, "content") and chunk.content:
        print(chunk.content, end="", flush=True)

Nesting Subgraphs Multiple Levels Deep

Nothing stops a subgraph from containing its own subgraphs. A common shape in larger systems is three levels: a top-level supervisor graph, mid-level agent graphs (coder, researcher, reviewer), and low-level tool-execution subgraphs inside each agent.

tool_subgraph = StateGraph(ToolState).compile()

def coder_node(state: AgentState, config: dict) -> AgentState:
    tool_input = {"command": state["pending_command"], "output": ""}
    tool_result = tool_subgraph.invoke(tool_input, config=config)
    return {"last_output": tool_result["output"]}

coder_builder = StateGraph(AgentState)
coder_builder.add_node("run_tool", coder_node)
coder_builder.add_edge(START, "run_tool")
coder_builder.add_edge("run_tool", END)
coder_subgraph = coder_builder.compile()

supervisor_builder = StateGraph(SupervisorState)
supervisor_builder.add_node("coder", coder_subgraph)

Each level should have the narrowest state schema it actually needs. Resist the temptation to thread every field through every level "just in case" -- that recreates the original monolith problem one layer down. If a tool subgraph only needs a command and an output field, give it exactly that, and let the adapter functions at each boundary do the translation work.

Debugging Subgraph Execution

Two tools make debugging nested graphs manageable instead of miserable.

Print the graph structure. Both parent_graph.get_graph() and, for nested graphs, parent_graph.get_graph(xray=True) render a picture of what is actually connected. The xray=True flag expands subgraphs inline instead of showing them as a single black-box node, which is invaluable when you are not sure whether an edge inside a subgraph is wired correctly.

print(parent_graph.get_graph(xray=True).draw_mermaid())

Paste the Mermaid output into any Mermaid renderer to see the whole nested structure as a diagram.

Use `stream_mode="debug"` with `subgraphs=True`. This gives you a step-by-step trace of every node execution across every level of nesting, including the exact state going in and coming out of each node. When a subgraph produces unexpected output, this is faster than adding print statements to every node function.

for chunk in parent_graph.stream(
    {"user_question": "what is a bloom filter", "answer": ""},
    subgraphs=True,
    stream_mode="debug",
):
    print(chunk)

If you are running with LangSmith tracing enabled (set the LANGSMITH_TRACING and LANGSMITH_API_KEY environment variables before running your script), every subgraph invocation shows up as a nested span in the trace UI automatically, with no code changes required. For anything beyond a toy example, turn this on early -- untangling a bug across three levels of nested subgraphs from console output alone is a bad way to spend an afternoon.

Common Patterns Worth Knowing

A few compositions come up often enough to name directly.

Supervisor plus workers. A top-level graph routes to one of several subgraphs based on a classification step, each subgraph fully encapsulating one worker's logic (a SQL agent, a web-search agent, a code-execution agent). The supervisor's state schema only needs the routing decision and the final answer; everything else stays inside each worker subgraph.

Retry wrapper. Wrap a subgraph invocation in a Python loop that retries with backoff or with a modified input if the subgraph's output fails a validation check. Because the subgraph is just a compiled runnable, ordinary Python control flow works around it without any special LangGraph API.

def run_with_retry(state: ParentState) -> ParentState:
    for attempt in range(3):
        sub_output = research_subgraph.invoke(
            {"query": state["user_question"], "search_results": [], "summary": ""}
        )
        if len(sub_output["summary"]) > 10:
            return {"answer": sub_output["summary"]}
    return {"answer": "unable to produce a summary"}

Shared checkpointer across parent and subgraph. When you compile the parent graph with a checkpointer (for persistence and human-in-the-loop interrupts), subgraphs inherit it automatically as long as you do not pass a separate checkpointer when compiling the subgraph itself. This means an interrupt() call inside a deeply nested subgraph correctly pauses and resumes the entire parent graph's execution, not just the subgraph.

FAQ

Do subgraphs need their own checkpointer? No, and in most cases you should not give them one. Compile subgraphs without a checkpointer argument so they inherit the parent graph's checkpointer at runtime. Giving a subgraph its own separate checkpointer breaks the ability to interrupt and resume the parent graph cleanly across that subgraph's nodes.

Can I add a compiled subgraph as a node if the schemas do not match at all? Not directly. add_node with a compiled graph only works when there is enough overlap in state keys for LangGraph to pass values through automatically. If the schemas are unrelated, wrap the subgraph call in a plain function that builds the subgraph's input from the parent's state and maps the subgraph's output back onto the parent's keys, as shown in the adapter-function example above.

How many levels of nesting is reasonable? Two or three levels covers the vast majority of real systems: a supervisor, a set of specialized agents, and each agent's internal tool-execution loop. Beyond that, the cost of tracing a bug across five or six levels tends to outweigh the modularity benefit. If you are nesting deeper than three levels, it is usually a sign that some of those "levels" should be plain function calls instead of graphs.

Does `subgraphs=True` change what the parent graph invoke returns? No, it only affects .stream(). A plain .invoke() call always returns just the final state, regardless of how many subgraphs ran underneath it. subgraphs=True is purely about visibility during streaming, letting you see intermediate updates from inside child graphs as they happen instead of only after they finish.

Is there a performance cost to using subgraphs versus one flat graph? The overhead is negligible for typical agent workloads; the cost of an LLM call or a tool call dwarfs the cost of an extra function call boundary between a parent and a subgraph. The real tradeoff is code complexity versus modularity, not runtime performance.

Can a subgraph call the same subgraph recursively? LangGraph does not restrict this at the API level, but you need an explicit termination condition in the subgraph's own logic (a depth counter passed through config, for example), because the graph's own edges will not automatically stop a subgraph from invoking itself indefinitely. Recursive subgraph patterns like this show up in agents that break a task into smaller identical subtasks, but they need careful bounds checking to avoid runaway recursion.