Ciclos y Loops: Retry Patterns

Descripción de la cápsula

Tu agente de investigación llama a una API de búsqueda web. El 90% de las veces, funciona. El 10% restante: timeout, 429 Too Many Requests, 503 Service Unavailable, o simplemente una conexión que se corta a mitad de la respuesta. Sin retry, tu agente crashea y el usuario ve un error. La investigación completa se pierde por un fallo transitorio que habría funcionado 2 segundos después.

Retry es el patrón más fundamental de sistemas distribuidos. Pero retry mal implementado es peor que no hacer retry. Si tu API devuelve 429 (rate limit) y tu código reintenta inmediatamente, le estás mandando más requests a una API que ya te dijo "para." Si 100 clientes hacen lo mismo simultáneamente, la API colapsa. Eso tiene nombre: DDoS involuntario.

En esta cápsula vas a implementar retry correctamente desde el principio: con backoff exponencial (esperas cada vez más entre intentos) y jitter aleatorio (no todos los clientes reintentan al mismo tiempo). Lo implementarás de dos formas: como ciclo en un StateGraph (conditional edge que loop back) y como while loop en la Functional API. Ambos approaches son válidos — la elección depende de si tu workflow es un grafo o una función.

El problema: APIs que fallan

Tu Research Agent v1 llama a una API de búsqueda web. Cuando la API devuelve 429 Too Many Requests o timeout, tu agente muere. Todo el pipeline — descomposición, búsqueda, síntesis — se pierde por un fallo transitorio que habría funcionado 2 segundos después.

No todos los errores merecen retry. La distinción crítica:

Retry solo tiene sentido para errores transitorios — donde el mismo request, enviado segundos después, tiene probabilidad razonable de funcionar.

Retry ingenuo: por qué no funciona

El primer instinto es: si falla, reintenta inmediatamente. Pero mira la timeline:

Timeline con retry inmediato (3 intentos):

t=0.00s  → Request 1 — 429 Too Many Requests
t=0.01s  → Request 2 — 429 Too Many Requests
t=0.02s  → Request 3 — 429 Too Many Requests
t=0.03s  → Exception: "Falló después de 3 intentos"

3 requests en 30 milisegundos a una API que te dijo "estoy saturada." Si 100 usuarios hacen lo mismo, la API recibe 300 requests en 30ms. Retry sin backoff amplifica el problema en vez de resolverlo.

Backoff exponencial con jitter: la solución correcta

La fórmula

import random

delay = min(base_delay * (2 ** attempt) + random.uniform(0, jitter), max_delay)

Cómo se ve en práctica

Timeline con backoff exponencial + jitter:

t=0.00s   → Request 1 — 429 Too Many Requests
t=1.23s   → Request 2 — 429 Too Many Requests  (esperó ~1.2s)
t=3.67s   → Request 3 — 200 OK ✅              (esperó ~2.4s)

Total: 3.67 segundos, 3 intentos, exitoso

Comparado con retry inmediato:

• Le diste tiempo a la API de recuperarse
• Cada intento espera más que el anterior
• El jitter aleatorio evita que 100 clientes reintentenexactamente al mismo segundo

Esa es la teoría. Ahora veamos cómo implementar el mismo concepto dentro de LangGraph, donde el retry es parte de la topología del grafo o del control flow de una función.

Ciclos en grafos: un edge que vuelve atrás

En StateGraph, un ciclo es un conditional edge que rutea un nodo de vuelta a sí mismo o a un nodo anterior:

            ┌──────────── retry ──────────────┐
            │                                  │
            ▼                                  │
START → search_node → should_retry ─── success → END
                          │
                          └── max_retries_exceeded → fallback_node → END

Implementación completa con StateGraph

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


class RetryState(TypedDict):
    query: str
    result: str
    retry_count: int
    max_retries: int
    last_error: str
    status: str


def search_node(state: RetryState) -> dict:
    if random.random() < 0.6:
        error_msg = f"503 Service Unavailable (intento {state['retry_count'] + 1})"
        print(f"[SEARCH] ❌ Falló: {error_msg}")
        return {"last_error": error_msg, "status": "error"}
    print(f"[SEARCH] ✅ Éxito en intento {state['retry_count'] + 1}")
    return {"result": f"Resultados para: {state['query']}", "last_error": "", "status": "success"}


def wait_with_backoff(state: RetryState) -> dict:
    attempt = state["retry_count"]
    delay = min(1.0 * (2 ** attempt) + random.uniform(0, 0.5), 30.0)
    print(f"[BACKOFF] Esperando {delay:.2f}s (intento {attempt + 1})")
    time.sleep(delay)
    return {"retry_count": attempt + 1}


def fallback_node(state: RetryState) -> dict:
    return {"result": f"Resultado parcial para '{state['query']}' — fuente no disponible", "status": "fallback"}


def should_retry(state: RetryState) -> str:
    if state["status"] == "success":
        return "done"
    if state["retry_count"] >= state["max_retries"]:
        return "fallback"
    return "retry"


builder = StateGraph(RetryState)
builder.add_node("search", search_node)
builder.add_node("wait_backoff", wait_with_backoff)
builder.add_node("fallback", fallback_node)
builder.add_edge(START, "search")
builder.add_conditional_edges("search", should_retry, {
    "done": END, "retry": "wait_backoff", "fallback": "fallback"
})
builder.add_edge("wait_backoff", "search")
builder.add_edge("fallback", END)

graph = builder.compile()

result = graph.invoke({
    "query": "prompt engineering best practices",
    "result": "", "retry_count": 0, "max_retries": 3, "last_error": "", "status": ""
})
print(f"\nResultado: {result['result']}")
print(f"Status: {result['status']}, Intentos: {result['retry_count'] + 1}")
# Output esperado (varía por el random):
# [SEARCH] ❌ Falló: 503 Service Unavailable (intento 1)
# [BACKOFF] Esperando 1.23s (intento 1)
# [SEARCH] ✅ Éxito en intento 3
# Resultado: Resultados para: prompt engineering best practices
# Status: success, Intentos: 3

El ciclo es el edge wait_backoff → search — un loop explícito en la topología del grafo. El conditional edge should_retry decide en cada iteración: éxito → END, fallo con intentos restantes → backoff → search (loop), fallo sin intentos → fallback → END.

Retry en la Functional API: for loop con try/except

El mismo patrón sin topología de grafo — Python puro con @entrypoint y @task:

import time
import random
from langgraph.func import entrypoint, task


@task
def search_web(query: str) -> str:
    if random.random() < 0.6:
        raise ConnectionError(f"503 Service Unavailable buscando '{query}'")
    return f"Resultados de búsqueda para: {query}"


@entrypoint()
def research_with_retry(query: str) -> dict:
    max_retries, base_delay = 3, 1.0

    for attempt in range(max_retries):
        try:
            result = search_web(query).result()
            return {"query": query, "result": result, "status": "success"}
        except Exception as e:
            print(f"[RETRY] Intento {attempt + 1} falló: {e}")
            if attempt < max_retries - 1:
                delay = min(base_delay * (2 ** attempt) + random.uniform(0, 0.5), 30.0)
                time.sleep(delay)

    return {"query": query, "result": f"Resultado parcial — fuente no disponible", "status": "fallback"}


result = research_with_retry.invoke("prompt engineering best practices")
print(f"Resultado: {result['result']}, Status: {result['status']}")
# Output esperado (varía):
# [RETRY] Intento 1 falló: 503 Service Unavailable buscando '...'
# Resultado: Resultados de búsqueda para: prompt engineering best practices, Status: success

Más compacto. Misma lógica. El for loop reemplaza al ciclo en el grafo, el try/except reemplaza al conditional edge, y el código después del loop reemplaza al fallback node.

Comparación: retry en StateGraph vs Functional API

Regla de decisión: si el usuario necesita ver "reintentando búsqueda..." en el stream → StateGraph (el ciclo es visible y checkpointeable). Si el retry es un detalle interno → Functional API (más compacto). En la mayoría de los casos, el retry es interno.

Filtrar errores: qué merece retry y qué no

Un retry que no distingue entre errores transitorios y permanentes desperdicia tiempo. Si la API devuelve 401 Unauthorized, reintentar 3 veces con 7 segundos de backoff no va a cambiar nada — tu API key sigue siendo inválida.

import httpx

RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
RETRYABLE_EXCEPTIONS = (httpx.TimeoutException, httpx.ConnectError, ConnectionError, TimeoutError)


def is_retryable(error: Exception) -> bool:
    """Determina si un error merece retry."""
    if isinstance(error, httpx.HTTPStatusError):
        return error.response.status_code in RETRYABLE_STATUS_CODES
    return isinstance(error, RETRYABLE_EXCEPTIONS)

Integrado en el retry, el patrón es: captura la excepción, verifica si es retryable, y solo entonces aplica backoff. Si no es retryable, propaga el error inmediatamente con raise:

import time
import random
import httpx

def search_with_smart_retry(query: str, max_retries: int = 3, base_delay: float = 1.0) -> dict:
    """Retry inteligente — solo reintenta errores transitorios."""
    for attempt in range(max_retries):
        try:
            response = httpx.get("https://api.example.com/search", params={"q": query}, timeout=10.0)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            if not is_retryable(e):
                print(f"[SEARCH] Error permanente (no retryable): {e}")
                raise
            if attempt < max_retries - 1:
                delay = min(base_delay * (2 ** attempt) + random.uniform(0, 0.5), 30.0)
                print(f"[RETRY] Intento {attempt + 1}: {e}. Esperando {delay:.2f}s")
                time.sleep(delay)

    raise Exception(f"Búsqueda falló después de {max_retries} intentos")

429 → reintenta con backoff. 401 → propaga inmediatamente. Esa distinción evita desperdiciar tiempo en errores irrecuperables.

Break conditions: cuándo dejar de reintentar

max_retries no es la única condición de parada. En producción, necesitas un timeout budget — un tiempo total máximo para toda la operación de retry. No importa si tienes 5 intentos restantes; si ya pasaron 30 segundos, el usuario no va a esperar más.

import time
import random


def search_with_timeout_budget(query: str, max_retries: int = 5, timeout_budget: float = 15.0) -> dict:
    """Retry con presupuesto de tiempo total."""
    start_time = time.time()

    for attempt in range(max_retries):
        elapsed = time.time() - start_time
        remaining = timeout_budget - elapsed

        if remaining <= 0:
            return {"status": "timeout", "query": query, "attempts": attempt}

        try:
            if random.random() < 0.5:
                raise ConnectionError("Simulated failure")
            return {"status": "success", "result": f"Results for: {query}", "attempts": attempt + 1}
        except Exception as e:
            if attempt < max_retries - 1:
                delay = min(1.0 * (2 ** attempt) + random.uniform(0, 0.5), remaining)
                if delay <= 0:
                    break
                print(f"[RETRY] Intento {attempt + 1}: {e}. Esperando {delay:.2f}s ({remaining:.1f}s restantes)")
                time.sleep(delay)

    return {"status": "failed", "query": query, "attempts": max_retries}


result = search_with_timeout_budget("AI safety", timeout_budget=10.0)
print(f"Status: {result['status']}, Intentos: {result['attempts']}")
# Output esperado (varía):
# [RETRY] Intento 1: Simulated failure. Esperando 1.23s (9.8s restantes)
# Status: success, Intentos: 2

El detalle clave: delay = min(..., remaining). Si solo quedan 2 segundos de budget y el backoff sugiere 4 segundos, espera solo 2. Dos condiciones de parada, la que se alcance primero gana: max_retries O timeout_budget.

recursion_limit: el safety net de LangGraph

LangGraph tiene un mecanismo de seguridad para ciclos: recursion_limit. Si un grafo ejecuta más de N supersteps (transiciones entre nodos), lanza GraphRecursionError. El default es 25 supersteps.

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


class CounterState(TypedDict):
    counter: int


def increment(state: CounterState) -> dict:
    print(f"Counter: {state['counter']}")
    return {"counter": state["counter"] + 1}


def always_loop(state: CounterState) -> str:
    return "loop"


builder = StateGraph(CounterState)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_conditional_edges("increment", always_loop, {"loop": "increment"})

graph = builder.compile()

try:
    result = graph.invoke({"counter": 0}, {"recursion_limit": 5})
except Exception as e:
    print(f"\nError: {type(e).__name__}: {e}")
# Output esperado:
# Counter: 0
# Counter: 1
# ...
# Counter: 4
# Error: GraphRecursionError: Recursion limit of 5 reached...

Si tu retry loop tiene un bug y should_retry nunca retorna "done", el recursion_limit evita un loop infinito. Siempre configura ambos: el límite de negocio (max_retries) y el límite de seguridad (recursion_limit). Para un ciclo de retry con 2 nodos por iteración (search + backoff), configura: recursion_limit = max_retries * 2 + 5.

Retry con context: llevar información entre intentos

A veces la API devuelve un header Retry-After que te dice cuánto esperar. Agrega suggested_delay al estado para que el nodo de backoff lo use:

def adaptive_backoff(state: dict) -> dict:
    attempt = state["retry_count"]
    if state.get("suggested_delay", 0) > 0:
        delay = state["suggested_delay"] + random.uniform(0, 0.5)
    else:
        delay = min(1.0 * (2 ** attempt) + random.uniform(0, 0.5), 30.0)
    time.sleep(delay)
    return {"retry_count": attempt + 1}

El estado del grafo lleva información entre iteraciones del ciclo. Datos del intento N influyen en la estrategia del intento N+1.

Troubleshooting

Problema 1: "GraphRecursionError" inesperado

Síntoma: GraphRecursionError antes de agotar los reintentos. Causa: recursion_limit default = 25. Cada reintento con 2 nodos (search + backoff) consume 2 supersteps. Solución: graph.invoke(state, {"recursion_limit": max_retries * 2 + 5}).

Problema 2: Todos los clientes reintentan al mismo tiempo

Síntoma: Después de un pico, todos los retries causan otro pico. Causa: Backoff sin jitter — todos calculan el mismo delay. Solución: Siempre incluye random.uniform(0, jitter) sumado al delay.

Problema 3: Retry reintenta errores permanentes

Síntoma: 7 segundos de retry contra un 401 Unauthorized. Causa: No filtras el tipo de error. Solución: Implementa is_retryable(error) — solo reintenta 429, 500, 502, 503, 504.

Problema 4: El delay crece indefinidamente

Síntoma: En el intento 10, el delay es 1024 segundos. Causa: Backoff sin max_delay. Solución: Siempre usa min(delay, max_delay) con max_delay=30.

Problema 5: Tests inconsistentes

Síntoma: Tests pasan a veces por depender de random.random(). Solución: Usa random.seed(42) al inicio del test para determinismo.

Ejercicios

Ejercicio 1: Backoff exponencial básico (Fácil)

Escribe una función calculate_delays que reciba max_retries, base_delay, y jitter=0 (sin jitter para verificar el patrón) y retorne una lista con el delay de cada intento. Verifica que los delays se duplican: [1.0, 2.0, 4.0, 8.0] para base_delay=1.0 y max_retries=4.

def calculate_delays(
    max_retries: int,
    base_delay: float = 1.0,
    jitter: float = 0.0,
    max_delay: float = 60.0
) -> list[float]:
    """Calcula la secuencia de delays para retry con backoff exponencial."""
    import random
    delays = []
    for attempt in range(max_retries):
        delay = min(
            base_delay * (2 ** attempt) + random.uniform(0, jitter),
            max_delay
        )
        delays.append(round(delay, 2))
    return delays


delays_no_jitter = calculate_delays(max_retries=4, base_delay=1.0, jitter=0.0)
print(f"Sin jitter: {delays_no_jitter}")
# Output esperado: Sin jitter: [1.0, 2.0, 4.0, 8.0]

delays_with_jitter = calculate_delays(max_retries=4, base_delay=1.0, jitter=0.5)
print(f"Con jitter: {delays_with_jitter}")
# Output esperado: Con jitter: [1.23, 2.41, 4.15, 8.33] (varía por random)

delays_with_cap = calculate_delays(max_retries=6, base_delay=1.0, jitter=0.0, max_delay=10.0)
print(f"Con cap: {delays_with_cap}")
# Output esperado: Con cap: [1.0, 2.0, 4.0, 8.0, 10.0, 10.0]

assert delays_no_jitter == [1.0, 2.0, 4.0, 8.0], "Backoff debe duplicarse"
assert all(d <= 10.0 for d in delays_with_cap), "max_delay debe respetarse"
print("\n✅ Todos los assertions pasaron")

Ejercicio 2: Retry con la Functional API (Fácil)

Crea un @entrypoint con un @task que simula una función que falla las primeras 2 veces y funciona en la tercera. Implementa retry con backoff dentro del entrypoint. Verifica que retorna exitosamente al tercer intento.

import time
import random
from langgraph.func import entrypoint, task

call_count = 0

@task
def flaky_api_call(query: str) -> str:
    """Simula una API que falla las primeras 2 veces."""
    global call_count
    call_count += 1
    if call_count <= 2:
        raise ConnectionError(f"Fallo simulado (intento {call_count})")
    return f"Respuesta exitosa para: {query}"


@entrypoint()
def agent_with_retry(query: str) -> dict:
    max_retries = 4
    base_delay = 0.1  # Delays cortos para el ejercicio

    for attempt in range(max_retries):
        try:
            result = flaky_api_call(query).result()
            return {"result": result, "attempts": attempt + 1, "status": "success"}
        except Exception as e:
            print(f"[RETRY] Intento {attempt + 1}: {e}")
            if attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 0.05)
                time.sleep(delay)

    return {"result": "", "attempts": max_retries, "status": "failed"}


call_count = 0
result = agent_with_retry.invoke("test query")
print(f"\nResultado: {result['result']}")
print(f"Intentos: {result['attempts']}")
print(f"Status: {result['status']}")
# Output esperado:
# [RETRY] Intento 1: Fallo simulado (intento 1)
# [RETRY] Intento 2: Fallo simulado (intento 2)
#
# Resultado: Respuesta exitosa para: test query
# Intentos: 3
# Status: success

assert result["status"] == "success"
assert result["attempts"] == 3
print("✅ Retry funcionó correctamente")

Ejercicio 3: Retry con StateGraph y error log acumulado (Medio)

Implementa un StateGraph con un ciclo de retry donde el estado tenga errors: Annotated[list[str], operator.add] como reducer para acumular errores. Incluye nodos process, backoff, fallback, y un conditional edge should_retry. Al final, imprime la lista de errores para ver el historial completo de fallos.

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


class ProcessState(TypedDict):
    input_data: str
    result: str
    retry_count: int
    max_retries: int
    errors: Annotated[list[str], operator.add]
    status: str


def process_node(state: ProcessState) -> dict:
    if random.random() < 0.7:
        return {"errors": [f"Intento {state['retry_count'] + 1}: failed"], "status": "error"}
    return {"result": f"Procesado: {state['input_data']}", "status": "success"}


def backoff_node(state: ProcessState) -> dict:
    delay = min(0.1 * (2 ** state["retry_count"]) + random.uniform(0, 0.05), 5.0)
    time.sleep(delay)
    return {"retry_count": state["retry_count"] + 1}


def fallback_node(state: ProcessState) -> dict:
    return {"result": f"Fallback para: {state['input_data']}", "status": "fallback"}


def should_retry(state: ProcessState) -> str:
    if state["status"] == "success":
        return "done"
    return "fallback" if state["retry_count"] >= state["max_retries"] else "retry"


builder = StateGraph(ProcessState)
builder.add_node("process", process_node)
builder.add_node("backoff", backoff_node)
builder.add_node("fallback", fallback_node)
builder.add_edge(START, "process")
builder.add_conditional_edges("process", should_retry, {"done": END, "retry": "backoff", "fallback": "fallback"})
builder.add_edge("backoff", "process")
builder.add_edge("fallback", END)
graph = builder.compile()

random.seed(42)
result = graph.invoke({"input_data": "datos de prueba", "result": "", "retry_count": 0, "max_retries": 3, "errors": [], "status": ""})
print(f"Status: {result['status']}, Errores: {result['errors']}")
# Output esperado: Status: success/fallback, Errores: ['Intento 1: failed', ...]

Ejercicio 4: Filtro de errores retryable vs permanente (Medio)

Crea una función smart_retry que reciba una función y dos tuplas: retryable_exceptions y permanent_exceptions. Si la función lanza una excepción retryable, reintenta con backoff. Si lanza una permanente, la propaga inmediatamente. Testea con ValueError (permanente) y ConnectionError (retryable).

import time
import random


def smart_retry(func, max_retries=3, base_delay=0.1,
                retryable=(ConnectionError, TimeoutError),
                permanent=(ValueError, TypeError)):
    last_exc = None
    for attempt in range(max_retries):
        try:
            return func()
        except permanent as e:
            print(f"[PERMANENT] {type(e).__name__}: {e}")
            raise
        except retryable as e:
            last_exc = e
            print(f"[RETRY] {attempt + 1}/{max_retries}: {type(e).__name__}: {e}")
            if attempt < max_retries - 1:
                time.sleep(min(base_delay * (2 ** attempt) + random.uniform(0, 0.05), 10.0))
    raise last_exc


# Test 1: retryable — funciona al tercer intento
call_count = 0
def flaky():
    global call_count
    call_count += 1
    if call_count <= 2:
        raise ConnectionError(f"Fallo {call_count}")
    return "OK"

call_count = 0
print(smart_retry(flaky, max_retries=4))
# Output: [RETRY] 1/4: ... → [RETRY] 2/4: ... → OK

# Test 2: permanente — no reintenta
try:
    smart_retry(lambda: (_ for _ in ()).throw(ValueError("bad input")), max_retries=4)
except ValueError as e:
    print(f"Capturado: {e}")
# Output: [PERMANENT] ValueError: bad input → Capturado: bad input

Ejercicio 5: Retry con timeout budget en @entrypoint (Medio-Avanzado)

Crea un @entrypoint que busque con retry y dos condiciones de parada: max_retries=10 y timeout_budget=3.0. Usa un @task que siempre falle para verificar que el budget se agota en ~3 segundos, no en 10 intentos. Loggea el tiempo transcurrido en cada intento.

import time
import random
from langgraph.func import entrypoint, task


@task
def always_fail(query: str) -> str:
    raise ConnectionError(f"Service unavailable for: {query}")


@entrypoint()
def search_with_budget(inputs: dict) -> dict:
    query, max_retries = inputs["query"], inputs.get("max_retries", 10)
    timeout_budget, base_delay = inputs.get("timeout_budget", 10.0), inputs.get("base_delay", 1.0)
    start_time = time.time()
    attempts = 0

    for attempt in range(max_retries):
        remaining = timeout_budget - (time.time() - start_time)
        if remaining <= 0:
            break
        attempts += 1
        try:
            return {"status": "success", "result": always_fail(query).result(), "attempts": attempts}
        except Exception as e:
            print(f"[RETRY] Intento {attempts} a t={time.time() - start_time:.2f}s: {e}")
            if attempt < max_retries - 1:
                delay = min(base_delay * (2 ** attempt) + random.uniform(0, 0.3), remaining)
                if delay > 0:
                    time.sleep(delay)

    elapsed = round(time.time() - start_time, 2)
    return {"status": "budget_exhausted", "attempts": attempts, "elapsed": elapsed}


result = search_with_budget.invoke({"query": "test", "max_retries": 10, "timeout_budget": 3.0, "base_delay": 1.0})
print(f"\nStatus: {result['status']}, Intentos: {result['attempts']}, Tiempo: {result['elapsed']}s")
assert result["elapsed"] <= 4.0, f"Budget de 3s excedido: {result['elapsed']}s"
print(f"✅ Budget respetado: {result['elapsed']}s <= 4.0s")
# Output esperado:
# [RETRY] Intento 1 a t=0.00s: Service unavailable for: test
# [RETRY] Intento 2 a t=1.12s: Service unavailable for: test
# Status: budget_exhausted, Intentos: 3, Tiempo: ~3.0s
# ✅ Budget respetado

Ejercicio 6: Retry encapsulado como building block (Avanzado)

Crea un @task llamado search_source_with_retry que encapsule toda la lógica de retry internamente. Recibe {"source": str, "query": str}, simula búsqueda con fallos, implementa retry con backoff + jitter, y retorna {"status", "source", "results", "attempts"}. Luego usa ese task desde un @entrypoint que busque en 3 fuentes. Este building block se reutilizará en la Cápsula 03 para branching paralelo.

import time
import random
from langgraph.func import entrypoint, task

FAILURE_RATES = {"web": 0.3, "papers": 0.5, "news": 0.2}


def simulate_search(source: str, query: str) -> list[dict]:
    failure_rate = FAILURE_RATES.get(source, 0.3)
    if random.random() < failure_rate:
        raise ConnectionError(f"{source}: 503 Service Unavailable")
    return [{"title": f"[{source}] Resultado para '{query}'", "relevance": 0.9}]


@task
def search_source_with_retry(inputs: dict) -> dict:
    source, query = inputs["source"], inputs["query"]
    max_retries, base_delay = 3, 0.5
    errors = []

    for attempt in range(max_retries):
        try:
            results = simulate_search(source, query)
            return {"status": "success", "source": source, "results": results, "attempts": attempt + 1}
        except ConnectionError as e:
            errors.append(str(e))
            if attempt < max_retries - 1:
                delay = min(base_delay * (2 ** attempt) + random.uniform(0, 0.3), 10.0)
                time.sleep(delay)

    return {"status": "exhausted", "source": source, "results": [], "attempts": max_retries, "errors": errors}


@entrypoint()
def multi_source_search(query: str) -> dict:
    sources = ["web", "papers", "news"]
    futures = [search_source_with_retry({"source": s, "query": query}) for s in sources]
    results = [f.result() for f in futures]

    successful = [r for r in results if r["status"] == "success"]
    return {"query": query, "successful": len(successful), "total": len(sources), "results": results}


random.seed(123)
report = multi_source_search.invoke("retry patterns")
print(f"Fuentes exitosas: {report['successful']}/{report['total']}")
for r in report["results"]:
    icon = "✅" if r["status"] == "success" else "⚠️"
    print(f"  {icon} {r['source']}: {r['status']} ({r['attempts']} intentos)")
# Output esperado (varía con seed):
# Fuentes exitosas: 3/3
#   ✅ web: success (1 intentos)
#   ✅ papers: success (2 intentos)
#   ✅ news: success (1 intentos)

Resumen

En esta cápsula aprendiste:

• Retry sin backoff es un anti-patrón. Reintentar inmediatamente amplifica el problema — le mandas más requests a una API que ya te dijo "para." Siempre usa backoff exponencial con jitter
• La fórmula: delay = min(base * 2^attempt + random.uniform(0, jitter), max_delay). Duplica la espera en cada intento, agrega variación aleatoria, y respeta un tope máximo
• Distingue errores transitorios de permanentes. 429 y 503 merecen retry. 401 y 404 no. Reintentar errores permanentes desperdicia tiempo
• En StateGraph, el retry es un ciclo explícito: conditional edge que rutea de vuelta a un nodo. El estado lleva retry_count y last_error. La topología es visible y streameable
• En Functional API, el retry es un for/while loop con try/except y time.sleep(). Más compacto, mismo efecto. El retry es un detalle de implementación oculto
• Dos condiciones de parada: max_retries (número de intentos) y timeout_budget (tiempo total). La que se alcance primero, gana
• recursion_limit es el safety net de LangGraph contra loops infinitos. Configúralo siempre por encima de tu max_retries esperado
• El retry con context lleva información entre intentos (como Retry-After headers) para adaptar la estrategia

Próxima cápsula: Branching y Merge: Ejecución Paralela — tu Research Agent busca en 3 fuentes secuencialmente (9 segundos). Vas a ejecutar las 3 búsquedas en paralelo (3 segundos) usando fan-out/fan-in y la Send API de LangGraph.

Recursos adicionales

1. Exponential Backoff and Jitter (AWS Architecture Blog) — El artículo definitivo sobre backoff con jitter. Compara full jitter, equal jitter, y decorrelated jitter con simulaciones
2. LangGraph — Concepts: Cycles — Documentación oficial sobre cómo LangGraph maneja ciclos y el recursion_limit
3. LangGraph — How to control graph recursion limit — Guía práctica para configurar y manejar el límite de recursión
4. Retry Pattern — Microsoft Azure Architecture — Documentación de Microsoft sobre el patrón de retry con guía de implementación
5. Circuit Breaker Pattern — Martin Fowler — El siguiente nivel después de retry: cuando dejar de intentar por completo. Preview del Módulo 7 Cápsula 06
6. httpx — Timeouts — Referencia de timeouts en httpx, la librería HTTP usada en los ejemplos

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

Siguiente cápsula: Branching y Merge: Ejecución Paralela — aprenderás fan-out/fan-in para ejecutar búsquedas en paralelo, la Send API de LangGraph, y cómo mergear resultados de múltiples fuentes con deduplicación.