LangGraph Subgraphs: Composing Complex Agents from Smaller Graphs
Why Your Agent Graph Is Turning Into Spaghetti
If you have built more than one or two LangGraph agents, you already know the feeling. You start with a clean graph: a node that calls a tool, a node that reasons, an edge that routes back on failure. Then the requirements grow. Now you need a research step that itself has multiple stages, a validation loop, a human-in-the-loop checkpoint, and a summarization pass. Before long your single StateGraph has thirty nodes, a state schema with twenty keys that only three nodes ever touch, and conditional edges that route to conditional edges. Nobody on the team can hold the whole thing in their head anymore.
This is the exact problem subgraphs solve. A subgraph is just a LangGraph graph that gets used as a node inside a bigger graph. Instead of flattening every concern into one giant state machine, you build small, focused graphs — a "research graph," a "critique graph," a "tool-execution graph" — and compose them the way you compose functions in normal software. Each subgraph has its own state schema, its own nodes, its own edges, and its own reasons to exist. The parent graph doesn't need to know how the research subgraph does research; it only needs to know what goes in and what comes out.
This article walks through how subgraphs actually work in LangGraph, the two ways state gets shared (or doesn't) between parent and child graphs, how to stream and debug across subgraph boundaries, and the architectural patterns that make subgraphs worth the extra abstraction. We will write real code throughout, not just describe the concept in the abstract. By the end you should be able to look at a sprawling single-file agent and see exactly where the seams should go.
What a Subgraph Actually Is
In LangGraph, a compiled graph is a Runnable. That is the whole trick. Because a compiled StateGraph behaves like any other runnable node, you can add it directly into another graph with add_node. There is no special "subgraph" class you need to import — the composition is structural, not a separate API surface.
Here is the smallest possible example: a child graph that does a two-step calculation, used inside a parent graph.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class ChildState(TypedDict):
number: int
doubled: int
incremented: int
def double_it(state: ChildState) -> dict:
return {"doubled": state["number"] * 2}
def increment_it(state: ChildState) -> dict:
return {"incremented": state["doubled"] + 1}
child_builder = StateGraph(ChildState)
child_builder.add_node("double_it", double_it)
child_builder.add_node("increment_it", increment_it)
child_builder.add_edge(START, "double_it")
child_builder.add_edge("double_it", "increment_it")
child_builder.add_edge("increment_it", END)
child_graph = child_builder.compile()
class ParentState(TypedDict):
number: int
doubled: int
incremented: int
label: str
def add_label(state: ParentState) -> dict:
return {"label": f"processed {state['incremented']}"}
parent_builder = StateGraph(ParentState)
parent_builder.add_node("compute", child_graph) # subgraph used as a node
parent_builder.add_node("add_label", add_label)
parent_builder.add_edge(START, "compute")
parent_builder.add_edge("compute", "add_label")
parent_builder.add_edge("add_label", END)
parent_graph = parent_builder.compile()
result = parent_graph.invoke({"number": 5, "label": ""})
print(result)
# {'number': 5, 'doubled': 10, 'incremented': 11, 'label': 'processed 11'}Notice that ChildState and ParentState share overlapping keys here (number, doubled, incremented). That is deliberate for this first example — it is the easy case, where the child and parent speak the same schema. Most real systems are not this tidy, which brings us to the actual design decision you need to make with every subgraph: how much of the parent's state should the child see.
Two Ways to Share State: Shared Schema vs. Different Schema
LangGraph gives you two patterns for wiring a subgraph into a parent, and picking the right one is the single most important decision when composing graphs.
Pattern 1 — shared state keys. If the child graph's state schema shares one or more keys with the parent's state schema, you can add the compiled subgraph directly as a node, exactly like the example above. LangGraph will pass the overlapping keys through automatically and merge the results back using each key's reducer. This is convenient, but it also means the subgraph is coupled to the parent's schema — rename a key in the parent and you have silently broken the child.
Pattern 2 — different (invocation) schema. If the subgraph has a completely different state schema — which is the more common and more maintainable case for anything non-trivial — you cannot add it directly as a node. Instead, you wrap it in a regular Python function, translate the parent's state into the child's input shape, invoke the compiled subgraph, and translate the output back into the parent's state shape.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class ResearchState(TypedDict):
topic: str
sources: list[str]
summary: str
def gather_sources(state: ResearchState) -> dict:
return {"sources": [f"source-on-{state['topic']}-1", f"source-on-{state['topic']}-2"]}
def summarize_sources(state: ResearchState) -> dict:
joined = ", ".join(state["sources"])
return {"summary": f"Summary of {joined}"}
research_builder = StateGraph(ResearchState)
research_builder.add_node("gather_sources", gather_sources)
research_builder.add_node("summarize_sources", summarize_sources)
research_builder.add_edge(START, "gather_sources")
research_builder.add_edge("gather_sources", "summarize_sources")
research_builder.add_edge("summarize_sources", END)
research_graph = research_builder.compile()
class AgentState(TypedDict):
user_question: str
research_summary: str
final_answer: str
def run_research_subgraph(state: AgentState) -> dict:
# Translate parent state -> child input schema
child_input = {"topic": state["user_question"], "sources": [], "summary": ""}
child_result = research_graph.invoke(child_input)
# Translate child output -> parent state update
return {"research_summary": child_result["summary"]}
def compose_answer(state: AgentState) -> dict:
return {"final_answer": f"Based on research: {state['research_summary']}"}
agent_builder = StateGraph(AgentState)
agent_builder.add_node("run_research_subgraph", run_research_subgraph)
agent_builder.add_node("compose_answer", compose_answer)
agent_builder.add_edge(START, "run_research_subgraph")
agent_builder.add_edge("run_research_subgraph", "compose_answer")
agent_builder.add_edge("compose_answer", END)
agent_graph = agent_builder.compile()
output = agent_graph.invoke({"user_question": "LangGraph subgraphs", "research_summary": "", "final_answer": ""})
print(output["final_answer"])This is the pattern to reach for by default. It costs you one small translation function, and in exchange the research subgraph becomes a genuinely independent unit — you can unit test research_graph in complete isolation, swap its internals, or reuse it in a different parent graph without touching AgentState at all.
Why Bother? The Real Reasons to Use Subgraphs
It is worth being explicit about what problem this actually solves, because "composability" can sound like an academic virtue rather than something that saves you time on a Tuesday afternoon.
- Team ownership boundaries. When multiple people work on one agent, subgraphs let each person own a self-contained piece — one engineer owns the retrieval subgraph, another owns the tool-calling subgraph — without merge conflicts on a single 800-line file.
- Independent testing. A subgraph compiles to a runnable with its own
invoke,stream, andainvoke. You can write unit tests against it directly, feeding it a minimal state and asserting on its output, with zero dependency on the parent graph's plumbing. - Multi-agent architectures. Subgraphs are the natural building block for multi-agent systems where each "agent" is really its own graph — a planner agent, a coder agent, a reviewer agent — coordinated by a parent orchestrator graph. This maps directly onto supervisor and hierarchical agent-team patterns.
- Reuse across projects. A well-scoped subgraph, like a "web research" graph or a "SQL generation and validation" graph, becomes an internal library component you drop into different parent agents.
- Readability at the top level. The parent graph becomes a short, high-level description of the workflow — "plan, then research, then write, then critique" — instead of a wall of low-level nodes. Anyone reading the parent graph's node list gets the executive summary for free.
- Isolated checkpointing granularity. Each subgraph can maintain its own internal state history, which matters a lot once you start using persistence and want to resume or inspect a specific stage without wading through the whole system's state.
None of these are exotic wins. They are the same reasons you break a function that is doing too much into smaller functions, applied to graphs of LLM calls instead of graphs of function calls.
Streaming Output From Inside Subgraphs
One thing that trips people up the first time: by default, stream() on the parent graph only emits events at the level of the parent's nodes. If the "interesting" work — the token-by-token generation, or the individual steps of the research process — happens inside a subgraph, you will not see it unless you ask for it explicitly.
LangGraph exposes this with the subgraphs=True flag on stream.
for chunk in agent_graph.stream(
{"user_question": "LangGraph subgraphs", "research_summary": "", "final_answer": ""},
subgraphs=True,
):
print(chunk)When subgraphs=True, each streamed chunk is a tuple of (namespace, data) rather than just data. The namespace tells you which subgraph, and which invocation of it, produced the event — this matters when a subgraph is invoked multiple times, for example inside a loop over several documents. Without the namespace you would have no way to tell which iteration a given chunk belongs to.
for namespace, chunk in agent_graph.stream(
{"user_question": "LangGraph subgraphs", "research_summary": "", "final_answer": ""},
subgraphs=True,
stream_mode="updates",
):
print(f"namespace={namespace}")
print(f"update={chunk}")This same principle applies to stream_mode="messages" if your subgraph contains an LLM call you want to stream token-by-token up to the parent — pass subgraphs=True and filter on the namespace to isolate which part of the system is currently talking.
Persistence and Checkpointing Across Subgraph Boundaries
Checkpointing is where subgraph composition gets a little more subtle, because it determines whether a subgraph can be resumed independently or only as part of the whole parent run.
If you compile the parent graph with a checkpointer, you generally do not need to also pass a checkpointer to the subgraph — the parent's checkpointer covers the whole tree by default, and each subgraph invocation gets its own checkpoint namespace automatically, keyed by the parent's thread and the subgraph's position in the run.
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent_graph = agent_builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "conversation-42"}}
result = agent_graph.invoke(
{"user_question": "LangGraph subgraphs", "research_summary": "", "final_answer": ""},
config=config,
)Where this gets important is interrupts. If your subgraph itself needs a human-in-the-loop pause — say the research subgraph wants approval before hitting a paid search API — you add interrupt() inside the subgraph's node exactly as you would in a top-level graph. Because the checkpointer is shared, resuming the parent graph with Command(resume=...) will correctly resume execution from inside the subgraph, not restart the subgraph from scratch. This is one of the more pleasant surprises in LangGraph: interrupts compose across graph boundaries without extra plumbing, as long as the checkpointer is attached at the top level.
The one case where you deliberately want a separate checkpointer is when a subgraph should be able to run and resume completely independently of the parent's thread — for instance, a subgraph that also gets invoked stand-alone as its own tool outside of this parent. In that scenario, compile the subgraph with its own checkpointer instance rather than inheriting the parent's.
A Realistic Pattern: Supervisor With Worker Subgraphs
The pattern that shows up most often in production is a supervisor graph that routes to one of several worker subgraphs based on the task. Each worker is a complete graph in its own right — with its own tool loop, its own retry logic — and the supervisor's only job is routing and aggregation.
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
# --- Worker subgraph: code generation ---
class CoderState(TypedDict):
task: str
code: str
def write_code(state: CoderState) -> dict:
return {"code": f"# code for: {state['task']}\nprint('done')"}
coder_builder = StateGraph(CoderState)
coder_builder.add_node("write_code", write_code)
coder_builder.add_edge(START, "write_code")
coder_builder.add_edge("write_code", END)
coder_graph = coder_builder.compile()
# --- Worker subgraph: research ---
class ResearcherState(TypedDict):
task: str
findings: str
def research_task(state: ResearcherState) -> dict:
return {"findings": f"findings for: {state['task']}"}
researcher_builder = StateGraph(ResearcherState)
researcher_builder.add_node("research_task", research_task)
researcher_builder.add_edge(START, "research_task")
researcher_builder.add_edge("research_task", END)
researcher_graph = researcher_builder.compile()
# --- Supervisor graph ---
class SupervisorState(TypedDict):
task: str
route: Literal["coder", "researcher"]
result: str
def decide_route(state: SupervisorState) -> dict:
if "code" in state["task"].lower():
return {"route": "coder"}
return {"route": "researcher"}
def run_coder(state: SupervisorState) -> dict:
output = coder_graph.invoke({"task": state["task"], "code": ""})
return {"result": output["code"]}
def run_researcher(state: SupervisorState) -> dict:
output = researcher_graph.invoke({"task": state["task"], "findings": ""})
return {"result": output["findings"]}
def route_selector(state: SupervisorState) -> str:
return state["route"]
supervisor_builder = StateGraph(SupervisorState)
supervisor_builder.add_node("decide_route", decide_route)
supervisor_builder.add_node("run_coder", run_coder)
supervisor_builder.add_node("run_researcher", run_researcher)
supervisor_builder.add_edge(START, "decide_route")
supervisor_builder.add_conditional_edges(
"decide_route",
route_selector,
{"coder": "run_coder", "researcher": "run_researcher"},
)
supervisor_builder.add_edge("run_coder", END)
supervisor_builder.add_edge("run_researcher", END)
supervisor_graph = supervisor_builder.compile()
print(supervisor_graph.invoke({"task": "write code to sort a list", "route": "coder", "result": ""}))
print(supervisor_graph.invoke({"task": "find papers on graph theory", "route": "coder", "result": ""}))Notice each worker subgraph — coder_graph and researcher_graph — has a completely different state schema from SupervisorState and from each other. The supervisor never needs to know that internally. If you later want the coder subgraph to gain a self-review loop with three additional internal nodes, you change exactly one file and the supervisor graph is untouched. That isolation is the entire point.
Visualizing and Debugging Nested Graphs
Once you have more than one level of nesting, being able to see the structure matters. LangGraph's built-in graph drawing will, by default, expand subgraphs so you can see their internal nodes inline with the parent — which is useful for documentation, but can get noisy fast on deeply nested systems.
# Renders the parent graph, with subgraph internals expanded by default
supervisor_graph.get_graph(xray=True).draw_mermaid_png(output_file_path="supervisor_graph.png")
# Collapse subgraphs to a single box instead
supervisor_graph.get_graph(xray=False).draw_mermaid_png(output_file_path="supervisor_graph_collapsed.png")For debugging a failing run, get_state_history() on a checkpointed parent graph will include entries tagged with the subgraph's namespace, so you can walk back through exactly which node inside which subgraph produced a bad value. When something goes wrong three levels deep, resist the urge to add print statements everywhere — pull the state history for the specific thread and look at the last few checkpoints before the failure. It is almost always faster than re-running the whole thing with logging sprinkled in.
A practical debugging habit: give each subgraph invocation a distinguishable identity when you call it multiple times in the same run (for example, looping the research subgraph over several documents). Since streaming and state history both surface the subgraph's position by namespace, keeping your loop indices or document IDs inside the state you pass to the subgraph makes the namespace immediately readable instead of a string of coordinates you have to decode.
Common Mistakes When Composing Subgraphs
- Forcing shared schema everywhere. Trying to make every subgraph share the parent's state schema so you can skip the wrapper function seems convenient at first, but it recreates the exact coupling problem subgraphs are meant to fix. Reach for the translation-function pattern by default; only share schema when the child truly is a natural extension of the same state.
- Forgetting `subgraphs=True` and thinking streaming is broken. This is the single most common "why can't I see my subgraph's output" support question. The events are there; they are just filtered out of the default stream.
- Over-nesting. Subgraphs inside subgraphs inside subgraphs can technically go as deep as you want, but each extra level adds a namespace layer to reason about during debugging. Two levels — parent plus workers — covers the overwhelming majority of real designs. Reach for a third level only when a worker itself is complex enough to deserve its own decomposition.
- Mismatched checkpointer expectations. Assuming a subgraph checkpoints independently when it actually inherited the parent's checkpointer (or vice versa) leads to confusing resume behavior. Decide explicitly, per subgraph, whether it needs standalone persistence.
- Skipping tests on the subgraph in isolation. Because a compiled subgraph is just a runnable, it is easy to test on its own with a handful of representative inputs before you ever wire it into the parent. Skipping this step means every bug surfaces at the full-system level, where it is much more expensive to localize.
Bringing It Together
Subgraphs are not a separate feature bolted onto LangGraph — they fall directly out of the fact that a compiled graph is a runnable, and runnables compose. That simplicity is exactly why the pattern scales so well: the same mental model you use for a two-node graph applies whether you are nesting one subgraph or coordinating five worker subgraphs under a supervisor. The two decisions that matter every time are whether to share state schema directly or translate at the boundary, and whether streaming and checkpointing need to reach across that boundary or stay contained within it.
Start by identifying the piece of your current agent that has grown its own internal logic — a retry loop, a validation pass, a multi-step tool chain — and pull it out into its own graph with a translation function at the seam. You will feel the difference in testability almost immediately, and your parent graph will read like a table of contents instead of a maze.
If you want to go deeper on this with guided, hands-on exercises — building multi-agent supervisor systems, wiring interrupts through nested subgraphs, and handling streaming across namespaces in a real project — that is exactly what we cover step by step inside the LangGraph Tutorial course on teachyou.ai.
BootcampA 30-day guided bootcamp: build, harden and ship a production autonomous agent from scratch.
AI AgentsUnderstand how AI agents really work: the loop, the tools, the memory, and why most agent projects fail.