Agentes Human-in-the-Loop con LangGraph

La promesa de los agentes autónomos siempre es “maneja toda la tarea de principio a fin.” La realidad al desplegar uno es que casi nadie quiere eso realmente para nada que sea consecuente —enviar un email a un cliente, eliminar una fila de base de datos, aprobar un reembolso. Lo que quieren es un agente que haga el trabajo y se detenga justo antes de la parte irreversible, espere a que una persona lo revise, y luego termine. Ese es un problema de ingeniería completamente diferente a “llamar al modelo en un loop”, porque ahora tu agente tiene que sobrevivir pausado por una cantidad desconocida de tiempo —segundos si alguien está mirando un dashboard, días si el aprobador está de vacaciones.

La mayoría de los frameworks de agentes no soportan eso realmente. Soportan “pausa”, en el sentido de que puedes dejar de llamar .invoke(), pero no soportan “reanudar con todo intacto”, porque nunca hubo un lugar durable donde poner el estado. LangGraph es uno de los pocos que trata esto como un escenario de primera clase en lugar de algo que atornillas con una tabla de base de datos y mucho código pegamento personalizado. Este artículo cubre las dos piezas que lo hacen funcionar: la función interrupt(), que pausa un grafo en ejecución a mitad de un nodo, y un checkpointer respaldado por Postgres, que es lo que hace que “reanudar días después” sea verdad en lugar de aspiracional.

Aprenderás:

• Por qué una pausa en memoria no es lo mismo que una durable
• Cómo interrupt() detiene un nodo del grafo y devuelve el control a tu aplicación
• Cómo reanudar un grafo pausado con Command(resume=...)
• Configurar PostgresSaver para que el estado pausado sobreviva un reinicio del proceso
• Qué se escribe realmente en Postgres en cada checkpoint, y por qué eso importa para el debugging
• Cómo diseñar el paso de revisión humana en sí para que no se convierta en el cuello de botella
• Los modos de fallo específicos de grafos pausados durante mucho tiempo, y cómo protegerse contra ellos

Tabla de contenidos

1. Por qué un botón de pausa necesita ser durable
2. La función interrupt
3. Reanudando con Command
4. Checkpointers: qué se guarda realmente
5. Checkpointing con Postgres en producción
6. Reanudar un grafo días después
7. Diseñando el paso de revisión en sí
8. Un flujo de aprobación completo, de extremo a extremo
9. Probando pausa y reanudación localmente
10. Modos de fallo y cómo protegerse
11. Conclusión

Por qué un botón de pausa necesita ser durable

Vale la pena ser preciso sobre qué requiere realmente “human in the loop”, porque es fácil construir algo que parece correcto en una demo y se cae en producción. Una pausa a nivel de demo es: mantener el proceso Python abierto, bloquearse en input(), y continuar una vez que alguien escribe algo. Eso funciona exactamente mientras el proceso siga vivo y una persona esté mirando una terminal.

El human-in-the-loop en producción necesita tres cosas que una llamada bloqueante de input() no puede dar: la capacidad de pausar a través de un reinicio del proceso (los deploys ocurren; los servidores se reprograman), la capacidad de que el evento “reanudar” venga de un proceso completamente diferente —un webhook de un botón de aprobación de Slack, un clic en un dashboard interno— y la capacidad de tener muchos agentes pausados simultáneamente sin mantener un hilo o una conexión abiertos para cada uno. Ese es el requisito real, y es por eso que no es solucionable con un loop y una bandera —necesita que el estado del grafo esté escrito en algún lugar durable, indexado por un ID que puedas buscar después. Si aún no has tomado la decisión de usar un framework basado en grafos en absoluto, ese punto de decisión —chains versus grafos— está cubierto en un artículo dedicado; este asume que ya llegaste a LangGraph y retoma desde ahí.

La función interrupt

La función interrupt() de LangGraph es una función que llamas desde dentro de un nodo. Cuando se alcanza, el grafo detiene la ejecución en ese punto exacto, y lo que sea que hayas pasado a interrupt() se muestra a quien esté ejecutando .invoke().

from langgraph.types import interrupt

def request_approval(state: AgentState):
    decision = interrupt({
        "action": "send_refund",
        "amount": state["refund_amount"],
        "customer": state["customer_id"],
        "reason": state["reason"],
    })
    return {"approved": decision["approved"]}

Cuando request_approval se ejecuta y llega a interrupt(...), la ejecución del grafo se detiene y el payload que pasaste se devuelve al llamador como parte del resultado de la invocación, marcado como un interrupt en lugar de una finalización normal:

result = graph.invoke({"refund_amount": 42.50, "customer_id": "cust_9", "reason": "damaged item"}, config)

if "__interrupt__" in result:
    interrupt_data = result["__interrupt__"][0].value
    print(interrupt_data)
    # {'action': 'send_refund', 'amount': 42.5, 'customer': 'cust_9', 'reason': 'damaged item'}

Nada después de la llamada a interrupt() ha corrido todavía —la función del nodo en sí ni siquiera ha devuelto. Este es el detalle que confunde a la gente: interrupt() no devuelve un valor en esta llamada, eleva el control completamente fuera del grafo. La línea decision = interrupt(...) solo asigna un valor en la siguiente invocación, después de que reanudes —LangGraph reproduce la función del nodo desde el inicio y esta vez interrupt() devuelve inmediatamente con lo que pasaste en la reanudación.

Reanudando con Command

Para reanudar, no llamas .invoke() con un nuevo input —lo llamas con un objeto Command que envuelve el valor que la llamada pausada de interrupt() debería recibir.

from langgraph.types import Command

result = graph.invoke(
    Command(resume={"approved": True}),
    config,  # same config, same thread_id, as the original call
)

El config —específicamente su thread_id— es lo que vincula esta reanudación con la ejecución pausada exacta. LangGraph busca el estado checkpointed para ese hilo, reproduce el nodo interrumpido con interrupt() devolviendo ahora {"approved": True}, y continúa la ejecución desde ahí a través del resto del grafo como de costumbre.

config = {"configurable": {"thread_id": f"refund-{refund_id}"}}

# First call — pauses at request_approval
result = graph.invoke({"refund_amount": 42.50, "customer_id": "cust_9"}, config)

# ... hours or days pass, a human clicks "approve" in a dashboard ...

# Second call — resumes and completes
final = graph.invoke(Command(resume={"approved": True}), config)

Un nodo también puede llamar interrupt() más de una vez a lo largo de la vida de un grafo —una revisión, luego un segundo paso de confirmación más adelante— y cada uno pausa y reanuda independientemente usando el mismo patrón Command(resume=...), con clave en el mismo thread_id.

Checkpointers: qué se guarda realmente

Nada de esto funciona sin un checkpointer, porque interrupt() por sí solo es solo una señal de control flow —es el checkpointer el que realmente persiste el estado para que un proceso diferente, posiblemente minutos o días después, pueda retomarlo. Cada vez que un nodo termina de ejecutarse, el checkpointer escribe una instantánea del estado completo del grafo, más metadatos sobre qué nodo sigue, con clave en thread_id.

from langgraph.checkpoint.memory import MemorySaver

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

MemorySaver está bien para el desarrollo local, pero por debajo es solo un dict de Python —mata el proceso y cada hilo pausado desaparece. Para cualquier cosa que necesite sobrevivir más allá de la vida de un proceso, necesitas un checkpointer respaldado por almacenamiento real.

Checkpointing con Postgres en producción

PostgresSaver (del paquete langgraph-checkpoint-postgres) persiste el mismo estado en una tabla real de Postgres, lo que convierte “pausar” en “pausar indefinidamente, a través de reinicios, reanudable desde cualquier proceso que tenga el connection string.”

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@localhost:5432/agent_db"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # creates the checkpoint tables on first run
    graph = builder.compile(checkpointer=checkpointer)
    result = graph.invoke(initial_state, config)

Para un proceso de servidor de larga duración —una app FastAPI manejando muchos agentes concurrentes— generalmente quieres abrir la conexión una vez al inicio en lugar de por solicitud, y usar la variante async para que pausar un agente no bloquee a otros:

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async def build_graph():
    checkpointer = AsyncPostgresSaver.from_conn_string(DB_URI)
    await checkpointer.setup()
    return builder.compile(checkpointer=checkpointer)

# in a request handler
result = await graph.ainvoke(initial_state, config)

checkpointer.setup() es idempotente y solo necesita ejecutarse una vez por base de datos —en un deploy real normalmente lo ejecutarías como parte de un paso de migración en lugar de en cada inicio de la app, de la misma manera que tratarías cualquier otra migración de esquema.

Reanudar un grafo días después

Vale la pena decirlo explícitamente, porque es el punto entero de usar Postgres en lugar de MemorySaver: el proceso que reanuda un agente no tiene que ser el proceso que lo inició. Un agente de aprobación de reembolso pausado el viernes puede ser reanudado por una instancia de servidor completamente diferente el lunes, siempre que tenga el thread_id y una conexión a la misma base de datos Postgres.

# On a completely different server, days later, triggered by a Slack button click:

async def handle_approval_webhook(refund_id: str, approved: bool):
    checkpointer = AsyncPostgresSaver.from_conn_string(DB_URI)
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": f"refund-{refund_id}"}}
    result = await graph.ainvoke(Command(resume={"approved": approved}), config)
    return result

Esta es la diferencia práctica entre una pausa de juguete y una real: el thread_id es efectivamente un puntero durable dentro de Postgres, y cualquier proceso con acceso a la base de datos puede retomar exactamente donde el grafo lo dejó —historial completo de mensajes, todo el estado intermedio, sin volver a ejecutar nodos que ya completaron. Puedes inspeccionar qué está pausado en cualquier momento con graph.get_state(config), que devuelve el estado checkpointed actual y, críticamente, .next —la tupla de nombres de nodos que se ejecutarán cuando reanudes, para que un dashboard pueda mostrar “esperando en: request_approval” sin adivinar.

Diseñando el paso de revisión en sí

La mecánica anterior es necesaria pero no suficiente —un paso human-in-the-loop que es mecánicamente correcto pero mal diseñado simplemente se convierte en un cuello de botella lento y resentido. Algunas cosas que hacen que el paso de revisión en sí realmente funcione:

• Pásale al revisor todo lo que necesita en el payload del interrupt, no una referencia para ir a buscar. Si interrupt() devuelve {"action": "send_refund", "amount": 42.50, "reason": "damaged item"}, un revisor puede decidir solo con eso. Si devuelve {"refund_id": "r_123"}, acabas de convertir cada aprobación en un proceso de dos pasos.
• Diseña la forma del payload de reanudación para que coincida con lo que el nodo realmente bifurca. No reanudes con un booleano crudo si eventualmente necesitarás también un motivo de rechazo —diseña {"approved": bool, "note": str} desde el inicio, incluso si aún no usas note.
• Delimita temporalmente qué significa “pausado” para tu caso de uso. Una aprobación de reembolso que sigue pendiente después de cinco días es probablemente un fallo del proceso, no un revisor paciente —agrega lógica de expiración (un job programado que verifica updated_at en hilos obsoletos y escala o auto-rechaza) en lugar de dejar que los hilos se queden pausados indefinidamente.
• Registra el payload del interrupt independientemente del checkpoint. Las tablas de checkpoint de Postgres no están pensadas para ser un log de auditoría legible por humanos —escribe una fila separada en una tabla approvals cuando se dispara interrupt(), para que tengas un historial consultable que no dependa de deserializar el formato interno de estado de LangGraph.

Un flujo de aprobación completo, de extremo a extremo

Ayuda ver las piezas anteriores ensambladas en algo más cercano a un servicio real —un agente que redacta un reembolso, pausa para aprobación, y o bien lo emite o notifica al cliente sobre un rechazo, expuesto a través de una pequeña app FastAPI.

from fastapi import FastAPI
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

class RefundState(TypedDict):
    customer_id: str
    amount: float
    reason: str
    approved: bool | None

def draft_refund(state: RefundState):
    # Any model calls or business logic that decide the refund amount happen here.
    return {"amount": round(state["amount"], 2)}

def request_approval(state: RefundState):
    decision = interrupt({
        "action": "approve_refund",
        "customer_id": state["customer_id"],
        "amount": state["amount"],
        "reason": state["reason"],
    })
    return {"approved": decision["approved"]}

def issue_refund(state: RefundState):
    process_payment_refund(state["customer_id"], state["amount"])
    return {}

def notify_rejection(state: RefundState):
    send_customer_email(state["customer_id"], "Your refund request was declined.")
    return {}

def route_on_approval(state: RefundState) -> str:
    return "issue_refund" if state["approved"] else "notify_rejection"

builder = StateGraph(RefundState)
builder.add_node("draft_refund", draft_refund)
builder.add_node("request_approval", request_approval)
builder.add_node("issue_refund", issue_refund)
builder.add_node("notify_rejection", notify_rejection)
builder.add_edge(START, "draft_refund")
builder.add_edge("draft_refund", "request_approval")
builder.add_conditional_edges("request_approval", route_on_approval)
builder.add_edge("issue_refund", END)
builder.add_edge("notify_rejection", END)

app = FastAPI()

@app.post("/refunds/{refund_id}/start")
async def start_refund(refund_id: str, customer_id: str, amount: float, reason: str):
    checkpointer = AsyncPostgresSaver.from_conn_string(DB_URI)
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": f"refund-{refund_id}"}}
    result = await graph.ainvoke(
        {"customer_id": customer_id, "amount": amount, "reason": reason, "approved": None},
        config,
    )
    return {"status": "pending_review", "interrupt": result.get("__interrupt__")}

@app.post("/refunds/{refund_id}/decide")
async def decide_refund(refund_id: str, approved: bool):
    checkpointer = AsyncPostgresSaver.from_conn_string(DB_URI)
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": f"refund-{refund_id}"}}
    result = await graph.ainvoke(Command(resume={"approved": approved}), config)
    return {"status": "completed"}

Nótese que draft_refund —cualquier cosa que pueda llamar a un modelo o a un servicio de precios externo— corre antes del interrupt, e issue_refund —el efecto secundario real— solo corre después de una reanudación, controlado por route_on_approval. Ese orden es deliberado: es la misma regla “sin efectos secundarios antes de un interrupt que podrías repetir” de la sección de modos de fallo a continuación, solo aplicada a un flujo concreto en lugar de planteada abstractamente.

Probando pausa y reanudación localmente

Dado que el punto entero de este patrón es sobrevivir un límite de proceso, vale la pena probar ese límite explícitamente en lugar de confiar en que el comportamiento de MemorySaver en desarrollo se traslade a Postgres en producción. Una prueba local útil: levanta una instancia real de Postgres (Docker está bien), inicia un grafo, mata el proceso Python completamente, inicia un proceso fresco, y reanuda usando solo el thread_id y el connection string —sin memoria compartida, sin imports llevados del primer proceso.

import subprocess, time

# Process 1: start the graph, let it pause at the interrupt, then exit.
subprocess.run(["python", "start_refund.py", "--refund-id", "test-1"])

# Simulate a deploy, a crash, whatever — nothing from process 1 is still running.
time.sleep(1)

# Process 2: a completely fresh interpreter, resuming purely from Postgres state.
subprocess.run(["python", "resume_refund.py", "--refund-id", "test-1", "--approved", "true"])

Si este roundtrip funciona con dos procesos genuinamente separados, has validado la propiedad que realmente importa —no “el código compila”, sino “el estado sobrevive un límite que el código tiene cero visibilidad a través.” Probar pausa/reanudación dentro de un solo proceso (llamando .invoke() dos veces en el mismo script) pasará incluso con un checkpointer que secretamente filtra estado a través de una variable a nivel de módulo, y no lo captará hasta que sea un incidente en producción.

Modos de fallo y cómo protegerse

Reanudar con el thread_id incorrecto. LangGraph no dará un error fuerte si pasas un thread_id para un hilo que no está realmente pausado en un interrupt —en el peor caso, simplemente corre desde cualquier estado que exista (o nada, para un ID fresco). Genera thread_ids determinísticamente desde un ID de entidad real (f"refund-{refund_id}", no un UUID aleatorio que tendrías que almacenar en algún otro lugar) para que no haya una tabla de búsqueda separada que pueda desincronizarse.

Código de nodo que no es seguro de reproducir. Porque LangGraph reproduce la función del nodo interrumpido desde el inicio al reanudar, cualquier código antes de la llamada interrupt() dentro de ese nodo corre de nuevo al reanudar. Si ese código envía un email o escribe a un sistema externo, enviarás doble. Mantén los efectos secundarios después de la llamada interrupt(), no antes, o hazlos idempotentes.

Drift de esquema entre pausa y reanudación. Si cambias tu TypedDict de AgentState —agregas un campo requerido— entre cuando un hilo fue pausado y cuando se reanuda, deserializar el checkpoint antiguo puede romperse. Trata los cambios de esquema de estado de la misma manera que tratarías una migración de base de datos: aditivo por defecto, y maneja los campos faltantes defensivamente en el código del nodo en lugar de asumir que cada checkpoint coincide con el esquema de hoy.

Agotamiento del pool de conexiones bajo muchos hilos pausados. Los hilos pausados no mantienen una conexión de base de datos abierta —son solo filas en Postgres— pero un spike de reanudaciones simultáneas puede agotar tu pool si AsyncPostgresSaver no está configurado con un tamaño de pool razonable para tu concurrencia esperada. Dimensiona el pool para tu tasa de reanudación, no tu tasa de pausa.

Tratar get_state() como sustituto de tu propio audit trail. graph.get_state(config) es genuinamente útil para debuggear interactivamente un único hilo pausado, pero no está construido para responder “muéstrame cada reembolso pendiente de aprobación ahora mismo” —esa es una consulta que el esquema de checkpoint de LangGraph no fue diseñado para servir eficientemente a escala. Si un dashboard necesita listar aprobaciones pendientes, mantenla como una tabla normal a nivel de aplicación (approvals con una columna de estado) escrita junto al checkpoint, en lugar de intentar escanear el estado del checkpoint para ello.

Conclusión

La brecha entre “nuestro agente puede pausar” y “nuestro agente puede pausar una semana y reanudarse en una máquina diferente sin perder nada” es exactamente la brecha entre mantener el estado en una variable de Python y escribirlo en Postgres en cada paso. interrupt() te da el primitivo de control flow —una forma limpia de detener un nodo a mitad de ejecución y devolver el control a quien esté conduciendo el grafo— y PostgresSaver te da la durabilidad que hace que la parte de “reanudar días después” sea verdad en lugar de un truco de demo. Una vez que ambas piezas están en su lugar, la revisión humana deja de ser un caso especial alrededor del cual tu arquitectura tiene que trabajar y se convierte simplemente en otro nodo en el grafo, uno que resulta esperar a una persona en lugar de a un modelo.