lesson-18

Welcome back, AI Architects, to our LangGraph Masterclass. It's your old friend here.

Over the past 17 episodes, our "AI Content Agency" has really started taking shape. We have a strategic Planner, a hardworking Researcher digging up data, a prolific Writer drafting away, and a meticulous Editor.

However, as our business logic deepens, have you noticed a headache-inducing problem? Our Main Graph is starting to look like a plate of spaghetti!

Think back to the last episode. To enable the Researcher to "Search -> Scrape -> Summarize -> Evaluate (and search again if needed)", we added a massive cluster of nodes and conditional edges. The result? The main graph is now densely packed with the Researcher's internal logic. The Planner and Writer are shoved into the corners, and the overall readability and maintainability of our architecture have plummeted.

Think of it this way: as the CEO (the Main Graph) of a company, you don't need—and shouldn't try—to micromanage how the Research Department (the Researcher) holds meetings, looks up data, or argues internally. You just need to hand the "research requirements" to the Head of Research and wait for them to hand you back the "research report."

This brings us to today's ultimate weapon: Sub-graphs.

Today, we are going to perform a "surgical-level refactoring" on our system. We will 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 standard Node.

Grab your coffee, let's dive in!

🎯 Learning Objectives

1. Master the Core Philosophy of Sub-graphs: Understand the underlying logic of "graphs within graphs" to drastically reduce system complexity.
2. Bridge Parent-Child State Mapping: Figure out exactly how data flows in and out between the Main Graph and the Sub-graph (this is where most people trip up).
3. Refactor the Agency Architecture: Extract the Researcher's "Search-Scrape-Summarize" loop into an independent sub-graph, returning the main flow to a minimalist "Planner -> Researcher -> Writer" pipeline.
4. Master Independent Sub-graph Debugging: Learn how to unit-test a sub-graph in isolation without booting up the entire system.

📖 Principle Analysis

In LangGraph, any StateGraph that has been compiled via compile() can be used directly as a Node in another StateGraph.

This sounds simple, but it is incredibly powerful. It means you can nest infinitely: a company contains departments, departments contain teams, teams contain individuals. This Fractal Architecture is the only viable solution for building enterprise-grade, complex Multi-Agent systems.

Let's compare the architecture before and after refactoring.

The "Spaghetti Graph" Before Refactoring (What Not to Do)

If you cram all the logic into a single graph, it looks like this: After the Planner finishes, it enters the Search node, then evaluates if Scraping is needed, then goes to Summarize, then checks if the data is sufficient... The entire main storyline is severely fragmented.

The "Modular" Architecture After Refactoring (Today's Goal)

We encapsulate the Researcher's internal logic into a black box (Sub-graph).

Take a look at our architectural design for today:

graph TD
    %% Main Graph Node Definitions
    Start((START))
    Planner[Planner Node\nGenerate Outline & Research Needs]
    Writer[Writer Node\nDraft Content from Data]
    Editor[Editor Node\nReview & Polish]
    End((END))

    %% Sub-graph Definition (Folded inside the Researcher Node)
    subgraph Researcher_SubGraph ["🔍 Researcher Node (Sub-graph Internal Logic)"]
        direction TB
        R_Start((R_Start))
        Search[Web Search\nSearch Keywords]
        Scrape[Web Scrape\nScrape Web Content]
        Summarize[Summarize\nExtract Core Info]
        Eval{Enough Data for Draft?}
        R_End((R_End))

        R_Start --> Search
        Search --> Scrape
        Scrape --> Summarize
        Summarize --> Eval
        Eval -- "No (Search More)" --> Search
        Eval -- "Yes (Research Done)" --> R_End
    end

    %% Main Graph Flow Logic
    Start --> Planner
    Planner -- "Pass Research Needs" --> 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 Challenge: State Isolation and Merging

When you place one graph as a node inside another, the most critical question is: How does the data interact?

In LangGraph, there are two common approaches:

1. Shared State: The parent graph and sub-graph use the exact same TypedDict or Pydantic model. The sub-graph directly reads and writes the parent graph's state. This is easy but destroys encapsulation. The sub-graph shouldn't know about irrelevant data like draft_content existing in the parent graph.
2. State Mapping / Wrapper: The parent graph and sub-graph have their own independent States. When the parent calls the sub-graph, we only pass in the data the sub-graph needs. Once the sub-graph finishes, we map its results back into the parent graph's State. (This is the standard for senior architects, and the method we'll use in today's practice.)

💻 Practical Code Walkthrough

Let's jump straight into the code. To ensure you can run this immediately, I've used Mock functions instead of real LLM calls, focusing entirely on Graph routing and nesting logic.

Please read the bilingual comments in the code carefully—they are the essence of my 10 years of debugging experience.

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 cares about: what to search (query), what was found (raw_docs), and the final 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 # Track iteration count to prevent infinite loops

# 2. Main Graph State (Agency State)
# The main graph has a global view, containing outlines, reports, drafts, etc.
class AgencyState(TypedDict):
    topic: str
    planner_outline: str
    # Note: The main graph doesn't need raw_docs; it only wants the final summary from the researcher.
    final_research_report: str 
    draft: str

# =====================================================================
# Part 2: Build Sub-graph (Researcher's internal workflow)
# =====================================================================

def search_node(state: ResearcherState):
    print(f"  [Researcher-Search] Searching engine for: {state['research_query']}")
    # Mock search result
    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 info from {len(state['raw_docs'])} docs...")
    # Mock summarize logic
    summary = f"[In-Depth Report] Based on {len(state['raw_docs'])} docs, core conclusions are..."
    return {"research_summary": summary}

def check_enough_data(state: ResearcherState):
    # Mock evaluation logic: Assume data is sufficient after 2 iterations
    if state["iteration_count"] >= 2:
        print("  [Researcher-Eval] Data is sufficient. Ending research.")
        return "sufficient"
    else:
        print("  [Researcher-Eval] Data insufficient. Digging deeper!")
        return "insufficient"

# Assemble 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 search again
    }
)

# Compile the sub-graph! It now becomes a callable Runnable object
researcher_graph = researcher_builder.compile()

# =====================================================================
# Part 3: Build Main Graph (AI Content Agency main flow)
# =====================================================================

def planner_node(state: AgencyState):
    print(f"\n[Planner] Received topic: {state['topic']}. Generating outline and research needs...")
    outline = f"Outline for '{state['topic']}': 1. Background 2. Current Status 3. Trends"
    return {"planner_outline": outline}

# 💡 Core Magic: Wrapper Function (State Mapper)
# Since AgencyState and ResearcherState differ, we need a 'translator'
def researcher_wrapper_node(state: AgencyState):
    print(f"\n[Main Graph] Waking up Researcher Dept, handing over research task...")
    
    # 1. Extract info from main state to construct initial state for sub-graph
    initial_researcher_state = ResearcherState(
        research_query=f"Deep dive: {state['topic']} ({state['planner_outline']})",
        raw_docs=[],
        research_summary="",
        iteration_count=0
    )
    
    # 2. Invoke sub-graph (just like invoking a standard LLM or Tool)
    # invoke() synchronously executes all sub-graph logic until it reaches END
    final_researcher_state = researcher_graph.invoke(initial_researcher_state)
    
    print(f"[Main Graph] Researcher Dept finished work. Receiving report.")
    
    # 3. Map sub-graph output back to the main state
    return {"final_research_report": final_researcher_state["research_summary"]}

def writer_node(state: AgencyState):
    print(f"\n[Writer] Drafting based on research report...\nReferences: {state['final_research_report']}")
    draft = f"This is a viral draft article about {state['topic']}!"
    return {"draft": draft}

# Assemble Main Graph
agency_builder = StateGraph(AgencyState)
agency_builder.add_node("planner", planner_node)
agency_builder.add_node("researcher_dept", researcher_wrapper_node) # Mount 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 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 Trends",
        planner_outline="",
        final_research_report="",
        draft=""
    )
    
    # Execute main graph
    final_state = agency_graph.invoke(initial_state)
    
    print("\n" + "="*40 + "\n🎉 Final Output:")
    print(final_state["draft"])

Simulated Execution Output

When you run this code, you will see an incredibly clear hierarchical structure: the main graph advances steadily, the sub-graph grinds away internally (looping), but the main graph doesn't care about the internal chaos—it just waits for the result.

🚀 Starting AI Content Agency Workflow...
========================================

[Planner] Received topic: 2024 Humanoid Robot Industry Trends. Generating outline and research needs...

[Main Graph] Waking up Researcher Dept, handing over research task...
  [Researcher-Search] Searching engine for: Deep dive: 2024 Humanoid Robot Industry Trends (Outline for '2024 Humanoid Robot Industry Trends': 1. Background 2. Current Status 3. Trends)
  [Researcher-Summarize] Extracting info from 1 docs...
  [Researcher-Eval] Data insufficient. Digging deeper!
  [Researcher-Search] Searching engine for: Deep dive: 2024 Humanoid Robot Industry Trends (Outline for '2024 Humanoid Robot Industry Trends': 1. Background 2. Current Status 3. Trends)
  [Researcher-Summarize] Extracting info from 2 docs...
  [Researcher-Eval] Data is sufficient. Ending research.
[Main Graph] Researcher Dept finished work. Receiving report.

[Writer] Drafting based on research report...
References: [In-Depth Report] Based on 2 docs, core conclusions are...

========================================
🎉 Final Output:
This is a viral draft article about 2024 Humanoid Robot Industry Trends!

Pitfalls & How to Avoid Them

Introducing Sub-graphs in real-world projects often causes agonizing pain for beginners. As your mentor, I've already cleared the minefield for you. Pay attention to these three points:

💣 Pitfall 1: Mixing Parent/Child States Causing Data Pollution

Symptom: To save time, you pass AgencyState directly into researcher_graph. As a result, a bug inside the Researcher overwrites the Writer's draft field, wiping it blank. Solution: 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): a sub-graph should only know the data it needs to know and return the data it needs to return.

💣 Pitfall 2: Sub-graph Infinite Loops Freezing the Main Graph

Symptom: Due to LLM hallucinations, the Researcher constantly feels "data is insufficient" and calls Search infinitely. This causes the entire Agency system to freeze and your API bill to explode. Solution: You must add an iteration_count field to the sub-graph's State. In the sub-graph's Conditional Edge, enforce a maximum loop limit (e.g., if count >= 3: return END). Never fully trust an LLM's evaluation logic.

💣 Pitfall 3: Losing Hierarchy in Streaming Outputs

Symptom: When using .stream() to monitor the main graph, you notice all execution info for nodes inside the sub-graph is missing. You only see researcher_dept start and end. Solution: In LangGraph, if you want to track the streaming output of underlying nodes while using a Sub-graph, you need to pass the subgraphs=True parameter when calling .stream() on the main graph. For example: agency_graph.stream(initial_state, subgraphs=True). This tells LangGraph to yield the nested hierarchy as well.

📝 Episode Summary

Today, we achieved an architectural elevation.

Through Sub-graphs, we refactored a bloated system into a highly cohesive, loosely coupled modular architecture.

• To the Main Graph, the Researcher is just a standard node that takes a "topic" as input and outputs a "report".
• To the Researcher, it possesses its own independent state machine, independent loop logic, and independent retry mechanisms.

This design allows you to swap out or upgrade the Researcher team at any time in the future. You could even package the Researcher Sub-graph as a standalone microservice and deploy it—without changing a single line of code in the Main Graph! That is the beauty of architecture design.

Spoiler Alert: Right now, our Agency is running smoothly. But what if the Editor thinks the Writer's draft is terrible, or the Planner thinks it's completely off-topic? In the next episode (Part 19), we will introduce LangGraph's advanced features: Human-in-the-loop and Breakpoints. We'll make the system pause at critical nodes, waiting for your (the boss's) approval before continuing execution.

Stay passionate, and see you in the next episode! Keep coding!