Module 7: Advanced Flows

Advanced Error Handling

Next →

Capsule overview

Your agent calls 3 external APIs. One is down. Without error handling, the whole system crashes — the user gets nothing. With error handling, 2 out of 3 sources keep working — the user gets a partial but useful result. That's the difference between a prototype and a production system.

Error handling is not an afterthought. It isn't something you add at the end "if there's time left." It's fundamental engineering — as critical as the business logic. A system that works 99% of the time but crashes catastrophically the other 1% is not production-ready. In the real world, APIs fail, rate limits kick in, connections drop, and LLMs hallucinate. Your system has to survive all of it.

In this capsule you'll learn the patterns that make the difference: fallback nodes for when the main path fails, graceful degradation to keep going with partial results, a circuit breaker so you don't hammer a service that's already down, and how to combine retry + fallback + circuit breaker into the full production-grade pattern.

The concrete problem

Your Research Assistant looks for information in 3 sources: Wikipedia, arXiv and a news API. Each source is an external API that can fail:

Scenario 1: Everything works → 3/3 sources respond → complete result
Scenario 2: arXiv times out → 2/3 sources respond → partial result (useful)
Scenario 3: Rate limit on all of them → 0/3 sources respond → ???

Without error handling, scenarios 2 and 3 crash the system. With error handling:

  • Scenario 2: the system continues with Wikipedia and news, notes that arXiv failed, and tells the user the result is partial
  • Scenario 3: the system fires a fallback (cache, default result, or an explanatory message), and doesn't crash

The user always gets something useful. That's graceful degradation.

Fallback nodes: the graph's plan B

A fallback node is a node that fires when the main path fails. You implement it with a conditional edge that checks whether there was an error:

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 langchain_core.messages import AnyMessage, HumanMessage
from IPython.display import Image, display

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    search_result: str
    error: str
    source_used: str

def primary_search(state: State) -> dict:
    """Primary search — it can fail."""
    try:
        raise ConnectionError("Primary search API unavailable")
    except Exception as e:
        return {"error": str(e), "search_result": "", "source_used": ""}

def route_after_search(state: State) -> str:
    if state.get("error"):
        return "fallback"
    return "process"

def fallback_search(state: State) -> dict:
    """Plan B: use cached results or an alternative source."""
    cached = "Python is a high-level programming language created by Guido van Rossum."
    return {
        "search_result": cached,
        "error": "",
        "source_used": "cache",
    }

def process_result(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    source = state.get("source_used", "primary")
    response = model.invoke(
        f"Generate an answer based on this information (source: {source}):\n\n"
        f"{state['search_result']}"
    )
    return {"messages": [response]}

graph_builder = StateGraph(State)
graph_builder.add_node("primary_search", primary_search)
graph_builder.add_node("fallback_search", fallback_search)
graph_builder.add_node("process", process_result)

graph_builder.add_edge(START, "primary_search")
graph_builder.add_conditional_edges(
    "primary_search", route_after_search,
    {"fallback": "fallback_search", "process": "process"}
)
graph_builder.add_edge("fallback_search", "process")
graph_builder.add_edge("process", END)

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

result = graph.invoke({
    "messages": [HumanMessage(content="What is Python?")],
    "search_result": "", "error": "", "source_used": "",
})
print(f"Source used: {result['source_used']}")
print(f"Answer: {result['messages'][-1].content[:120]}...")
# Expected output:
# Source used: cache
# Answer: Python is a high-level programming language, created by Guido van Rossum...

Anatomy of the fallback

  1. The primary node attempts the operation and captures errors into the state (error field)
  2. The routing function checks whether there's an error → "fallback" or success → "process"
  3. The fallback node supplies an alternative result and clears the error
  4. Both paths converge on "process" — the processing node doesn't know (and doesn't care) where the result came from

Fallback chain: A → B → C → default

When you have multiple alternative sources, chain them:

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

class State(TypedDict):
    query: str
    result: str
    source: str
    attempts: list

def search_premium_api(state: State) -> dict:
    """Source A: premium API (expensive but complete)."""
    try:
        raise ConnectionError("Premium API: rate limit exceeded")
    except Exception as e:
        return {
            "result": "",
            "source": "",
            "attempts": [{"source": "premium_api", "status": "error", "detail": str(e)}],
        }

def route_after_premium(state: State) -> str:
    return "try_free" if not state.get("result") else "done"

def search_free_api(state: State) -> dict:
    """Source B: free API (limited but functional)."""
    try:
        raise TimeoutError("Free API: connection timeout after 10s")
    except Exception as e:
        attempts = state.get("attempts", [])
        return {
            "result": "",
            "source": "",
            "attempts": attempts + [{"source": "free_api", "status": "error", "detail": str(e)}],
        }

def route_after_free(state: State) -> str:
    return "try_cache" if not state.get("result") else "done"

def search_cache(state: State) -> dict:
    """Source C: local cache (potentially stale data)."""
    attempts = state.get("attempts", [])
    return {
        "result": "Cached result: LangGraph is a framework for building agents with state graphs.",
        "source": "cache",
        "attempts": attempts + [{"source": "cache", "status": "ok", "detail": "Hit"}],
    }

def format_response(state: State) -> dict:
    return {}

graph_builder = StateGraph(State)
graph_builder.add_node("premium", search_premium_api)
graph_builder.add_node("free", search_free_api)
graph_builder.add_node("cache", search_cache)
graph_builder.add_node("done", format_response)

graph_builder.add_edge(START, "premium")
graph_builder.add_conditional_edges("premium", route_after_premium, {"try_free": "free", "done": "done"})
graph_builder.add_conditional_edges("free", route_after_free, {"try_cache": "cache", "done": "done"})
graph_builder.add_edge("cache", "done")
graph_builder.add_edge("done", END)

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

result = graph.invoke({
    "query": "What is LangGraph?",
    "result": "", "source": "", "attempts": [],
})
print(f"Final source: {result['source']}")
print(f"Result: {result['result']}")
print(f"Attempts:")
for a in result["attempts"]:
    status_icon = "✅" if a["status"] == "ok" else "❌"
    print(f"  {status_icon} {a['source']}: {a['detail']}")
# Expected output:
# Final source: cache
# Result: Cached result: LangGraph is a framework for building agents with state graphs.
# Attempts:
#   ❌ premium_api: Premium API: rate limit exceeded
#   ❌ free_api: Free API: connection timeout after 10s
#   ✅ cache: Hit

The graph tries A, then B, then C. The attempts field records every attempt for debugging. The user gets the result from the first source that works.

Graceful degradation: continuing with partial results

In parallel branching, when one of N branches fails, the others may have succeeded. Instead of throwing everything away, use what did work:

from dotenv import load_dotenv
load_dotenv()

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

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

def search_wikipedia(state: State) -> dict:
    return {"results": [{
        "source": "wikipedia",
        "status": "ok",
        "data": f"Wikipedia: encyclopedic information about '{state['query']}'.",
    }]}

def search_arxiv(state: State) -> dict:
    return {"results": [{
        "source": "arxiv",
        "status": "error",
        "data": None,
        "error": "arXiv API: 503 Service Unavailable",
    }]}

def search_news(state: State) -> dict:
    return {"results": [{
        "source": "news",
        "status": "ok",
        "data": f"News: latest news about '{state['query']}'.",
    }]}

def synthesize(state: State) -> dict:
    successes = [r for r in state["results"] if r["status"] == "ok"]
    failures = [r for r in state["results"] if r["status"] == "error"]

    if not successes:
        return {"results": [{
            "source": "system",
            "status": "error",
            "data": "Could not retrieve information from any source.",
        }]}

    model = init_chat_model("openai:gpt-4.1-mini")
    context = "\n".join(r["data"] for r in successes)

    degradation_note = ""
    if failures:
        failed_sources = ", ".join(r["source"] for r in failures)
        degradation_note = f"\n\nNote: {len(failures)} source(s) unavailable: {failed_sources}. "
        degradation_note += f"Result based on {len(successes)}/{len(state['results'])} sources."

    response = model.invoke(
        f"Synthesize this information about '{state['query']}':\n\n{context}{degradation_note}"
    )

    return {"results": [{
        "source": "synthesis",
        "status": "degraded" if failures else "ok",
        "data": response.content,
        "sources_used": len(successes),
        "sources_failed": len(failures),
    }]}

graph_builder = StateGraph(State)
graph_builder.add_node("wikipedia", search_wikipedia)
graph_builder.add_node("arxiv", search_arxiv)
graph_builder.add_node("news", search_news)
graph_builder.add_node("synthesize", synthesize)

graph_builder.add_edge(START, "wikipedia")
graph_builder.add_edge(START, "arxiv")
graph_builder.add_edge(START, "news")
graph_builder.add_edge("wikipedia", "synthesize")
graph_builder.add_edge("arxiv", "synthesize")
graph_builder.add_edge("news", "synthesize")
graph_builder.add_edge("synthesize", END)

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

result = graph.invoke({"query": "large language models", "results": []})
synthesis = [r for r in result["results"] if r["source"] == "synthesis"][0]
print(f"Status: {synthesis['status']}")
print(f"Sources used: {synthesis['sources_used']}/{synthesis['sources_used'] + synthesis['sources_failed']}")
print(f"Answer: {synthesis['data'][:150]}...")
# Expected output:
# Status: degraded
# Sources used: 2/3
# Answer: Large language models (LLMs) are AI systems based on...

The keys to graceful degradation

  1. Every node catches its own errors — it returns {"status": "error"} instead of raising exceptions
  2. The synthesis node separates successes from failures — it works only with the successes
  3. The user is told about the degradation — they know the result is partial
  4. It never returns empty — even if every source fails, it returns an explanatory message

Circuit breaker: protecting services that are down

When an API keeps failing, continuing to try is counterproductive: you waste time, you pile unnecessary load onto a service that's already down, and you delay the user's answer. The circuit breaker solves this:

CLOSED state (normal):        → Calls go through normally
OPEN state (service down):    → Calls are rejected immediately, without trying
HALF-OPEN state (probing):    → Allows ONE test call to see whether it recovered

from dotenv import load_dotenv
load_dotenv()

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

class CircuitState(TypedDict):
    query: str
    result: str
    circuit_breaker: dict

def init_circuit_breaker() -> dict:
    return {
        "failure_count": 0,
        "max_failures": 3,
        "state": "closed",
        "last_failure_time": 0,
        "cooldown_seconds": 30,
    }

def check_circuit(cb: dict) -> dict:
    """Checks whether the circuit breaker allows the call."""
    if cb["state"] == "closed":
        return {**cb, "allowed": True}

    if cb["state"] == "open":
        elapsed = time.time() - cb["last_failure_time"]
        if elapsed >= cb["cooldown_seconds"]:
            return {**cb, "state": "half-open", "allowed": True}
        return {**cb, "allowed": False}

    if cb["state"] == "half-open":
        return {**cb, "allowed": True}

    return {**cb, "allowed": False}

def record_success(cb: dict) -> dict:
    """Records a successful call."""
    return {**cb, "failure_count": 0, "state": "closed"}

def record_failure(cb: dict) -> dict:
    """Records a failed call."""
    new_count = cb["failure_count"] + 1
    new_state = "open" if new_count >= cb["max_failures"] else cb["state"]
    return {
        **cb,
        "failure_count": new_count,
        "state": new_state,
        "last_failure_time": time.time(),
    }

CALL_COUNT = 0

def call_api(state: CircuitState) -> dict:
    cb = state.get("circuit_breaker", init_circuit_breaker())
    cb = check_circuit(cb)

    if not cb.get("allowed"):
        return {
            "result": f"[CIRCUIT OPEN] Service unavailable, using cache. Failures: {cb['failure_count']}",
            "circuit_breaker": cb,
        }

    global CALL_COUNT
    CALL_COUNT += 1

    try:
        if CALL_COUNT <= 3:
            raise ConnectionError(f"API error on attempt #{CALL_COUNT}")
        return {
            "result": f"[OK] Data fetched successfully on attempt #{CALL_COUNT}",
            "circuit_breaker": record_success(cb),
        }
    except Exception as e:
        return {
            "result": f"[ERROR] {str(e)}",
            "circuit_breaker": record_failure(cb),
        }

graph_builder = StateGraph(CircuitState)
graph_builder.add_node("call_api", call_api)
graph_builder.add_edge(START, "call_api")
graph_builder.add_edge("call_api", END)

graph = graph_builder.compile()

CALL_COUNT = 0
cb = init_circuit_breaker()
for i in range(6):
    result = graph.invoke({
        "query": "test",
        "result": "",
        "circuit_breaker": cb,
    })
    cb = result["circuit_breaker"]
    print(f"Call {i+1}: {result['result']} | Circuit: {cb['state']} ({cb['failure_count']} failures)")
# Expected output:
# Call 1: [ERROR] API error on attempt #1 | Circuit: closed (1 failures)
# Call 2: [ERROR] API error on attempt #2 | Circuit: closed (2 failures)
# Call 3: [ERROR] API error on attempt #3 | Circuit: open (3 failures)
# Call 4: [CIRCUIT OPEN] Service unavailable, using cache. Failures: 3 | Circuit: open (3 failures)
# Call 5: [CIRCUIT OPEN] Service unavailable, using cache. Failures: 3 | Circuit: open (3 failures)
# Call 6: [CIRCUIT OPEN] Service unavailable, using cache. Failures: 3 | Circuit: open (3 failures)

How to wire the circuit breaker into the state

The circuit breaker lives in the graph's state as a dict. Every node that calls an external API:

  1. Reads the circuit breaker from the state
  2. Checks whether it's allowed
  3. If not, returns a fallback result immediately
  4. If so, attempts the call
  5. Updates the circuit breaker (success or failure)

In a system with multiple APIs, keep one circuit breaker per service:

class State(TypedDict):
    circuit_breakers: dict  # {"wikipedia": {...}, "arxiv": {...}, "news": {...}}

Error handling in parallel branches

When you run branches in parallel (fan-out), each branch can fail independently. The pattern: each branch handles its own errors and returns a result with a status. The merge node filters:

from dotenv import load_dotenv
load_dotenv()

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

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

def safe_search(source_name: str, should_fail: bool = False):
    """Factory that creates search nodes with internal error handling."""
    def node(state: State) -> dict:
        try:
            if should_fail:
                raise ConnectionError(f"{source_name}: service unavailable")
            return {"branch_results": [{
                "source": source_name,
                "status": "ok",
                "data": f"Results from {source_name} about '{state['query']}'.",
            }]}
        except Exception as e:
            return {"branch_results": [{
                "source": source_name,
                "status": "error",
                "data": None,
                "error": str(e),
            }]}
    return node

def smart_merge(state: State) -> dict:
    ok = [r for r in state["branch_results"] if r["status"] == "ok"]
    errors = [r for r in state["branch_results"] if r["status"] == "error"]

    if ok:
        model = init_chat_model("openai:gpt-4.1-mini")
        context = "\n".join(r["data"] for r in ok)
        response = model.invoke(
            f"Synthesize about '{state['query']}':\n\n{context}"
        )
        answer = response.content
    else:
        answer = "Could not retrieve information. Try again later."

    error_note = ""
    if errors:
        failed = [e["source"] for e in errors]
        error_note = f"\n[Unavailable sources: {', '.join(failed)}]"

    return {"final_answer": answer + error_note}

graph_builder = StateGraph(State)
graph_builder.add_node("wiki", safe_search("Wikipedia"))
graph_builder.add_node("arxiv", safe_search("arXiv", should_fail=True))
graph_builder.add_node("news", safe_search("News"))
graph_builder.add_node("merge", smart_merge)

graph_builder.add_edge(START, "wiki")
graph_builder.add_edge(START, "arxiv")
graph_builder.add_edge(START, "news")
graph_builder.add_edge("wiki", "merge")
graph_builder.add_edge("arxiv", "merge")
graph_builder.add_edge("news", "merge")
graph_builder.add_edge("merge", END)

graph = graph_builder.compile()

result = graph.invoke({
    "query": "machine learning",
    "branch_results": [],
    "final_answer": "",
})
print(result["final_answer"])
# Expected output:
# Machine learning is a branch of artificial intelligence that allows...
# [Unavailable sources: arXiv]

The safe_search factory function wraps each source with try/except. The merge node knows how to work with partial results. The user gets value out of the sources that did work.

Error handling with the Functional API

In the Functional API, error handling is plain Python — try/except inside the @entrypoint:

from dotenv import load_dotenv
load_dotenv()

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

@task
def fetch_source_a(query: str) -> str:
    return f"Source A: data about {query}."

@task
def fetch_source_b(query: str) -> str:
    raise ConnectionError("Source B unavailable")

@task
def fetch_source_c(query: str) -> str:
    return f"Source C: news about {query}."

@task
def synthesize(query: str, results: list, errors: list) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    context = "\n".join(results)
    response = model.invoke(
        f"Synthesize about '{query}' (available sources: {len(results)}):\n\n{context}"
    )
    return {
        "answer": response.content,
        "sources_ok": len(results),
        "sources_failed": len(errors),
        "errors": errors,
    }

@entrypoint()
def resilient_research(query: str) -> dict:
    sources = {
        "A": fetch_source_a(query),
        "B": fetch_source_b(query),
        "C": fetch_source_c(query),
    }

    results = []
    errors = []

    for name, future in sources.items():
        try:
            data = future.result()
            results.append(f"[{name}] {data}")
        except Exception as e:
            errors.append(f"[{name}] {str(e)}")

    if not results:
        return {
            "answer": "Every source failed. Try again.",
            "sources_ok": 0,
            "sources_failed": len(errors),
            "errors": errors,
        }

    return synthesize(query, results, errors).result()

result = resilient_research.invoke("neural networks")
print(f"Sources OK: {result['sources_ok']}, Failed: {result['sources_failed']}")
if result["errors"]:
    print(f"Errors: {result['errors']}")
print(f"Answer: {result['answer'][:150]}...")
# Expected output:
# Sources OK: 2, Failed: 1
# Errors: ['[B] Source B unavailable']
# Answer: Neural networks are computational models inspired by...

The pattern is identical to the Graph API's: launch every source in parallel, collect with try/except, synthesize with whatever worked.

Combining retry + fallback + circuit breaker

The production-grade pattern combines all three techniques. For each external service:

  1. The circuit breaker decides whether it's worth trying
  2. Retry with backoff retries transient errors
  3. The fallback kicks in if everything fails
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    query: str
    result: str
    metadata: dict

ATTEMPT_COUNTER = 0

def resilient_call(
    service_name: str,
    cb_state: dict,
    max_retries: int = 3,
    base_delay: float = 1.0,
) -> dict:
    """Call with retry + backoff + circuit breaker."""
    if cb_state.get("state") == "open":
        elapsed = time.time() - cb_state.get("last_failure_time", 0)
        if elapsed < cb_state.get("cooldown_seconds", 30):
            return {
                "status": "circuit_open",
                "data": None,
                "cb_state": cb_state,
                "attempts": 0,
            }
        cb_state = {**cb_state, "state": "half-open"}

    global ATTEMPT_COUNTER
    last_error = None

    for attempt in range(max_retries):
        try:
            ATTEMPT_COUNTER += 1
            if ATTEMPT_COUNTER <= 2:
                raise ConnectionError(f"{service_name}: timeout on attempt #{ATTEMPT_COUNTER}")

            return {
                "status": "ok",
                "data": f"Data from {service_name} fetched on attempt {attempt + 1}",
                "cb_state": {**cb_state, "failure_count": 0, "state": "closed"},
                "attempts": attempt + 1,
            }

        except Exception as e:
            last_error = str(e)
            if attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
                time.sleep(min(delay, 0.1))

    new_failures = cb_state.get("failure_count", 0) + 1
    new_state = "open" if new_failures >= cb_state.get("max_failures", 5) else cb_state.get("state", "closed")

    return {
        "status": "error",
        "data": None,
        "error": last_error,
        "cb_state": {
            **cb_state,
            "failure_count": new_failures,
            "state": new_state,
            "last_failure_time": time.time(),
        },
        "attempts": max_retries,
    }

def search_with_resilience(state: State) -> dict:
    cb = state.get("metadata", {}).get("circuit_breaker", {
        "failure_count": 0, "max_failures": 5, "state": "closed",
        "last_failure_time": 0, "cooldown_seconds": 30,
    })

    call_result = resilient_call("search_api", cb, max_retries=3)

    if call_result["status"] == "ok":
        return {
            "result": call_result["data"],
            "metadata": {
                "source": "primary",
                "attempts": call_result["attempts"],
                "circuit_breaker": call_result["cb_state"],
            },
        }

    return {
        "result": "",
        "metadata": {
            "source": "none",
            "error": call_result.get("error", "circuit open"),
            "attempts": call_result["attempts"],
            "circuit_breaker": call_result.get("cb_state", cb),
        },
    }

def route_after_search(state: State) -> str:
    return "use_result" if state.get("result") else "use_fallback"

def use_fallback(state: State) -> dict:
    return {
        "result": "Cached result: general information available.",
        "metadata": {**state["metadata"], "source": "fallback"},
    }

def format_output(state: State) -> dict:
    return {}

graph_builder = StateGraph(State)
graph_builder.add_node("search", search_with_resilience)
graph_builder.add_node("fallback", use_fallback)
graph_builder.add_node("output", format_output)

graph_builder.add_edge(START, "search")
graph_builder.add_conditional_edges(
    "search", route_after_search,
    {"use_result": "output", "use_fallback": "fallback"}
)
graph_builder.add_edge("fallback", "output")
graph_builder.add_edge("output", END)

graph = graph_builder.compile()

ATTEMPT_COUNTER = 0
result = graph.invoke({
    "query": "AI agents",
    "result": "",
    "metadata": {},
})
print(f"Result: {result['result']}")
print(f"Source: {result['metadata']['source']}")
print(f"Attempts: {result['metadata']['attempts']}")
# Expected output:
# Result: Data from search_api fetched on attempt 3
# Source: primary
# Attempts: 3

The complete flow

Call to the service
  ↓
[Circuit breaker] → Is it open? → Yes → Immediate fallback
  ↓ No
[Retry 1] → Success? → Yes → Return the result
  ↓ No
[Retry 2 (delay * 2)] → Success? → Yes → Return the result
  ↓ No
[Retry 3 (delay * 4)] → Success? → Yes → Return the result
  ↓ No
[Update the circuit breaker] → Fallback

Structured logging: errors as data

Errors aren't just handled — they're recorded as structured data in the state for debugging, monitoring and continuous improvement:

import time
from typing import TypedDict, Annotated
import operator

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

def log_error(source: str, error: Exception, context: dict = None) -> dict:
    """Builds a structured error record."""
    return {
        "timestamp": time.time(),
        "source": source,
        "error_type": type(error).__name__,
        "error_message": str(error),
        "context": context or {},
    }

def node_with_logging(state: State) -> dict:
    try:
        raise TimeoutError("API response took > 10s")
    except Exception as e:
        error_entry = log_error(
            source="search_api",
            error=e,
            context={"query": state["query"], "attempt": 1},
        )
        return {
            "result": "fallback result",
            "error_log": [error_entry],
        }

What to include in the log

FieldWhy
timestampTo sort chronologically and spot patterns
sourceWhich service/node failed
error_typeConnectionError, TimeoutError, ValueError — so you can filter
error_messageThe detail of the error
contextQuery, attempt, parameters — so you can reproduce it

With Annotated[list[dict], operator.add] on error_log, every node can append errors and the state accumulates them. At the end of the run, you have a complete record of everything that failed.

Default responses: when everything fails

The last resort is a default response — a reasonable answer when literally nothing works:

def ultimate_fallback(state: State) -> dict:
    """When retry, fallback, and cache all fail."""
    error_count = len(state.get("error_log", []))
    sources_tried = set(e["source"] for e in state.get("error_log", []))

    return {
        "result": (
            f"We couldn't retrieve information right now. "
            f"{len(sources_tried)} source(s) were tried with {error_count} error(s). "
            f"Please try again in a few minutes."
        ),
        "metadata": {"source": "default", "degradation_level": "total"},
    }

A default response:

  • ✅ Never crashes — it always returns something
  • ✅ Is informative — it tells the user what happened
  • ✅ Suggests an action — "try again in a few minutes"
  • ❌ Doesn't make up data — it never fabricates an answer when it has no information

Troubleshooting

Problem 1: "My fallback never fires"

Symptom: The primary node raises an exception and the graph crashes instead of going to the fallback.

Cause: The exception escapes the node, or the node doesn't catch it and doesn't put it in the state.

Fix: The node must catch the exception internally and record the error in the state:

# ❌ The exception escapes the node
def my_node(state):
    raise ConnectionError("API down")  # Crashes the graph

# ✅ Exception caught, error in the state
def my_node(state):
    try:
        raise ConnectionError("API down")
    except Exception as e:
        return {"error": str(e), "result": ""}

Problem 2: "The circuit breaker never opens"

Symptom: The service keeps failing but the circuit breaker stays "closed".

Cause: The failure counter isn't updated correctly between graph invocations.

Fix: Make sure the circuit breaker state is passed between invocations:

cb = result["metadata"]["circuit_breaker"]
next_result = graph.invoke({..., "metadata": {"circuit_breaker": cb}})

Problem 3: "Errors from parallel branches get lost"

Symptom: The merge node has no information about which branches failed.

Cause: The branches raise exceptions instead of returning results with a status.

Fix: Each branch must catch its own errors:

def safe_branch(state):
    try:
        result = risky_operation()
        return {"results": [{"status": "ok", "data": result}]}
    except Exception as e:
        return {"results": [{"status": "error", "error": str(e)}]}

Problem 4: "Retry without backoff causes rate limiting"

Symptom: The retries are so fast the service blocks you.

Cause: Retry with no delay = hammering the service = more rate limiting.

Fix: Always use exponential backoff with jitter:

import time
import random

for attempt in range(max_retries):
    try:
        return call_api()
    except Exception:
        if attempt < max_retries - 1:
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            time.sleep(delay)

Problem 5: "The graph returns an empty result with no explanation"

Symptom: Everything fails silently and the user gets an empty string.

Cause: There's no default response for when every source fails.

Fix: Always add a "last resort" path:

def merge(state):
    ok = [r for r in state["results"] if r["status"] == "ok"]
    if not ok:
        return {"final": "We couldn't get any results. Try again later."}
    ...

Exercises

Exercise 1: Simple fallback with two sources (Easy)

Create a graph with a "primary" node that always fails and a "fallback" node that returns a cached result. Use a conditional edge to route to the fallback when primary fails. Show which source was used.

See solution 
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 State(TypedDict):
    query: str
    result: str
    error: str
    source: str

def primary(state: State) -> dict:
    try:
        raise ConnectionError("Primary service unavailable")
    except Exception as e:
        return {"error": str(e), "result": "", "source": ""}

def route(state: State) -> str:
    return "fallback" if state.get("error") else "output"

def fallback(state: State) -> dict:
    return {
        "result": f"Cached result for '{state['query']}': general information available.",
        "error": "",
        "source": "cache",
    }

def output(state: State) -> dict:
    if not state.get("source"):
        return {"source": "primary"}
    return {}

graph_builder = StateGraph(State)
graph_builder.add_node("primary", primary)
graph_builder.add_node("fallback", fallback)
graph_builder.add_node("output", output)

graph_builder.add_edge(START, "primary")
graph_builder.add_conditional_edges("primary", route, {"fallback": "fallback", "output": "output"})
graph_builder.add_edge("fallback", "output")
graph_builder.add_edge("output", END)

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

result = graph.invoke({"query": "Python asyncio", "result": "", "error": "", "source": ""})
print(f"Source: {result['source']}")
print(f"Result: {result['result']}")
# Expected output:
# Source: cache
# Result: Cached result for 'Python asyncio': general information available.

Exercise 2: Graceful degradation with 3 sources (Easy)

Create a graph with 3 parallel branches (search sources). One of them fails. The merge node has to use the 2 successful sources and add a note about the source that failed.

See solution 
from dotenv import load_dotenv
load_dotenv()

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

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

def source_a(state: State) -> dict:
    return {"results": [{"source": "A", "status": "ok", "data": f"Source A: info about {state['query']}."}]}

def source_b(state: State) -> dict:
    return {"results": [{"source": "B", "status": "error", "data": None, "error": "Timeout"}]}

def source_c(state: State) -> dict:
    return {"results": [{"source": "C", "status": "ok", "data": f"Source C: technical data about {state['query']}."}]}

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"]

    model = init_chat_model("openai:gpt-4.1-mini")
    context = "\n".join(r["data"] for r in ok)
    response = model.invoke(f"Synthesize about '{state['query']}':\n\n{context}")

    note = ""
    if errors:
        failed = ", ".join(r["source"] for r in errors)
        note = f"\n\n[Note: unavailable sources: {failed}. Partial result with {len(ok)}/{len(state['results'])} sources.]"

    return {"final_answer": response.content + note}

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

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

graph = graph_builder.compile()

result = graph.invoke({"query": "transformers", "results": [], "final_answer": ""})
print(result["final_answer"])
# Expected output:
# Transformers are a deep learning architecture based on...
#
# [Note: unavailable sources: B. Partial result with 2/3 sources.]

Exercise 3: A 3-level fallback chain (Medium)

Implement a chain of 3 sources where each one fails: premium API (rate limited) → free API (timeout) → local cache (always works). The state must record every attempt. At the end, show the full attempt history.

See solution 
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 State(TypedDict):
    query: str
    result: str
    source: str
    attempt_log: list

def try_premium(state: State) -> dict:
    log = state.get("attempt_log", [])
    try:
        raise ConnectionError("Premium API: 429 Rate Limit Exceeded")
    except Exception as e:
        return {
            "result": "",
            "source": "",
            "attempt_log": log + [{"source": "premium", "status": "error", "detail": str(e)}],
        }

def route_premium(state: State) -> str:
    return "try_free" if not state.get("result") else "done"

def try_free(state: State) -> dict:
    try:
        raise TimeoutError("Free API: timeout after 15 seconds")
    except Exception as e:
        return {
            "result": "",
            "source": "",
            "attempt_log": state["attempt_log"] + [{"source": "free", "status": "error", "detail": str(e)}],
        }

def route_free(state: State) -> str:
    return "try_cache" if not state.get("result") else "done"

def try_cache(state: State) -> dict:
    return {
        "result": f"[Cache] Previously stored information about '{state['query']}'.",
        "source": "cache",
        "attempt_log": state["attempt_log"] + [{"source": "cache", "status": "ok", "detail": "Cache hit"}],
    }

def done(state: State) -> dict:
    return {}

graph_builder = StateGraph(State)
graph_builder.add_node("premium", try_premium)
graph_builder.add_node("free", try_free)
graph_builder.add_node("cache", try_cache)
graph_builder.add_node("done", done)

graph_builder.add_edge(START, "premium")
graph_builder.add_conditional_edges("premium", route_premium, {"try_free": "free", "done": "done"})
graph_builder.add_conditional_edges("free", route_free, {"try_cache": "cache", "done": "done"})
graph_builder.add_edge("cache", "done")
graph_builder.add_edge("done", END)

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

result = graph.invoke({
    "query": "LangGraph checkpointing",
    "result": "", "source": "", "attempt_log": [],
})
print(f"Final source: {result['source']}")
print(f"Result: {result['result']}")
print(f"\nAttempt history:")
for i, attempt in enumerate(result["attempt_log"], 1):
    icon = "✅" if attempt["status"] == "ok" else "❌"
    print(f"  {i}. {icon} {attempt['source']}: {attempt['detail']}")
# Expected output:
# Final source: cache
# Result: [Cache] Previously stored information about 'LangGraph checkpointing'.
#
# Attempt history:
#   1. ❌ premium: Premium API: 429 Rate Limit Exceeded
#   2. ❌ free: Free API: timeout after 15 seconds
#   3. ✅ cache: Cache hit

Exercise 4: A circuit breaker per service (Medium)

Implement a system with 2 services, each with its own circuit breaker. Simulate service A failing 3 times in a row (opening its circuit breaker) while service B works normally. Show the state of each circuit breaker after the calls.

See solution 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    query: str
    result_a: str
    result_b: str
    circuit_breakers: dict

def init_cb() -> dict:
    return {"failure_count": 0, "max_failures": 3, "state": "closed", "last_failure_time": 0}

SERVICE_A_CALLS = 0

def call_service_a(state: State) -> dict:
    cbs = state.get("circuit_breakers", {"a": init_cb(), "b": init_cb()})
    cb_a = cbs["a"]

    if cb_a["state"] == "open":
        return {
            "result_a": "[CIRCUIT OPEN] Service A unavailable",
            "circuit_breakers": cbs,
        }

    global SERVICE_A_CALLS
    SERVICE_A_CALLS += 1

    try:
        if SERVICE_A_CALLS <= 4:
            raise ConnectionError(f"Service A: error #{SERVICE_A_CALLS}")
        return {
            "result_a": "Service A: data fetched",
            "circuit_breakers": {**cbs, "a": {**cb_a, "failure_count": 0, "state": "closed"}},
        }
    except Exception as e:
        new_count = cb_a["failure_count"] + 1
        new_state = "open" if new_count >= cb_a["max_failures"] else "closed"
        return {
            "result_a": f"[ERROR] {str(e)}",
            "circuit_breakers": {
                **cbs,
                "a": {**cb_a, "failure_count": new_count, "state": new_state, "last_failure_time": time.time()},
            },
        }

def call_service_b(state: State) -> dict:
    cbs = state.get("circuit_breakers", {"a": init_cb(), "b": init_cb()})
    return {
        "result_b": f"Service B: data about '{state['query']}' fetched successfully.",
        "circuit_breakers": {**cbs, "b": {**cbs.get("b", init_cb()), "failure_count": 0, "state": "closed"}},
    }

graph_builder = StateGraph(State)
graph_builder.add_node("service_a", call_service_a)
graph_builder.add_node("service_b", call_service_b)

graph_builder.add_edge(START, "service_a")
graph_builder.add_edge("service_a", "service_b")
graph_builder.add_edge("service_b", END)

graph = graph_builder.compile()

SERVICE_A_CALLS = 0
cbs = {"a": init_cb(), "b": init_cb()}

for i in range(5):
    result = graph.invoke({
        "query": "AI safety",
        "result_a": "", "result_b": "",
        "circuit_breakers": cbs,
    })
    cbs = result["circuit_breakers"]
    print(f"Round {i+1}:")
    print(f"  A: {result['result_a']}")
    print(f"  B: {result['result_b'][:50]}")
    print(f"  CB-A: {cbs['a']['state']} ({cbs['a']['failure_count']} failures)")
    print(f"  CB-B: {cbs['b']['state']} ({cbs['b']['failure_count']} failures)")
# Expected output:
# Round 1:
#   A: [ERROR] Service A: error #1
#   B: Service B: data about 'AI safety' fetched successf
#   CB-A: closed (1 failures)
#   CB-B: closed (0 failures)
# Round 2:
#   A: [ERROR] Service A: error #2
#   B: Service B: data about 'AI safety' fetched successf
#   CB-A: closed (2 failures)
#   CB-B: closed (0 failures)
# Round 3:
#   A: [ERROR] Service A: error #3
#   B: Service B: data about 'AI safety' fetched successf
#   CB-A: open (3 failures)
#   CB-B: closed (0 failures)
# Round 4:
#   A: [CIRCUIT OPEN] Service A unavailable
#   B: Service B: data about 'AI safety' fetched successf
#   CB-A: open (3 failures)
#   CB-B: closed (0 failures)
# Round 5:
#   A: [CIRCUIT OPEN] Service A unavailable
#   B: Service B: data about 'AI safety' fetched successf
#   CB-A: open (3 failures)
#   CB-B: closed (0 failures)

Service A's circuit breaker opens after 3 failures. From round 4 on, it doesn't even try to call the service — it saves time and doesn't add unnecessary load. Service B works independently.

Exercise 5: Complete error handling with the Functional API (Medium)

Implement a research pipeline with 4 sources using the Functional API. Two sources fail. Use try/except to handle errors, accumulate an error log, and synthesize with the successful sources. Return a dict with: answer, sources_ok, sources_failed, error_log.

See solution 
from dotenv import load_dotenv
load_dotenv()

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

@task
def search_wiki(query: str) -> str:
    return f"Wikipedia: encyclopedic information about {query}."

@task
def search_arxiv(query: str) -> str:
    raise ConnectionError("arXiv: 503 Service Unavailable")

@task
def search_news(query: str) -> str:
    return f"News: recent news about {query}."

@task
def search_blogs(query: str) -> str:
    raise TimeoutError("Blogs API: timeout after 20s")

@task
def synthesize(query: str, sources: list) -> str:
    model = init_chat_model("openai:gpt-4.1-mini")
    context = "\n".join(sources)
    return model.invoke(
        f"Synthesize about '{query}' with these sources:\n\n{context}"
    ).content

@entrypoint()
def resilient_pipeline(query: str) -> dict:
    search_tasks = {
        "wiki": search_wiki(query),
        "arxiv": search_arxiv(query),
        "news": search_news(query),
        "blogs": search_blogs(query),
    }

    sources = []
    error_log = []

    for name, future in search_tasks.items():
        try:
            data = future.result()
            sources.append(f"[{name}] {data}")
        except Exception as e:
            error_log.append({
                "source": name,
                "error_type": type(e).__name__,
                "message": str(e),
            })

    if not sources:
        return {
            "answer": "Every source failed. Try again later.",
            "sources_ok": 0,
            "sources_failed": len(error_log),
            "error_log": error_log,
        }

    answer = synthesize(query, sources).result()

    return {
        "answer": answer,
        "sources_ok": len(sources),
        "sources_failed": len(error_log),
        "error_log": error_log,
    }

result = resilient_pipeline.invoke("prompt engineering techniques")
print(f"Sources OK: {result['sources_ok']}")
print(f"Failed sources: {result['sources_failed']}")
print(f"\nErrors:")
for err in result["error_log"]:
    print(f"  ❌ {err['source']}: [{err['error_type']}] {err['message']}")
print(f"\nAnswer: {result['answer'][:200]}...")
# Expected output:
# Sources OK: 2
# Failed sources: 2
#
# Errors:
#   ❌ arxiv: [ConnectionError] arXiv: 503 Service Unavailable
#   ❌ blogs: [TimeoutError] Blogs API: timeout after 20s
#
# Answer: Prompt engineering techniques include zero-shot, few-shot, and chain-of-thought...

Exercise 6: Complete retry + fallback + circuit breaker system (Advanced)

Implement a graph that combines all three patterns to call an API. The flow: (1) the circuit breaker checks whether it's worth trying, (2) retry with exponential backoff up to 3 attempts, (3) fallback to cache if everything fails. Set up the simulation so the API fails the first 2 times and works on the third. Show the complete process log.

See solution 
from dotenv import load_dotenv
load_dotenv()

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

class State(TypedDict):
    query: str
    result: str
    source: str
    execution_log: list
    circuit_breaker: dict

GLOBAL_ATTEMPTS = 0

def init_cb() -> dict:
    return {
        "failure_count": 0, "max_failures": 5,
        "state": "closed", "last_failure_time": 0, "cooldown_seconds": 30,
    }

def attempt_with_retry(state: State) -> dict:
    cb = state.get("circuit_breaker", init_cb())
    log = list(state.get("execution_log", []))

    if cb["state"] == "open":
        elapsed = time.time() - cb.get("last_failure_time", 0)
        if elapsed < cb["cooldown_seconds"]:
            log.append({"step": "circuit_breaker", "action": "blocked", "detail": "Circuit is OPEN"})
            return {"result": "", "source": "", "execution_log": log, "circuit_breaker": cb}
        cb = {**cb, "state": "half-open"}
        log.append({"step": "circuit_breaker", "action": "half-open", "detail": "Probing with one call"})

    global GLOBAL_ATTEMPTS
    max_retries = 3
    base_delay = 0.01

    for attempt in range(max_retries):
        GLOBAL_ATTEMPTS += 1
        try:
            if GLOBAL_ATTEMPTS <= 2:
                raise ConnectionError(f"API: transient error (global attempt #{GLOBAL_ATTEMPTS})")

            log.append({
                "step": "retry",
                "action": "success",
                "detail": f"Attempt {attempt + 1}/{max_retries}: success",
            })
            cb_updated = {**cb, "failure_count": 0, "state": "closed"}
            return {
                "result": f"Data fetched successfully (attempt {attempt + 1})",
                "source": "api",
                "execution_log": log,
                "circuit_breaker": cb_updated,
            }

        except Exception as e:
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.01)
            log.append({
                "step": "retry",
                "action": "failed",
                "detail": f"Attempt {attempt + 1}/{max_retries}: {str(e)} (delay: {delay:.3f}s)",
            })
            if attempt < max_retries - 1:
                time.sleep(delay)

    new_failures = cb["failure_count"] + 1
    new_state = "open" if new_failures >= cb["max_failures"] else cb["state"]
    cb_updated = {**cb, "failure_count": new_failures, "state": new_state, "last_failure_time": time.time()}
    log.append({"step": "retry", "action": "exhausted", "detail": f"All retries exhausted"})

    return {"result": "", "source": "", "execution_log": log, "circuit_breaker": cb_updated}

def route_after_retry(state: State) -> str:
    return "output" if state.get("result") else "fallback"

def fallback_cache(state: State) -> dict:
    log = list(state.get("execution_log", []))
    log.append({"step": "fallback", "action": "cache_hit", "detail": "Using the cached result"})
    return {
        "result": f"[Cache] Stored information about '{state['query']}'.",
        "source": "cache",
        "execution_log": log,
    }

def output(state: State) -> dict:
    return {}

graph_builder = StateGraph(State)
graph_builder.add_node("retry", attempt_with_retry)
graph_builder.add_node("fallback", fallback_cache)
graph_builder.add_node("output", output)

graph_builder.add_edge(START, "retry")
graph_builder.add_conditional_edges("retry", route_after_retry, {"output": "output", "fallback": "fallback"})
graph_builder.add_edge("fallback", "output")
graph_builder.add_edge("output", END)

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

GLOBAL_ATTEMPTS = 0
result = graph.invoke({
    "query": "error handling patterns",
    "result": "", "source": "",
    "execution_log": [],
    "circuit_breaker": init_cb(),
})

print(f"Source: {result['source']}")
print(f"Result: {result['result']}")
print(f"Circuit breaker: {result['circuit_breaker']['state']}")
print(f"\nExecution log:")
for entry in result["execution_log"]:
    icons = {"success": "✅", "failed": "❌", "exhausted": "⚠️", "cache_hit": "📦", "blocked": "🚫", "half-open": "🔄"}
    icon = icons.get(entry["action"], "•")
    print(f"  {icon} [{entry['step']}] {entry['detail']}")
# Expected output:
# Source: api
# Result: Data fetched successfully (attempt 3)
# Circuit breaker: closed
#
# Execution log:
#   ❌ [retry] Attempt 1/3: API: transient error (global attempt #1) (delay: 0.015s)
#   ❌ [retry] Attempt 2/3: API: transient error (global attempt #2) (delay: 0.025s)
#   ✅ [retry] Attempt 3/3: success

This exercise shows the complete production-grade pattern. The API fails twice (transient errors) and works on the third attempt. If all 3 retries had failed, the fallback to cache would have fired automatically.

Summary

In this capsule you learned:

  • Error handling is fundamental engineering — not an afterthought. A system that crashes 1% of the time is not production-ready
  • Fallback nodes provide a plan B when the main path fails — implemented with conditional edges that check the state's error field
  • Fallback chains (A → B → C → default) try multiple sources in order of preference until one works
  • Graceful degradation lets you continue with partial results when some parallel branches fail — each branch catches its errors and the merge node works with whatever succeeded
  • The circuit breaker protects services that are down: after N failures, it stops trying for a cooldown period — it keeps you from hammering a service that's already in trouble
  • The production-grade pattern combines retry + fallback + circuit breaker: the circuit breaker decides whether to try, the retry handles transient errors, the fallback covers you when everything fails
  • Structured logging records errors as data in the state — timestamp, source, error_type, context — for debugging and monitoring
  • Default responses are the last resort: they never return empty, they always tell the user what happened and suggest an action

Next capsule: Production patterns — per-node timeouts, internal rate limiting, and how to monitor a LangGraph system in real production.

Additional resources

  1. LangGraph Error Handling — Error handling patterns in LangGraph agents
  2. Retry patterns in distributed systems — AWS: Timeouts, Retries and Backoff with Jitter
  3. Circuit Breaker Pattern — Martin Fowler: Circuit Breaker
  4. LangGraph Branching — Fan-out/fan-in with error handling
  5. LangGraph Functional API — Native error handling with try/except
  6. Graceful Degradation Patterns — Microsoft: the retry pattern and degradation
  7. Python logging best practices — Structured logging in Python

Module 7 — LangChain & LangGraph: From Chains to Agents

← PreviousMap-Reduce and Deferred NodesNext →Production Patterns