Interrupts: Pausar Ejecución

Descripción de la cápsula

Tu agente de investigación ejecuta 5 nodos en secuencia: descompone la consulta, busca en fuentes, analiza hallazgos, genera el reporte, y lo publica. El proceso completo toma 3 minutos. En el paso 2, el agente decide buscar en una API de papers que cobra $0.10 por consulta. Ejecuta 50 consultas — $5.00 que nadie aprobó.

Con interrupt(), el agente se pausa antes de la búsqueda costosa: "Planeo buscar 50 papers en Google Scholar (costo estimado: $5.00). ¿Procedo? [sí/no/reducir]". Tú respondes "reduce a 10." El agente ajusta y continúa — $1.00 en lugar de $5.00, con la misma calidad.

interrupt() es la función que pausa tu grafo a mitad de ejecución. El grafo guarda su estado en un checkpoint, espera input humano indefinidamente, y resume cuando el humano responde. Es la base de todos los patrones de HITL que implementarás en este módulo.

En la cápsula anterior entendiste por qué HITL importa y cuándo aplicarlo. Ahora aprendes el cómo: la mecánica de interrupt(), Command(resume=), y el flujo completo de pausa/reanudación.

La API: interrupt() y Command

Dos imports. Eso es todo lo que necesitas para HITL:

from langgraph.types import interrupt, Command

• interrupt(value) — pausa el grafo y envía value al caller (el humano). value puede ser cualquier dato JSON-serializable: string, dict, lista, número, booleano.
• Command(resume=value) — reanuda el grafo pausado. value se convierte en el valor de retorno de interrupt() dentro del nodo.

Cómo funciona interrupt(): paso a paso

1. El grafo ejecuta nodos normalmente
   [START] → [nodo_A] → [nodo_B] → ...

2. Un nodo llama a interrupt("mensaje para el humano")
   [nodo_B] ejecuta → interrupt("¿Apruebas?") → PAUSA

3. El grafo se SUSPENDE
   Estado guardado en checkpoint (gracias al checkpointer)
   El valor de interrupt aparece en result["__interrupt__"]

4. El humano ve el mensaje y toma una decisión
   Esto puede tomar segundos, minutos, horas, o días

5. El humano envía Command(resume=valor)
   graph.invoke(Command(resume="sí"), config)

6. El nodo se RE-EJECUTA desde el principio
   interrupt() ahora retorna "sí" (el valor del resume)
   El nodo continúa con ese valor

7. El grafo sigue ejecutando los nodos restantes
   [nodo_B] completa → [nodo_C] → ... → [END]

El paso 6 es crítico y contra-intuitivo: el nodo entero se re-ejecuta desde el principio, no desde la línea donde estaba interrupt(). El interrupt() retorna el valor del resume en lugar de pausar. Cualquier código antes de interrupt() se ejecuta de nuevo.

CRÍTICO: checkpointer es obligatorio

Sin checkpointer, interrupt() no funciona. El grafo no puede pausarse si no tiene dónde guardar su estado.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

Regla absoluta: si usas interrupt(), configura un checkpointer. Si compilas sin checkpointer y un nodo llama a interrupt(), obtendrás un error. Es una dependencia técnica, no una recomendación.

Ejemplo básico: simulación completa de UX

Este es el ejemplo más importante de la cápsula. Muestra el flujo completo: el agente planea, se pausa, el humano decide, y el agente continúa.

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from typing import TypedDict


class ResearchState(TypedDict):
    query: str
    search_plan: str
    human_decision: str
    results: str
    status: str


def plan_search(state: ResearchState) -> dict:
    plan = (
        "Fuentes a consultar:\n"
        "  1. Google Scholar ($0.10/consulta × 10 = $1.00)\n"
        "  2. Web search (gratis)\n"
        "  3. ArXiv (gratis)\n"
        f"Costo total estimado: $1.00\n"
        f"Query: '{state['query']}'"
    )
    return {"search_plan": plan, "status": "plan_ready"}


def approve_search(state: ResearchState) -> dict:
    decision = interrupt(
        f"Plan de búsqueda:\n{state['search_plan']}\n\n"
        "¿Proceder? [sí / no / editar]"
    )
    return {"human_decision": decision, "status": "decision_received"}


def execute_search(state: ResearchState) -> dict:
    if state["human_decision"] == "no":
        return {"results": "Búsqueda cancelada.", "status": "cancelled"}

    if state["human_decision"].startswith("editar:"):
        edited = state["human_decision"].replace("editar:", "").strip()
        return {
            "results": f"Búsqueda ejecutada con plan editado: {edited}",
            "status": "completed_edited"
        }

    return {
        "results": f"Búsqueda completada para: '{state['query']}'. 15 resultados encontrados.",
        "status": "completed"
    }


builder = StateGraph(ResearchState)
builder.add_node("plan", plan_search)
builder.add_node("approve", approve_search)
builder.add_node("execute", execute_search)
builder.add_edge(START, "plan")
builder.add_edge("plan", "approve")
builder.add_edge("approve", "execute")
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "research_001"}}

print("=== PASO 1: Ejecutar hasta el interrupt ===")
result = graph.invoke(
    {"query": "Estado del arte en RAG 2025", "search_plan": "",
     "human_decision": "", "results": "", "status": ""},
    config
)
print(f"Estado: {result['status']}")
for item in result.get("__interrupt__", []):
    print(f"\n--- AGENTE PREGUNTA ---\n{item.value}\n")

print("=== PASO 2: Humano responde 'sí' ===")
result = graph.invoke(Command(resume="sí"), config)
print(f"Estado: {result['status']}")
print(f"Resultados: {result['results']}")
# Output esperado:
# === PASO 1: Ejecutar hasta el interrupt ===
# Estado: plan_ready
#
# --- AGENTE PREGUNTA ---
# Plan de búsqueda:
# Fuentes a consultar:
#   1. Google Scholar ($0.10/consulta × 10 = $1.00)
#   2. Web search (gratis)
#   3. ArXiv (gratis)
# Costo total estimado: $1.00
# Query: 'Estado del arte en RAG 2025'
#
# ¿Proceder? [sí / no / editar]
#
# === PASO 2: Humano responde 'sí' ===
# Estado: completed
# Resultados: Búsqueda completada para: 'Estado del arte en RAG 2025'. 15 resultados encontrados.

El primer invoke() ejecuta plan y approve. Cuando approve llama a interrupt(), el grafo se pausa. El resultado incluye __interrupt__ con el mensaje del agente. El segundo invoke() con Command(resume="sí") reanuda el grafo — approve recibe "sí" como retorno de interrupt(), y luego execute corre normalmente.

Tres caminos desde el mismo interrupt

El ejemplo anterior soporta 3 respuestas. Cada thread_id crea un camino independiente:

init_state = {
    "query": "RAG 2025", "search_plan": "",
    "human_decision": "", "results": "", "status": ""
}

for decision, tid in [("sí", "demo_yes"), ("no", "demo_no"), ("editar: solo ArXiv", "demo_edit")]:
    cfg = {"configurable": {"thread_id": tid}}
    graph.invoke(init_state, cfg)
    r = graph.invoke(Command(resume=decision), cfg)
    print(f"'{decision}' → {r['status']}: {r['results'][:60]}")
# Output esperado:
# 'sí' → completed: Búsqueda completada para: 'RAG 2025'. 15 resultados e
# 'no' → cancelled: Búsqueda cancelada.
# 'editar: solo ArXiv' → completed_edited: Búsqueda ejecutada con plan editado: solo ArXiv

HITL no es binario. El humano puede aprobar, rechazar, o modificar.

Qué pasa durante la pausa

Cuando interrupt() se ejecuta, el grafo se congela:

• ✅ Checkpointer guardó todo el estado del grafo
• ✅ El nodo con interrupt quedó suspendido — nodos siguientes no han corrido
• ✅ Puedes inspeccionar con graph.get_state(config)
• ✅ Puedes esperar minutos, horas, o días
• ❌ Con MemorySaver, si el proceso muere → estado perdido
• ✅ Con PostgresSaver, sobrevive incluso a reinicios

Puedes verificar qué nodo está esperando:

snapshot = graph.get_state(config)
print(f"Estado: {snapshot.values}")
print(f"Nodo en espera: {snapshot.next}")  # ('approval',) si está pausado, () si terminó

snapshot.next te dice qué nodo espera el resume. Si es una tupla vacía (), el grafo ya terminó.

Re-ejecución del nodo: la regla más importante

Cuando el grafo se reanuda, el nodo que contiene interrupt() se re-ejecuta desde el principio. No se reanuda desde la línea exacta — se ejecuta todo el nodo de nuevo, y interrupt() retorna el valor del resume en lugar de pausar.

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from typing import TypedDict

call_counter = 0


class State(TypedDict):
    result: str


def my_node(state: State) -> dict:
    global call_counter
    call_counter += 1
    print(f"  my_node ejecutado (vez #{call_counter})")

    decision = interrupt("¿Continuar?")
    print(f"  interrupt retornó: '{decision}'")
    return {"result": decision}


builder = StateGraph(State)
builder.add_node("my_node", my_node)
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "reexec_demo"}}

print("--- Primera invocación (se pausa) ---")
graph.invoke({"result": ""}, config)

print("\n--- Segunda invocación (resume) ---")
result = graph.invoke(Command(resume="aprobado"), config)
print(f"\nResultado: {result['result']}")
# Output esperado:
# --- Primera invocación (se pausa) ---
#   my_node ejecutado (vez #1)
#
# --- Segunda invocación (resume) ---
#   my_node ejecutado (vez #2)
#   interrupt retornó: 'aprobado'
#
# Resultado: aprobado

El nodo se ejecutó 2 veces. El print antes de interrupt() se ejecuta ambas veces.

Implicación: cualquier código antes de interrupt() debe ser idempotente — seguro de ejecutar múltiples veces sin efectos secundarios problemáticos. No hagas API calls, no insertes registros, no envíes emails antes de interrupt(). Ponlo después.

Valores del interrupt: qué puedes enviar y recibir

Lo que pasas a interrupt() es lo que el humano ve. Lo que el humano pone en Command(resume=) es lo que interrupt() retorna. Ambos deben ser JSON-serializable:

# Enviar al humano — string o dict
decision = interrupt("¿Apruebas esta acción?")
decision = interrupt({"action": "search", "cost": 5.00, "question": "¿Proceder?"})

# Recibir del humano — cualquier tipo JSON-serializable
graph.invoke(Command(resume="sí"), config)                       # string
graph.invoke(Command(resume=True), config)                       # bool
graph.invoke(Command(resume={"approved": True, "max": 2.0}), config)  # dict

Dentro del nodo, interrupt() retorna exactamente lo que el humano envió en Command(resume=).

Múltiples interrupts en el mismo grafo

Un grafo puede tener interrupts en diferentes nodos. Cada Command(resume=) avanza al siguiente:

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from typing import TypedDict


class MultiState(TypedDict):
    plan_approved: str
    budget_approved: str
    status: str


def approve_plan(state: MultiState) -> dict:
    return {"plan_approved": interrupt("¿Apruebas el plan? [sí/no]")}


def approve_budget(state: MultiState) -> dict:
    return {"budget_approved": interrupt("¿Apruebas el presupuesto: $5.00? [sí/no]")}


def execute(state: MultiState) -> dict:
    both_yes = state["plan_approved"] == "sí" and state["budget_approved"] == "sí"
    return {"status": "executed" if both_yes else "cancelled"}


builder = StateGraph(MultiState)
builder.add_node("approve_plan", approve_plan)
builder.add_node("approve_budget", approve_budget)
builder.add_node("execute", execute)
builder.add_edge(START, "approve_plan")
builder.add_edge("approve_plan", "approve_budget")
builder.add_edge("approve_budget", "execute")
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "multi_interrupt"}}

r1 = graph.invoke({"plan_approved": "", "budget_approved": "", "status": ""}, config)
print(f"Interrupt 1: {r1['__interrupt__'][0].value}")

r2 = graph.invoke(Command(resume="sí"), config)
print(f"Interrupt 2: {r2['__interrupt__'][0].value}")

r3 = graph.invoke(Command(resume="sí"), config)
print(f"Estado final: {r3['status']}")
# Output esperado:
# Interrupt 1: ¿Apruebas el plan? [sí/no]
# Interrupt 2: ¿Apruebas el presupuesto: $5.00? [sí/no]
# Estado final: executed

El grafo se pausa dos veces. Cada Command(resume=) avanza al siguiente interrupt.

Routing post-aprobación con Command(goto=)

Command no solo resume — también redirige. Si un nodo retorna Command(goto="nombre_nodo"), el grafo salta a ese nodo:

from typing import Literal, TypedDict
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END


class ApprovalState(TypedDict):
    action: str
    status: str


def approval_gate(state: ApprovalState) -> Command[Literal["execute", "cancel"]]:
    decision = interrupt({
        "question": f"¿Apruebas: '{state['action']}'?",
        "options": ["aprobar", "rechazar"]
    })
    if decision == "aprobar":
        return Command(goto="execute")
    return Command(goto="cancel")


def execute_node(state: ApprovalState) -> dict:
    return {"status": f"Ejecutado: {state['action']}"}


def cancel_node(state: ApprovalState) -> dict:
    return {"status": f"Cancelado: {state['action']}"}


builder = StateGraph(ApprovalState)
builder.add_node("approval", approval_gate)
builder.add_node("execute", execute_node)
builder.add_node("cancel", cancel_node)
builder.add_edge(START, "approval")
builder.add_edge("execute", END)
builder.add_edge("cancel", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

init = {"action": "Enviar reporte", "status": ""}

cfg_yes = {"configurable": {"thread_id": "route_yes"}}
graph.invoke(init, cfg_yes)
r = graph.invoke(Command(resume="aprobar"), cfg_yes)
print(f"Aprobar: {r['status']}")

cfg_no = {"configurable": {"thread_id": "route_no"}}
graph.invoke(init, cfg_no)
r = graph.invoke(Command(resume="rechazar"), cfg_no)
print(f"Rechazar: {r['status']}")
# Output esperado:
# Aprobar: Ejecutado: Enviar reporte
# Rechazar: Cancelado: Enviar reporte

El type hint Command[Literal["execute", "cancel"]] declara los destinos válidos. Esto es más limpio que edges condicionales para approval gates.

Manejo de respuestas inesperadas

Un while True + interrupt() crea un loop de validación que re-pregunta hasta obtener una respuesta válida:

from langgraph.types import interrupt
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from typing import TypedDict


class ValidatedState(TypedDict):
    result: str


def validated_interrupt(state: ValidatedState) -> dict:
    valid = ["sí", "no", "editar"]
    prompt = "¿Apruebas? [sí/no/editar]"

    while True:
        response = interrupt(prompt)
        if response in valid:
            return {"result": f"Respuesta válida: {response}"}
        prompt = f"'{response}' no es válido. Opciones: {', '.join(valid)}"

Cada respuesta inválida re-pausa el grafo con un mensaje de error más claro. El loop solo termina cuando el input es válido.

Troubleshooting

Problema 1: "interrupt() no pausa — ejecuta hasta el final"

Síntoma: Invocas el grafo esperando que se pause, pero ejecuta todos los nodos sin detenerse. Causa: No pasaste un checkpointer al compilar. Solución: Verifica que builder.compile(checkpointer=checkpointer) incluya el checkpointer.

Problema 2: "El nodo con interrupt() se ejecuta 2 veces"

Síntoma: Los print() o side effects dentro del nodo se ejecutan dos veces. Causa: Comportamiento esperado. Al resumir, el nodo se re-ejecuta desde el principio. interrupt() retorna el valor del resume en lugar de pausar. Solución: Coloca side effects después de interrupt(), no antes. El código pre-interrupt debe ser idempotente.

Problema 3: "Command(resume=) lanza error de thread_id"

Síntoma: Error al invocar graph.invoke(Command(resume="valor"), config). Causa: El config no tiene el mismo thread_id que se usó en la invocación original. Solución: Usa exactamente el mismo dict config con el mismo thread_id.

Problema 4: "No veo interrupt en el resultado"

Síntoma: El primer invoke() retorna sin el campo __interrupt__. Causa: El grafo terminó antes de llegar al nodo con interrupt() — un routing condicional lo saltó. Solución: Verifica con graph.get_state(config) que snapshot.next contiene el nodo esperado.

Problema 5: "Envuelvo interrupt() en try/except y no funciona"

Síntoma: El interrupt se "traga" y el nodo continúa sin pausarse. Causa: interrupt() funciona lanzando una excepción especial. Un try/except Exception la captura. Solución: Nunca envuelvas interrupt() en try/except genérico. Usa excepciones específicas (except ValueError) o pon el try/except después del interrupt.

Ejercicios

Ejercicio 1: Tres caminos — aprobar, rechazar, editar (Fácil)

Crea un grafo con 3 nodos (prepare → approve → execute). El nodo prepare establece recipient = "equipo@empresa.com". El nodo approve usa interrupt() para preguntar "¿Enviar email a {recipient}? [sí / no / editar:]". El nodo execute verifica la decisión: si "sí" envía, si "no" cancela, si "editar:X" envía a X. Usa 3 thread_ids para probar los 3 caminos.

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from typing import TypedDict


class EmailState(TypedDict):
    recipient: str
    decision: str
    status: str


def prepare(state: EmailState) -> dict:
    return {"recipient": "equipo@empresa.com", "status": "prepared"}


def approve(state: EmailState) -> dict:
    decision = interrupt(
        f"Enviar email a {state['recipient']}. "
        "¿Proceder? [sí / no / editar:<destinatario>]"
    )
    return {"decision": decision}


def execute(state: EmailState) -> dict:
    d = state["decision"]
    if d == "no":
        return {"status": "cancelled"}
    if d.startswith("editar:"):
        new = d.replace("editar:", "").strip()
        return {"recipient": new, "status": f"sent to {new}"}
    return {"status": f"sent to {state['recipient']}"}


builder = StateGraph(EmailState)
builder.add_node("prepare", prepare)
builder.add_node("approve", approve)
builder.add_node("execute", execute)
builder.add_edge(START, "prepare")
builder.add_edge("prepare", "approve")
builder.add_edge("approve", "execute")
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

init = {"recipient": "", "decision": "", "status": ""}

for decision, tid in [("sí", "ex1_yes"), ("no", "ex1_no"), ("editar:jefe@empresa.com", "ex1_edit")]:
    cfg = {"configurable": {"thread_id": tid}}
    graph.invoke(init, cfg)
    r = graph.invoke(Command(resume=decision), cfg)
    print(f"'{decision}' → {r['status']}")

# Output esperado:
# 'sí' → sent to equipo@empresa.com
# 'no' → cancelled
# 'editar:jefe@empresa.com' → sent to jefe@empresa.com

Ejercicio 2: Múltiples interrupts secuenciales (Medio)

Crea un grafo de compra con 3 nodos, cada uno con su propio interrupt: confirm_item ("¿Confirmas: Laptop?"), confirm_payment ("¿Pagas $999?"), confirm_shipping ("¿Envío a CDMX?"). Invoca y resume 3 veces. Verifica el resultado final.

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from typing import TypedDict


class PurchaseState(TypedDict):
    item: str
    price: float
    city: str
    item_ok: bool
    payment_ok: bool
    shipping_ok: bool
    status: str


def confirm_item(state: PurchaseState) -> dict:
    r = interrupt(f"¿Confirmas el producto: {state['item']}? [sí/no]")
    return {"item_ok": r == "sí"}


def confirm_payment(state: PurchaseState) -> dict:
    if not state["item_ok"]:
        return {"status": "cancelled_at_item"}
    r = interrupt(f"¿Confirmas el pago: ${state['price']:.2f}? [sí/no]")
    return {"payment_ok": r == "sí"}


def confirm_shipping(state: PurchaseState) -> dict:
    if not state["payment_ok"]:
        return {"status": "cancelled_at_payment"}
    r = interrupt(f"¿Envío a: {state['city']}? [sí/no]")
    ok = r == "sí"
    return {"shipping_ok": ok, "status": "completed" if ok else "cancelled_at_shipping"}


builder = StateGraph(PurchaseState)
builder.add_node("confirm_item", confirm_item)
builder.add_node("confirm_payment", confirm_payment)
builder.add_node("confirm_shipping", confirm_shipping)
builder.add_edge(START, "confirm_item")
builder.add_edge("confirm_item", "confirm_payment")
builder.add_edge("confirm_payment", "confirm_shipping")
builder.add_edge("confirm_shipping", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "ex2_purchase"}}
init = {
    "item": "Laptop Pro", "price": 999.00, "city": "Ciudad de México",
    "item_ok": False, "payment_ok": False, "shipping_ok": False, "status": ""
}

r1 = graph.invoke(init, config)
print(f"1: {r1['__interrupt__'][0].value}")

r2 = graph.invoke(Command(resume="sí"), config)
print(f"2: {r2['__interrupt__'][0].value}")

r3 = graph.invoke(Command(resume="sí"), config)
print(f"3: {r3['__interrupt__'][0].value}")

r4 = graph.invoke(Command(resume="sí"), config)
assert r4["status"] == "completed"
print(f"\n✅ Compra completada: {r4['status']}")
# Output esperado:
# 1: ¿Confirmas el producto: Laptop Pro? [sí/no]
# 2: ¿Confirmas el pago: $999.00? [sí/no]
# 3: ¿Envío a: Ciudad de México? [sí/no]
#
# ✅ Compra completada: completed

Ejercicio 3: Validación con retry loop (Medio)

Crea un nodo que pida al humano un número entre 1 y 10 usando interrupt() en un loop. Si el valor es inválido, re-pregunta. Prueba con "abc" (no numérico), 50 (fuera de rango), y 7 (válido).

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Optional


class NumberState(TypedDict):
    chosen_number: Optional[int]


def ask_number(state: NumberState) -> dict:
    prompt = "Elige un número entre 1 y 10:"

    while True:
        answer = interrupt(prompt)
        try:
            num = int(answer)
        except (ValueError, TypeError):
            prompt = f"'{answer}' no es un número. Elige entre 1 y 10:"
            continue
        if 1 <= num <= 10:
            return {"chosen_number": num}
        prompt = f"{num} está fuera de rango. Elige entre 1 y 10:"


builder = StateGraph(NumberState)
builder.add_node("ask", ask_number)
builder.add_edge(START, "ask")
builder.add_edge("ask", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "ex3_number"}}

r = graph.invoke({"chosen_number": None}, config)
print(f"Pregunta: {r['__interrupt__'][0].value}")

r = graph.invoke(Command(resume="abc"), config)
print(f"Re-pregunta: {r['__interrupt__'][0].value}")

r = graph.invoke(Command(resume="50"), config)
print(f"Re-pregunta: {r['__interrupt__'][0].value}")

r = graph.invoke(Command(resume="7"), config)
assert r["chosen_number"] == 7
print(f"✅ Número elegido: {r['chosen_number']}")
# Output esperado:
# Pregunta: Elige un número entre 1 y 10:
# Re-pregunta: 'abc' no es un número. Elige entre 1 y 10:
# Re-pregunta: 50 está fuera de rango. Elige entre 1 y 10:
# ✅ Número elegido: 7

Ejercicio 4: Routing con Command(goto=) — tres destinos (Avanzado)

Crea un grafo con un nodo gate que usa interrupt() y retorna Command(goto=...). Si el humano responde "aprobar", salta a fast_track. Si "rechazar", a review. Si "escalar", a escalate. Prueba los 3 caminos con thread_ids diferentes.

from typing import Literal, TypedDict
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END


class RequestState(TypedDict):
    request: str
    status: str


def gate(state: RequestState) -> Command[Literal["fast_track", "review", "escalate"]]:
    decision = interrupt({
        "request": state["request"],
        "options": ["aprobar", "rechazar", "escalar"]
    })
    routes = {"aprobar": "fast_track", "rechazar": "review", "escalar": "escalate"}
    return Command(goto=routes.get(decision, "review"))


def fast_track(state: RequestState) -> dict:
    return {"status": "approved_fast"}


def review(state: RequestState) -> dict:
    return {"status": "sent_to_review"}


def escalate(state: RequestState) -> dict:
    return {"status": "escalated"}


builder = StateGraph(RequestState)
builder.add_node("gate", gate)
builder.add_node("fast_track", fast_track)
builder.add_node("review", review)
builder.add_node("escalate", escalate)
builder.add_edge(START, "gate")
builder.add_edge("fast_track", END)
builder.add_edge("review", END)
builder.add_edge("escalate", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

init = {"request": "Acceso a producción", "status": ""}

for decision, tid, expected in [
    ("aprobar", "ex4_a", "approved_fast"),
    ("rechazar", "ex4_r", "sent_to_review"),
    ("escalar", "ex4_e", "escalated"),
]:
    cfg = {"configurable": {"thread_id": tid}}
    graph.invoke(init, cfg)
    r = graph.invoke(Command(resume=decision), cfg)
    assert r["status"] == expected
    print(f"'{decision}' → {r['status']}")

print("\n✅ Routing con 3 destinos verificado")
# Output esperado:
# 'aprobar' → approved_fast
# 'rechazar' → sent_to_review
# 'escalar' → escalated
#
# ✅ Routing con 3 destinos verificado

Resumen

• interrupt(value) pausa el grafo y envía value al humano. El grafo guarda su estado en un checkpoint y espera indefinidamente. Es la base de todo HITL en LangGraph
• Command(resume=value) reanuda el grafo. El valor se convierte en el retorno de interrupt() dentro del nodo. Puede ser cualquier dato JSON-serializable
• Checkpointer es obligatorio. Sin checkpointer, interrupt() no puede guardar el estado. Si usas interrupt(), configura MemorySaver (dev) o PostgresSaver (prod)
• El nodo se re-ejecuta desde el principio al resumir. El código antes de interrupt() corre de nuevo. El código pre-interrupt debe ser idempotente. Side effects van después del interrupt
• Un grafo puede tener múltiples interrupts en nodos diferentes. Cada Command(resume=) avanza al siguiente interrupt
• Command(goto=) permite routing post-aprobación. El nodo decide dinámicamente a qué nodo saltar basándose en la decisión humana
• Validación con retry: un while True + interrupt() crea loops que re-preguntan hasta obtener una respuesta válida

Próxima cápsula: Approval Gates: Validar Antes de Actuar — el patrón más común de HITL. Implementarás gates que piden permiso antes de acciones costosas, con estimación de costos, niveles de aprobación, y routing condicional.

Recursos adicionales

1. LangGraph — Interrupts — Documentación oficial de interrupt() y Command(resume=): flujo, reglas, y anti-patterns
2. interrupt() API Reference — Referencia de la API con firma, parámetros, y ejemplos
3. How to add human-in-the-loop — Guía práctica con patrones de approval, review, y feedback
4. LangGraph — Persistence — Documentación de checkpointing que habilita interrupts
5. How to wait for user input — Patrón específico para esperar input humano durante ejecución

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