Reusing subgraphs across multiple parent graphs
Why this matters
As your agentic workflows grow, you'll write the same patterns repeatedly: research nodes, validation logic, tool-calling sequences. Subgraphs let you define once, reuse everywhere, reducing maintenance burden and enabling teams to share workflow components like libraries.
Explanation
A subgraph is a StateGraph that runs as a node inside another StateGraph. Instead of inlining all your logic into one massive graph, you extract reusable workflows: like 'research this topic' or 'validate and correct user input': into their own graphs, then call them from parent graphs that compose different subgraphs together.
Mechanically, langgraph 0.2.x treats a compiled subgraph as a callable node. You create a StateGraph, compile it, then use it directly as a node in another StateGraph via .add_node(). The parent graph passes its state to the subgraph, which runs its own nodes and edges, then returns control to the parent. The key constraint: subgraph input/output state must be compatible with the parent's state shape: typically via shared keys or explicit state reduction.
Use subgraphs when you have a self-contained workflow (research, validation, code generation) that multiple parent graphs need. This pattern scales teams: one group owns the 'research' subgraph, another owns 'fact-checking,' and orchestration graphs compose them without touching the implementation.
Analogy
Think of a subgraph like a library function. A function takes inputs and returns outputs; you use it in many programs without rewriting it. A subgraph does the same for entire workflows: it's a function-sized unit of graph logic you can call from anywhere.
Code
from langgraph.graph import StateGraph, START, END
from langgraph.graph.graph import CompiledGraph
from typing import TypedDict
import time
class ResearchState(TypedDict):
topic: str
research_notes: str
class ValidationState(TypedDict):
content: str
is_valid: bool
feedback: str
class OrchestratorState(TypedDict):
topic: str
research_notes: str
content: str
is_valid: bool
feedback: str
def research_step(state: ResearchState) -> ResearchState:
topic = state["topic"]
notes = f"Researched '{topic}': Found 3 key sources, learned about history and current trends."
return {"topic": topic, "research_notes": notes}
def validate_step(state: ValidationState) -> ValidationState:
content = state["content"]
is_valid = len(content) > 10
feedback = "Content looks good" if is_valid else "Content too short"
return {"content": content, "is_valid": is_valid, "feedback": feedback}
research_graph = StateGraph(ResearchState)
research_graph.add_node("research", research_step)
research_graph.add_edge(START, "research")
research_graph.add_edge("research", END)
research_subgraph = research_graph.compile()
validation_graph = StateGraph(ValidationState)
validation_graph.add_node("validate", validate_step)
validation_graph.add_edge(START, "validate")
validation_graph.add_edge("validate", END)
validation_subgraph = validation_graph.compile()
def call_research_subgraph(state: OrchestratorState) -> OrchestratorState:
subgraph_input = {"topic": state["topic"]}
result = research_subgraph.invoke(subgraph_input)
return {
"topic": state["topic"],
"research_notes": result["research_notes"],
"content": state.get("content", ""),
"is_valid": state.get("is_valid", False),
"feedback": state.get("feedback", "")
}
def call_validation_subgraph(state: OrchestratorState) -> OrchestratorState:
subgraph_input = {"content": state["research_notes"]}
result = validation_subgraph.invoke(subgraph_input)
return {
"topic": state["topic"],
"research_notes": state["research_notes"],
"content": result["content"],
"is_valid": result["is_valid"],
"feedback": result["feedback"]
}
orchestrator_graph = StateGraph(OrchestratorState)
orchestrator_graph.add_node("run_research", call_research_subgraph)
orchestrator_graph.add_node("run_validation", call_validation_subgraph)
orchestrator_graph.add_edge(START, "run_research")
orchestrator_graph.add_edge("run_research", "run_validation")
orchestrator_graph.add_edge("run_validation", END)
orchestrator_compiled = orchestrator_graph.compile()
result = orchestrator_compiled.invoke({"topic": "Machine Learning"})
print(f"Topic: {result['topic']}")
print(f"Research Notes: {result['research_notes']}")
print(f"Is Valid: {result['is_valid']}")
print(f"Feedback: {result['feedback']}")Topic: Machine Learning Research Notes: Researched 'Machine Learning': Found 3 key sources, learned about history and current trends. Is Valid: True Feedback: Content looks good
What just happened?
We defined two standalone StateGraphs (research_subgraph and validation_subgraph), compiled them, then created wrapper nodes that invoke these subgraphs from an orchestrator graph. The orchestrator passes its state (topic) to the research subgraph, collects the output, then passes that result to the validation subgraph. Each subgraph ran independently with its own state shape, then results were mapped back into the parent's state. The orchestrator coordinated the flow from research → validation → end.
Common gotcha
The most common mistake is assuming subgraph state automatically merges with parent state. It doesn't. When you call a subgraph from a parent node, you must explicitly extract what the subgraph needs from the parent state, pass it in as a dict matching the subgraph's StateDict schema, invoke the subgraph, then manually merge the result back into the parent's state. Forgetting this step causes TypeErrors or lost data.
Error recovery
TypeError: invoke() got an unexpected keyword argumentKeyError when accessing state['missing_key']State mutation across subgraph boundariesExperienced dev note
Real teams version subgraphs separately from orchestrators. A subgraph becomes an 'asset': you test it in isolation, version it, and the orchestrator depends on a specific subgraph version. This prevents cascading failures: if you update the research subgraph logic, orchestrators don't break as long as the input/output state keys stay the same. Also: subgraphs shine when multiple teams own different pieces. The research team updates research_subgraph independently; the orchestration team imports it as a stable dependency. Without this boundary, you end up with one giant graph that no one understands.
Check your understanding
If you had two parent graphs (one for 'market research' and one for 'academic research') that both need the validation_subgraph, but they each pass different state keys to it, how would you handle that without duplicating the validation logic?
Show answer hint
A correct answer explains that you'd create a wrapper node in each parent that translates its local state keys into the validation_subgraph's expected keys, calls the subgraph, then maps results back. The validation logic itself stays in one place: only the mapping layer differs per parent.