Módulo 7: Flujos Avanzados

Error Handling Avanzado

Descripción de la cápsula

Tu agente llama a 3 APIs externas. Una está caída. Sin error handling, todo el sistema crashea — el usuario no recibe nada. Con error handling, 2 de 3 fuentes siguen funcionando — el usuario recibe un resultado parcial pero útil. Esa es la diferencia entre un prototipo y un sistema de producción.

Error handling no es un apéndice. No es algo que agregas al final "si queda tiempo." Es ingeniería fundamental — tan crítico como la lógica de negocio. Un sistema que funciona el 99% del tiempo pero crashea catastróficamente el 1% restante no es production-ready. En el mundo real, las APIs fallan, los rate limits se activan, las conexiones se cortan, y los LLMs alucinan. Tu sistema debe sobrevivir a todo eso.

En esta cápsula aprenderás los patrones que hacen la diferencia: fallback nodes para cuando el camino principal falla, graceful degradation para continuar con resultados parciales, circuit breaker para no bombardear un servicio caído, y cómo combinar retry + fallback + circuit breaker en el patrón production-grade completo.

El problema concreto

Tu Research Assistant busca información en 3 fuentes: Wikipedia, arXiv y un API de noticias. Cada fuente es una API externa que puede fallar:

Escenario 1: Todo funciona → 3/3 fuentes responden → resultado completo
Escenario 2: arXiv timeout → 2/3 fuentes responden → resultado parcial (útil)
Escenario 3: Rate limit en todas → 0/3 fuentes responden → ???

Sin error handling, los escenarios 2 y 3 crashean el sistema. Con error handling:

Escenario 2: el sistema continúa con Wikipedia y noticias, nota que arXiv falló, e informa al usuario que el resultado es parcial
Escenario 3: el sistema activa un fallback (caché, resultado por defecto, o un mensaje explicativo), y no crashea

El usuario siempre recibe algo útil. Eso es graceful degradation.

Fallback nodes: el plan B del grafo

Un fallback node es un nodo que se activa cuando el camino principal falla. Se implementa con un conditional edge que evalúa si hubo 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:
    """Búsqueda principal — puede fallar."""
    try:
        raise ConnectionError("API de búsqueda principal no disponible")
    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: usa resultados cacheados o una fuente alternativa."""
    cached = "Python es un lenguaje de programación de alto nivel creado por 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"Genera una respuesta basada en esta información (fuente: {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="¿Qué es Python?")],
    "search_result": "", "error": "", "source_used": "",
})
print(f"Fuente usada: {result['source_used']}")
print(f"Respuesta: {result['messages'][-1].content[:120]}...")
# Output esperado:
# Fuente usada: cache
# Respuesta: Python es un lenguaje de programación de alto nivel, creado por Guido van Rossum...

Anatomía del fallback

Nodo primario intenta la operación y captura errores en el estado (error field)
Routing function evalúa si hay error → "fallback" o éxito → "process"
Nodo fallback provee un resultado alternativo y limpia el error
Ambos caminos convergen en "process" — el nodo de procesamiento no sabe (ni le importa) de dónde vino el resultado

Fallback chain: A → B → C → default

Cuando tienes múltiples fuentes alternativas, encadénalas:

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:
    """Fuente A: API premium (costosa pero completa)."""
    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:
    """Fuente B: API gratuita (limitada pero funcional)."""
    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:
    """Fuente C: caché local (datos potencialmente desactualizados)."""
    attempts = state.get("attempts", [])
    return {
        "result": "Resultado del caché: LangGraph es un framework para construir agentes con grafos de estado.",
        "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": "¿Qué es LangGraph?",
    "result": "", "source": "", "attempts": [],
})
print(f"Fuente final: {result['source']}")
print(f"Resultado: {result['result']}")
print(f"Intentos:")
for a in result["attempts"]:
    status_icon = "✅" if a["status"] == "ok" else "❌"
    print(f"  {status_icon} {a['source']}: {a['detail']}")
# Output esperado:
# Fuente final: cache
# Resultado: Resultado del caché: LangGraph es un framework para construir agentes con grafos de estado.
# Intentos:
#   ❌ premium_api: Premium API: rate limit exceeded
#   ❌ free_api: Free API: connection timeout after 10s
#   ✅ cache: Hit

El grafo intenta A, luego B, luego C. El campo attempts registra cada intento para debugging. El usuario recibe el resultado de la primera fuente que funcione.

Graceful degradation: continuar con resultados parciales

En branching paralelo, cuando uno de N branches falla, los otros pueden haber tenido éxito. En vez de descartar todo, usa lo que sí funcionó:

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: información enciclopédica sobre '{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: últimas noticias sobre '{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": "No se pudo obtener información de ninguna fuente.",
        }]}

    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\nNota: {len(failures)} fuente(s) no disponible(s): {failed_sources}. "
        degradation_note += f"Resultado basado en {len(successes)}/{len(state['results'])} fuentes."

    response = model.invoke(
        f"Sintetiza esta información sobre '{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"Fuentes usadas: {synthesis['sources_used']}/{synthesis['sources_used'] + synthesis['sources_failed']}")
print(f"Respuesta: {synthesis['data'][:150]}...")
# Output esperado:
# Status: degraded
# Fuentes usadas: 2/3
# Respuesta: Los modelos de lenguaje grandes (LLMs) son sistemas de IA basados en...

Claves del graceful degradation

Cada nodo captura sus propios errores — retorna {"status": "error"} en vez de lanzar excepciones
El nodo de síntesis separa éxitos de fallos — trabaja solo con los éxitos
El usuario recibe información sobre la degradación — sabe que el resultado es parcial
Nunca retorna vacío — incluso si todas las fuentes fallan, retorna un mensaje explicativo

Circuit breaker: proteger servicios caídos

Cuando una API falla repetidamente, seguir intentando es contraproducente: desperdicias tiempo, generas carga innecesaria en el servicio caído, y retrasas la respuesta al usuario. El circuit breaker soluciona esto:

Estado CLOSED (normal):     → Llamadas pasan normalmente
Estado OPEN (servicio caído): → Llamadas se rechazan inmediatamente, sin intentar
Estado HALF-OPEN (probando):  → Permite UNA llamada de prueba para ver si se recuperó

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:
    """Evalúa si el circuit breaker permite la llamada."""
    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:
    """Registra una llamada exitosa."""
    return {**cb, "failure_count": 0, "state": "closed"}

def record_failure(cb: dict) -> dict:
    """Registra una llamada fallida."""
    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] Servicio no disponible, usando caché. Fallos: {cb['failure_count']}",
            "circuit_breaker": cb,
        }

    global CALL_COUNT
    CALL_COUNT += 1

    try:
        if CALL_COUNT <= 3:
            raise ConnectionError(f"API error en intento #{CALL_COUNT}")
        return {
            "result": f"[OK] Datos obtenidos exitosamente en intento #{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"Llamada {i+1}: {result['result']} | Circuit: {cb['state']} ({cb['failure_count']} fallos)")
# Output esperado:
# Llamada 1: [ERROR] API error en intento #1 | Circuit: closed (1 fallos)
# Llamada 2: [ERROR] API error en intento #2 | Circuit: closed (2 fallos)
# Llamada 3: [ERROR] API error en intento #3 | Circuit: open (3 fallos)
# Llamada 4: [CIRCUIT OPEN] Servicio no disponible, usando caché. Fallos: 3 | Circuit: open (3 fallos)
# Llamada 5: [CIRCUIT OPEN] Servicio no disponible, usando caché. Fallos: 3 | Circuit: open (3 fallos)
# Llamada 6: [CIRCUIT OPEN] Servicio no disponible, usando caché. Fallos: 3 | Circuit: open (3 fallos)

Cómo integrar el circuit breaker en el estado

El circuit breaker vive en el estado del grafo como un dict. Cada nodo que llama a una API externa:

Lee el circuit breaker del estado
Verifica si está allowed
Si no, retorna un resultado fallback inmediatamente
Si sí, intenta la llamada
Actualiza el circuit breaker (éxito o fallo)

En un sistema con múltiples APIs, mantén un circuit breaker por servicio:

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

Error handling en branches paralelos

Cuando ejecutas branches en paralelo (fan-out), cada branch puede fallar independientemente. El patrón: cada branch maneja sus propios errores y retorna un resultado con status. El nodo de merge filtra:

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 que crea nodos de búsqueda con error handling interno."""
    def node(state: State) -> dict:
        try:
            if should_fail:
                raise ConnectionError(f"{source_name}: servicio no disponible")
            return {"branch_results": [{
                "source": source_name,
                "status": "ok",
                "data": f"Resultados de {source_name} sobre '{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"Sintetiza sobre '{state['query']}':\n\n{context}"
        )
        answer = response.content
    else:
        answer = "No se pudo obtener información. Intenta de nuevo más tarde."

    error_note = ""
    if errors:
        failed = [e["source"] for e in errors]
        error_note = f"\n[Fuentes no disponibles: {', '.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"])
# Output esperado:
# Machine learning es una rama de la inteligencia artificial que permite...
# [Fuentes no disponibles: arXiv]

La función factory safe_search envuelve cada fuente con try/except. El nodo de merge sabe trabajar con resultados parciales. El usuario recibe valor de las fuentes que sí funcionaron.

Error handling con Functional API

En la Functional API, el error handling es Python puro — try/except dentro del @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: datos sobre {query}."

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

@task
def fetch_source_c(query: str) -> str:
    return f"Source C: noticias sobre {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"Sintetiza sobre '{query}' (fuentes disponibles: {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": "Todas las fuentes fallaron. Intenta de nuevo.",
            "sources_ok": 0,
            "sources_failed": len(errors),
            "errors": errors,
        }

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

result = resilient_research.invoke("neural networks")
print(f"Fuentes OK: {result['sources_ok']}, Fallidas: {result['sources_failed']}")
if result["errors"]:
    print(f"Errores: {result['errors']}")
print(f"Respuesta: {result['answer'][:150]}...")
# Output esperado:
# Fuentes OK: 2, Fallidas: 1
# Errores: ['[B] Source B no disponible']
# Respuesta: Las redes neuronales son modelos computacionales inspirados en...

El patrón es idéntico al de la Graph API: lanza todas las fuentes en paralelo, recolecta con try/except, sintetiza con lo que funcionó.

Combinando retry + fallback + circuit breaker

El patrón production-grade combina las tres técnicas. Para cada servicio externo:

Circuit breaker decide si vale la pena intentar
Retry con backoff reintenta errores transitorios
Fallback se activa si todo falla

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:
    """Llamada con 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 en intento #{ATTEMPT_COUNTER}")

            return {
                "status": "ok",
                "data": f"Datos de {service_name} obtenidos en intento {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": "Resultado del caché: información general disponible.",
        "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"Resultado: {result['result']}")
print(f"Fuente: {result['metadata']['source']}")
print(f"Intentos: {result['metadata']['attempts']}")
# Output esperado:
# Resultado: Datos de search_api obtenidos en intento 3
# Fuente: primary
# Intentos: 3

El flujo completo

Llamada al servicio
  ↓
[Circuit breaker] → ¿Está open? → Sí → Fallback inmediato
  ↓ No
[Retry 1] → ¿Éxito? → Sí → Retorna resultado
  ↓ No
[Retry 2 (delay * 2)] → ¿Éxito? → Sí → Retorna resultado
  ↓ No
[Retry 3 (delay * 4)] → ¿Éxito? → Sí → Retorna resultado
  ↓ No
[Actualizar circuit breaker] → Fallback

Logging estructurado: errores como datos

Los errores no solo se manejan — se registran como datos estructurados en el estado para debugging, monitoring y mejora continua:

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:
    """Genera un registro de error estructurado."""
    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],
        }

Qué incluir en el log

Campo	Por qué
`timestamp`	Para ordenar cronológicamente y detectar patrones
`source`	Qué servicio/nodo falló
`error_type`	`ConnectionError`, `TimeoutError`, `ValueError` — para filtrar
`error_message`	El detalle del error
`context`	Query, intento, parámetros — para reproducir

Con Annotated[list[dict], operator.add] en error_log, cada nodo puede agregar errores y el estado los acumula. Al final de la ejecución, tienes un registro completo de todo lo que falló.

Default responses: cuando todo falla

El último recurso es un default response — una respuesta razonable cuando literalmente nada funciona:

def ultimate_fallback(state: State) -> dict:
    """Cuando retry, fallback, y caché fallan."""
    error_count = len(state.get("error_log", []))
    sources_tried = set(e["source"] for e in state.get("error_log", []))

    return {
        "result": (
            f"No pudimos obtener información en este momento. "
            f"Se intentaron {len(sources_tried)} fuente(s) con {error_count} error(es). "
            f"Por favor intenta de nuevo en unos minutos."
        ),
        "metadata": {"source": "default", "degradation_level": "total"},
    }

Un default response:

✅ Nunca crashea — siempre retorna algo
✅ Es informativo — le dice al usuario qué pasó
✅ Sugiere acción — "intenta de nuevo en unos minutos"
❌ No inventa datos — nunca genera una respuesta falsa cuando no tiene información

Troubleshooting

Problema 1: "Mi fallback nunca se activa"

Síntoma: El nodo primario lanza una excepción y el grafo crashea en vez de ir al fallback.

Causa: La excepción se lanza fuera del nodo, o el nodo no la captura y no la pone en el estado.

Solución: El nodo debe capturar la excepción internamente y registrar el error en el estado:

# ❌ Excepción escapa del nodo
def my_node(state):
    raise ConnectionError("API down")  # Crashea el grafo

# ✅ Excepción capturada, error en estado
def my_node(state):
    try:
        raise ConnectionError("API down")
    except Exception as e:
        return {"error": str(e), "result": ""}

Problema 2: "El circuit breaker nunca se abre"

Síntoma: El servicio falla continuamente pero el circuit breaker sigue en "closed".

Causa: El contador de fallos no se actualiza correctamente entre invocaciones del grafo.

Solución: Asegúrate de que el circuit breaker state se pasa entre invocaciones:

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

Problema 3: "Los errores de branches paralelos se pierden"

Síntoma: El nodo de merge no tiene información sobre qué branches fallaron.

Causa: Los branches lanzan excepciones en vez de retornar resultados con status.

Solución: Cada branch debe capturar sus propios errores:

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)}]}

Problema 4: "Retry sin backoff causa rate limiting"

Síntoma: Las retries son tan rápidas que el servicio te bloquea.

Causa: Retry sin delay = bombardeo al servicio = más rate limiting.

Solución: Siempre usa backoff exponencial con 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)

Problema 5: "El grafo retorna un resultado vacío sin explicación"

Síntoma: Todo falla silenciosamente y el usuario recibe un string vacío.

Causa: No hay un default response cuando todas las fuentes fallan.

Solución: Agrega siempre un camino de "último recurso":

def merge(state):
    ok = [r for r in state["results"] if r["status"] == "ok"]
    if not ok:
        return {"final": "No pudimos obtener resultados. Intenta más tarde."}
    ...

Ejercicios

Ejercicio 1: Fallback simple con dos fuentes (Fácil)

Crea un grafo con un nodo "primary" que siempre falla y un nodo "fallback" que retorna un resultado cacheado. Usa un conditional edge para rutear al fallback cuando primary falla. Muestra qué fuente se usó.

Ver solución

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("Servicio principal no disponible")
    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"Resultado cacheado para '{state['query']}': información general disponible.",
        "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"Fuente: {result['source']}")
print(f"Resultado: {result['result']}")
# Output esperado:
# Fuente: cache
# Resultado: Resultado cacheado para 'Python asyncio': información general disponible.

Ejercicio 2: Graceful degradation con 3 fuentes (Fácil)

Crea un grafo con 3 branches paralelos (fuentes de búsqueda). Una de ellas falla. El nodo de merge debe usar las 2 fuentes exitosas y agregar una nota sobre la fuente que falló.

Ver solución

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 sobre {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: datos técnicos sobre {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"Sintetiza sobre '{state['query']}':\n\n{context}")

    note = ""
    if errors:
        failed = ", ".join(r["source"] for r in errors)
        note = f"\n\n[Nota: fuentes no disponibles: {failed}. Resultado parcial con {len(ok)}/{len(state['results'])} fuentes.]"

    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"])
# Output esperado:
# Los transformers son una arquitectura de deep learning basada en...
#
# [Nota: fuentes no disponibles: B. Resultado parcial con 2/3 fuentes.]

Ejercicio 3: Fallback chain de 3 niveles (Medio)

Implementa una cadena de 3 fuentes donde cada una falla: premium API (rate limited) → free API (timeout) → caché local (siempre funciona). El estado debe registrar cada intento. Al final, muestra el historial completo de intentos.

Ver solución

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"[Caché] Información previamente almacenada sobre '{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"Fuente final: {result['source']}")
print(f"Resultado: {result['result']}")
print(f"\nHistorial de intentos:")
for i, attempt in enumerate(result["attempt_log"], 1):
    icon = "✅" if attempt["status"] == "ok" else "❌"
    print(f"  {i}. {icon} {attempt['source']}: {attempt['detail']}")
# Output esperado:
# Fuente final: cache
# Resultado: [Caché] Información previamente almacenada sobre 'LangGraph checkpointing'.
#
# Historial de intentos:
#   1. ❌ premium: Premium API: 429 Rate Limit Exceeded
#   2. ❌ free: Free API: timeout after 15 seconds
#   3. ✅ cache: Cache hit

Ejercicio 4: Circuit breaker por servicio (Medio)

Implementa un sistema con 2 servicios, cada uno con su propio circuit breaker. Simula que el servicio A falla 3 veces seguidas (abriendo su circuit breaker) mientras el servicio B funciona normalmente. Muestra el estado de cada circuit breaker después de las llamadas.

Ver solución

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 no disponible",
            "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: datos obtenidos",
            "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: datos sobre '{state['query']}' obtenidos correctamente.",
        "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"Ronda {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']} fallos)")
    print(f"  CB-B: {cbs['b']['state']} ({cbs['b']['failure_count']} fallos)")
# Output esperado:
# Ronda 1:
#   A: [ERROR] Service A: error #1
#   B: Service B: datos sobre 'AI safety' obtenidos cor
#   CB-A: closed (1 fallos)
#   CB-B: closed (0 fallos)
# Ronda 2:
#   A: [ERROR] Service A: error #2
#   B: Service B: datos sobre 'AI safety' obtenidos cor
#   CB-A: closed (2 fallos)
#   CB-B: closed (0 fallos)
# Ronda 3:
#   A: [ERROR] Service A: error #3
#   B: Service B: datos sobre 'AI safety' obtenidos cor
#   CB-A: open (3 fallos)
#   CB-B: closed (0 fallos)
# Ronda 4:
#   A: [CIRCUIT OPEN] Service A no disponible
#   B: Service B: datos sobre 'AI safety' obtenidos cor
#   CB-A: open (3 fallos)
#   CB-B: closed (0 fallos)
# Ronda 5:
#   A: [CIRCUIT OPEN] Service A no disponible
#   B: Service B: datos sobre 'AI safety' obtenidos cor
#   CB-A: open (3 fallos)
#   CB-B: closed (0 fallos)

El circuit breaker de A se abre después de 3 fallos. A partir de la ronda 4, ni siquiera intenta llamar al servicio — ahorra tiempo y no genera carga innecesaria. El servicio B funciona independientemente.

Ejercicio 5: Error handling completo con Functional API (Medio)

Implementa un pipeline de investigación con 4 fuentes usando la Functional API. Dos fuentes fallan. Usa try/except para manejar errores, acumula un log de errores, y sintetiza con las fuentes exitosas. Retorna un dict con: answer, sources_ok, sources_failed, error_log.

Ver solución

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: información enciclopédica sobre {query}."

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

@task
def search_news(query: str) -> str:
    return f"News: noticias recientes sobre {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"Sintetiza sobre '{query}' con estas fuentes:\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": "Todas las fuentes fallaron. Intenta más tarde.",
            "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"Fuentes OK: {result['sources_ok']}")
print(f"Fuentes fallidas: {result['sources_failed']}")
print(f"\nErrores:")
for err in result["error_log"]:
    print(f"  ❌ {err['source']}: [{err['error_type']}] {err['message']}")
print(f"\nRespuesta: {result['answer'][:200]}...")
# Output esperado:
# Fuentes OK: 2
# Fuentes fallidas: 2
#
# Errores:
#   ❌ arxiv: [ConnectionError] arXiv: 503 Service Unavailable
#   ❌ blogs: [TimeoutError] Blogs API: timeout after 20s
#
# Respuesta: Las técnicas de prompt engineering incluyen zero-shot, few-shot, y chain-of-thought...

Ejercicio 6: Sistema completo retry + fallback + circuit breaker (Avanzado)

Implementa un grafo que combine los tres patrones para llamar a una API. El flujo: (1) circuit breaker verifica si vale la pena intentar, (2) retry con backoff exponencial hasta 3 intentos, (3) fallback a caché si todo falla. Configura la simulación para que la API falle las primeras 2 veces y funcione en la tercera. Muestra el log completo del proceso.

Ver solución

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": "Probando una llamada"})

    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: error transitorio (intento global #{GLOBAL_ATTEMPTS})")

            log.append({
                "step": "retry",
                "action": "success",
                "detail": f"Intento {attempt + 1}/{max_retries}: éxito",
            })
            cb_updated = {**cb, "failure_count": 0, "state": "closed"}
            return {
                "result": f"Datos obtenidos exitosamente (intento {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"Intento {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"Todos los reintentos agotados"})

    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": "Usando resultado del caché"})
    return {
        "result": f"[Caché] Información almacenada sobre '{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"Fuente: {result['source']}")
print(f"Resultado: {result['result']}")
print(f"Circuit breaker: {result['circuit_breaker']['state']}")
print(f"\nLog de ejecución:")
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']}")
# Output esperado:
# Fuente: api
# Resultado: Datos obtenidos exitosamente (intento 3)
# Circuit breaker: closed
#
# Log de ejecución:
#   ❌ [retry] Intento 1/3: API: error transitorio (intento global #1) (delay: 0.015s)
#   ❌ [retry] Intento 2/3: API: error transitorio (intento global #2) (delay: 0.025s)
#   ✅ [retry] Intento 3/3: éxito

Este ejercicio muestra el patrón production-grade completo. La API falla 2 veces (errores transitorios) y funciona en el tercer intento. Si los 3 reintentos fallaran, el fallback al caché se activaría automáticamente.

Resumen

En esta cápsula aprendiste:

Error handling es ingeniería fundamental — no un apéndice. Un sistema que crashea el 1% del tiempo no es production-ready
Fallback nodes proveen un plan B cuando el camino principal falla — implementados con conditional edges que evalúan el campo error del estado
Fallback chains (A → B → C → default) intentan múltiples fuentes en orden de preferencia hasta que una funciona
Graceful degradation permite continuar con resultados parciales cuando algunos branches paralelos fallan — cada branch captura sus errores y el nodo de merge trabaja con lo que funcionó
Circuit breaker protege servicios caídos: después de N fallos, deja de intentar por un período de cooldown — previene bombardear un servicio que ya está en problemas
El patrón production-grade combina retry + fallback + circuit breaker: el circuit breaker decide si intentar, el retry maneja errores transitorios, el fallback cubre cuando todo falla
Logging estructurado registra errores como datos en el estado — timestamp, source, error_type, context — para debugging y monitoring
Default responses son el último recurso: nunca retornan vacío, siempre informan al usuario qué pasó y sugieren una acción

Próxima cápsula: Patrones de producción — timeouts por nodo, rate limiting interno, y cómo monitorear un sistema LangGraph en producción real.

Recursos adicionales

LangGraph Error Handling — Patrones de error handling en agentes LangGraph
Retry patterns in distributed systems — AWS: Timeouts, Retries and Backoff with Jitter
Circuit Breaker Pattern — Martin Fowler: Circuit Breaker
LangGraph Branching — Fan-out/fan-in con error handling
LangGraph Functional API — Error handling nativo con try/except
Graceful Degradation Patterns — Microsoft: Retry pattern y degradación
Python logging best practices — Logging estructurado en Python

Módulo 7 — LangChain & LangGraph: From Chains to Agents

← AnteriorMap-Reduce y Deferred Nodes Siguiente →Patterns de Producción