mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-21 15:12:26 +02:00
536 lines
21 KiB
Python
536 lines
21 KiB
Python
import asyncio
|
|
from collections.abc import AsyncIterator, Iterator, Sequence
|
|
from contextlib import asynccontextmanager
|
|
from typing import Any, Optional
|
|
|
|
from langchain_core.runnables import RunnableConfig
|
|
from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
|
|
from psycopg.rows import DictRow, dict_row
|
|
from psycopg.types.json import Jsonb
|
|
from psycopg_pool import AsyncConnectionPool
|
|
|
|
from langgraph.checkpoint.base import (
|
|
WRITES_IDX_MAP,
|
|
ChannelVersions,
|
|
Checkpoint,
|
|
CheckpointMetadata,
|
|
CheckpointTuple,
|
|
get_checkpoint_id,
|
|
get_checkpoint_metadata,
|
|
)
|
|
from langgraph.checkpoint.postgres import _ainternal
|
|
from langgraph.checkpoint.postgres.base import BasePostgresSaver
|
|
from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
|
|
from langgraph.checkpoint.serde.base import SerializerProtocol
|
|
|
|
Conn = _ainternal.Conn # For backward compatibility
|
|
|
|
|
|
class AsyncPostgresSaver(BasePostgresSaver):
|
|
"""Asynchronous checkpointer that stores checkpoints in a Postgres database."""
|
|
|
|
lock: asyncio.Lock
|
|
|
|
def __init__(
|
|
self,
|
|
conn: _ainternal.Conn,
|
|
pipe: Optional[AsyncPipeline] = None,
|
|
serde: Optional[SerializerProtocol] = None,
|
|
) -> None:
|
|
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: Optional[SerializerProtocol] = None,
|
|
) -> AsyncIterator["AsyncPostgresSaver"]:
|
|
"""Create a new AsyncPostgresSaver instance from a connection string.
|
|
|
|
Args:
|
|
conn_string: The Postgres connection info string.
|
|
pipeline: whether to use AsyncPipeline
|
|
|
|
Returns:
|
|
AsyncPostgresSaver: A new AsyncPostgresSaver 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 :],
|
|
):
|
|
await cur.execute(migration)
|
|
await cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
|
|
if self.pipe:
|
|
await self.pipe.sync()
|
|
|
|
async def alist(
|
|
self,
|
|
config: Optional[RunnableConfig],
|
|
*,
|
|
filter: Optional[dict[str, Any]] = None,
|
|
before: Optional[RunnableConfig] = None,
|
|
limit: Optional[int] = 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. The checkpoints are ordered by checkpoint ID in descending order (newest first).
|
|
|
|
Args:
|
|
config: Base configuration for filtering checkpoints.
|
|
filter: Additional filtering criteria for metadata.
|
|
before: If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
|
|
limit: Maximum number of checkpoints to return.
|
|
|
|
Yields:
|
|
AsyncIterator[CheckpointTuple]: An asynchronous iterator of matching checkpoint tuples.
|
|
"""
|
|
where, args = self._search_where(config, filter, before)
|
|
query = self.SELECT_SQL + where + " ORDER BY checkpoint_id DESC"
|
|
if limit:
|
|
query += f" LIMIT {limit}"
|
|
# if we change this to use .stream() we need to make sure to close the cursor
|
|
async with self._cursor() as cur:
|
|
await cur.execute(query, args, binary=True)
|
|
async for value in cur:
|
|
yield CheckpointTuple(
|
|
{
|
|
"configurable": {
|
|
"thread_id": value["thread_id"],
|
|
"checkpoint_ns": value["checkpoint_ns"],
|
|
"checkpoint_id": value["checkpoint_id"],
|
|
}
|
|
},
|
|
await asyncio.to_thread(
|
|
self._load_checkpoint,
|
|
value["checkpoint"],
|
|
value["channel_values"],
|
|
value["pending_sends"],
|
|
),
|
|
self._load_metadata(value["metadata"]),
|
|
(
|
|
{
|
|
"configurable": {
|
|
"thread_id": value["thread_id"],
|
|
"checkpoint_ns": value["checkpoint_ns"],
|
|
"checkpoint_id": value["parent_checkpoint_id"],
|
|
}
|
|
}
|
|
if value["parent_checkpoint_id"]
|
|
else None
|
|
),
|
|
await asyncio.to_thread(self._load_writes, value["pending_writes"]),
|
|
)
|
|
|
|
async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
|
|
"""Get a checkpoint tuple from the database asynchronously.
|
|
|
|
This method retrieves a checkpoint tuple from the Postgres 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:
|
|
Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
|
|
"""
|
|
thread_id = config["configurable"]["thread_id"]
|
|
checkpoint_id = get_checkpoint_id(config)
|
|
checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
|
|
if checkpoint_id:
|
|
args: tuple[Any, ...] = (thread_id, checkpoint_ns, checkpoint_id)
|
|
where = "WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id = %s"
|
|
else:
|
|
args = (thread_id, checkpoint_ns)
|
|
where = "WHERE thread_id = %s AND checkpoint_ns = %s ORDER BY checkpoint_id DESC LIMIT 1"
|
|
|
|
async with self._cursor() as cur:
|
|
await cur.execute(
|
|
self.SELECT_SQL + where,
|
|
args,
|
|
binary=True,
|
|
)
|
|
|
|
async for value in cur:
|
|
return CheckpointTuple(
|
|
{
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": value["checkpoint_id"],
|
|
}
|
|
},
|
|
await asyncio.to_thread(
|
|
self._load_checkpoint,
|
|
value["checkpoint"],
|
|
value["channel_values"],
|
|
value["pending_sends"],
|
|
),
|
|
self._load_metadata(value["metadata"]),
|
|
(
|
|
{
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": value["parent_checkpoint_id"],
|
|
}
|
|
}
|
|
if value["parent_checkpoint_id"]
|
|
else None
|
|
),
|
|
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 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.
|
|
"""
|
|
configurable = config["configurable"].copy()
|
|
thread_id = configurable.pop("thread_id")
|
|
checkpoint_ns = configurable.pop("checkpoint_ns")
|
|
checkpoint_id = configurable.pop(
|
|
"checkpoint_id", configurable.pop("thread_ts", None)
|
|
)
|
|
|
|
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.executemany(
|
|
self.UPSERT_CHECKPOINT_BLOBS_SQL,
|
|
await asyncio.to_thread(
|
|
self._dump_blobs,
|
|
thread_id,
|
|
checkpoint_ns,
|
|
copy.pop("channel_values"), # type: ignore[misc]
|
|
new_versions,
|
|
),
|
|
)
|
|
await cur.execute(
|
|
self.UPSERT_CHECKPOINTS_SQL,
|
|
(
|
|
thread_id,
|
|
checkpoint_ns,
|
|
checkpoint["id"],
|
|
checkpoint_id,
|
|
Jsonb(self._dump_checkpoint(copy)),
|
|
self._dump_metadata(get_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)
|
|
|
|
async def adelete_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
|
|
"""
|
|
async with self._cursor(pipeline=True) as cur:
|
|
await cur.execute(
|
|
"DELETE FROM checkpoints WHERE thread_id = %s",
|
|
(str(thread_id),),
|
|
)
|
|
await cur.execute(
|
|
"DELETE FROM checkpoint_blobs WHERE thread_id = %s",
|
|
(str(thread_id),),
|
|
)
|
|
await cur.execute(
|
|
"DELETE FROM checkpoint_writes WHERE thread_id = %s",
|
|
(str(thread_id),),
|
|
)
|
|
|
|
@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 AsyncPostgresSaver 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: Optional[RunnableConfig],
|
|
*,
|
|
filter: Optional[dict[str, Any]] = None,
|
|
before: Optional[RunnableConfig] = None,
|
|
limit: Optional[int] = 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. The checkpoints are ordered by checkpoint ID in descending order (newest first).
|
|
|
|
Args:
|
|
config: Base configuration for filtering checkpoints.
|
|
filter: Additional filtering criteria for metadata.
|
|
before: If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.
|
|
limit: Maximum number of checkpoints to return.
|
|
|
|
Yields:
|
|
Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.
|
|
"""
|
|
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 AsyncPostgresSaver are only allowed from a "
|
|
"different thread. From the main thread, use the async interface. "
|
|
"For example, use `checkpointer.alist(...)` or `await "
|
|
"graph.ainvoke(...)`."
|
|
)
|
|
except RuntimeError:
|
|
pass
|
|
aiter_ = self.alist(config, filter=filter, before=before, limit=limit)
|
|
while True:
|
|
try:
|
|
yield asyncio.run_coroutine_threadsafe(
|
|
anext(aiter_), # noqa: F821
|
|
self.loop,
|
|
).result()
|
|
except StopAsyncIteration:
|
|
break
|
|
|
|
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
|
|
"""Get a checkpoint tuple from the database.
|
|
|
|
This method retrieves a checkpoint tuple from the Postgres 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:
|
|
Optional[CheckpointTuple]: 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 AsyncPostgresSaver 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 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.
|
|
"""
|
|
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()
|
|
|
|
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
|
|
"""
|
|
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 AsyncPostgresSaver 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.adelete_thread(thread_id), self.loop
|
|
).result()
|
|
|
|
|
|
__all__ = ["AsyncPostgresSaver", "AsyncShallowPostgresSaver", "Conn"]
|