Subgraphs: Modular Composition

Capsule overview

Your "search + parse + summarize" pattern repeats for every source. You have the same 3 steps (fetch → parse → summarize) duplicated 3 times — once for web, once for academic, once for news. If you need to change the parsing logic, you have to modify 3 places. If you add a fourth source, you copy and paste the same 3 nodes all over again.

What if you could define that pipeline once and reuse it? Just as a function can call another function, a graph can contain another graph. You encapsulate complex logic in a standalone graph and use it as a node inside a bigger graph. That's a subgraph.

Subgraphs are the equivalent of functions in modular programming: they encapsulate, abstract, and enable reuse. The difference from a regular function is that a subgraph gives you checkpointing for every internal step, flow visualization, and streaming — things a plain Python function doesn't offer.

The problem: duplicated logic

Look at this graph with 3 search pipelines that are structurally identical:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    query: str
    results: Annotated[list[str], operator.add]

def fetch_web(state: State) -> dict:
    return {"results": [f"[fetch_web] Raw data from web for '{state['query']}'"]}

def parse_web(state: State) -> dict:
    raw = state["results"][-1]
    return {"results": [f"[parse_web] Parsed: {raw}"]}

def summarize_web(state: State) -> dict:
    parsed = state["results"][-1]
    return {"results": [f"[summarize_web] Summary: {parsed}"]}

def fetch_academic(state: State) -> dict:
    return {"results": [f"[fetch_academic] Raw data from academic for '{state['query']}'"]}

def parse_academic(state: State) -> dict:
    raw = state["results"][-1]
    return {"results": [f"[parse_academic] Parsed: {raw}"]}

def summarize_academic(state: State) -> dict:
    parsed = state["results"][-1]
    return {"results": [f"[summarize_academic] Summary: {parsed}"]}

def fetch_news(state: State) -> dict:
    return {"results": [f"[fetch_news] Raw data from news for '{state['query']}'"]}

def parse_news(state: State) -> dict:
    raw = state["results"][-1]
    return {"results": [f"[parse_news] Parsed: {raw}"]}

def summarize_news(state: State) -> dict:
    parsed = state["results"][-1]
    return {"results": [f"[summarize_news] Summary: {parsed}"]}

graph = StateGraph(State)

graph.add_node("fetch_web", fetch_web)
graph.add_node("parse_web", parse_web)
graph.add_node("summarize_web", summarize_web)
graph.add_node("fetch_academic", fetch_academic)
graph.add_node("parse_academic", parse_academic)
graph.add_node("summarize_academic", summarize_academic)
graph.add_node("fetch_news", fetch_news)
graph.add_node("parse_news", parse_news)
graph.add_node("summarize_news", summarize_news)

graph.add_edge(START, "fetch_web")
graph.add_edge("fetch_web", "parse_web")
graph.add_edge("parse_web", "summarize_web")
graph.add_edge("summarize_web", "fetch_academic")
graph.add_edge("fetch_academic", "parse_academic")
graph.add_edge("parse_academic", "summarize_academic")
graph.add_edge("summarize_academic", "fetch_news")
graph.add_edge("fetch_news", "parse_news")
graph.add_edge("parse_news", "summarize_news")
graph.add_edge("summarize_news", END)

app = graph.compile()
result = app.invoke({"query": "Python 3.12", "results": []})
print(f"Total nodes: 9")
print(f"Results: {len(result['results'])}")
# Expected output:
# Total nodes: 9
# Results: 9

9 nodes for 3 pipelines that do the same thing with different data. If you need to add validation between parse and summarize, you have to touch 3 places. If you need a fourth source, you copy 3 more nodes. This doesn't scale.

Subgraphs = functions: the fundamental analogy

Think about Python functions:

def process_source(source_name: str, query: str) -> str:
    raw = fetch(source_name, query)
    parsed = parse(raw)
    summary = summarize(parsed)
    return summary

result_web = process_source("web", "Python 3.12")
result_academic = process_source("academic", "Python 3.12")
result_news = process_source("news", "Python 3.12")

You define the logic once and call it 3 times with different arguments. A subgraph does exactly the same thing, but with LangGraph's advantages:

Creating a subgraph: define a complete StateGraph

A subgraph is a regular StateGraph — you define it, add nodes and edges, and compile it. The only difference is that afterwards you use it as a node inside another graph:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class SearchPipelineState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    parsed_data: str
    summary: str

def fetch(state: SearchPipelineState) -> dict:
    return {"raw_data": f"Raw data from {state['source_name']} for '{state['query']}'"}

def parse(state: SearchPipelineState) -> dict:
    return {"parsed_data": f"Parsed: {state['raw_data']}"}

def summarize(state: SearchPipelineState) -> dict:
    return {"summary": f"Summary of {state['source_name']}: {state['parsed_data'][:50]}..."}

search_pipeline = StateGraph(SearchPipelineState)
search_pipeline.add_node("fetch", fetch)
search_pipeline.add_node("parse", parse)
search_pipeline.add_node("summarize", summarize)

search_pipeline.add_edge(START, "fetch")
search_pipeline.add_edge("fetch", "parse")
search_pipeline.add_edge("parse", "summarize")
search_pipeline.add_edge("summarize", END)

search_app = search_pipeline.compile()

result = search_app.invoke({
    "source_name": "web",
    "query": "Python 3.12",
    "raw_data": "",
    "parsed_data": "",
    "summary": "",
})
print(result["summary"])
# Expected output:
# Summary of web: Parsed: Raw data from web for 'Python 3.12'...

This graph works on its own. It has its own state (SearchPipelineState), its own nodes, and it can be invoked directly. Now let's use it as a node of a parent graph.

Using a subgraph as a node

To use a compiled subgraph as a node, you pass it straight to add_node:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display

class SearchPipelineState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    parsed_data: str
    summary: str

def fetch(state: SearchPipelineState) -> dict:
    return {"raw_data": f"Raw data from {state['source_name']} for '{state['query']}'"}

def parse(state: SearchPipelineState) -> dict:
    return {"parsed_data": f"Parsed: {state['raw_data']}"}

def summarize(state: SearchPipelineState) -> dict:
    return {"summary": f"Summary of {state['source_name']}: {state['parsed_data'][:50]}..."}

search_pipeline = StateGraph(SearchPipelineState)
search_pipeline.add_node("fetch", fetch)
search_pipeline.add_node("parse", parse)
search_pipeline.add_node("summarize", summarize)
search_pipeline.add_edge(START, "fetch")
search_pipeline.add_edge("fetch", "parse")
search_pipeline.add_edge("parse", "summarize")
search_pipeline.add_edge("summarize", END)

search_subgraph = search_pipeline.compile()

class ParentState(TypedDict):
    query: str
    source_name: str
    raw_data: str
    parsed_data: str
    summary: str
    all_summaries: Annotated[list[str], operator.add]

def collect_summary(state: ParentState) -> dict:
    return {"all_summaries": [state["summary"]]}

parent = StateGraph(ParentState)
parent.add_node("search_pipeline", search_subgraph)
parent.add_node("collect", collect_summary)

parent.add_edge(START, "search_pipeline")
parent.add_edge("search_pipeline", "collect")
parent.add_edge("collect", END)

app = parent.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "query": "LangGraph subgraphs",
    "source_name": "web",
    "raw_data": "",
    "parsed_data": "",
    "summary": "",
    "all_summaries": [],
})
print(f"Summary: {result['summary']}")
print(f"All summaries: {result['all_summaries']}")
# Expected output:
# Summary: Summary of web: Parsed: Raw data from web for 'LangGraph subgrap...
# All summaries: ['Summary of web: Parsed: Raw data from web for \'LangGraph subgrap...']

parent.add_node("search_pipeline", search_subgraph) — the compiled graph is used directly as a node. LangGraph takes care of running all the subgraph's internal nodes when it's that "node's" turn in the parent graph.

The state interface between parent and subgraph

When the parent and the subgraph share the same fields in their TypedDict, LangGraph maps the state automatically. Fields with the same name are passed straight through:

class SearchPipelineState(TypedDict):
    source_name: str    # ← shared with the parent
    query: str          # ← shared with the parent
    raw_data: str       # internal to the subgraph
    parsed_data: str    # internal to the subgraph
    summary: str        # ← shared with the parent

class ParentState(TypedDict):
    query: str          # ← shared with the subgraph
    source_name: str    # ← shared with the subgraph
    summary: str        # ← shared with the subgraph
    all_summaries: Annotated[list[str], operator.add]  # parent only

Fields that exist in both (query, source_name, summary) are shared. Fields that only exist in the subgraph (raw_data, parsed_data) are internal — the parent never sees them. Fields that only exist in the parent (all_summaries) don't affect the subgraph.

The mapping rule

Different state: a subgraph with its own TypedDict

Sometimes the subgraph needs a state that's completely different from the parent's. In that case, you use a wrapper function that translates between states:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class AnalysisState(TypedDict):
    text: str
    word_count: int
    sentiment: str

def count_words(state: AnalysisState) -> dict:
    return {"word_count": len(state["text"].split())}

def detect_sentiment(state: AnalysisState) -> dict:
    text = state["text"].lower()
    if any(w in text for w in ["great", "excellent", "amazing", "good"]):
        return {"sentiment": "positive"}
    elif any(w in text for w in ["bad", "terrible", "awful", "poor"]):
        return {"sentiment": "negative"}
    return {"sentiment": "neutral"}

analysis_graph = StateGraph(AnalysisState)
analysis_graph.add_node("count", count_words)
analysis_graph.add_node("sentiment", detect_sentiment)
analysis_graph.add_edge(START, "count")
analysis_graph.add_edge("count", "sentiment")
analysis_graph.add_edge("sentiment", END)
analysis_subgraph = analysis_graph.compile()

class ParentState(TypedDict):
    query: str
    search_result: str
    analysis: dict
    summaries: Annotated[list[str], operator.add]

def search(state: ParentState) -> dict:
    return {"search_result": f"Great results about '{state['query']}': Python is an amazing language."}

def analyze_wrapper(state: ParentState) -> dict:
    analysis_input = {"text": state["search_result"], "word_count": 0, "sentiment": ""}
    result = analysis_subgraph.invoke(analysis_input)
    return {"analysis": {"word_count": result["word_count"], "sentiment": result["sentiment"]}}

def report(state: ParentState) -> dict:
    a = state["analysis"]
    return {"summaries": [f"Analysis: {a['word_count']} words, sentiment: {a['sentiment']}"]}

parent = StateGraph(ParentState)
parent.add_node("search", search)
parent.add_node("analyze", analyze_wrapper)
parent.add_node("report", report)

parent.add_edge(START, "search")
parent.add_edge("search", "analyze")
parent.add_edge("analyze", "report")
parent.add_edge("report", END)

app = parent.compile()
result = app.invoke({"query": "Python", "search_result": "", "analysis": {}, "summaries": []})
print(f"Result: {result['analysis']}")
print(f"Report: {result['summaries']}")
# Expected output:
# Result: {'word_count': 9, 'sentiment': 'positive'}
# Report: ["Analysis: 9 words, sentiment: positive"]

The analyze_wrapper function translates from the parent's state into the subgraph's state, invokes the subgraph, and translates the result back. It's the same idea as an adapter pattern — a bridge between two different interfaces.

When to use shared state vs different state

Reuse: one subgraph used multiple times

The real power of subgraphs is that you define them once and reuse them. Here we use the same search pipeline for 3 different sources:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from IPython.display import Image, display

class SearchState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    parsed_data: str
    summary: str

def fetch_data(state: SearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Simulate raw data from the '{state['source_name']}' source about '{state['query']}'. "
        f"Generate 2-3 sentences of unprocessed data."
    )
    return {"raw_data": response.content}

def parse_data(state: SearchState) -> dict:
    return {"parsed_data": f"[{state['source_name']}] {state['raw_data'][:200]}"}

def summarize_data(state: SearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Summarize in 1 sentence: {state['parsed_data']}")
    return {"summary": response.content}

search_pipeline = StateGraph(SearchState)
search_pipeline.add_node("fetch", fetch_data)
search_pipeline.add_node("parse", parse_data)
search_pipeline.add_node("summarize", summarize_data)
search_pipeline.add_edge(START, "fetch")
search_pipeline.add_edge("fetch", "parse")
search_pipeline.add_edge("parse", "summarize")
search_pipeline.add_edge("summarize", END)
search_subgraph = search_pipeline.compile()

class OrchestratorState(TypedDict):
    query: str
    source_name: str
    raw_data: str
    parsed_data: str
    summary: str
    all_summaries: Annotated[list[dict], operator.add]

def set_source(name: str):
    def node(state: OrchestratorState) -> dict:
        return {"source_name": name}
    return node

def collect_result(state: OrchestratorState) -> dict:
    return {"all_summaries": [{"source": state["source_name"], "summary": state["summary"]}]}

parent = StateGraph(OrchestratorState)

parent.add_node("set_web", set_source("web"))
parent.add_node("pipeline_web", search_subgraph)
parent.add_node("collect_web", collect_result)

parent.add_node("set_academic", set_source("academic"))
parent.add_node("pipeline_academic", search_subgraph)
parent.add_node("collect_academic", collect_result)

parent.add_node("set_news", set_source("news"))
parent.add_node("pipeline_news", search_subgraph)
parent.add_node("collect_news", collect_result)

parent.add_edge(START, "set_web")
parent.add_edge("set_web", "pipeline_web")
parent.add_edge("pipeline_web", "collect_web")

parent.add_edge(START, "set_academic")
parent.add_edge("set_academic", "pipeline_academic")
parent.add_edge("pipeline_academic", "collect_academic")

parent.add_edge(START, "set_news")
parent.add_edge("set_news", "pipeline_news")
parent.add_edge("pipeline_news", "collect_news")

parent.add_edge("collect_web", END)
parent.add_edge("collect_academic", END)
parent.add_edge("collect_news", END)

app = parent.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "query": "RAG techniques",
    "source_name": "", "raw_data": "", "parsed_data": "", "summary": "",
    "all_summaries": [],
})
print(f"Sources processed: {len(result['all_summaries'])}")
for s in result["all_summaries"]:
    print(f"  [{s['source']}] {s['summary'][:80]}...")
# Expected output:
# Sources processed: 3
#   [web] RAG combines document retrieval with text generation...
#   [academic] Studies show that RAG improves factual accuracy...
#   [news] Companies like Google and OpenAI are adopting RAG techniques...

What we gained

• ❌ Before: 9 different functions (3 × fetch, parse, summarize) with duplicated logic
• ✅ After: 3 functions + 1 subgraph reused 3 times
• If you change the parsing logic, you change it in a single place

The same search_subgraph is used as pipeline_web, pipeline_academic and pipeline_news. All 3 instances share the same logic but operate on different data thanks to set_source.

Nesting: subgraphs inside subgraphs

Subgraphs can contain other subgraphs. But keep the maximum depth at 2 levels (parent → child → grandchild). More levels make debugging hard:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display

class InnerState(TypedDict):
    text: str
    processed: str

def clean_text(state: InnerState) -> dict:
    return {"processed": state["text"].strip().lower()}

inner = StateGraph(InnerState)
inner.add_node("clean", clean_text)
inner.add_edge(START, "clean")
inner.add_edge("clean", END)
inner_compiled = inner.compile()

class MiddleState(TypedDict):
    text: str
    processed: str
    word_count: int

def count_words(state: MiddleState) -> dict:
    return {"word_count": len(state["processed"].split())}

middle = StateGraph(MiddleState)
middle.add_node("preprocess", inner_compiled)
middle.add_node("count", count_words)
middle.add_edge(START, "preprocess")
middle.add_edge("preprocess", "count")
middle.add_edge("count", END)
middle_compiled = middle.compile()

class OuterState(TypedDict):
    text: str
    processed: str
    word_count: int
    report: str

def make_report(state: OuterState) -> dict:
    return {"report": f"Text '{state['processed']}' has {state['word_count']} words"}

outer = StateGraph(OuterState)
outer.add_node("analysis", middle_compiled)
outer.add_node("report", make_report)
outer.add_edge(START, "analysis")
outer.add_edge("analysis", "report")
outer.add_edge("report", END)

app = outer.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({"text": "  Hello World  ", "processed": "", "word_count": 0, "report": ""})
print(result["report"])
# Expected output:
# Text 'hello world' has 2 words

Three levels: outer → middle → inner. The flow is: outer.analysis → middle.preprocess → inner.clean → middle.count → outer.report. Each level has its own state and its own checkpoints.

The 2-level rule

If you need more than 2 levels, your design probably needs refactoring: combine levels or extract logic into regular functions.

Subgraphs with the Functional API: @entrypoint as a reusable module

In the Functional API, a compiled @entrypoint can act as a reusable module — similar to how a subgraph is a graph inside another graph:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

@task
def fetch_data(source: str, query: str) -> str:
    response = model.invoke(f"Simulate data from {source} about '{query}'. 2-3 sentences.")
    return response.content

@task
def parse_data(source: str, raw: str) -> str:
    return f"[{source}] {raw[:200]}"

@task
def summarize_data(parsed: str) -> str:
    response = model.invoke(f"Summarize in 1 sentence: {parsed}")
    return response.content

@entrypoint()
def search_pipeline(config: dict) -> dict:
    source = config["source"]
    query = config["query"]

    raw = fetch_data(source, query).result()
    parsed = parse_data(source, raw).result()
    summary = summarize_data(parsed).result()

    return {"source": source, "summary": summary}

@entrypoint()
def research_orchestrator(query: str) -> dict:
    web_fut = search_pipeline.ainvoke({"source": "web", "query": query})
    academic_fut = search_pipeline.ainvoke({"source": "academic", "query": query})
    news_fut = search_pipeline.ainvoke({"source": "news", "query": query})

    results = []
    for label, fut in [("web", web_fut), ("academic", academic_fut), ("news", news_fut)]:
        try:
            r = fut.result()
            results.append(r)
        except Exception as e:
            results.append({"source": label, "summary": f"Error: {e}"})

    return {"results": results, "total": len(results)}

result = research_orchestrator.invoke("transformer architectures")
print(f"Total: {result['total']}")
for r in result["results"]:
    print(f"  [{r['source']}] {r['summary'][:80]}...")
# Expected output:
# Total: 3
#   [web] Transformer architectures have revolutionized natural language...
#   [academic] Research shows transformers outperform RNNs in sequence...
#   [news] Major tech companies are investing heavily in transformer-based...

search_pipeline is an @entrypoint that encapsulates the fetch-parse-summarize logic. research_orchestrator calls it 3 times with different sources. Each invocation has its own internal checkpoints.

Subgraph vs regular function: when to use each

The key question: why not use a plain Python function instead of a subgraph?

def search_and_summarize(source: str, query: str) -> str:
    raw = fetch(source, query)
    parsed = parse(raw)
    summary = summarize(parsed)
    return summary

This function does exactly the same thing as the subgraph. It's simpler, shorter and easier to understand. So when is a subgraph worth it?

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model

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

def search_as_function(source: str, query: str) -> str:
    raw = model.invoke(f"Simulate data from {source} about '{query}'").content
    parsed = f"[{source}] {raw[:200]}"
    summary = model.invoke(f"Summarize in 1 sentence: {parsed}").content
    return summary

class SubgraphState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    parsed_data: str
    summary: str

def sg_fetch(state: SubgraphState) -> dict:
    result = model.invoke(f"Simulate data from {state['source_name']} about '{state['query']}'")
    return {"raw_data": result.content}

def sg_parse(state: SubgraphState) -> dict:
    return {"parsed_data": f"[{state['source_name']}] {state['raw_data'][:200]}"}

def sg_summarize(state: SubgraphState) -> dict:
    result = model.invoke(f"Summarize in 1 sentence: {state['parsed_data']}")
    return {"summary": result.content}

search_graph = StateGraph(SubgraphState)
search_graph.add_node("fetch", sg_fetch)
search_graph.add_node("parse", sg_parse)
search_graph.add_node("summarize", sg_summarize)
search_graph.add_edge(START, "fetch")
search_graph.add_edge("fetch", "parse")
search_graph.add_edge("parse", "summarize")
search_graph.add_edge("summarize", END)
search_subgraph = search_graph.compile()

func_result = search_as_function("web", "Python 3.12")
print(f"Function: {func_result[:80]}...")

sg_result = search_subgraph.invoke({
    "source_name": "web", "query": "Python 3.12",
    "raw_data": "", "parsed_data": "", "summary": "",
})
print(f"Subgraph: {sg_result['summary'][:80]}...")
# Expected output:
# Function: Python 3.12 ships significant improvements to typing...
# Subgraph: Python 3.12 ships significant improvements to typing...

Both produce the same result. The difference shows up when something goes wrong or when you need observability:

Decision rule

• ✅ Use a regular function when the logic is simple (1-3 steps), you don't need checkpointing, and you don't need to visualize the internal flow
• ✅ Use a subgraph when the logic is complex (3+ steps), you need recovery from failures, you want visualization/streaming of the internal flow, or the pipeline gets reused across multiple graphs by different teams

Complete example: a modular Research Agent with subgraphs

Let's put it all together: reusable subgraphs, parallel branching (from the previous capsule) and smart merging:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from IPython.display import Image, display

class PipelineState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    summary: str

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

def pipeline_fetch(state: PipelineState) -> dict:
    response = model.invoke(
        f"Simulate a search on {state['source_name']} about '{state['query']}'. "
        f"Generate 2-3 informative sentences."
    )
    return {"raw_data": response.content}

def pipeline_summarize(state: PipelineState) -> dict:
    response = model.invoke(
        f"Summarize the following from [{state['source_name']}] in 1 sentence: {state['raw_data']}"
    )
    return {"summary": response.content}

pipeline = StateGraph(PipelineState)
pipeline.add_node("fetch", pipeline_fetch)
pipeline.add_node("summarize", pipeline_summarize)
pipeline.add_edge(START, "fetch")
pipeline.add_edge("fetch", "summarize")
pipeline.add_edge("summarize", END)
pipeline_compiled = pipeline.compile()

class ResearchState(TypedDict):
    query: str
    source_name: str
    raw_data: str
    summary: str
    all_results: Annotated[list[dict], operator.add]
    final_report: str

def set_source(name: str):
    def node(state: ResearchState) -> dict:
        return {"source_name": name, "raw_data": "", "summary": ""}
    return node

def collect(state: ResearchState) -> dict:
    return {"all_results": [{"source": state["source_name"], "summary": state["summary"]}]}

def final_merge(state: ResearchState) -> dict:
    context = "\n".join(f"- [{r['source']}]: {r['summary']}" for r in state["all_results"])
    response = model.invoke(
        f"Generate a 3-4 sentence executive summary based on these sources "
        f"about '{state['query']}':\n\n{context}"
    )
    return {"final_report": response.content}

research = StateGraph(ResearchState)

for source in ["web", "academic", "news"]:
    research.add_node(f"set_{source}", set_source(source))
    research.add_node(f"pipeline_{source}", pipeline_compiled)
    research.add_node(f"collect_{source}", collect)

    research.add_edge(START, f"set_{source}")
    research.add_edge(f"set_{source}", f"pipeline_{source}")
    research.add_edge(f"pipeline_{source}", f"collect_{source}")
    research.add_edge(f"collect_{source}", "merge")

research.add_node("merge", final_merge)
research.add_edge("merge", END)

app = research.compile()
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({
    "query": "How does RAG improve the accuracy of LLMs?",
    "source_name": "", "raw_data": "", "summary": "",
    "all_results": [], "final_report": "",
})

print(f"Sources: {len(result['all_results'])}")
for r in result["all_results"]:
    print(f"  [{r['source']}] {r['summary'][:80]}...")
print(f"\nFinal report: {result['final_report'][:200]}...")
# Expected output:
# Sources: 3
#   [web] RAG combines document search with text generation to...
#   [academic] Recent studies show that RAG reduces hallucinations...
#   [news] Companies like Google and Microsoft are integrating RAG into their...
# Final report: RAG (Retrieval-Augmented Generation) has proven to be a technique...

The graph architecture

START → set_web → pipeline_web (subgraph: fetch → summarize) → collect_web ─┐
START → set_academic → pipeline_academic (subgraph: fetch → summarize) → collect_academic ─┤→ merge → END
START → set_news → pipeline_news (subgraph: fetch → summarize) → collect_news ─┘

• Subgraph: defined once, used 3 times
• Branching: the 3 branches run in parallel
• Merge: receives the 3 results and synthesizes with an LLM

If tomorrow you need to add a fourth source ("social_media"), you add 3 lines to the for source in [...] loop and you're done.

Troubleshooting

Problem 1: "The subgraph isn't receiving the parent state's fields"

Symptom: The subgraph receives empty fields or default values.

Cause: The field names in the subgraph's TypedDict don't match the parent's.

Fix: Check that the shared fields have exactly the same name and type in both TypedDicts:

# ❌ Different names
class ParentState(TypedDict):
    search_query: str    # "search_query"

class SubgraphState(TypedDict):
    query: str           # "query" — doesn't match

# ✅ Same names
class ParentState(TypedDict):
    query: str           # "query"

class SubgraphState(TypedDict):
    query: str           # "query" — matches

Problem 2: "The subgraph overwrites parent fields it shouldn't touch"

Symptom: After running the subgraph, parent fields that aren't in the subgraph change or get lost.

Cause: The subgraph can only modify fields it shares with the parent. If a parent field doesn't exist in the subgraph, the subgraph doesn't affect it. But if a shared field has a reducer in the parent (operator.add) and the subgraph returns a value for that field, the reducer processes it.

Fix: Be careful with reducers on shared fields. If you don't want the subgraph to affect a field with a reducer, don't include that field in the subgraph's TypedDict.

Problem 3: "I can't reuse the same subgraph with different configurations"

Symptom: You want to use the same subgraph for "web" and "academic", but both instances receive the same data.

Cause: The subgraph reads from the parent's state, which is shared. If you don't change source_name before each instance, both read the same value.

Fix: Use a set_source node before each subgraph instance to configure the fields it needs:

def set_source(name: str):
    def node(state):
        return {"source_name": name}
    return node

graph.add_node("set_web", set_source("web"))
graph.add_node("pipeline_web", search_subgraph)
graph.add_edge("set_web", "pipeline_web")

Problem 4: "The diagram doesn't show the subgraph's internal nodes"

Symptom: draw_mermaid_png() shows the subgraph as an opaque box with no internal detail.

Cause: By default, the parent graph's visualization shows subgraphs as collapsed nodes.

Fix: Use the xray parameter to expand the subgraphs in the visualization:

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

With xray=True, you'll see the internal nodes of each subgraph inside their respective boxes.

Problem 5: "The subgraph fails and the error isn't caught in the parent"

Symptom: An error inside the subgraph propagates unhandled up to the parent level and stops the entire workflow.

Cause: Exceptions inside subgraph nodes propagate upward just like in any Python call stack.

Fix: If you use the subgraph as a direct node (add_node("pipeline", subgraph)), errors propagate. To catch them, use a wrapper:

def safe_pipeline(state):
    try:
        result = search_subgraph.invoke({"source_name": state["source_name"], "query": state["query"], ...})
        return {"summary": result["summary"]}
    except Exception as e:
        return {"summary": f"Pipeline error: {str(e)}"}

Exercises

Exercise 1: Basic subgraph (Easy)

Create a subgraph that processes text in 2 steps: (1) clean it (strip + lowercase), (2) count words. Use that subgraph as a node of a parent graph that adds a final reporting step. Visualize both graphs.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display

class TextState(TypedDict):
    text: str
    cleaned: str
    word_count: int

def clean(state: TextState) -> dict:
    return {"cleaned": state["text"].strip().lower()}

def count(state: TextState) -> dict:
    return {"word_count": len(state["cleaned"].split())}

text_processor = StateGraph(TextState)
text_processor.add_node("clean", clean)
text_processor.add_node("count", count)
text_processor.add_edge(START, "clean")
text_processor.add_edge("clean", "count")
text_processor.add_edge("count", END)
text_subgraph = text_processor.compile()

print("=== Subgraph ===")
display(Image(text_subgraph.get_graph().draw_mermaid_png()))

class ParentState(TypedDict):
    text: str
    cleaned: str
    word_count: int
    report: str

def make_report(state: ParentState) -> dict:
    return {"report": f"'{state['cleaned']}' has {state['word_count']} words"}

parent = StateGraph(ParentState)
parent.add_node("process", text_subgraph)
parent.add_node("report", make_report)
parent.add_edge(START, "process")
parent.add_edge("process", "report")
parent.add_edge("report", END)

app = parent.compile()
print("=== Parent graph ===")
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({"text": "  Hello World From LangGraph  ", "cleaned": "", "word_count": 0, "report": ""})
print(result["report"])
# Expected output:
# 'hello world from langgraph' has 4 words

Exercise 2: Subgraph with a different state (Easy)

Create a sentiment analysis subgraph with its own TypedDict (fields: text, sentiment, confidence). The parent graph has a different TypedDict (fields: query, search_result, analysis). Use a wrapper function to connect them.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class SentimentState(TypedDict):
    text: str
    sentiment: str
    confidence: float

def analyze_sentiment(state: SentimentState) -> dict:
    text = state["text"].lower()
    positive_words = ["good", "great", "excellent", "amazing", "love", "best"]
    negative_words = ["bad", "terrible", "awful", "hate", "worst", "poor"]
    pos = sum(1 for w in positive_words if w in text)
    neg = sum(1 for w in negative_words if w in text)
    total = pos + neg
    if total == 0:
        return {"sentiment": "neutral", "confidence": 0.5}
    if pos > neg:
        return {"sentiment": "positive", "confidence": round(pos / total, 2)}
    return {"sentiment": "negative", "confidence": round(neg / total, 2)}

sentiment_graph = StateGraph(SentimentState)
sentiment_graph.add_node("analyze", analyze_sentiment)
sentiment_graph.add_edge(START, "analyze")
sentiment_graph.add_edge("analyze", END)
sentiment_subgraph = sentiment_graph.compile()

class ParentState(TypedDict):
    query: str
    search_result: str
    analysis: dict

def simulate_search(state: ParentState) -> dict:
    return {"search_result": f"Great results! Python is an excellent and amazing language for AI."}

def analyze_wrapper(state: ParentState) -> dict:
    result = sentiment_subgraph.invoke({
        "text": state["search_result"],
        "sentiment": "",
        "confidence": 0.0,
    })
    return {"analysis": {"sentiment": result["sentiment"], "confidence": result["confidence"]}}

parent = StateGraph(ParentState)
parent.add_node("search", simulate_search)
parent.add_node("analyze", analyze_wrapper)
parent.add_edge(START, "search")
parent.add_edge("search", "analyze")
parent.add_edge("analyze", END)

app = parent.compile()
result = app.invoke({"query": "Python for AI", "search_result": "", "analysis": {}})
print(f"Search: {result['search_result'][:60]}...")
print(f"Analysis: {result['analysis']}")
# Expected output:
# Search: Great results! Python is an excellent and amazing language...
# Analysis: {'sentiment': 'positive', 'confidence': 1.0}

Exercise 3: A subgraph reused in parallel (Medium)

Use the same "fetch + summarize" subgraph in 3 parallel branches (web, academic, news). Each branch configures the source before running the subgraph. The results accumulate with operator.add. Visualize with xray=True.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display

class FetchState(TypedDict):
    source: str
    query: str
    raw: str
    summary: str

def fetch(state: FetchState) -> dict:
    return {"raw": f"Data from {state['source']} about '{state['query']}'"}

def summarize(state: FetchState) -> dict:
    return {"summary": f"[{state['source']}] Summary: {state['raw'][:50]}"}

fetch_graph = StateGraph(FetchState)
fetch_graph.add_node("fetch", fetch)
fetch_graph.add_node("summarize", summarize)
fetch_graph.add_edge(START, "fetch")
fetch_graph.add_edge("fetch", "summarize")
fetch_graph.add_edge("summarize", END)
fetch_subgraph = fetch_graph.compile()

class OrchestratorState(TypedDict):
    query: str
    source: str
    raw: str
    summary: str
    all_summaries: Annotated[list[str], operator.add]

def set_src(name: str):
    def node(state: OrchestratorState) -> dict:
        return {"source": name, "raw": "", "summary": ""}
    return node

def collect(state: OrchestratorState) -> dict:
    return {"all_summaries": [state["summary"]]}

graph = StateGraph(OrchestratorState)
graph.add_node("merge", lambda state: {})

for src in ["web", "academic", "news"]:
    graph.add_node(f"set_{src}", set_src(src))
    graph.add_node(f"pipe_{src}", fetch_subgraph)
    graph.add_node(f"collect_{src}", collect)
    graph.add_edge(START, f"set_{src}")
    graph.add_edge(f"set_{src}", f"pipe_{src}")
    graph.add_edge(f"pipe_{src}", f"collect_{src}")
    graph.add_edge(f"collect_{src}", "merge")

graph.add_edge("merge", END)

app = graph.compile()
display(Image(app.get_graph(xray=True).draw_mermaid_png()))

result = app.invoke({
    "query": "LangGraph subgraphs", "source": "", "raw": "", "summary": "",
    "all_summaries": [],
})
print(f"Summaries: {len(result['all_summaries'])}")
for s in result["all_summaries"]:
    print(f"  {s}")
# Expected output:
# Summaries: 3
#   [web] Summary: Data from web about 'LangGraph subgraphs'
#   [academic] Summary: Data from academic about 'LangGraph subgrap
#   [news] Summary: Data from news about 'LangGraph subgraphs'

Exercise 4: Functional API with a reusable module (Medium)

Implement the same pattern with the Functional API: an @entrypoint called process_source that does fetch + summarize, and an orchestrator that calls it 3 times in parallel with different sources. Handle individual errors.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

@task
def fetch_from_source(source: str, query: str) -> str:
    response = model.invoke(f"Simulate data from {source} about '{query}'. 2 sentences.")
    return response.content

@task
def summarize_result(source: str, raw: str) -> str:
    response = model.invoke(f"Summarize in 1 sentence: [{source}] {raw}")
    return response.content

@entrypoint()
def process_source(config: dict) -> dict:
    source = config["source"]
    query = config["query"]
    raw = fetch_from_source(source, query).result()
    summary = summarize_result(source, raw).result()
    return {"source": source, "summary": summary}

@task
def final_synthesis(results: list) -> str:
    context = "\n".join(f"- [{r['source']}]: {r['summary']}" for r in results)
    response = model.invoke(f"Synthesize in 2 sentences:\n{context}")
    return response.content

@entrypoint()
def research_agent(query: str) -> dict:
    sources = ["web", "academic", "news"]
    futures = [process_source.ainvoke({"source": s, "query": query}) for s in sources]

    results = []
    errors = []
    for source, fut in zip(sources, futures):
        try:
            results.append(fut.result())
        except Exception as e:
            errors.append({"source": source, "error": str(e)})

    synthesis = final_synthesis(results).result()

    return {
        "results": results,
        "errors": errors,
        "synthesis": synthesis,
    }

result = research_agent.invoke("fine-tuning techniques for LLMs")
print(f"Sources OK: {len(result['results'])}, Errors: {len(result['errors'])}")
for r in result["results"]:
    print(f"  [{r['source']}] {r['summary'][:80]}...")
print(f"Synthesis: {result['synthesis'][:150]}...")
# Expected output:
# Sources OK: 3, Errors: 0
#   [web] Fine-tuning techniques for LLMs include LoRA and full fine-tuning...
#   [academic] Recent research compares parameter-efficient fine-tuning methods...
#   [news] Companies are adopting fine-tuning to customize LLMs for specific...
# Synthesis: Fine-tuning LLMs has evolved with parameter-efficient techniques...

Exercise 5: Subgraph vs function — a practical comparison (Advanced)

Implement the same pipeline (fetch → parse → summarize) both as a regular function and as a subgraph. Invoke both with the same data and compare: do they produce the same result? What do you get extra with the subgraph? Show how to visualize the subgraph and how you can't visualize the function.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from IPython.display import Image, display

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

def pipeline_as_function(source: str, query: str) -> dict:
    raw = model.invoke(f"Simulate data from {source} about '{query}'. 1 sentence.").content
    parsed = f"[{source}] {raw}"
    summary = model.invoke(f"Summarize: {parsed}").content
    return {"source": source, "raw": raw, "parsed": parsed, "summary": summary}

class PipeState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    parsed_data: str
    summary: str

def sg_fetch(state: PipeState) -> dict:
    result = model.invoke(f"Simulate data from {state['source_name']} about '{state['query']}'. 1 sentence.")
    return {"raw_data": result.content}

def sg_parse(state: PipeState) -> dict:
    return {"parsed_data": f"[{state['source_name']}] {state['raw_data']}"}

def sg_summarize(state: PipeState) -> dict:
    result = model.invoke(f"Summarize: {state['parsed_data']}")
    return {"summary": result.content}

pipe = StateGraph(PipeState)
pipe.add_node("fetch", sg_fetch)
pipe.add_node("parse", sg_parse)
pipe.add_node("summarize", sg_summarize)
pipe.add_edge(START, "fetch")
pipe.add_edge("fetch", "parse")
pipe.add_edge("parse", "summarize")
pipe.add_edge("summarize", END)
pipeline_subgraph = pipe.compile()

print("=== Subgraph (visualizable) ===")
display(Image(pipeline_subgraph.get_graph().draw_mermaid_png()))

print("\n=== Function (not visualizable — it's a black box) ===")
print("No diagram available for regular Python functions.\n")

func_result = pipeline_as_function("web", "Python async")
print(f"Function → {func_result['summary'][:80]}...")

sg_result = pipeline_subgraph.invoke({
    "source_name": "web", "query": "Python async",
    "raw_data": "", "parsed_data": "", "summary": "",
})
print(f"Subgraph → {sg_result['summary'][:80]}...")

print("\n=== Differences ===")
print("✅ Subgraph: checkpointing per step, visualization, streaming, resume on failure")
print("✅ Function: simpler, less code, enough for trivial logic")
print("❌ Function: no checkpointing, no visualization, if it fails it re-runs everything")
# Expected output:
# === Subgraph (visualizable) ===
# [Diagram: fetch → parse → summarize]
# === Function (not visualizable — it's a black box) ===
# No diagram available for regular Python functions.
# Function → Python's async capabilities enable concurrent execution...
# Subgraph → Python's async capabilities enable concurrent execution...
# === Differences ===
# ✅ Subgraph: checkpointing per step, visualization, streaming, resume on failure
# ✅ Function: simpler, less code, enough for trivial logic
# ❌ Function: no checkpointing, no visualization, if it fails it re-runs everything

Exercise 6: Complete modular Research Agent (Advanced)

Build a Research Agent that combines subgraphs + branching + LLM merge. Requirements: (1) a reusable search subgraph (fetch → summarize), (2) a parent graph that runs the subgraph in parallel for 3 sources, (3) a merge node that uses an LLM to synthesize, (4) error handling in each branch. Visualize with xray=True.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from IPython.display import Image, display

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

class SearchState(TypedDict):
    source_name: str
    query: str
    raw_data: str
    summary: str

def sg_fetch(state: SearchState) -> dict:
    response = model.invoke(
        f"Simulate data from '{state['source_name']}' about '{state['query']}'. 2-3 sentences."
    )
    return {"raw_data": response.content}

def sg_summarize(state: SearchState) -> dict:
    response = model.invoke(f"Summarize in 1 sentence: [{state['source_name']}] {state['raw_data']}")
    return {"summary": response.content}

search_graph = StateGraph(SearchState)
search_graph.add_node("fetch", sg_fetch)
search_graph.add_node("summarize", sg_summarize)
search_graph.add_edge(START, "fetch")
search_graph.add_edge("fetch", "summarize")
search_graph.add_edge("summarize", END)
search_compiled = search_graph.compile()

class AgentState(TypedDict):
    query: str
    source_name: str
    raw_data: str
    summary: str
    collected: Annotated[list[dict], operator.add]
    report: str

def set_src(name: str):
    def node(state: AgentState) -> dict:
        return {"source_name": name, "raw_data": "", "summary": ""}
    return node

def safe_collect(state: AgentState) -> dict:
    if state.get("summary"):
        return {"collected": [{"source": state["source_name"], "summary": state["summary"], "status": "ok"}]}
    return {"collected": [{"source": state["source_name"], "summary": "No data", "status": "error"}]}

def merge_report(state: AgentState) -> dict:
    ok = [r for r in state["collected"] if r["status"] == "ok"]
    if not ok:
        return {"report": "Could not retrieve information from any source."}
    context = "\n".join(f"- [{r['source']}]: {r['summary']}" for r in ok)
    response = model.invoke(
        f"Generate a 3-4 sentence report about '{state['query']}' "
        f"based on these sources:\n\n{context}"
    )
    return {"report": response.content}

agent = StateGraph(AgentState)
agent.add_node("merge", merge_report)

for src in ["web", "academic", "news"]:
    agent.add_node(f"set_{src}", set_src(src))
    agent.add_node(f"search_{src}", search_compiled)
    agent.add_node(f"collect_{src}", safe_collect)
    agent.add_edge(START, f"set_{src}")
    agent.add_edge(f"set_{src}", f"search_{src}")
    agent.add_edge(f"search_{src}", f"collect_{src}")
    agent.add_edge(f"collect_{src}", "merge")

agent.add_edge("merge", END)

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

result = app.invoke({
    "query": "How does fine-tuning of LLMs work?",
    "source_name": "", "raw_data": "", "summary": "",
    "collected": [], "report": "",
})

print(f"Sources: {len(result['collected'])}")
for r in result["collected"]:
    print(f"  [{r['source']}] ({r['status']}) {r['summary'][:70]}...")
print(f"\nReport: {result['report'][:200]}...")
# Expected output:
# Sources: 3
#   [web] (ok) Fine-tuning adapts pre-trained LLMs to specific tasks...
#   [academic] (ok) Research shows parameter-efficient methods like LoRA...
#   [news] (ok) Companies are increasingly using fine-tuning for custom...
# Report: Fine-tuning LLMs is a process that adapts pre-trained models...

Summary

In this capsule you learned:

• Subgraphs = functions: just as a function can call another function, a graph can contain another graph. You encapsulate complex logic and reuse it
• Creating a subgraph means creating a regular StateGraph, compiling it, and using it as a node in a parent graph with add_node("name", compiled_subgraph)
• Shared state: fields with the same name in parent and subgraph are mapped automatically. Fields exclusive to each one are invisible to the other
• Different state: when parent and subgraph have different TypedDicts, use a wrapper function that translates between the two interfaces
• Reuse: you define the subgraph once and use it N times with different configurations. If you change the logic, you change it in a single place
• Nesting: subgraphs inside subgraphs is possible, but keep it to a maximum depth of 2 levels for your own sanity
• Functional API: a compiled @entrypoint can act as a reusable module, with the same encapsulation philosophy
• Subgraph vs regular function: the subgraph gives you per-step checkpointing, visualization and streaming. The function is simpler but it's a black box. Use a subgraph when the logic is complex and you need observability
• xray=True in draw_mermaid_png() expands the subgraphs to show their internal nodes

Next capsule: Map-Reduce — how to process collections of data in parallel, sending each element to its own instance of a subgraph.

Additional resources

1. LangGraph Subgraphs — How-to Guide — Official subgraph tutorial with complete examples
2. LangGraph Concepts: Subgraphs — Conceptual documentation on subgraphs
3. State Interface Between Graphs — How to handle state between parent and subgraph
4. Visualizing Subgraphs — Using xray to see internal nodes
5. LangGraph Branching — Combining branching with subgraphs
6. Functional API — @entrypoint as a reusable module
7. LangGraph Persistence — Checkpointing in subgraphs (each internal step has its own checkpoint)

Module 7 — LangChain & LangGraph: From Chains to Agents