Branching and Merging

Capsule overview

Your Research Agent looks for information in 3 sources: web, academic, and news. Right now, the searches run one after another: 3 seconds + 3 seconds + 3 seconds = 9 seconds. But the searches don't depend on each other — none of them needs another's result to start. What if you ran them at the same time? In parallel: max(3s, 3s, 3s) = 3 seconds. Three times faster, without changing the search logic.

This pattern is called fan-out/fan-in: one node fires multiple nodes that run in parallel (fan-out), and then a merge node collects all the results (fan-in). It's the pattern that turns sequential latency into parallel latency — and users notice.

Branching without merge is an incomplete graph. If you fire off 3 searches but never bring the results back together, you don't have a workflow — you have 3 disconnected workflows. This capsule covers the full cycle: diverge → run in parallel → converge → process the combined results.

The problem: sequential searches

Look at this graph that searches 3 sources sequentially:

from dotenv import load_dotenv
load_dotenv()

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

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

def search_web(state: ResearchState) -> dict:
    time.sleep(1)
    return {"results": [f"[Web] Results on '{state['query']}': Python 3.12 improves typing and performance."]}

def search_academic(state: ResearchState) -> dict:
    time.sleep(1)
    return {"results": [f"[Academic] Paper on '{state['query']}': Comparative analysis of type checkers."]}

def search_news(state: ResearchState) -> dict:
    time.sleep(1)
    return {"results": [f"[News] News on '{state['query']}': Python dominates AI according to a 2026 survey."]}

graph = StateGraph(ResearchState)
graph.add_node("search_web", search_web)
graph.add_node("search_academic", search_academic)
graph.add_node("search_news", search_news)

graph.add_edge(START, "search_web")
graph.add_edge("search_web", "search_academic")
graph.add_edge("search_academic", "search_news")
graph.add_edge("search_news", END)

app = graph.compile()

start = time.time()
result = app.invoke({"query": "Python 3.12", "results": [], "total_time": 0.0})
elapsed = time.time() - start

print(f"Total time: {elapsed:.1f}s")
print(f"Results: {len(result['results'])}")
for r in result["results"]:
    print(f"  {r[:70]}...")
# Expected output:
# Total time: 3.0s
# Results: 3
#   [Web] Results on 'Python 3.12': Python 3.12 improves typing and perfo...
#   [Academic] Paper on 'Python 3.12': Comparative analysis of type checke...
#   [News] News on 'Python 3.12': Python dominates AI according to a 2026 ...

3 seconds. Each search waits for the previous one to finish. The flow is search_web → search_academic → search_news. But no search uses another's result — they're completely independent.

Fan-out: one node fires multiple nodes in parallel

Fan-out is when a node (or the entry point) has edges to multiple nodes. LangGraph detects these branches and runs the destination nodes concurrently:

graph.add_edge(START, "search_web")
graph.add_edge(START, "search_academic")
graph.add_edge(START, "search_news")

Those 3 lines say: "when the graph starts, launch search_web, search_academic, and search_news at the same time." You don't need threading, asyncio, or any concurrency library — LangGraph handles it internally.

How it works under the hood

LangGraph uses an internal scheduler. When a node has multiple outgoing edges going to different nodes, or when multiple edges point from the same origin (like START), the scheduler:

1. Identifies which nodes can run with no pending dependencies
2. Launches them in parallel (using threads internally)
3. Waits for all of them to finish before moving on to nodes that depend on them

You don't control the threading directly. You just declare the graph structure and LangGraph optimizes the execution.

Fan-in: multiple nodes converge into one

Fan-in is fan-out's complement: multiple parallel nodes point their edges toward a single merge node:

graph.add_edge("search_web", "merge_results")
graph.add_edge("search_academic", "merge_results")
graph.add_edge("search_news", "merge_results")

The merge_results node runs only when ALL the upstream nodes have completed. You don't need manual synchronization — LangGraph guarantees the merge receives every result.

The merge rule

The merge node runs exactly once, when the last parallel node finishes. If search_web takes 1s, search_academic takes 2s, and search_news takes 3s, the merge runs at 3 seconds — it waited for the slowest one. But the total time is 3s, not 1+2+3=6s.

State designed for branching: Annotated with operator.add

When multiple nodes write to the same state field, you need a reducer that combines the values instead of overwriting them. Without a reducer, the last node to finish would overwrite the earlier ones' results:

from typing import TypedDict, Annotated
import operator

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

Annotated[list[str], operator.add] says: "when a node returns {"results": [...]}, concatenate the new list with the existing one instead of replacing it." That way each parallel node appends its results and in the end you have all of them.

You already know this design from Module 5 (typed state with TypedDict and Annotated). Here you apply it in a context where it's critical: without the reducer, branching loses data.

Full example: Research Agent with parallel search

Now we put fan-out + fan-in + state with a reducer together:

from dotenv import load_dotenv
load_dotenv()

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

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

def search_web(state: ResearchState) -> dict:
    time.sleep(1)
    return {"results": [
        {"source": "web", "content": f"Web results on '{state['query']}': Python 3.12 introduces typing improvements, faster comprehensions, and better f-strings."}
    ]}

def search_academic(state: ResearchState) -> dict:
    time.sleep(1)
    return {"results": [
        {"source": "academic", "content": f"Paper on '{state['query']}': A comparative study of static type checkers shows a 15% improvement in bug detection with Python 3.12."}
    ]}

def search_news(state: ResearchState) -> dict:
    time.sleep(1)
    return {"results": [
        {"source": "news", "content": f"News on '{state['query']}': Python holds the #1 spot in the 2026 TIOBE index, driven by AI/ML adoption."}
    ]}

def merge_results(state: ResearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    sources_text = "\n\n".join(
        f"[{r['source'].upper()}]: {r['content']}" for r in state["results"]
    )
    response = model.invoke(
        f"Synthesize these sources into a 3-4 sentence executive summary about '{state['query']}':\n\n{sources_text}"
    )
    return {"results": [{"source": "synthesis", "content": response.content}]}

graph = StateGraph(ResearchState)
graph.add_node("search_web", search_web)
graph.add_node("search_academic", search_academic)
graph.add_node("search_news", search_news)
graph.add_node("merge_results", merge_results)

graph.add_edge(START, "search_web")
graph.add_edge(START, "search_academic")
graph.add_edge(START, "search_news")

graph.add_edge("search_web", "merge_results")
graph.add_edge("search_academic", "merge_results")
graph.add_edge("search_news", "merge_results")

graph.add_edge("merge_results", END)

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

start = time.time()
result = app.invoke({"query": "Python 3.12", "results": []})
elapsed = time.time() - start

print(f"Total time: {elapsed:.1f}s (vs ~3.0s sequential)")
print(f"Results collected: {len(result['results'])}")
for r in result["results"]:
    print(f"  [{r['source']}] {r['content'][:70]}...")
# Expected output:
# Total time: ~1.5s (vs ~3.0s sequential)
# Results collected: 4
#   [web] Web results on 'Python 3.12': Python 3.12 introduces typing imp...
#   [academic] Paper on 'Python 3.12': A comparative study of static type ...
#   [news] News on 'Python 3.12': Python holds the #1 spot in the 2026 TIO...
#   [synthesis] Python 3.12 consolidates significant improvements in typin...

Anatomy of the pattern

1. Fan-out: 3 edges from START to the 3 search nodes
2. Parallel execution: LangGraph runs the 3 nodes concurrently
3. Fan-in: 3 edges from the search nodes to merge_results
4. Reducer: Annotated[list[dict], operator.add] accumulates results from every node
5. Merge: merge_results receives the state with all 3 searches and synthesizes

The diagram that draw_mermaid_png() generates clearly shows the diamond shape: one divergence point (START), 3 parallel paths, and one convergence point (merge_results).

Fan-out with a planner node

In the previous example, the fan-out comes straight out of START. But in a real workflow, there's often a planner node that decides what to search for before launching the searches:

from dotenv import load_dotenv
load_dotenv()

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

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

def planner(state: PlanState) -> dict:
    return {"plan": f"Search '{state['query']}' across web, academic, and news simultaneously"}

def search_web(state: PlanState) -> dict:
    time.sleep(1)
    return {"results": [f"[Web] Info on '{state['query']}'"]}

def search_academic(state: PlanState) -> dict:
    time.sleep(1)
    return {"results": [f"[Academic] Papers on '{state['query']}'"]}

def search_news(state: PlanState) -> dict:
    time.sleep(1)
    return {"results": [f"[News] News on '{state['query']}'"]}

def synthesizer(state: PlanState) -> dict:
    combined = " | ".join(state["results"])
    return {"results": [f"[Synthesis] {combined}"]}

graph = StateGraph(PlanState)
graph.add_node("planner", planner)
graph.add_node("search_web", search_web)
graph.add_node("search_academic", search_academic)
graph.add_node("search_news", search_news)
graph.add_node("synthesizer", synthesizer)

graph.add_edge(START, "planner")
graph.add_edge("planner", "search_web")
graph.add_edge("planner", "search_academic")
graph.add_edge("planner", "search_news")
graph.add_edge("search_web", "synthesizer")
graph.add_edge("search_academic", "synthesizer")
graph.add_edge("search_news", "synthesizer")
graph.add_edge("synthesizer", END)

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

start = time.time()
result = app.invoke({"query": "LangGraph branching", "plan": "", "results": []})
elapsed = time.time() - start

print(f"Time: {elapsed:.1f}s")
print(f"Plan: {result['plan']}")
print(f"Results: {len(result['results'])}")
# Expected output:
# Time: ~1.1s
# Plan: Search 'LangGraph branching' across web, academic, and news simultaneously
# Results: 4

The flow is: START → planner → [search_web, search_academic, search_news] → synthesizer → END. The planner runs first, then the 3 searches in parallel, and finally the synthesizer.

Merge strategies

The merge node receives every result. But "bringing results together" can mean different things depending on your use case:

Simple concatenation

The most basic strategy. It joins all the results into one list:

def merge_concatenate(state: ResearchState) -> dict:
    all_content = "\n\n".join(r["content"] for r in state["results"])
    return {"results": [{"source": "merged", "content": all_content}]}

Deduplication

For when multiple sources can find the same information:

def merge_deduplicate(state: ResearchState) -> dict:
    seen = set()
    unique_results = []
    for r in state["results"]:
        content_key = r["content"][:100]
        if content_key not in seen:
            seen.add(content_key)
            unique_results.append(r)
    return {"results": unique_results}

Ranking by confidence

For when each source has a relevance score:

def merge_ranked(state: ResearchState) -> dict:
    source_priority = {"academic": 3, "web": 2, "news": 1}
    sorted_results = sorted(
        state["results"],
        key=lambda r: source_priority.get(r["source"], 0),
        reverse=True,
    )
    return {"results": sorted_results}

Synthesis with an LLM

The most sophisticated strategy — use an LLM to combine and synthesize:

from langchain.chat_models import init_chat_model

def merge_synthesize(state: ResearchState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    context = "\n\n".join(f"[{r['source']}]: {r['content']}" for r in state["results"])
    response = model.invoke(
        f"Synthesize these sources into a coherent summary:\n\n{context}"
    )
    return {"results": [{"source": "synthesis", "content": response.content}]}

Fan-out/fan-in with the Functional API: Futures in parallel

In Module 6, Pattern 3, you saw how the Functional API runs tasks in parallel with Futures. That same pattern is the equivalent of the Graph API's fan-out/fan-in:

from dotenv import load_dotenv
load_dotenv()

import time
from langgraph.func import entrypoint, task

@task
def search_web_fn(query: str) -> dict:
    time.sleep(1)
    return {"source": "web", "content": f"Web results for '{query}'"}

@task
def search_academic_fn(query: str) -> dict:
    time.sleep(1)
    return {"source": "academic", "content": f"Academic papers on '{query}'"}

@task
def search_news_fn(query: str) -> dict:
    time.sleep(1)
    return {"source": "news", "content": f"News about '{query}'"}

@task
def synthesize_fn(results: list) -> str:
    return f"Synthesis of {len(results)} sources: " + " | ".join(r["content"] for r in results)

@entrypoint()
def parallel_research(query: str) -> dict:
    web_future = search_web_fn(query)
    academic_future = search_academic_fn(query)
    news_future = search_news_fn(query)

    results = [
        web_future.result(),
        academic_future.result(),
        news_future.result(),
    ]

    summary = synthesize_fn(results).result()

    return {"results": results, "summary": summary}

start = time.time()
result = parallel_research.invoke("transformer architectures")
elapsed = time.time() - start

print(f"Time: {elapsed:.1f}s")
print(f"Sources: {len(result['results'])}")
print(f"Synthesis: {result['summary'][:80]}...")
# Expected output:
# Time: ~1.1s
# Sources: 3
# Synthesis: Synthesis of 3 sources: Web results for 'transformer architectures'...

The mechanics are identical

# Fan-out: launch without waiting
web_future = search_web_fn(query)        # Launches (doesn't wait)
academic_future = search_academic_fn(query)  # Launches (doesn't wait)
news_future = search_news_fn(query)      # Launches (doesn't wait)

# Fan-in: collect every result
results = [
    web_future.result(),    # Waits for web
    academic_future.result(),  # Waits for academic
    news_future.result(),   # Waits for news
]

The "fan-out" is launching multiple @task without .result(). The "fan-in" is collecting all the .result() afterwards. It's exactly what you saw in Module 6, but now with the right vocabulary: fan-out and fan-in.

Comparison: Graph API branching vs Functional API Futures

When to use each one

• ✅ Graph API when you need flow visualization, a non-technical team needs to understand the workflow, or you have multiple merge strategies
• ✅ Functional API when the parallelism is simple (launch N tasks, collect results), you don't need visualization, or you prefer writing native Python

When branching helps vs when it's overkill

Branching isn't free. It carries scheduling overhead, synchronization, and state complexity. Use it when the benefit outweighs the cost:

The golden rule: branching helps when you have multiple I/O-bound operations that don't depend on each other. API calls, database lookups, LLM invocations — all of them are I/O-bound and benefit from parallelism.

Branching with retry: combining patterns

In capsule 02 you learned retry with exponential backoff to handle APIs that fail. Now combine retry with branching — each parallel search can have its own independent retry:

from dotenv import load_dotenv
load_dotenv()

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

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

def make_search_with_retry(source_name: str, fail_rate: float = 0.5):
    def search(state: RetryState) -> dict:
        max_retries = 3
        for attempt in range(max_retries):
            try:
                if random.random() < fail_rate and attempt < max_retries - 1:
                    raise ConnectionError(f"{source_name} timeout")
                time.sleep(0.5)
                return {"results": [{
                    "source": source_name,
                    "content": f"Results from {source_name} for '{state['query']}'",
                    "attempts": attempt + 1,
                }]}
            except ConnectionError:
                wait = (2 ** attempt) * 0.1 + random.uniform(0, 0.05)
                time.sleep(wait)

        return {"results": [{
            "source": source_name,
            "content": f"[FALLBACK] {source_name} unavailable after {max_retries} attempts",
            "attempts": max_retries,
        }]}
    return search

def merge_with_status(state: RetryState) -> dict:
    successful = [r for r in state["results"] if not r["content"].startswith("[FALLBACK]")]
    failed = [r for r in state["results"] if r["content"].startswith("[FALLBACK]")]
    summary = f"Successful sources: {len(successful)}, failed: {len(failed)}"
    return {"results": [{"source": "merge", "content": summary, "attempts": 0}]}

graph = StateGraph(RetryState)
graph.add_node("search_web", make_search_with_retry("web", fail_rate=0.3))
graph.add_node("search_academic", make_search_with_retry("academic", fail_rate=0.3))
graph.add_node("search_news", make_search_with_retry("news", fail_rate=0.3))
graph.add_node("merge", merge_with_status)

graph.add_edge(START, "search_web")
graph.add_edge(START, "search_academic")
graph.add_edge(START, "search_news")
graph.add_edge("search_web", "merge")
graph.add_edge("search_academic", "merge")
graph.add_edge("search_news", "merge")
graph.add_edge("merge", END)

app = graph.compile()

result = app.invoke({"query": "LangGraph patterns", "results": []})
for r in result["results"]:
    print(f"  [{r['source']}] (attempts: {r['attempts']}) {r['content'][:60]}...")
# Expected output (varies with the randomness):
#   [web] (attempts: 2) Results from web for 'LangGraph patterns'...
#   [academic] (attempts: 1) Results from academic for 'LangGraph patterns'...
#   [news] (attempts: 3) [FALLBACK] news unavailable after 3 attempts...
#   [merge] (attempts: 0) Successful sources: 2, failed: 1...

Each search node handles its own retries internally. The merge receives every result (successful and fallback) and can decide how to proceed. The retries happen in parallel — if search_web needs 2 attempts and search_news needs 3, both retry at the same time.

Conditional branching: selective fan-out

You don't always want to search every source. Sometimes a planner decides which sources to consult. For that you combine conditional edges with branching:

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 SelectiveState(TypedDict):
    query: str
    query_type: str
    results: Annotated[list[str], operator.add]

def classifier(state: SelectiveState) -> dict:
    q = state["query"].lower()
    if any(w in q for w in ["paper", "study", "research"]):
        return {"query_type": "academic"}
    elif any(w in q for w in ["news", "today", "recent"]):
        return {"query_type": "news"}
    return {"query_type": "general"}

def route_searches(state: SelectiveState) -> list[str]:
    qt = state["query_type"]
    if qt == "academic":
        return ["search_web", "search_academic"]
    elif qt == "news":
        return ["search_web", "search_news"]
    return ["search_web", "search_academic", "search_news"]

def search_web(state: SelectiveState) -> dict:
    return {"results": [f"[Web] Results for '{state['query']}'"]}

def search_academic(state: SelectiveState) -> dict:
    return {"results": [f"[Academic] Papers on '{state['query']}'"]}

def search_news(state: SelectiveState) -> dict:
    return {"results": [f"[News] News on '{state['query']}'"]}

def merge(state: SelectiveState) -> dict:
    return {"results": [f"[Merge] Combined {len(state['results'])} results"]}

graph = StateGraph(SelectiveState)
graph.add_node("classifier", classifier)
graph.add_node("search_web", search_web)
graph.add_node("search_academic", search_academic)
graph.add_node("search_news", search_news)
graph.add_node("merge", merge)

graph.add_edge(START, "classifier")
graph.add_conditional_edges("classifier", route_searches, ["search_web", "search_academic", "search_news"])
graph.add_edge("search_web", "merge")
graph.add_edge("search_academic", "merge")
graph.add_edge("search_news", "merge")
graph.add_edge("merge", END)

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

for query in ["I'm looking for papers on transformers", "What news is there today?", "Explain what RAG is"]:
    result = app.invoke({"query": query, "query_type": "", "results": []})
    sources = [r for r in result["results"] if not r.startswith("[Merge]")]
    print(f"'{query}' → {len(sources)} sources: {sources}")
# Expected output:
# 'I'm looking for papers on transformers' → 2 sources: ['[Web]...', '[Academic]...']
# 'What news is there today?' → 2 sources: ['[Web]...', '[News]...']
# 'Explain what RAG is' → 3 sources: ['[Web]...', '[Academic]...', '[News]...']

The routing function returns a list of nodes instead of a string. LangGraph runs only the nodes in the list, in parallel. The unselected nodes don't run, but their edges to the merge still exist — the merge waits only for the nodes that actually ran.

Troubleshooting

Problem 1: "The merge node runs before all the parallel nodes finish"

Symptom: The merge receives incomplete results — sources are missing.

Cause: A parallel node has no edge to the merge, or the parallel node isn't returning state correctly.

Fix: Check that every parallel node has an edge to the merge. Use draw_mermaid_png() to confirm it visually:

display(Image(app.get_graph().draw_mermaid_png()))

If a node appears disconnected from the merge in the diagram, an add_edge is missing.

Problem 2: "The parallel nodes' results overwrite each other"

Symptom: The final state only has one node's result, not all of them.

Cause: The state field has no reducer. Without Annotated[list, operator.add], the last node to write overwrites the earlier ones.

Fix: Add the reducer to the field that accumulates results:

# ❌ Without a reducer (last one wins)
class State(TypedDict):
    results: list[str]

# ✅ With a reducer (everything accumulates)
class State(TypedDict):
    results: Annotated[list[str], operator.add]

Problem 3: "The result order is unpredictable"

Symptom: The results arrive in a different order on every run.

Cause: Parallel nodes finish in a non-deterministic order. Whichever finishes first appends to the state first.

Fix: If order matters, add metadata (timestamp, source name) and sort in the merge:

def merge_ordered(state: ResearchState) -> dict:
    source_order = {"web": 0, "academic": 1, "news": 2}
    sorted_results = sorted(state["results"], key=lambda r: source_order.get(r["source"], 99))
    return {"results": sorted_results}

Problem 4: "One parallel node fails and the whole graph goes down"

Symptom: An exception in one search node stops the entire workflow.

Cause: Uncaught exceptions in nodes propagate to the scheduler. In branching, a failure in any branch cancels all of them.

Fix: Catch exceptions inside each node and return a fallback result:

def search_web(state: ResearchState) -> dict:
    try:
        result = do_actual_search(state["query"])
        return {"results": [{"source": "web", "content": result, "status": "ok"}]}
    except Exception as e:
        return {"results": [{"source": "web", "content": str(e), "status": "error"}]}

Problem 5: "I see no parallelism — the time matches the sequential run"

Symptom: 3 nodes of 1s each take 3s in total, not ~1s.

Cause: The edges are chained instead of fanning out:

# ❌ Sequential (chained)
graph.add_edge(START, "search_web")
graph.add_edge("search_web", "search_academic")
graph.add_edge("search_academic", "search_news")

# ✅ Parallel (fan-out from the same node)
graph.add_edge(START, "search_web")
graph.add_edge(START, "search_academic")
graph.add_edge(START, "search_news")

Check that all the fan-out edges leave from the same node.

Exercises

Exercise 1: Basic fan-out (Easy)

Create a graph with 2 parallel nodes ("uppercase" and "word_count") that process the same text. uppercase converts to uppercase, word_count counts words. Both write to a field with a reducer. A "report" node at the end combines the results. Visualize it.

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 State(TypedDict):
    text: str
    analyses: Annotated[list[dict], operator.add]

def uppercase(state: State) -> dict:
    return {"analyses": [{"type": "uppercase", "result": state["text"].upper()}]}

def word_count(state: State) -> dict:
    count = len(state["text"].split())
    return {"analyses": [{"type": "word_count", "result": str(count)}]}

def report(state: State) -> dict:
    results = {a["type"]: a["result"] for a in state["analyses"]}
    summary = f"Text: {results.get('uppercase', 'N/A')} | Words: {results.get('word_count', 'N/A')}"
    return {"analyses": [{"type": "report", "result": summary}]}

graph = StateGraph(State)
graph.add_node("uppercase", uppercase)
graph.add_node("word_count", word_count)
graph.add_node("report", report)

graph.add_edge(START, "uppercase")
graph.add_edge(START, "word_count")
graph.add_edge("uppercase", "report")
graph.add_edge("word_count", "report")
graph.add_edge("report", END)

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

result = app.invoke({"text": "hello world from langgraph", "analyses": []})
for a in result["analyses"]:
    print(f"  [{a['type']}] {a['result']}")
# Expected output:
#   [uppercase] HELLO WORLD FROM LANGGRAPH
#   [word_count] 4
#   [report] Text: HELLO WORLD FROM LANGGRAPH | Words: 4

Exercise 2: Merge with deduplication (Easy)

Create 3 parallel search nodes that return results with overlap (2 of the 3 nodes return the same result). The merge node has to deduplicate by content before reporting. Use operator.add in the state.

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[dict], operator.add]

def source_a(state: State) -> dict:
    return {"results": [
        {"content": "Python is a programming language", "source": "A"},
        {"content": "Python was created by Guido van Rossum", "source": "A"},
    ]}

def source_b(state: State) -> dict:
    return {"results": [
        {"content": "Python is a programming language", "source": "B"},
        {"content": "Python is popular in data science", "source": "B"},
    ]}

def source_c(state: State) -> dict:
    return {"results": [
        {"content": "Python was created by Guido van Rossum", "source": "C"},
        {"content": "Python 3.12 is the latest version", "source": "C"},
    ]}

def merge_dedup(state: State) -> dict:
    seen_content = set()
    unique = []
    for r in state["results"]:
        if r["content"] not in seen_content:
            seen_content.add(r["content"])
            unique.append(r)
    return {"results": [{"content": f"Unique results: {len(unique)} out of {len(state['results'])} total", "source": "merge"}]}

graph = StateGraph(State)
graph.add_node("source_a", source_a)
graph.add_node("source_b", source_b)
graph.add_node("source_c", source_c)
graph.add_node("merge", merge_dedup)

graph.add_edge(START, "source_a")
graph.add_edge(START, "source_b")
graph.add_edge(START, "source_c")
graph.add_edge("source_a", "merge")
graph.add_edge("source_b", "merge")
graph.add_edge("source_c", "merge")
graph.add_edge("merge", END)

app = graph.compile()
result = app.invoke({"query": "Python", "results": []})
for r in result["results"]:
    print(f"  [{r['source']}] {r['content']}")
# Expected output:
#   [A] Python is a programming language
#   [A] Python was created by Guido van Rossum
#   [B] Python is popular in data science
#   [C] Python 3.12 is the latest version
#   [merge] Unique results: 4 out of 6 total

Exercise 3: Fan-out with per-branch error handling (Medium)

Create 3 parallel search nodes where one always fails (raises an exception). Each node has to handle its own error internally and return a result with status "error" or "ok". The merge reports how many sources succeeded.

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[dict], operator.add]

def safe_search(source_name: str, should_fail: bool = False):
    def search(state: State) -> dict:
        try:
            if should_fail:
                raise ConnectionError(f"{source_name} is down")
            return {"results": [{
                "source": source_name,
                "content": f"Results from {source_name} for '{state['query']}'",
                "status": "ok",
            }]}
        except Exception as e:
            return {"results": [{
                "source": source_name,
                "content": str(e),
                "status": "error",
            }]}
    return search

def merge(state: State) -> dict:
    ok = [r for r in state["results"] if r["status"] == "ok"]
    errors = [r for r in state["results"] if r["status"] == "error"]
    summary = f"{len(ok)} sources OK, {len(errors)} with errors"
    return {"results": [{"source": "merge", "content": summary, "status": "done"}]}

graph = StateGraph(State)
graph.add_node("search_a", safe_search("source_a"))
graph.add_node("search_b", safe_search("source_b", should_fail=True))
graph.add_node("search_c", safe_search("source_c"))
graph.add_node("merge", merge)

graph.add_edge(START, "search_a")
graph.add_edge(START, "search_b")
graph.add_edge(START, "search_c")
graph.add_edge("search_a", "merge")
graph.add_edge("search_b", "merge")
graph.add_edge("search_c", "merge")
graph.add_edge("merge", END)

app = graph.compile()
result = app.invoke({"query": "test", "results": []})
for r in result["results"]:
    print(f"  [{r['source']}] ({r['status']}) {r['content']}")
# Expected output:
#   [source_a] (ok) Results from source_a for 'test'
#   [source_b] (error) source_b is down
#   [source_c] (ok) Results from source_c for 'test'
#   [merge] (done) 2 sources OK, 1 with errors

Exercise 4: Functional API with a custom merge (Medium)

Implement the same parallel search pattern with the Functional API. 3 search tasks in parallel, collection with error handling, and a merge task that sorts results by source priority (academic > web > news).

from dotenv import load_dotenv
load_dotenv()

import time
from langgraph.func import entrypoint, task

@task
def search_web(query: str) -> dict:
    time.sleep(0.5)
    return {"source": "web", "content": f"Web results for '{query}'", "priority": 2}

@task
def search_academic(query: str) -> dict:
    time.sleep(0.5)
    return {"source": "academic", "content": f"Papers on '{query}'", "priority": 3}

@task
def search_news(query: str) -> dict:
    time.sleep(0.5)
    return {"source": "news", "content": f"News about '{query}'", "priority": 1}

@task
def merge_ranked(results: list) -> dict:
    sorted_results = sorted(results, key=lambda r: r["priority"], reverse=True)
    return {
        "ranked_results": sorted_results,
        "order": [r["source"] for r in sorted_results],
    }

@entrypoint()
def ranked_research(query: str) -> dict:
    web_fut = search_web(query)
    academic_fut = search_academic(query)
    news_fut = search_news(query)

    results = []
    errors = []
    for name, fut in [("web", web_fut), ("academic", academic_fut), ("news", news_fut)]:
        try:
            results.append(fut.result())
        except Exception as e:
            errors.append({"source": name, "error": str(e)})

    merged = merge_ranked(results).result()

    return {
        "results": merged["ranked_results"],
        "order": merged["order"],
        "errors": errors,
    }

start = time.time()
result = ranked_research.invoke("machine learning")
elapsed = time.time() - start

print(f"Time: {elapsed:.1f}s")
print(f"Priority order: {result['order']}")
for r in result["results"]:
    print(f"  [{r['source']}] (priority {r['priority']}) {r['content']}")
# Expected output:
# Time: ~0.6s
# Priority order: ['academic', 'web', 'news']
#   [academic] (priority 3) Papers on 'machine learning'
#   [web] (priority 2) Web results for 'machine learning'
#   [news] (priority 1) News about 'machine learning'

Exercise 5: Full Research Agent with planner + branching + LLM merge (Advanced)

Build a Research Agent with the Graph API that has: (1) a planner node that analyzes the query, (2) fan-out to 3 search sources in parallel, (3) a merge node that uses an LLM to synthesize the results into an executive summary. Use init_chat_model for the merge. Visualize the graph.

from dotenv import load_dotenv
load_dotenv()

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

class AgentState(TypedDict):
    query: str
    plan: str
    results: Annotated[list[dict], operator.add]
    summary: str

def planner(state: AgentState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Analyze this research query and generate a brief plan (1-2 sentences) "
        f"for how to search for information:\n\n{state['query']}"
    )
    return {"plan": response.content}

def search_web(state: AgentState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Simulate a web search. Generate an informative paragraph about: {state['query']}"
    )
    return {"results": [{"source": "web", "content": response.content}]}

def search_academic(state: AgentState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Simulate an academic search. Generate a technical abstract about: {state['query']}"
    )
    return {"results": [{"source": "academic", "content": response.content}]}

def search_news(state: AgentState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Simulate a news search. Generate a journalistic summary about: {state['query']}"
    )
    return {"results": [{"source": "news", "content": response.content}]}

def merge_synthesize(state: AgentState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    context = "\n\n".join(
        f"[{r['source'].upper()}]:\n{r['content']}" for r in state["results"]
    )
    response = model.invoke(
        f"Research plan: {state['plan']}\n\n"
        f"Collected sources:\n\n{context}\n\n"
        f"Generate a 4-5 sentence executive summary that synthesizes the sources."
    )
    return {"summary": response.content}

graph = StateGraph(AgentState)
graph.add_node("planner", planner)
graph.add_node("search_web", search_web)
graph.add_node("search_academic", search_academic)
graph.add_node("search_news", search_news)
graph.add_node("merge", merge_synthesize)

graph.add_edge(START, "planner")
graph.add_edge("planner", "search_web")
graph.add_edge("planner", "search_academic")
graph.add_edge("planner", "search_news")
graph.add_edge("search_web", "merge")
graph.add_edge("search_academic", "merge")
graph.add_edge("search_news", "merge")
graph.add_edge("merge", END)

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

start = time.time()
result = app.invoke({"query": "How does RAG impact the accuracy of LLMs?", "plan": "", "results": [], "summary": ""})
elapsed = time.time() - start

print(f"Time: {elapsed:.1f}s")
print(f"Plan: {result['plan'][:100]}...")
print(f"Sources: {len(result['results'])}")
print(f"Summary: {result['summary'][:200]}...")
# Expected output:
# Time: ~3-5s
# Plan: Search for information about RAG and its impact on accuracy...
# Sources: 3
# Summary: RAG (Retrieval-Augmented Generation) significantly improves...

Exercise 6: Dynamic branching with a variable number of sources (Advanced)

Adapt the previous exercise so the planner decides how many and which sources to consult. If the query is technical, search web + academic. If it's about recent events, search web + news. If it's general, search all 3. Use conditional edges that return a list of nodes.

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 State(TypedDict):
    query: str
    query_type: str
    results: Annotated[list[str], operator.add]

def classifier(state: State) -> dict:
    q = state["query"].lower()
    if any(w in q for w in ["paper", "study", "technical", "algorithm"]):
        return {"query_type": "technical"}
    elif any(w in q for w in ["news", "today", "recent", "launch"]):
        return {"query_type": "current"}
    return {"query_type": "general"}

def route_to_sources(state: State) -> list[str]:
    qt = state["query_type"]
    if qt == "technical":
        return ["search_web", "search_academic"]
    elif qt == "current":
        return ["search_web", "search_news"]
    return ["search_web", "search_academic", "search_news"]

def search_web(state: State) -> dict:
    return {"results": [f"[Web] Results for '{state['query']}'"]}

def search_academic(state: State) -> dict:
    return {"results": [f"[Academic] Papers on '{state['query']}'"]}

def search_news(state: State) -> dict:
    return {"results": [f"[News] News on '{state['query']}'"]}

def merge(state: State) -> dict:
    sources_used = len(state["results"])
    return {"results": [f"[Merge] Synthesized {sources_used} results"]}

graph = StateGraph(State)
graph.add_node("classifier", classifier)
graph.add_node("search_web", search_web)
graph.add_node("search_academic", search_academic)
graph.add_node("search_news", search_news)
graph.add_node("merge", merge)

graph.add_edge(START, "classifier")
graph.add_conditional_edges(
    "classifier",
    route_to_sources,
    ["search_web", "search_academic", "search_news"],
)
graph.add_edge("search_web", "merge")
graph.add_edge("search_academic", "merge")
graph.add_edge("search_news", "merge")
graph.add_edge("merge", END)

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

test_queries = [
    "I'm looking for a paper on transformers",
    "What's the AI news today?",
    "Explain what machine learning is",
]

for query in test_queries:
    result = app.invoke({"query": query, "query_type": "", "results": []})
    sources = [r for r in result["results"] if not r.startswith("[Merge]")]
    print(f"'{query}' → type: {result['query_type']}, sources: {len(sources)}")
    for s in sources:
        print(f"    {s}")
# Expected output:
# 'I'm looking for a paper on transformers' → type: technical, sources: 2
#     [Web] Results for 'I'm looking for a paper on transformers'
#     [Academic] Papers on 'I'm looking for a paper on transformers'
# 'What's the AI news today?' → type: current, sources: 2
#     [Web] Results for 'What's the AI news today?'
#     [News] News on 'What's the AI news today?'
# 'Explain what machine learning is' → type: general, sources: 3
#     [Web] Results for 'Explain what machine learning is'
#     [Academic] Papers on 'Explain what machine learning is'
#     [News] News on 'Explain what machine learning is'

Summary

In this capsule you learned:

• Fan-out is when one node fires multiple nodes in parallel — you declare it with multiple add_edge calls from the same source node
• Fan-in is when multiple parallel nodes converge into a single merge node — you declare it with multiple add_edge calls to the same destination node
• The merge waits for everyone: the convergence node only runs when the last parallel node finishes
• Annotated with operator.add is critical: without a reducer, parallel nodes overwrite each other's results
• Merge strategies: concatenation (join everything), deduplication (drop repeats), ranking (sort by priority), LLM synthesis (generate a coherent summary)
• In the Functional API, fan-out = launch @task without .result(), fan-in = collect .result() afterwards
• Conditional branching lets a planner decide which branches to run, by returning a list of destination nodes
• Retry + branching combine: each parallel branch can have its own independent retry
• When to use branching: independent I/O-bound operations (APIs, LLMs, searches). Not for fast or CPU-bound operations

Next capsule: Subgraphs — how to encapsulate whole flows as reusable nodes. Just as a function can call another function, a graph can contain another graph.

Additional resources

1. LangGraph Branching — How-to Guide — Fan-out and fan-in with complete examples
2. LangGraph Concepts: Nodes and Edges — Edge documentation, including branching
3. State Reducers — How reducers like operator.add work
4. Functional API Parallelism — Futures and parallelism in the Functional API
5. LangGraph Visualization — Visualizing graphs with draw_mermaid_png
6. Python operator module — Reference for operator.add and other operators
7. Concurrent Futures (Python docs) — The analogous Futures concept in standard Python

Module 7 — LangChain & LangGraph: From Chains to Agents