Map-Reduce and Deferred Nodes

Capsule overview

You have a system that searches multiple sources. In the branching capsule, you defined 3 fixed branches — one per source. It worked perfectly. But now a new problem shows up: the user enters a query, and the number of results varies. Sometimes it's 2, sometimes 7, sometimes 15. You can't hardcode branches for every possible number of results.

This is the problem map-reduce solves: creating branches dynamically at runtime, based on the data — not on the graph's structure. LangGraph implements it with the Send API, which lets you "send" data to a node N times, where N is determined during execution. Afterwards, a deferred node waits for all the branches to finish before continuing.

If branching is "I have 3 fixed sources and I search them in parallel," map-reduce is "I don't know how many items I have until I see them, but I want to process each one in parallel and then combine the results."

The problem: collections of dynamic size

Picture this concrete scenario: your Research Assistant searches for papers and gets back a list of results. You need to summarize each paper. But you don't know how many papers it's going to find:

# Search 1: "transformer architecture" → 5 papers
# Search 2: "quantum error correction" → 2 papers
# Search 3: "CRISPR gene editing" → 8 papers

With static branching, you'd need to define a branch for every possible paper — impossible. What you need is:

1. Dynamic fan-out: create a branch for each item in the list, no matter how many there are
2. Independent processing: each item processed by the same node, in parallel
3. Automatic fan-in: all the results collected into the state when they finish

This is the map-reduce pattern.

Map-reduce: the pattern

The map-reduce pattern has three phases:

In functional programming, it's map() followed by reduce(). In LangGraph, it's the Send API followed by a deferred node with a reducer in the state.

The Send API: dynamic fan-out

The key is Send from langgraph.types. Instead of a conditional edge returning a string (the name of the next node), it returns a list of Send objects. Each Send creates an independent branch toward the same node, but with different data:

from langgraph.types import Send

def route_to_processors(state):
    """Creates a dynamic branch for each result."""
    return [
        Send("process_result", {"item": item, "index": i})
        for i, item in enumerate(state["raw_results"])
    ]

Each Send("process_result", data) does two things:

1. Creates an invocation of the "process_result" node
2. Passes data as the input state for that invocation

If state["raw_results"] has 3 items, 3 parallel invocations are created. If it has 10, 10 are created. The number is determined at runtime.

Send vs static branching

Complete example: processing N search results

Let's build a graph that searches for papers, and then summarizes each paper dynamically:

from dotenv import load_dotenv
load_dotenv()

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

class MainState(TypedDict):
    query: str
    raw_results: list[str]
    summaries: Annotated[list[str], operator.add]

class ProcessorState(TypedDict):
    item: str
    index: int

def search_papers(state: MainState) -> dict:
    """Simulates a search that returns N results."""
    query = state["query"]
    papers = [
        f"Paper 1: '{query}' - Proposes a new framework based on attention mechanisms.",
        f"Paper 2: '{query}' - Analyzes the limitations of current approaches.",
        f"Paper 3: '{query}' - Presents experimental results with a 15% improvement.",
        f"Paper 4: '{query}' - State-of-the-art review with 200+ references.",
    ]
    return {"raw_results": papers}

def route_to_processors(state: MainState) -> list[Send]:
    """Creates one Send per paper found."""
    return [
        Send("summarize_paper", {"item": paper, "index": i})
        for i, paper in enumerate(state["raw_results"])
    ]

def summarize_paper(state: ProcessorState) -> dict:
    """Processes a single paper — runs N times in parallel."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Summarize this paper in one technical sentence:\n\n{state['item']}"
    )
    return {"summaries": [f"[{state['index']}] {response.content}"]}

def compile_results(state: MainState) -> dict:
    """Combines all the summaries into a final result."""
    model = init_chat_model("openai:gpt-4.1-mini")
    context = "\n".join(state["summaries"])
    response = model.invoke(
        f"Generate an executive summary of these papers about '{state['query']}':\n\n{context}"
    )
    return {"summaries": [f"\n--- EXECUTIVE SUMMARY ---\n{response.content}"]}

graph_builder = StateGraph(MainState)
graph_builder.add_node("search", search_papers)
graph_builder.add_node("summarize_paper", summarize_paper)
graph_builder.add_node("compile", compile_results)

graph_builder.add_edge(START, "search")
graph_builder.add_conditional_edges("search", route_to_processors)
graph_builder.add_edge("summarize_paper", "compile")
graph_builder.add_edge("compile", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({"query": "transformer architecture", "raw_results": [], "summaries": []})
print(f"Papers found: {len(result['raw_results'])}")
print(f"Summaries generated: {len(result['summaries']) - 1}")
for s in result["summaries"]:
    print(s)
# Expected output:
# Papers found: 4
# Summaries generated: 4
# [0] This paper proposes a new framework based on attention mechanisms...
# [1] The study analyzes the limitations of current approaches in...
# [2] Experimental results are presented showing a 15% improvement...
# [3] A comprehensive state-of-the-art review with over 200 references...
#
# --- EXECUTIVE SUMMARY ---
# Research on transformer architecture shows advances in...

Anatomy of the example

1. search_papers returns a variable-length list in raw_results
2. route_to_processors reads that list and creates one Send per item — the dynamic fan-out
3. summarize_paper receives a ProcessorState (not the full MainState) — only the data it needs
4. The operator.add reducer on summaries accumulates the results from every branch
5. compile is the deferred node — it waits for every summarize_paper to finish before running

The processor state: separating data per branch

Notice that summarize_paper uses ProcessorState, not MainState:

class MainState(TypedDict):
    query: str
    raw_results: list[str]
    summaries: Annotated[list[str], operator.add]

class ProcessorState(TypedDict):
    item: str
    index: int

Each Send("summarize_paper", {"item": paper, "index": i}) creates an invocation with only the data that branch needs. The summarize_paper node never sees the query or the other papers — just its item and its index. That's intentional:

• ✅ Isolation: each branch is independent, it can't interfere with the others
• ✅ Efficiency: it doesn't copy the full state N times
• ✅ Clarity: the ProcessorState type documents what data each branch receives

The return {"summaries": [...]} is written back into MainState because summaries has the operator.add reducer. Each branch appends its result to the shared list.

Deferred nodes: the automatic fan-in

The compile node in the example is a deferred node — a node that waits for all the upstream branches to finish before it runs. You don't need to configure anything special: LangGraph infers it from the graph's structure.

When you define graph_builder.add_edge("summarize_paper", "compile"), LangGraph knows that:

1. summarize_paper can run N times (because of the Sends)
2. compile comes after summarize_paper
3. Therefore, compile must wait for the N instances of summarize_paper to finish

The deferred node has access to the full state with all the results accumulated by the reducer. When compile runs, state["summaries"] holds the summaries of every paper — whether there were 2 or 20.

How LangGraph knows "everyone finished"

LangGraph keeps an internal counter of active branches. Each Send increments the counter. Each summarize_paper completion decrements it. When it reaches 0, the deferred node fires. You never touch this counter — it's automatic.

Reducers: the key piece of the fan-in

Without a reducer, the fan-in doesn't work. If summaries were a plain list[str] without Annotated[..., operator.add], each branch would overwrite the previous one's result:

# ❌ Without a reducer: only the last branch survives
class BadState(TypedDict):
    summaries: list[str]

# ✅ With a reducer: every branch accumulates
class GoodState(TypedDict):
    summaries: Annotated[list[str], operator.add]

The operator.add reducer concatenates lists. If branch 1 returns ["summary A"] and branch 2 returns ["summary B"], the state ends up with ["summary A", "summary B"].

Other reducers that come in handy for map-reduce:

import operator

# Accumulate lists
results: Annotated[list[str], operator.add]

# Count successes
success_count: Annotated[int, operator.add]

# Custom reducer: keep only the best results
def keep_top_scores(current: list[dict], new: list[dict]) -> list[dict]:
    combined = current + new
    return sorted(combined, key=lambda x: x["score"], reverse=True)[:5]

top_results: Annotated[list[dict], keep_top_scores]

When to use map-reduce vs static branching

Rule: if you know the branches when you write the code, use static branching. If the branches depend on the data, use Send.

Consensus patterns: combining results from parallel branches

When N branches return results, the deferred node needs a strategy to combine them. These are the most common patterns:

Pattern 1: Simple concatenation

All the results get joined into one list. Useful when each result is independent and valuable:

def compile_all(state: MainState) -> dict:
    return {"final_output": "\n".join(state["summaries"])}

Pattern 2: Majority vote

Multiple branches analyze the same input and vote. Useful for classification or verification:

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from collections import Counter
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langchain.chat_models import init_chat_model

class VoteState(TypedDict):
    text: str
    votes: Annotated[list[str], operator.add]

class VoterInput(TypedDict):
    text: str
    voter_id: int

def create_voters(state: VoteState) -> list[Send]:
    return [Send("vote", {"text": state["text"], "voter_id": i}) for i in range(3)]

def vote(state: VoterInput) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Classify this text as 'positive', 'negative', or 'neutral'. "
        f"Reply with ONLY the classification.\n\n{state['text']}"
    )
    return {"votes": [response.content.strip().lower()]}

def tally_votes(state: VoteState) -> dict:
    counts = Counter(state["votes"])
    winner = counts.most_common(1)[0][0]
    return {"votes": [f"RESULT: {winner} ({dict(counts)})"]}

graph_builder = StateGraph(VoteState)
graph_builder.add_node("vote", vote)
graph_builder.add_node("tally", tally_votes)

graph_builder.add_conditional_edges(START, create_voters)
graph_builder.add_edge("vote", "tally")
graph_builder.add_edge("tally", END)

graph = graph_builder.compile()

result = graph.invoke({
    "text": "The product is good but the after-sales service is terrible.",
    "votes": [],
})
print(f"Votes: {result['votes']}")
# Expected output:
# Votes: ['negative', 'negative', 'neutral', 'RESULT: negative ({negative: 2, neutral: 1})']

Pattern 3: Weighted average

Each branch returns a score. The deferred node computes the average:

import operator
from typing import TypedDict, Annotated

class ScoreState(TypedDict):
    scores: Annotated[list[float], operator.add]

def aggregate_scores(state: ScoreState) -> dict:
    avg = sum(state["scores"]) / len(state["scores"]) if state["scores"] else 0
    return {"scores": [avg]}

Pattern 4: Quality filtering

Each branch returns a result with a confidence score. The deferred node filters out the low-quality ones:

import operator
from typing import TypedDict, Annotated

class FilterState(TypedDict):
    results: Annotated[list[dict], operator.add]

def filter_high_quality(state: FilterState) -> dict:
    good = [r for r in state["results"] if r.get("confidence", 0) >= 0.7]
    return {"results": good if good else state["results"][:1]}

Map-reduce with the Functional API

The same pattern is possible with the Functional API using Futures for parallelism:

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 search(query: str) -> list:
    return [
        f"Paper 1 about {query}: a new framework is proposed.",
        f"Paper 2 about {query}: analysis of limitations.",
        f"Paper 3 about {query}: 15% experimental improvement.",
    ]

@task
def summarize_one(paper: str) -> str:
    return model.invoke(f"Summarize in one sentence: {paper}").content

@task
def compile_summaries(query: str, summaries: list) -> str:
    context = "\n".join(f"- {s}" for s in summaries)
    return model.invoke(
        f"Executive summary about '{query}':\n\n{context}"
    ).content

@entrypoint()
def map_reduce_functional(query: str) -> dict:
    papers = search(query).result()

    summary_futures = [summarize_one(p) for p in papers]
    summaries = [f.result() for f in summary_futures]

    executive = compile_summaries(query, summaries).result()

    return {
        "papers_found": len(papers),
        "summaries": summaries,
        "executive_summary": executive,
    }

result = map_reduce_functional.invoke("retrieval augmented generation")
print(f"Papers: {result['papers_found']}")
for s in result["summaries"]:
    print(f"  - {s[:80]}...")
print(f"\nExecutive summary: {result['executive_summary'][:150]}...")
# Expected output:
# Papers: 3
#   - This paper proposes a new framework for RAG based on...
#   - The study analyzes the main limitations of current approaches...
#   - Experimental results show a 15% improvement in...
# Executive summary: Research on RAG shows significant advances...

The key difference: with the Functional API you don't have Send — you use list comprehensions and Futures. This works fine when the processing is simple. If you need subgraphs, typed state, or visualization, the Graph API with Send is more powerful.

Advanced example: map-reduce with conditional processing

Not every item needs the same processing. You can use Send to route items to different nodes based on their characteristics:

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langchain.chat_models import init_chat_model

class MainState(TypedDict):
    items: list[dict]
    results: Annotated[list[str], operator.add]

class ItemState(TypedDict):
    item: dict

def classify_and_route(state: MainState) -> list[Send]:
    sends = []
    for item in state["items"]:
        if item.get("type") == "short":
            sends.append(Send("process_short", {"item": item}))
        else:
            sends.append(Send("process_long", {"item": item}))
    return sends

def process_short(state: ItemState) -> dict:
    text = state["item"]["text"]
    return {"results": [f"[SHORT] Fast-processed: {text[:50]}"]}

def process_long(state: ItemState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    text = state["item"]["text"]
    response = model.invoke(f"Summarize this long text in one sentence:\n\n{text}")
    return {"results": [f"[LONG] {response.content}"]}

def merge_results(state: MainState) -> dict:
    return {}

graph_builder = StateGraph(MainState)
graph_builder.add_node("process_short", process_short)
graph_builder.add_node("process_long", process_long)
graph_builder.add_node("merge", merge_results)

graph_builder.add_conditional_edges(START, classify_and_route)
graph_builder.add_edge("process_short", "merge")
graph_builder.add_edge("process_long", "merge")
graph_builder.add_edge("merge", END)

graph = graph_builder.compile()

result = graph.invoke({
    "items": [
        {"type": "short", "text": "Python is a programming language."},
        {"type": "long", "text": "Transformers are a deep learning architecture that uses attention mechanisms to process sequences of data in parallel, removing the need for recurrent processing."},
        {"type": "short", "text": "LangGraph handles graphs."},
    ],
    "results": [],
})
for r in result["results"]:
    print(r)
# Expected output:
# [SHORT] Fast-processed: Python is a programming language.
# [LONG] Transformers are an attention-based architecture that...
# [SHORT] Fast-processed: LangGraph handles graphs.

The Sends can point to different nodes — not every item has to go to the same processor. The routing function decides which node processes each item.

Troubleshooting

Problem 1: "The map results are empty"

Symptom: The deferred node receives an empty list in the accumulated field.

Cause: The processing node isn't returning data in the right field, or the field has no reducer.

Fix: Check two things:

# 1. The state field has a reducer
class State(TypedDict):
    results: Annotated[list[str], operator.add]  # ✅ With a reducer

# 2. The processing node returns into the right field
def process(state: ProcessorState) -> dict:
    return {"results": ["my result"]}  # ✅ Key matches

Problem 2: "Send returns an error: node not found"

Symptom: ValueError: Node 'process_result' not found in graph.

Cause: The node name in Send("process_result", data) doesn't match the name registered with add_node.

Fix: Check that the string in Send matches exactly:

graph_builder.add_node("process_result", my_function)  # Registration
Send("process_result", data)                            # Usage — must match

Problem 3: "I only get the result from the last branch"

Symptom: The deferred node only has one result instead of N.

Cause: The state field doesn't have operator.add as a reducer — each branch overwrites the previous one.

Fix:

# ❌ Without a reducer — the last branch wins
results: list[str]

# ✅ With a reducer — every branch accumulates
results: Annotated[list[str], operator.add]

Problem 4: "The order of the results is unpredictable"

Symptom: The map-reduce results come back in a different order every time.

Cause: The branches run in parallel — the order they finish in isn't deterministic.

Fix: Include an index in the Send data and sort in the deferred node:

def route(state):
    return [
        Send("process", {"item": item, "index": i})
        for i, item in enumerate(state["items"])
    ]

def merge(state):
    sorted_results = sorted(state["results"], key=lambda r: r["index"])
    return {"results": sorted_results}

Problem 5: "One branch fails and the deferred node never runs"

Symptom: If one of the N branches raises an exception, the whole graph fails.

Cause: By default, an exception in any branch cancels the execution.

Fix: Handle errors inside the processing node:

def process(state: ProcessorState) -> dict:
    try:
        result = expensive_operation(state["item"])
        return {"results": [{"status": "ok", "data": result}]}
    except Exception as e:
        return {"results": [{"status": "error", "error": str(e)}]}

Exercises

Exercise 1: Basic map-reduce with Send (Easy)

Create a graph that takes a list of cities in the state, uses Send to create a branch per city that fetches its "weather" (simulated), and a deferred node that combines all the weather into a report.

from dotenv import load_dotenv
load_dotenv()

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

class MainState(TypedDict):
    cities: list[str]
    weather_reports: Annotated[list[str], operator.add]

class CityState(TypedDict):
    city: str

MOCK_WEATHER = {
    "madrid": "Sunny, 28°C",
    "london": "Cloudy, 14°C",
    "tokyo": "Light rain, 22°C",
    "new york": "Partly cloudy, 25°C",
}

def route_to_cities(state: MainState) -> list[Send]:
    return [Send("get_weather", {"city": city}) for city in state["cities"]]

def get_weather(state: CityState) -> dict:
    city = state["city"]
    weather = MOCK_WEATHER.get(city.lower(), "Data not available")
    return {"weather_reports": [f"{city}: {weather}"]}

def compile_report(state: MainState) -> dict:
    return {}

graph_builder = StateGraph(MainState)
graph_builder.add_node("get_weather", get_weather)
graph_builder.add_node("compile", compile_report)

graph_builder.add_conditional_edges(START, route_to_cities)
graph_builder.add_edge("get_weather", "compile")
graph_builder.add_edge("compile", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({
    "cities": ["Madrid", "London", "Tokyo"],
    "weather_reports": [],
})
print("Weather report:")
for report in result["weather_reports"]:
    print(f"  {report}")
# Expected output:
# Weather report:
#   Madrid: Sunny, 28°C
#   London: Cloudy, 14°C
#   Tokyo: Light rain, 22°C

Exercise 2: Map-reduce with LLM processing (Easy)

Adapt the previous exercise so that, instead of fetching weather, each branch uses an LLM to generate a fun fact about the city. The deferred node should compile all the facts into a paragraph using the LLM.

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langchain.chat_models import init_chat_model

class MainState(TypedDict):
    cities: list[str]
    fun_facts: Annotated[list[str], operator.add]
    compiled: str

class CityState(TypedDict):
    city: str

def route_cities(state: MainState) -> list[Send]:
    return [Send("get_fact", {"city": c}) for c in state["cities"]]

def get_fact(state: CityState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Generate ONE fun fact about {state['city']} in a single sentence."
    )
    return {"fun_facts": [f"{state['city']}: {response.content}"]}

def compile_facts(state: MainState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    context = "\n".join(state["fun_facts"])
    response = model.invoke(
        f"Combine these fun facts into one entertaining paragraph:\n\n{context}"
    )
    return {"compiled": response.content}

graph_builder = StateGraph(MainState)
graph_builder.add_node("get_fact", get_fact)
graph_builder.add_node("compile", compile_facts)

graph_builder.add_conditional_edges(START, route_cities)
graph_builder.add_edge("get_fact", "compile")
graph_builder.add_edge("compile", END)

graph = graph_builder.compile()

result = graph.invoke({
    "cities": ["Paris", "Tokyo", "Buenos Aires", "Cairo"],
    "fun_facts": [],
    "compiled": "",
})
print(f"Cities processed: {len(result['fun_facts'])}")
for fact in result["fun_facts"]:
    print(f"  {fact[:80]}...")
print(f"\nCompiled paragraph: {result['compiled'][:200]}...")
# Expected output:
# Cities processed: 4
#   Paris: The Eiffel Tower was built as a temporary structure for...
#   Tokyo: Tokyo has more Michelin-starred restaurants than...
#   Buenos Aires: Buenos Aires has the widest avenue in the world...
#   Cairo: The Giza pyramids are the only wonders of the...
# Compiled paragraph: Around the world there are fascinating facts...

Exercise 3: Voting with 5 judges (Medium)

Implement a sentiment classification system where 5 LLM "judges" classify a text independently. Use Send to create the 5 judges. The deferred node counts the votes and decides by majority. Show the vote breakdown.

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from collections import Counter
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langchain.chat_models import init_chat_model

class MainState(TypedDict):
    text: str
    num_judges: int
    votes: Annotated[list[dict], operator.add]
    verdict: str

class JudgeState(TypedDict):
    text: str
    judge_id: int

def create_judges(state: MainState) -> list[Send]:
    return [
        Send("judge", {"text": state["text"], "judge_id": i})
        for i in range(state["num_judges"])
    ]

def judge(state: JudgeState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"You are judge #{state['judge_id']}. Classify the sentiment of this text "
        f"as EXACTLY one of these options: positive, negative, neutral.\n"
        f"Reply with ONLY the classification, nothing else.\n\n"
        f"Text: {state['text']}"
    )
    classification = response.content.strip().lower()
    valid = {"positive", "negative", "neutral"}
    if classification not in valid:
        classification = "neutral"
    return {"votes": [{"judge": state["judge_id"], "vote": classification}]}

def tally(state: MainState) -> dict:
    vote_values = [v["vote"] for v in state["votes"]]
    counts = Counter(vote_values)
    winner, count = counts.most_common(1)[0]
    total = len(vote_values)
    breakdown = ", ".join(f"{k}: {v}/{total}" for k, v in counts.most_common())
    return {"verdict": f"{winner} (consensus: {breakdown})"}

graph_builder = StateGraph(MainState)
graph_builder.add_node("judge", judge)
graph_builder.add_node("tally", tally)

graph_builder.add_conditional_edges(START, create_judges)
graph_builder.add_edge("judge", "tally")
graph_builder.add_edge("tally", END)

graph = graph_builder.compile()

result = graph.invoke({
    "text": "The product arrived fast and works well, although the packaging was damaged.",
    "num_judges": 5,
    "votes": [],
    "verdict": "",
})
print(f"Individual votes:")
for v in result["votes"]:
    print(f"  Judge #{v['judge']}: {v['vote']}")
print(f"\nVerdict: {result['verdict']}")
# Expected output:
# Individual votes:
#   Judge #0: positive
#   Judge #1: positive
#   Judge #2: neutral
#   Judge #3: positive
#   Judge #4: positive
# Verdict: positive (consensus: positive: 4/5, neutral: 1/5)

Exercise 4: Map-reduce with error handling (Medium)

Create a map-reduce graph where some items fail to process (simulate errors for specific items). The deferred node should separate successes from errors, and generate a report that includes both. The system must not crash because of a failed item.

from dotenv import load_dotenv
load_dotenv()

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

class MainState(TypedDict):
    urls: list[str]
    results: Annotated[list[dict], operator.add]

class FetchState(TypedDict):
    url: str
    index: int

MOCK_RESPONSES = {
    "https://api.example.com/data1": "Data from endpoint 1 fetched successfully.",
    "https://api.example.com/data2": None,
    "https://api.example.com/data3": "Data from endpoint 3 fetched successfully.",
    "https://api.example.com/data4": None,
    "https://api.example.com/data5": "Data from endpoint 5 fetched successfully.",
}

def route_to_fetchers(state: MainState) -> list[Send]:
    return [
        Send("fetch_url", {"url": url, "index": i})
        for i, url in enumerate(state["urls"])
    ]

def fetch_url(state: FetchState) -> dict:
    url = state["url"]
    response = MOCK_RESPONSES.get(url)
    if response is None:
        return {"results": [{
            "index": state["index"],
            "url": url,
            "status": "error",
            "data": None,
            "error": f"Connection timeout for {url}",
        }]}
    return {"results": [{
        "index": state["index"],
        "url": url,
        "status": "ok",
        "data": response,
        "error": None,
    }]}

def compile_report(state: MainState) -> dict:
    sorted_results = sorted(state["results"], key=lambda r: r["index"])
    successes = [r for r in sorted_results if r["status"] == "ok"]
    failures = [r for r in sorted_results if r["status"] == "error"]
    print(f"Successes: {len(successes)}/{len(sorted_results)}")
    print(f"Errors: {len(failures)}/{len(sorted_results)}")
    for f in failures:
        print(f"  ⚠️ {f['url']}: {f['error']}")
    for s in successes:
        print(f"  ✅ {s['url']}: {s['data'][:50]}")
    return {}

graph_builder = StateGraph(MainState)
graph_builder.add_node("fetch_url", fetch_url)
graph_builder.add_node("compile", compile_report)

graph_builder.add_conditional_edges(START, route_to_fetchers)
graph_builder.add_edge("fetch_url", "compile")
graph_builder.add_edge("compile", END)

graph = graph_builder.compile()

result = graph.invoke({
    "urls": list(MOCK_RESPONSES.keys()),
    "results": [],
})
# Expected output:
# Successes: 3/5
# Errors: 2/5
#   ⚠️ https://api.example.com/data2: Connection timeout for https://api.example.com/data2
#   ⚠️ https://api.example.com/data4: Connection timeout for https://api.example.com/data4
#   ✅ https://api.example.com/data1: Data from endpoint 1 fetched successfully.
#   ✅ https://api.example.com/data3: Data from endpoint 3 fetched successfully.
#   ✅ https://api.example.com/data5: Data from endpoint 5 fetched successfully.

Exercise 5: Map-reduce with conditional routing by type (Advanced)

Implement a graph that takes a list of mixed documents (emails, tweets, articles). Use Send to route each document to a different processing node depending on its type: "process_email", "process_tweet", "process_article". Each processor extracts different information. The deferred node combines everything.

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langchain.chat_models import init_chat_model

class MainState(TypedDict):
    documents: list[dict]
    extractions: Annotated[list[dict], operator.add]
    summary: str

class DocState(TypedDict):
    doc: dict

def route_by_type(state: MainState) -> list[Send]:
    sends = []
    for doc in state["documents"]:
        doc_type = doc.get("type", "article")
        node_name = f"process_{doc_type}"
        sends.append(Send(node_name, {"doc": doc}))
    return sends

def process_email(state: DocState) -> dict:
    doc = state["doc"]
    return {"extractions": [{
        "type": "email",
        "from": doc.get("from", "unknown"),
        "subject": doc.get("subject", ""),
        "urgency": "high" if "urgent" in doc.get("body", "").lower() else "normal",
    }]}

def process_tweet(state: DocState) -> dict:
    doc = state["doc"]
    body = doc.get("body", "")
    hashtags = [w for w in body.split() if w.startswith("#")]
    return {"extractions": [{
        "type": "tweet",
        "author": doc.get("author", "unknown"),
        "hashtags": hashtags,
        "char_count": len(body),
    }]}

def process_article(state: DocState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Summarize this article in one sentence:\n\n{state['doc'].get('body', '')}"
    )
    return {"extractions": [{
        "type": "article",
        "title": state["doc"].get("title", ""),
        "summary": response.content,
    }]}

def compile_all(state: MainState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    context = "\n".join(str(e) for e in state["extractions"])
    response = model.invoke(
        f"Generate an executive summary of these {len(state['extractions'])} processed documents:\n\n{context}"
    )
    return {"summary": response.content}

graph_builder = StateGraph(MainState)
graph_builder.add_node("process_email", process_email)
graph_builder.add_node("process_tweet", process_tweet)
graph_builder.add_node("process_article", process_article)
graph_builder.add_node("compile", compile_all)

graph_builder.add_conditional_edges(START, route_by_type)
graph_builder.add_edge("process_email", "compile")
graph_builder.add_edge("process_tweet", "compile")
graph_builder.add_edge("process_article", "compile")
graph_builder.add_edge("compile", END)

graph = graph_builder.compile()

result = graph.invoke({
    "documents": [
        {"type": "email", "from": "boss@company.com", "subject": "Q4 Report", "body": "I need the report urgently for the board meeting."},
        {"type": "tweet", "author": "@techguru", "body": "LangGraph is amazing for AI workflows #AI #LangGraph #Python"},
        {"type": "article", "title": "RAG in 2026", "body": "The RAG technique has evolved significantly, integrating semantic search with generative models for more accurate answers."},
        {"type": "email", "from": "team@dev.com", "subject": "Deploy v2.1", "body": "The deploy was successful, no incidents."},
    ],
    "extractions": [],
    "summary": "",
})
print(f"Documents processed: {len(result['extractions'])}")
for ext in result["extractions"]:
    print(f"  [{ext['type']}] {ext}")
print(f"\nSummary: {result['summary'][:200]}...")
# Expected output:
# Documents processed: 4
#   [email] {'type': 'email', 'from': 'boss@company.com', 'subject': 'Q4 Report', 'urgency': 'high'}
#   [tweet] {'type': 'tweet', 'author': '@techguru', 'hashtags': ['#AI', '#LangGraph', '#Python'], 'char_count': 61}
#   [article] {'type': 'article', 'title': 'RAG in 2026', 'summary': 'RAG has evolved by integrating...'}
#   [email] {'type': 'email', 'from': 'team@dev.com', 'subject': 'Deploy v2.1', 'urgency': 'normal'}
# Summary: 4 documents were processed: 2 emails (1 urgent about the Q4 report...

Exercise 6: Nested map-reduce (Advanced)

Implement a two-level system: the first map-reduce processes a list of topics (generating sub-questions per topic), and the second level processes each sub-question. Use Send at both levels. The final result is a dictionary of topic → [answers].

from dotenv import load_dotenv
load_dotenv()

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from langchain.chat_models import init_chat_model

class MainState(TypedDict):
    topics: list[str]
    sub_questions: Annotated[list[dict], operator.add]
    answers: Annotated[list[dict], operator.add]

class TopicState(TypedDict):
    topic: str

class QuestionState(TypedDict):
    topic: str
    question: str

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

def route_topics(state: MainState) -> list[Send]:
    return [Send("generate_questions", {"topic": t}) for t in state["topics"]]

def generate_questions(state: TopicState) -> dict:
    response = model.invoke(
        f"Generate exactly 2 technical questions about '{state['topic']}'. "
        f"Reply with ONLY the questions, one per line."
    )
    questions = [q.strip() for q in response.content.strip().split("\n") if q.strip()][:2]
    return {"sub_questions": [
        {"topic": state["topic"], "question": q} for q in questions
    ]}

def route_questions(state: MainState) -> list[Send]:
    return [
        Send("answer_question", {"topic": sq["topic"], "question": sq["question"]})
        for sq in state["sub_questions"]
    ]

def answer_question(state: QuestionState) -> dict:
    response = model.invoke(
        f"Answer this technical question in 2 sentences:\n{state['question']}"
    )
    return {"answers": [{
        "topic": state["topic"],
        "question": state["question"],
        "answer": response.content,
    }]}

def compile_final(state: MainState) -> dict:
    return {}

graph_builder = StateGraph(MainState)
graph_builder.add_node("generate_questions", generate_questions)
graph_builder.add_node("collect_questions", lambda state: {})
graph_builder.add_node("answer_question", answer_question)
graph_builder.add_node("compile", compile_final)

graph_builder.add_conditional_edges(START, route_topics)
graph_builder.add_edge("generate_questions", "collect_questions")
graph_builder.add_conditional_edges("collect_questions", route_questions)
graph_builder.add_edge("answer_question", "compile")
graph_builder.add_edge("compile", END)

graph = graph_builder.compile()

result = graph.invoke({
    "topics": ["RAG", "Fine-tuning"],
    "sub_questions": [],
    "answers": [],
})

print(f"Topics: {len(result['topics'])}")
print(f"Sub-questions generated: {len(result['sub_questions'])}")
print(f"Answers: {len(result['answers'])}")
for topic in result["topics"]:
    topic_answers = [a for a in result["answers"] if a["topic"] == topic]
    print(f"\n--- {topic} ---")
    for a in topic_answers:
        print(f"  Q: {a['question'][:60]}...")
        print(f"  A: {a['answer'][:80]}...")
# Expected output:
# Topics: 2
# Sub-questions generated: 4
# Answers: 4
#
# --- RAG ---
#   Q: What are the main chunking strategies for RAG...
#   A: Chunking strategies include fixed-size, semantic, and recursive...
#   Q: How do you evaluate the quality of a RAG system...
#   A: The quality of a RAG system is evaluated with metrics like faithfulness...
#
# --- Fine-tuning ---
#   Q: When is fine-tuning preferable to few-shot prompting...
#   A: Fine-tuning is preferable when you have enough training data...
#   Q: What efficient fine-tuning techniques exist for LLMs...
#   A: PEFT techniques like LoRA and QLoRA enable efficient fine-tuning...

Summary

In this capsule you learned:

• The problem: processing collections of variable size — you can't hardcode N branches when N changes at runtime
• The Send API (Send("node", data)) creates branches dynamically at runtime — the routing function returns a list of Send instead of a string
• A separate ProcessorState lets each branch receive only the data it needs, without copying the full state
• Reducers (Annotated[list, operator.add]) are mandatory so the results from multiple branches accumulate instead of overwriting each other
• Deferred nodes automatically wait for all the upstream branches to finish — LangGraph keeps the counter internally
• Consensus patterns: concatenation, majority vote, weighted average, quality filtering — each one solves a different kind of result combination
• A Send can point to different nodes — not every item has to go to the same processor
• The Functional API can pull off map-reduce with Futures and list comprehensions, but the Graph API with Send is more powerful for complex cases

Next capsule: Advanced Error Handling — what happens when your branches fail, and how to build systems that degrade gracefully instead of crashing.

Additional resources

1. Map-Reduce in LangGraph — Official tutorial for the map-reduce pattern with the Send API
2. Send API Reference — Complete reference for the Send class
3. LangGraph Branching — Static vs dynamic branching
4. State Reducers — How reducers work in LangGraph
5. Fan-out / Fan-in patterns — Dynamic fan-out concepts
6. Subgraphs vs Map-Reduce — When to use subgraphs vs map-reduce
7. Python operator module — Reference for operator.add and other operators used as reducers

Module 7 — LangChain & LangGraph: From Chains to Agents