Issue 18 | Sub-graphs: Condensing the Underlying Graph into a Regular Node
Welcome back to our "LangGraph Multi-Agent Expert Course", fellow AI architects. I am your old friend.
In the previous 17 issues, our "AI Content Agency" has begun to take shape. We have the strategizing Planner, the hard-working Researcher looking up information, the furiously typing Writer, and the nitpicking Editor.
However, as the business deepens, have you noticed a scalp-tingling problem? Our Main Graph is becoming more and more like a plate of spaghetti!
Recall that in the last issue, to enable the Researcher to "search -> scrape web -> summarize -> evaluate information quality (and search again if inadequate)", we added a whole bunch of nodes and conditional edges to it. The result? The main graph is densely packed with the Researcher's internal logic. The Planner and Writer are shivering in the corner, and the readability and maintainability of the entire architecture have plummeted.
It's like you, as the CEO of a company (the Main Graph), don't need to, and shouldn't, manage how the research department (Researcher) holds meetings, looks up information, or argues internally. You just need to throw the "research requirement" to the research director, and then wait for him to hand you the "research report".
This is the ultimate weapon we are going to talk about today—Sub-graphs.
Today, we are going to perform a "surgical-level refactoring" on the system: fold and condense the Researcher's complex internal workflow into a Sub-graph, and then mount it onto our Main Graph just like calling a regular Node.
Get your coffee ready, let's go!
🎯 Learning Objectives for this Issue
- Master the core philosophy of Sub-graphs: Understand the underlying logic of "graphs nested within graphs" to achieve a dimensionality reduction strike on system complexity.
- Bridge the State Mapping between parent and child graphs: Figure out how data goes in and comes out between the Main Graph and the Sub-graph (this is where it's easiest to crash and burn).
- Complete the Agency project architecture refactoring: Strip the Researcher's "search-scrape-summarize" loop into an independent sub-graph, returning the main workflow to the minimalist form of "Planner -> Researcher -> Writer".
- Master independent debugging techniques for sub-graphs: Learn how to unit test a sub-graph independently without starting the entire system.
📖 Principle Analysis
In LangGraph, any StateGraph that has been compiled via compile() can be directly used as a Node in another StateGraph.
This sounds simple, but it is infinitely powerful. It means you can nest infinitely: a company contains departments, a department contains teams, and a team contains individuals. This Fractal architecture is the only solution for building enterprise-level complex Multi-Agent systems.
Let's look at the architecture comparison before and after refactoring.
Pre-refactoring "Spaghetti Graph" Architecture (Negative Example)
If all the logic is stuffed into one graph, it looks like this: After the Planner finishes planning, it enters the Search node, then determines whether Scrape is needed, then enters Summarize, then determines if the materials are sufficient... The entire main thread is severely fragmented.
Post-refactoring "Modular" Architecture (Today's Goal)
We encapsulate the Researcher's internal logic into a black box (Sub-graph).
Let's look at our architecture design diagram for today:
graph TD
%% Main Graph Node Definitions
Start((START))
Planner[Planner Node\nGenerates outline and research requirements]
Writer[Writer Node\nWrites first draft based on materials]
Editor[Editor Node\nReviews and polishes]
End((END))
%% Sub-graph Definition (Folded inside the Researcher Node)
subgraph Researcher_SubGraph ["🔍 Researcher Node (Internal logic of Sub-graph)"]
direction TB
R_Start((R_Start))
Search[Web Search\nSearch keywords]
Scrape[Web Scrape\nScrape web content]
Summarize[Summarize\nExtract core information]
Eval{Enough to write?}
R_End((R_End))
R_Start --> Search
Search --> Scrape
Scrape --> Summarize
Summarize --> Eval
Eval -- "No (Supplementary search)" --> Search
Eval -- "Yes (Research complete)" --> R_End
end
%% Main Graph Flow Logic
Start --> Planner
Planner -- "Pass research requirements" --> Researcher_SubGraph
Researcher_SubGraph -- "Return research report" --> Writer
Writer --> Editor
Editor --> End
%% Styling
classDef mainNode fill:#2d3436,stroke:#74b9ff,stroke-width:2px,color:#fff;
classDef subNode fill:#0984e3,stroke:#00cec9,stroke-width:2px,color:#fff;
classDef subGraphBox fill:#dfe6e9,stroke:#b2bec3,stroke-width:2px,stroke-dasharray: 5 5,color:#2d3436;
class Planner,Writer,Editor mainNode;
class Search,Scrape,Summarize,Eval subNode;
class Researcher_SubGraph subGraphBox;Core Difficulty: State Isolation and Merging
When you put one graph as a node into another graph, the most critical question is: How does the data interact?
In LangGraph, there are two common approaches:
- Shared State: The parent graph and child graph use the exact same
TypedDictorPydanticmodel. The child graph directly reads and writes the parent graph's state. This approach is simple but destroys encapsulation. The child graph shouldn't know about data likedraft_contentin the parent graph that has nothing to do with it. - State Mapping / Wrapper: The parent graph and child graph have their own independent States. When the parent graph calls the child graph, we only pass in the data the child graph needs. After the child graph finishes running, the results are mapped back into the parent graph's State. (This is the standard for senior architects, and the method we will use in today's practical exercise)
💻 Practical Code Walkthrough
Let's dive straight into the code. To allow you to run this directly, I used Mock functions instead of real LLM calls, focusing on demonstrating the Graph routing and nesting logic.
Please read the bilingual comments in the code carefully; this is the essence I've gained from 10 years of trial and error.
import operator
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, START, END
# =====================================================================
# Part 1: Define States
# =====================================================================
# 1. Sub-graph State (Researcher State)
# The researcher only needs to care about: what to search (query), what was found (raw_docs), and the final summary (research_summary)
class ResearcherState(TypedDict):
research_query: str
raw_docs: Annotated[List[str], operator.add] # Use operator.add to automatically append to the list
research_summary: str
iteration_count: int # Record the number of iterations to prevent infinite loops
# 2. Main Graph State (Agency State)
# The main graph is a global perspective, containing all information such as the plan, research report, first draft, etc.
class AgencyState(TypedDict):
topic: str
planner_outline: str
# Note: The main graph does not need raw_docs; it only wants the researcher's final summary
final_research_report: str
draft: str
# =====================================================================
# Part 2: Build the Sub-graph (Researcher's internal workflow)
# =====================================================================
def search_node(state: ResearcherState):
print(f" [Researcher-Search] Searching in search engine: {state['research_query']}")
# Mock search results
new_doc = f"Web snippet about {state['research_query']}..."
return {"raw_docs": [new_doc], "iteration_count": state.get("iteration_count", 0) + 1}
def summarize_node(state: ResearcherState):
print(f" [Researcher-Summarize] Extracting from {len(state['raw_docs'])} documents...")
# Mock summarize logic
summary = f"[In-depth Research Report] Based on {len(state['raw_docs'])} documents, the core conclusions are as follows..."
return {"research_summary": summary}
def check_enough_data(state: ResearcherState):
# Mock evaluation logic: Assume materials are sufficient after 2 searches
if state["iteration_count"] >= 2:
print(" [Researcher-Eval] Materials are sufficient, ending research.")
return "sufficient"
else:
print(" [Researcher-Eval] Materials are insufficient, digging deeper!")
return "insufficient"
# Assemble the Researcher Sub-graph
researcher_builder = StateGraph(ResearcherState)
researcher_builder.add_node("search", search_node)
researcher_builder.add_node("summarize", summarize_node)
researcher_builder.add_edge(START, "search")
researcher_builder.add_edge("search", "summarize")
researcher_builder.add_conditional_edges(
"summarize",
check_enough_data,
{
"sufficient": END,
"insufficient": "search" # Loop back to continue searching
}
)
# Compile the sub-graph! Now it becomes a callable Runnable object
researcher_graph = researcher_builder.compile()
# =====================================================================
# Part 3: Build the Main Graph (AI Content Agency main workflow)
# =====================================================================
def planner_node(state: AgencyState):
print(f"\n[Planner] Received topic: {state['topic']}. Generating outline and research requirements...")
outline = f"Outline for <{state['topic']}>: 1. Background 2. Current Status 3. Trends"
return {"planner_outline": outline}
# 💡 Core Magic: Wrapper function (State Mapper)
# Because AgencyState and ResearcherState are different, we need a "translator"
def researcher_wrapper_node(state: AgencyState):
print(f"\n[Main Graph] Waking up Researcher department, handing over research task...")
# 1. Extract information from the main state to construct the initial state required by the sub-graph
initial_researcher_state = ResearcherState(
research_query=f"Deep dive: {state['topic']} ({state['planner_outline']})",
raw_docs=[],
research_summary="",
iteration_count=0
)
# 2. Call the sub-graph (just like calling a regular LLM or Tool)
# invoke() will synchronously execute all logic of the sub-graph until it reaches END
final_researcher_state = researcher_graph.invoke(initial_researcher_state)
print(f"[Main Graph] Researcher department work completed, receiving report.")
# 3. Map the output of the sub-graph back to the main state
return {"final_research_report": final_researcher_state["research_summary"]}
def writer_node(state: AgencyState):
print(f"\n[Writer] Starting to write based on the research report...\nReferences: {state['final_research_report']}")
draft = f"This is a viral first draft about {state['topic']}!"
return {"draft": draft}
# Assemble the Main Graph
agency_builder = StateGraph(AgencyState)
agency_builder.add_node("planner", planner_node)
agency_builder.add_node("researcher_dept", researcher_wrapper_node) # Mount the sub-graph Wrapper
agency_builder.add_node("writer", writer_node)
agency_builder.add_edge(START, "planner")
agency_builder.add_edge("planner", "researcher_dept")
agency_builder.add_edge("researcher_dept", "writer")
agency_builder.add_edge("writer", END)
# Compile the main graph
agency_graph = agency_builder.compile()
# =====================================================================
# Part 4: Run Tests
# =====================================================================
if __name__ == "__main__":
print("🚀 Starting AI Content Agency workflow...\n" + "="*40)
initial_state = AgencyState(
topic="2024 Humanoid Robot Industry Development Trends",
planner_outline="",
final_research_report="",
draft=""
)
# Execute the main graph
final_state = agency_graph.invoke(initial_state)
print("\n" + "="*40 + "\n🎉 Final Output Result:")
print(final_state["draft"])
Simulation Output
When you run this code, you will see an extremely clear hierarchical structure: the main graph advances steadily, the sub-graph frantically grinds (loops) internally, but the main graph doesn't care about this at all and only waits for the result.
🚀 Starting AI Content Agency workflow...
========================================
[Planner] Received topic: 2024 Humanoid Robot Industry Development Trends. Generating outline and research requirements...
[Main Graph] Waking up Researcher department, handing over research task...
[Researcher-Search] Searching in search engine: Deep dive: 2024 Humanoid Robot Industry Development Trends (Outline for <2024 Humanoid Robot Industry Development Trends>: 1. Background 2. Current Status 3. Trends)
[Researcher-Summarize] Extracting from 1 documents...
[Researcher-Eval] Materials are insufficient, digging deeper!
[Researcher-Search] Searching in search engine: Deep dive: 2024 Humanoid Robot Industry Development Trends (Outline for <2024 Humanoid Robot Industry Development Trends>: 1. Background 2. Current Status 3. Trends)
[Researcher-Summarize] Extracting from 2 documents...
[Researcher-Eval] Materials are sufficient, ending research.
[Main Graph] Researcher department work completed, receiving report.
[Writer] Starting to write based on the research report...
References: [In-depth Research Report] Based on 2 documents, the core conclusions are as follows...
========================================
🎉 Final Output Result:
This is a viral first draft about 2024 Humanoid Robot Industry Development Trends!
Pitfalls & Guidelines
Introducing Sub-graphs in practice often makes beginners miserable. As your mentor, I have already cleared the mines for you. Pay attention to the following three points:
💣 Pitfall 1: Data pollution caused by mixing parent and child states
Symptom: To save trouble, directly passing AgencyState to researcher_graph. As a result, a bug inside the Researcher overwrites the Writer's draft field to empty.
Guideline: Always, always, always use a Wrapper function for state isolation (just like the researcher_wrapper_node we wrote in the code). Adhere to the Law of Demeter (Principle of Least Knowledge); the sub-graph only needs to know the data it should know and return the data it should return.
💣 Pitfall 2: Sub-graph falls into an infinite loop, causing the main graph to freeze completely
Symptom: Due to LLM hallucinations, the Researcher constantly feels "materials are insufficient" and calls Search an infinite number of times, causing the entire Agency system to hang and API bills to explode.
Guideline: Be sure to add an iteration_count field in the sub-graph's State. In the sub-graph's Conditional Edge, forcefully set a maximum number of loops (e.g., if count >= 3: return END). Do not completely trust the LLM's judgment logic.
💣 Pitfall 3: Loss of hierarchy in Streaming output
Symptom: When using .stream() to monitor the main graph, you find that the execution information of the nodes inside the sub-graph is completely lost, and you can only see the start and end of researcher_dept.
Guideline: In LangGraph, if you want to still track the streaming output of underlying nodes when using a Sub-graph, you need to pass the subgraphs=True parameter when calling the main graph's .stream(). For example: agency_graph.stream(initial_state, subgraphs=True). This way, LangGraph will yield the nested hierarchy together.
📝 Summary for this Issue
Today we completed an architectural sublimation.
Through Sub-graphs technology, we refactored an originally bloated system into a highly cohesive, loosely coupled modular architecture.
- For the Main Graph, the Researcher is just a regular node that inputs a "topic" and outputs a "report".
- For the Researcher, it has its own independent state machine, independent loop logic, and independent retry mechanism.
This design allows you to replace or upgrade the Researcher team at any time in the future, or even package the Researcher Sub-graph independently as a microservice and publish it, without changing a single line of code in the Main Graph! This is the charm of architecture design.
Spoiler Alert: Right now, our Agency is running very smoothly, but what if the Editor thinks the article written by the Writer is no good, or even the Planner thinks it's off-topic? In the next issue (Issue 19), we will introduce LangGraph's advanced features: Human-in-the-loop and Breakpoints. We will make the system pause at critical nodes and wait for your (the boss's) approval before continuing execution.
Stay passionate, see you in the next issue! Keep coding!