InMemorySaver + SqliteSaver: checkpoints en proceso y a nivel de archivo

源码版本1.2.9

Responsabilidades

BaseCheckpointSaver define la forma de la interfaz; los dos saver más usados que la materializan son InMemorySaver y SqliteSaver —el primero mete el estado completo en un defaultdict en memoria del proceso, el segundo descompone los mismos campos en dos tablas SQLite: checkpoints y writes. Juntos cubren los dos escenarios de depuración y persistencia de una sola máquina; para producción a gran escala se cambia a PostgresSaver.

InMemorySaver vive en libs/checkpoint/langgraph/checkpoint/memory/__init__.py (InMemorySaver:33) y usa tres defaultdict para tres cosas: storage guarda el cuerpo del checkpoint + los metadatos + el parent_id; writes mapea (thread_id, ns, checkpoint_id) → (task_id, idx) → registro de escritura; blobs mapea (thread_id, ns, channel, version) → binario serializado (storage / writes / blobs:68-83). Esta estructura de tres cubos se corresponde uno a uno con la de tres tablas de Postgres/SQLite; sólo que aquí se usan dicts en lugar de tablas.

SqliteSaver está en libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py (SqliteSaver:45). Hereda directamente de BaseCheckpointSaver[str], recibe una sqlite3.Connection y crea dos tablas en setup() (<SrcLink path="libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py" lines="129-166" label="setup"/>); put / put_writes usan SQL estándar INSERT OR REPLACE / INSERT OR IGNORE. Trae su propio threading.Lock para serializar el acceso a una única conexión, así que check_same_thread=False es seguro.

Además MemorySaver = InMemorySaver es un alias histórico (MemorySaver alias:631); el from langgraph.checkpoint.memory import MemorySaver tan común en código antiguo entrega la misma clase.

Motivación de diseño

Los tres cubos dict de InMemory se corresponden con tres tablas: los tres campos storage / writes / blobs (data members:68-83) se mapean uno a uno con las tablas checkpoints / writes de SQLite y con las tres tablas checkpoints / checkpoint_writes / checkpoint_blobs de Postgres. InMemory también separa los blobs para que la serialización de un valor de canal ocurra una sola vez cuando cambia la versión, y no en cada put de todo el checkpoint.
PersistentDict para persistencia opcional: el módulo memory además define un PersistentDict(defaultdict) (PersistentDict:634) que vuelca el dict a un archivo con pickle y lo recarga al reiniciar. El __init__ de InMemorySaver acepta un parámetro factory (__init__:85-99) precisamente para que, con factory=PersistentDict, se enganchen automáticamente los gestores de contexto —es el camino intermedio para «uso de depuración pero con estado que sobrevive a reinicios».
SQLite usa WAL + un único lock: la primera sentencia de setup() es PRAGMA journal_mode=WAL (WAL:141) para que las lecturas no bloqueen escrituras; las escrituras siguen serializadas por self.lock, porque sqlite3.Connection no es thread-safe. La docstring lo deja claro: «no escala a múltiples hilos» (note:48-54); para concurrencia, ir a Postgres.
La tabla de escrituras distingue REPLACE y IGNORE: en SQLite, put_writes elige entre INSERT OR REPLACE y INSERT OR IGNORE según todas las escrituras caigan o no en WRITES_IDX_MAP (put_writes query:462-466): las escrituras a canales de control (ERROR / INTERRUPT / RESUME) pueden sobrescribir; las escrituras de negocio normales se ignoran si colisionan —la semántica es la misma que en InMemory con if inner_key[1] >= 0 ... continue (InMemorySaver.put_writes:499-509).
InMemory usa max(checkpoints.keys()) como latest: sin un ORDER BY explícito, se apoya en que checkpoint_id crezca monótonamente como cadena (latest:282-283); por eso checkpoint_id debe ser un UUID6/UUID7 comparable lexicográficamente, no un UUID aleatorio puro.

Archivos clave

InMemorySaver 类定义:33-83 — declaración de la clase + los tres campos defaultdict: storage / writes / blobs.
InMemorySaver.put:427-471 — descompone channel_values en blobs, guarda el cuerpo en storage y devuelve el nuevo config.
InMemorySaver.put_writes:473-509 — salta las escrituras de control con idx negativo ya existentes; las escrituras normales se indexan por (task_id, idx).
InMemorySaver.get_tuple:236-316 — busca exactamente por checkpoint_id o toma el más reciente con max(keys()), y reconstruye el CheckpointTuple.
MemorySaver alias:631 — alias histórico, equivalente a InMemorySaver.
SqliteSaver 类:45-95 — recibe sqlite3.Connection y trae un threading.Lock para serializar escrituras.
SqliteSaver.setup:129-166 — crea tablas + activa WAL.
SqliteSaver.get_tuple:191-263 — dos SELECT preparados: búsqueda exacta por checkpoint_id o ORDER BY checkpoint_id DESC LIMIT 1 para el último, con JOIN a writes.
SqliteSaver.put:387-443 — INSERT OR REPLACE INTO checkpoints; la metadata se serializa como bytes en JSON.
AsyncSqliteSaver:38 — versión asíncrona; aput / aget_tuple / alist se reescriben con aiosqlite.

Flujo de datos

El cuerpo de InMemorySaver.put hace dos cosas a la vez: «descomponer channel_values en blobs» y «guardar el cuerpo en storage». Se ve con claridad cómo un saver descompone un Checkpoint TypedDict en una estructura de tres niveles:

python

c = checkpoint.copy()
thread_id = config["configurable"]["thread_id"]
checkpoint_ns = config["configurable"]["checkpoint_ns"]
values: dict[str, Any] = c.pop("channel_values")  # type: ignore[misc]
for k, v in new_versions.items():
    self.blobs[(thread_id, checkpoint_ns, k, v)] = (
        self.serde.dumps_typed(values[k]) if k in values else ("empty", b"")
    )
self.storage[thread_id][checkpoint_ns].update(
    {
        checkpoint["id"]: (
            self.serde.dumps_typed(c),
            self.serde.dumps_typed(get_checkpoint_metadata(config, metadata)),
            config["configurable"].get("checkpoint_id"),  # parent
        )
    }
)

(put body:448-464) Nota: blobs se indexa por (thread_id, checkpoint_ns, channel, version), mientras que storage se indexa por (thread_id, checkpoint_ns, checkpoint_id) —dos índices sin solapamiento; el valor del canal queda desacoplado del cuerpo del checkpoint, así los canales cuya versión no cambió no se reserializan. Al leer de vuelta, get_tuple llama a _load_blobs para deserializar cada blob correspondiente a un par (channel, version) de channel_versions y recomponer channel_values (_load_blobs:259-265), y lo envuelve en un CheckpointTuple.

SqliteSaver sigue la misma lógica, sólo que los dicts se vuelven SQL y el cubo de blobs pasa a ser la columna checkpoint BLOB de la tabla principal + la columna value BLOB de la tabla de writes —no hay tabla de blobs independiente; todos los valores de canal se serializan directamente en el campo checkpoint blob:

python

cur.execute(
    "INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
    (
        str(config["configurable"]["thread_id"]),
        checkpoint_ns,
        checkpoint["id"],
        config["configurable"].get("checkpoint_id"),
        type_,
        serialized_checkpoint,
        serialized_metadata,
    ),
)

(SqliteSaver.put INSERT:425-436)

Límites y fallos

Si el proceso que usa InMemorySaver cae, se pierde todo: la docstring lo deja claro: «usar InMemorySaver sólo para depuración o pruebas» (warning:40-44); para persistencia, cambiar a Sqlite o Postgres.
InMemorySaver obtiene latest por orden lexicográfico de checkpoint_id: max(checkpoints.keys()) (max:282-283) asume IDs monótonos; si un saver personalizado usa IDs no monótonos, get_tuple sin checkpoint_id devolverá la fila equivocada.
SqliteSaver usa una única conexión y un único lock: el gestor de contexto cursor() toma self.lock (cursor:181-189) y serializa las escrituras concurrentes; una transacción larga bloquea a otras lecturas. Para concurrencia real hace falta el modo pipeline de Postgres.
SqliteSaver usa ? como placeholder con orden de columnas hardcodeado: el orden de columnas debe coincidir con el INSERT INTO checkpoints (...) (INSERT columns:426); añadir una columna en una migración obliga a tocar ambos sitios. La columna task_path, por ejemplo, se añadió después —véase el final de MIGRATIONS en Postgres (task_path migration:90).
AsyncSqliteSaver no es sólo un to_thread: AsyncSqliteSaver:38 es una clase independiente; aput / aget_tuple / alist se reescriben con aiosqlite (aput:509), no usan el asyncio.to_thread por defecto de la clase base —así la ruta async es realmente concurrente y no la serializa el lock global.
Con factory=PersistentDict hay que llamar close() al salir: PersistentDict.sync() vuelca el pickle a archivo (sync:657); sin llamar a __exit__ no se persiste automáticamente. El patrón idiomático es with InMemorySaver(factory=PersistentDict, ...).

Resumen

InMemorySaver es la implementación de referencia de «tres cubos dict»; SqliteSaver despliega los mismos tres cubos en dos tablas SQLite + WAL. Juntos cubren depuración y persistencia en una sola máquina. Las convenciones de primary key, separación de blobs y el idx negativo para canales de control se ven con detalle en estas dos implementaciones. Para producción a escala, véase PostgresSaver; para la forma de la interfaz, BaseCheckpointSaver.

Véase la documentación oficial: documentación de LangGraph · README

InMemorySaver + SqliteSaver: checkpoints en proceso y a nivel de archivo ​

Responsabilidades ​

Motivación de diseño ​

Archivos clave ​

Flujo de datos ​

Límites y fallos ​

Resumen ​