From 0cad7019cb077cfe19ca3feafb93d2ae3d0889d6 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Fri, 13 Jun 2025 16:37:26 -0700 Subject: [PATCH 01/13] Restore shallow checkpointer - This should definitely be removed soon, but let's give people more time to update --- .../langgraph/checkpoint/postgres/__init__.py | 3 +- .../langgraph/checkpoint/postgres/aio.py | 3 +- .../langgraph/checkpoint/postgres/shallow.py | 959 ++++++++++++++++++ libs/checkpoint-postgres/tests/test_async.py | 41 +- libs/checkpoint-postgres/tests/test_sync.py | 34 +- 5 files changed, 1030 insertions(+), 10 deletions(-) create mode 100644 libs/checkpoint-postgres/langgraph/checkpoint/postgres/shallow.py diff --git a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py index 8e4595a37..691654e6b 100644 --- a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py +++ b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py @@ -23,6 +23,7 @@ from langgraph.checkpoint.base import ( ) from langgraph.checkpoint.postgres import _internal from langgraph.checkpoint.postgres.base import BasePostgresSaver +from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver from langgraph.checkpoint.serde.base import SerializerProtocol Conn = _internal.Conn # For backward compatibility @@ -456,4 +457,4 @@ class PostgresSaver(BasePostgresSaver): ) -__all__ = ["PostgresSaver", "BasePostgresSaver", "Conn"] +__all__ = ["PostgresSaver", "BasePostgresSaver", "ShallowPostgresSaver", "Conn"] diff --git a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py index 3d6396f29..9fc7673b2 100644 --- a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py +++ b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py @@ -23,6 +23,7 @@ from langgraph.checkpoint.base import ( ) 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 @@ -559,4 +560,4 @@ class AsyncPostgresSaver(BasePostgresSaver): ).result() -__all__ = ["AsyncPostgresSaver", "Conn"] +__all__ = ["AsyncPostgresSaver", "AsyncShallowPostgresSaver", "Conn"] diff --git a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/shallow.py b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/shallow.py new file mode 100644 index 000000000..af2aacef8 --- /dev/null +++ b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/shallow.py @@ -0,0 +1,959 @@ +import asyncio +import threading +import warnings +from collections.abc import AsyncIterator, Iterator, Sequence +from contextlib import asynccontextmanager, contextmanager +from typing import Any, Optional + +from langchain_core.runnables import RunnableConfig +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.base import ( + WRITES_IDX_MAP, + ChannelVersions, + Checkpoint, + CheckpointMetadata, + CheckpointTuple, + get_checkpoint_metadata, +) +from langgraph.checkpoint.postgres import _ainternal, _internal +from langgraph.checkpoint.postgres.base import BasePostgresSaver +from langgraph.checkpoint.serde.base import SerializerProtocol +from langgraph.checkpoint.serde.types import TASKS + +""" +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 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] order by 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.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, Optional[bytes]]]: + 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: Optional[Pipeline] = None, + serde: Optional[SerializerProtocol] = 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(..., checkpoint_during=False)`.", + 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 :], + ): + cur.execute(migration) + cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})") + if self.pipe: + self.pipe.sync() + + 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. 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 + if limit: + query += f" LIMIT {limit}" + 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 [], + } + 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) -> Optional[CheckpointTuple]: + """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: + Optional[CheckpointTuple]: 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(...) + """ # noqa + 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_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: Optional[AsyncPipeline] = None, + serde: Optional[SerializerProtocol] = 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(..., checkpoint_during=False)`.", + 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: Optional[SerializerProtocol] = 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 :], + ): + 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. 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 + if limit: + query += f" LIMIT {limit}" + 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 [], + } + 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) -> 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 (matching the thread ID in the config). + + 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_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_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: 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. 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_), # 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 (matching the thread ID in the config). + + 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 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() diff --git a/libs/checkpoint-postgres/tests/test_async.py b/libs/checkpoint-postgres/tests/test_async.py index 59669d17c..f0196f845 100644 --- a/libs/checkpoint-postgres/tests/test_async.py +++ b/libs/checkpoint-postgres/tests/test_async.py @@ -15,7 +15,10 @@ from langgraph.checkpoint.base import ( Checkpoint, CheckpointMetadata, ) -from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver +from langgraph.checkpoint.postgres.aio import ( + AsyncPostgresSaver, + AsyncShallowPostgresSaver, +) from langgraph.checkpoint.serde.types import TASKS from tests.checkpoint_utils import create_checkpoint, empty_checkpoint from tests.conftest import DEFAULT_POSTGRES_URI @@ -108,11 +111,41 @@ async def _base_saver(): await conn.execute(f"DROP DATABASE {database}") +@asynccontextmanager +async def _shallow_saver(): + """Fixture for shallow connection mode testing.""" + database = f"test_{uuid4().hex[:16]}" + # create unique db + async with await AsyncConnection.connect( + DEFAULT_POSTGRES_URI, autocommit=True + ) as conn: + await conn.execute(f"CREATE DATABASE {database}") + try: + async with await AsyncConnection.connect( + DEFAULT_POSTGRES_URI + database, + autocommit=True, + prepare_threshold=0, + row_factory=dict_row, + ) as conn: + checkpointer = AsyncShallowPostgresSaver(conn) + await checkpointer.setup() + yield checkpointer + finally: + # drop unique db + async with await AsyncConnection.connect( + DEFAULT_POSTGRES_URI, autocommit=True + ) as conn: + await conn.execute(f"DROP DATABASE {database}") + + @asynccontextmanager async def _saver(name: str): if name == "base": async with _base_saver() as saver: yield saver + elif name == "shallow": + async with _shallow_saver() as saver: + yield saver elif name == "pool": async with _pool_saver() as saver: yield saver @@ -172,7 +205,7 @@ def test_data(): } -@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"]) +@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"]) async def test_combined_metadata(saver_name: str, test_data) -> None: async with _saver(saver_name) as saver: config = { @@ -199,7 +232,7 @@ async def test_combined_metadata(saver_name: str, test_data) -> None: } -@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"]) +@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"]) async def test_asearch(saver_name: str, test_data) -> None: async with _saver(saver_name) as saver: configs = test_data["configs"] @@ -250,7 +283,7 @@ async def test_asearch(saver_name: str, test_data) -> None: } == {"", "inner"} -@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"]) +@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"]) async def test_null_chars(saver_name: str, test_data) -> None: async with _saver(saver_name) as saver: config = await saver.aput( diff --git a/libs/checkpoint-postgres/tests/test_sync.py b/libs/checkpoint-postgres/tests/test_sync.py index b6eea12bf..3ce48c8da 100644 --- a/libs/checkpoint-postgres/tests/test_sync.py +++ b/libs/checkpoint-postgres/tests/test_sync.py @@ -16,7 +16,7 @@ from langgraph.checkpoint.base import ( Checkpoint, CheckpointMetadata, ) -from langgraph.checkpoint.postgres import PostgresSaver +from langgraph.checkpoint.postgres import PostgresSaver, ShallowPostgresSaver from langgraph.checkpoint.serde.types import TASKS from tests.checkpoint_utils import create_checkpoint, empty_checkpoint from tests.conftest import DEFAULT_POSTGRES_URI @@ -97,11 +97,37 @@ def _base_saver(): conn.execute(f"DROP DATABASE {database}") +@contextmanager +def _shallow_saver(): + """Fixture for regular connection mode testing with a shallow checkpointer.""" + database = f"test_{uuid4().hex[:16]}" + # create unique db + with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn: + conn.execute(f"CREATE DATABASE {database}") + try: + with Connection.connect( + DEFAULT_POSTGRES_URI + database, + autocommit=True, + prepare_threshold=0, + row_factory=dict_row, + ) as conn: + checkpointer = ShallowPostgresSaver(conn) + checkpointer.setup() + yield checkpointer + finally: + # drop unique db + with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn: + conn.execute(f"DROP DATABASE {database}") + + @contextmanager def _saver(name: str): if name == "base": with _base_saver() as saver: yield saver + elif name == "shallow": + with _shallow_saver() as saver: + yield saver elif name == "pool": with _pool_saver() as saver: yield saver @@ -161,7 +187,7 @@ def test_data(): } -@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"]) +@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"]) def test_combined_metadata(saver_name: str, test_data) -> None: with _saver(saver_name) as saver: config = { @@ -188,7 +214,7 @@ def test_combined_metadata(saver_name: str, test_data) -> None: } -@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"]) +@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"]) def test_search(saver_name: str, test_data) -> None: with _saver(saver_name) as saver: configs = test_data["configs"] @@ -237,7 +263,7 @@ def test_search(saver_name: str, test_data) -> None: } == {"", "inner"} -@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"]) +@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"]) def test_null_chars(saver_name: str, test_data) -> None: with _saver(saver_name) as saver: config = saver.put( From 21906d2b7bd4abebb207c86e776a4029353567c3 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Fri, 13 Jun 2025 17:33:19 -0700 Subject: [PATCH 02/13] Add migration for pending_sends - Checkpoints saved on older versions of langgraph will be compatible with langgraph 0.5 and 1.0 --- libs/langgraph/langgraph/graph/state.py | 1 + libs/langgraph/langgraph/pregel/__init__.py | 7 +++++- libs/langgraph/tests/conftest.py | 5 ++++ libs/langgraph/tests/conftest_checkpointer.py | 11 ++++++++- libs/langgraph/tests/memory_assert.py | 24 +++++++++++++++++++ 5 files changed, 46 insertions(+), 2 deletions(-) diff --git a/libs/langgraph/langgraph/graph/state.py b/libs/langgraph/langgraph/graph/state.py index f0d9e46eb..7081402c7 100644 --- a/libs/langgraph/langgraph/graph/state.py +++ b/libs/langgraph/langgraph/graph/state.py @@ -1101,6 +1101,7 @@ class CompiledStateGraph( def _migrate_checkpoint(self, checkpoint: Checkpoint) -> None: """Migrate a checkpoint to new channel layout.""" + super()._migrate_checkpoint(checkpoint) values = checkpoint["channel_values"] versions = checkpoint["channel_versions"] diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index 3ee6420fa..aef5c7ce3 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -908,7 +908,12 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou def _migrate_checkpoint(self, checkpoint: Checkpoint) -> None: """Migrate a saved checkpoint to new channel layout.""" - pass + if checkpoint["v"] < 4 and checkpoint.get("pending_sends"): + pending_sends: list[Send] = checkpoint.pop("pending_sends") + checkpoint["channel_values"][TASKS] = pending_sends + checkpoint["channel_versions"][TASKS] = max( + checkpoint["channel_versions"].values() + ) def _prepare_state_snapshot( self, diff --git a/libs/langgraph/tests/conftest.py b/libs/langgraph/tests/conftest.py index c6226cb72..269c3e66b 100644 --- a/libs/langgraph/tests/conftest.py +++ b/libs/langgraph/tests/conftest.py @@ -12,6 +12,7 @@ from langgraph.checkpoint.base import BaseCheckpointSaver from langgraph.store.base import BaseStore from tests.conftest_checkpointer import ( _checkpointer_memory, + _checkpointer_memory_migrate_sends, _checkpointer_postgres, _checkpointer_postgres_aio, _checkpointer_postgres_aio_pipe, @@ -125,6 +126,7 @@ async def async_store(request: pytest.FixtureRequest) -> AsyncIterator[BaseStore if NO_DOCKER else [ "memory", + "memory_migrate_sends", "sqlite", "sqlite_aes", "postgres", @@ -139,6 +141,9 @@ def sync_checkpointer( if checkpointer_name == "memory": with _checkpointer_memory() as checkpointer: yield checkpointer + elif checkpointer_name == "memory_migrate_sends": + with _checkpointer_memory_migrate_sends() as checkpointer: + yield checkpointer elif checkpointer_name == "sqlite": with _checkpointer_sqlite() as checkpointer: yield checkpointer diff --git a/libs/langgraph/tests/conftest_checkpointer.py b/libs/langgraph/tests/conftest_checkpointer.py index ba15a8251..deb8802f4 100644 --- a/libs/langgraph/tests/conftest_checkpointer.py +++ b/libs/langgraph/tests/conftest_checkpointer.py @@ -14,7 +14,10 @@ from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver pytest.register_assert_rewrite("tests.memory_assert") -from tests.memory_assert import MemorySaverAssertImmutable # noqa: E402 +from tests.memory_assert import ( # noqa: E402 + MemorySaverAssertImmutable, + MemorySaverNeedsPendingSendsMigration, +) DEFAULT_POSTGRES_URI = "postgres://postgres:postgres@localhost:5442/" @@ -24,6 +27,11 @@ def _checkpointer_memory(): yield MemorySaverAssertImmutable() +@contextmanager +def _checkpointer_memory_migrate_sends(): + yield MemorySaverNeedsPendingSendsMigration() + + @contextmanager def _checkpointer_sqlite(): with SqliteSaver.from_conn_string(":memory:") as checkpointer: @@ -187,6 +195,7 @@ async def _checkpointer_postgres_aio_pool(): __all__ = [ "_checkpointer_memory", + "_checkpointer_memory_migrate_sends", "_checkpointer_sqlite", "_checkpointer_sqlite_aes", "_checkpointer_postgres", diff --git a/libs/langgraph/tests/memory_assert.py b/libs/langgraph/tests/memory_assert.py index 3a1ef4536..43eb1aee6 100644 --- a/libs/langgraph/tests/memory_assert.py +++ b/libs/langgraph/tests/memory_assert.py @@ -7,6 +7,7 @@ from typing import Any, Optional from langchain_core.runnables import RunnableConfig from langgraph.checkpoint.base import ( + BaseCheckpointSaver, ChannelVersions, Checkpoint, CheckpointMetadata, @@ -14,6 +15,7 @@ from langgraph.checkpoint.base import ( SerializerProtocol, ) from langgraph.checkpoint.memory import InMemorySaver, PersistentDict +from langgraph.constants import TASKS class NoopSerializer(SerializerProtocol): @@ -24,6 +26,28 @@ class NoopSerializer(SerializerProtocol): return "type", obj +class MemorySaverNeedsPendingSendsMigration(BaseCheckpointSaver): + def __init__(self) -> None: + self.saver = InMemorySaver() + + def __getattribute__(self, name): + if name in ("saver", "__class__", "get_tuple"): + return object.__getattribute__(self, name) + return getattr(self.saver, name) + + def get_tuple(self, config): + if tup := self.saver.get_tuple(config): + if tup.checkpoint["v"] == 4 and tup.checkpoint["channel_values"].get(TASKS): + tup.checkpoint["v"] = 3 + tup.checkpoint["pending_sends"] = tup.checkpoint["channel_values"].pop( + TASKS + ) + tup.checkpoint["channel_versions"].pop(TASKS) + for seen in tup.checkpoint["versions_seen"].values(): + seen.pop(TASKS, None) + return tup + + class MemorySaverAssertImmutable(InMemorySaver): storage_for_copies: defaultdict[str, dict[str, dict[str, Checkpoint]]] From 25a59447c1cbe9fdb448081a83ee3b3752fb7be3 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 16 Jun 2025 08:47:45 -0700 Subject: [PATCH 03/13] Introduce "tasks" and "checkpoints" stream modes - These are split out of "debug" stream mode, which is now an alias for ["tasks", "checkpoints"] --- libs/langgraph/langgraph/pregel/__init__.py | 5 ++++- libs/langgraph/langgraph/pregel/loop.py | 11 ++++++----- libs/langgraph/langgraph/types.py | 5 +++-- 3 files changed, 13 insertions(+), 8 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index 3ee6420fa..84eef0abe 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -2236,6 +2236,8 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou stream_mode = ["values"] elif stream_mode is None: stream_mode = self.stream_mode + elif stream_mode == "debug": + stream_mode = ["checkpoints", "tasks"] if not isinstance(stream_mode, list): stream_mode = [stream_mode] if self.checkpointer is False: @@ -2298,7 +2300,8 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou - `"custom"`: Emit custom data from inside nodes or tasks using `StreamWriter`. - `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks. Will be emitted as 2-tuples `(LLM token, metadata)`. - - `"debug"`: Emit debug events with as much information as possible for each step. + - `"checkpoints"`: Emit an event when a checkpoint is created, in the same format as returned by get_state(). + - `"tasks"`: Emit events when tasks start and finish, including their results and errors. You can pass a list as the `stream_mode` parameter to stream multiple modes at once. The streamed outputs will be tuples of `(mode, data)`. diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index ee4382557..9043d6041 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -119,6 +119,7 @@ from langgraph.types import ( PregelScratchpad, RetryPolicy, StreamChunk, + StreamMode, StreamProtocol, ) from langgraph.utils.config import patch_configurable @@ -422,7 +423,7 @@ class PregelLoop: ), ): # produce debug output - self._emit("debug", map_debug_tasks, self.step, [pushed]) + self._emit("tasks", map_debug_tasks, self.step, [pushed]) # debug flag if self.debug: print_step_tasks(self.step, [pushed]) @@ -472,7 +473,7 @@ class PregelLoop: # produce debug output if self._checkpointer_put_after_previous is not None: self._emit( - "debug", + "checkpoints", map_debug_checkpoint, self.step - 1, # printing checkpoint for previous step { @@ -509,7 +510,7 @@ class PregelLoop: raise GraphInterrupt() # produce debug output - self._emit("debug", map_debug_tasks, self.step, self.tasks.values()) + self._emit("tasks", map_debug_tasks, self.step, self.tasks.values()) # debug flag if self.debug: @@ -834,7 +835,7 @@ class PregelLoop: def _emit( self, - mode: str, + mode: StreamMode, values: Callable[P, Iterator[Any]], *args: P.args, **kwargs: P.kwargs, @@ -885,7 +886,7 @@ class PregelLoop: ) if not cached: self._emit( - "debug", + "tasks", map_debug_task_results, self.step, (task, writes), diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 3e907e3eb..a5004ab2d 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -46,7 +46,7 @@ Checkpointer = Union[None, bool, BaseCheckpointSaver] - False disables checkpointing, even if the parent graph has a checkpointer. - None inherits checkpointer from the parent graph.""" -StreamMode = Literal["values", "updates", "debug", "messages", "custom"] +StreamMode = Literal["values", "updates", "checkpoints", "tasks", "messages", "custom"] """How the stream method should emit outputs. - `"values"`: Emit all values in the state after each step, including interrupts. @@ -55,7 +55,8 @@ StreamMode = Literal["values", "updates", "debug", "messages", "custom"] If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately. - `"custom"`: Emit custom data using from inside nodes or tasks using `StreamWriter`. - `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks. -- `"debug"`: Emit debug events with as much information as possible for each step. +- `"checkpoints"`: Emit an event when a checkpoint is created, in the same format as returned by get_state(). +- `"tasks"`: Emit events when tasks start and finish, including their results and errors. """ StreamWriter = Callable[[Any], None] From 417103066bde3680530e83b5dda0cc88b3daae77 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 16 Jun 2025 08:51:18 -0700 Subject: [PATCH 04/13] Lint --- libs/langgraph/langgraph/pregel/__init__.py | 2 - libs/langgraph/langgraph/pregel/debug.py | 149 +++++++------------- libs/langgraph/langgraph/pregel/loop.py | 34 ++++- libs/langgraph/langgraph/types.py | 5 +- libs/sdk-py/langgraph_sdk/schema.py | 12 +- 5 files changed, 90 insertions(+), 112 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index 84eef0abe..3946ac746 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -2236,8 +2236,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou stream_mode = ["values"] elif stream_mode is None: stream_mode = self.stream_mode - elif stream_mode == "debug": - stream_mode = ["checkpoints", "tasks"] if not isinstance(stream_mode, list): stream_mode = [stream_mode] if self.checkpointer is False: diff --git a/libs/langgraph/langgraph/pregel/debug.py b/libs/langgraph/langgraph/pregel/debug.py index 1733d1ff3..fff84ac45 100644 --- a/libs/langgraph/langgraph/pregel/debug.py +++ b/libs/langgraph/langgraph/pregel/debug.py @@ -3,13 +3,8 @@ from __future__ import annotations from collections import defaultdict from collections.abc import Iterable, Iterator, Mapping, Sequence from dataclasses import asdict -from datetime import datetime, timezone from pprint import pformat -from typing import ( - Any, - Literal, - Union, -) +from typing import Any from uuid import UUID from langchain_core.runnables.config import RunnableConfig @@ -17,7 +12,7 @@ from langchain_core.utils.input import get_bolded_text, get_colored_text from typing_extensions import TypedDict from langgraph.channels.base import BaseChannel -from langgraph.checkpoint.base import Checkpoint, CheckpointMetadata, PendingWrite +from langgraph.checkpoint.base import CheckpointMetadata, PendingWrite from langgraph.constants import ( CONF, CONFIG_KEY_CHECKPOINT_NS, @@ -66,82 +61,43 @@ class CheckpointPayload(TypedDict): tasks: list[CheckpointTask] -class DebugOutputBase(TypedDict): - timestamp: str - step: int - - -class DebugOutputTask(DebugOutputBase): - type: Literal["task"] - payload: TaskPayload - - -class DebugOutputTaskResult(DebugOutputBase): - type: Literal["task_result"] - payload: TaskResultPayload - - -class DebugOutputCheckpoint(DebugOutputBase): - type: Literal["checkpoint"] - payload: CheckpointPayload - - -DebugOutput = Union[DebugOutputTask, DebugOutputTaskResult, DebugOutputCheckpoint] - - TASK_NAMESPACE = UUID("6ba7b831-9dad-11d1-80b4-00c04fd430c8") -def map_debug_tasks( - step: int, tasks: Iterable[PregelExecutableTask] -) -> Iterator[DebugOutputTask]: +def map_debug_tasks(tasks: Iterable[PregelExecutableTask]) -> Iterator[TaskPayload]: """Produce "task" events for stream_mode=debug.""" - ts = datetime.now(timezone.utc).isoformat() for task in tasks: if task.config is not None and TAG_HIDDEN in task.config.get("tags", []): continue yield { - "type": "task", - "timestamp": ts, - "step": step, - "payload": { - "id": task.id, - "name": task.name, - "input": task.input, - "triggers": task.triggers, - }, + "id": task.id, + "name": task.name, + "input": task.input, + "triggers": task.triggers, } def map_debug_task_results( - step: int, task_tup: tuple[PregelExecutableTask, Sequence[tuple[str, Any]]], stream_keys: str | Sequence[str], -) -> Iterator[DebugOutputTaskResult]: +) -> Iterator[TaskResultPayload]: """Produce "task_result" events for stream_mode=debug.""" stream_channels_list = ( [stream_keys] if isinstance(stream_keys, str) else stream_keys ) task, writes = task_tup yield { - "type": "task_result", - "timestamp": datetime.now(timezone.utc).isoformat(), - "step": step, - "payload": { - "id": task.id, - "name": task.name, - "error": next((w[1] for w in writes if w[0] == ERROR), None), - "result": [ - w for w in writes if w[0] in stream_channels_list or w[0] == RETURN - ], - "interrupts": [ - asdict(v) - for w in writes - if w[0] == INTERRUPT - for v in (w[1] if isinstance(w[1], Sequence) else [w[1]]) - ], - }, + "id": task.id, + "name": task.name, + "error": next((w[1] for w in writes if w[0] == ERROR), None), + "result": [w for w in writes if w[0] in stream_channels_list or w[0] == RETURN], + "interrupts": [ + asdict(v) + for w in writes + if w[0] == INTERRUPT + for v in (w[1] if isinstance(w[1], Sequence) else [w[1]]) + ], } @@ -159,17 +115,15 @@ def rm_pregel_keys(config: RunnableConfig | None) -> RunnableConfig | None: def map_debug_checkpoint( - step: int, config: RunnableConfig, channels: Mapping[str, BaseChannel], stream_channels: str | Sequence[str], metadata: CheckpointMetadata, - checkpoint: Checkpoint, tasks: Iterable[PregelExecutableTask], pending_writes: list[PendingWrite], parent_config: RunnableConfig | None, output_keys: str | Sequence[str], -) -> Iterator[DebugOutputCheckpoint]: +) -> Iterator[CheckpointPayload]: """Produce "checkpoint" events for stream_mode=debug.""" parent_ns = config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, "") @@ -193,42 +147,35 @@ def map_debug_checkpoint( } yield { - "type": "checkpoint", - "timestamp": checkpoint["ts"], - "step": step, - "payload": { - "config": rm_pregel_keys(patch_checkpoint_map(config, metadata)), - "parent_config": rm_pregel_keys( - patch_checkpoint_map(parent_config, metadata) - ), - "values": read_channels(channels, stream_channels), - "metadata": metadata, - "next": [t.name for t in tasks], - "tasks": [ - { - "id": t.id, - "name": t.name, - "error": t.error, - "state": t.state, - } - if t.error - else { - "id": t.id, - "name": t.name, - "result": t.result, - "interrupts": tuple(asdict(i) for i in t.interrupts), - "state": t.state, - } - if t.result - else { - "id": t.id, - "name": t.name, - "interrupts": tuple(asdict(i) for i in t.interrupts), - "state": t.state, - } - for t in tasks_w_writes(tasks, pending_writes, task_states, output_keys) - ], - }, + "config": rm_pregel_keys(patch_checkpoint_map(config, metadata)), + "parent_config": rm_pregel_keys(patch_checkpoint_map(parent_config, metadata)), + "values": read_channels(channels, stream_channels), + "metadata": metadata, + "next": [t.name for t in tasks], + "tasks": [ + { + "id": t.id, + "name": t.name, + "error": t.error, + "state": t.state, + } + if t.error + else { + "id": t.id, + "name": t.name, + "result": t.result, + "interrupts": tuple(asdict(i) for i in t.interrupts), + "state": t.state, + } + if t.result + else { + "id": t.id, + "name": t.name, + "interrupts": tuple(asdict(i) for i in t.interrupts), + "state": t.state, + } + for t in tasks_w_writes(tasks, pending_writes, task_states, output_keys) + ], } diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index 9043d6041..07d4a97ad 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -11,6 +11,7 @@ from contextlib import ( AsyncExitStack, ExitStack, ) +from datetime import datetime, timezone from inspect import signature from types import TracebackType from typing import ( @@ -423,7 +424,7 @@ class PregelLoop: ), ): # produce debug output - self._emit("tasks", map_debug_tasks, self.step, [pushed]) + self._emit("tasks", map_debug_tasks, [pushed]) # debug flag if self.debug: print_step_tasks(self.step, [pushed]) @@ -475,7 +476,6 @@ class PregelLoop: self._emit( "checkpoints", map_debug_checkpoint, - self.step - 1, # printing checkpoint for previous step { **self.checkpoint_config, CONF: { @@ -486,7 +486,6 @@ class PregelLoop: self.channels, self.stream_keys, self.checkpoint_metadata, - self.checkpoint, self.tasks.values(), self.checkpoint_pending_writes, self.prev_checkpoint_config, @@ -510,7 +509,7 @@ class PregelLoop: raise GraphInterrupt() # produce debug output - self._emit("tasks", map_debug_tasks, self.step, self.tasks.values()) + self._emit("tasks", map_debug_tasks, self.tasks.values()) # debug flag if self.debug: @@ -842,10 +841,32 @@ class PregelLoop: ) -> None: if self.stream is None: return - if mode not in self.stream.modes: + debug_remap = mode in ("checkpoints", "tasks") and "debug" in self.stream.modes + if mode not in self.stream.modes and not debug_remap: return for v in values(*args, **kwargs): - self.stream((self.checkpoint_ns, mode, v)) + if mode in self.stream.modes: + self.stream((self.checkpoint_ns, mode, v)) + # "debug" mode is "checkpoints" or "tasks" with a wrapper dict + if debug_remap: + self.stream( + ( + self.checkpoint_ns, + "debug", + { + "step": self.step - 1 + if mode == "checkpoints" + else self.step, + "timestamp": datetime.now(timezone.utc).isoformat(), + "type": "checkpoint" + if mode == "checkpoints" + else "task_result" + if "result" in v + else "task", + "payload": v, + }, + ) + ) def output_writes( self, task_id: str, writes: WritesT, *, cached: bool = False @@ -888,7 +909,6 @@ class PregelLoop: self._emit( "tasks", map_debug_task_results, - self.step, (task, writes), self.stream_keys, ) diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index a5004ab2d..292deee03 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -46,7 +46,9 @@ Checkpointer = Union[None, bool, BaseCheckpointSaver] - False disables checkpointing, even if the parent graph has a checkpointer. - None inherits checkpointer from the parent graph.""" -StreamMode = Literal["values", "updates", "checkpoints", "tasks", "messages", "custom"] +StreamMode = Literal[ + "values", "updates", "checkpoints", "tasks", "debug", "messages", "custom" +] """How the stream method should emit outputs. - `"values"`: Emit all values in the state after each step, including interrupts. @@ -57,6 +59,7 @@ StreamMode = Literal["values", "updates", "checkpoints", "tasks", "messages", "c - `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks. - `"checkpoints"`: Emit an event when a checkpoint is created, in the same format as returned by get_state(). - `"tasks"`: Emit events when tasks start and finish, including their results and errors. +- `"debug"`: Emit "checlkpoints" and "tasks" events, for debugging purposes. """ StreamWriter = Callable[[Any], None] diff --git a/libs/sdk-py/langgraph_sdk/schema.py b/libs/sdk-py/langgraph_sdk/schema.py index c3e135603..5aa413be9 100644 --- a/libs/sdk-py/langgraph_sdk/schema.py +++ b/libs/sdk-py/langgraph_sdk/schema.py @@ -36,7 +36,15 @@ Represents the status of a thread: """ StreamMode = Literal[ - "values", "messages", "updates", "events", "debug", "custom", "messages-tuple" + "values", + "messages", + "updates", + "events", + "tasks", + "checkpoints", + "debug", + "custom", + "messages-tuple", ] """ Defines the mode of streaming: @@ -44,6 +52,8 @@ Defines the mode of streaming: - "messages": Stream complete messages. - "updates": Stream updates to the state. - "events": Stream events occurring during execution. +- "checkpoints": Stream checkpoints as they are created. +- "tasks": Stream task start and finish events. - "debug": Stream detailed debug information. - "custom": Stream custom events. """ From 3488ee47e05e93bec4ea8a879e66f2c955369e36 Mon Sep 17 00:00:00 2001 From: hari-dhanushkodi Date: Mon, 16 Jun 2025 13:21:42 -0400 Subject: [PATCH 05/13] chore: add docs for lgp deployment monitoring (#5104) --- docs/docs/cloud/deployment/cloud.md | 9 +++++++++ docs/docs/concepts/langgraph_control_plane.md | 10 ++++++++++ 2 files changed, 19 insertions(+) diff --git a/docs/docs/cloud/deployment/cloud.md b/docs/docs/cloud/deployment/cloud.md index 8dfe00a40..89e703599 100644 --- a/docs/docs/cloud/deployment/cloud.md +++ b/docs/docs/cloud/deployment/cloud.md @@ -62,6 +62,15 @@ Starting from the `LangGraph Platform` view... 1. In the panel, select the `Server` tab to view server logs for the revision. Server logs are only available after a revision has been deployed. 1. Within the `Server` tab, adjust the date/time range picker as needed. By default, the date/time range picker is set to the `Last 7 days`. +## View Deployment Metrics + +Starting from the LangSmith UI... + +1. In the left-hand navigation panel, select `LangGraph Platform`. The `LangGraph Platform` view contains a list of existing LangGraph Platform deployments. +1. Select an existing deployment to monitor. +1. Select the `Monitoring` tab to view the deployment metrics. See a list of [all available metrics](../../concepts/langgraph_control_plane.md#monitoring). +1. Within the `Monitoring` tab, use the date/time range picker as needed. By default, the date/time range picker is set to the `Last 15 minutes`. + ## Interrupt Revision Interrupting a revision will stop deployment of the revision. diff --git a/docs/docs/concepts/langgraph_control_plane.md b/docs/docs/concepts/langgraph_control_plane.md index 3d54ee7c5..aac377a93 100644 --- a/docs/docs/concepts/langgraph_control_plane.md +++ b/docs/docs/concepts/langgraph_control_plane.md @@ -19,6 +19,7 @@ From the control plane UI, you can: - Update a deployment. - Update environment variables for a deployment. - View build and server logs of a deployment. +- View deployment metrics like CPU and memory usage. - Delete a deployment. The Control Plane UI is embedded in [LangSmith](https://docs.smith.langchain.com/langgraph_cloud). @@ -88,6 +89,15 @@ Infrastructure for deployments and revisions are provisioned and deployed asynch The control plane and [LangGraph Data Plane](./langgraph_data_plane.md) "listener" application coordinate to achieve asynchronous deployments. +### Monitoring + +After a deployment is ready, the control plane monitors the deployment and records various metrics, such as: + +- CPU and memory usage of the deployment. +- Number of container restarts. + +These metrics are displayed as charts in the Control Plane UI. + ### LangSmith Integration A [LangSmith](https://docs.smith.langchain.com/) tracing project is automatically created for each deployment. The tracing project has the same name as the deployment. When creating a deployment, the `LANGCHAIN_TRACING` and `LANGSMITH_API_KEY`/`LANGCHAIN_API_KEY` environment variables do not need to be specified; they are set automatically by the control plane. From c1371693257f2fe4478a214d4a8781405e5fb6aa Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 16 Jun 2025 12:57:46 -0700 Subject: [PATCH 06/13] Preparation for 0.5 release - Update deprecation warnings to mention 0.5, no 1.0 - Add back type hint support for Runnable arg to add_node --- libs/langgraph/langgraph/func/__init__.py | 6 +++--- libs/langgraph/langgraph/graph/state.py | 13 +++++++------ libs/langgraph/langgraph/warnings.py | 6 +++--- libs/langgraph/tests/test_deprecation.py | 14 +++++++------- 4 files changed, 20 insertions(+), 19 deletions(-) diff --git a/libs/langgraph/langgraph/func/__init__.py b/libs/langgraph/langgraph/func/__init__.py index 2b3095b23..ba0b3c4cb 100644 --- a/libs/langgraph/langgraph/func/__init__.py +++ b/libs/langgraph/langgraph/func/__init__.py @@ -38,7 +38,7 @@ from langgraph.pregel.read import PregelNode from langgraph.pregel.write import ChannelWrite, ChannelWriteEntry from langgraph.store.base import BaseStore from langgraph.types import _DC_KWARGS, CachePolicy, RetryPolicy, StreamMode -from langgraph.warnings import LangGraphDeprecatedSinceV10 +from langgraph.warnings import LangGraphDeprecatedSinceV05 class TaskFunction(Generic[P, T]): @@ -179,7 +179,7 @@ def task( if (retry := kwargs.get("retry", UNSET)) is not UNSET: warnings.warn( "`retry` is deprecated and will be removed. Please use `retry_policy` instead.", - category=LangGraphDeprecatedSinceV10, + category=LangGraphDeprecatedSinceV05, ) if retry_policy is None: retry_policy = retry # type: ignore[assignment] @@ -383,7 +383,7 @@ class entrypoint: if (retry := kwargs.get("retry", UNSET)) is not UNSET: warnings.warn( "`retry` is deprecated and will be removed. Please use `retry_policy` instead.", - category=LangGraphDeprecatedSinceV10, + category=LangGraphDeprecatedSinceV05, ) if retry_policy is None: retry_policy = retry # type: ignore[assignment] diff --git a/libs/langgraph/langgraph/graph/state.py b/libs/langgraph/langgraph/graph/state.py index 7081402c7..151b1b278 100644 --- a/libs/langgraph/langgraph/graph/state.py +++ b/libs/langgraph/langgraph/graph/state.py @@ -86,7 +86,7 @@ from langgraph.utils.fields import ( ) from langgraph.utils.pydantic import create_model from langgraph.utils.runnable import coerce_to_runnable -from langgraph.warnings import LangGraphDeprecatedSinceV10 +from langgraph.warnings import LangGraphDeprecatedSinceV05 logger = logging.getLogger(__name__) @@ -160,6 +160,7 @@ StateNode: TypeAlias = Union[ _NodeWithConfigWriter[StateT_contra], _NodeWithConfigStore[StateT_contra], _NodeWithConfigWriterStore[StateT_contra], + Runnable[StateT_contra, Any], ] @@ -261,7 +262,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]): if (input_ := kwargs.get("input", UNSET)) is not UNSET: warnings.warn( "`input` is deprecated and will be removed. Please use `input_schema` instead.", - category=LangGraphDeprecatedSinceV10, + category=LangGraphDeprecatedSinceV05, stacklevel=2, ) if input_schema is None: @@ -270,7 +271,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]): if (output := kwargs.get("output", UNSET)) is not UNSET: warnings.warn( "`output` is deprecated and will be removed. Please use `output_schema` instead.", - category=LangGraphDeprecatedSinceV10, + category=LangGraphDeprecatedSinceV05, stacklevel=2, ) if output_schema is None: @@ -436,7 +437,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]): if (retry := kwargs.get("retry", UNSET)) is not UNSET: warnings.warn( "`retry` is deprecated and will be removed. Please use `retry_policy` instead.", - category=LangGraphDeprecatedSinceV10, + category=LangGraphDeprecatedSinceV05, ) if retry_policy is None: retry_policy = retry # type: ignore[assignment] @@ -444,7 +445,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]): if (input_ := kwargs.get("input", UNSET)) is not UNSET: warnings.warn( "`input` is deprecated and will be removed. Please use `input_schema` instead.", - category=LangGraphDeprecatedSinceV10, + category=LangGraphDeprecatedSinceV05, ) if input_schema is None: input_schema = cast(Union[type[InputT], None], input_) @@ -535,7 +536,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]): if input_schema is not None: self._add_schema(input_schema) self.nodes[node] = StateNodeSpec( - coerce_to_runnable(action, name=node, trace=False), # type: ignore + coerce_to_runnable(action, name=node, trace=False), metadata, input=input_schema or self.state_schema, retry_policy=retry_policy, diff --git a/libs/langgraph/langgraph/warnings.py b/libs/langgraph/langgraph/warnings.py index 00a5dea5a..e8fd59d88 100644 --- a/libs/langgraph/langgraph/warnings.py +++ b/libs/langgraph/langgraph/warnings.py @@ -41,8 +41,8 @@ class LangGraphDeprecationWarning(DeprecationWarning): return message -class LangGraphDeprecatedSinceV10(LangGraphDeprecationWarning): - """A specific `LangGraphDeprecationWarning` subclass defining functionality deprecated since LangGraph v1.0.0""" +class LangGraphDeprecatedSinceV05(LangGraphDeprecationWarning): + """A specific `LangGraphDeprecationWarning` subclass defining functionality deprecated since LangGraph v0.5.0""" def __init__(self, message: str, *args: object) -> None: - super().__init__(message, *args, since=(1, 0), expected_removal=(2, 0)) + super().__init__(message, *args, since=(0, 5), expected_removal=(2, 0)) diff --git a/libs/langgraph/tests/test_deprecation.py b/libs/langgraph/tests/test_deprecation.py index 456961390..2a217edae 100644 --- a/libs/langgraph/tests/test_deprecation.py +++ b/libs/langgraph/tests/test_deprecation.py @@ -4,7 +4,7 @@ from typing_extensions import TypedDict from langgraph.func import entrypoint, task from langgraph.graph import StateGraph from langgraph.types import RetryPolicy -from langgraph.warnings import LangGraphDeprecatedSinceV10 +from langgraph.warnings import LangGraphDeprecatedSinceV05 class PlainState(TypedDict): ... @@ -14,7 +14,7 @@ def test_add_node_retry_arg() -> None: builder = StateGraph(PlainState) with pytest.warns( - LangGraphDeprecatedSinceV10, + LangGraphDeprecatedSinceV05, match="`retry` is deprecated and will be removed. Please use `retry_policy` instead.", ): builder.add_node("test_node", lambda state: state, retry=RetryPolicy()) # type: ignore[arg-type] @@ -22,7 +22,7 @@ def test_add_node_retry_arg() -> None: def test_task_retry_arg() -> None: with pytest.warns( - LangGraphDeprecatedSinceV10, + LangGraphDeprecatedSinceV05, match="`retry` is deprecated and will be removed. Please use `retry_policy` instead.", ): @@ -33,7 +33,7 @@ def test_task_retry_arg() -> None: def test_entrypoint_retry_arg() -> None: with pytest.warns( - LangGraphDeprecatedSinceV10, + LangGraphDeprecatedSinceV05, match="`retry` is deprecated and will be removed. Please use `retry_policy` instead.", ): @@ -44,7 +44,7 @@ def test_entrypoint_retry_arg() -> None: def test_state_graph_input_schema() -> None: with pytest.warns( - LangGraphDeprecatedSinceV10, + LangGraphDeprecatedSinceV05, match="`input` is deprecated and will be removed. Please use `input_schema` instead.", ): StateGraph(PlainState, input=PlainState) # type: ignore[arg-type] @@ -52,7 +52,7 @@ def test_state_graph_input_schema() -> None: def test_state_graph_output_schema() -> None: with pytest.warns( - LangGraphDeprecatedSinceV10, + LangGraphDeprecatedSinceV05, match="`output` is deprecated and will be removed. Please use `output_schema` instead.", ): StateGraph(PlainState, output=PlainState) # type: ignore[arg-type] @@ -62,7 +62,7 @@ def test_add_node_input_schema() -> None: builder = StateGraph(PlainState) with pytest.warns( - LangGraphDeprecatedSinceV10, + LangGraphDeprecatedSinceV05, match="`input` is deprecated and will be removed. Please use `input_schema` instead.", ): builder.add_node("test_node", lambda state: state, input=PlainState) # type: ignore[arg-type] From 33feba4877f92d5cb70b6f826c2cdfe65f08ca4e Mon Sep 17 00:00:00 2001 From: Lauren Hirata Singh Date: Mon, 16 Jun 2025 16:33:32 -0400 Subject: [PATCH 07/13] Remove cookie consent --- docs/mkdocs.yml | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index 5c8ff289d..a1b53f559 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -364,16 +364,6 @@ markdown_extensions: hooks: - _scripts/notebook_hooks.py extra: - consent: - title: Cookie consent - actions: - - accept - - reject - description: >- - We use cookies to recognize your repeated visits and preferences, as well - as to measure the effectiveness of our documentation and whether users - find what they're searching for. Clicking "Accept" makes our - documentation better. Thank you! ❤️ social: - icon: fontawesome/brands/js link: https://langchain-ai.github.io/langgraphjs/ From 1134017d076ff66508f948752dbf5f596b049a84 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 16 Jun 2025 14:57:11 -0700 Subject: [PATCH 08/13] Preparation for 0.5 release: langgraph-checkpoint (#5124) Prepare langgraph-checkpoint for 0.5 - Given we have no upper bound on langgraph-checkpoint dep need to undo all changes in langgraph-checkpoint that might break previous versions of langgraph --- .../langgraph/checkpoint/postgres/base.py | 4 +- .../tests/checkpoint_utils.py | 53 ------------ libs/checkpoint-postgres/tests/test_async.py | 3 +- libs/checkpoint-postgres/tests/test_sync.py | 3 +- .../langgraph/checkpoint/sqlite/__init__.py | 2 +- .../langgraph/checkpoint/sqlite/aio.py | 2 +- .../tests/checkpoint_utils.py | 53 ------------ .../checkpoint-sqlite/tests/test_aiosqlite.py | 3 +- libs/checkpoint-sqlite/tests/test_sqlite.py | 3 +- .../langgraph/checkpoint/base/__init__.py | 81 +++++++++++++++---- .../langgraph/checkpoint/memory/__init__.py | 2 +- libs/checkpoint/tests/checkpoint_utils.py | 53 ------------ libs/checkpoint/tests/test_memory.py | 4 +- libs/langgraph/langgraph/pregel/__init__.py | 2 +- libs/langgraph/langgraph/pregel/algo.py | 7 +- libs/langgraph/langgraph/pregel/checkpoint.py | 11 +++ libs/langgraph/langgraph/pregel/loop.py | 19 +---- .../tests/test_checkpoint_migration.py | 7 +- libs/langgraph/tests/test_pregel.py | 2 +- libs/langgraph/tests/test_pregel_async.py | 2 +- .../langgraph/prebuilt/chat_agent_executor.py | 10 +-- libs/prebuilt/tests/memory_assert.py | 2 +- 22 files changed, 109 insertions(+), 219 deletions(-) delete mode 100644 libs/checkpoint-postgres/tests/checkpoint_utils.py delete mode 100644 libs/checkpoint-sqlite/tests/checkpoint_utils.py delete mode 100644 libs/checkpoint/tests/checkpoint_utils.py diff --git a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py index 8c502a7fe..44b8ee397 100644 --- a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py +++ b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py @@ -168,7 +168,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]): checkpoint["channel_versions"][TASKS] = ( max(checkpoint["channel_versions"].values()) if checkpoint["channel_versions"] - else self.get_next_version(None) + else self.get_next_version(None, None) ) def _load_blobs( @@ -246,7 +246,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]): for idx, (channel, value) in enumerate(writes) ] - def get_next_version(self, current: str | None) -> str: + def get_next_version(self, current: str | None, channel: None) -> str: if current is None: current_v = 0 elif isinstance(current, int): diff --git a/libs/checkpoint-postgres/tests/checkpoint_utils.py b/libs/checkpoint-postgres/tests/checkpoint_utils.py deleted file mode 100644 index f38afd740..000000000 --- a/libs/checkpoint-postgres/tests/checkpoint_utils.py +++ /dev/null @@ -1,53 +0,0 @@ -from __future__ import annotations - -from collections.abc import Mapping -from datetime import datetime, timezone -from typing import Any, Protocol - -from langgraph.checkpoint.base import Checkpoint, EmptyChannelError -from langgraph.checkpoint.base.id import uuid6 - - -class ChannelProtocol(Protocol): - def checkpoint(self) -> Any | None: ... - - -def empty_checkpoint() -> Checkpoint: - return Checkpoint( - v=1, - id=str(uuid6(clock_seq=-2)), - ts=datetime.now(timezone.utc).isoformat(), - channel_values={}, - channel_versions={}, - versions_seen={}, - ) - - -def create_checkpoint( - checkpoint: Checkpoint, - channels: Mapping[str, ChannelProtocol] | None, - step: int, - *, - id: str | None = None, -) -> Checkpoint: - """Create a checkpoint for the given channels.""" - ts = datetime.now(timezone.utc).isoformat() - if channels is None: - values = checkpoint["channel_values"] - else: - values = {} - for k, v in channels.items(): - if k not in checkpoint["channel_versions"]: - continue - try: - values[k] = v.checkpoint() - except EmptyChannelError: - pass - return Checkpoint( - v=1, - ts=ts, - id=id or str(uuid6(clock_seq=step)), - channel_values=values, - channel_versions=checkpoint["channel_versions"], - versions_seen=checkpoint["versions_seen"], - ) diff --git a/libs/checkpoint-postgres/tests/test_async.py b/libs/checkpoint-postgres/tests/test_async.py index f0196f845..905aa8968 100644 --- a/libs/checkpoint-postgres/tests/test_async.py +++ b/libs/checkpoint-postgres/tests/test_async.py @@ -14,13 +14,14 @@ from langgraph.checkpoint.base import ( EXCLUDED_METADATA_KEYS, Checkpoint, CheckpointMetadata, + create_checkpoint, + empty_checkpoint, ) from langgraph.checkpoint.postgres.aio import ( AsyncPostgresSaver, AsyncShallowPostgresSaver, ) from langgraph.checkpoint.serde.types import TASKS -from tests.checkpoint_utils import create_checkpoint, empty_checkpoint from tests.conftest import DEFAULT_POSTGRES_URI diff --git a/libs/checkpoint-postgres/tests/test_sync.py b/libs/checkpoint-postgres/tests/test_sync.py index 3ce48c8da..b010b5bbe 100644 --- a/libs/checkpoint-postgres/tests/test_sync.py +++ b/libs/checkpoint-postgres/tests/test_sync.py @@ -15,10 +15,11 @@ from langgraph.checkpoint.base import ( EXCLUDED_METADATA_KEYS, Checkpoint, CheckpointMetadata, + create_checkpoint, + empty_checkpoint, ) from langgraph.checkpoint.postgres import PostgresSaver, ShallowPostgresSaver from langgraph.checkpoint.serde.types import TASKS -from tests.checkpoint_utils import create_checkpoint, empty_checkpoint from tests.conftest import DEFAULT_POSTGRES_URI diff --git a/libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py b/libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py index caf9cdf3a..e716b1f47 100644 --- a/libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py +++ b/libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py @@ -536,7 +536,7 @@ class SqliteSaver(BaseCheckpointSaver[str]): """ raise NotImplementedError(_AIO_ERROR_MSG) - def get_next_version(self, current: str | None) -> str: + 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. diff --git a/libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py b/libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py index dd4a61ab9..6ee30b259 100644 --- a/libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py +++ b/libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py @@ -591,7 +591,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]): ) await self.conn.commit() - def get_next_version(self, current: str | None) -> str: + 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. diff --git a/libs/checkpoint-sqlite/tests/checkpoint_utils.py b/libs/checkpoint-sqlite/tests/checkpoint_utils.py deleted file mode 100644 index f38afd740..000000000 --- a/libs/checkpoint-sqlite/tests/checkpoint_utils.py +++ /dev/null @@ -1,53 +0,0 @@ -from __future__ import annotations - -from collections.abc import Mapping -from datetime import datetime, timezone -from typing import Any, Protocol - -from langgraph.checkpoint.base import Checkpoint, EmptyChannelError -from langgraph.checkpoint.base.id import uuid6 - - -class ChannelProtocol(Protocol): - def checkpoint(self) -> Any | None: ... - - -def empty_checkpoint() -> Checkpoint: - return Checkpoint( - v=1, - id=str(uuid6(clock_seq=-2)), - ts=datetime.now(timezone.utc).isoformat(), - channel_values={}, - channel_versions={}, - versions_seen={}, - ) - - -def create_checkpoint( - checkpoint: Checkpoint, - channels: Mapping[str, ChannelProtocol] | None, - step: int, - *, - id: str | None = None, -) -> Checkpoint: - """Create a checkpoint for the given channels.""" - ts = datetime.now(timezone.utc).isoformat() - if channels is None: - values = checkpoint["channel_values"] - else: - values = {} - for k, v in channels.items(): - if k not in checkpoint["channel_versions"]: - continue - try: - values[k] = v.checkpoint() - except EmptyChannelError: - pass - return Checkpoint( - v=1, - ts=ts, - id=id or str(uuid6(clock_seq=step)), - channel_values=values, - channel_versions=checkpoint["channel_versions"], - versions_seen=checkpoint["versions_seen"], - ) diff --git a/libs/checkpoint-sqlite/tests/test_aiosqlite.py b/libs/checkpoint-sqlite/tests/test_aiosqlite.py index 1e18fbb5e..503b7ade2 100644 --- a/libs/checkpoint-sqlite/tests/test_aiosqlite.py +++ b/libs/checkpoint-sqlite/tests/test_aiosqlite.py @@ -6,9 +6,10 @@ from langchain_core.runnables import RunnableConfig from langgraph.checkpoint.base import ( Checkpoint, CheckpointMetadata, + create_checkpoint, + empty_checkpoint, ) from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver -from tests.checkpoint_utils import create_checkpoint, empty_checkpoint class TestAsyncSqliteSaver: diff --git a/libs/checkpoint-sqlite/tests/test_sqlite.py b/libs/checkpoint-sqlite/tests/test_sqlite.py index 05bea2907..2a027fa3b 100644 --- a/libs/checkpoint-sqlite/tests/test_sqlite.py +++ b/libs/checkpoint-sqlite/tests/test_sqlite.py @@ -6,10 +6,11 @@ from langchain_core.runnables import RunnableConfig from langgraph.checkpoint.base import ( Checkpoint, CheckpointMetadata, + create_checkpoint, + empty_checkpoint, ) from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.checkpoint.sqlite.utils import _metadata_predicate, search_where -from tests.checkpoint_utils import create_checkpoint, empty_checkpoint class TestSqliteSaver: diff --git a/libs/checkpoint/langgraph/checkpoint/base/__init__.py b/libs/checkpoint/langgraph/checkpoint/base/__init__.py index 80b3466ed..e9350a993 100644 --- a/libs/checkpoint/langgraph/checkpoint/base/__init__.py +++ b/libs/checkpoint/langgraph/checkpoint/base/__init__.py @@ -1,10 +1,8 @@ from __future__ import annotations -from collections.abc import AsyncIterator, Iterator, Sequence -from inspect import signature +from collections.abc import AsyncIterator, Iterator, Mapping, Sequence from typing import ( # noqa: UP035 Any, - ClassVar, Generic, Literal, NamedTuple, @@ -15,6 +13,7 @@ from typing import ( # noqa: UP035 from langchain_core.runnables import RunnableConfig +from langgraph.checkpoint.base.id import uuid6 from langgraph.checkpoint.serde.base import SerializerProtocol, maybe_add_typed_methods from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer from langgraph.checkpoint.serde.types import ( @@ -22,6 +21,7 @@ from langgraph.checkpoint.serde.types import ( INTERRUPT, RESUME, SCHEDULED, + ChannelProtocol, ) V = TypeVar("V", int, float, str) @@ -91,6 +91,7 @@ def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint: channel_values=checkpoint["channel_values"].copy(), channel_versions=checkpoint["channel_versions"].copy(), versions_seen={k: v.copy() for k, v in checkpoint["versions_seen"].items()}, + pending_sends=checkpoint.get("pending_sends", []).copy(), ) @@ -118,19 +119,8 @@ class BaseCheckpointSaver(Generic[V]): versions to avoid blocking the main thread. """ - _get_next_version_legacy: ClassVar[bool] = False - """Flag indicating if get_next_version method is legacy (takes two parameters).""" - serde: SerializerProtocol = JsonPlusSerializer() - def __init_subclass__(cls) -> None: - cls._get_next_version_legacy = ( - len(signature(cls.get_next_version).parameters) > 2 # self + current - if hasattr(cls, "get_next_version") - else False - ) - return super().__init_subclass__() - def __init__( self, *, @@ -138,6 +128,15 @@ class BaseCheckpointSaver(Generic[V]): ) -> None: self.serde = maybe_add_typed_methods(serde or self.serde) + @property + def config_specs(self) -> list: + """Define the configuration options for the checkpoint saver. + + Returns: + list: List of configuration field specs. + """ + return [] + def get(self, config: RunnableConfig) -> Checkpoint | None: """Fetch a checkpoint using the given configuration. @@ -347,7 +346,7 @@ class BaseCheckpointSaver(Generic[V]): """ raise NotImplementedError - def get_next_version(self, current: V | None) -> V: + def get_next_version(self, current: V | None, channel: None) -> V: """Generate the next version ID for a channel. Default is to use integer versions, incrementing by 1. If you override, you can use str/int/float versions, @@ -355,6 +354,7 @@ class BaseCheckpointSaver(Generic[V]): Args: current: The current version identifier (int, float, or str). + channel: Deprecated argument, kept for backwards compatibility. Returns: V: The next version identifier, which must be increasing. @@ -417,3 +417,54 @@ EXCLUDED_METADATA_KEYS = { "checkpoint_ns", "checkpoint_map", } + +# --- below are deprecated utilities used by past versions of LangGraph --- + +LATEST_VERSION = 2 + + +def empty_checkpoint() -> Checkpoint: + from datetime import datetime, timezone + + return Checkpoint( + v=LATEST_VERSION, + id=str(uuid6(clock_seq=-2)), + ts=datetime.now(timezone.utc).isoformat(), + channel_values={}, + channel_versions={}, + versions_seen={}, + pending_sends=[], + ) + + +def create_checkpoint( + checkpoint: Checkpoint, + channels: Mapping[str, ChannelProtocol] | None, + step: int, + *, + id: str | None = None, +) -> Checkpoint: + """Create a checkpoint for the given channels.""" + from datetime import datetime, timezone + + ts = datetime.now(timezone.utc).isoformat() + if channels is None: + values = checkpoint["channel_values"] + else: + values = {} + for k, v in channels.items(): + if k not in checkpoint["channel_versions"]: + continue + try: + values[k] = v.checkpoint() + except EmptyChannelError: + pass + return Checkpoint( + v=LATEST_VERSION, + ts=ts, + id=id or str(uuid6(clock_seq=step)), + channel_values=values, + channel_versions=checkpoint["channel_versions"], + versions_seen=checkpoint["versions_seen"], + pending_sends=checkpoint.get("pending_sends", []), + ) diff --git a/libs/checkpoint/langgraph/checkpoint/memory/__init__.py b/libs/checkpoint/langgraph/checkpoint/memory/__init__.py index dc6d089e6..14f2a9547 100644 --- a/libs/checkpoint/langgraph/checkpoint/memory/__init__.py +++ b/libs/checkpoint/langgraph/checkpoint/memory/__init__.py @@ -512,7 +512,7 @@ class InMemorySaver( """ return self.delete_thread(thread_id) - def get_next_version(self, current: str | None) -> str: + def get_next_version(self, current: str | None, channel: None) -> str: if current is None: current_v = 0 elif isinstance(current, int): diff --git a/libs/checkpoint/tests/checkpoint_utils.py b/libs/checkpoint/tests/checkpoint_utils.py deleted file mode 100644 index f38afd740..000000000 --- a/libs/checkpoint/tests/checkpoint_utils.py +++ /dev/null @@ -1,53 +0,0 @@ -from __future__ import annotations - -from collections.abc import Mapping -from datetime import datetime, timezone -from typing import Any, Protocol - -from langgraph.checkpoint.base import Checkpoint, EmptyChannelError -from langgraph.checkpoint.base.id import uuid6 - - -class ChannelProtocol(Protocol): - def checkpoint(self) -> Any | None: ... - - -def empty_checkpoint() -> Checkpoint: - return Checkpoint( - v=1, - id=str(uuid6(clock_seq=-2)), - ts=datetime.now(timezone.utc).isoformat(), - channel_values={}, - channel_versions={}, - versions_seen={}, - ) - - -def create_checkpoint( - checkpoint: Checkpoint, - channels: Mapping[str, ChannelProtocol] | None, - step: int, - *, - id: str | None = None, -) -> Checkpoint: - """Create a checkpoint for the given channels.""" - ts = datetime.now(timezone.utc).isoformat() - if channels is None: - values = checkpoint["channel_values"] - else: - values = {} - for k, v in channels.items(): - if k not in checkpoint["channel_versions"]: - continue - try: - values[k] = v.checkpoint() - except EmptyChannelError: - pass - return Checkpoint( - v=1, - ts=ts, - id=id or str(uuid6(clock_seq=step)), - channel_values=values, - channel_versions=checkpoint["channel_versions"], - versions_seen=checkpoint["versions_seen"], - ) diff --git a/libs/checkpoint/tests/test_memory.py b/libs/checkpoint/tests/test_memory.py index b0eeb319b..ad2dbdb1e 100644 --- a/libs/checkpoint/tests/test_memory.py +++ b/libs/checkpoint/tests/test_memory.py @@ -6,12 +6,10 @@ from langchain_core.runnables import RunnableConfig from langgraph.checkpoint.base import ( Checkpoint, CheckpointMetadata, -) -from langgraph.checkpoint.memory import InMemorySaver -from tests.checkpoint_utils import ( create_checkpoint, empty_checkpoint, ) +from langgraph.checkpoint.memory import InMemorySaver class TestMemorySaver: diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index 6fbd1f6b2..5cf977637 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -32,7 +32,6 @@ from langgraph.checkpoint.base import ( BaseCheckpointSaver, Checkpoint, CheckpointTuple, - copy_checkpoint, ) from langgraph.config import get_config from langgraph.constants import ( @@ -79,6 +78,7 @@ from langgraph.pregel.algo import ( from langgraph.pregel.call import identifier from langgraph.pregel.checkpoint import ( channels_from_checkpoint, + copy_checkpoint, create_checkpoint, empty_checkpoint, ) diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index ce2b22411..9d15aff3c 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -83,7 +83,7 @@ from langgraph.types import ( ) from langgraph.utils.config import merge_configs, patch_config -GetNextVersion = Callable[[Optional[V]], V] +GetNextVersion = Callable[[Optional[V], None], V] SUPPORTS_EXC_NOTES = sys.version_info >= (3, 11) @@ -214,7 +214,7 @@ def local_read( return values -def increment(current: int | None) -> int: +def increment(current: int | None, channel: None) -> int: """Default channel versioning function, increments the current int version.""" return current + 1 if current is not None else 1 @@ -265,7 +265,8 @@ def apply_writes( next_version = get_next_version( max(checkpoint["channel_versions"].values()) if checkpoint["channel_versions"] - else None + else None, + None, ) # Consume all channels that were read diff --git a/libs/langgraph/langgraph/pregel/checkpoint.py b/libs/langgraph/langgraph/pregel/checkpoint.py index b8ca90db4..b404ee550 100644 --- a/libs/langgraph/langgraph/pregel/checkpoint.py +++ b/libs/langgraph/langgraph/pregel/checkpoint.py @@ -71,3 +71,14 @@ def channels_from_checkpoint( }, managed_specs, ) + + +def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint: + return Checkpoint( + v=checkpoint["v"], + ts=checkpoint["ts"], + id=checkpoint["id"], + channel_values=checkpoint["channel_values"].copy(), + channel_versions=checkpoint["channel_versions"].copy(), + versions_seen={k: v.copy() for k, v in checkpoint["versions_seen"].items()}, + ) diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index 07d4a97ad..5ff4771b3 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -29,7 +29,6 @@ from typing_extensions import ParamSpec, Self from langgraph.cache.base import BaseCache from langgraph.channels.base import BaseChannel -from langgraph.channels.last_value import LastValue from langgraph.checkpoint.base import ( EXCLUDED_METADATA_KEYS, WRITES_IDX_MAP, @@ -39,7 +38,6 @@ from langgraph.checkpoint.base import ( CheckpointMetadata, CheckpointTuple, PendingWrite, - copy_checkpoint, ) from langgraph.constants import ( CONF, @@ -86,6 +84,7 @@ from langgraph.pregel.algo import ( ) from langgraph.pregel.checkpoint import ( channels_from_checkpoint, + copy_checkpoint, create_checkpoint, empty_checkpoint, ) @@ -963,13 +962,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager): ) self.stack = ExitStack() if checkpointer: - if checkpointer._get_next_version_legacy: - empty_channel: LastValue[Any] = LastValue(Any) - self.checkpointer_get_next_version = ( - lambda c: checkpointer.get_next_version(c, empty_channel) # type: ignore[call-arg] - ) - else: - self.checkpointer_get_next_version = checkpointer.get_next_version + self.checkpointer_get_next_version = checkpointer.get_next_version self.checkpointer_put_writes = checkpointer.put_writes self.checkpointer_put_writes_accepts_task_path = ( signature(checkpointer.put_writes).parameters.get("task_path") @@ -1142,13 +1135,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager): ) self.stack = AsyncExitStack() if checkpointer: - if checkpointer._get_next_version_legacy: - empty_channel: LastValue[Any] = LastValue(Any) - self.checkpointer_get_next_version = ( - lambda c: checkpointer.get_next_version(c, empty_channel) # type: ignore[call-arg] - ) - else: - self.checkpointer_get_next_version = checkpointer.get_next_version + self.checkpointer_get_next_version = checkpointer.get_next_version self.checkpointer_put_writes = checkpointer.aput_writes self.checkpointer_put_writes_accepts_task_path = ( signature(checkpointer.aput_writes).parameters.get("task_path") diff --git a/libs/langgraph/tests/test_checkpoint_migration.py b/libs/langgraph/tests/test_checkpoint_migration.py index e284af3d5..85229c7a8 100644 --- a/libs/langgraph/tests/test_checkpoint_migration.py +++ b/libs/langgraph/tests/test_checkpoint_migration.py @@ -7,12 +7,9 @@ from typing import Annotated, Literal, Optional, Union import pytest from typing_extensions import TypedDict -from langgraph.checkpoint.base import ( - BaseCheckpointSaver, - CheckpointTuple, - copy_checkpoint, -) +from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointTuple from langgraph.graph.state import StateGraph +from langgraph.pregel.checkpoint import copy_checkpoint from langgraph.types import Command, Interrupt, PregelTask, StateSnapshot, interrupt from langgraph.utils.config import patch_configurable from tests.any_int import AnyInt diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 05561bc78..3d38558ab 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -159,7 +159,7 @@ def test_checkpoint_errors() -> None: raise ValueError("Faulty put_writes") class FaultyVersionCheckpointer(InMemorySaver): - def get_next_version(self, current: Optional[int]) -> int: + def get_next_version(self, current: Optional[int], channel: None) -> int: raise ValueError("Faulty get_next_version") def logic(inp: str) -> str: diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index f7934a9f5..ff5e8496d 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -103,7 +103,7 @@ async def test_checkpoint_errors() -> None: raise ValueError("Faulty put_writes") class FaultyVersionCheckpointer(InMemorySaver): - def get_next_version(self, current: Optional[int]) -> int: + def get_next_version(self, current: Optional[int], channel: None) -> int: raise ValueError("Faulty get_next_version") def logic(inp: str) -> str: diff --git a/libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py b/libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py index 9bf35a941..8f3149663 100644 --- a/libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py +++ b/libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py @@ -591,7 +591,7 @@ def create_react_agent( workflow = StateGraph(state_schema, config_schema=config_schema) workflow.add_node( "agent", - RunnableCallable(call_model, acall_model), # type: ignore[call-overload] + RunnableCallable(call_model, acall_model), input_schema=input_schema, ) if pre_model_hook is not None: @@ -610,7 +610,7 @@ def create_react_agent( if response_format is not None: workflow.add_node( "generate_structured_response", - RunnableCallable( # type: ignore[call-overload] + RunnableCallable( generate_structured_response, agenerate_structured_response, ), @@ -660,10 +660,10 @@ def create_react_agent( # Define the two nodes we will cycle between workflow.add_node( "agent", - RunnableCallable(call_model, acall_model), # type: ignore[call-overload] + RunnableCallable(call_model, acall_model), input_schema=input_schema, ) - workflow.add_node("tools", tool_node) # type: ignore[call-overload] + workflow.add_node("tools", tool_node) # Optionally add a pre-model hook node that will be called # every time before the "agent" (LLM-calling node) @@ -693,7 +693,7 @@ def create_react_agent( if response_format is not None: workflow.add_node( "generate_structured_response", - RunnableCallable( # type: ignore[call-overload] + RunnableCallable( generate_structured_response, agenerate_structured_response, ), diff --git a/libs/prebuilt/tests/memory_assert.py b/libs/prebuilt/tests/memory_assert.py index f88f0358f..10b93fdbd 100644 --- a/libs/prebuilt/tests/memory_assert.py +++ b/libs/prebuilt/tests/memory_assert.py @@ -13,9 +13,9 @@ from langgraph.checkpoint.base import ( CheckpointMetadata, CheckpointTuple, SerializerProtocol, - copy_checkpoint, ) from langgraph.checkpoint.memory import InMemorySaver, PersistentDict +from langgraph.pregel.checkpoint import copy_checkpoint class NoopSerializer(SerializerProtocol): From 63a00283729812207f0c893104702ea4fa747877 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 16 Jun 2025 14:58:50 -0700 Subject: [PATCH 09/13] langgraph-checkpoint 2.1.0 --- libs/checkpoint/pyproject.toml | 2 +- libs/checkpoint/uv.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/libs/checkpoint/pyproject.toml b/libs/checkpoint/pyproject.toml index c0b17a606..b7bc90bce 100644 --- a/libs/checkpoint/pyproject.toml +++ b/libs/checkpoint/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "langgraph-checkpoint" -version = "2.0.26" +version = "2.1.0" description = "Library with base interfaces for LangGraph checkpoint savers." authors = [] requires-python = ">=3.9" diff --git a/libs/checkpoint/uv.lock b/libs/checkpoint/uv.lock index 348b8083c..0fad33ab7 100644 --- a/libs/checkpoint/uv.lock +++ b/libs/checkpoint/uv.lock @@ -324,7 +324,7 @@ wheels = [ [[package]] name = "langgraph-checkpoint" -version = "2.0.26" +version = "2.1.0" source = { editable = "." } dependencies = [ { name = "langchain-core" }, From dfcaf97c732119e6f9ab4deae417c08a992c8208 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 16 Jun 2025 15:17:56 -0700 Subject: [PATCH 10/13] langgraph 0.5.0rc0 --- libs/langgraph/pyproject.toml | 4 ++-- libs/langgraph/uv.lock | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/libs/langgraph/pyproject.toml b/libs/langgraph/pyproject.toml index 04037001e..05d5985b4 100644 --- a/libs/langgraph/pyproject.toml +++ b/libs/langgraph/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "langgraph" -version = "0.4.7" +version = "0.5.0rc0" description = "Building stateful, multi-actor applications with LLMs" authors = [] requires-python = ">=3.9" @@ -13,7 +13,7 @@ license = "MIT" license-files = ['LICENSE'] dependencies = [ "langchain-core>=0.1", - "langgraph-checkpoint>=2.0.26", + "langgraph-checkpoint>=2.1.0", "langgraph-sdk>=0.1.42", "langgraph-prebuilt>=0.2.0", "xxhash>=3.5.0", diff --git a/libs/langgraph/uv.lock b/libs/langgraph/uv.lock index 7c170f6c8..5c14c0ef7 100644 --- a/libs/langgraph/uv.lock +++ b/libs/langgraph/uv.lock @@ -1201,7 +1201,7 @@ wheels = [ [[package]] name = "langgraph" -version = "0.4.7" +version = "0.5.0rc0" source = { editable = "." } dependencies = [ { name = "langchain-core" }, @@ -1310,7 +1310,7 @@ wheels = [ [[package]] name = "langgraph-checkpoint" -version = "2.0.26" +version = "2.1.0" source = { editable = "../checkpoint" } dependencies = [ { name = "langchain-core" }, From edfb65fd3ac41a3ea9ab6ddcf1438175b0195093 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 16 Jun 2025 17:47:21 -0700 Subject: [PATCH 11/13] langgraph-prebuilt 0.5.0rc0 --- libs/prebuilt/pyproject.toml | 4 ++-- libs/prebuilt/uv.lock | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/libs/prebuilt/pyproject.toml b/libs/prebuilt/pyproject.toml index 2cb75c62b..cf7e3a244 100644 --- a/libs/prebuilt/pyproject.toml +++ b/libs/prebuilt/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "langgraph-prebuilt" -version = "0.2.2" +version = "0.5.0rc0" description = "Library with high-level APIs for creating and executing LangGraph agents and tools." authors = [] requires-python = ">=3.9" @@ -12,7 +12,7 @@ readme = "README.md" license = "MIT" license-files = ['LICENSE'] dependencies = [ - "langgraph-checkpoint>=2.0.10", + "langgraph-checkpoint>=2.1.0", "langchain-core>=0.3.22", ] diff --git a/libs/prebuilt/uv.lock b/libs/prebuilt/uv.lock index d85d7c934..5fde75d82 100644 --- a/libs/prebuilt/uv.lock +++ b/libs/prebuilt/uv.lock @@ -320,7 +320,7 @@ wheels = [ [[package]] name = "langgraph" -version = "0.4.7" +version = "0.5.0rc0" source = { editable = "../langgraph" } dependencies = [ { name = "langchain-core" }, @@ -371,7 +371,7 @@ dev = [ [[package]] name = "langgraph-checkpoint" -version = "2.0.26" +version = "2.1.0" source = { editable = "../checkpoint" } dependencies = [ { name = "langchain-core" }, @@ -464,7 +464,7 @@ dev = [ [[package]] name = "langgraph-prebuilt" -version = "0.2.2" +version = "0.5.0rc0" source = { editable = "." } dependencies = [ { name = "langchain-core" }, From 771c6150a4f039c69903a86c8201de7127f37006 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 16 Jun 2025 17:52:13 -0700 Subject: [PATCH 12/13] langgraph 0.5.0rc1 --- libs/langgraph/pyproject.toml | 4 ++-- libs/langgraph/uv.lock | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/libs/langgraph/pyproject.toml b/libs/langgraph/pyproject.toml index 05d5985b4..98104a931 100644 --- a/libs/langgraph/pyproject.toml +++ b/libs/langgraph/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "langgraph" -version = "0.5.0rc0" +version = "0.5.0rc1" description = "Building stateful, multi-actor applications with LLMs" authors = [] requires-python = ">=3.9" @@ -15,7 +15,7 @@ dependencies = [ "langchain-core>=0.1", "langgraph-checkpoint>=2.1.0", "langgraph-sdk>=0.1.42", - "langgraph-prebuilt>=0.2.0", + "langgraph-prebuilt>=0.5.0rc0", "xxhash>=3.5.0", "pydantic>=2.7.4", ] diff --git a/libs/langgraph/uv.lock b/libs/langgraph/uv.lock index 5c14c0ef7..3df4c7f09 100644 --- a/libs/langgraph/uv.lock +++ b/libs/langgraph/uv.lock @@ -1201,7 +1201,7 @@ wheels = [ [[package]] name = "langgraph" -version = "0.5.0rc0" +version = "0.5.0rc1" source = { editable = "." } dependencies = [ { name = "langchain-core" }, @@ -1423,7 +1423,7 @@ inmem = [ [[package]] name = "langgraph-prebuilt" -version = "0.2.2" +version = "0.5.0rc0" source = { editable = "../prebuilt" } dependencies = [ { name = "langchain-core" }, From 1309243b29d7fcdce8d62f6d4b4304d1ded5d157 Mon Sep 17 00:00:00 2001 From: lc-arjun Date: Tue, 17 Jun 2025 12:32:04 -0700 Subject: [PATCH 13/13] docs: studio evals (#5129) * docs: studio evals * docs: added studio evals images (#5076) * docs: added studio evals images * Update docs/docs/cloud/how-tos/studio/run_evals.md Co-authored-by: lc-arjun * Update docs/docs/cloud/how-tos/studio/run_evals.md Co-authored-by: lc-arjun * Update docs/docs/cloud/how-tos/studio/run_evals.md Co-authored-by: lc-arjun * docs: updated studio evals * Update docs/docs/cloud/how-tos/studio/run_evals.md Co-authored-by: lc-arjun * docs: removed images --------- Co-authored-by: lc-arjun * final changes * i think its this --------- Co-authored-by: Marco Perini --- docs/docs/cloud/how-tos/studio/run_evals.md | 57 +++++++++++++++++++++ docs/docs/concepts/langgraph_studio.md | 3 +- docs/mkdocs.yml | 1 + 3 files changed, 60 insertions(+), 1 deletion(-) create mode 100644 docs/docs/cloud/how-tos/studio/run_evals.md diff --git a/docs/docs/cloud/how-tos/studio/run_evals.md b/docs/docs/cloud/how-tos/studio/run_evals.md new file mode 100644 index 000000000..d7a367cd8 --- /dev/null +++ b/docs/docs/cloud/how-tos/studio/run_evals.md @@ -0,0 +1,57 @@ +# Run experiments over a dataset + +LangGraph Studio supports evaluations by allowing you to run your assistant over a pre-defined LangSmith dataset. This enables you to understand how your application performs over a variety of inputs, compare the results to reference outputs, and score the results using [evaluators](../../../agents/evals.md). + +This guide shows you how to run an experiment end-to-end from Studio. + +--- + +## Prerequisites + +Before running an experiment, ensure you have the following: + +1. **A LangSmith dataset**: Your dataset should contain the inputs you want to test and optionally, reference outputs for comparison. + + - The schema for the inputs must match the required input schema for the assistant. For more information on schemas, see [here](../../../concepts/low_level.md#schema). + - For more on creating datasets, see [How to Manage Datasets](https://docs.smith.langchain.com/evaluation/how_to_guides/manage_datasets_in_application#set-up-your-dataset). + +2. **(Optional) Evaluators**: You can attach evaluators (e.g., LLM-as-a-Judge, heuristics, or custom functions) to your dataset in LangSmith. These will run automatically after the graph has processed all inputs. + + - To learn more, read about [Evaluation Concepts](https://docs.smith.langchain.com/evaluation/concepts#evaluators). + +3. **A running application**: The experiment can be run against: + - An application deployed on [LangGraph Platform](../../quick_start.md). + - A locally running application started via the [langgraph-cli](../../../tutorials/langgraph-platform/local-server.md). + +--- + +## Step-by-step guide + +### 1. Launch the experiment + +Click the **Run experiment** button in the top right corner of the Studio page. + +### 2. Select your dataset + +In the modal that appears, select the dataset (or a specific dataset split) to use for the experiment and click **Start**. + +### 3. Monitor the progress + +All of the inputs in the dataset will now be run against the active assistant. Monitor the experiment's progress via the badge in the top right corner. + +You can continue to work in Studio while the experiment runs in the background. Click the arrow icon button at any time to navigate to LangSmith and view the detailed experiment results. + +--- + +## Troubleshooting + +### "Run experiment" button is disabled + +If the "Run experiment" button is disabled, check the following: + +- **Deployed application**: If your application is deployed on LangGraph Platform, you may need to create a new revision to enable this feature. +- **Local development server**: If you are running your application locally, make sure you have upgraded to the latest version of the `langgraph-cli` (`pip install -U langgraph-cli`). Additionally, ensure you have tracing enabled by setting the `LANGSMITH_API_KEY` in your project's `.env` file. + +### Evaluator results are missing + +When you run an experiment, any attached evaluators are scheduled for execution in a queue. If you don't see results immediately, it likely means they are still pending. diff --git a/docs/docs/concepts/langgraph_studio.md b/docs/docs/concepts/langgraph_studio.md index cdd8ef00c..6011de041 100644 --- a/docs/docs/concepts/langgraph_studio.md +++ b/docs/docs/concepts/langgraph_studio.md @@ -24,6 +24,7 @@ Key features of LangGraph Studio: - [Manage assistants](../cloud/how-tos/studio/manage_assistants.md) - [Manage threads](../cloud/how-tos/threads_studio.md) - [Iterate on prompts](../cloud/how-tos/iterate_graph_studio.md) +- [Run experiments over a dataset](../cloud/how-tos/studio/run_evals.md) - Manage [long term memory](memory.md) - Debug agent state via [time travel](time-travel.md) @@ -41,4 +42,4 @@ Chat mode is a simpler UI for iterating on and testing chat-specific agents. It ## Learn more -- See this guide on how to [get started](../cloud/how-tos/studio/quick_start.md) with LangGraph Studio. \ No newline at end of file +- See this guide on how to [get started](../cloud/how-tos/studio/quick_start.md) with LangGraph Studio. diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index a1b53f559..b1f932409 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -179,6 +179,7 @@ nav: - cloud/how-tos/studio/manage_assistants.md - cloud/how-tos/threads_studio.md - cloud/how-tos/iterate_graph_studio.md + - cloud/how-tos/studio/run_evals.md - cloud/how-tos/clone_traces_studio.md - cloud/how-tos/datasets_studio.md - LangGraph SDK: concepts/sdk.md