mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-10-14 12:15:25 +02:00
2b93e8b79f
Fixes langchain-ai/langgraph#8382 `DeltaChannel` rebuilds its value by replaying ancestor writes through the reducer. Every saver ordered a checkpoint's writes by `(task_id, idx)`, but live execution applies a super-step's writes in task-path order (`apply_writes` sorts tasks by `task_path_str(task.path[:3])`). `task_id` is a hash of the path, so when two or more tasks write one `DeltaChannel` in the same super-step, replay applies them in an arbitrary permutation. Reducers must be batching-invariant, not order-invariant, so the permutation changes the value: `get_state` disagrees with what `invoke` returned, and continuing the thread saves the reordered value as the base for every later write. ## Fix Replay orders by `(task_path, task_id, idx)`, the order `SELECT_PENDING_SENDS_SQL` already uses for sends. `apply_writes` and every saver now sort with one `writes_sort_key` from `langgraph.checkpoint.base`, in Python, so live and replay can't drift apart and don't depend on a database collation. `apply_writes` sorts by the full task path, the string `put_writes` stores, instead of `path[:3]`. The paths that reach `apply_writes` no longer carry task ids, so the live order doesn't change. `BaseCheckpointSaver`'s default `get_delta_channel_history` can't sort by path, since `PendingWrite` has no task path, so it replays each checkpoint's writes in `get_tuple` order. `get_tuple` now has to return `pending_writes` in `writes_sort_key` order, and memory, SQLite and Postgres do, so the default walk is right for any saver that follows that. `AsyncPostgresSaver` also gets the sync `get_delta_channel_history` that `AsyncSqliteSaver` already had; sync reads used to fall back to the default walk. Memory and Postgres already stored `task_path`. SQLite accepted it on `put_writes` and dropped it, so `writes` gains the column and `setup()` adds it to existing databases. Writes without a task path (graph input, `update_state` updates, rows written before the column existed) sort first by `task_id`, which is also where live execution applies input. ## Compatibility - Memory and Postgres already had `task_path` on disk, so existing threads replay in the corrected order after upgrade. - On SQLite, rows from before the upgrade read back as `''` and keep their old order. Only new writes get the corrected order, since there is no stored path to recover. - SQLite has no `ADD COLUMN IF NOT EXISTS`, so `setup()` runs the `ALTER` and treats `duplicate column name` as success, which also covers two processes migrating one file. That error comes while SQLite prepares the statement, before it asks for the write lock, so `setup()` on an up-to-date database doesn't wait on other writers. A read-only database from before the column is read with empty paths instead of failing setup. - Downgrade is safe: older code's `INSERT` omits the column and the column has a default. - `get_tuple` and `list` return `pending_writes` in `writes_sort_key` order instead of task id order (SQLite, Postgres) or put order (memory). Nothing in langgraph depends on the old order. - Postgres reads `task_path` and `idx` with each pending write, so a subclass that pairs `SELECT_SQL` with its own `_load_writes`, or the other way round, has to match the new row shape. - A third-party saver has to replay in `writes_sort_key` order: in its own `get_delta_channel_history` if it has one, otherwise by returning `pending_writes` from `get_tuple` in that order. Otherwise it fails the new conformance tests; the `get_tuple` one is a base test, so it runs for every saver. - langgraph, checkpoint-sqlite and checkpoint-postgres import `writes_sort_key` from langgraph-checkpoint, so this bumps langgraph-checkpoint to 4.3.0 and requires it there. 4.3.0 has to be released first. ## Limits - Several updates in one `bulk_update_state` super-step share a task id, so the saver keeps only the first one's writes and the `DeltaChannel` loses the rest (plain channels are fine). That's on main too; #9128 fixes it and stores a task path per update, so with this PR those updates also replay in the order given. - Exit durability stores its writes differently; resume ordering there is fixed in #9114, stacked on this PR. - langgraphjs applies `DeltaChannel` writes in task id order, live and on replay (langchain-ai/langgraphjs#2544), so its reads match too. Moving JS to task path order is a follow-up. ## Tests Each fails on main and passes here: - `test_delta_channel_parallel_order.py`, across memory, SQLite and Postgres: parallel writers and a `Send` fan-out, comparing `get_state` with the live result, including sync `get_state` on the async savers. - Three conformance tests whose task ids sort opposite to their task paths, so they cannot pass by chance: two on `get_delta_channel_history`, one on `get_tuple`'s `pending_writes` order. - SQLite migration tests from a pre-column database, sync and async, including a read-only file. One more that passes on main too: SQLite `setup()` while another connection holds the write lock, with no busy timeout, sync and async. It keeps the `ALTER` from waiting on that lock. Thanks to @ErenAta16 for the report, the reproduction, and for tracing `task_path` through all three backends, and to @ragnarok268 for the conformance-test design with task ids ordered opposite to their paths. --------- Co-authored-by: ErenAta16 <149434812+ErenAta16@users.noreply.github.com> Co-authored-by: ragnarok268 <58264829+ragnarok268@users.noreply.github.com>
968 lines
36 KiB
Python
968 lines
36 KiB
Python
import asyncio
|
|
import threading
|
|
import warnings
|
|
from collections.abc import AsyncIterator, Iterator, Sequence
|
|
from contextlib import asynccontextmanager, contextmanager
|
|
from typing import Any
|
|
|
|
from langchain_core.runnables import RunnableConfig
|
|
from langgraph.checkpoint.base import (
|
|
WRITES_IDX_MAP,
|
|
ChannelVersions,
|
|
Checkpoint,
|
|
CheckpointMetadata,
|
|
CheckpointTuple,
|
|
get_serializable_checkpoint_metadata,
|
|
)
|
|
from langgraph.checkpoint.serde.base import SerializerProtocol
|
|
from langgraph.checkpoint.serde.types import TASKS
|
|
from psycopg import (
|
|
AsyncConnection,
|
|
AsyncCursor,
|
|
AsyncPipeline,
|
|
Capabilities,
|
|
Connection,
|
|
Cursor,
|
|
Pipeline,
|
|
)
|
|
from psycopg.rows import DictRow, dict_row
|
|
from psycopg.types.json import Jsonb
|
|
from psycopg_pool import AsyncConnectionPool, ConnectionPool
|
|
|
|
from langgraph.checkpoint.postgres import _ainternal, _internal
|
|
from langgraph.checkpoint.postgres.base import BasePostgresSaver
|
|
|
|
"""
|
|
To add a new migration, add a new string to the MIGRATIONS list.
|
|
The position of the migration in the list is the version number.
|
|
"""
|
|
MIGRATIONS = [
|
|
"""CREATE TABLE IF NOT EXISTS checkpoint_migrations (
|
|
v INTEGER PRIMARY KEY
|
|
);""",
|
|
"""CREATE TABLE IF NOT EXISTS checkpoints (
|
|
thread_id TEXT NOT NULL,
|
|
checkpoint_ns TEXT NOT NULL DEFAULT '',
|
|
type TEXT,
|
|
checkpoint JSONB NOT NULL,
|
|
metadata JSONB NOT NULL DEFAULT '{}',
|
|
PRIMARY KEY (thread_id, checkpoint_ns)
|
|
);""",
|
|
"""CREATE TABLE IF NOT EXISTS checkpoint_blobs (
|
|
thread_id TEXT NOT NULL,
|
|
checkpoint_ns TEXT NOT NULL DEFAULT '',
|
|
channel TEXT NOT NULL,
|
|
type TEXT NOT NULL,
|
|
blob BYTEA,
|
|
PRIMARY KEY (thread_id, checkpoint_ns, channel)
|
|
);""",
|
|
"""CREATE TABLE IF NOT EXISTS checkpoint_writes (
|
|
thread_id TEXT NOT NULL,
|
|
checkpoint_ns TEXT NOT NULL DEFAULT '',
|
|
checkpoint_id TEXT NOT NULL,
|
|
task_id TEXT NOT NULL,
|
|
idx INTEGER NOT NULL,
|
|
channel TEXT NOT NULL,
|
|
type TEXT,
|
|
blob BYTEA NOT NULL,
|
|
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
|
|
);""",
|
|
"""
|
|
CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoints_thread_id_idx ON checkpoints(thread_id);
|
|
""",
|
|
"""
|
|
CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoint_blobs_thread_id_idx ON checkpoint_blobs(thread_id);
|
|
""",
|
|
"""
|
|
CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoint_writes_thread_id_idx ON checkpoint_writes(thread_id);
|
|
""",
|
|
"""
|
|
ALTER TABLE checkpoint_writes ADD COLUMN IF NOT EXISTS task_path TEXT NOT NULL DEFAULT '';
|
|
""",
|
|
]
|
|
|
|
SELECT_SQL = f"""
|
|
select
|
|
thread_id,
|
|
checkpoint,
|
|
checkpoint_ns,
|
|
metadata,
|
|
(
|
|
select array_agg(array[bl.channel::bytea, bl.type::bytea, bl.blob])
|
|
from jsonb_each_text(checkpoint -> 'channel_versions')
|
|
inner join checkpoint_blobs bl
|
|
on bl.thread_id = checkpoints.thread_id
|
|
and bl.checkpoint_ns = checkpoints.checkpoint_ns
|
|
and bl.channel = jsonb_each_text.key
|
|
) as channel_values,
|
|
(
|
|
select
|
|
array_agg(array[cw.task_id::text::bytea, cw.channel::bytea, cw.type::bytea, cw.blob, convert_to(cw.task_path, 'UTF8'), cw.idx::text::bytea])
|
|
from checkpoint_writes cw
|
|
where cw.thread_id = checkpoints.thread_id
|
|
and cw.checkpoint_ns = checkpoints.checkpoint_ns
|
|
and cw.checkpoint_id = (checkpoint->>'id')
|
|
) as pending_writes,
|
|
(
|
|
select array_agg(array[cw.type::bytea, cw.blob] order by cw.task_path, cw.task_id, cw.idx)
|
|
from checkpoint_writes cw
|
|
where cw.thread_id = checkpoints.thread_id
|
|
and cw.checkpoint_ns = checkpoints.checkpoint_ns
|
|
and cw.channel = '{TASKS}'
|
|
) as pending_sends
|
|
from checkpoints """
|
|
|
|
UPSERT_CHECKPOINT_BLOBS_SQL = """
|
|
INSERT INTO checkpoint_blobs (thread_id, checkpoint_ns, channel, type, blob)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
ON CONFLICT (thread_id, checkpoint_ns, channel) DO UPDATE SET
|
|
type = EXCLUDED.type,
|
|
blob = EXCLUDED.blob;
|
|
"""
|
|
|
|
UPSERT_CHECKPOINTS_SQL = """
|
|
INSERT INTO checkpoints (thread_id, checkpoint_ns, checkpoint, metadata)
|
|
VALUES (%s, %s, %s, %s)
|
|
ON CONFLICT (thread_id, checkpoint_ns)
|
|
DO UPDATE SET
|
|
checkpoint = EXCLUDED.checkpoint,
|
|
metadata = EXCLUDED.metadata;
|
|
"""
|
|
|
|
UPSERT_CHECKPOINT_WRITES_SQL = """
|
|
INSERT INTO checkpoint_writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, blob)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) DO UPDATE SET
|
|
channel = EXCLUDED.channel,
|
|
type = EXCLUDED.type,
|
|
blob = EXCLUDED.blob;
|
|
"""
|
|
|
|
INSERT_CHECKPOINT_WRITES_SQL = """
|
|
INSERT INTO checkpoint_writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, blob)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) DO NOTHING
|
|
"""
|
|
|
|
|
|
def _dump_blobs(
|
|
serde: SerializerProtocol,
|
|
thread_id: str,
|
|
checkpoint_ns: str,
|
|
values: dict[str, Any],
|
|
versions: ChannelVersions,
|
|
) -> list[tuple[str, str, str, str, bytes | None]]:
|
|
if not versions:
|
|
return []
|
|
|
|
return [
|
|
(
|
|
thread_id,
|
|
checkpoint_ns,
|
|
k,
|
|
*(serde.dumps_typed(values[k]) if k in values else ("empty", None)),
|
|
)
|
|
for k in versions
|
|
]
|
|
|
|
|
|
class ShallowPostgresSaver(BasePostgresSaver):
|
|
"""A checkpoint saver that uses Postgres to store checkpoints.
|
|
|
|
This checkpointer ONLY stores the most recent checkpoint and does NOT retain any history.
|
|
It is meant to be a light-weight drop-in replacement for the PostgresSaver that
|
|
supports most of the LangGraph persistence functionality with the exception of time travel.
|
|
"""
|
|
|
|
SELECT_SQL = SELECT_SQL
|
|
MIGRATIONS = MIGRATIONS
|
|
UPSERT_CHECKPOINT_BLOBS_SQL = UPSERT_CHECKPOINT_BLOBS_SQL
|
|
UPSERT_CHECKPOINTS_SQL = UPSERT_CHECKPOINTS_SQL
|
|
UPSERT_CHECKPOINT_WRITES_SQL = UPSERT_CHECKPOINT_WRITES_SQL
|
|
INSERT_CHECKPOINT_WRITES_SQL = INSERT_CHECKPOINT_WRITES_SQL
|
|
|
|
lock: threading.Lock
|
|
|
|
def __init__(
|
|
self,
|
|
conn: _internal.Conn,
|
|
pipe: Pipeline | None = None,
|
|
serde: SerializerProtocol | None = None,
|
|
) -> None:
|
|
warnings.warn(
|
|
"ShallowPostgresSaver is deprecated as of version 2.0.20 and will be removed in 3.0.0. "
|
|
"Use PostgresSaver instead, and invoke the graph with `graph.invoke(..., durability='exit')`.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
super().__init__(serde=serde)
|
|
if isinstance(conn, ConnectionPool) and pipe is not None:
|
|
raise ValueError(
|
|
"Pipeline should be used only with a single Connection, not ConnectionPool."
|
|
)
|
|
|
|
self.conn = conn
|
|
self.pipe = pipe
|
|
self.lock = threading.Lock()
|
|
self.supports_pipeline = Capabilities().has_pipeline()
|
|
|
|
@classmethod
|
|
@contextmanager
|
|
def from_conn_string(
|
|
cls, conn_string: str, *, pipeline: bool = False
|
|
) -> Iterator["ShallowPostgresSaver"]:
|
|
"""Create a new ShallowPostgresSaver instance from a connection string.
|
|
|
|
Args:
|
|
conn_string: The Postgres connection info string.
|
|
pipeline: whether to use Pipeline
|
|
|
|
Returns:
|
|
ShallowPostgresSaver: A new ShallowPostgresSaver instance.
|
|
"""
|
|
with Connection.connect(
|
|
conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
|
|
) as conn:
|
|
if pipeline:
|
|
with conn.pipeline() as pipe:
|
|
yield cls(conn, pipe)
|
|
else:
|
|
yield cls(conn)
|
|
|
|
def setup(self) -> None:
|
|
"""Set up the checkpoint database asynchronously.
|
|
|
|
This method creates the necessary tables in the Postgres database if they don't
|
|
already exist and runs database migrations. It MUST be called directly by the user
|
|
the first time checkpointer is used.
|
|
"""
|
|
with self._cursor() as cur:
|
|
cur.execute(self.MIGRATIONS[0])
|
|
results = cur.execute(
|
|
"SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
|
|
)
|
|
row = results.fetchone()
|
|
if row is None:
|
|
version = -1
|
|
else:
|
|
version = row["v"]
|
|
for v, migration in zip(
|
|
range(version + 1, len(self.MIGRATIONS)),
|
|
self.MIGRATIONS[version + 1 :],
|
|
strict=False,
|
|
):
|
|
cur.execute(migration)
|
|
cur.execute("INSERT INTO checkpoint_migrations (v) VALUES (%s)", (v,))
|
|
if self.pipe:
|
|
self.pipe.sync()
|
|
|
|
def list(
|
|
self,
|
|
config: RunnableConfig | None,
|
|
*,
|
|
filter: dict[str, Any] | None = None,
|
|
before: RunnableConfig | None = None,
|
|
limit: int | None = None,
|
|
) -> Iterator[CheckpointTuple]:
|
|
"""List checkpoints from the database.
|
|
|
|
This method retrieves a list of checkpoint tuples from the Postgres database based
|
|
on the provided config. For ShallowPostgresSaver, this method returns a list with
|
|
ONLY the most recent checkpoint.
|
|
"""
|
|
where, args = self._search_where(config, filter, before)
|
|
query = self.SELECT_SQL + where
|
|
params = list(args)
|
|
if limit is not None:
|
|
query += " LIMIT %s"
|
|
params.append(int(limit))
|
|
with self._cursor() as cur:
|
|
cur.execute(query, params, binary=True)
|
|
for value in cur:
|
|
checkpoint: Checkpoint = {
|
|
**value["checkpoint"],
|
|
"channel_values": self._load_blobs(value["channel_values"]),
|
|
"pending_sends": [
|
|
self.serde.loads_typed((t.decode(), v))
|
|
for t, v in value["pending_sends"]
|
|
]
|
|
if value["pending_sends"]
|
|
else [],
|
|
}
|
|
yield CheckpointTuple(
|
|
config={
|
|
"configurable": {
|
|
"thread_id": value["thread_id"],
|
|
"checkpoint_ns": value["checkpoint_ns"],
|
|
"checkpoint_id": checkpoint["id"],
|
|
}
|
|
},
|
|
checkpoint=checkpoint,
|
|
metadata=value["metadata"],
|
|
pending_writes=self._load_writes(value["pending_writes"]),
|
|
)
|
|
|
|
def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
|
|
"""Get a checkpoint tuple from the database.
|
|
|
|
This method retrieves a checkpoint tuple from the Postgres database based on the
|
|
provided config (matching the thread ID in the config).
|
|
|
|
Args:
|
|
config: The config to use for retrieving the checkpoint.
|
|
|
|
Returns:
|
|
The retrieved checkpoint tuple, or None if no matching checkpoint was found.
|
|
|
|
Examples:
|
|
|
|
Basic:
|
|
>>> config = {"configurable": {"thread_id": "1"}}
|
|
>>> checkpoint_tuple = memory.get_tuple(config)
|
|
>>> print(checkpoint_tuple)
|
|
CheckpointTuple(...)
|
|
|
|
With timestamp:
|
|
|
|
>>> config = {
|
|
... "configurable": {
|
|
... "thread_id": "1",
|
|
... "checkpoint_ns": "",
|
|
... "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
|
|
... }
|
|
... }
|
|
>>> checkpoint_tuple = memory.get_tuple(config)
|
|
>>> print(checkpoint_tuple)
|
|
CheckpointTuple(...)
|
|
"""
|
|
thread_id = config["configurable"]["thread_id"]
|
|
checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
|
|
args = (thread_id, checkpoint_ns)
|
|
where = "WHERE thread_id = %s AND checkpoint_ns = %s"
|
|
|
|
with self._cursor() as cur:
|
|
cur.execute(
|
|
self.SELECT_SQL + where,
|
|
args,
|
|
binary=True,
|
|
)
|
|
|
|
for value in cur:
|
|
checkpoint: Checkpoint = {
|
|
**value["checkpoint"],
|
|
"channel_values": self._load_blobs(value["channel_values"]),
|
|
"pending_sends": [
|
|
self.serde.loads_typed((t.decode(), v))
|
|
for t, v in value["pending_sends"]
|
|
]
|
|
if value["pending_sends"]
|
|
else [],
|
|
}
|
|
return CheckpointTuple(
|
|
config={
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": checkpoint["id"],
|
|
}
|
|
},
|
|
checkpoint=checkpoint,
|
|
metadata=value["metadata"],
|
|
pending_writes=self._load_writes(value["pending_writes"]),
|
|
)
|
|
|
|
def put(
|
|
self,
|
|
config: RunnableConfig,
|
|
checkpoint: Checkpoint,
|
|
metadata: CheckpointMetadata,
|
|
new_versions: ChannelVersions,
|
|
) -> RunnableConfig:
|
|
"""Save a checkpoint to the database.
|
|
|
|
This method saves a checkpoint to the Postgres database. The checkpoint is associated
|
|
with the provided config. For ShallowPostgresSaver, this method saves ONLY the most recent
|
|
checkpoint and overwrites a previous checkpoint, if it exists.
|
|
|
|
Args:
|
|
config: The config to associate with the checkpoint.
|
|
checkpoint: The checkpoint to save.
|
|
metadata: Additional metadata to save with the checkpoint.
|
|
new_versions: New channel versions as of this write.
|
|
|
|
Returns:
|
|
RunnableConfig: Updated configuration after storing the checkpoint.
|
|
|
|
Examples:
|
|
|
|
>>> from langgraph.checkpoint.postgres import ShallowPostgresSaver
|
|
>>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
|
|
>>> with ShallowPostgresSaver.from_conn_string(DB_URI) as memory:
|
|
>>> config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
|
|
>>> checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
|
|
>>> saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
|
|
>>> print(saved_config)
|
|
{'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}
|
|
"""
|
|
configurable = config["configurable"].copy()
|
|
thread_id = configurable.pop("thread_id")
|
|
checkpoint_ns = configurable.pop("checkpoint_ns")
|
|
|
|
copy = checkpoint.copy()
|
|
next_config = {
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": checkpoint["id"],
|
|
}
|
|
}
|
|
|
|
with self._cursor(pipeline=True) as cur:
|
|
cur.execute(
|
|
"""DELETE FROM checkpoint_writes
|
|
WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id NOT IN (%s, %s)""",
|
|
(
|
|
thread_id,
|
|
checkpoint_ns,
|
|
checkpoint["id"],
|
|
configurable.get("checkpoint_id", ""),
|
|
),
|
|
)
|
|
cur.executemany(
|
|
self.UPSERT_CHECKPOINT_BLOBS_SQL,
|
|
_dump_blobs(
|
|
self.serde,
|
|
thread_id,
|
|
checkpoint_ns,
|
|
copy.pop("channel_values"), # type: ignore[misc]
|
|
new_versions,
|
|
),
|
|
)
|
|
cur.execute(
|
|
self.UPSERT_CHECKPOINTS_SQL,
|
|
(
|
|
thread_id,
|
|
checkpoint_ns,
|
|
Jsonb(copy),
|
|
Jsonb(get_serializable_checkpoint_metadata(config, metadata)),
|
|
),
|
|
)
|
|
return next_config
|
|
|
|
def put_writes(
|
|
self,
|
|
config: RunnableConfig,
|
|
writes: Sequence[tuple[str, Any]],
|
|
task_id: str,
|
|
task_path: str = "",
|
|
) -> None:
|
|
"""Store intermediate writes linked to a checkpoint.
|
|
|
|
This method saves intermediate writes associated with a checkpoint to the Postgres database.
|
|
|
|
Args:
|
|
config: Configuration of the related checkpoint.
|
|
writes: List of writes to store.
|
|
task_id: Identifier for the task creating the writes.
|
|
"""
|
|
query = (
|
|
self.UPSERT_CHECKPOINT_WRITES_SQL
|
|
if all(w[0] in WRITES_IDX_MAP for w in writes)
|
|
else self.INSERT_CHECKPOINT_WRITES_SQL
|
|
)
|
|
with self._cursor(pipeline=True) as cur:
|
|
cur.executemany(
|
|
query,
|
|
self._dump_writes(
|
|
config["configurable"]["thread_id"],
|
|
config["configurable"]["checkpoint_ns"],
|
|
config["configurable"]["checkpoint_id"],
|
|
task_id,
|
|
task_path,
|
|
writes,
|
|
),
|
|
)
|
|
|
|
@contextmanager
|
|
def _cursor(self, *, pipeline: bool = False) -> Iterator[Cursor[DictRow]]:
|
|
"""Create a database cursor as a context manager.
|
|
|
|
Args:
|
|
pipeline: whether to use pipeline for the DB operations inside the context manager.
|
|
Will be applied regardless of whether the ShallowPostgresSaver instance was initialized with a pipeline.
|
|
If pipeline mode is not supported, will fall back to using transaction context manager.
|
|
"""
|
|
with _internal.get_connection(self.conn) as conn:
|
|
if self.pipe:
|
|
# a connection in pipeline mode can be used concurrently
|
|
# in multiple threads/coroutines, but only one cursor can be
|
|
# used at a time
|
|
try:
|
|
with conn.cursor(binary=True, row_factory=dict_row) as cur:
|
|
yield cur
|
|
finally:
|
|
if pipeline:
|
|
self.pipe.sync()
|
|
elif pipeline:
|
|
# a connection not in pipeline mode can only be used by one
|
|
# thread/coroutine at a time, so we acquire a lock
|
|
if self.supports_pipeline:
|
|
with (
|
|
self.lock,
|
|
conn.pipeline(),
|
|
conn.cursor(binary=True, row_factory=dict_row) as cur,
|
|
):
|
|
yield cur
|
|
else:
|
|
# Use connection's transaction context manager when pipeline mode not supported
|
|
with (
|
|
self.lock,
|
|
conn.transaction(),
|
|
conn.cursor(binary=True, row_factory=dict_row) as cur,
|
|
):
|
|
yield cur
|
|
else:
|
|
with self.lock, conn.cursor(binary=True, row_factory=dict_row) as cur:
|
|
yield cur
|
|
|
|
|
|
class AsyncShallowPostgresSaver(BasePostgresSaver):
|
|
"""A checkpoint saver that uses Postgres to store checkpoints asynchronously.
|
|
|
|
This checkpointer ONLY stores the most recent checkpoint and does NOT retain any history.
|
|
It is meant to be a light-weight drop-in replacement for the AsyncPostgresSaver that
|
|
supports most of the LangGraph persistence functionality with the exception of time travel.
|
|
"""
|
|
|
|
SELECT_SQL = SELECT_SQL
|
|
MIGRATIONS = MIGRATIONS
|
|
UPSERT_CHECKPOINT_BLOBS_SQL = UPSERT_CHECKPOINT_BLOBS_SQL
|
|
UPSERT_CHECKPOINTS_SQL = UPSERT_CHECKPOINTS_SQL
|
|
UPSERT_CHECKPOINT_WRITES_SQL = UPSERT_CHECKPOINT_WRITES_SQL
|
|
INSERT_CHECKPOINT_WRITES_SQL = INSERT_CHECKPOINT_WRITES_SQL
|
|
lock: asyncio.Lock
|
|
|
|
def __init__(
|
|
self,
|
|
conn: _ainternal.Conn,
|
|
pipe: AsyncPipeline | None = None,
|
|
serde: SerializerProtocol | None = None,
|
|
) -> None:
|
|
warnings.warn(
|
|
"AsyncShallowPostgresSaver is deprecated as of version 2.0.20 and will be removed in 3.0.0. "
|
|
"Use AsyncPostgresSaver instead, and invoke the graph with `await graph.ainvoke(..., durability='exit')`.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
super().__init__(serde=serde)
|
|
if isinstance(conn, AsyncConnectionPool) and pipe is not None:
|
|
raise ValueError(
|
|
"Pipeline should be used only with a single AsyncConnection, not AsyncConnectionPool."
|
|
)
|
|
|
|
self.conn = conn
|
|
self.pipe = pipe
|
|
self.lock = asyncio.Lock()
|
|
self.loop = asyncio.get_running_loop()
|
|
self.supports_pipeline = Capabilities().has_pipeline()
|
|
|
|
@classmethod
|
|
@asynccontextmanager
|
|
async def from_conn_string(
|
|
cls,
|
|
conn_string: str,
|
|
*,
|
|
pipeline: bool = False,
|
|
serde: SerializerProtocol | None = None,
|
|
) -> AsyncIterator["AsyncShallowPostgresSaver"]:
|
|
"""Create a new AsyncShallowPostgresSaver instance from a connection string.
|
|
|
|
Args:
|
|
conn_string: The Postgres connection info string.
|
|
pipeline: whether to use AsyncPipeline
|
|
|
|
Returns:
|
|
AsyncShallowPostgresSaver: A new AsyncShallowPostgresSaver instance.
|
|
"""
|
|
async with await AsyncConnection.connect(
|
|
conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
|
|
) as conn:
|
|
if pipeline:
|
|
async with conn.pipeline() as pipe:
|
|
yield cls(conn=conn, pipe=pipe, serde=serde)
|
|
else:
|
|
yield cls(conn=conn, serde=serde)
|
|
|
|
async def setup(self) -> None:
|
|
"""Set up the checkpoint database asynchronously.
|
|
|
|
This method creates the necessary tables in the Postgres database if they don't
|
|
already exist and runs database migrations. It MUST be called directly by the user
|
|
the first time checkpointer is used.
|
|
"""
|
|
async with self._cursor() as cur:
|
|
await cur.execute(self.MIGRATIONS[0])
|
|
results = await cur.execute(
|
|
"SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
|
|
)
|
|
row = await results.fetchone()
|
|
if row is None:
|
|
version = -1
|
|
else:
|
|
version = row["v"]
|
|
for v, migration in zip(
|
|
range(version + 1, len(self.MIGRATIONS)),
|
|
self.MIGRATIONS[version + 1 :],
|
|
strict=False,
|
|
):
|
|
await cur.execute(migration)
|
|
await cur.execute(
|
|
"INSERT INTO checkpoint_migrations (v) VALUES (%s)", (v,)
|
|
)
|
|
if self.pipe:
|
|
await self.pipe.sync()
|
|
|
|
async def alist(
|
|
self,
|
|
config: RunnableConfig | None,
|
|
*,
|
|
filter: dict[str, Any] | None = None,
|
|
before: RunnableConfig | None = None,
|
|
limit: int | None = None,
|
|
) -> AsyncIterator[CheckpointTuple]:
|
|
"""List checkpoints from the database asynchronously.
|
|
|
|
This method retrieves a list of checkpoint tuples from the Postgres database based
|
|
on the provided config. For ShallowPostgresSaver, this method returns a list with
|
|
ONLY the most recent checkpoint.
|
|
"""
|
|
where, args = self._search_where(config, filter, before)
|
|
query = self.SELECT_SQL + where
|
|
params = list(args)
|
|
if limit is not None:
|
|
query += " LIMIT %s"
|
|
params.append(int(limit))
|
|
async with self._cursor() as cur:
|
|
await cur.execute(query, params, binary=True)
|
|
async for value in cur:
|
|
checkpoint: Checkpoint = {
|
|
**value["checkpoint"],
|
|
"channel_values": self._load_blobs(value["channel_values"]),
|
|
"pending_sends": [
|
|
self.serde.loads_typed((t.decode(), v))
|
|
for t, v in value["pending_sends"]
|
|
]
|
|
if value["pending_sends"]
|
|
else [],
|
|
}
|
|
yield CheckpointTuple(
|
|
config={
|
|
"configurable": {
|
|
"thread_id": value["thread_id"],
|
|
"checkpoint_ns": value["checkpoint_ns"],
|
|
"checkpoint_id": checkpoint["id"],
|
|
}
|
|
},
|
|
checkpoint=checkpoint,
|
|
metadata=value["metadata"],
|
|
pending_writes=await asyncio.to_thread(
|
|
self._load_writes, value["pending_writes"]
|
|
),
|
|
)
|
|
|
|
async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
|
|
"""Get a checkpoint tuple from the database asynchronously.
|
|
|
|
This method retrieves a checkpoint tuple from the Postgres database based on the
|
|
provided config (matching the thread ID in the config).
|
|
|
|
Args:
|
|
config: The config to use for retrieving the checkpoint.
|
|
|
|
Returns:
|
|
The retrieved checkpoint tuple, or None if no matching checkpoint was found.
|
|
"""
|
|
thread_id = config["configurable"]["thread_id"]
|
|
checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
|
|
args = (thread_id, checkpoint_ns)
|
|
where = "WHERE thread_id = %s AND checkpoint_ns = %s"
|
|
|
|
async with self._cursor() as cur:
|
|
await cur.execute(
|
|
self.SELECT_SQL + where,
|
|
args,
|
|
binary=True,
|
|
)
|
|
|
|
async for value in cur:
|
|
checkpoint: Checkpoint = {
|
|
**value["checkpoint"],
|
|
"channel_values": self._load_blobs(value["channel_values"]),
|
|
"pending_sends": [
|
|
self.serde.loads_typed((t.decode(), v))
|
|
for t, v in value["pending_sends"]
|
|
]
|
|
if value["pending_sends"]
|
|
else [],
|
|
}
|
|
return CheckpointTuple(
|
|
config={
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": checkpoint["id"],
|
|
}
|
|
},
|
|
checkpoint=checkpoint,
|
|
metadata=value["metadata"],
|
|
pending_writes=await asyncio.to_thread(
|
|
self._load_writes, value["pending_writes"]
|
|
),
|
|
)
|
|
|
|
async def aput(
|
|
self,
|
|
config: RunnableConfig,
|
|
checkpoint: Checkpoint,
|
|
metadata: CheckpointMetadata,
|
|
new_versions: ChannelVersions,
|
|
) -> RunnableConfig:
|
|
"""Save a checkpoint to the database asynchronously.
|
|
|
|
This method saves a checkpoint to the Postgres database. The checkpoint is associated
|
|
with the provided config. For AsyncShallowPostgresSaver, this method saves ONLY the most recent
|
|
checkpoint and overwrites a previous checkpoint, if it exists.
|
|
|
|
Args:
|
|
config: The config to associate with the checkpoint.
|
|
checkpoint: The checkpoint to save.
|
|
metadata: Additional metadata to save with the checkpoint.
|
|
new_versions: New channel versions as of this write.
|
|
|
|
Returns:
|
|
RunnableConfig: Updated configuration after storing the checkpoint.
|
|
"""
|
|
configurable = config["configurable"].copy()
|
|
thread_id = configurable.pop("thread_id")
|
|
checkpoint_ns = configurable.pop("checkpoint_ns")
|
|
|
|
copy = checkpoint.copy()
|
|
next_config = {
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": checkpoint["id"],
|
|
}
|
|
}
|
|
|
|
async with self._cursor(pipeline=True) as cur:
|
|
await cur.execute(
|
|
"""DELETE FROM checkpoint_writes
|
|
WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id NOT IN (%s, %s)""",
|
|
(
|
|
thread_id,
|
|
checkpoint_ns,
|
|
checkpoint["id"],
|
|
configurable.get("checkpoint_id", ""),
|
|
),
|
|
)
|
|
await cur.executemany(
|
|
self.UPSERT_CHECKPOINT_BLOBS_SQL,
|
|
_dump_blobs(
|
|
self.serde,
|
|
thread_id,
|
|
checkpoint_ns,
|
|
copy.pop("channel_values"), # type: ignore[misc]
|
|
new_versions,
|
|
),
|
|
)
|
|
await cur.execute(
|
|
self.UPSERT_CHECKPOINTS_SQL,
|
|
(
|
|
thread_id,
|
|
checkpoint_ns,
|
|
Jsonb(copy),
|
|
Jsonb(get_serializable_checkpoint_metadata(config, metadata)),
|
|
),
|
|
)
|
|
return next_config
|
|
|
|
async def aput_writes(
|
|
self,
|
|
config: RunnableConfig,
|
|
writes: Sequence[tuple[str, Any]],
|
|
task_id: str,
|
|
task_path: str = "",
|
|
) -> None:
|
|
"""Store intermediate writes linked to a checkpoint asynchronously.
|
|
|
|
This method saves intermediate writes associated with a checkpoint to the database.
|
|
|
|
Args:
|
|
config: Configuration of the related checkpoint.
|
|
writes: List of writes to store, each as (channel, value) pair.
|
|
task_id: Identifier for the task creating the writes.
|
|
"""
|
|
query = (
|
|
self.UPSERT_CHECKPOINT_WRITES_SQL
|
|
if all(w[0] in WRITES_IDX_MAP for w in writes)
|
|
else self.INSERT_CHECKPOINT_WRITES_SQL
|
|
)
|
|
params = await asyncio.to_thread(
|
|
self._dump_writes,
|
|
config["configurable"]["thread_id"],
|
|
config["configurable"]["checkpoint_ns"],
|
|
config["configurable"]["checkpoint_id"],
|
|
task_id,
|
|
task_path,
|
|
writes,
|
|
)
|
|
async with self._cursor(pipeline=True) as cur:
|
|
await cur.executemany(query, params)
|
|
|
|
@asynccontextmanager
|
|
async def _cursor(
|
|
self, *, pipeline: bool = False
|
|
) -> AsyncIterator[AsyncCursor[DictRow]]:
|
|
"""Create a database cursor as a context manager.
|
|
|
|
Args:
|
|
pipeline: whether to use pipeline for the DB operations inside the context manager.
|
|
Will be applied regardless of whether the AsyncShallowPostgresSaver instance was initialized with a pipeline.
|
|
If pipeline mode is not supported, will fall back to using transaction context manager.
|
|
"""
|
|
async with _ainternal.get_connection(self.conn) as conn:
|
|
if self.pipe:
|
|
# a connection in pipeline mode can be used concurrently
|
|
# in multiple threads/coroutines, but only one cursor can be
|
|
# used at a time
|
|
try:
|
|
async with conn.cursor(binary=True, row_factory=dict_row) as cur:
|
|
yield cur
|
|
finally:
|
|
if pipeline:
|
|
await self.pipe.sync()
|
|
elif pipeline:
|
|
# a connection not in pipeline mode can only be used by one
|
|
# thread/coroutine at a time, so we acquire a lock
|
|
if self.supports_pipeline:
|
|
async with (
|
|
self.lock,
|
|
conn.pipeline(),
|
|
conn.cursor(binary=True, row_factory=dict_row) as cur,
|
|
):
|
|
yield cur
|
|
else:
|
|
# Use connection's transaction context manager when pipeline mode not supported
|
|
async with (
|
|
self.lock,
|
|
conn.transaction(),
|
|
conn.cursor(binary=True, row_factory=dict_row) as cur,
|
|
):
|
|
yield cur
|
|
else:
|
|
async with (
|
|
self.lock,
|
|
conn.cursor(binary=True, row_factory=dict_row) as cur,
|
|
):
|
|
yield cur
|
|
|
|
def list(
|
|
self,
|
|
config: RunnableConfig | None,
|
|
*,
|
|
filter: dict[str, Any] | None = None,
|
|
before: RunnableConfig | None = None,
|
|
limit: int | None = None,
|
|
) -> Iterator[CheckpointTuple]:
|
|
"""List checkpoints from the database.
|
|
|
|
This method retrieves a list of checkpoint tuples from the Postgres database based
|
|
on the provided config. For ShallowPostgresSaver, this method returns a list with
|
|
ONLY the most recent checkpoint.
|
|
"""
|
|
aiter_ = self.alist(config, filter=filter, before=before, limit=limit)
|
|
while True:
|
|
try:
|
|
yield asyncio.run_coroutine_threadsafe(
|
|
anext(aiter_), # type: ignore[arg-type]
|
|
self.loop,
|
|
).result()
|
|
except StopAsyncIteration:
|
|
break
|
|
|
|
def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
|
|
"""Get a checkpoint tuple from the database.
|
|
|
|
This method retrieves a checkpoint tuple from the Postgres database based on the
|
|
provided config (matching the thread ID in the config).
|
|
|
|
Args:
|
|
config: The config to use for retrieving the checkpoint.
|
|
|
|
Returns:
|
|
The retrieved checkpoint tuple, or None if no matching checkpoint was found.
|
|
"""
|
|
try:
|
|
# check if we are in the main thread, only bg threads can block
|
|
# we don't check in other methods to avoid the overhead
|
|
if asyncio.get_running_loop() is self.loop:
|
|
raise asyncio.InvalidStateError(
|
|
"Synchronous calls to AsyncShallowPostgresSaver are only allowed from a "
|
|
"different thread. From the main thread, use the async interface."
|
|
"For example, use `await checkpointer.aget_tuple(...)` or `await "
|
|
"graph.ainvoke(...)`."
|
|
)
|
|
except RuntimeError:
|
|
pass
|
|
return asyncio.run_coroutine_threadsafe(
|
|
self.aget_tuple(config), self.loop
|
|
).result()
|
|
|
|
def put(
|
|
self,
|
|
config: RunnableConfig,
|
|
checkpoint: Checkpoint,
|
|
metadata: CheckpointMetadata,
|
|
new_versions: ChannelVersions,
|
|
) -> RunnableConfig:
|
|
"""Save a checkpoint to the database.
|
|
|
|
This method saves a checkpoint to the Postgres database. The checkpoint is associated
|
|
with the provided config. For AsyncShallowPostgresSaver, this method saves ONLY the most recent
|
|
checkpoint and overwrites a previous checkpoint, if it exists.
|
|
|
|
Args:
|
|
config: The config to associate with the checkpoint.
|
|
checkpoint: The checkpoint to save.
|
|
metadata: Additional metadata to save with the checkpoint.
|
|
new_versions: New channel versions as of this write.
|
|
|
|
Returns:
|
|
RunnableConfig: Updated configuration after storing the checkpoint.
|
|
"""
|
|
return asyncio.run_coroutine_threadsafe(
|
|
self.aput(config, checkpoint, metadata, new_versions), self.loop
|
|
).result()
|
|
|
|
def put_writes(
|
|
self,
|
|
config: RunnableConfig,
|
|
writes: Sequence[tuple[str, Any]],
|
|
task_id: str,
|
|
task_path: str = "",
|
|
) -> None:
|
|
"""Store intermediate writes linked to a checkpoint.
|
|
|
|
This method saves intermediate writes associated with a checkpoint to the database.
|
|
|
|
Args:
|
|
config: Configuration of the related checkpoint.
|
|
writes: List of writes to store, each as (channel, value) pair.
|
|
task_id: Identifier for the task creating the writes.
|
|
task_path: Path of the task creating the writes.
|
|
"""
|
|
return asyncio.run_coroutine_threadsafe(
|
|
self.aput_writes(config, writes, task_id, task_path), self.loop
|
|
).result()
|