7. Subgraphs as Agent Modules

Overview

In the previous capsules you built agents as graphs: nodes, edges, conditional routing, cycles, the Functional API. Everything lived in a single StateGraph. While your agent had 4-5 nodes, that worked. But think about what's coming: in M8 you'll build a multi-agent system where a supervisor orchestrates specialized agents — a research agent, an analysis agent, a synthesis agent. If each agent is a block of 15 nodes flattened into one graph, your state machine becomes an unmanageable monster.

Subgraphs solve this problem. A subgraph is a complete StateGraph — with its own nodes, edges, state, and logic — used as a node inside another graph. It's the same idea as functions in programming: you encapsulate a complete capability behind a clear interface. The difference is that here it isn't a function — it's an entire graph with its own flow, its own cycles, and its own routing logic.

This capsule is the direct bridge to M8 (Multi-Agent Orchestration). The pattern parent.add_node("agent_name", compiled_subgraph) is exactly how agents are implemented in a multi-agent system. If you master subgraphs here, M8 will be a natural extension.

The Problem: Monolithic Graphs

What a graph that grew too big looks like

Imagine you evolve the Research Agent. You started with 4 nodes. Now research needs to search the web, query databases, and validate sources. Analysis needs to extract entities, classify, and evaluate. Flattened: 10+ nodes, 15+ conditional edges, 1 TypedDict with 20+ fields.

The concrete problems

1. Contaminated state. Every node shares a single TypedDict. source_validity, which only matters to research, is also visible to synthesis. A bug in one corrupts another's data.

2. You can't test in isolation. To check that analysis works, you have to spin up the entire graph.

3. You can't reuse. The analysis module is coupled to the global state and to the graph's edges.

4. Routing explosion. With 10 nodes and conditional edges, debugging becomes tracing which path it took among dozens of possibilities.

The solution: divide and compose

Instead of one monolithic graph, you build 3 subgraphs of 3-4 nodes each:

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│   RESEARCH   │     │   ANALYSIS   │     │  SYNTHESIS   │
│   subgraph   │     │   subgraph   │     │   subgraph   │
│              │     │              │     │              │
│ search →     │     │ extract →    │     │ draft →      │
│ validate →   │     │ classify →   │     │ review →     │
│ collect      │     │ evaluate     │     │ format       │
└──────┬───────┘     └──────┬───────┘     └──────┬───────┘
       └────────────────────┼────────────────────┘
                   ┌────────▼────────┐
                   │  PARENT GRAPH   │
                   │ plan → research │
                   │  → analysis     │
                   │  → synthesis    │
                   └─────────────────┘

Each subgraph has its own state, its own logic, and its own interface. The parent graph only knows they exist, not how they work internally.

Subgraphs: Graphs Inside Graphs

The concept

A subgraph is a StateGraph that: (1) is defined with its own typed state, (2) has its own nodes and edges, (3) is compiled with .compile(), and (4) is used as a node in another graph with add_node(). LangGraph treats a compiled subgraph like any other node.

Your first subgraph

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

model = init_chat_model("openai:gpt-4.1-mini")

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]

def search_node(state: ResearchState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Search for relevant information about the topic.")]
        + state["messages"]
    )
    return {"messages": [response]}

def process_node(state: ResearchState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Extract the 3 key points from the research.")]
        + state["messages"]
    )
    return {"messages": [response]}

research_graph = StateGraph(ResearchState)
research_graph.add_node("search", search_node)
research_graph.add_node("process", process_node)
research_graph.add_edge(START, "search")
research_graph.add_edge("search", "process")
research_graph.add_edge("process", END)

research_subgraph = research_graph.compile()

research_subgraph is a fully functional graph. You can invoke it directly:

result = research_subgraph.invoke({
    "messages": [HumanMessage(content="What is RAG?")]
})
print(result["messages"][-1].content)

Using it as a node in a parent graph

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

def planning_node(state: AgentState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Generate a research plan with 2 sub-questions.")]
        + state["messages"]
    )
    return {"messages": [response]}

def synthesis_node(state: AgentState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Synthesize all the information into a final answer.")]
        + state["messages"]
    )
    return {"messages": [response]}

parent_graph = StateGraph(AgentState)
parent_graph.add_node("plan", planning_node)
parent_graph.add_node("research", research_subgraph)  # Subgraph as a node
parent_graph.add_node("synthesize", synthesis_node)

parent_graph.add_edge(START, "plan")
parent_graph.add_edge("plan", "research")
parent_graph.add_edge("research", "synthesize")
parent_graph.add_edge("synthesize", END)

agent = parent_graph.compile()
result = agent.invoke({
    "messages": [HumanMessage(content="What are the benefits of RAG?")]
})
print(result["messages"][-1].content)

To the parent graph, "research" is a node like any other. It doesn't know that internally it has search → process.

Visualization

from IPython.display import Image, display

display(Image(agent.get_graph().draw_mermaid_png()))

display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

Without xray, the diagram shows plan → research → synthesize. With xray=True, LangGraph expands the subgraphs and shows their internal nodes. You get both views: high-level architecture and implementation detail.

State Mapping Between Parent and Subgraph

The mechanism: shared keys

When the parent invokes a subgraph, LangGraph maps the state using keys with the same name. If the parent has messages and the subgraph has messages, the value flows between them.

Parent State                    Subgraph State
┌─────────────────┐            ┌─────────────────┐
│ messages ───────│────────────│─── messages      │
│ plan            │            │                  │
│ iteration_count │            │                  │
└─────────────────┘            └─────────────────┘

The subgraph only sees the keys it shares. The keys it doesn't share (plan, iteration_count) are invisible. That's exactly the isolation we want.

Shared vs private keys

class ParentState(TypedDict):
    messages: Annotated[list, add_messages]
    plan: list[str]
    iteration_count: int

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]  # Shared — same name and type

class AnalysisState(TypedDict):
    messages: Annotated[list, add_messages]  # Shared
    analysis_depth: int                      # Subgraph-only — private

When the parent invokes the research subgraph: messages gets passed (shared key), plan and iteration_count don't (they don't exist in ResearchState). When the subgraph finishes, its updated messages flows back to the parent.

Explicit input/output with wrapper functions

When the states differ significantly, you use a wrapper function:

class ParentState(TypedDict):
    messages: Annotated[list, add_messages]
    raw_data: str

class AnalysisSubState(TypedDict):
    messages: Annotated[list, add_messages]
    input_text: str

def analysis_wrapper(state: ParentState) -> dict:
    sub_result = analysis_subgraph.invoke({
        "messages": state["messages"],
        "input_text": state.get("raw_data", ""),
    })
    return {"messages": sub_result["messages"],
            "raw_data": sub_result["messages"][-1].content}

The wrapper translates raw_data → input_text on the way in, giving you total control over the mapping when the interfaces don't line up.

Designing Modules for the Research Agent

Three modules, three responsibilities

Research Module (with a reason-act loop)

from langchain_core.tools import tool
from langchain_core.messages import ToolMessage

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    return f"Results for '{query}': relevant data found."

class ResearchModuleState(TypedDict):
    messages: Annotated[list, add_messages]

research_tools = [search_web]
research_model = model.bind_tools(research_tools)
tools_by_name = {t.name: t for t in research_tools}

def research_reason(state: ResearchModuleState) -> dict:
    response = research_model.invoke(
        [SystemMessage(content="Research the topic. Use tools. "
                       "When you have enough, answer directly.")]
        + state["messages"]
    )
    return {"messages": [response]}

def research_tools_node(state: ResearchModuleState) -> dict:
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:
        result = str(tools_by_name[tc["name"]].invoke(tc["args"]))
        results.append(ToolMessage(content=result, tool_call_id=tc["id"]))
    return {"messages": results}

def research_route(state: ResearchModuleState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return "done"

rg = StateGraph(ResearchModuleState)
rg.add_node("reason", research_reason)
rg.add_node("tools", research_tools_node)
rg.add_edge(START, "reason")
rg.add_conditional_edges("reason", research_route, {"tools": "tools", "done": END})
rg.add_edge("tools", "reason")
research_module = rg.compile()

Analysis and Synthesis Modules

The same pattern — its own state, internal nodes, compile, done:

class AnalysisModuleState(TypedDict):
    messages: Annotated[list, add_messages]

def extract_key_points(state: AnalysisModuleState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Extract the 5 key points.")] + state["messages"]
    )]}

def evaluate_quality(state: AnalysisModuleState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Evaluate the quality. Is it enough? What's missing?")]
        + state["messages"]
    )]}

ag = StateGraph(AnalysisModuleState)
ag.add_node("extract", extract_key_points)
ag.add_node("evaluate", evaluate_quality)
ag.add_edge(START, "extract")
ag.add_edge("extract", "evaluate")
ag.add_edge("evaluate", END)
analysis_module = ag.compile()

class SynthesisModuleState(TypedDict):
    messages: Annotated[list, add_messages]

def generate_draft(state: SynthesisModuleState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Generate a complete, structured answer.")]
        + state["messages"]
    )]}

def refine_response(state: SynthesisModuleState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Review it: improve clarity, remove redundancies.")]
        + state["messages"]
    )]}

syg = StateGraph(SynthesisModuleState)
syg.add_node("draft", generate_draft)
syg.add_node("refine", refine_response)
syg.add_edge(START, "draft")
syg.add_edge("draft", "refine")
syg.add_edge("refine", END)
synthesis_module = syg.compile()

Isolated testing

Each module gets tested separately without spinning up the whole agent:

test_msg = [HumanMessage(content="Explain retrieval-augmented generation")]

r1 = research_module.invoke({"messages": test_msg})
r2 = analysis_module.invoke({"messages": r1["messages"]})
r3 = synthesis_module.invoke({"messages": r2["messages"]})
print(r3["messages"][-1].content[:200])

Composing Subgraphs

Parent graph with conditional routing

Now we join the three modules with a parent graph that includes a quality check and a re-research cycle:

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    quality_score: float
    cycle_count: int

def planning_node(state: AgentState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Analyze the question and generate a plan with sub-questions.")]
        + state["messages"]
    )
    return {"messages": [response]}

def quality_check_node(state: AgentState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Rate the quality 0.0-1.0. Answer with ONLY the number.")]
        + state["messages"]
    )
    try:
        score = float(response.content.strip())
    except ValueError:
        score = 0.5
    return {
        "quality_score": score,
        "cycle_count": state.get("cycle_count", 0) + 1,
    }

def route_by_quality(state: AgentState) -> str:
    if state.get("cycle_count", 0) >= 3:
        return "synthesis"
    if state.get("quality_score", 0.0) >= 0.7:
        return "synthesis"
    return "research"

parent = StateGraph(AgentState)
parent.add_node("plan", planning_node)
parent.add_node("research", research_module)      # Subgraph
parent.add_node("analysis", analysis_module)       # Subgraph
parent.add_node("quality_check", quality_check_node)
parent.add_node("synthesis", synthesis_module)     # Subgraph

parent.add_edge(START, "plan")
parent.add_edge("plan", "research")
parent.add_edge("research", "analysis")
parent.add_edge("analysis", "quality_check")
parent.add_conditional_edges("quality_check", route_by_quality, {
    "research": "research",
    "synthesis": "synthesis",
})
parent.add_edge("synthesis", END)

agent = parent.compile()

result = agent.invoke({
    "messages": [HumanMessage(content="How do AI agents work with LangGraph?")],
    "quality_score": 0.0,
    "cycle_count": 0,
})
print(result["messages"][-1].content)

The parent has a cycle: research → analysis → quality_check → (research if quality is low). Each iteration invokes complete subgraphs. The research subgraph might internally run 3-4 iterations of its own reason-tools loop, but the parent neither knows nor cares.

from IPython.display import Image, display
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

With xray=True you see the complete architecture: the parent with its quality-driven cycle, and inside each subgraph its internal nodes.

Subgraphs vs Normal Functions

A normal function also encapsulates logic. Why bother with a subgraph?

Decision criteria

Practical rules

Use a normal function when:

1. The logic is a single model call or a 2-step sequence with no cycles
2. You don't need intermediate checkpointing
3. You won't reuse that logic in other agents

Use a subgraph when:

1. The logic has internal cycles (its own reason-act loop)
2. You need per-step checkpointing (production, fault tolerance)
3. The module will be reused in other agents or in M8 as a standalone agent
4. The module has 3+ nodes with routing between them

The key question: "Is this module a mini-agent with its own flow, or is it a simple step?" Mini-agent → Subgraph. Simple step → Normal function.

Connection to the Project

In this module's project (capsule 08, Research Agent State Machine), you'll build the Research Agent as a StateGraph with planning, research, analysis, and synthesis nodes. Now you know you can implement research and analysis as subgraphs. The project lets you choose: a flat graph or composition with subgraphs.

In later modules:

• M5 (Planning): The planning module can be a subgraph with its own plan → evaluate → re-plan cycle.
• M6 (Memory): Each subgraph can have its own checkpointer. Research persists its progress independently of the parent.
• M7 (MCP): The research subgraph can integrate MCP tools internally without the parent knowing.
• M8 (Multi-Agent): Each agent is a compiled subgraph. The supervisor is the parent graph. parent.add_node("researcher", researcher_subgraph) is literal — it's the code. What you learned here applies directly.

Troubleshooting

Problem 1: The subgraph doesn't receive the parent's state

Cause: The state keys don't match. LangGraph maps by key name.

# ❌ Different names — no mapping
class ParentState(TypedDict):
    messages: Annotated[list, add_messages]
class SubState(TypedDict):
    msgs: Annotated[list, add_messages]

# ✅ Same name
class SubState(TypedDict):
    messages: Annotated[list, add_messages]

Problem 2: The parent's state doesn't update after the subgraph

Cause: The subgraph returns keys that don't exist in the parent state. Only shared keys sync. Make sure the subgraph's output keys exist in ParentState with the same reducer.

Problem 3: TypeError when compiling the parent with a subgraph

Cause: You passed the StateGraph without compiling it.

# ❌ Not compiled
parent.add_node("research", research_graph)

# ✅ Compiled
parent.add_node("research", research_graph.compile())

Problem 4: draw_mermaid_png doesn't show the subgraph's internal nodes

Cause: By default, subgraphs are opaque boxes.

agent.get_graph().draw_mermaid_png()            # Only shows "research" as a block
agent.get_graph(xray=True).draw_mermaid_png()   # Shows the internal nodes

Problem 5: The subgraph enters an infinite loop

Cause: The subgraph has a cycle with no stop condition. The parent can't interrupt it — always include an iteration limit inside the subgraph with an internal_iterations field in its state.

Exercises

Exercise 1: Basic subgraph as a node (Easy)

Create a translation subgraph with two nodes (detect_language → translate). Compile it and use it as a node in a parent graph: process_input → translate_subgraph → format_output.

class TranslateState(TypedDict):
    messages: Annotated[list, add_messages]

def detect_language(state: TranslateState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Detect the language.")] + state["messages"])]}
def translate(state: TranslateState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content="Translate to English.")] + state["messages"])]}

tg = StateGraph(TranslateState)
tg.add_node("detect", detect_language)
tg.add_node("translate", translate)
tg.add_edge(START, "detect")
tg.add_edge("detect", "translate")
tg.add_edge("translate", END)
translate_module = tg.compile()

class ParentState(TypedDict):
    messages: Annotated[list, add_messages]

parent = StateGraph(ParentState)
parent.add_node("process", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Clean up the text.")] + s["messages"])]})
parent.add_node("translate", translate_module)
parent.add_node("format", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Format the translation.")] + s["messages"])]})
parent.add_edge(START, "process")
parent.add_edge("process", "translate")
parent.add_edge("translate", "format")
parent.add_edge("format", END)
agent = parent.compile()

result = agent.invoke({"messages": [HumanMessage(content="Hola, ¿cómo estás?")]})
print(result["messages"][-1].content)

Exercise 2: Subgraph with an internal cycle (Medium)

Create a research subgraph with an internal reason-act loop (maximum 3 iterations, using search_web). Insert it into a parent: plan → research_subgraph → summarize.

class RState(TypedDict):
    messages: Annotated[list, add_messages]
    iterations: int

model_with_tools = model.bind_tools([search_web])

def reason(state: RState) -> dict:
    return {"messages": [model_with_tools.invoke(
        [SystemMessage(content="Research. Answer when you have enough.")]
        + state["messages"])], "iterations": state.get("iterations", 0) + 1}

def exec_tools(state: RState) -> dict:
    last = state["messages"][-1]
    return {"messages": [ToolMessage(content=str(tools_by_name[tc["name"]].invoke(
        tc["args"])), tool_call_id=tc["id"]) for tc in last.tool_calls]}

def should_continue(state: RState) -> str:
    if state.get("iterations", 0) >= 3: return "done"
    last = state["messages"][-1]
    return "tools" if hasattr(last, "tool_calls") and last.tool_calls else "done"

rg = StateGraph(RState)
rg.add_node("reason", reason)
rg.add_node("tools", exec_tools)
rg.add_edge(START, "reason")
rg.add_conditional_edges("reason", should_continue, {"tools": "tools", "done": END})
rg.add_edge("tools", "reason")
research_sub = rg.compile()

class PState(TypedDict):
    messages: Annotated[list, add_messages]

parent = StateGraph(PState)
parent.add_node("plan", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Generate a brief plan.")] + s["messages"])]})
parent.add_node("research", research_sub)
parent.add_node("summarize", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Summarize in 3 points.")] + s["messages"])]})
parent.add_edge(START, "plan")
parent.add_edge("plan", "research")
parent.add_edge("research", "summarize")
parent.add_edge("summarize", END)
agent = parent.compile()

result = agent.invoke({"messages": [HumanMessage(content="What is vector search?")]})

Exercise 3: State mapping with a wrapper (Medium)

Create a parent with AgentState(messages, raw_data) and a subgraph with ProcessState(messages, input_text). Implement a wrapper that translates raw_data → input_text when invoking the subgraph.

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    raw_data: str

class ProcessState(TypedDict):
    messages: Annotated[list, add_messages]
    input_text: str

def process_text(state: ProcessState) -> dict:
    return {"messages": [model.invoke(
        [SystemMessage(content=f"Process: {state.get('input_text', '')}")]
        + state["messages"])]}

pg = StateGraph(ProcessState)
pg.add_node("process", process_text)
pg.add_edge(START, "process")
pg.add_edge("process", END)
process_sub = pg.compile()

def process_wrapper(state: AgentState) -> dict:
    sub_result = process_sub.invoke({
        "messages": state["messages"],
        "input_text": state.get("raw_data", ""),
    })
    return {"messages": sub_result["messages"],
            "raw_data": sub_result["messages"][-1].content}

parent = StateGraph(AgentState)
parent.add_node("collect", lambda s: {"raw_data": "AI agents combine LLMs with tools."})
parent.add_node("process", process_wrapper)
parent.add_edge(START, "collect")
parent.add_edge("collect", "process")
parent.add_edge("process", END)
agent = parent.compile()
result = agent.invoke({"messages": [HumanMessage(content="Analyze")], "raw_data": ""})
print(result["raw_data"][:200])

Exercise 4: Three subgraphs with conditional routing (Hard)

Implement research, analysis, and synthesis as subgraphs. The parent has a quality_check after analysis. If quality_score < 0.7, go back to research (maximum 3 cycles).

class QState(TypedDict):
    messages: Annotated[list, add_messages]
    quality_score: float
    cycle_count: int

# Uses research_module, analysis_module, synthesis_module defined earlier

def quality_check(state: QState) -> dict:
    response = model.invoke(
        [SystemMessage(content="Rate the quality 0.0-1.0. ONLY the number.")]
        + state["messages"])
    try: score = float(response.content.strip())
    except ValueError: score = 0.5
    return {"quality_score": score, "cycle_count": state.get("cycle_count", 0) + 1}

def quality_route(state: QState) -> str:
    if state.get("cycle_count", 0) >= 3: return "synthesis"
    return "synthesis" if state.get("quality_score", 0.0) >= 0.7 else "research"

parent = StateGraph(QState)
parent.add_node("research", research_module)
parent.add_node("analysis", analysis_module)
parent.add_node("quality_check", quality_check)
parent.add_node("synthesis", synthesis_module)
parent.add_edge(START, "research")
parent.add_edge("research", "analysis")
parent.add_edge("analysis", "quality_check")
parent.add_conditional_edges("quality_check", quality_route, {
    "research": "research", "synthesis": "synthesis"})
parent.add_edge("synthesis", END)

agent = parent.compile()
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))
result = agent.invoke({"messages": [HumanMessage(content="Explain transformers")],
    "quality_score": 0.0, "cycle_count": 0})
print(f"Cycles: {result['cycle_count']}, Score: {result['quality_score']:.2f}")

Exercise 5: Reuse a subgraph in two agents (Hard)

Take the analysis_module and use it in two different parent graphs: one research → analysis → synthesis, and another user_input → analysis → response. Show that the same compiled subgraph works in both contexts.

class SharedState(TypedDict):
    messages: Annotated[list, add_messages]

def analyze(state): return {"messages": [model.invoke(
    [SystemMessage(content="Analyze: topics, sentiment, key points.")] + state["messages"])]}
def rate(state): return {"messages": [model.invoke(
    [SystemMessage(content="Rate the importance: high, medium, low.")] + state["messages"])]}

sag = StateGraph(SharedState)
sag.add_node("analyze", analyze)
sag.add_node("rate", rate)
sag.add_edge(START, "analyze")
sag.add_edge("analyze", "rate")
sag.add_edge("rate", END)
shared_analysis = sag.compile()

# Agent 1: Research Pipeline — research → analysis → synthesis
p1 = StateGraph(SharedState)
p1.add_node("research", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Research it.")] + s["messages"])]})
p1.add_node("analysis", shared_analysis)
p1.add_node("synth", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Synthesize it.")] + s["messages"])]})
p1.add_edge(START, "research")
p1.add_edge("research", "analysis")
p1.add_edge("analysis", "synth")
p1.add_edge("synth", END)
agent1 = p1.compile()

# Agent 2: Feedback Pipeline — analysis → response
p2 = StateGraph(SharedState)
p2.add_node("analysis", shared_analysis)
p2.add_node("respond", lambda s: {"messages": [model.invoke(
    [SystemMessage(content="Give a recommendation.")] + s["messages"])]})
p2.add_edge(START, "analysis")
p2.add_edge("analysis", "respond")
p2.add_edge("respond", END)
agent2 = p2.compile()

r1 = agent1.invoke({"messages": [HumanMessage(content="What is fine-tuning?")]})
r2 = agent2.invoke({"messages": [HumanMessage(content="Low retention.")]})
print("Agent 1:", r1["messages"][-1].content[:100])
print("Agent 2:", r2["messages"][-1].content[:100])

Summary

In this capsule you learned:

• Monolithic graphs become unmanageable as an agent grows. Contaminated state, the inability to test in isolation, and routing explosion are the symptoms.
• Subgraphs are compiled StateGraphs used as nodes in a parent graph. Each subgraph has its own state, nodes, edges, and routing logic.
• State mapping works by keys with the same name. Shared keys flow between parent and subgraph; private ones stay isolated. For complex mappings, you use wrapper functions.
• Modular composition: The Research Agent decomposes into research, analysis, and synthesis as independent subgraphs that get tested, reused, and replaced without coupling.
• Subgraph vs normal function: Cycles, routing, or checkpointing → subgraph. A simple call → normal function.
• xray=True in draw_mermaid_png reveals each subgraph's internal nodes.
• Subgraphs = agents in M8. parent.add_node("agent_name", compiled_subgraph) is exactly how you'll implement multi-agent orchestration. Each agent is a subgraph. The supervisor is the parent.

Next capsule: Project — Research Agent with a State Machine. You'll put the whole module into practice: StateGraph, typed state, cycles, routing, and subgraphs to build the v1 of the agent that will evolve all the way to M10.

Additional Resources

1. LangGraph Subgraphs — Conceptual Guide — Official documentation on subgraphs
2. LangGraph Subgraphs — How-to Guide — Step-by-step tutorial for creating and composing subgraphs
3. LangGraph State Management — State mapping between graphs
4. LangGraph Visualization — draw_mermaid_png and xray=True
5. LangGraph Multi-Agent — Subgraph Pattern — A preview of subgraphs in M8