mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-10-01 22:15:11 +02:00
eb69f67b65
## Summary The sqlite delta history silently drops a parent checkpoint whose id sorts above its child's, losing that parent's stored value and its pending writes. The channel hydrates short with no error. Fixes #8550 ## Problem Stage 1 walked ancestors with: ```sql WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id <= ? ORDER BY checkpoint_id DESC ``` Ancestry is defined by the `parent_checkpoint_id` column. These two predicates add a second requirement: that every child's id sorts above its parent's. The contract promises monotonic ids, but that only holds within one process, so ids from processes with different clocks can break it. When the requirement is violated the parent is excluded from the stream and its seed and writes go with it. Dropping the range filter alone does not fix it: in `checkpoint_id DESC` order that parent arrives *before* the target, so the walk streams past it before it has started. ## Fix A recursive CTE anchored at the target, following `parent_checkpoint_id`: ```sql WITH RECURSIVE ancestors(checkpoint_id, parent_checkpoint_id, type, checkpoint) AS ( SELECT ... FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? UNION ALL SELECT c.... FROM ancestors a CROSS JOIN checkpoints c ON c.checkpoint_id = a.parent_checkpoint_id WHERE c.thread_id = ? AND c.checkpoint_ns = ? ) SELECT checkpoint_id, type, checkpoint FROM ancestors ``` Rows now arrive in walk order (target, parent, grandparent, ...), so `step_walk_with_row` no longer needs its off-path skip or its `parent_cid` tracking; both are removed. The query reads only true ancestors, where the old one read every row at or below the target including sibling branches. `CROSS JOIN` pins the join order. The saver never runs `ANALYZE`, and with a plain `JOIN` sqlite put `checkpoints` as the outer loop, scanning the whole thread on every recursion step. With `ancestors` outside, each step is one primary key lookup. Through `get_delta_channel_history`: | chain length | plain `JOIN` | `CROSS JOIN` | | -- | -- | -- | | 1000 | 0.032s | 0.001s | | 2000 | 0.124s | 0.003s | | 4000 | 0.475s | 0.006s | ## Cycle guard Following pointers can loop where a bounded id scan could not, and a loop is reachable through `put` alone: `put` writes with `INSERT OR REPLACE`, so re-putting an existing checkpoint id under a descendant's config repoints that checkpoint at its own descendant. The walk stops on a repeated `checkpoint_id` (one set insert per row, no depth ceiling that could truncate a long migrated thread). sqlite yields recursive rows lazily, so abandoning the cursor ends the recursion. `test_walk_terminates_when_put_makes_the_parent_chain_cycle` fails by hanging, not by asserting, if the guard regresses (confirmed by deleting the guard locally). The package has no `pytest-timeout`, so the CI job timeout is the backstop. ## Postgres No equivalent change needed. It pages the whole thread with no id bound and follows parent pointers in Python, and its upsert never rewrites `parent_checkpoint_id`, so it can neither miss this parent nor form the loop. `BaseCheckpointSaver` and `InMemorySaver` also walk parent pointers. ## Test plan New `libs/checkpoint-sqlite/tests/test_delta_parent_walk.py`: - [x] Sync and async, parametrised over both id orders; the sync case also asserts equality with `BaseCheckpointSaver` on the same rows. `parent_id_sorts_above_child` is the bug, `parent_id_sorts_below_child` the control. - [x] `test_walk_reaches_root_of_long_chain_with_descending_ids`: 40 checkpoints, only stored value at the root. - [x] `test_walk_terminates_when_put_makes_the_parent_chain_cycle`. - [x] `test_walk_step_looks_up_the_parent_by_primary_key`: asserts the recursive step's `EXPLAIN QUERY PLAN` is a key lookup, so a plain `JOIN` can't come back. Fails with it. - [x] On `main`: 3 of the 6 walk tests fail (both `parent_id_sorts_above_child` cases and the long chain). The cycle test passes on `main` too, since the old bounded scan could not loop; it guards the new path. - [x] #8550's repro returns `{'writes': [('task', 'ch', 'write-root')], 'seed': 'seed'}` sync and async (was `{'writes': []}` on `main`). - [x] `libs/checkpoint-sqlite`: `make format`, `make lint` clean; full suite 125 passed, 2 skipped. - [x] `libs/langgraph`: `-k "delta or sqlite"` 739 passed, 1 skipped. Thanks to @lylelllll for the report, the minimal repro, the base-saver comparison that isolated it to the fast path, and for suggesting the recursive CTE. Co-authored-by: lylelllll <59271327+lylelllll@users.noreply.github.com>
647 lines
25 KiB
Python
647 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import random
|
|
import sqlite3
|
|
import threading
|
|
from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
|
|
from contextlib import closing, contextmanager
|
|
from typing import Any, cast
|
|
|
|
from langchain_core.runnables import RunnableConfig
|
|
from langgraph.checkpoint.base import (
|
|
WRITES_IDX_MAP,
|
|
BaseCheckpointSaver,
|
|
ChannelVersions,
|
|
Checkpoint,
|
|
CheckpointMetadata,
|
|
CheckpointTuple,
|
|
DeltaChannelHistory,
|
|
SerializerProtocol,
|
|
get_checkpoint_id,
|
|
get_checkpoint_metadata,
|
|
)
|
|
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
|
|
|
|
from langgraph.checkpoint.sqlite._delta import (
|
|
DELTA_STAGE1_SQL,
|
|
build_delta_channels_writes_history,
|
|
build_delta_stage2_sql,
|
|
step_walk_with_row,
|
|
)
|
|
from langgraph.checkpoint.sqlite.utils import search_where
|
|
|
|
_AIO_ERROR_MSG = (
|
|
"The SqliteSaver does not support async methods. "
|
|
"Consider using AsyncSqliteSaver instead.\n"
|
|
"from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver\n"
|
|
"Note: AsyncSqliteSaver requires the aiosqlite package to use.\n"
|
|
"Install with:\n`pip install aiosqlite`\n"
|
|
"See https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver"
|
|
"for more information."
|
|
)
|
|
|
|
|
|
class SqliteSaver(BaseCheckpointSaver[str]):
|
|
"""A checkpoint saver that stores checkpoints in a SQLite database.
|
|
|
|
Note:
|
|
This class is meant for lightweight, synchronous use cases
|
|
(demos and small projects) and does not
|
|
scale to multiple threads.
|
|
For a similar sqlite saver with `async` support,
|
|
consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
|
|
|
|
Args:
|
|
conn (sqlite3.Connection): The SQLite database connection.
|
|
serde (Optional[SerializerProtocol]): The serializer to use for serializing and deserializing checkpoints. Defaults to JsonPlusSerializerCompat.
|
|
|
|
Examples:
|
|
|
|
>>> import sqlite3
|
|
>>> from langgraph.checkpoint.sqlite import SqliteSaver
|
|
>>> from langgraph.graph import StateGraph
|
|
>>>
|
|
>>> builder = StateGraph(int)
|
|
>>> builder.add_node("add_one", lambda x: x + 1)
|
|
>>> builder.set_entry_point("add_one")
|
|
>>> builder.set_finish_point("add_one")
|
|
>>> # Create a new SqliteSaver instance
|
|
>>> # Note: check_same_thread=False is OK as the implementation uses a lock
|
|
>>> # to ensure thread safety.
|
|
>>> conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
|
|
>>> memory = SqliteSaver(conn)
|
|
>>> graph = builder.compile(checkpointer=memory)
|
|
>>> config = {"configurable": {"thread_id": "1"}}
|
|
>>> graph.get_state(config)
|
|
>>> result = graph.invoke(3, config)
|
|
>>> graph.get_state(config)
|
|
StateSnapshot(values=4, next=(), config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '0c62ca34-ac19-445d-bbb0-5b4984975b2a'}}, parent_config=None)
|
|
"""
|
|
|
|
conn: sqlite3.Connection
|
|
is_setup: bool
|
|
|
|
def __init__(
|
|
self,
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
serde: SerializerProtocol | None = None,
|
|
) -> None:
|
|
super().__init__(serde=serde)
|
|
self.jsonplus_serde = JsonPlusSerializer()
|
|
self.conn = conn
|
|
self.is_setup = False
|
|
self.lock = threading.Lock()
|
|
|
|
@classmethod
|
|
@contextmanager
|
|
def from_conn_string(cls, conn_string: str) -> Iterator[SqliteSaver]:
|
|
"""Create a new SqliteSaver instance from a connection string.
|
|
|
|
Args:
|
|
conn_string: The SQLite connection string.
|
|
|
|
Yields:
|
|
SqliteSaver: A new SqliteSaver instance.
|
|
|
|
Examples:
|
|
|
|
In memory:
|
|
|
|
with SqliteSaver.from_conn_string(":memory:") as memory:
|
|
...
|
|
|
|
To disk:
|
|
|
|
with SqliteSaver.from_conn_string("checkpoints.sqlite") as memory:
|
|
...
|
|
"""
|
|
with closing(
|
|
sqlite3.connect(
|
|
conn_string,
|
|
# https://ricardoanderegg.com/posts/python-sqlite-thread-safety/
|
|
check_same_thread=False,
|
|
)
|
|
) as conn:
|
|
yield cls(conn)
|
|
|
|
def setup(self) -> None:
|
|
"""Set up the checkpoint database.
|
|
|
|
This method creates the necessary tables in the SQLite database if they don't
|
|
already exist. It is called automatically when needed and should not be called
|
|
directly by the user.
|
|
"""
|
|
if self.is_setup:
|
|
return
|
|
|
|
self.conn.executescript(
|
|
"""
|
|
PRAGMA journal_mode=WAL;
|
|
CREATE TABLE IF NOT EXISTS checkpoints (
|
|
thread_id TEXT NOT NULL,
|
|
checkpoint_ns TEXT NOT NULL DEFAULT '',
|
|
checkpoint_id TEXT NOT NULL,
|
|
parent_checkpoint_id TEXT,
|
|
type TEXT,
|
|
checkpoint BLOB,
|
|
metadata BLOB,
|
|
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS 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,
|
|
value BLOB,
|
|
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
|
|
);
|
|
"""
|
|
)
|
|
|
|
self.is_setup = True
|
|
|
|
@contextmanager
|
|
def cursor(self, transaction: bool = True) -> Iterator[sqlite3.Cursor]:
|
|
"""Get a cursor for the SQLite database.
|
|
|
|
This method returns a cursor for the SQLite database. It is used internally
|
|
by the SqliteSaver and should not be called directly by the user.
|
|
|
|
Args:
|
|
transaction (bool): Whether to commit the transaction when the cursor is closed. Defaults to True.
|
|
|
|
Yields:
|
|
sqlite3.Cursor: A cursor for the SQLite database.
|
|
"""
|
|
with self.lock:
|
|
self.setup()
|
|
cur = self.conn.cursor()
|
|
try:
|
|
yield cur
|
|
finally:
|
|
if transaction:
|
|
self.conn.commit()
|
|
cur.close()
|
|
|
|
def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
|
|
"""Get a checkpoint tuple from the database.
|
|
|
|
This method retrieves a checkpoint tuple from the SQLite database based on the
|
|
provided config. If the config contains a `checkpoint_id` key, the checkpoint with
|
|
the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint
|
|
for the given thread ID is retrieved.
|
|
|
|
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 checkpoint ID:
|
|
|
|
>>> config = {
|
|
... "configurable": {
|
|
... "thread_id": "1",
|
|
... "checkpoint_ns": "",
|
|
... "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
|
|
... }
|
|
... }
|
|
>>> checkpoint_tuple = memory.get_tuple(config)
|
|
>>> print(checkpoint_tuple)
|
|
CheckpointTuple(...)
|
|
"""
|
|
checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
|
|
with self.cursor(transaction=False) as cur:
|
|
# find the latest checkpoint for the thread_id
|
|
if checkpoint_id := get_checkpoint_id(config):
|
|
cur.execute(
|
|
"SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
|
|
(
|
|
str(config["configurable"]["thread_id"]),
|
|
checkpoint_ns,
|
|
checkpoint_id,
|
|
),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"SELECT thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata FROM checkpoints WHERE thread_id = ? AND checkpoint_ns = ? ORDER BY checkpoint_id DESC LIMIT 1",
|
|
(str(config["configurable"]["thread_id"]), checkpoint_ns),
|
|
)
|
|
# if a checkpoint is found, return it
|
|
if value := cur.fetchone():
|
|
(
|
|
thread_id,
|
|
checkpoint_id,
|
|
parent_checkpoint_id,
|
|
type,
|
|
checkpoint,
|
|
metadata,
|
|
) = value
|
|
if not get_checkpoint_id(config):
|
|
config = {
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": checkpoint_id,
|
|
}
|
|
}
|
|
# find any pending writes
|
|
cur.execute(
|
|
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
|
|
(
|
|
str(config["configurable"]["thread_id"]),
|
|
checkpoint_ns,
|
|
str(config["configurable"]["checkpoint_id"]),
|
|
),
|
|
)
|
|
# deserialize the checkpoint and metadata
|
|
return CheckpointTuple(
|
|
config,
|
|
self.serde.loads_typed((type, checkpoint)),
|
|
cast(
|
|
CheckpointMetadata,
|
|
json.loads(metadata) if metadata is not None else {},
|
|
),
|
|
(
|
|
{
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": parent_checkpoint_id,
|
|
}
|
|
}
|
|
if parent_checkpoint_id
|
|
else None
|
|
),
|
|
[
|
|
(task_id, channel, self.serde.loads_typed((type, value)))
|
|
for task_id, channel, type, value in 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 SQLite database based
|
|
on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).
|
|
|
|
Args:
|
|
config: The config to use for listing the checkpoints.
|
|
filter: Additional filtering criteria for metadata.
|
|
before: If provided, only checkpoints before the specified checkpoint ID are returned.
|
|
limit: The maximum number of checkpoints to return.
|
|
|
|
Yields:
|
|
An iterator of checkpoint tuples.
|
|
|
|
Examples:
|
|
>>> from langgraph.checkpoint.sqlite import SqliteSaver
|
|
>>> with SqliteSaver.from_conn_string(":memory:") as memory:
|
|
... # Run a graph, then list the checkpoints
|
|
>>> config = {"configurable": {"thread_id": "1"}}
|
|
>>> checkpoints = list(memory.list(config, limit=2))
|
|
>>> print(checkpoints)
|
|
[CheckpointTuple(...), CheckpointTuple(...)]
|
|
|
|
>>> config = {"configurable": {"thread_id": "1"}}
|
|
>>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
|
|
>>> with SqliteSaver.from_conn_string(":memory:") as memory:
|
|
... # Run a graph, then list the checkpoints
|
|
>>> checkpoints = list(memory.list(config, before=before))
|
|
>>> print(checkpoints)
|
|
[CheckpointTuple(...), ...]
|
|
"""
|
|
where, param_values = search_where(config, filter, before)
|
|
query = f"""SELECT thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata
|
|
FROM checkpoints
|
|
{where}
|
|
ORDER BY checkpoint_id DESC"""
|
|
if limit is not None:
|
|
query += " LIMIT ?"
|
|
param_values = (*param_values, limit)
|
|
with self.cursor(transaction=False) as cur, closing(self.conn.cursor()) as wcur:
|
|
cur.execute(query, param_values)
|
|
for (
|
|
thread_id,
|
|
checkpoint_ns,
|
|
checkpoint_id,
|
|
parent_checkpoint_id,
|
|
type,
|
|
checkpoint,
|
|
metadata,
|
|
) in cur:
|
|
wcur.execute(
|
|
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
|
|
(thread_id, checkpoint_ns, checkpoint_id),
|
|
)
|
|
yield CheckpointTuple(
|
|
{
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": checkpoint_id,
|
|
}
|
|
},
|
|
self.serde.loads_typed((type, checkpoint)),
|
|
cast(
|
|
CheckpointMetadata,
|
|
json.loads(metadata) if metadata is not None else {},
|
|
),
|
|
(
|
|
{
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": parent_checkpoint_id,
|
|
}
|
|
}
|
|
if parent_checkpoint_id
|
|
else None
|
|
),
|
|
[
|
|
(task_id, channel, self.serde.loads_typed((type, value)))
|
|
for task_id, channel, type, value in wcur
|
|
],
|
|
)
|
|
|
|
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 SQLite database. The checkpoint is associated
|
|
with the provided config and its parent config (if any).
|
|
|
|
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.sqlite import SqliteSaver
|
|
>>> with SqliteSaver.from_conn_string(":memory:") 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'}}
|
|
"""
|
|
thread_id = config["configurable"]["thread_id"]
|
|
checkpoint_ns = config["configurable"]["checkpoint_ns"]
|
|
type_, serialized_checkpoint = self.serde.dumps_typed(checkpoint)
|
|
serialized_metadata = json.dumps(
|
|
get_checkpoint_metadata(config, metadata), ensure_ascii=False
|
|
).encode("utf-8", "ignore")
|
|
with self.cursor() as cur:
|
|
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,
|
|
),
|
|
)
|
|
return {
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": checkpoint["id"],
|
|
}
|
|
}
|
|
|
|
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 SQLite 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.
|
|
"""
|
|
query = (
|
|
"INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
|
|
if all(w[0] in WRITES_IDX_MAP for w in writes)
|
|
else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
|
|
)
|
|
with self.cursor() as cur:
|
|
cur.executemany(
|
|
query,
|
|
[
|
|
(
|
|
str(config["configurable"]["thread_id"]),
|
|
str(config["configurable"]["checkpoint_ns"]),
|
|
str(config["configurable"]["checkpoint_id"]),
|
|
task_id,
|
|
WRITES_IDX_MAP.get(channel, idx),
|
|
channel,
|
|
*self.serde.dumps_typed(value),
|
|
)
|
|
for idx, (channel, value) in enumerate(writes)
|
|
],
|
|
)
|
|
|
|
def delete_thread(self, thread_id: str) -> None:
|
|
"""Delete all checkpoints and writes associated with a thread ID.
|
|
|
|
Args:
|
|
thread_id: The thread ID to delete.
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
with self.cursor() as cur:
|
|
cur.execute(
|
|
"DELETE FROM checkpoints WHERE thread_id = ?",
|
|
(str(thread_id),),
|
|
)
|
|
cur.execute(
|
|
"DELETE FROM writes WHERE thread_id = ?",
|
|
(str(thread_id),),
|
|
)
|
|
|
|
def get_delta_channel_history(
|
|
self, *, config: RunnableConfig, channels: Sequence[str]
|
|
) -> Mapping[str, DeltaChannelHistory]:
|
|
"""Fast-path override of `BaseCheckpointSaver.get_delta_channel_history`.
|
|
|
|
Two-stage query:
|
|
|
|
* Stage 1 (streamed): recursive CTE over `checkpoints` following
|
|
`parent_checkpoint_id` from the target, returning
|
|
`(checkpoint_id, type, checkpoint)` per ancestor. Sqlite has no
|
|
JSONB, so we ship the full serialized checkpoint blob and inspect
|
|
`channel_values` in Python. Stops reading when every channel has
|
|
found its seed or the chain is exhausted.
|
|
|
|
* Stage 2 (per-channel UNION ALL): one branch per channel reading
|
|
`writes` filtered to that channel's specific `chain_cids`. No
|
|
separate seed-blob fetch — sqlite stores `channel_values` inline
|
|
in the checkpoint blob, so seeds come back from stage 1.
|
|
"""
|
|
if not channels:
|
|
return {}
|
|
channels = list(channels)
|
|
thread_id = str(config["configurable"]["thread_id"])
|
|
checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
|
|
checkpoint_id = get_checkpoint_id(config)
|
|
if checkpoint_id is None:
|
|
target = self.get_tuple(config)
|
|
if target is None:
|
|
return {ch: {"writes": []} for ch in channels}
|
|
checkpoint_id = target.config["configurable"]["checkpoint_id"]
|
|
|
|
chain_by_ch: dict[str, list[str]] = {ch: [] for ch in channels}
|
|
seed_val_by_ch: dict[str, Any] = {}
|
|
walk_state: dict[str, Any] = {}
|
|
seeded: set[str] = set()
|
|
|
|
with self.cursor(transaction=False) as cur:
|
|
cur.execute(
|
|
DELTA_STAGE1_SQL,
|
|
(thread_id, checkpoint_ns, checkpoint_id, thread_id, checkpoint_ns),
|
|
)
|
|
for row in cur:
|
|
cid, type_tag, blob = row
|
|
if step_walk_with_row(
|
|
cid=cid,
|
|
type_tag=type_tag,
|
|
blob=blob,
|
|
target_id=checkpoint_id,
|
|
serde=self.serde,
|
|
chain_by_ch=chain_by_ch,
|
|
seed_val_by_ch=seed_val_by_ch,
|
|
walk_state=walk_state,
|
|
seeded=seeded,
|
|
channels=channels,
|
|
):
|
|
break
|
|
|
|
channels_with_chain = [ch for ch in channels if chain_by_ch[ch]]
|
|
stage2_sql = build_delta_stage2_sql(
|
|
chain_lens=[len(chain_by_ch[ch]) for ch in channels_with_chain],
|
|
)
|
|
if stage2_sql:
|
|
stage2_params: list[Any] = []
|
|
for ch in channels_with_chain:
|
|
stage2_params.extend(
|
|
[thread_id, checkpoint_ns, ch, *chain_by_ch[ch]]
|
|
)
|
|
cur.execute(stage2_sql, stage2_params)
|
|
stage2_rows = cast(
|
|
"list[tuple[str, str, str, int, str, bytes]]", cur.fetchall()
|
|
)
|
|
else:
|
|
stage2_rows = []
|
|
|
|
return build_delta_channels_writes_history(
|
|
channels=channels,
|
|
chain_by_ch=chain_by_ch,
|
|
seed_val_by_ch=seed_val_by_ch,
|
|
seeded=seeded,
|
|
stage2_rows=stage2_rows,
|
|
serde=self.serde,
|
|
)
|
|
|
|
async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
|
|
"""Get a checkpoint tuple from the database asynchronously.
|
|
|
|
Note:
|
|
This async method is not supported by the SqliteSaver class.
|
|
Use get_tuple() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
|
|
"""
|
|
raise NotImplementedError(_AIO_ERROR_MSG)
|
|
|
|
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.
|
|
|
|
Note:
|
|
This async method is not supported by the SqliteSaver class.
|
|
Use list() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
|
|
"""
|
|
raise NotImplementedError(_AIO_ERROR_MSG)
|
|
yield
|
|
|
|
async def aput(
|
|
self,
|
|
config: RunnableConfig,
|
|
checkpoint: Checkpoint,
|
|
metadata: CheckpointMetadata,
|
|
new_versions: ChannelVersions,
|
|
) -> RunnableConfig:
|
|
"""Save a checkpoint to the database asynchronously.
|
|
|
|
Note:
|
|
This async method is not supported by the SqliteSaver class.
|
|
Use put() instead, or consider using [AsyncSqliteSaver][langgraph.checkpoint.sqlite.aio.AsyncSqliteSaver].
|
|
"""
|
|
raise NotImplementedError(_AIO_ERROR_MSG)
|
|
|
|
def get_next_version(self, current: str | None, channel: None) -> str:
|
|
"""Generate the next version ID for a channel.
|
|
|
|
This method creates a new version identifier for a channel based on its current version.
|
|
|
|
Args:
|
|
current (Optional[str]): The current version identifier of the channel.
|
|
|
|
Returns:
|
|
str: The next version identifier, which is guaranteed to be monotonically increasing.
|
|
"""
|
|
if current is None:
|
|
current_v = 0
|
|
elif isinstance(current, int):
|
|
current_v = current
|
|
else:
|
|
current_v = int(current.split(".")[0])
|
|
next_v = current_v + 1
|
|
next_h = random.random()
|
|
return f"{next_v:032}.{next_h:016}"
|