Better handle sync calls

lib: Add support for returning multiple commands from a node

.github/workflows/_lint.yml

@@ -42,7 +42,6 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
           poetry-version: ${{ env.POETRY_VERSION }}
-          working-directory: ${{ inputs.working-directory }}
           cache-key: lint-${{ inputs.working-directory }}
 
       - name: Check Poetry File

.github/workflows/_test.yml

@@ -31,7 +31,6 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
           poetry-version: ${{ env.POETRY_VERSION }}
-          working-directory: ${{ inputs.working-directory }}
           cache-key: test-${{ inputs.working-directory }}
       - name: Login to Docker Hub
         uses: docker/login-action@v3

.github/workflows/_test_release.yml

@@ -29,7 +29,6 @@ jobs:
         with:
           python-version: ${{ env.PYTHON_VERSION }}
           poetry-version: ${{ env.POETRY_VERSION }}
-          working-directory: ${{ inputs.working-directory }}
           cache-key: release
 
       # We want to keep this build stage *separate* from the release stage,

.github/workflows/release.yml

@@ -31,7 +31,6 @@ jobs:
         with:
           python-version: ${{ env.PYTHON_VERSION }}
           poetry-version: ${{ env.POETRY_VERSION }}
-          working-directory: ${{ inputs.working-directory }}
           cache-key: release
 
       # We want to keep this build stage *separate* from the release stage,
@@ -169,7 +168,6 @@ jobs:
         with:
           python-version: ${{ env.PYTHON_VERSION }}
           poetry-version: ${{ env.POETRY_VERSION }}
-          working-directory: ${{ inputs.working-directory }}
 
       - name: Import published package
         shell: bash
@@ -256,7 +254,6 @@ jobs:
         with:
           python-version: ${{ env.PYTHON_VERSION }}
           poetry-version: ${{ env.POETRY_VERSION }}
-          working-directory: ${{ inputs.working-directory }}
           cache-key: release
 
       - uses: actions/download-artifact@v4
@@ -298,7 +295,6 @@ jobs:
         with:
           python-version: ${{ env.PYTHON_VERSION }}
           poetry-version: ${{ env.POETRY_VERSION }}
-          working-directory: ${{ inputs.working-directory }}
           cache-key: release
 
       - uses: actions/download-artifact@v4

docs/docs/cloud/deployment/setup.md

@@ -36,8 +36,8 @@ Dependencies can optionally be specified in one of the following files: `pyproje
 The dependencies below will be included in the image, you can also use them in your code, as long as with a compatible version range:
 
 ```
-langgraph>=0.2.30,<0.3.0
-langgraph-checkpoint>=1.0.14
+langgraph>=0.2.56,<0.3.0
+langgraph-checkpoint>=2.0.5,<3.0
 langchain-core>=0.2.38,<0.4.0
 langsmith>=0.1.63
 orjson>=3.9.7

docs/docs/cloud/deployment/setup_pyproject.md

@@ -36,8 +36,8 @@ Dependencies can optionally be specified in one of the following files: `pyproje
 The dependencies below will be included in the image, you can also use them in your code, as long as with a compatible version range:
 
 ```
-langgraph>=0.2.30,<0.3.0
-langgraph-checkpoint>=1.0.14
+langgraph>=0.2.56,<0.3.0
+langgraph-checkpoint>=2.0.5,<3.0
 langchain-core>=0.2.38,<0.4.0
 langsmith>=0.1.63
 orjson>=3.9.7

docs/docs/cloud/reference/cli.md

@@ -134,6 +134,11 @@ langgraph [OPTIONS] COMMAND [ARGS]
 
 Run LangGraph API server in development mode with hot reloading and debugging capabilities. This lightweight server requires no Docker installation and is suitable for development and testing. State is persisted to a local directory.
 
+!!! note "Python only"
+
+    Currently, the CLI only supports Python >= 3.11.
+    JS support is coming soon.
+
 **Installation**
 
 This command requires the "inmem" extra to be installed:

docs/docs/concepts/langgraph_cli.md

@@ -33,6 +33,11 @@ The `langgraph build` command builds a Docker image for the [LangGraph API serve
 !!! note "New in version 0.1.55"
     The `langgraph dev` command was introduced in langgraph-cli version 0.1.55.
 
+!!! note "Python only"
+
+    Currently, the CLI only supports Python >= 3.11.
+    JS support is coming soon.
+
 The `langgraph dev` command starts a lightweight development server that requires no Docker installation. This server is ideal for rapid development and testing, with features like:
 
 - Hot reloading: Changes to your code are automatically detected and reloaded

docs/docs/concepts/template_applications.md

@@ -1,14 +1,21 @@
 # Template Applications
 
-!!! note Prerequisites
-
-    - [LangGraph Studio](./langgraph_studio.md)
-
 Templates are open source reference applications designed to help you get started quickly when building with LangGraph. They provide working examples of common agentic workflows that can be customized to your needs.
 
-Templates can be accessed via [LangGraph Studio (macOS only)](langgraph_studio.md), or cloned directly from Github. You can download LangGraph Studio and see available templates [here](https://studio.langchain.com/).
+You can create an application from a template using the LangGraph CLI.
 
-## Available templates
+!!! info "Requirements"
+
+    - Python >= 3.11
+    - [LangGraph CLI](https://langchain-ai.github.io/langgraph/cloud/reference/cli/): Requires langchain-cli[inmem] >= 0.1.58
+
+## Install the LangGraph CLI
+
+```bash
+pip install "langgraph-cli[inmem]==0.1.58" python-dotenv
+```
+
+## Available Templates
 
 | Template                  | Description                                                                              | Python                                                           | JS/TS                                                               |
 |---------------------------|------------------------------------------------------------------------------------------|------------------------------------------------------------------|---------------------------------------------------------------------|
@@ -17,3 +24,39 @@ Templates can be accessed via [LangGraph Studio (macOS only)](langgraph_studio.m
 | **Memory Agent**          | A ReAct-style agent with an additional tool to store memories for use across threads.    | [Repo](https://github.com/langchain-ai/memory-agent)             | [Repo](https://github.com/langchain-ai/memory-agent-js)             |
 | **Retrieval Agent**       | An agent that includes a retrieval-based question-answering system.                      | [Repo](https://github.com/langchain-ai/retrieval-agent-template) | [Repo](https://github.com/langchain-ai/retrieval-agent-template-js) |
 | **Data-Enrichment Agent** | An agent that performs web searches and organizes its findings into a structured format. | [Repo](https://github.com/langchain-ai/data-enrichment)          | [Repo](https://github.com/langchain-ai/data-enrichment-js)          |
+
+
+## 🌱 Create a LangGraph App
+
+To create a new app from a template, use the `langgraph new` command.
+
+```bash
+langgraph new
+```
+
+## Next Steps
+
+Review the `README.md` file in the root of your new LangGraph app for more information about the template and how to customize it.
+
+After configuring the app properly and adding your API keys, you can start the app using the LangGraph CLI:
+
+```bash
+langgraph dev 
+```
+
+See the following guides for more information on how to deploy your app:
+
+- **[Launch Local LangGraph Server](../tutorials/langgraph-platform/local-server.md)**: This quick start guide shows how to start a LangGraph Server locally for the **ReAct Agent** template. The steps are similar for other templates.
+- **[Deploy to LangGraph Cloud](../cloud/quick_start.md)**: Deploy your LangGraph app using LangGraph Cloud.
+ 
+### LangGraph Framework
+
+- **[LangGraph Concepts](../concepts/index.md)**: Learn the foundational concepts of LangGraph.
+- **[LangGraph How-to Guides](../how-tos/index.md)**: Guides for common tasks with LangGraph.
+
+### 📚 Learn More about LangGraph Platform
+
+Expand your knowledge with these resources:
+
+- **[LangGraph Platform Concepts](../concepts/index.md#langgraph-platform)**: Understand the foundational concepts of the LangGraph Platform.
+- **[LangGraph Platform How-to Guides](../how-tos/index.md#langgraph-platform)**: Discover step-by-step guides to build and deploy applications.

docs/docs/tutorials/index.md

@@ -13,6 +13,7 @@ New to LangGraph or LLM app development? Read this material to get up and runnin
 - [LangGraph Quickstart](introduction.ipynb): Build a chatbot that can use tools and keep track of conversation history. Add human-in-the-loop capabilities and explore how time-travel works.
 - [LangGraph Server Quickstart](langgraph-platform/local-server.md): Launch a LangGraph server locally and interact with it using the REST API and LangGraph Studio Web UI.
 - [LangGraph Cloud QuickStart](../cloud/quick_start.md): Deploy a LangGraph app using LangGraph Cloud.
+- [LangGraph Template Quickstart](../concepts/template_applications.md): Quickly start building with LangGraph Platform using a template application.
 
 ## Use cases 🛠️
 

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

@@ -10,7 +10,7 @@ This is a quick start guide to help you get a LangGraph app up and running local
 ## Install the LangGraph CLI
 
 ```bash
-pip install "langgraph-cli[inmem]==0.1.58" python-dotenv
+pip install -U "langgraph-cli[inmem]" python-dotenv
 ```
 
 ## 🌱 Create a LangGraph App
@@ -250,4 +250,4 @@ Access detailed documentation for development and API usage:
 
 - **[LangGraph Server API Reference](../../cloud/reference/api/api_ref.html)**: Explore the LangGraph Server API documentation.  
 - **[Python SDK Reference](../../cloud/reference/sdk/python_sdk_ref.md)**: Explore the Python SDK API Reference.
-- **[JS/TS SDK Reference](../../cloud/reference/sdk/js_ts_sdk_ref.md)**: Explore the Python SDK API Reference.
+- **[JS/TS SDK Reference](../../cloud/reference/sdk/js_ts_sdk_ref.md)**: Explore the Python SDK API Reference.

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

@@ -57,6 +57,17 @@ MIGRATIONS = [
     PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
 );""",
     "ALTER TABLE checkpoint_blobs ALTER COLUMN blob DROP not null;",
+    """
+    """,
+    """
+    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);
+    """,
 ]
 
 SELECT_SQL = f"""

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

@@ -6,7 +6,6 @@ from typing import Any, Callable, Optional, Union, cast
 
 import orjson
 from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
-from psycopg.errors import UndefinedTable
 from psycopg.rows import DictRow, dict_row
 from psycopg_pool import AsyncConnectionPool
 
@@ -157,7 +156,12 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
         return results
 
     def batch(self, ops: Iterable[Op]) -> list[Result]:
-        return asyncio.run_coroutine_threadsafe(self.abatch(ops), self.loop).result()
+        futures = []
+        for op in ops:
+            fut = self._loop.create_future()
+            self._aqueue[fut] = op
+            futures.append(fut)
+        return [fut.result() for fut in asyncio.as_completed(futures)]
 
     @classmethod
     @asynccontextmanager
@@ -219,22 +223,19 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
         """
 
         async def _get_version(cur: AsyncCursor[DictRow], table: str) -> int:
-            try:
-                await cur.execute(f"SELECT v FROM {table} ORDER BY v DESC LIMIT 1")
-                row = await cur.fetchone()
-                if row is None:
-                    version = -1
-                else:
-                    version = row["v"]
-            except UndefinedTable:
-                version = -1
-                await cur.execute(
-                    f"""
-                    CREATE TABLE IF NOT EXISTS {table} (
-                        v INTEGER PRIMARY KEY
-                    )
-                """
+            await cur.execute(
+                f"""
+                CREATE TABLE IF NOT EXISTS {table} (
+                    v INTEGER PRIMARY KEY
                 )
+            """
+            )
+            await cur.execute(f"SELECT v FROM {table} ORDER BY v DESC LIMIT 1")
+            row = cast(dict, await cur.fetchone())
+            if row is None:
+                version = -1
+            else:
+                version = row["v"]
             return version
 
         async with self._cursor() as cur:

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

@@ -21,7 +21,6 @@ from typing import (
 
 import orjson
 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
@@ -73,7 +72,7 @@ CREATE TABLE IF NOT EXISTS store (
 """,
     """
 -- For faster lookups by prefix
-CREATE INDEX IF NOT EXISTS store_prefix_idx ON store USING btree (prefix text_pattern_ops);
+CREATE INDEX CONCURRENTLY IF NOT EXISTS store_prefix_idx ON store USING btree (prefix text_pattern_ops);
 """,
 ]
 
@@ -107,7 +106,7 @@ CREATE TABLE IF NOT EXISTS store_vectors (
     ),
     Migration(
         """
-CREATE INDEX IF NOT EXISTS store_vectors_embedding_idx ON store_vectors 
+CREATE INDEX CONCURRENTLY IF NOT EXISTS store_vectors_embedding_idx ON store_vectors 
     USING %(index_type)s (embedding %(ops)s)%(index_params)s;
 """,
         condition=lambda store: bool(
@@ -573,6 +572,7 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
 
         # Search by similarity
         results = store.search(("docs",), query="python programming")
+        ```
 
     Note:
         Semantic search is disabled by default. You can enable it by providing an `index` configuration
@@ -846,22 +846,19 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
         """
 
         def _get_version(cur: Cursor[dict[str, Any]], table: str) -> int:
-            try:
-                cur.execute(f"SELECT v FROM {table} ORDER BY v DESC LIMIT 1")
-                row = cast(dict, cur.fetchone())
-                if row is None:
-                    version = -1
-                else:
-                    version = row["v"]
-            except UndefinedTable:
-                version = -1
-                cur.execute(
-                    f"""
-                    CREATE TABLE IF NOT EXISTS {table} (
-                        v INTEGER PRIMARY KEY
-                    )
-                """
+            cur.execute(
+                f"""
+                CREATE TABLE IF NOT EXISTS {table} (
+                    v INTEGER PRIMARY KEY
                 )
+            """
+            )
+            cur.execute(f"SELECT v FROM {table} ORDER BY v DESC LIMIT 1")
+            row = cast(dict, cur.fetchone())
+            if row is None:
+                version = -1
+            else:
+                version = row["v"]
             return version
 
         with self._cursor() as cur:

libs/checkpoint-postgres/poetry.lock

@@ -13,24 +13,24 @@ files = [
 
 [[package]]
 name = "anyio"
-version = "4.6.2.post1"
+version = "4.7.0"
 description = "High level compatibility layer for multiple asynchronous event loop implementations"
 optional = false
 python-versions = ">=3.9"
 files = [
-    {file = "anyio-4.6.2.post1-py3-none-any.whl", hash = "sha256:6d170c36fba3bdd840c73d3868c1e777e33676a69c3a72cf0a0d5d6d8009b61d"},
-    {file = "anyio-4.6.2.post1.tar.gz", hash = "sha256:4c8bc31ccdb51c7f7bd251f51c609e038d63e34219b44aa86e47576389880b4c"},
+    {file = "anyio-4.7.0-py3-none-any.whl", hash = "sha256:ea60c3723ab42ba6fff7e8ccb0488c898ec538ff4df1f1d5e642c3601d07e352"},
+    {file = "anyio-4.7.0.tar.gz", hash = "sha256:2f834749c602966b7d456a7567cafcb309f96482b5081d14ac93ccd457f9dd48"},
 ]
 
 [package.dependencies]
 exceptiongroup = {version = ">=1.0.2", markers = "python_version < \"3.11\""}
 idna = ">=2.8"
 sniffio = ">=1.1"
-typing-extensions = {version = ">=4.1", markers = "python_version < \"3.11\""}
+typing_extensions = {version = ">=4.5", markers = "python_version < \"3.13\""}
 
 [package.extras]
-doc = ["Sphinx (>=7.4,<8.0)", "packaging", "sphinx-autodoc-typehints (>=1.2.0)", "sphinx-rtd-theme"]
-test = ["anyio[trio]", "coverage[toml] (>=7)", "exceptiongroup (>=1.2.0)", "hypothesis (>=4.0)", "psutil (>=5.9)", "pytest (>=7.0)", "pytest-mock (>=3.6.1)", "trustme", "truststore (>=0.9.1)", "uvloop (>=0.21.0b1)"]
+doc = ["Sphinx (>=7.4,<8.0)", "packaging", "sphinx-autodoc-typehints (>=1.2.0)", "sphinx_rtd_theme"]
+test = ["anyio[trio]", "coverage[toml] (>=7)", "exceptiongroup (>=1.2.0)", "hypothesis (>=4.0)", "psutil (>=5.9)", "pytest (>=7.0)", "pytest-mock (>=3.6.1)", "trustme", "truststore (>=0.9.1)", "uvloop (>=0.21)"]
 trio = ["trio (>=0.26.1)"]
 
 [[package]]
@@ -244,13 +244,13 @@ trio = ["trio (>=0.22.0,<1.0)"]
 
 [[package]]
 name = "httpx"
-version = "0.27.2"
+version = "0.28.0"
 description = "The next generation HTTP client."
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "httpx-0.27.2-py3-none-any.whl", hash = "sha256:7bb2708e112d8fdd7829cd4243970f0c223274051cb35ee80c03301ee29a3df0"},
-    {file = "httpx-0.27.2.tar.gz", hash = "sha256:f7c2be1d2f3c3c3160d441802406b206c2b76f5947b11115e6df10c6c65e66c2"},
+    {file = "httpx-0.28.0-py3-none-any.whl", hash = "sha256:dc0b419a0cfeb6e8b34e85167c0da2671206f5095f1baa9663d23bcfd6b535fc"},
+    {file = "httpx-0.28.0.tar.gz", hash = "sha256:0858d3bab51ba7e386637f22a61d8ccddaeec5f3fe4209da3a6168dbb91573e0"},
 ]
 
 [package.dependencies]
@@ -258,7 +258,6 @@ anyio = "*"
 certifi = "*"
 httpcore = "==1.*"
 idna = "*"
-sniffio = "*"
 
 [package.extras]
 brotli = ["brotli", "brotlicffi"]
@@ -342,7 +341,7 @@ typing-extensions = ">=4.7"
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "2.0.7"
+version = "2.0.8"
 description = "Library with base interfaces for LangGraph checkpoint savers."
 optional = false
 python-versions = "^3.9.0,<4.0"
@@ -741,13 +740,13 @@ typing-extensions = ">=4.6"
 
 [[package]]
 name = "pydantic"
-version = "2.10.2"
+version = "2.10.3"
 description = "Data validation using Python type hints"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "pydantic-2.10.2-py3-none-any.whl", hash = "sha256:cfb96e45951117c3024e6b67b25cdc33a3cb7b2fa62e239f7af1378358a1d99e"},
-    {file = "pydantic-2.10.2.tar.gz", hash = "sha256:2bc2d7f17232e0841cbba4641e65ba1eb6fafb3a08de3a091ff3ce14a197c4fa"},
+    {file = "pydantic-2.10.3-py3-none-any.whl", hash = "sha256:be04d85bbc7b65651c5f8e6b9976ed9c6f41782a55524cef079a34a0bb82144d"},
+    {file = "pydantic-2.10.3.tar.gz", hash = "sha256:cb5ac360ce894ceacd69c403187900a02c4b20b693a9dd1d643e1effab9eadf9"},
 ]
 
 [package.dependencies]

libs/checkpoint-postgres/pyproject.toml

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

libs/checkpoint-postgres/tests/test_async.py

@@ -63,9 +63,8 @@ async def _pipe_saver():
             prepare_threshold=0,
             row_factory=dict_row,
         ) as conn:
-            async with conn.pipeline() as pipe:
-                checkpointer = AsyncPostgresSaver(conn, pipe=pipe)
-                await checkpointer.setup()
+            checkpointer = AsyncPostgresSaver(conn)
+            await checkpointer.setup()
             async with conn.pipeline() as pipe:
                 checkpointer = AsyncPostgresSaver(conn, pipe=pipe)
                 yield checkpointer

libs/checkpoint-postgres/tests/test_async_store.py

@@ -1,8 +1,10 @@
 # type: ignore
+import asyncio
 import itertools
 import sys
 import uuid
 from collections.abc import AsyncIterator
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import asynccontextmanager
 from typing import Any, Optional
 
@@ -63,6 +65,96 @@ async def store(request) -> AsyncIterator[AsyncPostgresStore]:
             await conn.execute(f"DROP DATABASE {database}")
 
 
+async def test_large_batches(store: AsyncPostgresStore) -> None:
+    N = 1000
+    M = 10
+
+    with ThreadPoolExecutor(max_workers=10) as executor:
+        for m in range(M):
+            for i in range(N):
+                _ = [
+                    executor.submit(
+                        store.put,
+                        ("test", "foo", "bar", "baz", str(m % 2)),
+                        f"key{i}",
+                        value={"foo": "bar" + str(i)},
+                    ),
+                    executor.submit(
+                        store.get,
+                        ("test", "foo", "bar", "baz", str(m % 2)),
+                        f"key{i}",
+                    ),
+                    executor.submit(
+                        store.list_namespaces,
+                        prefix=None,
+                        max_depth=m + 1,
+                    ),
+                    executor.submit(
+                        store.search,
+                        ("test",),
+                    ),
+                    executor.submit(
+                        store.put,
+                        ("test", "foo", "bar", "baz", str(m % 2)),
+                        f"key{i}",
+                        value={"foo": "bar" + str(i)},
+                    ),
+                    executor.submit(
+                        store.put,
+                        ("test", "foo", "bar", "baz", str(m % 2)),
+                        f"key{i}",
+                        None,
+                    ),
+                ]
+
+
+async def test_large_batches_async(store: AsyncPostgresStore) -> None:
+    N = 1000
+    M = 10
+    coros = []
+    for m in range(M):
+        for i in range(N):
+            coros.append(
+                store.aput(
+                    ("test", "foo", "bar", "baz", str(m % 2)),
+                    f"key{i}",
+                    value={"foo": "bar" + str(i)},
+                )
+            )
+            coros.append(
+                store.aget(
+                    ("test", "foo", "bar", "baz", str(m % 2)),
+                    f"key{i}",
+                )
+            )
+            coros.append(
+                store.alist_namespaces(
+                    prefix=None,
+                    max_depth=m + 1,
+                )
+            )
+            coros.append(
+                store.asearch(
+                    ("test",),
+                )
+            )
+            coros.append(
+                store.aput(
+                    ("test", "foo", "bar", "baz", str(m % 2)),
+                    f"key{i}",
+                    value={"foo": "bar" + str(i)},
+                )
+            )
+            coros.append(
+                store.adelete(
+                    ("test", "foo", "bar", "baz", str(m % 2)),
+                    f"key{i}",
+                )
+            )
+
+    await asyncio.gather(*coros)
+
+
 async def test_abatch_order(store: AsyncPostgresStore) -> None:
     # Setup test data
     await store.aput(("test", "foo"), "key1", {"data": "value1"})

libs/checkpoint-postgres/tests/test_sync.py

@@ -57,9 +57,8 @@ def _pipe_saver():
             prepare_threshold=0,
             row_factory=dict_row,
         ) as conn:
-            with conn.pipeline() as pipe:
-                checkpointer = PostgresSaver(conn, pipe=pipe)
-                checkpointer.setup()
+            checkpointer = PostgresSaver(conn)
+            checkpointer.setup()
             with conn.pipeline() as pipe:
                 checkpointer = PostgresSaver(conn, pipe=pipe)
                 yield checkpointer

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

@@ -80,13 +80,16 @@ class Item:
 
     def dict(self) -> dict:
         return {
-            "value": self.value,
-            "key": self.key,
             "namespace": list(self.namespace),
+            "key": self.key,
+            "value": self.value,
             "created_at": self.created_at.isoformat(),
             "updated_at": self.updated_at.isoformat(),
         }
 
+    def __repr__(self) -> str:
+        return f"Item({', '.join(f'{k}={v!r}' for k, v in self.dict().items())})"
+
 
 class SearchItem(Item):
     """Represents an item returned from a search operation with additional metadata."""

libs/langgraph/langgraph/graph/graph.py

@@ -629,3 +629,10 @@ class CompiledGraph(Pregel):
                     add_edge(key, end, conditional=True)
 
         return graph
+
+    def _repr_mimebundle_(self, **kwargs: Any) -> dict[str, Any]:
+        """Mime bundle used by Jupyter to display the graph"""
+        return {
+            "text/plain": repr(self),
+            "image/png": self.get_graph().draw_mermaid_png(),
+        }

libs/langgraph/langgraph/graph/state.py

@@ -51,7 +51,11 @@ from langgraph.managed.base import (
     is_writable_managed_value,
 )
 from langgraph.pregel.read import ChannelRead, PregelNode
-from langgraph.pregel.write import SKIP_WRITE, ChannelWrite, ChannelWriteEntry
+from langgraph.pregel.write import (
+    ChannelWrite,
+    ChannelWriteEntry,
+    ChannelWriteTupleEntry,
+)
 from langgraph.store.base import BaseStore
 from langgraph.types import All, Checkpointer, Command, RetryPolicy
 from langgraph.utils.fields import get_field_default
@@ -608,33 +612,53 @@ class CompiledStateGraph(CompiledGraph):
                 if is_writable_managed_value(v)
             ]
 
-        def _get_root(input: Any) -> Any:
-            if isinstance(input, Command):
-                if input.graph == Command.PARENT:
-                    return SKIP_WRITE
-                return input.update
-            else:
-                return input
-
-        # to avoid name collision below
-        node_key = key
-
-        def _get_state_key(input: Union[None, dict, Any], *, key: str) -> Any:
-            if input is None:
-                return SKIP_WRITE
-            elif isinstance(input, dict):
-                if all(k not in output_keys for k in input):
-                    raise InvalidUpdateError(
-                        f"Expected node {node_key} to update at least one of {output_keys}, got {input}"
-                    )
-                return input.get(key, SKIP_WRITE)
+        def _get_root(input: Any) -> Optional[Sequence[tuple[str, Any]]]:
+            if (
+                isinstance(input, (list, tuple))
+                and input
+                and all(isinstance(i, Command) for i in input)
+            ):
+                updates: list[tuple[str, Any]] = []
+                for i in input:
+                    if i.graph == Command.PARENT:
+                        continue
+                    updates.extend(i._update_as_tuples())
+                return updates
             elif isinstance(input, Command):
                 if input.graph == Command.PARENT:
-                    return SKIP_WRITE
-                return _get_state_key(input.update, key=key)
+                    return ()
+                return input._update_as_tuples()
+            elif input is not None:
+                return [("__root__", input)]
+
+        def _get_updates(
+            input: Union[None, dict, Any],
+        ) -> Optional[Sequence[tuple[str, Any]]]:
+            if input is None:
+                return None
+            elif isinstance(input, dict):
+                return [(k, v) for k, v in input.items() if k in output_keys]
+            elif isinstance(input, Command):
+                if input.graph == Command.PARENT:
+                    return None
+                return input._update_as_tuples()
+            elif (
+                isinstance(input, (list, tuple))
+                and input
+                and all(isinstance(i, Command) for i in input)
+            ):
+                updates: list[tuple[str, Any]] = []
+                for i in input:
+                    if i.graph == Command.PARENT:
+                        continue
+                    updates.extend(i._update_as_tuples())
+                return updates
             elif get_type_hints(type(input)):
-                value = getattr(input, key, SKIP_WRITE)
-                return value if value is not None else SKIP_WRITE
+                return [
+                    (k, getattr(input, k))
+                    for k in output_keys
+                    if getattr(input, k, None) is not None
+                ]
             else:
                 msg = create_error_message(
                     message=f"Expected dict, got {input}",
@@ -643,14 +667,11 @@ class CompiledStateGraph(CompiledGraph):
                 raise InvalidUpdateError(msg)
 
         # state updaters
-        write_entries = (
-            [ChannelWriteEntry("__root__", skip_none=True, mapper=_get_root)]
-            if output_keys == ["__root__"]
-            else [
-                ChannelWriteEntry(key, mapper=partial(_get_state_key, key=key))
-                for key in output_keys
-            ]
-        )
+        write_entries: list[Union[ChannelWriteEntry, ChannelWriteTupleEntry]] = [
+            ChannelWriteTupleEntry(
+                mapper=_get_root if output_keys == ["__root__"] else _get_updates
+            )
+        ]
 
         # add node and output channel
         if key == START:
@@ -685,7 +706,7 @@ class CompiledStateGraph(CompiledGraph):
                 writers=[
                     # publish to this channel and state keys
                     ChannelWrite(
-                        [ChannelWriteEntry(key, key)] + write_entries,
+                        write_entries + [ChannelWriteEntry(key, key)],
                         tags=[TAG_HIDDEN],
                     ),
                 ],
@@ -811,34 +832,54 @@ def _coerce_state(schema: Type[Any], input: dict[str, Any]) -> dict[str, Any]:
 def _control_branch(value: Any) -> Sequence[Union[str, Send]]:
     if isinstance(value, Send):
         return [value]
-    if not isinstance(value, Command):
-        return EMPTY_SEQ
-    if value.graph == Command.PARENT:
-        raise ParentCommand(value)
-    rtn: list[Union[str, Send]] = []
-    if isinstance(value.goto, Send):
-        rtn.append(value.goto)
-    elif isinstance(value.goto, str):
-        rtn.append(value.goto)
+    commands: list[Command] = []
+    if isinstance(value, Command):
+        commands.append(value)
+    elif (
+        isinstance(value, (list, tuple))
+        and value
+        and all(isinstance(i, Command) for i in value)
+    ):
+        commands.extend(value)
     else:
-        rtn.extend(value.goto)
+        return EMPTY_SEQ
+    rtn: list[Union[str, Send]] = []
+    for command in commands:
+        if command.graph == Command.PARENT:
+            raise ParentCommand(command)
+        if isinstance(command.goto, Send):
+            rtn.append(command.goto)
+        elif isinstance(command.goto, str):
+            rtn.append(command.goto)
+        else:
+            rtn.extend(command.goto)
     return rtn
 
 
 async def _acontrol_branch(value: Any) -> Sequence[Union[str, Send]]:
     if isinstance(value, Send):
         return [value]
-    if not isinstance(value, Command):
-        return EMPTY_SEQ
-    if value.graph == Command.PARENT:
-        raise ParentCommand(value)
-    rtn: list[Union[str, Send]] = []
-    if isinstance(value.goto, Send):
-        rtn.append(value.goto)
-    elif isinstance(value.goto, str):
-        rtn.append(value.goto)
+    commands: list[Command] = []
+    if isinstance(value, Command):
+        commands.append(value)
+    elif (
+        isinstance(value, (list, tuple))
+        and value
+        and all(isinstance(i, Command) for i in value)
+    ):
+        commands.extend(value)
     else:
-        rtn.extend(value.goto)
+        return EMPTY_SEQ
+    rtn: list[Union[str, Send]] = []
+    for command in commands:
+        if command.graph == Command.PARENT:
+            raise ParentCommand(command)
+        if isinstance(command.goto, Send):
+            rtn.append(command.goto)
+        elif isinstance(command.goto, str):
+            rtn.append(command.goto)
+        else:
+            rtn.extend(command.goto)
     return rtn
 
 

libs/langgraph/langgraph/prebuilt/tool_node.py

@@ -1,15 +1,10 @@
-from __future__ import annotations
-
 import asyncio
 import inspect
 import json
 from copy import copy
 from typing import (
-    TYPE_CHECKING,
     Any,
     Callable,
-    Dict,
-    List,
     Literal,
     Optional,
     Sequence,
@@ -35,22 +30,20 @@ from langchain_core.runnables.utils import Input
 from langchain_core.tools import BaseTool, InjectedToolArg
 from langchain_core.tools import tool as create_tool
 from langchain_core.tools.base import get_all_basemodel_annotations
+from pydantic import BaseModel
 from typing_extensions import Annotated, get_args, get_origin
 
 from langgraph.errors import GraphBubbleUp
 from langgraph.store.base import BaseStore
 from langgraph.utils.runnable import RunnableCallable
 
-if TYPE_CHECKING:
-    from pydantic import BaseModel
-
 INVALID_TOOL_NAME_ERROR_TEMPLATE = (
     "Error: {requested_tool} is not a valid tool, try one of [{available_tools}]."
 )
 TOOL_CALL_ERROR_TEMPLATE = "Error: {error}\n Please fix your mistakes."
 
 
-def msg_content_output(output: Any) -> str | List[dict]:
+def msg_content_output(output: Any) -> Union[str, list[dict]]:
     recognized_content_block_types = ("image", "image_url", "text", "json")
     if isinstance(output, str):
         return output
@@ -95,7 +88,7 @@ def _handle_tool_error(
     return content
 
 
-def _infer_handled_types(handler: Callable[..., str]) -> tuple[type[Exception]]:
+def _infer_handled_types(handler: Callable[..., str]) -> tuple[type[Exception], ...]:
     sig = inspect.signature(handler)
     params = list(sig.parameters.values())
     if params:
@@ -194,9 +187,9 @@ class ToolNode(RunnableCallable):
         messages_key: str = "messages",
     ) -> None:
         super().__init__(self._func, self._afunc, name=name, tags=tags, trace=False)
-        self.tools_by_name: Dict[str, BaseTool] = {}
-        self.tool_to_state_args: Dict[str, Dict[str, Optional[str]]] = {}
-        self.tool_to_store_arg: Dict[str, Optional[str]] = {}
+        self.tools_by_name: dict[str, BaseTool] = {}
+        self.tool_to_state_args: dict[str, dict[str, Optional[str]]] = {}
+        self.tool_to_store_arg: dict[str, Optional[str]] = {}
         self.handle_tool_errors = handle_tool_errors
         self.messages_key = messages_key
         for tool_ in tools:
@@ -346,7 +339,7 @@ class ToolNode(RunnableCallable):
             BaseModel,
         ],
         store: BaseStore,
-    ) -> Tuple[List[ToolCall], Literal["list", "dict"]]:
+    ) -> Tuple[list[ToolCall], Literal["list", "dict"]]:
         if isinstance(input, list):
             output_type = "list"
             message: AnyMessage = input[-1]
@@ -656,9 +649,9 @@ def _is_injection(
     return False
 
 
-def _get_state_args(tool: BaseTool) -> Dict[str, Optional[str]]:
+def _get_state_args(tool: BaseTool) -> dict[str, Optional[str]]:
     full_schema = tool.get_input_schema()
-    tool_args_to_state_fields: Dict = {}
+    tool_args_to_state_fields: dict = {}
 
     for name, type_ in get_all_basemodel_annotations(full_schema).items():
         injections = [

libs/langgraph/langgraph/pregel/io.py

@@ -95,11 +95,7 @@ def map_command(
         else:
             yield (NULL_TASK_ID, RESUME, cmd.resume)
     if cmd.update:
-        if not isinstance(cmd.update, dict):
-            raise TypeError(
-                f"Expected cmd.update to be a dict mapping channel names to update values, got {type(cmd.update).__name__}"
-            )
-        for k, v in cmd.update.items():
+        for k, v in cmd._update_as_tuples():
             yield (NULL_TASK_ID, k, v)
 
 

libs/langgraph/langgraph/pregel/write.py

@@ -36,31 +36,40 @@ class ChannelWriteEntry(NamedTuple):
     """Function to transform the value before writing."""
 
 
+class ChannelWriteTupleEntry(NamedTuple):
+    mapper: Callable[[Any], Optional[Sequence[tuple[str, Any]]]]
+    """Function to extract tuples from value."""
+    value: Any = PASSTHROUGH
+    """Value to write, or PASSTHROUGH to use the input."""
+
+
 class ChannelWrite(RunnableCallable):
-    """Implements th logic for sending writes to CONFIG_KEY_SEND.
+    """Implements the logic for sending writes to CONFIG_KEY_SEND.
     Can be used as a runnable or as a static method to call imperatively."""
 
-    writes: list[Union[ChannelWriteEntry, Send]]
+    writes: list[Union[ChannelWriteEntry, ChannelWriteTupleEntry, Send]]
     """Sequence of write entries or Send objects to write."""
     require_at_least_one_of: Optional[Sequence[str]]
     """If defined, at least one of these channels must be written to."""
 
     def __init__(
         self,
-        writes: Sequence[Union[ChannelWriteEntry, Send]],
+        writes: Sequence[Union[ChannelWriteEntry, ChannelWriteTupleEntry, Send]],
         *,
         tags: Optional[Sequence[str]] = None,
         require_at_least_one_of: Optional[Sequence[str]] = None,
     ):
         super().__init__(func=self._write, afunc=self._awrite, name=None, tags=tags)
-        self.writes = cast(list[Union[ChannelWriteEntry, Send]], writes)
+        self.writes = cast(
+            list[Union[ChannelWriteEntry, ChannelWriteTupleEntry, Send]], writes
+        )
         self.require_at_least_one_of = require_at_least_one_of
 
     def get_name(
         self, suffix: Optional[str] = None, *, name: Optional[str] = None
     ) -> str:
         if not name:
-            name = f"ChannelWrite<{','.join(w.channel if isinstance(w, ChannelWriteEntry) else w.node for w in self.writes)}>"
+            name = f"ChannelWrite<{','.join(w.channel if isinstance(w, ChannelWriteEntry) else '...' if isinstance(w, ChannelWriteTupleEntry) else w.node for w in self.writes)}>"
         return super().get_name(suffix, name=name)
 
     @property
@@ -79,6 +88,8 @@ class ChannelWrite(RunnableCallable):
         writes = [
             ChannelWriteEntry(write.channel, input, write.skip_none, write.mapper)
             if isinstance(write, ChannelWriteEntry) and write.value is PASSTHROUGH
+            else ChannelWriteTupleEntry(write.mapper, input)
+            if isinstance(write, ChannelWriteTupleEntry) and write.value is PASSTHROUGH
             else write
             for write in self.writes
         ]
@@ -93,6 +104,8 @@ class ChannelWrite(RunnableCallable):
         writes = [
             ChannelWriteEntry(write.channel, input, write.skip_none, write.mapper)
             if isinstance(write, ChannelWriteEntry) and write.value is PASSTHROUGH
+            else ChannelWriteTupleEntry(write.mapper, input)
+            if isinstance(write, ChannelWriteTupleEntry) and write.value is PASSTHROUGH
             else write
             for write in self.writes
         ]
@@ -106,7 +119,7 @@ class ChannelWrite(RunnableCallable):
     @staticmethod
     def do_write(
         config: RunnableConfig,
-        writes: Sequence[Union[ChannelWriteEntry, Send]],
+        writes: Sequence[Union[ChannelWriteEntry, ChannelWriteTupleEntry, Send]],
         require_at_least_one_of: Optional[Sequence[str]] = None,
     ) -> None:
         # validate
@@ -118,32 +131,34 @@ class ChannelWrite(RunnableCallable):
                     )
                 if w.value is PASSTHROUGH:
                     raise InvalidUpdateError("PASSTHROUGH value must be replaced")
-        # split packets and entries
-        sends = [
-            (PUSH if FF_SEND_V2 else TASKS, packet)
-            for packet in writes
-            if isinstance(packet, Send)
-        ]
-        entries = [write for write in writes if isinstance(write, ChannelWriteEntry)]
-        # process entries into values
-        values = [
-            write.mapper(write.value) if write.mapper is not None else write.value
-            for write in entries
-        ]
-        values = [
-            (write.channel, val)
-            for val, write in zip(values, entries)
-            if not write.skip_none or val is not None
-        ]
-        # filter out SKIP_WRITE values
-        filtered = [(chan, val) for chan, val in values if val is not SKIP_WRITE]
+            if isinstance(w, ChannelWriteTupleEntry):
+                if w.value is PASSTHROUGH:
+                    raise InvalidUpdateError("PASSTHROUGH value must be replaced")
+        # assemble writes
+        tuples: list[tuple[str, Any]] = []
+        for w in writes:
+            if isinstance(w, Send):
+                tuples.append((PUSH if FF_SEND_V2 else TASKS, w))
+            elif isinstance(w, ChannelWriteTupleEntry):
+                if ww := w.mapper(w.value):
+                    tuples.extend(ww)
+            elif isinstance(w, ChannelWriteEntry):
+                value = w.mapper(w.value) if w.mapper is not None else w.value
+                if value is SKIP_WRITE:
+                    continue
+                if w.skip_none and value is None:
+                    continue
+                tuples.append((w.channel, value))
+            else:
+                raise ValueError(f"Invalid write entry: {w}")
+        # assert required channels
         if require_at_least_one_of is not None:
-            if not {chan for chan, _ in filtered} & set(require_at_least_one_of):
+            if not {chan for chan, _ in tuples} & set(require_at_least_one_of):
                 raise InvalidUpdateError(
                     f"Must write to at least one of {require_at_least_one_of}"
                 )
         write: TYPE_SEND = config[CONF][CONFIG_KEY_SEND]
-        write(sends + filtered)
+        write(tuples)
 
     @staticmethod
     def is_writer(runnable: Runnable) -> bool:

libs/langgraph/langgraph/types.py

@@ -251,7 +251,7 @@ class Command(Generic[N]):
         graph: graph to send the command to. Supported values are:
 
             - None: the current graph (default)
-            - GraphCommand.PARENT: closest parent graph
+            - Command.PARENT: closest parent graph
         update: update to apply to the graph's state.
         resume: value to resume execution with. To be used together with [`interrupt()`][langgraph.types.interrupt].
         goto: can be one of the following:
@@ -263,7 +263,7 @@ class Command(Generic[N]):
     """
 
     graph: Optional[str] = None
-    update: Optional[dict[str, Any]] = None
+    update: Union[dict[str, Any], Sequence[tuple[str, Any]]] = ()
     resume: Optional[Union[Any, dict[str, Any]]] = None
     goto: Union[Send, Sequence[Union[Send, str]], str] = ()
 
@@ -276,6 +276,17 @@ class Command(Generic[N]):
         )
         return f"Command({contents})"
 
+    def _update_as_tuples(self) -> Sequence[tuple[str, Any]]:
+        if isinstance(self.update, dict):
+            return list(self.update.items())
+        elif isinstance(self.update, (list, tuple)) and all(
+            isinstance(t, tuple) and len(t) == 2 and isinstance(t[0], str)
+            for t in self.update
+        ):
+            return self.update
+        else:
+            return [("__root__", self.update)]
+
     PARENT: ClassVar[Literal["__parent__"]] = "__parent__"
 
 

libs/langgraph/tests/test_pregel.py

@@ -220,18 +220,6 @@ def test_graph_validation() -> None:
     class State(TypedDict):
         hello: str
 
-    def node_a(state: State) -> State:
-        # typo
-        return {"hell": "world"}
-
-    builder = StateGraph(State)
-    builder.add_node("a", node_a)
-    builder.set_entry_point("a")
-    builder.set_finish_point("a")
-    graph = builder.compile()
-    with pytest.raises(InvalidUpdateError):
-        graph.invoke({"hello": "there"})
-
     graph = StateGraph(State)
     graph.add_node("start", lambda x: x)
     graph.add_edge("__start__", "start")
@@ -1919,7 +1907,7 @@ def test_send_sequences() -> None:
                 else ["|".join((self.name, str(state)))]
             )
             if isinstance(state, Command):
-                return replace(state, update=update)
+                return [state, Command(update=update)]
             else:
                 return update
 

libs/sdk-js/src/types.ts

@@ -40,9 +40,11 @@ export interface Command {
   resume?: unknown;
 
   /**
-   * A single, or array of `Send` commands to trigger nodes.
+   * Determine the next node to navigate to. Can be one of the following:
+   * - Name(s) of the node names to navigate to next.
+   * - `Send` command(s) to execute node(s) with provided input.
    */
-  send?: Send | Send[];
+  goto?: Send | Send[] | string | string[];
 }
 
 interface RunsInvokePayload {