17. Persistencia de estado y checkpointing

En el capítulo 15 reconstruimos el bucle del agente como un StateGraph, y en el capítulo 16 usamos componentes preconstruidos y enrutamiento multirrama para construir algo más elaborado. Cada grafo que hemos escrito hasta ahora comparte una limitación: el grafo no conserva su estado.

El grafo conserva y gestiona el estado durante una única llamada a invoke(), y ni un momento más. Cuando lo llamas, LangGraph crea un estado nuevo, ejecuta los nodos, fusiona cada valor de retorno en el estado según las reglas del reducer y devuelve el estado final al llamador. Una vez que ha entregado ese estado, el grafo ya no lo recuerda. La siguiente llamada a invoke() empieza desde un estado completamente nuevo, sin conexión con la llamada anterior.

De aquí surgen dos problemas. Primero, messages también forma parte del estado, así que el agente no puede recordar nada de lo que dijiste antes. Segundo, si una ejecución falla a mitad de camino, todo lo que había logrado hasta ese punto desaparece. Digamos que el tercer nodo lanza una excepción: los resultados producidos por los dos primeros nodos se van con ella, y tienes que empezar de nuevo desde el principio. Allá en la sección 15.1 mencionamos "recuperarse de interrupciones" como una de las razones para recurrir a LangGraph: este es el problema que teníamos en mente.

LangGraph resuelve esto a nivel del framework. Adjunta un checkpointer a un grafo y LangGraph guardará automáticamente una instantánea del estado en cada paso de la ejecución. Esas instantáneas guardadas sobreviven a la llamada a invoke(), de modo que la siguiente llamada puede retomar donde se quedó la anterior. Esta propiedad —que el estado sobreviva más allá de una única ejecución— se llama persistencia.

En realidad ya has usado un checkpointer antes. En el capítulo 11, cuando dotamos al agente de RAG conversacional de memoria multiturno, pasamos create_agent(..., checkpointer=InMemorySaver()) y un thread_id. En aquel momento, todo lo que necesitabas saber era que el checkpointer mantiene el historial de conversación por thread_id; nunca explicamos cómo. Y en el capítulo 16, cuando presentamos el parámetro checkpointer, dijimos "cubriremos cómo funciona esto en el capítulo 17". Este es ese capítulo.

Viene en tres partes. En la 17.1 adjuntamos un checkpointer a un grafo y mantenemos conversaciones multiturno con thread_id. En la 17.2 abrimos los checkpoints guardados para ver qué sabía el agente en cualquier momento dado: los checkpoints son tu principal herramienta para rastrear por qué un agente se comportó mal. En la 17.3 tomamos un grafo que falló a mitad de ejecución y lo reanudamos desde donde se detuvo, en lugar de desde el principio.

17.1) Llevar el estado a través de las llamadas

17.1.1) Un grafo que olvida

Abrimos este capítulo diciendo que un grafo no conserva su estado. Confirmémoslo en código.

El grafo de abajo tiene la misma forma que el grafo say_hello de la sección 15.2. La única diferencia es que el nodo devuelve una respuesta del LLM en lugar de una cadena fija.

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
 
def llm_call(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
 
graph = builder.compile()   # sin checkpointer

Ahora tengamos una conversación de dos turnos. Le decimos al modelo nuestro nombre en la primera llamada y luego le preguntamos cuál es nuestro nombre en la segunda.

# Primera llamada — damos nuestro nombre
graph.invoke({"messages": [{"role": "user", "content": "Hola, me llamo Bob."}]})
 
# Segunda llamada — lo pedimos de vuelta
result = graph.invoke({"messages": [{"role": "user", "content": "¿Cómo me llamo?"}]})
print(result["messages"][-1].content)

Salida:

Lo siento, pero no tengo tu nombre. ¿Podrías decirme cuál es?

El único mensaje que el modelo recibió en la segunda llamada fue "¿Cómo me llamo?". El estado de la primera llamada ya no está en el grafo, así que los mensajes anteriores —los que llevaban el nombre— nunca llegaron al modelo.

Por supuesto, podríamos arreglarlo nosotros mismos. Conservar los messages devueltos por la primera llamada y pasarlos junto con la segunda. Así es exactamente como gestionamos el historial de conversación allá en el capítulo 8. Pero entonces estaríamos escribiendo nuestro propio código para almacenar y recuperar el historial de cada conversación y de cada usuario. Ese es el trabajo que un checkpointer te quita de las manos.

17.1.2) Checkpoints y checkpointers

Un checkpointer es un objeto cuya función es guardar el estado. Creas una instancia —InMemorySaver(), por ejemplo— y se la pasas a builder.compile(checkpointer=...) para adjuntarla a tu grafo.

Una vez que un checkpointer está adjunto, el grafo copia todo el estado y lo guarda a medida que avanza la ejecución. Cada una de esas copias guardadas se llama un checkpoint. Piénsalo como una fotografía: todo el estado en ese instante, preservado exactamente tal como era.

Un autoguardado en un videojuego es la imagen mental adecuada. El juego registra silenciosamente tu progreso cada vez que pasas un punto significativo, para que puedas salir y volver más tarde, o morir sin tener que empezar de nuevo desde el principio. Un checkpointer hace exactamente esto para un grafo.

Entonces, ¿cuándo es un "punto significativo"? LangGraph divide la ejecución de un grafo en etapas, y cada etapa se llama un super-step. Se guarda un checkpoint cada vez que termina un super-step.

La razón por la que es un super-step y no simplemente un step es que una única etapa puede ejecutar varios nodos a la vez. En un grafo como el de la sección 17.1.1, donde los nodos forman una línea recta, ejecutar un nodo es un super-step. Pero en un grafo donde varios nodos se ejecutan en paralelo, todos esos nodos juntos conforman un único super-step.

Así que incluso una única llamada a invoke() deja atrás varios checkpoints. Los extraeremos y veremos exactamente qué contiene cada uno en la sección 17.2.

El InMemorySaver que hemos estado usando como ejemplo es el checkpointer más simple que existe. Como sugiere su nombre, almacena checkpoints en la memoria del proceso (RAM). No hay nada que instalar ni nada que configurar, lo que lo hace una buena opción para aprender y para el desarrollo local. La contrapartida es que cada checkpoint guardado desaparece cuando el proceso se reinicia. Veremos las alternativas de producción en la sección 17.1.5.

17.1.3) Añadir un checkpointer

Adjuntar un checkpointer solo requiere dos cosas.

1. Crear una instancia del checkpointer y pasarla a compile().
2. Pasar un config que contenga un thread_id cada vez que llames a invoke().

Enseguida veremos por qué se necesita lo segundo. Por ahora, basta con saber que le indica al checkpointer cuál de sus conversaciones guardadas quieres continuar.

Apliquemos ambas cosas al grafo de la sección 17.1.1.

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
 
def llm_call(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
 
# 1. Crea un checkpointer y pásalo a compile()
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
 
# 2. Pasa un config que lleve un thread_id a invoke()
config = {"configurable": {"thread_id": "1"}}
 
graph.invoke(
    {"messages": [{"role": "user", "content": "Hola, me llamo Bob."}]},
    config,
)
result = graph.invoke(
    {"messages": [{"role": "user", "content": "¿Cómo me llamo?"}]},
    config,
)
print(result["messages"][-1].content)

Salida:

Te llamas Bob.

Las mismas dos llamadas que en la sección 17.1.1, y un resultado diferente. Esta vez el nombre se mantiene.

Aquí está el porqué. Un grafo con un checkpointer adjunto carga el estado guardado antes de ejecutar el nodo llm_call. Ese estado ya contiene el primer intercambio. El nuevo mensaje que pasamos se fusiona entonces en él. Como vimos en la sección 15.2.2, el campo messages lleva el reducer add_messages, así que el nuevo mensaje se añade a la lista existente. El modelo termina recibiendo tres mensajes: el saludo, su propia primera respuesta y la nueva pregunta.

Enviamos solo el nuevo mensaje, y LangGraph carga la conversación anterior desde el último checkpoint. El historial de conversación que gestionamos a mano en el capítulo 8 ahora lo gestiona LangGraph.

17.1.4) thread_id: el identificador que separa conversaciones

Un thread_id es un identificador que distingue una conversación de otra. Tú eliges el valor. Usamos "1" en la sección 17.1.3, pero cualquier cadena sirve. Llama al grafo con el mismo thread_id y continúas esa conversación; llámalo con uno diferente e inicias una conversación separada.

Confirmémoslo. Haremos que Alice y Bob mantengan conversaciones diferentes a través del mismo grafo.

def send(thread_id: str, text: str) -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(
        {"messages": [{"role": "user", "content": text}]},
        config,
    )
    return result["messages"][-1].content
 
# La conversación de Alice
send("alice", "Mi color favorito es el verde azulado.")
 
# La conversación de Bob — un thread_id diferente
send("bob", "Mi color favorito es el naranja.")
 
# Preguntamos a cada uno de nuevo
print("Alice:", send("alice", "¿Cuál es mi color favorito?"))
print("Bob:  ", send("bob", "¿Cuál es mi color favorito?"))

Salida:

Alice: Tu color favorito es el verde azulado.
Bob:   Tu color favorito es el naranja.

Ambas conversaciones pasaron por el mismo objeto graph y el mismo checkpointer, y sin embargo nunca se mezclaron. El thread_id es la clave primaria que el checkpointer usa para almacenar y buscar el estado. Claves diferentes, almacenamiento completamente separado.

Entonces, ¿qué pasa si dejas el valor fuera?

graph.invoke({"messages": [{"role": "user", "content": "Hola"}]})

Salida:

ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id

El grafo no se ejecuta en absoluto. Una vez que un checkpointer está adjunto, thread_id no es opcional: es obligatorio.

Dado lo que acabamos de ver, eso tiene sentido. Antes de ejecutar un nodo, el checkpointer tiene que cargar el estado guardado, y el thread_id es lo que le indica el estado de qué conversación cargar.

Esta es la forma básica de un servicio de chatbot: un grafo, un checkpointer y un thread_id por usuario o por sala de chat.

17.1.5) Los límites de InMemorySaver y las alternativas de producción

Dijimos antes que InMemorySaver mantiene los checkpoints en memoria. Dos limitaciones vienen con esa elección.

Reinicia el proceso y todo desaparece. Redespliega el servicio o reinicia el servidor, y toda conversación acumulada hasta ese momento desaparece.

Procesos separados no pueden compartirlo. Un servicio real reparte las solicitudes entrantes entre varios procesos. Cada proceso tiene su propia memoria, así que una conversación guardada por el proceso A es invisible para el proceso B. Un usuario puede enviar el mismo thread_id cada vez y aun así ver la conversación desmoronarse, dependiendo de qué proceso resulte que atienda la solicitud.

Por eso en producción se usan checkpointers que almacenan los checkpoints en una base de datos.

• SqliteSaver / AsyncSqliteSaver (langgraph-checkpoint-sqlite) — almacena todo en un único archivo. Una buena opción para un servicio pequeño que se ejecuta en un solo servidor, o para un prototipo local.
• PostgresSaver / AsyncPostgresSaver (langgraph-checkpoint-postgres) — almacena los checkpoints en un servidor de base de datos. Añade más servidores y cada proceso seguirá viendo los mismos checkpoints. Este es el checkpointer que la documentación de LangGraph recomienda para producción.

Todos ellos implementan la misma interfaz que InMemorySaver. El código de tu grafo, tus nodos y la manera en que usas thread_id se quedan exactamente como están. Lo único que cambia es cómo creas el checkpointer.

create_agent, que conociste en el capítulo 16, usa los checkpointers de la misma manera. Pasa un checkpointer a su parámetro checkpointer, y pasa un config que lleve un thread_id a invoke(). Esto es lo que hizo que el agente de RAG conversacional del capítulo 11 recordara turnos anteriores.

Seguiremos usando InMemorySaver durante el resto de este capítulo. Sea donde sea que los checkpoints acaben viviendo, la manera de trabajar con ellos es la misma.

17.2) Inspeccionar el estado y depurar

En la sección 17.1.2 dijimos que el checkpointer guarda el estado en cada super-step. Extraigamos esos checkpoints y examinémoslos.

Abrir los checkpoints guardados no es mera curiosidad. Los agentes se comportan mal. Llaman a herramientas en un bucle que nunca termina, pierden el hilo de turnos anteriores, toman una rama que nunca esperabas. Para averiguar por qué, necesitas saber cómo se veía el estado en ese momento, y los checkpoints tienen la respuesta.

LangGraph te da dos métodos.

• graph.get_state(config) — devuelve el checkpoint más reciente de esa conversación.
• graph.get_state_history(config) — devuelve todos los checkpoints de esa conversación, del más nuevo al más antiguo.

Ambos necesitan un thread_id en el config, ya que tienen que saber de quién son los checkpoints que estás pidiendo.

Los checkpoints que estos dos métodos devuelven se representan como objetos StateSnapshot.

17.2.1) StateSnapshot: qué hay dentro de un checkpoint

Veamos qué contiene realmente un checkpoint. No usaremos un LLM para este ejemplo. Un LLM devuelve algo diferente cada vez, lo que no ayuda cuando queremos recorrer cada campo uno por uno. En su lugar, construiremos un grafo pequeño que devuelve valores fijos.

El estado tendrá dos tipos de campos. foo no tiene reducer, así que se sobrescribe; bar tiene uno, así que se acumula. Es la misma disposición que el reducer add_messages que adjuntamos a messages en la sección 15.2.2; aquí usamos el operator.add de Python como reducer para concatenar listas.

from operator import add
from typing_extensions import TypedDict, Annotated
 
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
 
class State(TypedDict):
    foo: str                        # sin reducer → sobrescribe
    bar: Annotated[list[str], add]  # reducer add → acumula
 
def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}
 
def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}
 
builder = StateGraph(State)
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
 
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": "", "bar": []}, config)
 
snapshot = graph.get_state(config)
print(snapshot)

Salida:

StateSnapshot(
    values={'foo': 'b', 'bar': ['a', 'b']},
    next=(),
    config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
                             'checkpoint_id': '1f17da03-9654-65d8-8002-9e59231bb481'}},
    metadata={'source': 'loop', 'step': 2, 'parents': {}},
    created_at='2026-07-12T03:17:33.637368+00:00',
    parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
                                    'checkpoint_id': '1f17da03-9653-6ca0-8001-7c24c8ec66f2'}},
    tasks=(),
    interrupts=()
)

Ocho campos. Vamos uno por uno.

• values — el estado tal como estaba en este checkpoint. bar es ['a', 'b'] porque el reducer add acumuló lo que devolvieron ambos nodos; foo es 'b' porque no tiene reducer, así que ganó la última escritura. Este es el campo que responde a "¿cómo se veía el estado en ese momento?"
• next — una tupla de nombres de nodos que se ejecutarán después de este checkpoint. Una tupla vacía () significa que no queda nada por ejecutar, es decir, que el grafo terminó. ('node_b',) significa que node_b todavía está por delante.
• config — la dirección de este checkpoint. thread_id identifica la conversación, y checkpoint_id identifica qué momento dentro de ella. LangGraph asigna el checkpoint_id automáticamente cada vez que guarda un checkpoint.
• metadata — información de control sobre la ejecución. source te dice de dónde vino el checkpoint: "input" significa que se construyó a partir de la entrada que le pasaste a invoke(), y "loop" significa que se produjo mientras el grafo se ejecutaba. step es el número de super-step.
• created_at — cuándo se guardó el checkpoint. Útil cuando estás cuadrando cosas contra tus logs.
• parent_config — el config del checkpoint inmediatamente anterior a este. Síguelo y podrás retroceder a través de la ejecución. Es None para el primer checkpoint de todos.
• tasks — el registro de ejecución de los nodos listados en next. En el momento en que se guarda un checkpoint, esos nodos todavía no se han ejecutado; una vez que lo hacen, su resultado se adjunta a este checkpoint. Un nodo que tuvo éxito deja su valor de retorno en result, y un nodo que falló deja su excepción en error.
• interrupts — dónde el grafo se pausó para devolver el control a una persona. LangGraph puede detenerse a mitad de ejecución y esperar a que alguien apruebe un paso o proporcione un valor, y este campo registra esas pausas.

Los campos se leen como atributos. metadata es un diccionario, así que extraes valores de él con una clave.

snapshot = graph.get_state(config)
 
print(snapshot.values)            # {'foo': 'b', 'bar': ['a', 'b']}
print(snapshot.next)              # ()
print(snapshot.metadata["step"])  # 2

De todos ellos, el que más consultarás mientras depuras es next. Si next no está vacío, el grafo no llegó hasta el final: se detuvo en algún punto intermedio. Y cuando reanudemos un grafo fallido en la sección 17.3, este campo es donde empezamos.

17.2.2) Recorrer el historial de checkpoints

get_state() te muestra solo el checkpoint más reciente. Pero depurar a menudo significa preguntar "¿cómo acabamos aquí?", y para eso necesitas toda la trayectoria de la ejecución. get_state_history() te la da.

for snap in graph.get_state_history(config):
    print(f"step={snap.metadata['step']:>2}  "
          f"next={str(snap.next):<16}  values={snap.values}")

Salida:

step= 2  next=()                values={'foo': 'b', 'bar': ['a', 'b']}
step= 1  next=('node_b',)       values={'foo': 'a', 'bar': ['a']}
step= 0  next=('node_a',)       values={'foo': '', 'bar': []}
step=-1  next=('__start__',)    values={'bar': []}

El checkpoint más nuevo viene primero, así que trabaja de abajo hacia arriba para seguir la ejecución en orden.

• step -1 — justo después de que invoke() recibiera la entrada. Fíjate en que el {"foo": "", "bar": []} que pasamos no aparece en values. Meter la entrada en el estado es en sí mismo una etapa, y esa etapa aún no se ha ejecutado. El __start__ en next es el nodo interno que lo hace.

  bar aparece como [], pero ese no es el valor que pasamos. Un campo con un reducer arranca con un valor vacío para que las escrituras se acumulen en él. foo no tiene reducer, así que no tiene ningún valor inicial, y por eso no aparece aquí.

• step 0 — __start__ se ha ejecutado y la entrada ya está en el estado. foo='' y bar=[] son los valores que pasamos. node_a es el siguiente.
• step 1 — el resultado de ejecutar node_a. foo ahora es 'a' y bar es ['a'], con node_b como siguiente.
• step 2 — el resultado de ejecutar node_b. next está vacío, así que el grafo terminó.

Un next no vacío en el step 0 o el step 1 no significa que el grafo se detuviera ahí. Un checkpoint tomado mientras el grafo todavía se ejecuta tiene naturalmente un nodo alineado como siguiente. Cuando en la sección 17.2.1 dijimos "un next no vacío significa que el grafo se detuvo", estábamos hablando de el checkpoint en el punto donde algo salió mal. En medio del historial, next simplemente te muestra qué camino tomó el grafo.

17.3) Reanudar desde donde falló

Como el estado se guarda al final de cada super-step, un fallo a mitad de ejecución no se lleva consigo el trabajo completado: sigue ahí en los checkpoints. No hay razón para empezar de nuevo desde el principio. Retomas desde donde se detuvo.

17.3.1) Reanudar con invoke(None, config)

Reanudar es simple: pasa None donde va la entrada.

graph.invoke(None, config)

Significa "no hay nueva entrada; continúa desde el estado guardado." El config todavía necesita un thread_id, por supuesto, ya que LangGraph tiene que saber qué conversación continuar.

Preparemos un fallo y reanudemos desde él. Construiremos un grafo de dos nodos cuyo segundo nodo falla solo en su primera ejecución. Tiene que tener éxito en el reintento, o nunca veríamos la reanudación funcionar de verdad.

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
 
class State(TypedDict):
    step_1_done: bool
    step_2_done: bool
 
first_try = True   # bandera para hacer que la primera ejecución falle
 
def step_1(state: State):
    print("step_1 ejecutándose (trabajo costoso)")
    return {"step_1_done": True}
 
def step_2(state: State):
    global first_try
    if first_try:
        first_try = False
        print("step_2 falló (timeout de la API)")
        raise RuntimeError("External API timed out")
    print("step_2 ejecutándose")
    return {"step_2_done": True}
 
builder = StateGraph(State)
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", END)
 
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "job-42"}}
 
try:
    graph.invoke({"step_1_done": False, "step_2_done": False}, config)
except RuntimeError as e:
    print("Falló:", e)

Salida:

step_1 ejecutándose (trabajo costoso)
step_2 falló (timeout de la API)
Falló: External API timed out

step_1 tuvo éxito y step_2 lanzó una excepción. Averigüemos dónde se detuvo el grafo, usando el get_state() que aprendimos en la sección 17.2.

snapshot = graph.get_state(config)
print("next   =", snapshot.next)
print("values =", snapshot.values)

Salida:

next   = ('step_2',)
values = {'step_1_done': True, 'step_2_done': False}

next es ('step_2',), lo que nos dice que el grafo se detuvo en medio de la ejecución de step_2. Y en values, step_1_done es True: el resultado de step_1 sigue ahí en el checkpoint.

Ahora reanudamos con None.

result = graph.invoke(None, config)
print("Final =", result)

Salida:

step_2 ejecutándose
Final = {'step_1_done': True, 'step_2_done': True}

step_1 ejecutándose (trabajo costoso) nunca se imprimió. step_1 no se ejecutó una segunda vez. LangGraph cargó el estado guardado y retomó en step_2. No pagamos por ese costoso primer paso dos veces.

17.3.2) Qué tener en cuenta al reanudar

Cuando reanudas, el nodo que falló se ejecuta de nuevo. Si ese nodo llama a un LLM o accede a una API externa, esas llamadas también ocurren de nuevo, y podrían devolver algo diferente.

Ahí está la trampa. Si step_2 envió un correo electrónico y luego falló, reanudar envía un segundo correo. LangGraph solo garantiza que no volverá a ejecutar los nodos que tuvieron éxito.

Así que cualquier nodo que pueda ejecutarse de nuevo tiene que ser idempotente: hacer lo mismo dos veces debería dejarte en el mismo lugar. Comprueba si el correo ya se ha enviado antes de enviarlo; pon una clave única en la tabla de la base de datos para que no pueda ocurrir una inserción duplicada.