advanced⏱40 minutesLesson 3 of 10
Hierarchical Agents
Implement hierarchical multi-agent architectures with manager agents, sub-agents, subgraphs, and state passing between hierarchy levels.
Hierarchical Agents
Hierarchical agents organize work into levels — a manager agent delegates tasks to sub-agents, which can themselves be graphs with internal nodes. This enables scalable, maintainable, and encapsulated agent systems.
Why Hierarchical?
Flat multi-agent systems have limitations:
- State complexity: All agents share one state dict, which grows unmanageable
- Coordination cost: Every agent sees all messages, even irrelevant ones
- Encapsulation: Agent internals leak into the shared state
- Scalability: Adding agents increases complexity linearly
Hierarchical architectures solve these with encapsulation and delegation:
Manager Agent
├── Research Sub-Agent (internal graph)
│ ├── plan_research
│ ├── execute_search
│ └── summarize
├── Write Sub-Agent (internal graph)
│ ├── draft
│ ├── review
│ └── finalize
└── Quality Sub-Agent (internal graph)
├── check_facts
└── validate_output
Subgraphs
Subgraphs are graphs used as nodes within a parent graph. They have their own state, nodes, and edges — fully encapsulated.
python
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
# === Subgraph: Research ===
class ResearchState(TypedDict):
topic: str
findings: str
status: str
def plan_research(state: ResearchState) -> dict:
plan = f"Research plan for: {state['topic']}"
return {"status": "planned"}
def execute_research(state: ResearchState) -> dict:
findings = f"Findings about {state['topic']}: ..."
return {"findings": findings, "status": "completed"}
research_builder = StateGraph(ResearchState)
research_builder.add_node("plan", plan_research)
research_builder.add_node("execute", execute_research)
research_builder.add_edge(START, "plan")
research_builder.add_edge("plan", "execute")
research_builder.add_edge("execute", END)
research_subgraph = research_builder.compile()Note
The research subgraph has its own state (ResearchState). It doesn't know about the parent graph's state, and the parent doesn't know about internal research details.
Parent Graph Using Subgraph
python
class ManagerState(TypedDict):
query: str
research_findings: str
final_report: str
status: str
def manager_router(state: ManagerState) -> str:
if not state.get("research_findings"):
return "research_subgraph"
return "finalize"
def finalize_report(state: ManagerState) -> dict:
report = f"Report based on: {state['research_findings']}"
return {"final_report": report, "status": "completed"}
# Parent graph
builder = StateGraph(ManagerState)
builder.add_node("research_subgraph", research_subgraph) # Subgraph as a node
builder.add_node("finalize", finalize_report)
builder.add_edge(START, "research_subgraph")
builder.add_edge("research_subgraph", "finalize")
builder.add_edge("finalize", END)
app = builder.compile()Success
A subgraph is added as a node just like a regular function. LangGraph handles state mapping between parent and subgraph automatically.
State Mapping Between Parent and Subgraph
The parent passes a subset of its state to the subgraph. The subgraph returns its state, which is merged back.
python
class ParentState(TypedDict):
query: str
research: str
output: str
class ResearchState(TypedDict):
query: str # Mapped from parent
findings: str # Mapped back to parent
depth: str
# When invoke is called on the subgraph from the parent node:
def research_node(parent_state: ParentState) -> dict:
# Map parent state to subgraph state
sub_input = {
"query": parent_state["query"],
"findings": "",
"depth": "deep"
}
# Invoke the subgraph
sub_result = research_subgraph.invoke(sub_input)
# Map subgraph result back to parent state
return {"research": sub_result["findings"]}Multi-Level Hierarchy
python
# Level 3: Sub-subgraph
class FactCheckState(TypedDict):
claim: str
verified: bool
def verify_claim(state: FactCheckState) -> dict:
verified = "verified" in state["claim"].lower()
return {"verified": verified}
fact_builder = StateGraph(FactCheckState)
fact_builder.add_node("verify", verify_claim)
fact_builder.add_edge(START, "verify")
fact_builder.add_edge("verify", END)
fact_subgraph = fact_builder.compile()
# Level 2: Subgraph uses fact_subgraph
class ResearchState(TypedDict):
topic: str
claims: list
verified_claims: list
summary: str
def check_claims(state: ResearchState) -> dict:
verified = []
for claim in state["claims"]:
result = fact_subgraph.invoke({"claim": claim, "verified": False})
verified.append({"claim": claim, "verified": result["verified"]})
return {"verified_claims": verified}
research_builder = StateGraph(ResearchState)
research_builder.add_node("check", check_claims)
research_builder.add_edge(START, "check")
research_builder.add_edge("check", END)
research_graph = research_builder.compile()
# Level 1: Parent uses research_graph
builder = StateGraph(ParentState)
builder.add_node("research", research_graph)
builder.add_edge(START, "research")
builder.add_edge("research", END)
app = builder.compile()Important
Subgraphs at any level can contain their own subgraphs. This nesting enables arbitrarily complex agent hierarchies while maintaining encapsulation at each level.
Shared Context Across Levels
Use a context object that flows through all hierarchy levels:
python
class GlobalContext(TypedDict):
user_id: str
session_id: str
constraints: List[str]
preferences: dict
class TaskState(TypedDict):
context: GlobalContext # Passed through all levels
task: str
result: str
# Each subgraph receives and passes the context
def research_subgraph_node(state: TaskState) -> dict:
ctx = state["context"]
result = perform_research(state["task"], ctx["constraints"])
return {"result": result} # Context flows through automatically?Warning
State mapping between parent and subgraph must be explicit. The parent decides which fields to pass to the subgraph and which subgraph outputs to read.
Parallel Subgraph Execution
Run multiple subgraphs in parallel:
python
def run_parallel_research(state: ManagerState) -> dict:
topics = [state["query"], f"{state['query']} advanced"]
results = []
for topic in topics:
result = research_subgraph.invoke({
"topic": topic,
"findings": "",
"status": ""
})
results.append(result["findings"])
return {"research_findings": "\n\n".join(results)}For true parallelism, use fan-out:
python
builder.add_edge(START, "research_topic_1")
builder.add_edge(START, "research_topic_2")
builder.add_edge("research_topic_1", "merge")
builder.add_edge("research_topic_2", "merge")Complete Hierarchical Agent Example
python
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from typing import List, Annotated
from operator import add
# === Level 2: Search Subgraph ===
class SearchState(TypedDict):
query: str
results: List[str]
def search_web(state: SearchState) -> dict:
return {"results": [f"Result for: {state['query']}"]}
search_builder = StateGraph(SearchState)
search_builder.add_node("search", search_web)
search_builder.add_edge(START, "search")
search_builder.add_edge("search", END)
search_graph = search_builder.compile()
# === Level 2: Analyze Subgraph ===
class AnalyzeState(TypedDict):
data: List[str]
analysis: str
def analyze_data(state: AnalyzeState) -> dict:
return {"analysis": f"Analysis of {len(state['data'])} sources"}
analyze_builder = StateGraph(AnalyzeState)
analyze_builder.add_node("analyze", analyze_data)
analyze_builder.add_edge(START, "analyze")
analyze_builder.add_edge("analyze", END)
analyze_graph = analyze_builder.compile()
# === Level 1: Manager ===
class ManagerState(TypedDict):
query: str
results: List[str]
analysis: str
final: str
logs: Annotated[List[str], add]
def search_manager(state: ManagerState) -> dict:
sub_result = search_graph.invoke({"query": state["query"], "results": []})
return {"results": sub_result["results"], "logs": ["Search completed"]}
def analyze_manager(state: ManagerState) -> dict:
sub_result = analyze_graph.invoke({"data": state["results"], "analysis": ""})
return {"analysis": sub_result["analysis"], "logs": ["Analysis completed"]}
def finalize(state: ManagerState) -> dict:
return {"final": f"Query: {state['query']}\nAnalysis: {state['analysis']}"}
builder = StateGraph(ManagerState)
builder.add_node("search", search_manager)
builder.add_node("analyze", analyze_manager)
builder.add_node("finalize", finalize)
builder.add_edge(START, "search")
builder.add_edge("search", "analyze")
builder.add_edge("analyze", "finalize")
builder.add_edge("finalize", END)
app = builder.compile()
result = app.invoke({
"query": "LangGraph hierarchy",
"results": [],
"analysis": "",
"final": "",
"logs": []
})
print(result["final"])Benefits of Hierarchical Design
| Benefit | Description |
|---|---|
| Encapsulation | Subgraph internals don't leak to parent level |
| Reusability | Subgraphs can be used in multiple parent graphs |
| Testability | Each subgraph is tested independently |
| Complexity management | Each level only deals with its own concerns |
| Parallelism | Multiple subgraphs can run in parallel |
| Clear interfaces | Explicit input/output contracts between levels |
Practice Questions
Practice Question
What is a subgraph in LangGraph?
Practice Question
What is the main advantage of hierarchical agent architecture?
Practice Question
How does a parent graph pass data to a subgraph?
Practice Question
Can a subgraph contain its own subgraphs?
Practice Question
What is a benefit of subgraph encapsulation?
Practice Question
How do you add a subgraph as a node to a parent graph?
Practice Question
What happens to the parent state after a subgraph completes?
Practice Question
What problem does hierarchical architecture solve in multi-agent systems?
Practice Question
How can you run multiple subgraphs in parallel from a parent?
Practice Question
What is the role of a manager agent in a hierarchical system?
Success
Key Takeaways
- Subgraphs are compiled graphs used as nodes in parent graphs
- Each subgraph has its own encapsulated state
- State mapping between levels is explicit (manual)
- Subgraphs can be nested to any depth
- Reusability: same subgraph can be used in multiple parents
- Testability: each subgraph is independently testable
- Parallel subgraph execution via fan-out edges
- Hierarchical design manages complexity in large agent systems
Progress30%