ANN support testing

docs/docs/tutorials/langgraph-platform/local-server.md

@@ -35,10 +35,10 @@ Create a new app from the `react-agent` template. This template is a simple agen
 
 ## Install Dependencies
 
-In the root of your new LangGraph app, install the dependencies in `edit` mode so your local changes are used by the server:
+In the root of your new LangGraph app, install the dependencies:
 
 ```shell
-pip install -e .
+pip install .
 ```
 
 ## Create a `.env` file

libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

@@ -5,6 +5,7 @@ from typing import Any, Optional
 
 from langchain_core.runnables import RunnableConfig
 from psycopg import Capabilities, Connection, Cursor, Pipeline
+from psycopg.errors import UndefinedTable
 from psycopg.rows import DictRow, dict_row
 from psycopg.types.json import Jsonb
 from psycopg_pool import ConnectionPool
@@ -75,15 +76,16 @@ class PostgresSaver(BasePostgresSaver):
         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:
+            try:
+                row = cur.execute(
+                    "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
+                ).fetchone()
+                if row is None:
+                    version = -1
+                else:
+                    version = row["v"]
+            except UndefinedTable:
                 version = -1
-            else:
-                version = row["v"]
             for v, migration in zip(
                 range(version + 1, len(self.MIGRATIONS)),
                 self.MIGRATIONS[version + 1 :],

libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

@@ -5,6 +5,7 @@ from typing import Any, Optional
 
 from langchain_core.runnables import RunnableConfig
 from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
+from psycopg.errors import UndefinedTable
 from psycopg.rows import DictRow, dict_row
 from psycopg.types.json import Jsonb
 from psycopg_pool import AsyncConnectionPool
@@ -80,15 +81,17 @@ class AsyncPostgresSaver(BasePostgresSaver):
         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:
+            try:
+                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"]
+            except UndefinedTable:
                 version = -1
-            else:
-                version = row["v"]
             for v, migration in zip(
                 range(version + 1, len(self.MIGRATIONS)),
                 self.MIGRATIONS[version + 1 :],

libs/checkpoint-postgres/langgraph/store/postgres/base.py

@@ -56,10 +56,9 @@ class Migration(NamedTuple):
 
     sql: str
     params: Optional[dict[str, Any]] = None
-    condition: Optional[Callable[["BasePostgresStore"], bool]] = None
 
 
-MIGRATIONS: Sequence[str] = [
+MIGRATIONS: Sequence[Union[str, Migration]] = [
     """
 CREATE TABLE IF NOT EXISTS store (
     -- 'prefix' represents the doc's 'namespace'
@@ -110,9 +109,6 @@ CREATE TABLE IF NOT EXISTS store_vectors (
 CREATE INDEX IF NOT EXISTS store_vectors_embedding_idx ON store_vectors 
     USING %(index_type)s (embedding %(ops)s)%(index_params)s;
 """,
-        condition=lambda store: bool(
-            store.index_config and _get_index_params(store)[0] != "flat"
-        ),
         params={
             "index_type": lambda store: _get_index_params(store)[0],
             "ops": lambda store: _get_vector_type_ops(store),
@@ -159,7 +155,7 @@ class PoolConfig(TypedDict, total=False):
 class ANNIndexConfig(TypedDict, total=False):
     """Configuration for vector index in PostgreSQL store."""
 
-    kind: Literal["hnsw", "ivfflat", "flat"]
+    kind: Literal["hnsw", "ivfflat"]
     """Type of index to use: 'hnsw' for Hierarchical Navigable Small World, or 'ivfflat' for Inverted File Flat."""
     vector_type: Literal["vector", "halfvec"]
     """Type of vector storage to use.
@@ -394,18 +390,17 @@ class BasePostgresStore(Generic[C]):
 
                 vectors_per_doc_estimate = self.index_config["__estimated_num_vectors"]
                 expanded_limit = (op.limit * vectors_per_doc_estimate * 2) + 1
-
                 # Vector search with CTE for proper score handling
                 filter_str = (
                     ""
                     if not filter_conditions
                     else " AND " + " AND ".join(filter_conditions)
                 )
+                ns_args = []
                 if op.namespace_prefix:
-                    prefix_filter_str = f"WHERE s.prefix LIKE %s {filter_str} "
-                    ns_args: Sequence = (f"{_namespace_to_text(op.namespace_prefix)}%",)
+                    prefix_filter_str = f"WHERE s.prefix = %s {filter_str} "
+                    ns_args = [f"{_namespace_to_text(op.namespace_prefix)}"]
                 else:
-                    ns_args = ()
                     if filter_str:
                         prefix_filter_str = f"WHERE {filter_str} "
                     else:
@@ -762,6 +757,20 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
                         _paramslist[i] = embedding
 
         for (idx, _), (query, params) in zip(search_ops, queries):
+            # Get and print pgvector version
+            cur.execute("SELECT extversion FROM pg_extension WHERE extname = 'vector'")
+            version = cur.fetchone()
+            if version:
+                print(f"pgvector version: {list(version.values())[0]}", flush=True)
+
+            # Run EXPLAIN on the query, verbose to get the query plan
+            cur.execute(f"EXPLAIN {query}", params)
+            # Print the query plan line by line. Truncate at 300 chars per line
+            print("^" * 80, flush=True)
+            for line in cur.fetchall():
+                print(list(line.values())[0][:300], flush=True)
+            print("*" * 80, flush=True)
+            # Execute the actual query
             cur.execute(query, params)
             rows = cast(list[Row], cur.fetchall())
             results[idx] = [
@@ -824,8 +833,6 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
                 for v, migration in enumerate(
                     self.VECTOR_MIGRATIONS[version + 1 :], start=version + 1
                 ):
-                    if migration.condition and not migration.condition(self):
-                        continue
                     sql = migration.sql
                     if migration.params:
                         params = {
@@ -1016,15 +1023,14 @@ def _get_distance_operator(store: Any) -> tuple[str, str]:
     # a DESCENDING ORDER sort clause and the user's expectations of what the similarity score
     # should be.
     if distance_type == "l2":
-        # Final: "-(sv.embedding <-> %s::%s)"
-        # We return the "l2 similarity" so that the sorting order is the same
-        return "sv.embedding <-> %s::%s", "-scored.neg_score"
+        # Final: "1 - (sv.embedding <-> %s::%s)"
+        return "sv.embedding <-> %s::%s", "1 - (scored.neg_score)"
     elif distance_type == "inner_product":
         # Final: "-(sv.embedding <#> %s::%s)"
         return "sv.embedding <#> %s::%s", "-(scored.neg_score)"
-    else:  # cosine similarity
+    else:  # cosine
         # Final:  "1 - (sv.embedding <=> %s::%s)"
-        return "sv.embedding <=> %s::%s", "1 - scored.neg_score"
+        return "sv.embedding <=> %s::%s", "1 - (scored.neg_score)"
 
 
 def _ensure_index_config(

libs/checkpoint-postgres/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph-checkpoint-postgres"
-version = "2.0.7"
+version = "2.0.6"
 description = "Library with a Postgres implementation of LangGraph checkpoint saver."
 authors = []
 license = "MIT"

libs/checkpoint-postgres/tests/conftest.py

@@ -7,7 +7,6 @@ from psycopg.rows import DictRow, dict_row
 
 from tests.embed_test_utils import CharacterEmbeddings
 
-DEFAULT_POSTGRES_URI = "postgres://postgres:postgres@localhost:5441/"
 DEFAULT_URI = "postgres://postgres:postgres@localhost:5441/postgres?sslmode=disable"
 
 

libs/checkpoint-postgres/tests/test_async.py

@@ -1,14 +1,7 @@
-# type: ignore
-
-from contextlib import asynccontextmanager
 from typing import Any
-from uuid import uuid4
 
 import pytest
 from langchain_core.runnables import RunnableConfig
-from psycopg import AsyncConnection
-from psycopg.rows import dict_row
-from psycopg_pool import AsyncConnectionPool
 
 from langgraph.checkpoint.base import (
     Checkpoint,
@@ -17,212 +10,104 @@ from langgraph.checkpoint.base import (
     empty_checkpoint,
 )
 from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
-from tests.conftest import DEFAULT_POSTGRES_URI
+from tests.conftest import DEFAULT_URI
 
 
-@asynccontextmanager
-async def _pool_saver():
-    """Fixture for pool 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:
-        # yield checkpointer
-        async with AsyncConnectionPool(
-            DEFAULT_POSTGRES_URI + database,
-            max_size=10,
-            kwargs={"autocommit": True, "row_factory": dict_row},
-        ) as pool:
-            checkpointer = AsyncPostgresSaver(pool)
-            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 _pipe_saver():
-    """Fixture for pipeline 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:
-            async with conn.pipeline() as pipe:
-                checkpointer = AsyncPostgresSaver(conn, pipe=pipe)
-                await checkpointer.setup()
-            async with conn.pipeline() as pipe:
-                checkpointer = AsyncPostgresSaver(conn, pipe=pipe)
-                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 _base_saver():
-    """Fixture for regular 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 = AsyncPostgresSaver(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 == "pool":
-        async with _pool_saver() as saver:
-            yield saver
-    elif name == "pipe":
-        async with _pipe_saver() as saver:
-            yield saver
-
-
-@pytest.fixture
-def test_data():
-    """Fixture providing test data for checkpoint tests."""
-    config_1: RunnableConfig = {
-        "configurable": {
-            "thread_id": "thread-1",
-            # for backwards compatibility testing
-            "thread_ts": "1",
-            "checkpoint_ns": "",
+class TestAsyncPostgresSaver:
+    @pytest.fixture(autouse=True)
+    async def setup(self) -> None:
+        # objects for test setup
+        self.config_1: RunnableConfig = {
+            "configurable": {
+                "thread_id": "thread-1",
+                # for backwards compatibility testing
+                "thread_ts": "1",
+                "checkpoint_ns": "",
+            }
         }
-    }
-    config_2: RunnableConfig = {
-        "configurable": {
-            "thread_id": "thread-2",
-            "checkpoint_id": "2",
-            "checkpoint_ns": "",
+        self.config_2: RunnableConfig = {
+            "configurable": {
+                "thread_id": "thread-2",
+                "checkpoint_id": "2",
+                "checkpoint_ns": "",
+            }
         }
-    }
-    config_3: RunnableConfig = {
-        "configurable": {
-            "thread_id": "thread-2",
-            "checkpoint_id": "2-inner",
-            "checkpoint_ns": "inner",
+        self.config_3: RunnableConfig = {
+            "configurable": {
+                "thread_id": "thread-2",
+                "checkpoint_id": "2-inner",
+                "checkpoint_ns": "inner",
+            }
         }
-    }
 
-    chkpnt_1: Checkpoint = empty_checkpoint()
-    chkpnt_2: Checkpoint = create_checkpoint(chkpnt_1, {}, 1)
-    chkpnt_3: Checkpoint = empty_checkpoint()
+        self.chkpnt_1: Checkpoint = empty_checkpoint()
+        self.chkpnt_2: Checkpoint = create_checkpoint(self.chkpnt_1, {}, 1)
+        self.chkpnt_3: Checkpoint = empty_checkpoint()
 
-    metadata_1: CheckpointMetadata = {
-        "source": "input",
-        "step": 2,
-        "writes": {},
-        "score": 1,
-    }
-    metadata_2: CheckpointMetadata = {
-        "source": "loop",
-        "step": 1,
-        "writes": {"foo": "bar"},
-        "score": None,
-    }
-    metadata_3: CheckpointMetadata = {}
-
-    return {
-        "configs": [config_1, config_2, config_3],
-        "checkpoints": [chkpnt_1, chkpnt_2, chkpnt_3],
-        "metadata": [metadata_1, metadata_2, metadata_3],
-    }
-
-
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
-async def test_asearch(request, saver_name: str, test_data) -> None:
-    async with _saver(saver_name) as saver:
-        configs = test_data["configs"]
-        checkpoints = test_data["checkpoints"]
-        metadata = test_data["metadata"]
-
-        await saver.aput(configs[0], checkpoints[0], metadata[0], {})
-        await saver.aput(configs[1], checkpoints[1], metadata[1], {})
-        await saver.aput(configs[2], checkpoints[2], metadata[2], {})
-
-        # call method / assertions
-        query_1 = {"source": "input"}  # search by 1 key
-        query_2 = {
+        self.metadata_1: CheckpointMetadata = {
+            "source": "input",
+            "step": 2,
+            "writes": {},
+            "score": 1,
+        }
+        self.metadata_2: CheckpointMetadata = {
+            "source": "loop",
             "step": 1,
             "writes": {"foo": "bar"},
-        }  # search by multiple keys
-        query_3: dict[str, Any] = {}  # search by no keys, return all checkpoints
-        query_4 = {"source": "update", "step": 1}  # no match
+            "score": None,
+        }
+        self.metadata_3: CheckpointMetadata = {}
+        async with AsyncPostgresSaver.from_conn_string(DEFAULT_URI) as saver:
+            await saver.setup()
 
-        search_results_1 = [c async for c in saver.alist(None, filter=query_1)]
-        assert len(search_results_1) == 1
-        assert search_results_1[0].metadata == metadata[0]
+    async def test_asearch(self) -> None:
+        async with AsyncPostgresSaver.from_conn_string(DEFAULT_URI) as saver:
+            await saver.aput(self.config_1, self.chkpnt_1, self.metadata_1, {})
+            await saver.aput(self.config_2, self.chkpnt_2, self.metadata_2, {})
+            await saver.aput(self.config_3, self.chkpnt_3, self.metadata_3, {})
 
-        search_results_2 = [c async for c in saver.alist(None, filter=query_2)]
-        assert len(search_results_2) == 1
-        assert search_results_2[0].metadata == metadata[1]
+            # call method / assertions
+            query_1 = {"source": "input"}  # search by 1 key
+            query_2 = {
+                "step": 1,
+                "writes": {"foo": "bar"},
+            }  # search by multiple keys
+            query_3: dict[str, Any] = {}  # search by no keys, return all checkpoints
+            query_4 = {"source": "update", "step": 1}  # no match
 
-        search_results_3 = [c async for c in saver.alist(None, filter=query_3)]
-        assert len(search_results_3) == 3
+            search_results_1 = [c async for c in saver.alist(None, filter=query_1)]
+            assert len(search_results_1) == 1
+            assert search_results_1[0].metadata == self.metadata_1
 
-        search_results_4 = [c async for c in saver.alist(None, filter=query_4)]
-        assert len(search_results_4) == 0
+            search_results_2 = [c async for c in saver.alist(None, filter=query_2)]
+            assert len(search_results_2) == 1
+            assert search_results_2[0].metadata == self.metadata_2
 
-        # search by config (defaults to checkpoints across all namespaces)
-        search_results_5 = [
-            c async for c in saver.alist({"configurable": {"thread_id": "thread-2"}})
-        ]
-        assert len(search_results_5) == 2
-        assert {
-            search_results_5[0].config["configurable"]["checkpoint_ns"],
-            search_results_5[1].config["configurable"]["checkpoint_ns"],
-        } == {"", "inner"}
+            search_results_3 = [c async for c in saver.alist(None, filter=query_3)]
+            assert len(search_results_3) == 3
 
+            search_results_4 = [c async for c in saver.alist(None, filter=query_4)]
+            assert len(search_results_4) == 0
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
-async def test_null_chars(request, saver_name: str, test_data) -> None:
-    async with _saver(saver_name) as saver:
-        config = await saver.aput(
-            test_data["configs"][0],
-            test_data["checkpoints"][0],
-            {"my_key": "\x00abc"},
-            {},
-        )
-        assert (await saver.aget_tuple(config)).metadata["my_key"] == "abc"  # type: ignore
-        assert [c async for c in saver.alist(None, filter={"my_key": "abc"})][
-            0
-        ].metadata["my_key"] == "abc"
+            # search by config (defaults to checkpoints across all namespaces)
+            search_results_5 = [
+                c
+                async for c in saver.alist({"configurable": {"thread_id": "thread-2"}})
+            ]
+            assert len(search_results_5) == 2
+            assert {
+                search_results_5[0].config["configurable"]["checkpoint_ns"],
+                search_results_5[1].config["configurable"]["checkpoint_ns"],
+            } == {"", "inner"}
+
+            # TODO: test before and limit params
+
+    async def test_null_chars(self) -> None:
+        async with AsyncPostgresSaver.from_conn_string(DEFAULT_URI) as saver:
+            config = await saver.aput(
+                self.config_1, self.chkpnt_1, {"my_key": "\x00abc"}, {}
+            )
+            assert (await saver.aget_tuple(config)).metadata["my_key"] == "abc"  # type: ignore
+            assert [c async for c in saver.alist(None, filter={"my_key": "abc"})][
+                0
+            ].metadata["my_key"] == "abc"

libs/checkpoint-postgres/tests/test_store.py

@@ -634,7 +634,6 @@ def test_embed_with_path_operation_config(
     distance_type: str,
 ) -> None:
     """Test operation-level field configuration for vector search."""
-
     with _create_vector_store(
         vector_type,
         distance_type,
@@ -696,89 +695,3 @@ def test_embed_with_path_operation_config(
         # assert len(results) == 3
         # doc5_result = next(r for r in results if r.key == "doc5")
         # assert doc5_result.score is None
-
-
-def _cosine_similarity(X: list[float], Y: list[list[float]]) -> list[float]:
-    """
-    Compute cosine similarity between a vector X and a matrix Y.
-    Lazy import numpy for efficiency.
-    """
-
-    similarities = []
-    for y in Y:
-        dot_product = sum(a * b for a, b in zip(X, y))
-        norm1 = sum(a * a for a in X) ** 0.5
-        norm2 = sum(a * a for a in y) ** 0.5
-        similarity = dot_product / (norm1 * norm2) if norm1 > 0 and norm2 > 0 else 0.0
-        similarities.append(similarity)
-
-    return similarities
-
-
-def _inner_product(X: list[float], Y: list[list[float]]) -> list[float]:
-    """
-    Compute inner product between a vector X and a matrix Y.
-    Lazy import numpy for efficiency.
-    """
-
-    similarities = []
-    for y in Y:
-        similarity = sum(a * b for a, b in zip(X, y))
-        similarities.append(similarity)
-
-    return similarities
-
-
-def _neg_l2_distance(X: list[float], Y: list[list[float]]) -> list[float]:
-    """
-    Compute l2 distance between a vector X and a matrix Y.
-    Lazy import numpy for efficiency.
-    """
-
-    similarities = []
-    for y in Y:
-        similarity = sum((a - b) ** 2 for a, b in zip(X, y)) ** 0.5
-        similarities.append(-similarity)
-
-    return similarities
-
-
-@pytest.mark.parametrize(
-    "vector_type,distance_type",
-    [
-        ("vector", "cosine"),
-        ("vector", "inner_product"),
-        ("halfvec", "l2"),
-    ],
-)
-@pytest.mark.parametrize("query", ["aaa", "bbb", "ccc", "abcd", "poisson"])
-def test_scores(
-    fake_embeddings: CharacterEmbeddings,
-    vector_type: str,
-    distance_type: str,
-    query: str,
-) -> None:
-    """Test operation-level field configuration for vector search."""
-    with _create_vector_store(
-        vector_type,
-        distance_type,
-        fake_embeddings,
-        text_fields=["key0"],
-    ) as store:
-        doc = {
-            "key0": "aaa",
-        }
-        store.put(("test",), "doc", doc, index=["key0", "key1"])
-
-        results = store.search((), query=query)
-        vec0 = fake_embeddings.embed_query(doc["key0"])
-        vec1 = fake_embeddings.embed_query(query)
-        if distance_type == "cosine":
-            similarities = _cosine_similarity(vec1, [vec0])
-        elif distance_type == "inner_product":
-            similarities = _inner_product(vec1, [vec0])
-        elif distance_type == "l2":
-            similarities = _neg_l2_distance(vec1, [vec0])
-
-        assert len(results) == 1
-        assert results[0].score == pytest.approx(similarities[0], abs=1e-3)

libs/checkpoint-postgres/tests/test_sync.py

@@ -1,14 +1,7 @@
-# type: ignore
-
-from contextlib import contextmanager
 from typing import Any
-from uuid import uuid4
 
 import pytest
 from langchain_core.runnables import RunnableConfig
-from psycopg import Connection
-from psycopg.rows import dict_row
-from psycopg_pool import ConnectionPool
 
 from langgraph.checkpoint.base import (
     Checkpoint,
@@ -17,199 +10,103 @@ from langgraph.checkpoint.base import (
     empty_checkpoint,
 )
 from langgraph.checkpoint.postgres import PostgresSaver
-from tests.conftest import DEFAULT_POSTGRES_URI
+from tests.conftest import DEFAULT_URI
 
 
-@contextmanager
-def _pool_saver():
-    """Fixture for pool mode testing."""
-    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:
-        # yield checkpointer
-        with ConnectionPool(
-            DEFAULT_POSTGRES_URI + database,
-            max_size=10,
-            kwargs={"autocommit": True, "row_factory": dict_row},
-        ) as pool:
-            checkpointer = PostgresSaver(pool)
-            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 _pipe_saver():
-    """Fixture for pipeline mode testing."""
-    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:
-            with conn.pipeline() as pipe:
-                checkpointer = PostgresSaver(conn, pipe=pipe)
-                checkpointer.setup()
-            with conn.pipeline() as pipe:
-                checkpointer = PostgresSaver(conn, pipe=pipe)
-                yield checkpointer
-    finally:
-        # drop unique db
-        with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
-            conn.execute(f"DROP DATABASE {database}")
-
-
-@contextmanager
-def _base_saver():
-    """Fixture for regular connection mode testing."""
-    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 = PostgresSaver(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 == "pool":
-        with _pool_saver() as saver:
-            yield saver
-    elif name == "pipe":
-        with _pipe_saver() as saver:
-            yield saver
-
-
-@pytest.fixture
-def test_data():
-    """Fixture providing test data for checkpoint tests."""
-    config_1: RunnableConfig = {
-        "configurable": {
-            "thread_id": "thread-1",
-            # for backwards compatibility testing
-            "thread_ts": "1",
-            "checkpoint_ns": "",
+class TestPostgresSaver:
+    @pytest.fixture(autouse=True)
+    def setup(self) -> None:
+        # objects for test setup
+        self.config_1: RunnableConfig = {
+            "configurable": {
+                "thread_id": "thread-1",
+                # for backwards compatibility testing
+                "thread_ts": "1",
+                "checkpoint_ns": "",
+            }
         }
-    }
-    config_2: RunnableConfig = {
-        "configurable": {
-            "thread_id": "thread-2",
-            "checkpoint_id": "2",
-            "checkpoint_ns": "",
+        self.config_2: RunnableConfig = {
+            "configurable": {
+                "thread_id": "thread-2",
+                "checkpoint_id": "2",
+                "checkpoint_ns": "",
+            }
         }
-    }
-    config_3: RunnableConfig = {
-        "configurable": {
-            "thread_id": "thread-2",
-            "checkpoint_id": "2-inner",
-            "checkpoint_ns": "inner",
+        self.config_3: RunnableConfig = {
+            "configurable": {
+                "thread_id": "thread-2",
+                "checkpoint_id": "2-inner",
+                "checkpoint_ns": "inner",
+            }
         }
-    }
 
-    chkpnt_1: Checkpoint = empty_checkpoint()
-    chkpnt_2: Checkpoint = create_checkpoint(chkpnt_1, {}, 1)
-    chkpnt_3: Checkpoint = empty_checkpoint()
+        self.chkpnt_1: Checkpoint = empty_checkpoint()
+        self.chkpnt_2: Checkpoint = create_checkpoint(self.chkpnt_1, {}, 1)
+        self.chkpnt_3: Checkpoint = empty_checkpoint()
 
-    metadata_1: CheckpointMetadata = {
-        "source": "input",
-        "step": 2,
-        "writes": {},
-        "score": 1,
-    }
-    metadata_2: CheckpointMetadata = {
-        "source": "loop",
-        "step": 1,
-        "writes": {"foo": "bar"},
-        "score": None,
-    }
-    metadata_3: CheckpointMetadata = {}
-
-    return {
-        "configs": [config_1, config_2, config_3],
-        "checkpoints": [chkpnt_1, chkpnt_2, chkpnt_3],
-        "metadata": [metadata_1, metadata_2, metadata_3],
-    }
-
-
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
-def test_search(saver_name: str, test_data) -> None:
-    with _saver(saver_name) as saver:
-        configs = test_data["configs"]
-        checkpoints = test_data["checkpoints"]
-        metadata = test_data["metadata"]
-
-        saver.put(configs[0], checkpoints[0], metadata[0], {})
-        saver.put(configs[1], checkpoints[1], metadata[1], {})
-        saver.put(configs[2], checkpoints[2], metadata[2], {})
-
-        # call method / assertions
-        query_1 = {"source": "input"}  # search by 1 key
-        query_2 = {
+        self.metadata_1: CheckpointMetadata = {
+            "source": "input",
+            "step": 2,
+            "writes": {},
+            "score": 1,
+        }
+        self.metadata_2: CheckpointMetadata = {
+            "source": "loop",
             "step": 1,
             "writes": {"foo": "bar"},
-        }  # search by multiple keys
-        query_3: dict[str, Any] = {}  # search by no keys, return all checkpoints
-        query_4 = {"source": "update", "step": 1}  # no match
+            "score": None,
+        }
+        self.metadata_3: CheckpointMetadata = {}
+        with PostgresSaver.from_conn_string(DEFAULT_URI) as saver:
+            saver.setup()
 
-        search_results_1 = list(saver.list(None, filter=query_1))
-        assert len(search_results_1) == 1
-        assert search_results_1[0].metadata == metadata[0]
+    def test_search(self) -> None:
+        with PostgresSaver.from_conn_string(DEFAULT_URI) as saver:
+            # save checkpoints
+            saver.put(self.config_1, self.chkpnt_1, self.metadata_1, {})
+            saver.put(self.config_2, self.chkpnt_2, self.metadata_2, {})
+            saver.put(self.config_3, self.chkpnt_3, self.metadata_3, {})
 
-        search_results_2 = list(saver.list(None, filter=query_2))
-        assert len(search_results_2) == 1
-        assert search_results_2[0].metadata == metadata[1]
+            # call method / assertions
+            query_1 = {"source": "input"}  # search by 1 key
+            query_2 = {
+                "step": 1,
+                "writes": {"foo": "bar"},
+            }  # search by multiple keys
+            query_3: dict[str, Any] = {}  # search by no keys, return all checkpoints
+            query_4 = {"source": "update", "step": 1}  # no match
 
-        search_results_3 = list(saver.list(None, filter=query_3))
-        assert len(search_results_3) == 3
+            search_results_1 = list(saver.list(None, filter=query_1))
+            assert len(search_results_1) == 1
+            assert search_results_1[0].metadata == self.metadata_1
 
-        search_results_4 = list(saver.list(None, filter=query_4))
-        assert len(search_results_4) == 0
+            search_results_2 = list(saver.list(None, filter=query_2))
+            assert len(search_results_2) == 1
+            assert search_results_2[0].metadata == self.metadata_2
 
-        # search by config (defaults to checkpoints across all namespaces)
-        search_results_5 = list(saver.list({"configurable": {"thread_id": "thread-2"}}))
-        assert len(search_results_5) == 2
-        assert {
-            search_results_5[0].config["configurable"]["checkpoint_ns"],
-            search_results_5[1].config["configurable"]["checkpoint_ns"],
-        } == {"", "inner"}
+            search_results_3 = list(saver.list(None, filter=query_3))
+            assert len(search_results_3) == 3
 
+            search_results_4 = list(saver.list(None, filter=query_4))
+            assert len(search_results_4) == 0
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
-def test_null_chars(saver_name: str, test_data) -> None:
-    with _saver(saver_name) as saver:
-        config = saver.put(
-            test_data["configs"][0],
-            test_data["checkpoints"][0],
-            {"my_key": "\x00abc"},
-            {},
-        )
-        assert saver.get_tuple(config).metadata["my_key"] == "abc"  # type: ignore
-        assert (
-            list(saver.list(None, filter={"my_key": "abc"}))[0].metadata["my_key"]
-            == "abc"
-        )
+            # search by config (defaults to checkpoints across all namespaces)
+            search_results_5 = list(
+                saver.list({"configurable": {"thread_id": "thread-2"}})
+            )
+            assert len(search_results_5) == 2
+            assert {
+                search_results_5[0].config["configurable"]["checkpoint_ns"],
+                search_results_5[1].config["configurable"]["checkpoint_ns"],
+            } == {"", "inner"}
+
+            # TODO: test before and limit params
+
+    def test_null_chars(self) -> None:
+        with PostgresSaver.from_conn_string(DEFAULT_URI) as saver:
+            config = saver.put(self.config_1, self.chkpnt_1, {"my_key": "\x00abc"}, {})
+            assert saver.get_tuple(config).metadata["my_key"] == "abc"  # type: ignore
+            assert (
+                list(saver.list(None, filter={"my_key": "abc"}))[0].metadata["my_key"]  # type: ignore
+                == "abc"
+            )

libs/checkpoint/langgraph/store/memory/__init__.py

@@ -413,8 +413,6 @@ def _cosine_similarity(X: list[float], Y: list[list[float]]) -> list[float]:
     Compute cosine similarity between a vector X and a matrix Y.
     Lazy import numpy for efficiency.
     """
-    if not Y:
-        return []
     if _check_numpy():
         import numpy as np  # type: ignore
 

libs/checkpoint/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph-checkpoint"
-version = "2.0.8"
+version = "2.0.7"
 description = "Library with base interfaces for LangGraph checkpoint savers."
 authors = []
 license = "MIT"

libs/langgraph/langgraph/graph/graph.py

@@ -374,11 +374,6 @@ class Graph:
             if source not in self.nodes and source != START:
                 raise ValueError(f"Found edge starting at unknown node '{source}'")
 
-        if START not in all_sources:
-            raise ValueError(
-                "Graph must have an entrypoint: add at least one edge from START to another node"
-            )
-
         # assemble targets
         all_targets = {end for _, end in self._all_edges}
         for start, branches in self.branches.items():
@@ -400,6 +395,10 @@ class Graph:
         for name, spec in self.nodes.items():
             if spec.ends:
                 all_targets.update(spec.ends)
+        # validate targets
+        for node in self.nodes:
+            if node not in all_targets:
+                raise ValueError(f"Node `{node}` is not reachable")
         for target in all_targets:
             if target not in self.nodes and target != END:
                 raise ValueError(f"Found edge ending at unknown node `{target}`")

libs/langgraph/langgraph/graph/state.py

@@ -933,12 +933,12 @@ def _is_field_binop(typ: Type[Any]) -> Optional[BinaryOperatorAggregate]:
     if hasattr(typ, "__metadata__"):
         meta = typ.__metadata__
         if len(meta) >= 1 and callable(meta[-1]):
-            sig = signature(meta[-1])
+            sig = signature(meta[0])
             params = list(sig.parameters.values())
             if len(params) == 2 and all(
                 p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD) for p in params
             ):
-                return BinaryOperatorAggregate(typ, meta[-1])
+                return BinaryOperatorAggregate(typ, meta[0])
             else:
                 raise ValueError(
                     f"Invalid reducer signature. Expected (a, b) -> c. Got {sig}"

libs/langgraph/langgraph/pregel/algo.py

@@ -595,7 +595,7 @@ def prepare_single_task(
                                     for tid, c, v in pending_writes
                                     if tid in (NULL_TASK_ID, task_id) and c == RESUME
                                 ),
-                                configurable.get(CONFIG_KEY_RESUME_VALUE, MISSING),
+                                MISSING,
                             ),
                         },
                     ),
@@ -720,7 +720,7 @@ def prepare_single_task(
                                         if tid in (NULL_TASK_ID, task_id)
                                         and c == RESUME
                                     ),
-                                    configurable.get(CONFIG_KEY_RESUME_VALUE, MISSING),
+                                    MISSING,
                                 ),
                             },
                         ),

libs/langgraph/langgraph/pregel/remote.py

@@ -1,4 +1,3 @@
-from dataclasses import asdict
 from typing import (
     Any,
     AsyncIterator,
@@ -28,7 +27,6 @@ from langgraph_sdk.client import (
     get_sync_client,
 )
 from langgraph_sdk.schema import Checkpoint, ThreadState
-from langgraph_sdk.schema import Command as CommandSDK
 from langgraph_sdk.schema import StreamMode as StreamModeSDK
 from typing_extensions import Self
 
@@ -43,7 +41,7 @@ from langgraph.constants import (
 from langgraph.errors import GraphInterrupt
 from langgraph.pregel.protocol import PregelProtocol
 from langgraph.pregel.types import All, PregelTask, StateSnapshot, StreamMode
-from langgraph.types import Command, Interrupt, StreamProtocol
+from langgraph.types import Interrupt, StreamProtocol
 from langgraph.utils.config import merge_configs
 
 
@@ -575,7 +573,6 @@ class RemoteGraph(PregelProtocol):
         interrupt_before: Optional[Union[All, Sequence[str]]] = None,
         interrupt_after: Optional[Union[All, Sequence[str]]] = None,
         subgraphs: bool = False,
-        **kwargs: Any,
     ) -> Iterator[Union[dict[str, Any], Any]]:
         """Create a run and stream the results.
 
@@ -590,7 +587,6 @@ class RemoteGraph(PregelProtocol):
             interrupt_before: Interrupt the graph before these nodes.
             interrupt_after: Interrupt the graph after these nodes.
             subgraphs: Stream from subgraphs.
-            **kwargs: Additional params to pass to client.runs.stream.
 
         Yields:
             The output of the graph.
@@ -601,24 +597,17 @@ class RemoteGraph(PregelProtocol):
         stream_modes, requested, req_single, stream = self._get_stream_modes(
             stream_mode, config
         )
-        if isinstance(input, Command):
-            command: Optional[CommandSDK] = cast(CommandSDK, asdict(input))
-            input = None
-        else:
-            command = None
 
         for chunk in sync_client.runs.stream(
             thread_id=sanitized_config["configurable"].get("thread_id"),
             assistant_id=self.name,
             input=input,
-            command=command,
             config=sanitized_config,
             stream_mode=stream_modes,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
             stream_subgraphs=subgraphs or stream is not None,
             if_not_exists="create",
-            **kwargs,
         ):
             # split mode and ns
             if NS_SEP in chunk.event:
@@ -667,7 +656,6 @@ class RemoteGraph(PregelProtocol):
         interrupt_before: Optional[Union[All, Sequence[str]]] = None,
         interrupt_after: Optional[Union[All, Sequence[str]]] = None,
         subgraphs: bool = False,
-        **kwargs: Any,
     ) -> AsyncIterator[Union[dict[str, Any], Any]]:
         """Create a run and stream the results.
 
@@ -682,7 +670,6 @@ class RemoteGraph(PregelProtocol):
             interrupt_before: Interrupt the graph before these nodes.
             interrupt_after: Interrupt the graph after these nodes.
             subgraphs: Stream from subgraphs.
-            **kwargs: Additional params to pass to client.runs.stream.
 
         Yields:
             The output of the graph.
@@ -693,24 +680,17 @@ class RemoteGraph(PregelProtocol):
         stream_modes, requested, req_single, stream = self._get_stream_modes(
             stream_mode, config
         )
-        if isinstance(input, Command):
-            command: Optional[CommandSDK] = cast(CommandSDK, asdict(input))
-            input = None
-        else:
-            command = None
 
         async for chunk in client.runs.stream(
             thread_id=sanitized_config["configurable"].get("thread_id"),
             assistant_id=self.name,
             input=input,
-            command=command,
             config=sanitized_config,
             stream_mode=stream_modes,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
             stream_subgraphs=subgraphs or stream is not None,
             if_not_exists="create",
-            **kwargs,
         ):
             # split mode and ns
             if NS_SEP in chunk.event:
@@ -773,16 +753,18 @@ class RemoteGraph(PregelProtocol):
         *,
         interrupt_before: Optional[Union[All, Sequence[str]]] = None,
         interrupt_after: Optional[Union[All, Sequence[str]]] = None,
-        **kwargs: Any,
     ) -> Union[dict[str, Any], Any]:
         """Create a run, wait until it finishes and return the final state.
 
+        This method calls `POST /threads/{thread_id}/runs/wait` if a `thread_id`
+        is speciffed in the `configurable` field of the config or
+        `POST /runs/wait` otherwise.
+
         Args:
             input: Input to the graph.
             config: A `RunnableConfig` for graph invocation.
             interrupt_before: Interrupt the graph before these nodes.
             interrupt_after: Interrupt the graph after these nodes.
-            **kwargs: Additional params to pass to RemoteGraph.stream.
 
         Returns:
             The output of the graph.
@@ -793,7 +775,6 @@ class RemoteGraph(PregelProtocol):
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
             stream_mode="values",
-            **kwargs,
         ):
             pass
         try:
@@ -808,16 +789,18 @@ class RemoteGraph(PregelProtocol):
         *,
         interrupt_before: Optional[Union[All, Sequence[str]]] = None,
         interrupt_after: Optional[Union[All, Sequence[str]]] = None,
-        **kwargs: Any,
     ) -> Union[dict[str, Any], Any]:
         """Create a run, wait until it finishes and return the final state.
 
+        This method calls `POST /threads/{thread_id}/runs/wait` if a `thread_id`
+        is speciffed in the `configurable` field of the config or
+        `POST /runs/wait` otherwise.
+
         Args:
             input: Input to the graph.
             config: A `RunnableConfig` for graph invocation.
             interrupt_before: Interrupt the graph before these nodes.
             interrupt_after: Interrupt the graph after these nodes.
-            **kwargs: Additional params to pass to RemoteGraph.astream.
 
         Returns:
             The output of the graph.
@@ -828,7 +811,6 @@ class RemoteGraph(PregelProtocol):
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
             stream_mode="values",
-            **kwargs,
         ):
             pass
         try:

libs/langgraph/tests/test_pregel.py

@@ -160,7 +160,7 @@ def test_graph_validation() -> None:
     workflow = Graph()
     workflow.add_node("agent", logic)
     workflow.set_finish_point("agent")
-    with pytest.raises(ValueError, match="must have an entrypoint"):
+    with pytest.raises(ValueError, match="not reachable"):
         workflow.compile()
 
     workflow = Graph()
@@ -207,6 +207,18 @@ def test_graph_validation() -> None:
     with pytest.raises(ValueError, match="unknown"):  # extra is not defined
         workflow.compile()
 
+    workflow = Graph()
+    workflow.add_node("agent", logic)
+    workflow.add_node("tools", logic)
+    workflow.add_node("extra", logic)
+    workflow.set_entry_point("agent")
+    workflow.add_conditional_edges("agent", logic, {"continue": "tools", "exit": END})
+    workflow.add_edge("tools", "agent")
+    with pytest.raises(
+        ValueError, match="Node `extra` is not reachable"
+    ):  # extra is not reachable
+        workflow.compile()
+
     workflow = Graph()
     workflow.add_node("agent", logic)
     workflow.add_node("tools", logic)
@@ -264,25 +276,6 @@ def test_graph_validation() -> None:
         graph.invoke({"hello": "there"})
 
 
-def test_graph_validation_with_command() -> None:
-    class State(TypedDict):
-        foo: str
-        bar: str
-
-    def node_a(state: State):
-        return GraphCommand(goto="b", update={"foo": "bar"})
-
-    def node_b(state: State):
-        return GraphCommand(goto=END, update={"bar": "baz"})
-
-    builder = StateGraph(State)
-    builder.add_node("a", node_a)
-    builder.add_node("b", node_b)
-    builder.add_edge(START, "a")
-    graph = builder.compile()
-    assert graph.invoke({"foo": ""}) == {"foo": "bar", "bar": "baz"}
-
-
 def test_checkpoint_errors() -> None:
     class FaultyGetCheckpointer(MemorySaver):
         def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
@@ -8735,176 +8728,6 @@ def test_copy_checkpoint(
     )
 
 
-@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
-def test_dynamic_interrupt_subgraph(
-    request: pytest.FixtureRequest, checkpointer_name: str
-) -> None:
-    checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}")
-
-    class SubgraphState(TypedDict):
-        my_key: str
-        market: str
-
-    tool_two_node_count = 0
-
-    def tool_two_node(s: SubgraphState) -> SubgraphState:
-        nonlocal tool_two_node_count
-        tool_two_node_count += 1
-        if s["market"] == "DE":
-            answer = interrupt("Just because...")
-        else:
-            answer = " all good"
-        return {"my_key": answer}
-
-    subgraph = StateGraph(SubgraphState)
-    subgraph.add_node("do", tool_two_node, retry=RetryPolicy())
-    subgraph.add_edge(START, "do")
-
-    class State(TypedDict):
-        my_key: Annotated[str, operator.add]
-        market: str
-
-    tool_two_graph = StateGraph(State)
-    tool_two_graph.add_node("tool_two", subgraph.compile())
-    tool_two_graph.add_edge(START, "tool_two")
-    tool_two = tool_two_graph.compile()
-
-    tracer = FakeTracer()
-    assert tool_two.invoke(
-        {"my_key": "value", "market": "DE"}, {"callbacks": [tracer]}
-    ) == {
-        "my_key": "value",
-        "market": "DE",
-    }
-    assert tool_two_node_count == 1, "interrupts aren't retried"
-    assert len(tracer.runs) == 1
-    run = tracer.runs[0]
-    assert run.end_time is not None
-    assert run.error is None
-    assert run.outputs == {"market": "DE", "my_key": "value"}
-
-    assert tool_two.invoke({"my_key": "value", "market": "US"}) == {
-        "my_key": "value all good",
-        "market": "US",
-    }
-
-    tool_two = tool_two_graph.compile(checkpointer=checkpointer)
-
-    # missing thread_id
-    with pytest.raises(ValueError, match="thread_id"):
-        tool_two.invoke({"my_key": "value", "market": "DE"})
-
-    # flow: interrupt -> resume with answer
-    thread2 = {"configurable": {"thread_id": "2"}}
-    # stop when about to enter node
-    assert [
-        c for c in tool_two.stream({"my_key": "value ⛰️", "market": "DE"}, thread2)
-    ] == [
-        {
-            "__interrupt__": (
-                Interrupt(
-                    value="Just because...",
-                    resumable=True,
-                    ns=[AnyStr("tool_two:"), AnyStr("do:")],
-                ),
-            )
-        },
-    ]
-    # resume with answer
-    assert [c for c in tool_two.stream(Command(resume=" my answer"), thread2)] == [
-        {"tool_two": {"my_key": " my answer", "market": "DE"}},
-    ]
-
-    # flow: interrupt -> clear tasks
-    thread1 = {"configurable": {"thread_id": "1"}}
-    # stop when about to enter node
-    assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == {
-        "my_key": "value ⛰️",
-        "market": "DE",
-    }
-    assert [
-        c.metadata
-        for c in tool_two.checkpointer.list(
-            {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
-        )
-    ] == [
-        {
-            "parents": {},
-            "source": "loop",
-            "step": 0,
-            "writes": None,
-            "thread_id": "1",
-        },
-        {
-            "parents": {},
-            "source": "input",
-            "step": -1,
-            "writes": {"__start__": {"my_key": "value ⛰️", "market": "DE"}},
-            "thread_id": "1",
-        },
-    ]
-    assert tool_two.get_state(thread1) == StateSnapshot(
-        values={"my_key": "value ⛰️", "market": "DE"},
-        next=("tool_two",),
-        tasks=(
-            PregelTask(
-                AnyStr(),
-                "tool_two",
-                (PULL, "tool_two"),
-                interrupts=(
-                    Interrupt(
-                        value="Just because...",
-                        resumable=True,
-                        ns=[AnyStr("tool_two:"), AnyStr("do:")],
-                    ),
-                ),
-                state={
-                    "configurable": {
-                        "thread_id": "1",
-                        "checkpoint_ns": AnyStr("tool_two:"),
-                    }
-                },
-            ),
-        ),
-        config=tool_two.checkpointer.get_tuple(thread1).config,
-        created_at=tool_two.checkpointer.get_tuple(thread1).checkpoint["ts"],
-        metadata={
-            "parents": {},
-            "source": "loop",
-            "step": 0,
-            "writes": None,
-            "thread_id": "1",
-        },
-        parent_config=[
-            *tool_two.checkpointer.list(
-                {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}, limit=2
-            )
-        ][-1].config,
-    )
-    # clear the interrupt and next tasks
-    tool_two.update_state(thread1, None, as_node=END)
-    # interrupt and next tasks are cleared
-    assert tool_two.get_state(thread1) == StateSnapshot(
-        values={"my_key": "value ⛰️", "market": "DE"},
-        next=(),
-        tasks=(),
-        config=tool_two.checkpointer.get_tuple(thread1).config,
-        created_at=tool_two.checkpointer.get_tuple(thread1).checkpoint["ts"],
-        metadata={
-            "parents": {},
-            "source": "update",
-            "step": 1,
-            "writes": {},
-            "thread_id": "1",
-        },
-        parent_config=[
-            *tool_two.checkpointer.list(
-                {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}, limit=2
-            )
-        ][-1].config,
-    )
-
-
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
 def test_start_branch_then(
     snapshot: SnapshotAssertion, request: pytest.FixtureRequest, checkpointer_name: str
@@ -14648,35 +14471,3 @@ def test_parent_command(request: pytest.FixtureRequest, checkpointer_name: str)
         },
         tasks=(),
     )
-
-
-@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
-def test_interrupt_subgraph(request: pytest.FixtureRequest, checkpointer_name: str):
-    checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}")
-
-    class State(TypedDict):
-        baz: str
-
-    def foo(state):
-        return {"baz": "foo"}
-
-    def bar(state):
-        value = interrupt("Please provide baz value:")
-        return {"baz": value}
-
-    child_builder = StateGraph(State)
-    child_builder.add_node(bar)
-    child_builder.add_edge(START, "bar")
-
-    builder = StateGraph(State)
-    builder.add_node(foo)
-    builder.add_node("bar", child_builder.compile())
-    builder.add_edge(START, "foo")
-    builder.add_edge("foo", "bar")
-    graph = builder.compile(checkpointer=checkpointer)
-
-    thread1 = {"configurable": {"thread_id": "1"}}
-    # First run, interrupted at bar
-    assert graph.invoke({"baz": ""}, thread1)
-    # Resume with answer
-    assert graph.invoke(Command(resume="bar"), thread1)

libs/langgraph/tests/test_pregel_async.py

@@ -429,189 +429,6 @@ async def test_dynamic_interrupt(checkpointer_name: str) -> None:
         )
 
 
-@pytest.mark.skipif(
-    sys.version_info < (3, 11),
-    reason="Python 3.11+ is required for async contextvars support",
-)
-@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
-async def test_dynamic_interrupt_subgraph(checkpointer_name: str) -> None:
-    class SubgraphState(TypedDict):
-        my_key: str
-        market: str
-
-    tool_two_node_count = 0
-
-    def tool_two_node(s: SubgraphState) -> SubgraphState:
-        nonlocal tool_two_node_count
-        tool_two_node_count += 1
-        if s["market"] == "DE":
-            answer = interrupt("Just because...")
-        else:
-            answer = " all good"
-        return {"my_key": answer}
-
-    subgraph = StateGraph(SubgraphState)
-    subgraph.add_node("do", tool_two_node, retry=RetryPolicy())
-    subgraph.add_edge(START, "do")
-
-    class State(TypedDict):
-        my_key: Annotated[str, operator.add]
-        market: str
-
-    tool_two_graph = StateGraph(State)
-    tool_two_graph.add_node("tool_two", subgraph.compile())
-    tool_two_graph.add_edge(START, "tool_two")
-    tool_two = tool_two_graph.compile()
-
-    tracer = FakeTracer()
-    assert await tool_two.ainvoke(
-        {"my_key": "value", "market": "DE"}, {"callbacks": [tracer]}
-    ) == {
-        "my_key": "value",
-        "market": "DE",
-    }
-    assert tool_two_node_count == 1, "interrupts aren't retried"
-    assert len(tracer.runs) == 1
-    run = tracer.runs[0]
-    assert run.end_time is not None
-    assert run.error is None
-    assert run.outputs == {"market": "DE", "my_key": "value"}
-
-    assert await tool_two.ainvoke({"my_key": "value", "market": "US"}) == {
-        "my_key": "value all good",
-        "market": "US",
-    }
-
-    async with awith_checkpointer(checkpointer_name) as checkpointer:
-        tool_two = tool_two_graph.compile(checkpointer=checkpointer)
-
-        # missing thread_id
-        with pytest.raises(ValueError, match="thread_id"):
-            await tool_two.ainvoke({"my_key": "value", "market": "DE"})
-
-        # flow: interrupt -> resume with answer
-        thread2 = {"configurable": {"thread_id": "2"}}
-        # stop when about to enter node
-        assert [
-            c
-            async for c in tool_two.astream(
-                {"my_key": "value ⛰️", "market": "DE"}, thread2
-            )
-        ] == [
-            {
-                "__interrupt__": (
-                    Interrupt(
-                        value="Just because...",
-                        resumable=True,
-                        ns=[AnyStr("tool_two:"), AnyStr("do:")],
-                    ),
-                )
-            },
-        ]
-        # resume with answer
-        assert [
-            c async for c in tool_two.astream(Command(resume=" my answer"), thread2)
-        ] == [
-            {"tool_two": {"my_key": " my answer", "market": "DE"}},
-        ]
-
-        # flow: interrupt -> clear
-        thread1 = {"configurable": {"thread_id": "1"}}
-        thread1root = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
-        # stop when about to enter node
-        assert [
-            c
-            async for c in tool_two.astream(
-                {"my_key": "value ⛰️", "market": "DE"}, thread1
-            )
-        ] == [
-            {
-                "__interrupt__": (
-                    Interrupt(
-                        value="Just because...",
-                        resumable=True,
-                        ns=[AnyStr("tool_two:"), AnyStr("do:")],
-                    ),
-                )
-            },
-        ]
-        assert [c.metadata async for c in tool_two.checkpointer.alist(thread1root)] == [
-            {
-                "parents": {},
-                "source": "loop",
-                "step": 0,
-                "writes": None,
-                "thread_id": "1",
-            },
-            {
-                "parents": {},
-                "source": "input",
-                "step": -1,
-                "writes": {"__start__": {"my_key": "value ⛰️", "market": "DE"}},
-                "thread_id": "1",
-            },
-        ]
-        tup = await tool_two.checkpointer.aget_tuple(thread1)
-        assert await tool_two.aget_state(thread1) == StateSnapshot(
-            values={"my_key": "value ⛰️", "market": "DE"},
-            next=("tool_two",),
-            tasks=(
-                PregelTask(
-                    AnyStr(),
-                    "tool_two",
-                    (PULL, "tool_two"),
-                    interrupts=(
-                        Interrupt(
-                            value="Just because...",
-                            resumable=True,
-                            ns=[AnyStr("tool_two:"), AnyStr("do:")],
-                        ),
-                    ),
-                    state={
-                        "configurable": {
-                            "thread_id": "1",
-                            "checkpoint_ns": AnyStr("tool_two:"),
-                        }
-                    },
-                ),
-            ),
-            config=tup.config,
-            created_at=tup.checkpoint["ts"],
-            metadata={
-                "parents": {},
-                "source": "loop",
-                "step": 0,
-                "writes": None,
-                "thread_id": "1",
-            },
-            parent_config=[
-                c async for c in tool_two.checkpointer.alist(thread1root, limit=2)
-            ][-1].config,
-        )
-
-        # clear the interrupt and next tasks
-        await tool_two.aupdate_state(thread1, None, as_node=END)
-        # interrupt is cleared, as well as the next tasks
-        tup = await tool_two.checkpointer.aget_tuple(thread1)
-        assert await tool_two.aget_state(thread1) == StateSnapshot(
-            values={"my_key": "value ⛰️", "market": "DE"},
-            next=(),
-            tasks=(),
-            config=tup.config,
-            created_at=tup.checkpoint["ts"],
-            metadata={
-                "parents": {},
-                "source": "update",
-                "step": 1,
-                "writes": {},
-                "thread_id": "1",
-            },
-            parent_config=[
-                c async for c in tool_two.checkpointer.alist(thread1root, limit=2)
-            ][-1].config,
-        )
-
-
 @pytest.mark.skipif(not FF_SEND_V2, reason="send v2 is not enabled")
 @pytest.mark.skipif(
     sys.version_info < (3, 11),
@@ -12860,39 +12677,3 @@ async def test_parent_command(checkpointer_name: str) -> None:
             },
             tasks=(),
         )
-
-
-@pytest.mark.skipif(
-    sys.version_info < (3, 11),
-    reason="Python 3.11+ is required for async contextvars support",
-)
-@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
-async def test_interrupt_subgraph(checkpointer_name: str):
-    class State(TypedDict):
-        baz: str
-
-    def foo(state):
-        return {"baz": "foo"}
-
-    def bar(state):
-        value = interrupt("Please provide baz value:")
-        return {"baz": value}
-
-    child_builder = StateGraph(State)
-    child_builder.add_node(bar)
-    child_builder.add_edge(START, "bar")
-
-    builder = StateGraph(State)
-    builder.add_node(foo)
-    builder.add_node("bar", child_builder.compile())
-    builder.add_edge(START, "foo")
-    builder.add_edge("foo", "bar")
-
-    async with awith_checkpointer(checkpointer_name) as checkpointer:
-        graph = builder.compile(checkpointer=checkpointer)
-
-        thread1 = {"configurable": {"thread_id": "1"}}
-        # First run, interrupted at bar
-        assert await graph.ainvoke({"baz": ""}, thread1)
-        # Resume with answer
-        assert await graph.ainvoke(Command(resume="bar"), thread1)

libs/sdk-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@langchain/langgraph-sdk",
-  "version": "0.0.30",
+  "version": "0.0.29",
   "description": "Client library for interacting with the LangGraph API",
   "type": "module",
   "packageManager": "yarn@1.22.19",

libs/sdk-js/src/client.ts

@@ -7,7 +7,6 @@ import {
   GraphSchema,
   Metadata,
   Run,
-  RunStatus,
   Thread,
   ThreadState,
   Cron,
@@ -945,18 +944,12 @@ export class RunsClient extends BaseClient {
        * Defaults to 0.
        */
       offset?: number;
-
-      /**
-       * Status of the run to filter by.
-       */
-      status?: RunStatus;
     },
   ): Promise<Run[]> {
     return this.fetch<Run[]>(`/threads/${threadId}/runs`, {
       params: {
         limit: options?.limit ?? 10,
         offset: options?.offset ?? 0,
-        status: options?.status ?? undefined,
       },
     });
   }
@@ -1021,28 +1014,19 @@ export class RunsClient extends BaseClient {
    *
    * @param threadId The ID of the thread.
    * @param runId The ID of the run.
+   * @param signal An optional abort signal.
    * @returns An async generator yielding stream parts.
    */
   async *joinStream(
     threadId: string,
     runId: string,
-    options?:
-      | { signal?: AbortSignal; cancelOnDisconnect?: boolean }
-      | AbortSignal,
+    signal?: AbortSignal,
   ): AsyncGenerator<{ event: StreamEvent; data: any }> {
-    const opts =
-      typeof options === "object" &&
-      options != null &&
-      options instanceof AbortSignal
-        ? { signal: options }
-        : options;
-
     const response = await this.asyncCaller.fetch(
       ...this.prepareFetchOptions(`/threads/${threadId}/runs/${runId}/stream`, {
         method: "GET",
         timeoutMs: null,
-        signal: opts?.signal,
-        params: { cancel_on_disconnect: opts?.cancelOnDisconnect ? "1" : "0" },
+        signal,
       }),
     );
 
@@ -1057,7 +1041,7 @@ export class RunsClient extends BaseClient {
         async start(ctrl) {
           parser = createParser((event) => {
             if (
-              (opts?.signal && opts.signal.aborted) ||
+              (signal && signal.aborted) ||
               (event.type === "event" && event.data === "[DONE]")
             ) {
               ctrl.terminate();

libs/sdk-js/src/schema.ts

@@ -2,7 +2,7 @@ import type { JSONSchema7 } from "json-schema";
 
 type Optional<T> = T | null | undefined;
 
-export type RunStatus =
+type RunStatus =
   | "pending"
   | "running"
   | "error"

libs/sdk-py/langgraph_sdk/client.py

@@ -26,6 +26,7 @@ from typing import (
 )
 
 import httpx
+import httpx_sse
 import orjson
 from httpx._types import QueryParamTypes
 
@@ -50,7 +51,6 @@ from langgraph_sdk.schema import (
     OnConflictBehavior,
     Run,
     RunCreate,
-    RunStatus,
     SearchItemsResponse,
     StreamMode,
     StreamPart,
@@ -60,7 +60,6 @@ from langgraph_sdk.schema import (
     ThreadStatus,
     ThreadUpdateStateResponse,
 )
-from langgraph_sdk.sse import SSEDecoder, aiter_lines_raw, iter_lines_raw
 
 logger = logging.getLogger(__name__)
 
@@ -282,35 +281,22 @@ class HttpClient:
     ) -> AsyncIterator[StreamPart]:
         """Stream results using SSE."""
         headers, content = await aencode_json(json)
-        headers["Accept"] = "text/event-stream"
-        headers["Cache-Control"] = "no-store"
-
-        async with self.client.stream(
-            method, path, headers=headers, content=content
-        ) as res:
-            # check status
+        async with httpx_sse.aconnect_sse(
+            self.client, method, path, headers=headers, content=content
+        ) as sse:
             try:
-                res.raise_for_status()
+                sse.response.raise_for_status()
             except httpx.HTTPStatusError as e:
-                body = (await res.aread()).decode()
+                body = (await sse.response.aread()).decode()
                 if sys.version_info >= (3, 11):
                     e.add_note(body)
                 else:
                     logger.error(f"Error from langgraph-api: {body}", exc_info=e)
                 raise e
-            # check content type
-            content_type = res.headers.get("content-type", "").partition(";")[0]
-            if "text/event-stream" not in content_type:
-                raise httpx.TransportError(
-                    "Expected response header Content-Type to contain 'text/event-stream', "
-                    f"got {content_type!r}"
+            async for event in sse.aiter_sse():
+                yield StreamPart(
+                    event.event, orjson.loads(event.data) if event.data else None
                 )
-            # parse SSE
-            decoder = SSEDecoder()
-            async for line in aiter_lines_raw(res):
-                sse = decoder.decode(line=line.rstrip(b"\n"))
-                if sse is not None:
-                    yield sse
 
 
 async def aencode_json(json: Any) -> tuple[dict[str, str], bytes]:
@@ -1698,12 +1684,7 @@ class RunsClient:
         return response
 
     async def list(
-        self,
-        thread_id: str,
-        *,
-        limit: int = 10,
-        offset: int = 0,
-        status: Optional[RunStatus] = None,
+        self, thread_id: str, *, limit: int = 10, offset: int = 0
     ) -> List[Run]:
         """List runs.
 
@@ -1711,7 +1692,6 @@ class RunsClient:
             thread_id: The thread ID to list runs for.
             limit: The maximum number of results to return.
             offset: The number of results to skip.
-            status: The status of the run to filter by.
 
         Returns:
             List[Run]: The runs for the thread.
@@ -1725,13 +1705,9 @@ class RunsClient:
             )
 
         """  # noqa: E501
-        params = {
-            "limit": limit,
-            "offset": offset,
-        }
-        if status is not None:
-            params["status"] = status
-        return await self.http.get(f"/threads/{thread_id}/runs", params=params)
+        return await self.http.get(
+            f"/threads/{thread_id}/runs?limit={limit}&offset={offset}"
+        )
 
     async def get(self, thread_id: str, run_id: str) -> Run:
         """Get a run.
@@ -1809,9 +1785,7 @@ class RunsClient:
         """  # noqa: E501
         return await self.http.get(f"/threads/{thread_id}/runs/{run_id}/join")
 
-    def join_stream(
-        self, thread_id: str, run_id: str, *, cancel_on_disconnect: bool = False
-    ) -> AsyncIterator[StreamPart]:
+    def join_stream(self, thread_id: str, run_id: str) -> AsyncIterator[StreamPart]:
         """Stream output from a run in real-time, until the run is done.
         Output is not buffered, so any output produced before this call will
         not be received here.
@@ -1819,7 +1793,6 @@ class RunsClient:
         Args:
             thread_id: The thread ID to join.
             run_id: The run ID to join.
-            cancel_on_disconnect: Whether to cancel the run when the stream is disconnected.
 
         Returns:
             None
@@ -1832,11 +1805,7 @@ class RunsClient:
             )
 
         """  # noqa: E501
-        return self.http.stream(
-            f"/threads/{thread_id}/runs/{run_id}/stream",
-            "GET",
-            params={"cancel_on_disconnect": cancel_on_disconnect},
-        )
+        return self.http.stream(f"/threads/{thread_id}/runs/{run_id}/stream", "GET")
 
     async def delete(self, thread_id: str, run_id: str) -> None:
         """Delete a run.
@@ -2451,30 +2420,22 @@ class SyncHttpClient:
     ) -> Iterator[StreamPart]:
         """Stream the results of a request using SSE."""
         headers, content = encode_json(json)
-        with self.client.stream(method, path, headers=headers, content=content) as res:
-            # check status
+        with httpx_sse.connect_sse(
+            self.client, method, path, headers=headers, content=content
+        ) as sse:
             try:
-                res.raise_for_status()
+                sse.response.raise_for_status()
             except httpx.HTTPStatusError as e:
-                body = (res.read()).decode()
+                body = sse.response.read().decode()
                 if sys.version_info >= (3, 11):
                     e.add_note(body)
                 else:
                     logger.error(f"Error from langgraph-api: {body}", exc_info=e)
                 raise e
-            # check content type
-            content_type = res.headers.get("content-type", "").partition(";")[0]
-            if "text/event-stream" not in content_type:
-                raise httpx.TransportError(
-                    "Expected response header Content-Type to contain 'text/event-stream', "
-                    f"got {content_type!r}"
+            for event in sse.iter_sse():
+                yield StreamPart(
+                    event.event, orjson.loads(event.data) if event.data else None
                 )
-            # parse SSE
-            decoder = SSEDecoder()
-            for line in iter_lines_raw(res):
-                sse = decoder.decode(line.rstrip(b"\n"))
-                if sse is not None:
-                    yield sse
 
 
 def encode_json(json: Any) -> tuple[dict[str, str], bytes]:
@@ -3342,7 +3303,6 @@ class SyncRunsClient:
         assistant_id: str,
         *,
         input: Optional[dict] = None,
-        command: Optional[Command] = None,
         stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values",
         stream_subgraphs: bool = False,
         metadata: Optional[dict] = None,
@@ -3366,7 +3326,6 @@ class SyncRunsClient:
         assistant_id: str,
         *,
         input: Optional[dict] = None,
-        command: Optional[Command] = None,
         stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values",
         stream_subgraphs: bool = False,
         metadata: Optional[dict] = None,
@@ -3387,7 +3346,6 @@ class SyncRunsClient:
         assistant_id: str,
         *,
         input: Optional[dict] = None,
-        command: Optional[Command] = None,
         stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values",
         stream_subgraphs: bool = False,
         metadata: Optional[dict] = None,
@@ -3412,7 +3370,6 @@ class SyncRunsClient:
             assistant_id: The assistant ID or graph name to stream from.
                 If using graph name, will default to first assistant created from that graph.
             input: The input to the graph.
-            command: The command to execute.
             stream_mode: The stream mode(s) to use.
             stream_subgraphs: Whether to stream output from subgraphs.
             metadata: Metadata to assign to the run.
@@ -3463,7 +3420,6 @@ class SyncRunsClient:
         """  # noqa: E501
         payload = {
             "input": input,
-            "command": command,
             "config": config,
             "metadata": metadata,
             "stream_mode": stream_mode,
@@ -3497,7 +3453,6 @@ class SyncRunsClient:
         assistant_id: str,
         *,
         input: Optional[dict] = None,
-        command: Optional[Command] = None,
         stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values",
         stream_subgraphs: bool = False,
         metadata: Optional[dict] = None,
@@ -3517,7 +3472,6 @@ class SyncRunsClient:
         assistant_id: str,
         *,
         input: Optional[dict] = None,
-        command: Optional[Command] = None,
         stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values",
         stream_subgraphs: bool = False,
         metadata: Optional[dict] = None,
@@ -3538,7 +3492,6 @@ class SyncRunsClient:
         assistant_id: str,
         *,
         input: Optional[dict] = None,
-        command: Optional[Command] = None,
         stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values",
         stream_subgraphs: bool = False,
         metadata: Optional[dict] = None,
@@ -3561,7 +3514,6 @@ class SyncRunsClient:
             assistant_id: The assistant ID or graph name to stream from.
                 If using graph name, will default to first assistant created from that graph.
             input: The input to the graph.
-            command: The command to execute.
             stream_mode: The stream mode(s) to use.
             stream_subgraphs: Whether to stream output from subgraphs.
             metadata: Metadata to assign to the run.
@@ -3648,7 +3600,6 @@ class SyncRunsClient:
         """  # noqa: E501
         payload = {
             "input": input,
-            "command": command,
             "stream_mode": stream_mode,
             "stream_subgraphs": stream_subgraphs,
             "config": config,
@@ -3686,7 +3637,6 @@ class SyncRunsClient:
         assistant_id: str,
         *,
         input: Optional[dict] = None,
-        command: Optional[Command] = None,
         metadata: Optional[dict] = None,
         config: Optional[Config] = None,
         checkpoint: Optional[Checkpoint] = None,
@@ -3707,7 +3657,6 @@ class SyncRunsClient:
         assistant_id: str,
         *,
         input: Optional[dict] = None,
-        command: Optional[Command] = None,
         metadata: Optional[dict] = None,
         config: Optional[Config] = None,
         interrupt_before: Optional[Union[All, Sequence[str]]] = None,
@@ -3725,7 +3674,6 @@ class SyncRunsClient:
         assistant_id: str,
         *,
         input: Optional[dict] = None,
-        command: Optional[Command] = None,
         metadata: Optional[dict] = None,
         config: Optional[Config] = None,
         checkpoint: Optional[Checkpoint] = None,
@@ -3747,7 +3695,6 @@ class SyncRunsClient:
             assistant_id: The assistant ID or graph name to run.
                 If using graph name, will default to first assistant created from that graph.
             input: The input to the graph.
-            command: The command to execute.
             metadata: Metadata to assign to the run.
             config: The configuration for the assistant.
             checkpoint: The checkpoint to resume from.
@@ -3814,7 +3761,6 @@ class SyncRunsClient:
         """  # noqa: E501
         payload = {
             "input": input,
-            "command": command,
             "config": config,
             "metadata": metadata,
             "assistant_id": assistant_id,

libs/sdk-py/langgraph_sdk/schema.py

@@ -1,7 +1,7 @@
 """Data models for interacting with the LangGraph API."""
 
 from datetime import datetime
-from typing import Any, Dict, Literal, NamedTuple, Optional, Sequence, TypedDict, Union
+from typing import Any, Literal, NamedTuple, Optional, Sequence, TypedDict, Union
 
 Json = Optional[dict[str, Any]]
 """Represents a JSON-like structure, which can be None or a dictionary with string keys and any values."""
@@ -176,19 +176,6 @@ class Assistant(AssistantBase):
     """The name of the assistant"""
 
 
-class Interrupt(TypedDict, total=False):
-    """Represents an interruption in the execution flow."""
-
-    value: Any
-    """The value associated with the interrupt."""
-    when: Literal["during"]
-    """When the interrupt occurred."""
-    resumable: bool
-    """Whether the interrupt can be resumed."""
-    ns: Optional[list[str]]
-    """Optional namespace for the interrupt."""
-
-
 class Thread(TypedDict):
     """Represents a conversation thread."""
 
@@ -204,8 +191,6 @@ class Thread(TypedDict):
     """The status of the thread, one of 'idle', 'busy', 'interrupted'."""
     values: Json
     """The current state of the thread."""
-    interrupts: Dict[str, list[Interrupt]]
-    """Interrupts which were thrown in this thread"""
 
 
 class ThreadTask(TypedDict):
@@ -214,7 +199,7 @@ class ThreadTask(TypedDict):
     id: str
     name: str
     error: Optional[str]
-    interrupts: list[Interrupt]
+    interrupts: list[dict]
     checkpoint: Optional[Checkpoint]
     state: Optional["ThreadState"]
     result: Optional[dict[str, Any]]

libs/sdk-py/langgraph_sdk/sse.py

@@ -1,148 +0,0 @@
-"""Adapted from httpx_sse to split lines on \n, \r, \r\n per the SSE spec."""
-
-from typing import AsyncIterator, Iterator, Optional, Union
-
-import httpx
-import orjson
-
-from langgraph_sdk.schema import StreamPart
-
-BytesLike = Union[bytes, bytearray, memoryview]
-
-
-class BytesLineDecoder:
-    """
-    Handles incrementally reading lines from text.
-
-    Has the same behaviour as the stdllib bytes splitlines,
-    but handling the input iteratively.
-    """
-
-    def __init__(self) -> None:
-        self.buffer = bytearray()
-        self.trailing_cr: bool = False
-
-    def decode(self, text: bytes) -> list[BytesLike]:
-        # See https://docs.python.org/3/glossary.html#term-universal-newlines
-        NEWLINE_CHARS = b"\n\r"
-
-        # We always push a trailing `\r` into the next decode iteration.
-        if self.trailing_cr:
-            text = b"\r" + text
-            self.trailing_cr = False
-        if text.endswith(b"\r"):
-            self.trailing_cr = True
-            text = text[:-1]
-
-        if not text:
-            # NOTE: the edge case input of empty text doesn't occur in practice,
-            # because other httpx internals filter out this value
-            return []  # pragma: no cover
-
-        trailing_newline = text[-1] in NEWLINE_CHARS
-        lines = text.splitlines()
-
-        if len(lines) == 1 and not trailing_newline:
-            # No new lines, buffer the input and continue.
-            self.buffer.extend(lines[0])
-            return []
-
-        if self.buffer:
-            # Include any existing buffer in the first portion of the
-            # splitlines result.
-            self.buffer.extend(lines[0])
-            lines = [self.buffer] + lines[1:]
-            self.buffer = bytearray()
-
-        if not trailing_newline:
-            # If the last segment of splitlines is not newline terminated,
-            # then drop it from our output and start a new buffer.
-            self.buffer.extend(lines.pop())
-
-        return lines
-
-    def flush(self) -> list[BytesLike]:
-        if not self.buffer and not self.trailing_cr:
-            return []
-
-        lines = [self.buffer]
-        self.buffer = bytearray()
-        self.trailing_cr = False
-        return lines
-
-
-class SSEDecoder:
-    def __init__(self) -> None:
-        self._event = ""
-        self._data = bytearray()
-        self._last_event_id = ""
-        self._retry: Optional[int] = None
-
-    def decode(self, line: bytes) -> Optional[StreamPart]:
-        # See: https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation  # noqa: E501
-
-        if not line:
-            if (
-                not self._event
-                and not self._data
-                and not self._last_event_id
-                and self._retry is None
-            ):
-                return None
-
-            sse = StreamPart(
-                event=self._event,
-                data=orjson.loads(self._data) if self._data else None,
-            )
-
-            # NOTE: as per the SSE spec, do not reset last_event_id.
-            self._event = ""
-            self._data = bytearray()
-            self._retry = None
-
-            return sse
-
-        if line.startswith(b":"):
-            return None
-
-        fieldname, _, value = line.partition(b":")
-
-        if value.startswith(b" "):
-            value = value[1:]
-
-        if fieldname == b"event":
-            self._event = value.decode()
-        elif fieldname == b"data":
-            self._data.extend(value)
-        elif fieldname == b"id":
-            if b"\0" in value:
-                pass
-            else:
-                self._last_event_id = value.decode()
-        elif fieldname == b"retry":
-            try:
-                self._retry = int(value)
-            except (TypeError, ValueError):
-                pass
-        else:
-            pass  # Field is ignored.
-
-        return None
-
-
-async def aiter_lines_raw(response: httpx.Response) -> AsyncIterator[BytesLike]:
-    decoder = BytesLineDecoder()
-    async for chunk in response.aiter_bytes():
-        for line in decoder.decode(chunk):
-            yield line
-    for line in decoder.flush():
-        yield line
-
-
-def iter_lines_raw(response: httpx.Response) -> Iterator[BytesLike]:
-    decoder = BytesLineDecoder()
-    for chunk in response.iter_bytes():
-        for line in decoder.decode(chunk):
-            yield line
-    for line in decoder.flush():
-        yield line

libs/sdk-py/poetry.lock

@@ -141,6 +141,17 @@ cli = ["click (==8.*)", "pygments (==2.*)", "rich (>=10,<14)"]
 http2 = ["h2 (>=3,<5)"]
 socks = ["socksio (==1.*)"]
 
+[[package]]
+name = "httpx-sse"
+version = "0.4.0"
+description = "Consume Server-Sent Event (SSE) messages with HTTPX."
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "httpx-sse-0.4.0.tar.gz", hash = "sha256:1e81a3a3070ce322add1d3529ed42eb5f70817f45ed6ec915ab753f961139721"},
+    {file = "httpx_sse-0.4.0-py3-none-any.whl", hash = "sha256:f329af6eae57eaa2bdfd962b42524764af68075ea87370a2de920af5341e318f"},
+]
+
 [[package]]
 name = "idna"
 version = "3.7"
@@ -479,4 +490,4 @@ watchmedo = ["PyYAML (>=3.10)"]
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.9.0,<4.0"
-content-hash = "1262a6148df18cc44ade00466b6e0f8305897a460eea370c8de649d8d20cd7a2"
+content-hash = "832acea0ad21ce71ae74edef225a1ad6f8fb166f6bf1531d876fe80fac7495f0"

libs/sdk-py/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph-sdk"
-version = "0.1.42"
+version = "0.1.40"
 description = "SDK for interacting with LangGraph API"
 authors = []
 license = "MIT"
@@ -11,6 +11,7 @@ packages = [{ include = "langgraph_sdk" }]
 [tool.poetry.dependencies]
 python = "^3.9.0,<4.0"
 httpx = ">=0.25.2"
+httpx-sse = ">=0.4.0"
 orjson = ">=3.10.1"
 
 [tool.poetry.group.dev.dependencies]