LangGraph — Módulo 7: human-in-the-loop

Séptimo módulo del Curso completo de LangGraph. Ya tienes un agente que usa herramientas y que recuerda. Ahora la pregunta incómoda: ¿de verdad quieres que ejecute sola la tool que emite una factura, borra un registro o manda un correo a un cliente? Este módulo trata de poner un humano en medio.

La tesis del módulo

Un agente en producción no necesita más autonomía: necesita mejor supervisión. interrupt() pausa el grafo en mitad de un nodo, guarda el estado y te devuelve el control. Cuando respondes, la ejecución continúa desde ahí. Es la diferencia entre una demo y algo que puedes poner frente a un cliente.

Prerrequisito: checkpointer obligatorio

Pausar el grafo significa guardarlo y retomarlo después — posiblemente en otro proceso, minutos u horas más tarde. Eso es exactamente lo que hace un checkpointer. Sin checkpointer, interrupt() no funciona. Por eso el Módulo 6 venía antes.

interrupt(): pausar y preguntar

interrupt() se llama dentro de un nodo. Detiene la ejecución justo ahí y devuelve al llamador el valor que le pasaste (la pregunta para el humano).

from langgraph.types import interrupt

def revisar_envio(state: State) -> dict:
    respuesta = interrupt({
        "pregunta": "¿Autorizas enviar este correo?",
        "borrador": state["borrador"],
    })
    return {"aprobado": respuesta == "si"}

Al ejecutar, el grafo se detiene y el resultado trae la clave __interrupt__:

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "envio-1"}}

resultado = graph.invoke({"borrador": "Estimado cliente..."}, config)

if "__interrupt__" in resultado:
    print(resultado["__interrupt__"][0].value)
    # → {'pregunta': '¿Autorizas enviar este correo?', 'borrador': 'Estimado cliente...'}

El grafo está congelado. El proceso puede terminar, el servidor puede reiniciarse: el estado está en el checkpointer, esperando.

Command(resume=...): retomar

Para continuar, invocas el mismo thread pasando un Command con la respuesta:

from langgraph.types import Command

final = graph.invoke(Command(resume="si"), config)
print(final["aprobado"])   # → True

El valor que pones en resume= es exactamente lo que devuelve interrupt() dentro del nodo. Ese es todo el contrato.

 invoke(...) ──► nodo ──► interrupt() ──► PAUSA
                                            │ estado guardado
             minutos, horas, otro proceso   │
                                            ▼
 invoke(Command(resume="si"))
      ──► interrupt() devuelve "si" ──► sigue ──► END

El detalle que confunde a todo el mundo

Cuando retomas, el nodo se ejecuta desde el principio otra vez, no desde la línea del interrupt. LangGraph re-ejecuta el nodo completo y, al llegar de nuevo al interrupt(), en vez de pausar devuelve el valor que enviaste.

Esto tiene una consecuencia práctica que hay que tener presente:

def nodo(state: State) -> dict:
    enviar_correo(state["borrador"])          # ⚠️ se ejecutaría DOS veces
    ok = interrupt("¿Confirmas?")
    return {"ok": ok}

Todo lo que ocurra antes del interrupt() corre dos veces. La regla: pon los efectos secundarios después del interrupt, o hazlos idempotentes.

def nodo(state: State) -> dict:
    ok = interrupt("¿Confirmas?")             # ✅ primero preguntar
    if ok:
        enviar_correo(state["borrador"])      #    después actuar
    return {"ok": ok}

Otra regla derivada: nunca envuelvas un interrupt() en un try/except genérico. La pausa se implementa lanzando una excepción interna; si la capturas, el interrupt no llega a la superficie y el grafo se comporta de forma inexplicable.

El caso real: aprobar una tool

Este es el patrón que vas a usar de verdad. Retomamos el agente ReAct del Módulo 5 y metemos un nodo de aprobación entre el modelo y las herramientas.

from typing import Literal
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode
from langgraph.types import interrupt
from langchain.messages import ToolMessage

def aprobar(state: MessagesState) -> dict:
    """Muestra al humano la tool que el modelo quiere ejecutar."""
    llamada = state["messages"][-1].tool_calls[0]

    decision = interrupt({
        "accion": "aprobar_tool",
        "tool": llamada["name"],
        "argumentos": llamada["args"],
    })

    if decision == "aprobar":
        return {}   # sin cambios: el flujo sigue hacia "tools"

    # Rechazada: respondemos la tool call con una explicación,
    # para que el modelo sepa qué pasó y pueda reaccionar.
    return {"messages": [ToolMessage(
        content="El usuario rechazó esta acción. Ofrece una alternativa.",
        tool_call_id=llamada["id"],
    )]}

Fíjate en el detalle del rechazo: toda tool call necesita su ToolMessage de respuesta. No puedes simplemente saltarte la ejecución y volver al modelo; quedaría un mensaje sin contestar y la API del proveedor lo rechaza. Devolver un ToolMessage explicando el rechazo es además mucho más útil: el agente puede proponer otra cosa.

Ahora el routing. Después de aprobar, si el último mensaje ya es un ToolMessage significa que rechazamos; si no, adelante:

def tras_aprobacion(state: MessagesState) -> Literal["tools", "asistente"]:
    return "asistente" if isinstance(state["messages"][-1], ToolMessage) else "tools"

def ruta(state: MessagesState) -> Literal["aprobar", "__end__"]:
    return "aprobar" if state["messages"][-1].tool_calls else "__end__"

builder = StateGraph(MessagesState)
builder.add_node("asistente", asistente)
builder.add_node("aprobar", aprobar)
builder.add_node("tools", ToolNode(tools))

builder.add_edge(START, "asistente")
builder.add_conditional_edges("asistente", ruta)
builder.add_conditional_edges("aprobar", tras_aprobacion)
builder.add_edge("tools", "asistente")

agente = builder.compile(checkpointer=InMemorySaver())

 START ──► asistente ──► ? ──► aprobar ──► pausa ──► ?
               ▲         │                        ok │
               │         └──► END (sin tools)        ▼
               │                                   tools
               │      rechazo o tool ejecutada       │
               └─────────────────────────────────────┘

Con esto tienes un agente que pide permiso antes de actuar. Y si solo algunas tools son sensibles, condiciona el nodo aprobar por nombre de tool: consultar stock pasa directo, emitir factura pregunta.

Los cuatro usos de un interrupt

El mismo mecanismo cubre cuatro patrones:

1. Aprobar / rechazar — el del ejemplo. resume es un booleano o un string.
2. Editar el estado — muestras el borrador y resume trae la versión corregida.
3. Pedir información — el agente necesita un dato que no tiene y lo pregunta.
4. Revisar los argumentos de una tool — apruebas la acción pero corriges los parámetros.

Todos son interrupt(algo) + Command(resume=algo). Cambia solo qué pones dentro.

Breakpoints estáticos: pausar para depurar

Existe una variante que no requiere tocar tus nodos: pedirle al grafo que se detenga antes o después de un nodo determinado.

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["tools"],
)

graph.invoke(entrada, config)   # se detiene antes de "tools"
# ... inspeccionas con graph.get_state(config), incluso lo corriges con update_state ...
graph.invoke(None, config)      # None = "continúa desde donde quedaste"

Ese invoke(None, config) es la forma de reanudar un breakpoint estático (a diferencia de interrupt(), que se reanuda con Command(resume=...)).

Los breakpoints estáticos son ideales para depurar: ver el estado en un punto exacto sin ensuciar el código. Para lógica de negocio real —aprobaciones que forman parte del producto— usa interrupt(), que es dinámico y te deja llevar una pregunta concreta al humano.

Errores comunes

• Olvidar el checkpointer. Sin él no hay dónde guardar la pausa.
• Poner efectos secundarios antes del interrupt(). Se ejecutan dos veces.
• Capturar la excepción con try/except. Rompe el mecanismo por completo.
• Reanudar con otro thread_id. El estado pausado vive en su thread; con otro empiezas de cero.
• Rechazar una tool sin devolver ToolMessage. Queda una tool call sin respuesta y el proveedor devuelve error.
• Pasar objetos no serializables en el payload del interrupt: tiene que poder guardarse en el checkpointer.

La frase para llevarse

“La autonomía se demuestra pidiendo permiso. Un agente que sabe cuándo detenerse es el único que puedes dejar suelto en producción.”

Qué viene

En el Módulo 8 cerramos el bloque intermedio con lo que hace que todo esto se sienta como un producto: streaming (mostrar el pensamiento del agente en vivo, token a token) y observabilidad (saber qué pasó cuando algo falla).

Material relacionado

• Glosario de LangGraph
• Módulo 6: persistencia y memoria
• Módulo 5: tools y el patrón ReAct