feat: [CLI] Add support for building deps with uv (#4995)

blank issues

.github/ISSUE_TEMPLATE/config.yml

@@ -1,4 +1,4 @@
-blank_issues_enabled: false
+blank_issues_enabled: true
 version: 2.1
 contact_links:
   - name: 🤔 Question or Problem

.github/dependabot.yml

@@ -0,0 +1,11 @@
+# Please see the documentation for all configuration options:
+# https://docs.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
+# and
+# https://docs.github.com/code-security/dependabot/dependabot-version-updates/configuration-options-for-the-dependabot.yml-file
+
+version: 2
+updates:
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "weekly"

.github/workflows/_lint.yml

@@ -49,7 +49,7 @@ jobs:
 
       - name: Get .mypy_cache to speed up mypy
         if: steps.changed-files.outputs.all
-        uses: actions/cache@v3
+        uses: actions/cache@v4
         env:
           SEGMENT_DOWNLOAD_TIMEOUT_MIN: "2"
         with:
@@ -75,7 +75,7 @@ jobs:
 
       - name: Get .mypy_cache_test to speed up mypy
         if: steps.changed-files.outputs.all
-        uses: actions/cache@v3
+        uses: actions/cache@v4
         env:
           SEGMENT_DOWNLOAD_TIMEOUT_MIN: "2"
         with:

.github/workflows/_test_scheduler_kafka.yml

@@ -1,52 +0,0 @@
-name: test
-
-on:
-  workflow_call:
-
-jobs:
-  build:
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        python-version:
-          - "3.11"
-          - "3.12"
-
-    defaults:
-      run:
-        working-directory: libs/scheduler-kafka
-    name: "test #${{ matrix.python-version }}"
-    steps:
-      - uses: actions/checkout@v4
-      - name: Set up Python ${{ matrix.python-version }}
-        uses: astral-sh/setup-uv@v6
-        with:
-          python-version: ${{ matrix.python-version }}
-          enable-cache: true
-          cache-suffix: "test-scheduler-kafka"
-      - name: Login to Docker Hub
-        uses: docker/login-action@v3
-        if: ${{ !github.event.pull_request.head.repo.fork }}
-        with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_RO_TOKEN }}
-
-      - name: Install dependencies
-        shell: bash
-        run: uv sync --frozen --group dev
-
-      - name: Run tests
-        shell: bash
-        run: make test
-
-      - name: Ensure the tests did not create any additional files
-        shell: bash
-        run: |
-          set -eu
-
-          STATUS="$(git status)"
-          echo "$STATUS"
-
-          # grep will exit non-zero if the target message isn't found,
-          # and `set -e` above will cause the step to fail.
-          echo "$STATUS" | grep 'nothing to commit, working tree clean'

.github/workflows/ci.yml

@@ -35,7 +35,6 @@ jobs:
               - 'libs/checkpoint/**'
               - 'libs/checkpoint-sqlite/**'
               - 'libs/checkpoint-postgres/**'
-              - 'libs/scheduler-kafka/**'
               - 'libs/prebuilt/**'
             sdk-js:
               - 'libs/sdk-js/**'
@@ -53,7 +52,7 @@ jobs:
             "libs/checkpoint",
             "libs/checkpoint-sqlite",
             "libs/checkpoint-postgres",
-            "libs/scheduler-kafka",
+
             "libs/prebuilt",
           ]
     if: needs.changes.outputs.python == 'true'
@@ -89,14 +88,6 @@ jobs:
     uses: ./.github/workflows/_test_langgraph.yml
     secrets: inherit
 
-  # NOTE: we're testing scheduler-kafka separately because it requires a different matrix
-  test-scheduler-kafka:
-    needs: changes
-    if: needs.changes.outputs.python == 'true'
-    name: "cd libs/scheduler-kafka"
-    uses: ./.github/workflows/_test_scheduler_kafka.yml
-    secrets: inherit
-
   check-sdk-methods:
     needs: changes
     if: needs.changes.outputs.python == 'true'
@@ -166,9 +157,9 @@ jobs:
       run:
         working-directory: ${{ matrix.working-directory }}
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Setup Node.js (LTS)
-        uses: actions/setup-node@v3
+        uses: actions/setup-node@v4
         with:
           node-version: "20"
           cache: "yarn"
@@ -192,9 +183,9 @@ jobs:
       run:
         working-directory: ${{ matrix.working-directory }}
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Setup Node.js (LTS)
-        uses: actions/setup-node@v3
+        uses: actions/setup-node@v4
         with:
           node-version: "20"
           cache: "yarn"
@@ -212,7 +203,6 @@ jobs:
         lint-js,
         test,
         test-langgraph,
-        test-scheduler-kafka,
         check-sdk-methods,
         check-schema,
         integration-test,

.github/workflows/deploy_docs.yml

@@ -4,9 +4,11 @@ on:
   push:
     branches:
       - main
+      - v0
   pull_request:
     branches:
       - main
+      - v0
   workflow_dispatch:
 
 permissions:
@@ -82,9 +84,9 @@ jobs:
         run: make llms-text
       - name: Build site
         run: |
-          # If this is main branch, then we want to download stats. we do this 
+          # If this is v0 branch, then we want to download stats. we do this 
           # with the env variable DOWNLOAD_STATS=true
-          if [ "${{ github.ref }}" == "refs/heads/main" ]; then
+          if [ "${{ github.ref }}" == "refs/heads/v0" ]; then
             DOWNLOAD_STATS=true make build-docs
           else
             make build-docs
@@ -144,8 +146,8 @@ jobs:
           fi
 
       - name: Configure GitHub Pages
-        if: github.ref == 'refs/heads/main'
-        uses: actions/configure-pages@v4
+        if: github.ref == 'refs/heads/v0'
+        uses: actions/configure-pages@v5
 
       - name: Upload Pages Artifact
         # if: github.ref == 'refs/heads/main'
@@ -154,6 +156,6 @@ jobs:
           path: ./docs/site/
 
       - name: Deploy to GitHub Pages
-        if: github.ref == 'refs/heads/main'
+        if: github.ref == 'refs/heads/v0'
         id: deployment
         uses: actions/deploy-pages@v4

.github/workflows/release_js.yml

@@ -22,7 +22,7 @@ jobs:
       - uses: actions/checkout@v4
       # JS Build
       - name: Use Node.js
-        uses: actions/setup-node@v3
+        uses: actions/setup-node@v4
         with:
           node-version: "20"
           cache: "yarn"

.gitignore

@@ -181,3 +181,4 @@ Chinook.db
 .vercel
 .turbo
 .editorconfig
+.scratch

CONTRIBUTING.md

@@ -109,7 +109,7 @@ Here are some high-level tips on writing a good how-to guide:
 LangGraph's conceptual guides fall under the **Explanation** quadrant of Diataxis. They should cover LangChain terms and concepts
 in a more abstract way than how-to guides or tutorials, and should be geared towards curious users interested in
 gaining a deeper understanding of the framework. Try to avoid excessively large code examples. The goal here is to
-impart perspective to the user rather than to finish a practical project. These guides should cover **why** things work they way they do.
+impart perspective to the user rather than to finish a practical project. These guides should cover **why** things work the way they do.
 
 
 To quote the Diataxis website:

Makefile

@@ -0,0 +1,58 @@
+# Define the directories containing projects
+LIBS_DIRS := $(wildcard libs/*)
+
+# Default target
+.PHONY: all
+all: lint format lock test
+
+# Install dependencies for all projects
+.PHONY: install
+install:
+	@echo "Creating virtual environment..."
+	@uv venv
+	@for dir in $(LIBS_DIRS); do \
+		if [ -f $$dir/pyproject.toml ]; then \
+			echo "Installing dependencies for $$dir"; \
+			uv pip install -e $$dir; \
+		fi; \
+	done
+
+# Lint all projects
+.PHONY: lint
+lint:
+	@for dir in $(LIBS_DIRS); do \
+		if [ -f $$dir/Makefile ]; then \
+			echo "Running lint in $$dir"; \
+			$(MAKE) -C $$dir lint; \
+		fi; \
+	done
+
+# Format all projects
+.PHONY: format
+format:
+	@for dir in $(LIBS_DIRS); do \
+		if [ -f $$dir/Makefile ]; then \
+			echo "Running format in $$dir"; \
+			$(MAKE) -C $$dir format; \
+		fi; \
+	done
+
+# Lock all projects
+.PHONY: lock
+lock:
+	@for dir in $(LIBS_DIRS); do \
+		if [ -f $$dir/Makefile ]; then \
+			echo "Running lock in $$dir"; \
+			(cd $$dir && uv lock); \
+		fi; \
+	done
+
+# Test all projects
+.PHONY: test
+test:
+	@for dir in $(LIBS_DIRS); do \
+		if [ -f $$dir/Makefile ]; then \
+			echo "Running test in $$dir"; \
+			$(MAKE) -C $$dir test; \
+		fi; \
+	done

docs/_scripts/notebook_hooks.py

@@ -1,9 +1,16 @@
+"""mkdocs hooks for adding custom logic to documentation pipeline.
+
+Lifecycle events: https://www.mkdocs.org/dev-guide/plugins/#events
+"""
+
 import logging
 import os
 import posixpath
 import re
 from typing import Any, Dict
 
+from bs4 import BeautifulSoup
+from mkdocs.config.defaults import MkDocsConfig
 from mkdocs.structure.files import Files, File
 from mkdocs.structure.pages import Page
 
@@ -101,8 +108,7 @@ REDIRECT_MAP = {
     "how-tos/deploy-self-hosted.md": "cloud/deployment/self_hosted_data_plane.md",
     "concepts/self_hosted.md": "concepts/langgraph_self_hosted_data_plane.md",
     # assistant redirects
-    "cloud/how-tos/assistant_versioning.md": "cloud/how-tos/configuration_cloud.md"
-
+    "cloud/how-tos/assistant_versioning.md": "cloud/how-tos/configuration_cloud.md",
 }
 
 
@@ -292,7 +298,7 @@ Redirecting...
 """
 
 
-def write_html(site_dir, old_path, new_path):
+def _write_html(site_dir, old_path, new_path):
     """Write an HTML file in the site_dir with a meta redirect to the new page"""
     # Determine all relevant paths
     old_path_abs = os.path.join(site_dir, old_path)
@@ -308,6 +314,52 @@ def write_html(site_dir, old_path, new_path):
         f.write(content)
 
 
+def _inject_gtm(html: str) -> str:
+    """Inject Google Tag Manager code into the HTML.
+
+    Code to inject Google Tag Manager noscript tag immediately after <body>.
+
+    This is done via hooks rather than via a template because the MkDocs material
+    theme does not seem to allow placing the code immediately after the <body> tag
+    without modifying the template files directly.
+
+    Args:
+        html: The HTML content to modify.
+
+    Returns:
+        The modified HTML content with GTM code injected.
+    """
+    # Code was copied from Google Tag Manager setup instructions.
+    gtm_code = """
+<!-- Google Tag Manager (noscript) -->
+<noscript><iframe src="https://www.googletagmanager.com/ns.html?id=GTM-T35S4S46"
+height="0" width="0" style="display:none;visibility:hidden"></iframe></noscript>
+<!-- End Google Tag Manager (noscript) -->
+"""
+    soup = BeautifulSoup(html, "html.parser")
+    body = soup.body
+    if body:
+        # Insert the GTM code as raw HTML at the top of <body>
+        body.insert(0, BeautifulSoup(gtm_code, "html.parser"))
+        return str(soup)
+    else:
+        return html  # fallback if no <body> found
+
+
+def on_post_page(output: str, page: Page, config: MkDocsConfig) -> str:
+    """Inject Google Tag Manager noscript tag immediately after <body>.
+
+    Args:
+        output: The HTML output of the page.
+        page: The page instance.
+        config: The MkDocs configuration object.
+
+    Returns:
+        modified HTML output with GTM code injected.
+    """
+    return _inject_gtm(output)
+
+
 # Create HTML files for redirects after site dir has been built
 def on_post_build(config):
     use_directory_urls = config.get("use_directory_urls")
@@ -324,4 +376,4 @@ def on_post_build(config):
             + hash
             + suffix
         )
-        write_html(config["site_dir"], old_html_path, new_html_path)
+        _write_html(config["site_dir"], old_html_path, new_html_path)

docs/docs/agents/mcp.md

@@ -38,7 +38,7 @@ client = MultiServerMCPClient(
             "transport": "stdio",
         },
         "weather": {
-            # Ensure your start your weather server on port 8000
+            # Ensure you start your weather server on port 8000
             "url": "http://localhost:8000/mcp",
             "transport": "streamable_http",
         }

docs/docs/agents/memory.md

@@ -88,7 +88,7 @@ ny_response = agent.invoke(
 
 When the agent is invoked the second time with the same `thread_id`, the original message history from the first conversation is automatically included, allowing the agent to infer that the user is asking specifically about the **weather** in New York.
 
-!!! Note "LangGraph Platform providers a production-ready checkpointer"
+!!! Note "LangGraph Platform provides a production-ready checkpointer"
 
     If you're using [LangGraph Platform](./deployment.md), during deployment your checkpointer will be automatically configured to use a production-ready database.
 

docs/docs/agents/multi-agent.md

@@ -9,7 +9,7 @@ hide:
 
 # Multi-agent
 
-A single agent might struggle if it needs to specialize in multiple domains or manage many tools. To tackle this, you can break your agent into smaller, independent agents and composing them into a [multi-agent system](../concepts/multi_agent.md).
+A single agent might struggle if it needs to specialize in multiple domains or manage many tools. To tackle this, you can break your agent into smaller, independent agents and compose them into a [multi-agent system](../concepts/multi_agent.md).
 
 In multi-agent systems, agents need to communicate between each other. They do so via [handoffs](#handoffs) — a primitive that describes which agent to hand control to and the payload to send to that agent.
 

docs/docs/cloud/deployment/custom_docker.md

@@ -16,4 +16,4 @@ Users can add an array of additional lines to add to the Dockerfile following th
 }
 ```
 
-This would install the system packages required to use Pillow if we were working with `jpeq` or `png` image formats. 
+This would install the system packages required to use Pillow if we were working with `jpeg` or `png` image formats. 

docs/docs/cloud/deployment/graph_rebuild.md

@@ -20,7 +20,7 @@ my-app/
 |-- openai_agent.py     # code for your graph
 ```
 
-where the graph is defined in `openai_agent.py`. 
+where the graph is defined in `openai_agent.py`.
 
 ### No rebuild
 
@@ -28,11 +28,11 @@ In the standard LangGraph API configuration, the server uses the compiled graph
 
 ```python
 from langchain_openai import ChatOpenAI
-from langgraph.graph import END, START, MessageGraph
+from langgraph.graph import END, START, StateGraph, MessagesState
 
 model = ChatOpenAI(temperature=0)
 
-graph_workflow = MessageGraph()
+graph_workflow = StateGraph(MessagesState)
 
 graph_workflow.add_node("agent", model)
 graph_workflow.add_edge("agent", END)
@@ -61,7 +61,7 @@ To make your graph rebuild on each new run with custom configuration, you need t
 from typing import Annotated
 from typing_extensions import TypedDict
 from langchain_openai import ChatOpenAI
-from langgraph.graph import END, START, MessageGraph
+from langgraph.graph import END, START
 from langgraph.graph.state import StateGraph
 from langgraph.graph.message import add_messages
 from langgraph.prebuilt import ToolNode
@@ -144,4 +144,4 @@ Finally, you need to specify the path to your graph-making function (`make_graph
 }
 ```
 
-See more info on LangGraph API configuration file [here](../reference/cli.md#configuration-file)
+See more info on LangGraph API configuration file [here](../reference/cli.md#configuration-file)

docs/docs/cloud/deployment/self_hosted_control_plane.md

@@ -41,7 +41,8 @@ Before deploying, review the [conceptual guide for the Self-Hosted Control Plane
           pullPolicy: IfNotPresent
           tag: "aa9dff4"
 
-1. In your `values.yaml` file, enable the `langgraphPlatform` option. Note that you must also have a valid ingress setup:
+1. In your `langsmith_config.yaml` file, enable the `langgraphPlatform` option. Note that you must also have a valid ingress setup:
+
         config:
           langgraphPlatform:
             enabled: true

docs/docs/cloud/deployment/setup.md

@@ -95,7 +95,7 @@ my-app/
 
 ## Define Graphs
 
-Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledGraph][langgraph.graph.graph.CompiledGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).
+Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledStateGraph][langgraph.graph.state.CompiledStateGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).
 
 Example `agent.py` file, which shows how to import from other modules you define (code for the modules is not shown here, please see [this repository](https://github.com/langchain-ai/langgraph-example) to see their implementation):
 

docs/docs/cloud/deployment/setup_pyproject.md

@@ -108,7 +108,7 @@ my-app/
 
 ## Define Graphs
 
-Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledGraph][langgraph.graph.graph.CompiledGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).
+Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledStateGraph][langgraph.graph.state.CompiledStateGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).
 
 Example `agent.py` file, which shows how to import from other modules you define (code for the modules is not shown here, please see [this repository](https://github.com/langchain-ai/langgraph-example-pyproject) to see their implementation):
 

docs/docs/cloud/how-tos/studio/manage_assistants.md

@@ -2,7 +2,7 @@
 
 !!! info "Prerequisites"
 
-    - [Assistants Overview](../../concepts/assistants.md)
+    - [Assistants Overview](../../../concepts/assistants.md)
 
 LangGraph Studio lets you view, edit, and update your assistants, and allows you to run your graph using these assistant configurations.
 

docs/docs/cloud/how-tos/use_stream_react.md

@@ -1,8 +1,8 @@
-# How to integrate LangGraph into your React application
+How to integrate LangGraph into your React application# How to integrate LangGraph into your React application
 
-!!! info "Prerequisites" 
+!!! info "Prerequisites"
 
-    - [LangGraph Platform](../../concepts/langgraph_platform.md) 
+    - [LangGraph Platform](../../concepts/langgraph_platform.md)
     - [LangGraph Server](../../concepts/langgraph_server.md)
 
 The `useStream()` React hook provides a seamless way to integrate LangGraph into your React applications. It handles all the complexities of streaming, state management, and branching logic, letting you focus on building great chat experiences.
@@ -113,6 +113,115 @@ export default function App() {
 }
 ```
 
+### Resume a stream after page refresh
+
+The `useStream()` hook can automatically resume an ongoing run upon mounting by setting `reconnectOnMount: true`. This is useful for continuing a stream after a page refresh, ensuring no messages and events generated during the downtime are lost.
+
+```tsx
+const thread = useStream<{ messages: Message[] }>({
+  apiUrl: "http://localhost:2024",
+  assistantId: "agent",
+  reconnectOnMount: true,
+});
+```
+
+By default the ID of the created run is stored in `window.sessionStorage`, which can be swapped by passing a custom storage in `reconnectOnMount` instead. The storage is used to persist the in-flight run ID for a thread (under `lg:stream:${threadId}` key).
+
+```tsx
+const thread = useStream<{ messages: Message[] }>({
+  apiUrl: "http://localhost:2024",
+  assistantId: "agent",
+  reconnectOnMount: () => window.localStorage,
+});
+```
+
+You can also manually manage the resuming process by using the run callbacks to persist the run metadata and the `joinStream` function to resume the stream. Make sure to pass `streamResumable: true` when creating the run; otherwise some events might be lost.
+
+````tsx
+import type { Message } from "@langchain/langgraph-sdk";
+import { useStream } from "@langchain/langgraph-sdk/react";
+import { useCallback, useState, useEffect, useRef } from "react";
+
+export default function App() {
+  const [threadId, onThreadId] = useSearchParam("threadId");
+
+  const thread = useStream<{ messages: Message[] }>({
+    apiUrl: "http://localhost:2024",
+    assistantId: "agent",
+
+    threadId,
+    onThreadId,
+
+    onCreated: (run) => {
+      window.sessionStorage.setItem(`resume:${run.thread_id}`, run.run_id);
+    },
+    onFinish: (_, run) => {
+      window.sessionStorage.removeItem(`resume:${run?.thread_id}`);
+    },
+  });
+
+  // Ensure that we only join the stream once per thread.
+  const joinedThreadId = useRef<string | null>(null);
+  useEffect(() => {
+    if (!threadId) return;
+
+    const resume = window.sessionStorage.getItem(`resume:${threadId}`);
+    if (resume && joinedThreadId.current !== threadId) {
+      thread.joinStream(resume);
+      joinedThreadId.current = threadId;
+    }
+  }, [threadId]);
+
+  return (
+    <form
+      onSubmit={(e) => {
+        e.preventDefault();
+        const form = e.target as HTMLFormElement;
+        const message = new FormData(form).get("message") as string;
+        thread.submit(
+          { messages: [{ type: "human", content: message }] },
+          { streamResumable: true }
+        );
+      }}
+    >
+      <div>
+        {thread.messages.map((message) => (
+          <div key={message.id}>{message.content as string}</div>
+        ))}
+      </div>
+      <input type="text" name="message" />
+      <button type="submit">Send</button>
+    </form>
+  );
+}
+
+// Utility method to retrieve and persist data in URL as search param
+function useSearchParam(key: string) {
+  const [value, setValue] = useState<string | null>(() => {
+    const params = new URLSearchParams(window.location.search);
+    return params.get(key) ?? null;
+  });
+
+  const update = useCallback(
+    (value: string | null) => {
+      setValue(value);
+
+      const url = new URL(window.location.href);
+      if (value == null) {
+        url.searchParams.delete(key);
+      } else {
+        url.searchParams.set(key, value);
+      }
+
+      window.history.pushState({}, "", url.toString());
+    },
+    [key]
+  );
+
+  return [value, update] as const;
+}
+```
+
 ### Thread Management
 
 Keep track of conversations with built-in thread management. You can access the current thread ID and get notified when new threads are created:
@@ -127,7 +236,7 @@ const thread = useStream<{ messages: Message[] }>({
   threadId: threadId,
   onThreadId: setThreadId,
 });
-```
+````
 
 We recommend storing the `threadId` in your URL's query parameters to let users resume conversations after page refreshes.
 

docs/docs/cloud/reference/cli.md

@@ -43,6 +43,7 @@ The LangGraph CLI requires a JSON configuration file that follows this [schema](
     | <span style="white-space: nowrap;">`graphs`</span>           | **Required**. Mapping from graph ID to path where the compiled graph or a function that makes a graph is defined. Example: <ul><li>`./your_package/your_file.py:variable`, where `variable` is an instance of `langgraph.graph.state.CompiledStateGraph`</li><li>`./your_package/your_file.py:make_graph`, where `make_graph` is a function that takes a config dictionary (`langchain_core.runnables.RunnableConfig`) and returns an instance of `langgraph.graph.state.StateGraph` or `langgraph.graph.state.CompiledStateGraph`. See [how to rebuild a graph at runtime](../../cloud/deployment/graph_rebuild.md) for more details.</li></ul>                                    |
     | <span style="white-space: nowrap;">`auth`</span>             | _(Added in v0.0.11)_ Auth configuration containing the path to your authentication handler. Example: `./your_package/auth.py:auth`, where `auth` is an instance of `langgraph_sdk.Auth`. See [authentication guide](../../concepts/auth.md) for details.                                                                                                                                                                                                                                                                                                                        |
     | <span style="white-space: nowrap;">`base_image`</span>       | Optional. Base image to use for the LangGraph API server. Defaults to `langchain/langgraph-api` or `langchain/langgraphjs-api`. Use this to pin your builds to a particular version of the langgraph API, such as `"langchain/langgraph-server:0.2"`. See https://hub.docker.com/r/langchain/langgraph-server/tags for more details. (added in `langgraph-cli==0.2.8`) |
+    | <span style="white-space: nowrap;">`image_distro`</span>     | Optional. Linux distribution for the base image. Must be either `"debian"` or `"wolfi"`. If omitted, defaults to `"debian"`. Available in `langgraph-cli>=0.2.11`.|
     | <span style="white-space: nowrap;">`env`</span>              | Path to `.env` file or a mapping from environment variable to its value.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
     | <span style="white-space: nowrap;">`store`</span>            | Configuration for adding semantic search and/or time-to-live (TTL) to the BaseStore. Contains the following fields: <ul><li>`index` (optional): Configuration for semantic search indexing with fields `embed`, `dims`, and optional `fields`.</li><li>`ttl` (optional): Configuration for item expiration. An object with optional fields: `refresh_on_read` (boolean, defaults to `true`), `default_ttl` (float, lifespan in **minutes**, defaults to no expiration), and `sweep_interval_minutes` (integer, how often to check for expired items, defaults to no sweeping).</li></ul> |
     | <span style="white-space: nowrap;">`ui`</span>               | Optional. Named definitions of UI components emitted by the agent, each pointing to a JS/TS file. (added in `langgraph-cli==0.1.84`)                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
@@ -79,6 +80,20 @@ The LangGraph CLI requires a JSON configuration file that follows this [schema](
     }
     ```
 
+    #### Using Wolfi Base Images
+
+    You can specify the Linux distribution for your base image using the `image_distro` field. Valid options are `debian` or `wolfi`. Wolfi is the recommended option as it provides smaller and more secure images. This is available in `langgraph-cli>=0.2.11`.
+
+    ```json
+    {
+      "dependencies": ["."],
+      "graphs": {
+        "chat": "./chat/graph.py:graph"
+      },
+      "image_distro": "wolfi"
+    }
+    ```
+
     #### Adding semantic search to the store
 
     All deployments come with a DB-backed BaseStore. Adding an "index" configuration to your `langgraph.json` will enable [semantic search](../deployment/semantic_search.md) within the BaseStore of your deployment.

docs/docs/concepts/application_structure.md

@@ -32,7 +32,7 @@ Below are examples of directory structures for Python and JavaScript application
     │   ├── utils # utilities for your graph
     │   │   ├── __init__.py
     │   │   ├── tools.py # tools for your graph
-    │   │   ├── nodes.py # node functions for you graph
+    │   │   ├── nodes.py # node functions for your graph
     │   │   └── state.py # state definition of your graph
     │   ├── __init__.py
     │   └── agent.py # code for constructing your graph

docs/docs/concepts/assistants.md

@@ -26,4 +26,4 @@ Once you've created an assistant, subsequent edits to that assistant will create
 
 ## Learn more
 
-* The LangGraph Cloud API provides several endpoints for creating and managing assistants their versions. See the [API reference](../cloud/reference/api/api_ref.html#tag/assistants) for more details.
+* The LangGraph Cloud API provides several endpoints for creating and managing assistants and their versions. See the [API reference](../cloud/reference/api/api_ref.html#tag/assistants) for more details.

docs/docs/concepts/auth.md

@@ -198,7 +198,7 @@ async def add_owner(
 You can register handlers for specific resources and actions by chaining the resource and action names together with the [`@auth.on`](../cloud/reference/sdk/python_sdk_ref.md#langgraph_sdk.auth.Auth.on) decorator.
 When a request is made, the most specific handler that matches that resource and action is called. Below is an example of how to register handlers for specific resources and actions. For the following setup:
 
-1. Authenticated users are able to create threads, read thread, create runs on threads
+1. Authenticated users are able to create threads, read threads, and create runs on threads
 2. Only users with the "assistants:create" permission are allowed to create new assistants
 3. All other endpoints (e.g., e.g., delete assistant, crons, store) are disabled for all users.
 

docs/docs/concepts/deployment_options.md

@@ -59,7 +59,7 @@ For more information, please see:
 !!! info "Important"
     The Self-Hosted Control Plane deployment option is currently in beta stage and requires an [Enterprise](../concepts/plans.md) plan.
 
-The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option give you full control and responsibility of the control plane and data plane infrastructure.
+The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option gives you full control and responsibility of the control plane and data plane infrastructure.
 
 Build a Docker image using the [LangGraph CLI](./langgraph_cli.md) and deploy your LangGraph Server from the [control plane UI](./langgraph_control_plane.md#control-plane-ui).
 

docs/docs/concepts/faq.md

@@ -59,8 +59,8 @@ Yes! You can use LangGraph with any LLMs. The main reason we use LLMs that suppo
 
 Yes! LangGraph is totally ambivalent to what LLMs are used under the hood. The main reason we use closed LLMs in most of the tutorials is that they seamlessly support tool calling, while OSS LLMs often don't. But tool calling is not necessary (see [this section](#does-langgraph-work-with-llms-that-dont-support-tool-calling)) so you can totally use LangGraph with OSS LLMs.
 
-## Can I use LangGraph Studio without logging to LangSmith
+## Can I use LangGraph Studio without logging in to LangSmith
 
 Yes! You can use the [development version of LangGraph Server](../tutorials/langgraph-platform/local-server.md) to run the backend locally.
 This will connect to the studio frontend hosted as part of LangSmith.
-If you set an environment variable of `LANGSMITH_TRACING=false` then no traces will be sent to LangSmith.
+If you set an environment variable of `LANGSMITH_TRACING=false`, then no traces will be sent to LangSmith.

docs/docs/concepts/functional_api.md

@@ -186,7 +186,7 @@ When declaring an `entrypoint`, you can request access to additional parameters
 
 | Parameter    | Description                                                                                                                                                        |
 |--------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| **previous** | Access the the state associated with the previous `checkpoint` for the given thread. See [short-term-memory](#short-term-memory).                                  |
+| **previous** | Access the state associated with the previous `checkpoint` for the given thread. See [short-term-memory](#short-term-memory).                                  |
 | **store**    | An instance of [BaseStore][langgraph.store.base.BaseStore]. Useful for [long-term memory](../how-tos/use-functional-api.md#long-term-memory).                      |
 | **writer**   | Use to access the StreamWriter when working with Async Python < 3.11. See [streaming with functional API for details](../how-tos/use-functional-api.md#streaming). |
 | **config**   | For accessing run time configuration. See [RunnableConfig](https://python.langchain.com/docs/concepts/runnables/#runnableconfig) for information.                  |

docs/docs/concepts/langgraph_data_plane.md

@@ -9,7 +9,7 @@ The term "data plane" is used broadly to refer to [LangGraph Servers](./langgrap
 
 ## Server Infrastructure
 
-In addition to the [LangGraph Server](./langgraph_server.md) itself, the following infrastructure for each server are also included in the broad definition of "data plane":
+In addition to the [LangGraph Server](./langgraph_server.md) itself, the following infrastructure components for each server are also included in the broad definition of "data plane":
 
 - Postgres
 - Redis
@@ -44,7 +44,7 @@ All runs in a LangGraph Server are executed by a pool of background workers that
 
 ### Ephemeral metadata
 
-Runs in a LangGraph Server may be retried for specific failures (currently only for transient Postgres errors encountered during the run). In order to limit the number of retries (currently limited to 3 attempts per run) we record the attempt number in a Redis string when is picked up. This contains no run-specific info other than its ID, and expires after a short delay.
+Runs in a LangGraph Server may be retried for specific failures (currently only for transient Postgres errors encountered during the run). In order to limit the number of retries (currently limited to 3 attempts per run) we record the attempt number in a Redis string when it is picked up. This contains no run-specific info other than its ID, and expires after a short delay.
 
 ## Data Plane Features
 
@@ -62,7 +62,7 @@ For CPU utilization, the autoscaler targets 75% utilization. This means the auto
 
 For number of pending runs, the autoscaler targets 10 pending runs. For example, if the current number of containers is 1, but the number of pending runs in 20, the autoscaler will scale up the deployment to 2 containers (20 pending runs / 2 containers = 10 pending runs per container).
 
-Each metric is computed independently and the autoscaler will determine the scaling action based on the metric that results in the most number of containers.
+Each metric is computed independently and the autoscaler will determine the scaling action based on the metric that results in the largest number of containers.
 
 Scale down actions are delayed for 30 minutes before any action is taken. In other words, if the autoscaler decides to scale down a deployment, it will first wait for 30 minutes before scaling down. After 30 minutes, the metrics are recomputed and the deployment will scale down if the recomputed metrics result in a lower number of containers than the current number. Otherwise, the deployment remains scaled up. This "cool down" period ensures that deployments do not scale up and down too frequently.
 

docs/docs/concepts/langgraph_platform.md

@@ -9,7 +9,7 @@ Develop, deploy, scale, and manage agents with **LangGraph Platform** — the pu
 
 !!! tip "Get started with LangGraph Platform"
 
-    Check out the [LangGraph Platform quickstart](../tutorials/langgraph-platform/local-server.md) for instructions on how to use LangGraph Platform run a LangGraph application locally.
+    Check out the [LangGraph Platform quickstart](../tutorials/langgraph-platform/local-server.md) for instructions on how to use LangGraph Platform to run a LangGraph application locally.
 
 ## Why use LangGraph Platform?
 
@@ -33,4 +33,4 @@ LangGraph Platform makes it easy to get your agent running in production —  wh
 
 - **[LangGraph Studio](./langgraph_studio.md)**: Enables visualization, interaction, and debugging of agentic systems that implement the LangGraph Server API protocol. Studio also integrates with LangSmith to enable tracing, evaluation, and prompt engineering.
 
-- **[Deployment](./deployment_options.md)**: There are four ways to deploy on LangGraph Platform: [Cloud Saas](../concepts/langgraph_cloud.md), [Self-Hosted Data Plane](../concepts/langgraph_self_hosted_data_plane.md), [Self-Hosted Control Plane](../concepts/langgraph_self_hosted_control_plane.md), and [Standalone Container](../concepts/langgraph_standalone_container.md).
+- **[Deployment](./deployment_options.md)**: There are four ways to deploy on LangGraph Platform: [Cloud SaaS](../concepts/langgraph_cloud.md), [Self-Hosted Data Plane](../concepts/langgraph_self_hosted_data_plane.md), [Self-Hosted Control Plane](../concepts/langgraph_self_hosted_control_plane.md), and [Standalone Container](../concepts/langgraph_standalone_container.md).

docs/docs/concepts/langgraph_self_hosted_control_plane.md

@@ -9,10 +9,12 @@ There are two versions of the self-hosted deployment: [Self-Hosted Data Plane](.
 
 - You use `langgraph-cli` and/or [LangGraph Studio](./langgraph_studio.md) app to test graph locally.
 - You use `langgraph build` command to build image.
+- You have a Self-Hosted LangSmith instance deployed.
+- You are using Ingress for your LangSmith instance. All agents will be deployed as Kubernetes services behind this ingress.
 
 ## Self-Hosted Control Plane
 
-The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option give you full control and responsibility of the control plane and data plane infrastructure.
+The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option gives you full control and responsibility of the control plane and data plane infrastructure.
 
 |                   | [Control plane](../concepts/langgraph_control_plane.md) | [Data plane](../concepts/langgraph_data_plane.md) |
 |-------------------|-------------------|------------|
@@ -29,4 +31,4 @@ The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deploy
  - **Kubernetes**: The Self-Hosted Control Plane deployment option supports deploying control plane and data plane infrastructure to any Kubernetes cluster.
 
 !!! tip
-    If you would like to deploy to Kubernetes, you can use this [Helm chart](https://github.com/langchain-ai/helm/blob/main/charts/langgraph-cloud/README.md).
+    If you would like to enable this on your LangSmith instance, please follow the [Self-Hosted Control Plane deployment guide](../cloud/deployment/self_hosted_control_plane.md).

docs/docs/concepts/langgraph_self_hosted_data_plane.md

@@ -37,4 +37,4 @@ For information on how to deploy a [LangGraph Server](../concepts/langgraph_serv
 - **Amazon ECS**: Coming soon!
 
 !!! tip
-    If you would like to deploy to Kubernetes, you can use this [Helm chart](https://github.com/langchain-ai/helm/blob/main/charts/langgraph-cloud/README.md).
+    If you would like to deploy to Kubernetes, you can follow the [Self-Hosted Data Plane deployment guide](../cloud/deployment/self_hosted_data_plane.md).

docs/docs/concepts/langgraph_server.md

@@ -26,7 +26,7 @@ Feature Differences:
 |-------|------------|------------|
 | [Cron Jobs](../cloud/concepts/cron_jobs.md) |❌|✅|
 | [Custom Authentication](../concepts/auth.md) |❌|✅|
-| [Deployment options](../concepts/deployment_options.md) | Standalone container | Cloud Saas, Self-Hosted Data Plane, Self-Hosted Control Plane, Standalone container
+| [Deployment options](../concepts/deployment_options.md) | Standalone container | Cloud SaaS, Self-Hosted Data Plane, Self-Hosted Control Plane, Standalone container
 
 ## Application structure
 

docs/docs/concepts/langgraph_studio.md

@@ -21,7 +21,7 @@ Key features of LangGraph Studio:
 
 - Visualize your graph architecture
 - [Run and interact with your agent](../cloud/how-tos/invoke_studio.md)
-- [Manage assistants](../cloud/how-tos/studio/manage_assistants.md.md)
+- [Manage assistants](../cloud/how-tos/studio/manage_assistants.md)
 - [Manage threads](../cloud/how-tos/threads_studio.md)
 - [Iterate on prompts](../cloud/how-tos/iterate_graph_studio.md)
 - Manage [long term memory](memory.md)
@@ -33,7 +33,7 @@ Studio supports two modes:
 
 ### Graph mode
 
-Graph mode exposes the full feature-set of Studio and is useful when you would like as many details about the execution of your agent, including the nodes traversed, intermediate states, and LangSmith integrations (such as adding to datasets an playground).
+Graph mode exposes the full feature-set of Studio and is useful when you would like as many details about the execution of your agent, including the nodes traversed, intermediate states, and LangSmith integrations (such as adding to datasets and playground).
 
 ### Chat mode
 

docs/docs/concepts/low_level.md

@@ -89,7 +89,7 @@ def node_3(state: PrivateState) -> OutputState:
     # Read from PrivateState, write to OutputState
     return {"graph_output": state["bar"] + " Lance"}
 
-builder = StateGraph(OverallState,input=InputState,output=OutputState)
+builder = StateGraph(OverallState,input_schema=InputState,output_schema=OutputState)
 builder.add_node("node_1", node_1)
 builder.add_node("node_2", node_2)
 builder.add_node("node_3", node_3)
@@ -105,9 +105,9 @@ graph.invoke({"user_input":"My"})
 
 There are two subtle and important points to note here:
 
-1. We pass `state: InputState` as the input schema to `node_1`. But, we write out to `foo`, a channel in `OverallState`. How can we write out to a state channel that is not included in the input schema? This is because a node _can write to any state channel in the graph state._ The graph state is the union of of the state channels defined at initialization, which includes `OverallState` and the filters `InputState` and `OutputState`.
+1. We pass `state: InputState` as the input schema to `node_1`. But, we write out to `foo`, a channel in `OverallState`. How can we write out to a state channel that is not included in the input schema? This is because a node _can write to any state channel in the graph state._ The graph state is the union of the state channels defined at initialization, which includes `OverallState` and the filters `InputState` and `OutputState`.
 
-2. We initialize the graph with `StateGraph(OverallState,input=InputState,output=OutputState)`. So, how can we write to `PrivateState` in `node_2`? How does the graph gain access to this schema if it was not passed in the `StateGraph` initialization? We can do this because _nodes can also declare additional state channels_ as long as the state schema definition exists. In this case, the `PrivateState` schema is defined, so we can add `bar` as a new state channel in the graph and write to it.
+2. We initialize the graph with `StateGraph(OverallState,input_schema=InputState,output_schema=OutputState)`. So, how can we write to `PrivateState` in `node_2`? How does the graph gain access to this schema if it was not passed in the `StateGraph` initialization? We can do this because _nodes can also declare additional state channels_ as long as the state schema definition exists. In this case, the `PrivateState` schema is defined, so we can add `bar` as a new state channel in the graph and write to it.
 
 ### Reducers
 
@@ -167,7 +167,7 @@ In addition to keeping track of message IDs, the `add_messages` function will al
 {"messages": [{"type": "human", "content": "message"}]}
 ```
 
-Since the state updates are always deserialized into LangChain `Messages` when using `add_messages`, you should use dot notation to access message attributes, like `state["messages"][-1].content`. Below is an example of a graph that uses `add_messages` as it's reducer function.
+Since the state updates are always deserialized into LangChain `Messages` when using `add_messages`, you should use dot notation to access message attributes, like `state["messages"][-1].content`. Below is an example of a graph that uses `add_messages` as its reducer function.
 
 ```python
 from langchain_core.messages import AnyMessage
@@ -197,19 +197,25 @@ In LangGraph, nodes are typically python functions (sync or async) where the **f
 Similar to `NetworkX`, you add these nodes to a graph using the [add_node][langgraph.graph.StateGraph.add_node] method:
 
 ```python
+from typing_extensions import TypedDict
+
 from langchain_core.runnables import RunnableConfig
 from langgraph.graph import StateGraph
 
-builder = StateGraph(dict)
+class State(TypedDict):
+    input: str
+    results: str
+
+builder = StateGraph(State)
 
 
-def my_node(state: dict, config: RunnableConfig):
+def my_node(state: State, config: RunnableConfig):
     print("In node: ", config["configurable"]["user_id"])
     return {"results": f"Hello, {state['input']}!"}
 
 
 # The second argument is optional
-def my_other_node(state: dict):
+def my_other_node(state: State):
     return state
 
 

docs/docs/concepts/persistence.md

@@ -383,7 +383,7 @@ def update_memory(state: MessagesState, config: RunnableConfig, *, store: BaseSt
 
 ```
 
-As we showed above, we can also access the store in any node and use the `store.search` method to get memories. Recall the the memories are returned as a list of objects that can be converted to a dictionary.
+As we showed above, we can also access the store in any node and use the `store.search` method to get memories. Recall the memories are returned as a list of objects that can be converted to a dictionary.
 
 ```python
 memories[-1].dict()

docs/docs/concepts/pregel.md

@@ -25,7 +25,7 @@ Each step consists of three phases:
 
 Repeat until no **actors** are selected for execution, or a maximum number of steps is reached.
 
-## Actors 
+## Actors
 
 An **actor** is a `PregelNode`. It subscribes to channels, reads data from them, and writes data to them. It can be thought of as an **actor** in the Pregel algorithm. `PregelNodes` implement LangChain's Runnable interface.
 
@@ -39,7 +39,7 @@ Channels are used to communicate between actors (PregelNodes). Each channel has
 
 ## Examples
 
-While most users will interact with Pregel through the [StateGraph][langgraph.graph.StateGraph] API or 
+While most users will interact with Pregel through the [StateGraph][langgraph.graph.StateGraph] API or
 the [entrypoint][langgraph.func.entrypoint] decorator, it is possible to interact with Pregel directly.
 
 Below are a few different examples to give you a sense of the Pregel API.
@@ -49,12 +49,12 @@ Below are a few different examples to give you a sense of the Pregel API.
     ```python
 
     from langgraph.channels import EphemeralValue
-    from langgraph.pregel import Pregel, Channel 
+    from langgraph.pregel import Pregel, NodeBuilder
 
     node1 = (
-        Channel.subscribe_to("a")
-        | (lambda x: x + x)
-        | Channel.write_to("b")
+        NodeBuilder().subscribe_only("a")
+        .do(lambda x: x + x)
+        .write_to("b")
     )
 
     app = Pregel(
@@ -78,18 +78,18 @@ Below are a few different examples to give you a sense of the Pregel API.
 
     ```python
     from langgraph.channels import LastValue, EphemeralValue
-    from langgraph.pregel import Pregel, Channel 
+    from langgraph.pregel import Pregel, NodeBuilder
 
     node1 = (
-        Channel.subscribe_to("a")
-        | (lambda x: x + x)
-        | Channel.write_to("b")
+        NodeBuilder().subscribe_only("a")
+        .do(lambda x: x + x)
+        .write_to("b")
     )
 
     node2 = (
-        Channel.subscribe_to("b")
-        | (lambda x: x + x)
-        | Channel.write_to("c")
+        NodeBuilder().subscribe_only("b")
+        .do(lambda x: x + x)
+        .write_to("c")
     )
 
 
@@ -115,23 +115,18 @@ Below are a few different examples to give you a sense of the Pregel API.
 
     ```python
     from langgraph.channels import EphemeralValue, Topic
-    from langgraph.pregel import Pregel, Channel 
+    from langgraph.pregel import Pregel, NodeBuilder
 
     node1 = (
-        Channel.subscribe_to("a")
-        | (lambda x: x + x)
-        | {
-            "b": Channel.write_to("b"),
-            "c": Channel.write_to("c")
-        }
+        NodeBuilder().subscribe_only("a")
+        .do(lambda x: x + x)
+        .write_to("b", "c")
     )
 
     node2 = (
-        Channel.subscribe_to("b")
-        | (lambda x: x + x)
-        | {
-            "c": Channel.write_to("c"),
-        }
+        NodeBuilder().subscribe_to("b")
+        .do(lambda x: x["b"] + x["b"])
+        .write_to("c")
     )
 
     app = Pregel(
@@ -158,24 +153,19 @@ Below are a few different examples to give you a sense of the Pregel API.
 
     ```python
     from langgraph.channels import EphemeralValue, BinaryOperatorAggregate
-    from langgraph.pregel import Pregel, Channel
+    from langgraph.pregel import Pregel, NodeBuilder
 
 
     node1 = (
-        Channel.subscribe_to("a")
-        | (lambda x: x + x)
-        | {
-            "b": Channel.write_to("b"),
-            "c": Channel.write_to("c")
-        }
+        NodeBuilder().subscribe_only("a")
+        .do(lambda x: x + x)
+        .write_to("b", "c")
     )
 
     node2 = (
-        Channel.subscribe_to("b")
-        | (lambda x: x + x)
-        | {
-            "c": Channel.write_to("c"),
-        }
+        NodeBuilder().subscribe_only("b")
+        .do(lambda x: x + x)
+        .write_to("c")
     )
 
     def reducer(current, update):
@@ -197,8 +187,7 @@ Below are a few different examples to give you a sense of the Pregel API.
 
     app.invoke({"a": "foo"})
     ```
-    
-    
+
 === "Cycle"
 
     This example demonstrates how to introduce a cycle in the graph, by having
@@ -207,12 +196,12 @@ Below are a few different examples to give you a sense of the Pregel API.
 
     ```python
     from langgraph.channels import EphemeralValue
-    from langgraph.pregel import Pregel, Channel, ChannelWrite, ChannelWriteEntry
+    from langgraph.pregel import Pregel, NodeBuilder, ChannelWriteEntry
 
     example_node = (
-        Channel.subscribe_to("value")
-        | (lambda x: x + x if len(x) < 10 else None)
-        | ChannelWrite(writes=[ChannelWriteEntry(channel="value", skip_none=True)])
+        NodeBuilder().subscribe_only("value")
+        .do(lambda x: x + x if len(x) < 10 else None)
+        .write_to(ChannelWriteEntry("value", skip_none=True))
     )
 
     app = Pregel(
@@ -235,7 +224,6 @@ Below are a few different examples to give you a sense of the Pregel API.
 
 LangGraph provides two high-level APIs for creating a Pregel application: the [StateGraph (Graph API)](./low_level.md) and the [Functional API](functional_api.md).
 
-
 === "StateGraph (Graph API)"
 
     The [StateGraph (Graph API)][langgraph.graph.StateGraph] is a higher-level abstraction that simplifies the creation of Pregel applications. It allows you to define a graph of nodes and edges. When you compile the graph, the StateGraph API automatically creates the Pregel application for you.
@@ -266,7 +254,7 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S
     builder.add_node(score_essay)
     builder.add_edge(START, "write_essay")
 
-    # Compile the graph. 
+    # Compile the graph.
     # This will return a Pregel instance.
     graph = builder.compile()
     ```
@@ -279,7 +267,7 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S
 
     You will see something like this:
 
-    ```pycon 
+    ```pycon
     {'__start__': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1810>,
      'write_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba14d0>,
      'score_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1710>}
@@ -310,7 +298,7 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S
 === "Functional API"
 
     In the [Functional API](functional_api.md), you can use an [`entrypoint`][langgraph.func.entrypoint] to create
-    a Pregel application. The `entrypoint` decorator allows you to define a function that takes input and returns output. 
+    a Pregel application. The `entrypoint` decorator allows you to define a function that takes input and returns output.
 
     ```python
     from typing import TypedDict, Optional
@@ -339,8 +327,8 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S
     ```
 
     ```pycon
-    Nodes: 
+    Nodes:
     {'write_essay': <langgraph.pregel.read.PregelNode object at 0x7d05e2f9aad0>}
-    Channels: 
+    Channels:
     {'__start__': <langgraph.channels.ephemeral_value.EphemeralValue object at 0x7d05e2c906c0>, '__end__': <langgraph.channels.last_value.LastValue object at 0x7d05e2c90c40>, '__previous__': <langgraph.channels.last_value.LastValue object at 0x7d05e1007280>}
     ```

docs/docs/concepts/scalability_and_resilience.md

@@ -25,7 +25,7 @@ When a graceful shutdown request is received (SIGINT) an instance enters shutdow
 - gives any in-progress runs a limited number of seconds to finish (if not finished it will be put back in the queue)
 - stops the instance from picking up more runs from the queue
 
-If a hard shutdown occurs due to a server crash or an infrastructure failure, any runs that were in progress will be picked up by a internal sweeper task that looks for in-progress runs that have breached their heartbeat window. The sweeper runs every 2 minutes and will put the runs back in the queue for another instance to pick them up.
+If a hard shutdown occurs due to a server crash or an infrastructure failure, any runs that were in progress will be picked up by an internal sweeper task that looks for in-progress runs that have breached their heartbeat window. The sweeper runs every 2 minutes and will put the runs back in the queue for another instance to pick them up.
 
 ## Postgres resilience
 

docs/docs/concepts/server-mcp.md

@@ -94,7 +94,7 @@ def answer_node(state: InputState):
     return {"answer": "bye", "question": state["question"]}
 
 # Build the graph with explicit schemas
-builder = StateGraph(OverallState, input=InputState, output=OutputState)
+builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
 builder.add_node(answer_node)
 builder.add_edge(START, "answer_node")
 builder.add_edge("answer_node", END)

docs/docs/how-tos/auth/custom_auth.md

@@ -74,7 +74,7 @@ In your `langgraph.json`, add the path to your auth file:
 
 ## 3. Connect from the client
 
-Once you've set up authentication in your server, requests must include the the required authorization information based on your chosen scheme.
+Once you've set up authentication in your server, requests must include the required authorization information based on your chosen scheme.
 Assuming you are using JWT token authentication, you could access your deployments using any of the following methods:
 
 === "Python Client"

docs/docs/how-tos/graph-api.ipynb

@@ -439,7 +439,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 6,
+   "execution_count": null,
    "id": "6ec0eb77-874e-443e-8c73-93125b515106",
    "metadata": {},
    "outputs": [
@@ -478,7 +478,7 @@
     "\n",
     "\n",
     "# Build the graph with input and output schemas specified\n",
-    "builder = StateGraph(OverallState, input=InputState, output=OutputState)\n",
+    "builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)\n",
     "builder.add_node(answer_node)  # Add the answer node\n",
     "builder.add_edge(START, \"answer_node\")  # Define the starting edge\n",
     "builder.add_edge(\"answer_node\", END)  # Define the ending edge\n",
@@ -1198,7 +1198,7 @@
     "\n",
     "There are many use cases where you may wish for your node to have a custom retry policy, for example if you are calling an API, querying a database, or calling an LLM, etc. LangGraph lets you add retry policies to nodes.\n",
     "\n",
-    "To configure a retry policy, pass the `retry` parameter to the [add_node](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph.add_node). The `retry` parameter takes in a `RetryPolicy` named tuple object. Below we instantiate a `RetryPolicy` object with the default parameters and associate it with a node:\n",
+    "To configure a retry policy, pass the `retry_policy` parameter to the [add_node](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph.add_node). The `retry_policy` parameter takes in a `RetryPolicy` named tuple object. Below we instantiate a `RetryPolicy` object with the default parameters and associate it with a node:\n",
     "\n",
     "```python\n",
     "from langgraph.pregel import RetryPolicy\n",
@@ -1206,7 +1206,7 @@
     "builder.add_node(\n",
     "    \"node_name\",\n",
     "    node_function,\n",
-    "    retry=RetryPolicy(),\n",
+    "    retry_policy=RetryPolicy(),\n",
     ")\n",
     "```"
    ]
@@ -1241,7 +1241,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": null,
    "id": "ad92598c-b688-42fa-aae0-9de36273d584",
    "metadata": {},
    "outputs": [],
@@ -1276,9 +1276,9 @@
     "builder.add_node(\n",
     "    \"query_database\",\n",
     "    query_database,\n",
-    "    retry=RetryPolicy(retry_on=sqlite3.OperationalError),\n",
+    "    retry_policy=RetryPolicy(retry_on=sqlite3.OperationalError),\n",
     ")\n",
-    "builder.add_node(\"model\", call_model, retry=RetryPolicy(max_attempts=5))\n",
+    "builder.add_node(\"model\", call_model, retry_policy=RetryPolicy(max_attempts=5))\n",
     "builder.add_edge(START, \"model\")\n",
     "builder.add_edge(\"model\", \"query_database\")\n",
     "builder.add_edge(\"query_database\", END)\n",
@@ -3416,7 +3416,7 @@
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "Python 3 (ipykernel)",
+   "display_name": ".venv",
    "language": "python",
    "name": "python3"
   },
@@ -3430,7 +3430,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.10.4"
+   "version": "3.9.6"
   }
  },
  "nbformat": 4,

docs/docs/how-tos/human_in_the_loop/time-travel.ipynb

@@ -12,9 +12,9 @@
     "\n",
     "\n",
     "1. **Run the graph** with initial inputs using `invoke` or `stream` APIs.\n",
-    "2. **Identify a checkpoint in an existing thread**: Use the [`get_state_history()`][langgraph.graph.graph.CompiledGraph.get_state_history] method to retrieve the execution history for a specific `thread_id` and locate the desired `checkpoint_id`.  \n",
+    "2. **Identify a checkpoint in an existing thread**: Use the [`get_state_history()`][langgraph.graph.state.CompiledStateGraph.get_state_history] method to retrieve the execution history for a specific `thread_id` and locate the desired `checkpoint_id`.  \n",
     "   Alternatively, set a [breakpoint](../../../concepts/breakpoints/) before the node(s) where you want execution to pause. You can then find the most recent checkpoint recorded up to that breakpoint.\n",
-    "3. **(Optional) modify the graph state**: Use the [`update_state`][langgraph.graph.graph.CompiledGraph.update_state] method to modify the graph’s state at the checkpoint and resume execution from alternative state.\n",
+    "3. **(Optional) modify the graph state**: Use the [`update_state`][langgraph.graph.state.CompiledStateGraph.update_state] method to modify the graph’s state at the checkpoint and resume execution from alternative state.\n",
     "4. **Resume execution from the checkpoint**: Use the `invoke` or `stream` APIs with an input of `None` and a configuration containing the appropriate `thread_id` and `checkpoint_id`.\n",
     "\n",
     "## Example\n",

docs/docs/how-tos/many-tools.ipynb

@@ -405,7 +405,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 46,
+   "execution_count": null,
    "id": "1954a5f1-91e4-4b32-9be9-c8bc1cc43cb5",
    "metadata": {},
    "outputs": [],
@@ -465,7 +465,9 @@
     "\n",
     "graph_builder = StateGraph(State)\n",
     "graph_builder.add_node(\"agent\", agent)\n",
-    "graph_builder.add_node(\"select_tools\", select_tools, retry=RetryPolicy(max_attempts=3))\n",
+    "graph_builder.add_node(\n",
+    "    \"select_tools\", select_tools, retry_policy=RetryPolicy(max_attempts=3)\n",
+    ")\n",
     "\n",
     "tool_node = ToolNode(tools=tools)\n",
     "graph_builder.add_node(\"tools\", tool_node)\n",

docs/docs/how-tos/multi-agent-multi-turn-convo-functional.ipynb

@@ -207,7 +207,7 @@
    "id": "213d661e-6ba4-42b9-bc7f-6c8c423e3419",
    "metadata": {},
    "source": [
-    "Let's now create our agents using the the prebuilt [`create_react_agent`][langgraph.prebuilt.chat_agent_executor.create_react_agent] and our multi-agent workflow. Note that will be calling [`interrupt`][langgraph.types.interrupt] every time after we get the final response from each of the agents."
+    "Let's now create our agents using the prebuilt [`create_react_agent`][langgraph.prebuilt.chat_agent_executor.create_react_agent] and our multi-agent workflow. Note that will be calling [`interrupt`][langgraph.types.interrupt] every time after we get the final response from each of the agents."
    ]
   },
   {

docs/docs/how-tos/persistence.ipynb

@@ -739,7 +739,6 @@
     "                'id': '1f029ca3-1f5b-6704-8004-820c16b69a5a',\n",
     "                'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000006.0.3205149138784782', 'branch:to:call_model': '00000000000000000000000000000006.0.14611156755133758'}, 'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}},\n",
     "                'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today?), HumanMessage(content=\"what's my name?\"), AIMessage(content='Your name is Bob.')]},\n",
-    "                'pending_sends': []\n",
     "            },\n",
     "            metadata={\n",
     "                'source': 'loop',\n",
@@ -856,7 +855,7 @@
     "                    'id': '1f029ca3-1f5b-6704-8004-820c16b69a5a', \n",
     "                    'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000006.0.3205149138784782', 'branch:to:call_model': '00000000000000000000000000000006.0.14611156755133758'}, \n",
     "                    'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}},\n",
-    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\"), AIMessage(content='Your name is Bob.')]}, 'pending_sends': []\n",
+    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\"), AIMessage(content='Your name is Bob.')]},\n",
     "                },\n",
     "                metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}}, 'step': 4, 'parents': {}, 'thread_id': '1'}, \n",
     "                parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}}, \n",
@@ -870,8 +869,7 @@
     "                    'id': '1f029ca3-1790-6b0a-8003-baf965b6a38f', \n",
     "                    'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000005.0.7935064215293443', 'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}, \n",
     "                    'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, \n",
-    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\")], 'branch:to:call_model': None}, \n",
-    "                    'pending_sends': []\n",
+    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\")], 'branch:to:call_model': None}\n",
     "                }, \n",
     "                metadata={'source': 'loop', 'writes': None, 'step': 3, 'parents': {}, 'thread_id': '1'}, \n",
     "                parent_config={...}, \n",
@@ -885,8 +883,7 @@
     "                    'id': '1f029ca3-1790-616e-8002-9e021694a0cd', \n",
     "                    'channel_versions': {'__start__': '00000000000000000000000000000004.0.5736472536395331', 'messages': '00000000000000000000000000000003.0.7056767754077798', 'branch:to:call_model': '00000000000000000000000000000003.0.22059023329132854'}, \n",
     "                    'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, \n",
-    "                    'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"what's my name?\"}]}, 'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}, \n",
-    "                    'pending_sends': []\n",
+    "                    'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"what's my name?\"}]}, 'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}\n",
     "                }, \n",
     "                metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': \"what's my name?\"}]}}, 'step': 2, 'parents': {}, 'thread_id': '1'}, \n",
     "                parent_config={...}, \n",
@@ -900,8 +897,7 @@
     "                    'id': '1f029ca3-178d-6f54-8001-d7b180db0c89', \n",
     "                    'channel_versions': {'__start__': '00000000000000000000000000000002.0.18673090920108737', 'messages': '00000000000000000000000000000003.0.7056767754077798', 'branch:to:call_model': '00000000000000000000000000000003.0.22059023329132854'}, \n",
     "                    'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, \n",
-    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}, \n",
-    "                    'pending_sends': []\n",
+    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}\n",
     "                }, \n",
     "                metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}}, 'step': 1, 'parents': {}, 'thread_id': '1'}, \n",
     "                parent_config={...}, \n",
@@ -915,8 +911,7 @@
     "                    'id': '1f029ca3-0874-6612-8000-339f2abc83b1', \n",
     "                    'channel_versions': {'__start__': '00000000000000000000000000000002.0.18673090920108737', 'messages': '00000000000000000000000000000002.0.30296526818059655', 'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}, \n",
     "                    'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}}, \n",
-    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\")], 'branch:to:call_model': None}, \n",
-    "                    'pending_sends': []\n",
+    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\")], 'branch:to:call_model': None}\n",
     "                }, \n",
     "                metadata={'source': 'loop', 'writes': None, 'step': 0, 'parents': {}, 'thread_id': '1'}, \n",
     "                parent_config={...}, \n",
@@ -930,8 +925,7 @@
     "                    'id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565', \n",
     "                    'channel_versions': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, \n",
     "                    'versions_seen': {'__input__': {}}, \n",
-    "                    'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"hi! I'm bob\"}]}}, \n",
-    "                    'pending_sends': []\n",
+    "                    'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"hi! I'm bob\"}]}}\n",
     "                }, \n",
     "                metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': \"hi! I'm bob\"}]}}, 'step': -1, 'parents': {}, 'thread_id': '1'}, \n",
     "                parent_config=None, \n",

docs/docs/how-tos/use-functional-api.md

@@ -321,7 +321,7 @@ attempts = 0
 # The default RetryPolicy is optimized for retrying specific network errors.
 retry_policy = RetryPolicy(retry_on=ValueError)
 
-@task(retry=retry_policy) 
+@task(retry_policy=retry_policy) 
 def get_info():
     global attempts
     attempts += 1

docs/docs/reference/graphs.md

@@ -36,41 +36,6 @@
         - aget_subgraphs
         - with_config
 
-::: langgraph.graph.graph.Graph
-    options:
-      show_if_no_docstring: true
-      show_root_heading: true
-      show_root_full_path: false
-      members:
-        - add_node
-        - add_edge
-        - add_conditional_edges
-        - compile
-
-::: langgraph.graph.graph.CompiledGraph
-    options:
-      show_if_no_docstring: true
-      show_root_heading: true
-      show_root_full_path: false
-      members:
-        - stream
-        - astream
-        - invoke
-        - ainvoke
-        - get_state
-        - aget_state
-        - get_state_history
-        - aget_state_history
-        - update_state
-        - aupdate_state
-        - bulk_update_state
-        - abulk_update_state
-        - get_graph
-        - aget_graph
-        - get_subgraphs
-        - aget_subgraphs
-        - with_config
-
 ::: langgraph.graph.message
     options:
       members:

docs/docs/reference/index.md

@@ -22,7 +22,7 @@ Welcome to the LangGraph reference docs! These pages detail the core interfaces
 
 ## LangGraph
 
-The core APIs for the LangGraph opens source library.
+The core APIs for the LangGraph open source library.
 
 - [Graphs](graphs.md): Main graph abstraction and usage.
 - [Functional API](func.md): Functional programming interface for graphs.

docs/docs/reference/pregel.md

@@ -1,5 +1,21 @@
 # Pregel
 
+::: langgraph.pregel.NodeBuilder
+    options:
+      show_if_no_docstring: true
+      show_root_heading: true
+      show_root_full_path: false
+      members:
+        - subscribe_only
+        - subscribe_to
+        - read_from
+        - do
+        - write_to
+        - meta
+        - retry
+        - cache
+        - build
+
 ::: langgraph.pregel.Pregel
     options:
       show_if_no_docstring: true

docs/docs/tutorials/customer-support/customer-support.ipynb

@@ -833,7 +833,7 @@
     "@tool\n",
     "def book_excursion(recommendation_id: int) -> str:\n",
     "    \"\"\"\n",
-    "    Book a excursion by its recommendation ID.\n",
+    "    Book an excursion by its recommendation ID.\n",
     "\n",
     "    Args:\n",
     "        recommendation_id (int): The ID of the trip recommendation to book.\n",

docs/docs/tutorials/extraction/retries.ipynb

@@ -89,7 +89,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": null,
    "id": "baf669a0-04ee-492d-80d8-8fcb658ed128",
    "metadata": {},
    "outputs": [],
@@ -313,8 +313,8 @@
     "\n",
     "    builder.add_edge(\"finalizer\", END)\n",
     "\n",
-    "    # These functions let the step be used in a MessageGraph\n",
-    "    # or a StateGraph with 'messages' as the key.\n",
+    "    # These functions let the step be used in a\n",
+    "    # StateGraph with 'messages' as the key.\n",
     "    def encode(x: Union[Sequence[AnyMessage], PromptValue]) -> dict:\n",
     "        \"\"\"Ensure the input is the correct format.\"\"\"\n",
     "        if isinstance(x, PromptValue):\n",

docs/docs/tutorials/get-started/1-build-basic-chatbot.md

@@ -32,7 +32,7 @@ from typing import Annotated
 
 from typing_extensions import TypedDict
 
-from langgraph.graph import StateGraph, START
+from langgraph.graph import StateGraph, START, END
 from langgraph.graph.message import add_messages
 
 
@@ -100,7 +100,16 @@ Add an `entry` point to tell the graph **where to start its work** each time it
 graph_builder.add_edge(START, "chatbot")
 ```
 
-## 5. Compile the graph
+## 5. Add an `exit` point
+
+Add an `exit` point to indicate **where the graph should finish execution**. This is helpful for more complex flows, but even in a simple graph like this, adding an end node improves clarity.
+
+```python
+graph_builder.add_edge("chatbot", END)
+```
+This tells the graph to terminate after running the chatbot node.
+
+## 6. Compile the graph
 
 Before running the graph, we'll need to compile it. We can do so by calling `compile()`
 on the graph builder. This creates a `CompiledGraph` we can invoke on our state.
@@ -109,7 +118,7 @@ on the graph builder. This creates a `CompiledGraph` we can invoke on our state.
 graph = graph_builder.compile()
 ```
 
-## 6. Visualize the graph (optional)
+## 7. Visualize the graph (optional)
 
 You can visualize the graph using the `get_graph` method and one of the "draw" methods, like `draw_ascii` or `draw_png`. The `draw` methods each require additional dependencies.
 
@@ -126,7 +135,7 @@ except Exception:
 ![basic chatbot diagram](basic-chatbot.png)
 
 
-## 7. Run the chatbot
+## 8. Run the chatbot
 
 Now run the chatbot! 
 
@@ -171,7 +180,7 @@ from typing import Annotated
 from langchain.chat_models import init_chat_model
 from typing_extensions import TypedDict
 
-from langgraph.graph import StateGraph, START
+from langgraph.graph import StateGraph, START, END
 from langgraph.graph.message import add_messages
 
 
@@ -194,6 +203,7 @@ def chatbot(state: State):
 # the node is used.
 graph_builder.add_node("chatbot", chatbot)
 graph_builder.add_edge(START, "chatbot")
+graph_builder.add_edge("chatbot", END)
 graph = graph_builder.compile()
 ```
 

docs/docs/tutorials/get-started/2-add-tools.md

@@ -1,6 +1,6 @@
 # Add tools
 
-To handle queries you chatbot can't answer "from memory", integrate a web search tool. The chatbot can use this tool to find relevant information and provide better responses.
+To handle queries that your chatbot can't answer "from memory", integrate a web search tool. The chatbot can use this tool to find relevant information and provide better responses.
 
 !!! note
 

docs/docs/tutorials/get-started/3-add-memory.md

@@ -164,7 +164,7 @@ llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
 ```
 -->
 
-```python
+```python hl_lines="36 37"
 from typing import Annotated
 
 from langchain.chat_models import init_chat_model
@@ -206,4 +206,4 @@ graph = graph_builder.compile(checkpointer=memory)
 
 ## Next steps
 
-In the next tutorial, you will [add human-in-the-loop to the chatbot](./4-human-in-the-loop.md) to handle situations where it may need guidance or verification before proceeding.
+In the next tutorial, you will [add human-in-the-loop to the chatbot](./4-human-in-the-loop.md) to handle situations where it may need guidance or verification before proceeding.

docs/docs/tutorials/rag/langgraph_adaptive_rag.ipynb

@@ -516,11 +516,13 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 12,
+   "execution_count": null,
    "id": "b76b5ec3-0720-443d-85b1-c0e79659ca0a",
    "metadata": {},
    "outputs": [],
    "source": [
+    "from pprint import pprint\n",
+    "\n",
     "from langchain.schema import Document\n",
     "\n",
     "\n",
@@ -796,7 +798,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 14,
+   "execution_count": null,
    "id": "29acc541-d726-4b75-84d1-a215845fe88a",
    "metadata": {},
    "outputs": [
@@ -823,8 +825,6 @@
     }
    ],
    "source": [
-    "from pprint import pprint\n",
-    "\n",
     "# Run\n",
     "inputs = {\n",
     "    \"question\": \"What player at the Bears expected to draft first in the 2024 NFL draft?\"\n",

docs/docs/tutorials/reflection/reflection.ipynb

@@ -389,7 +389,7 @@
      "text": [
       "{'generate': {'messages': [AIMessage(content='Title: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nAntoine de Saint-Exupéry\\'s \"The Little Prince\" is a classic novella that has captured the hearts of millions since its publication in 1943. While it might be easy to dismiss this work as a children\\'s story, its profound themes and timeless message make it a relevant and topical piece in modern life. This essay will explore the allegorical nature of \"The Little Prince\" and discuss how its message can be applied to the complexities of the modern world.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\n\"The Little Prince\" is an allegorical tale that explores various aspects of the human condition through its whimsical characters and situations. The Little Prince himself represents innocence, curiosity, and the importance of human connection. As the story unfolds, readers encounter different characters that symbolize various aspects of adult life, such as vanity, materialism, and authority. These representations allow the story to transcend age and culture, making it relatable to a wide range of readers, even in the modern context.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe Little Prince\\'s message is centered around the importance of looking beyond superficial appearances and forming meaningful connections with others. In a world increasingly dominated by technology and social media, where surface-level interactions are commonplace, this message is more relevant than ever. The Little Prince encourages readers to cherish and nurture genuine relationships, reminding us that true happiness and fulfillment come from understanding and empathizing with others.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\n\"The Little Prince\" also offers a critique of modern society, highlighting the dangers of materialism, consumerism, and the pursuit of power. These themes resonate strongly in today\\'s world, where wealth inequality and environmental degradation are pressing issues. The story serves as a reminder that the pursuit of material possessions and status often comes at the expense of our own happiness and the well-being of our planet.\\n\\nConclusion:\\nIn conclusion, \"The Little Prince\" remains a topical and relevant work in modern life due to its allegorical nature, timeless message, and critique of modern society. Its exploration of human connections, materialism, and the pursuit of power offers valuable insights for readers of all ages. By embracing the story\\'s wisdom, we can better navigate the complexities of the modern world and foster a more compassionate, sustainable, and interconnected society.', response_metadata={'token_usage': {'prompt_tokens': 72, 'total_tokens': 632, 'completion_tokens': 560}, 'model_name': 'accounts/fireworks/models/mixtral-8x7b-instruct', 'system_fingerprint': '', 'finish_reason': 'stop', 'logprobs': None}, id='run-b39a25ab-24f6-42d0-96c2-0f74c3ecc8f7-0', usage_metadata={'input_tokens': 72, 'output_tokens': 560, 'total_tokens': 632})]}}\n",
       "---\n",
-      "{'reflect': {'messages': [HumanMessage(content='Essay Critique and Recommendations:\\n\\nTitle: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nThe introduction effectively sets the stage for the essay by providing background information on \"The Little Prince\" and its relevance in modern life. However, consider adding a hook to engage the reader\\'s attention and create a stronger first impression.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\nThis paragraph provides a clear explanation of the allegorical nature of \"The Little Prince.\" To enhance this section, consider offering specific examples from the text to illustrate how the characters and situations symbolize various aspects of adult life. This will strengthen your analysis and make it more engaging for the reader.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe relevance of the Little Prince\\'s message is well-articulated in this paragraph. To further strengthen your argument, consider discussing the consequences of ignoring this message in the context of modern society. This will help emphasize the importance of the Little Prince\\'s wisdom and its relevance to contemporary issues.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\nThis paragraph effectively highlights the story\\'s critique of modern society. To deepen your analysis, explore how the themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince\\'s message that could help address these issues.\\n\\nConclusion:\\nThe conclusion effectively summarizes the main points of the essay and emphasizes the relevance of \"The Little Prince\" in modern life. To further enhance this section, consider incorporating a thought-provoking question or statement that encourages readers to reflect on the story\\'s message and its implications for their own lives.\\n\\nRecommendations:\\n1. Expand the essay to approximately 1,200-1,500 words to allow for a more in-depth analysis.\\n2. Incorporate specific examples and quotes from \"The Little Prince\" to support your arguments and engage the reader.\\n3. Ensure that each body paragraph contains a clear thesis statement, supporting evidence, and analysis.\\n4. Consider discussing counterarguments or potential criticisms of the Little Prince\\'s message to add depth and complexity to your essay.\\n5. Revise and edit the essay for clarity, coherence, and grammar.')]}}\n",
+      "{'reflect': {'messages': [HumanMessage(content='Essay Critique and Recommendations:\\n\\nTitle: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nThe introduction effectively sets the stage for the essay by providing background information on \"The Little Prince\" and its relevance in modern life. However, consider adding a hook to engage the reader\\'s attention and create a stronger first impression.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\nThis paragraph provides a clear explanation of the allegorical nature of \"The Little Prince.\" To enhance this section, consider offering specific examples from the text to illustrate how the characters and situations symbolize various aspects of adult life. This will strengthen your analysis and make it more engaging for the reader.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe relevance of the Little Prince\\'s message is well-articulated in this paragraph. To further strengthen your argument, consider discussing the consequences of ignoring this message in the context of modern society. This will help emphasize the importance of the Little Prince\\'s wisdom and its relevance to contemporary issues.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\nThis paragraph effectively highlights the story\\'s critique of modern society. To deepen your analysis, explore how themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince\\'s message that could help address these issues.\\n\\nConclusion:\\nThe conclusion effectively summarizes the main points of the essay and emphasizes the relevance of \"The Little Prince\" in modern life. To further enhance this section, consider incorporating a thought-provoking question or statement that encourages readers to reflect on the story\\'s message and its implications for their own lives.\\n\\nRecommendations:\\n1. Expand the essay to approximately 1,200-1,500 words to allow for a more in-depth analysis.\\n2. Incorporate specific examples and quotes from \"The Little Prince\" to support your arguments and engage the reader.\\n3. Ensure that each body paragraph contains a clear thesis statement, supporting evidence, and analysis.\\n4. Consider discussing counterarguments or potential criticisms of the Little Prince\\'s message to add depth and complexity to your essay.\\n5. Revise and edit the essay for clarity, coherence, and grammar.')]}}\n",
       "---\n",
       "{'generate': {'messages': [AIMessage(content='Title: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nIn Antoine de Saint-Exupéry\\'s classic novella \"The Little Prince,\" a young boy embarks on a journey through the universe, meeting various characters that symbolize different aspects of adult life. This timeless tale, published in 1943, remains incredibly relevant in today\\'s modern world. Its allegorical nature, thought-provoking message, and critique of modern society offer invaluable insights for readers of all ages. This essay will explore the allegory of \"The Little Prince,\" analyze the relevance of its message, and discuss its critique of modern society, demonstrating its topicality in contemporary life.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\n\"The Little Prince\" is an allegorical tale that uses whimsical characters and situations to explore various aspects of the human condition. For instance, the king represents authority without substance, while the businessman embodies the futility of materialism. The fox, conversely, symbolizes the importance of forming genuine connections and nurturing meaningful relationships. These allegorical representations allow the story to transcend age and culture, making it relatable to a wide range of readers, even in the modern context.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe Little Prince\\'s message is centered around the importance of looking beyond superficial appearances and forming meaningful connections with others. In a world increasingly dominated by technology and social media, where surface-level interactions are commonplace, this message is more relevant than ever. Neglecting this message can lead to feelings of isolation, loneliness, and dissatisfaction. By embracing the story\\'s wisdom, we can prioritize genuine relationships, fostering a more compassionate and interconnected society.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\n\"The Little Prince\" offers a critique of modern society, highlighting the dangers of materialism, consumerism, and the pursuit of power. These themes resonate strongly in today\\'s world, where wealth inequality and environmental degradation are pressing issues. The story serves as a reminder that the pursuit of material possessions and status often comes at the expense of our own happiness and the well-being of our planet. To address these challenges, we must reevaluate our priorities, focusing on sustainability, empathy, and the cultivation of meaningful relationships.\\n\\nConclusion:\\nIn conclusion, \"The Little Prince\" remains a topical and relevant work in modern life due to its allegorical nature, timeless message, and critique of modern society. Its exploration of human connections, materialism, and the pursuit of power offers valuable insights for readers of all ages. By embracing the story\\'s wisdom, we can better navigate the complexities of the modern world and foster a more compassionate, sustainable, and interconnected society. As the Little Prince so eloquently states, \"What is essential is invisible to the eye,\" reminding us that true happiness and fulfillment come from understanding and empathizing with others.\\n\\nExpanded Essay Recommendations:\\n\\n1. Expand the essay to approximately 1,200-1,500 words to allow for a more in-depth analysis.\\n2. Incorporate specific examples and quotes from \"The Little Prince\" to support your arguments and engage the reader. For instance, use quotes like, \"You become responsible, forever, for what you have tamed,\" to emphasize the importance of forming genuine connections.\\n3. Ensure that each body paragraph contains a clear thesis statement, supporting evidence, and analysis.\\n4. Consider discussing counterarguments or potential criticisms of the Little Prince\\'s message to add depth and complexity to your essay. For example, explore the idea that the pursuit of material possessions can provide a sense of security and comfort.\\n5. Revise and edit the essay for clarity, coherence, and grammar. Ensure that transitions between paragraphs are smooth and that your arguments flow logically.', response_metadata={'token_usage': {'prompt_tokens': 1168, 'total_tokens': 2044, 'completion_tokens': 876}, 'model_name': 'accounts/fireworks/models/mixtral-8x7b-instruct', 'system_fingerprint': '', 'finish_reason': 'stop', 'logprobs': None}, id='run-9bfc9ff2-3186-43f5-8b75-498d532d8d1a-0', usage_metadata={'input_tokens': 1168, 'output_tokens': 876, 'total_tokens': 2044})]}}\n",
       "---\n",
@@ -478,7 +478,7 @@
       "The relevance of the Little Prince's message is well-articulated in this paragraph. To further strengthen your argument, consider discussing the consequences of ignoring this message in the context of modern society. This will help emphasize the importance of the Little Prince's wisdom and its relevance to contemporary issues.\n",
       "\n",
       "Body Paragraph 3 - The Critique of Modern Society:\n",
-      "This paragraph effectively highlights the story's critique of modern society. To deepen your analysis, explore how the themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince's message that could help address these issues.\n",
+      "This paragraph effectively highlights the story's critique of modern society. To deepen your analysis, explore how themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince's message that could help address these issues.\n",
       "\n",
       "Conclusion:\n",
       "The conclusion effectively summarizes the main points of the essay and emphasizes the relevance of \"The Little Prince\" in modern life. To further enhance this section, consider incorporating a thought-provoking question or statement that encourages readers to reflect on the story's message and its implications for their own lives.\n",

docs/docs/tutorials/usaco/usaco.ipynb

@@ -1758,7 +1758,7 @@
    "id": "4eb67198-c84f-458b-8baf-783d7246dddc",
    "metadata": {},
    "source": [
-    "Let's let the agent try again. Call `stream` with `None` to just use the inputs loaded from the memory. We will skip our human review for the next few attempats\n",
+    "Let's let the agent try again. Call `stream` with `None` to just use the inputs loaded from the memory. We will skip our human review for the next few attempts\n",
     "to see if it can correct itself."
    ]
   },

docs/overrides/main.html

@@ -1,5 +1,16 @@
 {% extends "base.html" %}
 
+{% block analytics %}
+<!-- Google Tag Manager -->
+<script>(function(w,d,s,l,i){w[l]=w[l]||[];w[l].push({'gtm.start':
+new Date().getTime(),event:'gtm.js'});var f=d.getElementsByTagName(s)[0],
+j=d.createElement(s),dl=l!='dataLayer'?'&l='+l:'';j.async=true;j.src=
+'https://www.googletagmanager.com/gtm.js?id='+i+dl;f.parentNode.insertBefore(j,f);
+})(window,document,'script','dataLayer','GTM-T35S4S46');</script>
+<!-- End Google Tag Manager -->
+{% endblock %}
+
+
 {% block extrahead %}
   <meta name="algolia-site-verification" content="165B7E7C89E49946" />
   <style>
@@ -185,7 +196,6 @@
   </style>
 {% endblock %}
 
-
 {% block content %}
 <div class="notebook-links">
   {% if page.nb_url %}
@@ -209,7 +219,6 @@
   {% endif %}
 {% endblock %}
 
-
 {% block announce %}
 <strong>We are growing and hiring for multiple roles for LangChain, LangGraph and LangSmith. <a href="https://www.langchain.com/careers" target="_blank" rel="noopener noreferrer"> Join our team!</a></strong>
 {% endblock %}

libs/checkpoint-postgres/README.md

@@ -25,7 +25,7 @@ with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
     # call .setup() the first time you're using the checkpointer
     checkpointer.setup()
     checkpoint = {
-        "v": 2,
+        "v": 4,
         "ts": "2024-07-31T20:14:19.804150+00:00",
         "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
         "channel_values": {
@@ -47,7 +47,6 @@ with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
             "start:node": 2
             }
         },
-        "pending_sends": [],
     }
 
     # store checkpoint
@@ -67,7 +66,7 @@ from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
 
 async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
     checkpoint = {
-        "v": 2,
+        "v": 4,
         "ts": "2024-07-31T20:14:19.804150+00:00",
         "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
         "channel_values": {
@@ -89,7 +88,6 @@ async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
             "start:node": 2
             }
         },
-        "pending_sends": [],
     }
 
     # store checkpoint

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

@@ -1,7 +1,10 @@
+from __future__ import annotations
+
 import threading
+from collections import defaultdict
 from collections.abc import Iterator, Sequence
 from contextlib import contextmanager
-from typing import Any, Optional
+from typing import Any
 
 from langchain_core.runnables import RunnableConfig
 from psycopg import Capabilities, Connection, Cursor, Pipeline
@@ -20,7 +23,6 @@ from langgraph.checkpoint.base import (
 )
 from langgraph.checkpoint.postgres import _internal
 from langgraph.checkpoint.postgres.base import BasePostgresSaver
-from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
 from langgraph.checkpoint.serde.base import SerializerProtocol
 
 Conn = _internal.Conn  # For backward compatibility
@@ -34,8 +36,8 @@ class PostgresSaver(BasePostgresSaver):
     def __init__(
         self,
         conn: _internal.Conn,
-        pipe: Optional[Pipeline] = None,
-        serde: Optional[SerializerProtocol] = None,
+        pipe: Pipeline | None = None,
+        serde: SerializerProtocol | None = None,
     ) -> None:
         super().__init__(serde=serde)
         if isinstance(conn, ConnectionPool) and pipe is not None:
@@ -52,7 +54,7 @@ class PostgresSaver(BasePostgresSaver):
     @contextmanager
     def from_conn_string(
         cls, conn_string: str, *, pipeline: bool = False
-    ) -> Iterator["PostgresSaver"]:
+    ) -> Iterator[PostgresSaver]:
         """Create a new PostgresSaver instance from a connection string.
 
         Args:
@@ -99,11 +101,11 @@ class PostgresSaver(BasePostgresSaver):
 
     def list(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> Iterator[CheckpointTuple]:
         """List checkpoints from the database.
 
@@ -143,8 +145,36 @@ class PostgresSaver(BasePostgresSaver):
             query += f" LIMIT {limit}"
         # if we change this to use .stream() we need to make sure to close the cursor
         with self._cursor() as cur:
-            cur.execute(query, args, binary=True)
-            for value in cur:
+            cur.execute(query, args)
+            values = cur.fetchall()
+            if not values:
+                return
+            # migrate pending sends if necessary
+            if to_migrate := [
+                v
+                for v in values
+                if v["checkpoint"]["v"] < 4 and v["parent_checkpoint_id"]
+            ]:
+                cur.execute(
+                    self.SELECT_PENDING_SENDS_SQL,
+                    (
+                        values[0]["thread_id"],
+                        [v["parent_checkpoint_id"] for v in to_migrate],
+                    ),
+                )
+                grouped_by_parent = defaultdict(list)
+                for value in to_migrate:
+                    grouped_by_parent[value["parent_checkpoint_id"]].append(value)
+                for sends in cur:
+                    for value in grouped_by_parent[sends["checkpoint_id"]]:
+                        if value["channel_values"] is None:
+                            value["channel_values"] = []
+                        self._migrate_pending_sends(
+                            sends["sends"],
+                            value["checkpoint"],
+                            value["channel_values"],
+                        )
+            for value in values:
                 yield CheckpointTuple(
                     {
                         "configurable": {
@@ -153,12 +183,11 @@ class PostgresSaver(BasePostgresSaver):
                             "checkpoint_id": value["checkpoint_id"],
                         }
                     },
-                    self._load_checkpoint(
-                        value["checkpoint"],
-                        value["channel_values"],
-                        value["pending_sends"],
-                    ),
-                    self._load_metadata(value["metadata"]),
+                    {
+                        **value["checkpoint"],
+                        "channel_values": self._load_blobs(value["channel_values"]),
+                    },
+                    value["metadata"],
                     (
                         {
                             "configurable": {
@@ -173,7 +202,7 @@ class PostgresSaver(BasePostgresSaver):
                     self._load_writes(value["pending_writes"]),
                 )
 
-    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the database.
 
         This method retrieves a checkpoint tuple from the Postgres database based on the
@@ -222,37 +251,52 @@ class PostgresSaver(BasePostgresSaver):
             cur.execute(
                 self.SELECT_SQL + where,
                 args,
-                binary=True,
             )
+            value = cur.fetchone()
+            if value is None:
+                return None
 
-            for value in cur:
-                return CheckpointTuple(
+            # migrate pending sends if necessary
+            if value["checkpoint"]["v"] < 4 and value["parent_checkpoint_id"]:
+                cur.execute(
+                    self.SELECT_PENDING_SENDS_SQL,
+                    (thread_id, [value["parent_checkpoint_id"]]),
+                )
+                if sends := cur.fetchone():
+                    if value["channel_values"] is None:
+                        value["channel_values"] = []
+                    self._migrate_pending_sends(
+                        sends["sends"],
+                        value["checkpoint"],
+                        value["channel_values"],
+                    )
+
+            return CheckpointTuple(
+                {
+                    "configurable": {
+                        "thread_id": thread_id,
+                        "checkpoint_ns": checkpoint_ns,
+                        "checkpoint_id": value["checkpoint_id"],
+                    }
+                },
+                {
+                    **value["checkpoint"],
+                    "channel_values": self._load_blobs(value["channel_values"]),
+                },
+                value["metadata"],
+                (
                     {
                         "configurable": {
                             "thread_id": thread_id,
                             "checkpoint_ns": checkpoint_ns,
-                            "checkpoint_id": value["checkpoint_id"],
+                            "checkpoint_id": value["parent_checkpoint_id"],
                         }
-                    },
-                    self._load_checkpoint(
-                        value["checkpoint"],
-                        value["channel_values"],
-                        value["pending_sends"],
-                    ),
-                    self._load_metadata(value["metadata"]),
-                    (
-                        {
-                            "configurable": {
-                                "thread_id": thread_id,
-                                "checkpoint_ns": checkpoint_ns,
-                                "checkpoint_id": value["parent_checkpoint_id"],
-                            }
-                        }
-                        if value["parent_checkpoint_id"]
-                        else None
-                    ),
-                    self._load_writes(value["pending_writes"]),
-                )
+                    }
+                    if value["parent_checkpoint_id"]
+                    else None
+                ),
+                self._load_writes(value["pending_writes"]),
+            )
 
     def put(
         self,
@@ -319,8 +363,8 @@ class PostgresSaver(BasePostgresSaver):
                     checkpoint_ns,
                     checkpoint["id"],
                     checkpoint_id,
-                    Jsonb(self._dump_checkpoint(copy)),
-                    self._dump_metadata(get_checkpoint_metadata(config, metadata)),
+                    Jsonb(copy),
+                    Jsonb(get_checkpoint_metadata(config, metadata)),
                 ),
             )
         return next_config
@@ -391,7 +435,7 @@ class PostgresSaver(BasePostgresSaver):
                 Will be applied regardless of whether the PostgresSaver instance was initialized with a pipeline.
                 If pipeline mode is not supported, will fall back to using transaction context manager.
         """
-        with _internal.get_connection(self.conn) as conn:
+        with self.lock, _internal.get_connection(self.conn) as conn:
             if self.pipe:
                 # a connection in pipeline mode can be used concurrently
                 # in multiple threads/coroutines, but only one cursor can be
@@ -407,7 +451,6 @@ class PostgresSaver(BasePostgresSaver):
                 # thread/coroutine at a time, so we acquire a lock
                 if self.supports_pipeline:
                     with (
-                        self.lock,
                         conn.pipeline(),
                         conn.cursor(binary=True, row_factory=dict_row) as cur,
                     ):
@@ -415,14 +458,13 @@ class PostgresSaver(BasePostgresSaver):
                 else:
                     # Use connection's transaction context manager when pipeline mode not supported
                     with (
-                        self.lock,
                         conn.transaction(),
                         conn.cursor(binary=True, row_factory=dict_row) as cur,
                     ):
                         yield cur
             else:
-                with self.lock, conn.cursor(binary=True, row_factory=dict_row) as cur:
+                with conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur
 
 
-__all__ = ["PostgresSaver", "BasePostgresSaver", "ShallowPostgresSaver", "Conn"]
+__all__ = ["PostgresSaver", "BasePostgresSaver", "Conn"]

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

@@ -1,7 +1,10 @@
+from __future__ import annotations
+
 import asyncio
+from collections import defaultdict
 from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import asynccontextmanager
-from typing import Any, Optional
+from typing import Any
 
 from langchain_core.runnables import RunnableConfig
 from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
@@ -20,7 +23,6 @@ from langgraph.checkpoint.base import (
 )
 from langgraph.checkpoint.postgres import _ainternal
 from langgraph.checkpoint.postgres.base import BasePostgresSaver
-from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
 from langgraph.checkpoint.serde.base import SerializerProtocol
 
 Conn = _ainternal.Conn  # For backward compatibility
@@ -34,8 +36,8 @@ class AsyncPostgresSaver(BasePostgresSaver):
     def __init__(
         self,
         conn: _ainternal.Conn,
-        pipe: Optional[AsyncPipeline] = None,
-        serde: Optional[SerializerProtocol] = None,
+        pipe: AsyncPipeline | None = None,
+        serde: SerializerProtocol | None = None,
     ) -> None:
         super().__init__(serde=serde)
         if isinstance(conn, AsyncConnectionPool) and pipe is not None:
@@ -56,8 +58,8 @@ class AsyncPostgresSaver(BasePostgresSaver):
         conn_string: str,
         *,
         pipeline: bool = False,
-        serde: Optional[SerializerProtocol] = None,
-    ) -> AsyncIterator["AsyncPostgresSaver"]:
+        serde: SerializerProtocol | None = None,
+    ) -> AsyncIterator[AsyncPostgresSaver]:
         """Create a new AsyncPostgresSaver instance from a connection string.
 
         Args:
@@ -104,11 +106,11 @@ class AsyncPostgresSaver(BasePostgresSaver):
 
     async def alist(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> AsyncIterator[CheckpointTuple]:
         """List checkpoints from the database asynchronously.
 
@@ -131,7 +133,35 @@ class AsyncPostgresSaver(BasePostgresSaver):
         # if we change this to use .stream() we need to make sure to close the cursor
         async with self._cursor() as cur:
             await cur.execute(query, args, binary=True)
-            async for value in cur:
+            values = await cur.fetchall()
+            if not values:
+                return
+            # migrate pending sends if necessary
+            if to_migrate := [
+                v
+                for v in values
+                if v["checkpoint"]["v"] < 4 and v["parent_checkpoint_id"]
+            ]:
+                await cur.execute(
+                    self.SELECT_PENDING_SENDS_SQL,
+                    (
+                        values[0]["thread_id"],
+                        [v["parent_checkpoint_id"] for v in to_migrate],
+                    ),
+                )
+                grouped_by_parent = defaultdict(list)
+                for value in to_migrate:
+                    grouped_by_parent[value["parent_checkpoint_id"]].append(value)
+                async for sends in cur:
+                    for value in grouped_by_parent[sends["checkpoint_id"]]:
+                        if value["channel_values"] is None:
+                            value["channel_values"] = []
+                        self._migrate_pending_sends(
+                            sends["sends"],
+                            value["checkpoint"],
+                            value["channel_values"],
+                        )
+            for value in values:
                 yield CheckpointTuple(
                     {
                         "configurable": {
@@ -140,13 +170,11 @@ class AsyncPostgresSaver(BasePostgresSaver):
                             "checkpoint_id": value["checkpoint_id"],
                         }
                     },
-                    await asyncio.to_thread(
-                        self._load_checkpoint,
-                        value["checkpoint"],
-                        value["channel_values"],
-                        value["pending_sends"],
-                    ),
-                    self._load_metadata(value["metadata"]),
+                    {
+                        **value["checkpoint"],
+                        "channel_values": self._load_blobs(value["channel_values"]),
+                    },
+                    value["metadata"],
                     (
                         {
                             "configurable": {
@@ -161,7 +189,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
                     await asyncio.to_thread(self._load_writes, value["pending_writes"]),
                 )
 
-    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the database asynchronously.
 
         This method retrieves a checkpoint tuple from the Postgres database based on the
@@ -191,36 +219,51 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 args,
                 binary=True,
             )
+            value = await cur.fetchone()
+            if value is None:
+                return None
 
-            async for value in cur:
-                return CheckpointTuple(
+            # migrate pending sends if necessary
+            if value["checkpoint"]["v"] < 4 and value["parent_checkpoint_id"]:
+                await cur.execute(
+                    self.SELECT_PENDING_SENDS_SQL,
+                    (thread_id, [value["parent_checkpoint_id"]]),
+                )
+                if sends := await cur.fetchone():
+                    if value["channel_values"] is None:
+                        value["channel_values"] = []
+                    self._migrate_pending_sends(
+                        sends["sends"],
+                        value["checkpoint"],
+                        value["channel_values"],
+                    )
+
+            return CheckpointTuple(
+                {
+                    "configurable": {
+                        "thread_id": thread_id,
+                        "checkpoint_ns": checkpoint_ns,
+                        "checkpoint_id": value["checkpoint_id"],
+                    }
+                },
+                {
+                    **value["checkpoint"],
+                    "channel_values": self._load_blobs(value["channel_values"]),
+                },
+                value["metadata"],
+                (
                     {
                         "configurable": {
                             "thread_id": thread_id,
                             "checkpoint_ns": checkpoint_ns,
-                            "checkpoint_id": value["checkpoint_id"],
+                            "checkpoint_id": value["parent_checkpoint_id"],
                         }
-                    },
-                    await asyncio.to_thread(
-                        self._load_checkpoint,
-                        value["checkpoint"],
-                        value["channel_values"],
-                        value["pending_sends"],
-                    ),
-                    self._load_metadata(value["metadata"]),
-                    (
-                        {
-                            "configurable": {
-                                "thread_id": thread_id,
-                                "checkpoint_ns": checkpoint_ns,
-                                "checkpoint_id": value["parent_checkpoint_id"],
-                            }
-                        }
-                        if value["parent_checkpoint_id"]
-                        else None
-                    ),
-                    await asyncio.to_thread(self._load_writes, value["pending_writes"]),
-                )
+                    }
+                    if value["parent_checkpoint_id"]
+                    else None
+                ),
+                await asyncio.to_thread(self._load_writes, value["pending_writes"]),
+            )
 
     async def aput(
         self,
@@ -277,8 +320,8 @@ class AsyncPostgresSaver(BasePostgresSaver):
                     checkpoint_ns,
                     checkpoint["id"],
                     checkpoint_id,
-                    Jsonb(self._dump_checkpoint(copy)),
-                    self._dump_metadata(get_checkpoint_metadata(config, metadata)),
+                    Jsonb(copy),
+                    Jsonb(get_checkpoint_metadata(config, metadata)),
                 ),
             )
         return next_config
@@ -350,7 +393,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 Will be applied regardless of whether the AsyncPostgresSaver instance was initialized with a pipeline.
                 If pipeline mode is not supported, will fall back to using transaction context manager.
         """
-        async with _ainternal.get_connection(self.conn) as conn:
+        async with self.lock, _ainternal.get_connection(self.conn) as conn:
             if self.pipe:
                 # a connection in pipeline mode can be used concurrently
                 # in multiple threads/coroutines, but only one cursor can be
@@ -366,7 +409,6 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 # thread/coroutine at a time, so we acquire a lock
                 if self.supports_pipeline:
                     async with (
-                        self.lock,
                         conn.pipeline(),
                         conn.cursor(binary=True, row_factory=dict_row) as cur,
                     ):
@@ -374,25 +416,21 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 else:
                     # Use connection's transaction context manager when pipeline mode not supported
                     async with (
-                        self.lock,
                         conn.transaction(),
                         conn.cursor(binary=True, row_factory=dict_row) as cur,
                     ):
                         yield cur
             else:
-                async with (
-                    self.lock,
-                    conn.cursor(binary=True, row_factory=dict_row) as cur,
-                ):
+                async with conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur
 
     def list(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> Iterator[CheckpointTuple]:
         """List checkpoints from the database.
 
@@ -430,7 +468,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
             except StopAsyncIteration:
                 break
 
-    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the database.
 
         This method retrieves a checkpoint tuple from the Postgres database based on the
@@ -532,4 +570,4 @@ class AsyncPostgresSaver(BasePostgresSaver):
         ).result()
 
 
-__all__ = ["AsyncPostgresSaver", "AsyncShallowPostgresSaver", "Conn"]
+__all__ = ["AsyncPostgresSaver", "Conn"]

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

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import random
 from collections.abc import Sequence
 from typing import Any, Optional, cast
@@ -9,12 +11,9 @@ from langgraph.checkpoint.base import (
     WRITES_IDX_MAP,
     BaseCheckpointSaver,
     ChannelVersions,
-    Checkpoint,
-    CheckpointMetadata,
     get_checkpoint_id,
 )
-from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
-from langgraph.checkpoint.serde.types import TASKS, ChannelProtocol
+from langgraph.checkpoint.serde.types import TASKS
 
 MetadataInput = Optional[dict[str, Any]]
 
@@ -72,7 +71,7 @@ MIGRATIONS = [
     """ALTER TABLE checkpoint_writes ADD COLUMN task_path TEXT NOT NULL DEFAULT '';""",
 ]
 
-SELECT_SQL = f"""
+SELECT_SQL = """
 select
     thread_id,
     checkpoint,
@@ -96,17 +95,20 @@ select
         where cw.thread_id = checkpoints.thread_id
             and cw.checkpoint_ns = checkpoints.checkpoint_ns
             and cw.checkpoint_id = checkpoints.checkpoint_id
-    ) as pending_writes,
-    (
-        select array_agg(array[cw.type::bytea, cw.blob] order by cw.task_path, cw.task_id, cw.idx)
-        from checkpoint_writes cw
-        where cw.thread_id = checkpoints.thread_id
-            and cw.checkpoint_ns = checkpoints.checkpoint_ns
-            and cw.checkpoint_id = checkpoints.parent_checkpoint_id
-            and cw.channel = '{TASKS}'
-    ) as pending_sends
+    ) as pending_writes
 from checkpoints """
 
+SELECT_PENDING_SENDS_SQL = f"""
+select
+    checkpoint_id,
+    array_agg(array[type::bytea, blob] order by task_path, task_id, idx) as sends
+from checkpoint_writes
+where thread_id = %s
+    and checkpoint_id = any(%s)
+    and channel = '{TASKS}'
+group by checkpoint_id
+"""
+
 UPSERT_CHECKPOINT_BLOBS_SQL = """
     INSERT INTO checkpoint_blobs (thread_id, checkpoint_ns, channel, version, type, blob)
     VALUES (%s, %s, %s, %s, %s, %s)
@@ -140,31 +142,34 @@ INSERT_CHECKPOINT_WRITES_SQL = """
 
 class BasePostgresSaver(BaseCheckpointSaver[str]):
     SELECT_SQL = SELECT_SQL
+    SELECT_PENDING_SENDS_SQL = SELECT_PENDING_SENDS_SQL
     MIGRATIONS = MIGRATIONS
     UPSERT_CHECKPOINT_BLOBS_SQL = UPSERT_CHECKPOINT_BLOBS_SQL
     UPSERT_CHECKPOINTS_SQL = UPSERT_CHECKPOINTS_SQL
     UPSERT_CHECKPOINT_WRITES_SQL = UPSERT_CHECKPOINT_WRITES_SQL
     INSERT_CHECKPOINT_WRITES_SQL = INSERT_CHECKPOINT_WRITES_SQL
 
-    jsonplus_serde = JsonPlusSerializer()
     supports_pipeline: bool
 
-    def _load_checkpoint(
+    def _migrate_pending_sends(
         self,
+        pending_sends: list[tuple[bytes, bytes]],
         checkpoint: dict[str, Any],
         channel_values: list[tuple[bytes, bytes, bytes]],
-        pending_sends: list[tuple[bytes, bytes]],
-    ) -> Checkpoint:
-        return {
-            **checkpoint,
-            "pending_sends": [
-                self.serde.loads_typed((c.decode(), b)) for c, b in pending_sends or []
-            ],
-            "channel_values": self._load_blobs(channel_values),
-        }
-
-    def _dump_checkpoint(self, checkpoint: Checkpoint) -> dict[str, Any]:
-        return {**checkpoint, "pending_sends": []}
+    ) -> None:
+        if not pending_sends:
+            return
+        # add to values
+        enc, blob = self.serde.dumps_typed(
+            [self.serde.loads_typed((c.decode(), b)) for c, b in pending_sends],
+        )
+        channel_values.append((TASKS.encode(), enc.encode(), blob))
+        # add to versions
+        checkpoint["channel_versions"][TASKS] = (
+            max(checkpoint["channel_versions"].values())
+            if checkpoint["channel_versions"]
+            else self.get_next_version(None)
+        )
 
     def _load_blobs(
         self, blob_values: list[tuple[bytes, bytes, bytes]]
@@ -183,7 +188,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
         checkpoint_ns: str,
         values: dict[str, Any],
         versions: ChannelVersions,
-    ) -> list[tuple[str, str, str, str, str, Optional[bytes]]]:
+    ) -> list[tuple[str, str, str, str, str, bytes | None]]:
         if not versions:
             return []
 
@@ -241,15 +246,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
             for idx, (channel, value) in enumerate(writes)
         ]
 
-    def _load_metadata(self, metadata: dict[str, Any]) -> CheckpointMetadata:
-        return self.jsonplus_serde.loads(self.jsonplus_serde.dumps(metadata))
-
-    def _dump_metadata(self, metadata: CheckpointMetadata) -> str:
-        serialized_metadata = self.jsonplus_serde.dumps(metadata)
-        # NOTE: we're using JSON serializer (not msgpack), so we need to remove null characters before writing
-        return serialized_metadata.decode().replace("\\u0000", "")
-
-    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
+    def get_next_version(self, current: str | None) -> str:
         if current is None:
             current_v = 0
         elif isinstance(current, int):
@@ -262,9 +259,9 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
 
     def _search_where(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         filter: MetadataInput,
-        before: Optional[RunnableConfig] = None,
+        before: RunnableConfig | None = None,
     ) -> tuple[str, list[Any]]:
         """Return WHERE clause predicates for alist() given config, filter, before.
 

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

@@ -1,941 +0,0 @@
-import asyncio
-import threading
-import warnings
-from collections.abc import AsyncIterator, Iterator, Sequence
-from contextlib import asynccontextmanager, contextmanager
-from typing import Any, Optional
-
-from langchain_core.runnables import RunnableConfig
-from psycopg import (
-    AsyncConnection,
-    AsyncCursor,
-    AsyncPipeline,
-    Capabilities,
-    Connection,
-    Cursor,
-    Pipeline,
-)
-from psycopg.rows import DictRow, dict_row
-from psycopg.types.json import Jsonb
-from psycopg_pool import AsyncConnectionPool, ConnectionPool
-
-from langgraph.checkpoint.base import (
-    WRITES_IDX_MAP,
-    ChannelVersions,
-    Checkpoint,
-    CheckpointMetadata,
-    CheckpointTuple,
-    get_checkpoint_metadata,
-)
-from langgraph.checkpoint.postgres import _ainternal, _internal
-from langgraph.checkpoint.postgres.base import BasePostgresSaver
-from langgraph.checkpoint.serde.base import SerializerProtocol
-from langgraph.checkpoint.serde.types import TASKS
-
-"""
-To add a new migration, add a new string to the MIGRATIONS list.
-The position of the migration in the list is the version number.
-"""
-MIGRATIONS = [
-    """CREATE TABLE IF NOT EXISTS checkpoint_migrations (
-    v INTEGER PRIMARY KEY
-);""",
-    """CREATE TABLE IF NOT EXISTS checkpoints (
-    thread_id TEXT NOT NULL,
-    checkpoint_ns TEXT NOT NULL DEFAULT '',
-    type TEXT,
-    checkpoint JSONB NOT NULL,
-    metadata JSONB NOT NULL DEFAULT '{}',
-    PRIMARY KEY (thread_id, checkpoint_ns)
-);""",
-    """CREATE TABLE IF NOT EXISTS checkpoint_blobs (
-    thread_id TEXT NOT NULL,
-    checkpoint_ns TEXT NOT NULL DEFAULT '',
-    channel TEXT NOT NULL,
-    type TEXT NOT NULL,
-    blob BYTEA,
-    PRIMARY KEY (thread_id, checkpoint_ns, channel)
-);""",
-    """CREATE TABLE IF NOT EXISTS checkpoint_writes (
-    thread_id TEXT NOT NULL,
-    checkpoint_ns TEXT NOT NULL DEFAULT '',
-    checkpoint_id TEXT NOT NULL,
-    task_id TEXT NOT NULL,
-    idx INTEGER NOT NULL,
-    channel TEXT NOT NULL,
-    type TEXT,
-    blob BYTEA NOT NULL,
-    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
-);""",
-    """
-    CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoints_thread_id_idx ON checkpoints(thread_id);
-    """,
-    """
-    CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoint_blobs_thread_id_idx ON checkpoint_blobs(thread_id);
-    """,
-    """
-    CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoint_writes_thread_id_idx ON checkpoint_writes(thread_id);
-    """,
-    """
-    ALTER TABLE checkpoint_writes ADD COLUMN task_path TEXT NOT NULL DEFAULT '';
-    """,
-]
-
-SELECT_SQL = f"""
-select
-    thread_id,
-    checkpoint,
-    checkpoint_ns,
-    metadata,
-    (
-        select array_agg(array[bl.channel::bytea, bl.type::bytea, bl.blob])
-        from jsonb_each_text(checkpoint -> 'channel_versions')
-        inner join checkpoint_blobs bl
-            on bl.thread_id = checkpoints.thread_id
-            and bl.checkpoint_ns = checkpoints.checkpoint_ns
-            and bl.channel = jsonb_each_text.key
-    ) as channel_values,
-    (
-        select
-        array_agg(array[cw.task_id::text::bytea, cw.channel::bytea, cw.type::bytea, cw.blob] order by cw.task_id, cw.idx)
-        from checkpoint_writes cw
-        where cw.thread_id = checkpoints.thread_id
-            and cw.checkpoint_ns = checkpoints.checkpoint_ns
-            and cw.checkpoint_id = (checkpoint->>'id')
-    ) as pending_writes,
-    (
-        select array_agg(array[cw.type::bytea, cw.blob] order by cw.task_path, cw.task_id, cw.idx)
-        from checkpoint_writes cw
-        where cw.thread_id = checkpoints.thread_id
-            and cw.checkpoint_ns = checkpoints.checkpoint_ns
-            and cw.channel = '{TASKS}'
-    ) as pending_sends
-from checkpoints """
-
-UPSERT_CHECKPOINT_BLOBS_SQL = """
-    INSERT INTO checkpoint_blobs (thread_id, checkpoint_ns, channel, type, blob)
-    VALUES (%s, %s, %s, %s, %s)
-    ON CONFLICT (thread_id, checkpoint_ns, channel) DO UPDATE SET
-        type = EXCLUDED.type,
-        blob = EXCLUDED.blob;
-"""
-
-UPSERT_CHECKPOINTS_SQL = """
-    INSERT INTO checkpoints (thread_id, checkpoint_ns, checkpoint, metadata)
-    VALUES (%s, %s, %s, %s)
-    ON CONFLICT (thread_id, checkpoint_ns)
-    DO UPDATE SET
-        checkpoint = EXCLUDED.checkpoint,
-        metadata = EXCLUDED.metadata;
-"""
-
-UPSERT_CHECKPOINT_WRITES_SQL = """
-    INSERT INTO checkpoint_writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, blob)
-    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
-    ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) DO UPDATE SET
-        channel = EXCLUDED.channel,
-        type = EXCLUDED.type,
-        blob = EXCLUDED.blob;
-"""
-
-INSERT_CHECKPOINT_WRITES_SQL = """
-    INSERT INTO checkpoint_writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, blob)
-    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
-    ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) DO NOTHING
-"""
-
-
-def _dump_blobs(
-    serde: SerializerProtocol,
-    thread_id: str,
-    checkpoint_ns: str,
-    values: dict[str, Any],
-    versions: ChannelVersions,
-) -> list[tuple[str, str, str, str, Optional[bytes]]]:
-    if not versions:
-        return []
-
-    return [
-        (
-            thread_id,
-            checkpoint_ns,
-            k,
-            *(serde.dumps_typed(values[k]) if k in values else ("empty", None)),
-        )
-        for k in versions
-    ]
-
-
-class ShallowPostgresSaver(BasePostgresSaver):
-    """A checkpoint saver that uses Postgres to store checkpoints.
-
-    This checkpointer ONLY stores the most recent checkpoint and does NOT retain any history.
-    It is meant to be a light-weight drop-in replacement for the PostgresSaver that
-    supports most of the LangGraph persistence functionality with the exception of time travel.
-    """
-
-    SELECT_SQL = SELECT_SQL
-    MIGRATIONS = MIGRATIONS
-    UPSERT_CHECKPOINT_BLOBS_SQL = UPSERT_CHECKPOINT_BLOBS_SQL
-    UPSERT_CHECKPOINTS_SQL = UPSERT_CHECKPOINTS_SQL
-    UPSERT_CHECKPOINT_WRITES_SQL = UPSERT_CHECKPOINT_WRITES_SQL
-    INSERT_CHECKPOINT_WRITES_SQL = INSERT_CHECKPOINT_WRITES_SQL
-
-    lock: threading.Lock
-
-    def __init__(
-        self,
-        conn: _internal.Conn,
-        pipe: Optional[Pipeline] = None,
-        serde: Optional[SerializerProtocol] = None,
-    ) -> None:
-        warnings.warn(
-            "ShallowPostgresSaver is deprecated as of version 2.0.20 and will be removed in 3.0.0. "
-            "Use PostgresSaver instead, and invoke the graph with `graph.invoke(..., checkpoint_during=False)`.",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-        super().__init__(serde=serde)
-        if isinstance(conn, ConnectionPool) and pipe is not None:
-            raise ValueError(
-                "Pipeline should be used only with a single Connection, not ConnectionPool."
-            )
-
-        self.conn = conn
-        self.pipe = pipe
-        self.lock = threading.Lock()
-        self.supports_pipeline = Capabilities().has_pipeline()
-
-    @classmethod
-    @contextmanager
-    def from_conn_string(
-        cls, conn_string: str, *, pipeline: bool = False
-    ) -> Iterator["ShallowPostgresSaver"]:
-        """Create a new ShallowPostgresSaver instance from a connection string.
-
-        Args:
-            conn_string: The Postgres connection info string.
-            pipeline: whether to use Pipeline
-
-        Returns:
-            ShallowPostgresSaver: A new ShallowPostgresSaver instance.
-        """
-        with Connection.connect(
-            conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
-        ) as conn:
-            if pipeline:
-                with conn.pipeline() as pipe:
-                    yield cls(conn, pipe)
-            else:
-                yield cls(conn)
-
-    def setup(self) -> None:
-        """Set up the checkpoint database asynchronously.
-
-        This method creates the necessary tables in the Postgres database if they don't
-        already exist and runs database migrations. It MUST be called directly by the user
-        the first time checkpointer is used.
-        """
-        with self._cursor() as cur:
-            cur.execute(self.MIGRATIONS[0])
-            results = cur.execute(
-                "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
-            )
-            row = results.fetchone()
-            if row is None:
-                version = -1
-            else:
-                version = row["v"]
-            for v, migration in zip(
-                range(version + 1, len(self.MIGRATIONS)),
-                self.MIGRATIONS[version + 1 :],
-            ):
-                cur.execute(migration)
-                cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
-        if self.pipe:
-            self.pipe.sync()
-
-    def list(
-        self,
-        config: Optional[RunnableConfig],
-        *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
-    ) -> Iterator[CheckpointTuple]:
-        """List checkpoints from the database.
-
-        This method retrieves a list of checkpoint tuples from the Postgres database based
-        on the provided config. For ShallowPostgresSaver, this method returns a list with
-        ONLY the most recent checkpoint.
-        """
-        where, args = self._search_where(config, filter, before)
-        query = self.SELECT_SQL + where
-        if limit:
-            query += f" LIMIT {limit}"
-        with self._cursor() as cur:
-            cur.execute(self.SELECT_SQL + where, args, binary=True)
-            for value in cur:
-                checkpoint = self._load_checkpoint(
-                    value["checkpoint"],
-                    value["channel_values"],
-                    value["pending_sends"],
-                )
-                yield CheckpointTuple(
-                    config={
-                        "configurable": {
-                            "thread_id": value["thread_id"],
-                            "checkpoint_ns": value["checkpoint_ns"],
-                            "checkpoint_id": checkpoint["id"],
-                        }
-                    },
-                    checkpoint=checkpoint,
-                    metadata=self._load_metadata(value["metadata"]),
-                    pending_writes=self._load_writes(value["pending_writes"]),
-                )
-
-    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
-        """Get a checkpoint tuple from the database.
-
-        This method retrieves a checkpoint tuple from the Postgres database based on the
-        provided config (matching the thread ID in the config).
-
-        Args:
-            config: The config to use for retrieving the checkpoint.
-
-        Returns:
-            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
-
-        Examples:
-
-            Basic:
-            >>> config = {"configurable": {"thread_id": "1"}}
-            >>> checkpoint_tuple = memory.get_tuple(config)
-            >>> print(checkpoint_tuple)
-            CheckpointTuple(...)
-
-            With timestamp:
-
-            >>> config = {
-            ...    "configurable": {
-            ...        "thread_id": "1",
-            ...        "checkpoint_ns": "",
-            ...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
-            ...    }
-            ... }
-            >>> checkpoint_tuple = memory.get_tuple(config)
-            >>> print(checkpoint_tuple)
-            CheckpointTuple(...)
-        """  # noqa
-        thread_id = config["configurable"]["thread_id"]
-        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
-        args = (thread_id, checkpoint_ns)
-        where = "WHERE thread_id = %s AND checkpoint_ns = %s"
-
-        with self._cursor() as cur:
-            cur.execute(
-                self.SELECT_SQL + where,
-                args,
-                binary=True,
-            )
-
-            for value in cur:
-                checkpoint = self._load_checkpoint(
-                    value["checkpoint"],
-                    value["channel_values"],
-                    value["pending_sends"],
-                )
-                return CheckpointTuple(
-                    config={
-                        "configurable": {
-                            "thread_id": thread_id,
-                            "checkpoint_ns": checkpoint_ns,
-                            "checkpoint_id": checkpoint["id"],
-                        }
-                    },
-                    checkpoint=checkpoint,
-                    metadata=self._load_metadata(value["metadata"]),
-                    pending_writes=self._load_writes(value["pending_writes"]),
-                )
-
-    def put(
-        self,
-        config: RunnableConfig,
-        checkpoint: Checkpoint,
-        metadata: CheckpointMetadata,
-        new_versions: ChannelVersions,
-    ) -> RunnableConfig:
-        """Save a checkpoint to the database.
-
-        This method saves a checkpoint to the Postgres database. The checkpoint is associated
-        with the provided config. For ShallowPostgresSaver, this method saves ONLY the most recent
-        checkpoint and overwrites a previous checkpoint, if it exists.
-
-        Args:
-            config: The config to associate with the checkpoint.
-            checkpoint: The checkpoint to save.
-            metadata: Additional metadata to save with the checkpoint.
-            new_versions: New channel versions as of this write.
-
-        Returns:
-            RunnableConfig: Updated configuration after storing the checkpoint.
-
-        Examples:
-
-            >>> from langgraph.checkpoint.postgres import ShallowPostgresSaver
-            >>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
-            >>> with ShallowPostgresSaver.from_conn_string(DB_URI) as memory:
-            >>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
-            >>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
-            >>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
-            >>> print(saved_config)
-            {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}
-        """
-        configurable = config["configurable"].copy()
-        thread_id = configurable.pop("thread_id")
-        checkpoint_ns = configurable.pop("checkpoint_ns")
-
-        copy = checkpoint.copy()
-        next_config = {
-            "configurable": {
-                "thread_id": thread_id,
-                "checkpoint_ns": checkpoint_ns,
-                "checkpoint_id": checkpoint["id"],
-            }
-        }
-
-        with self._cursor(pipeline=True) as cur:
-            cur.execute(
-                """DELETE FROM checkpoint_writes
-                WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id NOT IN (%s, %s)""",
-                (
-                    thread_id,
-                    checkpoint_ns,
-                    checkpoint["id"],
-                    configurable.get("checkpoint_id", ""),
-                ),
-            )
-            cur.executemany(
-                self.UPSERT_CHECKPOINT_BLOBS_SQL,
-                _dump_blobs(
-                    self.serde,
-                    thread_id,
-                    checkpoint_ns,
-                    copy.pop("channel_values"),  # type: ignore[misc]
-                    new_versions,
-                ),
-            )
-            cur.execute(
-                self.UPSERT_CHECKPOINTS_SQL,
-                (
-                    thread_id,
-                    checkpoint_ns,
-                    Jsonb(self._dump_checkpoint(copy)),
-                    self._dump_metadata(get_checkpoint_metadata(config, metadata)),
-                ),
-            )
-        return next_config
-
-    def put_writes(
-        self,
-        config: RunnableConfig,
-        writes: Sequence[tuple[str, Any]],
-        task_id: str,
-        task_path: str = "",
-    ) -> None:
-        """Store intermediate writes linked to a checkpoint.
-
-        This method saves intermediate writes associated with a checkpoint to the Postgres database.
-
-        Args:
-            config: Configuration of the related checkpoint.
-            writes: List of writes to store.
-            task_id: Identifier for the task creating the writes.
-        """
-        query = (
-            self.UPSERT_CHECKPOINT_WRITES_SQL
-            if all(w[0] in WRITES_IDX_MAP for w in writes)
-            else self.INSERT_CHECKPOINT_WRITES_SQL
-        )
-        with self._cursor(pipeline=True) as cur:
-            cur.executemany(
-                query,
-                self._dump_writes(
-                    config["configurable"]["thread_id"],
-                    config["configurable"]["checkpoint_ns"],
-                    config["configurable"]["checkpoint_id"],
-                    task_id,
-                    task_path,
-                    writes,
-                ),
-            )
-
-    @contextmanager
-    def _cursor(self, *, pipeline: bool = False) -> Iterator[Cursor[DictRow]]:
-        """Create a database cursor as a context manager.
-
-        Args:
-            pipeline: whether to use pipeline for the DB operations inside the context manager.
-                Will be applied regardless of whether the ShallowPostgresSaver instance was initialized with a pipeline.
-                If pipeline mode is not supported, will fall back to using transaction context manager.
-        """
-        with _internal.get_connection(self.conn) as conn:
-            if self.pipe:
-                # a connection in pipeline mode can be used concurrently
-                # in multiple threads/coroutines, but only one cursor can be
-                # used at a time
-                try:
-                    with conn.cursor(binary=True, row_factory=dict_row) as cur:
-                        yield cur
-                finally:
-                    if pipeline:
-                        self.pipe.sync()
-            elif pipeline:
-                # a connection not in pipeline mode can only be used by one
-                # thread/coroutine at a time, so we acquire a lock
-                if self.supports_pipeline:
-                    with (
-                        self.lock,
-                        conn.pipeline(),
-                        conn.cursor(binary=True, row_factory=dict_row) as cur,
-                    ):
-                        yield cur
-                else:
-                    # Use connection's transaction context manager when pipeline mode not supported
-                    with (
-                        self.lock,
-                        conn.transaction(),
-                        conn.cursor(binary=True, row_factory=dict_row) as cur,
-                    ):
-                        yield cur
-            else:
-                with self.lock, conn.cursor(binary=True, row_factory=dict_row) as cur:
-                    yield cur
-
-
-class AsyncShallowPostgresSaver(BasePostgresSaver):
-    """A checkpoint saver that uses Postgres to store checkpoints asynchronously.
-
-    This checkpointer ONLY stores the most recent checkpoint and does NOT retain any history.
-    It is meant to be a light-weight drop-in replacement for the AsyncPostgresSaver that
-    supports most of the LangGraph persistence functionality with the exception of time travel.
-    """
-
-    SELECT_SQL = SELECT_SQL
-    MIGRATIONS = MIGRATIONS
-    UPSERT_CHECKPOINT_BLOBS_SQL = UPSERT_CHECKPOINT_BLOBS_SQL
-    UPSERT_CHECKPOINTS_SQL = UPSERT_CHECKPOINTS_SQL
-    UPSERT_CHECKPOINT_WRITES_SQL = UPSERT_CHECKPOINT_WRITES_SQL
-    INSERT_CHECKPOINT_WRITES_SQL = INSERT_CHECKPOINT_WRITES_SQL
-    lock: asyncio.Lock
-
-    def __init__(
-        self,
-        conn: _ainternal.Conn,
-        pipe: Optional[AsyncPipeline] = None,
-        serde: Optional[SerializerProtocol] = None,
-    ) -> None:
-        warnings.warn(
-            "AsyncShallowPostgresSaver is deprecated as of version 2.0.20 and will be removed in 3.0.0. "
-            "Use AsyncPostgresSaver instead, and invoke the graph with `await graph.ainvoke(..., checkpoint_during=False)`.",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-        super().__init__(serde=serde)
-        if isinstance(conn, AsyncConnectionPool) and pipe is not None:
-            raise ValueError(
-                "Pipeline should be used only with a single AsyncConnection, not AsyncConnectionPool."
-            )
-
-        self.conn = conn
-        self.pipe = pipe
-        self.lock = asyncio.Lock()
-        self.loop = asyncio.get_running_loop()
-        self.supports_pipeline = Capabilities().has_pipeline()
-
-    @classmethod
-    @asynccontextmanager
-    async def from_conn_string(
-        cls,
-        conn_string: str,
-        *,
-        pipeline: bool = False,
-        serde: Optional[SerializerProtocol] = None,
-    ) -> AsyncIterator["AsyncShallowPostgresSaver"]:
-        """Create a new AsyncShallowPostgresSaver instance from a connection string.
-
-        Args:
-            conn_string: The Postgres connection info string.
-            pipeline: whether to use AsyncPipeline
-
-        Returns:
-            AsyncShallowPostgresSaver: A new AsyncShallowPostgresSaver instance.
-        """
-        async with await AsyncConnection.connect(
-            conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
-        ) as conn:
-            if pipeline:
-                async with conn.pipeline() as pipe:
-                    yield cls(conn=conn, pipe=pipe, serde=serde)
-            else:
-                yield cls(conn=conn, serde=serde)
-
-    async def setup(self) -> None:
-        """Set up the checkpoint database asynchronously.
-
-        This method creates the necessary tables in the Postgres database if they don't
-        already exist and runs database migrations. It MUST be called directly by the user
-        the first time checkpointer is used.
-        """
-        async with self._cursor() as cur:
-            await cur.execute(self.MIGRATIONS[0])
-            results = await cur.execute(
-                "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
-            )
-            row = await results.fetchone()
-            if row is None:
-                version = -1
-            else:
-                version = row["v"]
-            for v, migration in zip(
-                range(version + 1, len(self.MIGRATIONS)),
-                self.MIGRATIONS[version + 1 :],
-            ):
-                await cur.execute(migration)
-                await cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
-        if self.pipe:
-            await self.pipe.sync()
-
-    async def alist(
-        self,
-        config: Optional[RunnableConfig],
-        *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
-    ) -> AsyncIterator[CheckpointTuple]:
-        """List checkpoints from the database asynchronously.
-
-        This method retrieves a list of checkpoint tuples from the Postgres database based
-        on the provided config. For ShallowPostgresSaver, this method returns a list with
-        ONLY the most recent checkpoint.
-        """
-        where, args = self._search_where(config, filter, before)
-        query = self.SELECT_SQL + where
-        if limit:
-            query += f" LIMIT {limit}"
-        async with self._cursor() as cur:
-            await cur.execute(self.SELECT_SQL + where, args, binary=True)
-            async for value in cur:
-                checkpoint = await asyncio.to_thread(
-                    self._load_checkpoint,
-                    value["checkpoint"],
-                    value["channel_values"],
-                    value["pending_sends"],
-                )
-                yield CheckpointTuple(
-                    config={
-                        "configurable": {
-                            "thread_id": value["thread_id"],
-                            "checkpoint_ns": value["checkpoint_ns"],
-                            "checkpoint_id": checkpoint["id"],
-                        }
-                    },
-                    checkpoint=checkpoint,
-                    metadata=self._load_metadata(value["metadata"]),
-                    pending_writes=await asyncio.to_thread(
-                        self._load_writes, value["pending_writes"]
-                    ),
-                )
-
-    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
-        """Get a checkpoint tuple from the database asynchronously.
-
-        This method retrieves a checkpoint tuple from the Postgres database based on the
-        provided config (matching the thread ID in the config).
-
-        Args:
-            config: The config to use for retrieving the checkpoint.
-
-        Returns:
-            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
-        """
-        thread_id = config["configurable"]["thread_id"]
-        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
-        args = (thread_id, checkpoint_ns)
-        where = "WHERE thread_id = %s AND checkpoint_ns = %s"
-
-        async with self._cursor() as cur:
-            await cur.execute(
-                self.SELECT_SQL + where,
-                args,
-                binary=True,
-            )
-
-            async for value in cur:
-                checkpoint = await asyncio.to_thread(
-                    self._load_checkpoint,
-                    value["checkpoint"],
-                    value["channel_values"],
-                    value["pending_sends"],
-                )
-                return CheckpointTuple(
-                    config={
-                        "configurable": {
-                            "thread_id": thread_id,
-                            "checkpoint_ns": checkpoint_ns,
-                            "checkpoint_id": checkpoint["id"],
-                        }
-                    },
-                    checkpoint=checkpoint,
-                    metadata=self._load_metadata(value["metadata"]),
-                    pending_writes=await asyncio.to_thread(
-                        self._load_writes, value["pending_writes"]
-                    ),
-                )
-
-    async def aput(
-        self,
-        config: RunnableConfig,
-        checkpoint: Checkpoint,
-        metadata: CheckpointMetadata,
-        new_versions: ChannelVersions,
-    ) -> RunnableConfig:
-        """Save a checkpoint to the database asynchronously.
-
-        This method saves a checkpoint to the Postgres database. The checkpoint is associated
-        with the provided config. For AsyncShallowPostgresSaver, this method saves ONLY the most recent
-        checkpoint and overwrites a previous checkpoint, if it exists.
-
-        Args:
-            config: The config to associate with the checkpoint.
-            checkpoint: The checkpoint to save.
-            metadata: Additional metadata to save with the checkpoint.
-            new_versions: New channel versions as of this write.
-
-        Returns:
-            RunnableConfig: Updated configuration after storing the checkpoint.
-        """
-        configurable = config["configurable"].copy()
-        thread_id = configurable.pop("thread_id")
-        checkpoint_ns = configurable.pop("checkpoint_ns")
-
-        copy = checkpoint.copy()
-        next_config = {
-            "configurable": {
-                "thread_id": thread_id,
-                "checkpoint_ns": checkpoint_ns,
-                "checkpoint_id": checkpoint["id"],
-            }
-        }
-
-        async with self._cursor(pipeline=True) as cur:
-            await cur.execute(
-                """DELETE FROM checkpoint_writes
-                WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id NOT IN (%s, %s)""",
-                (
-                    thread_id,
-                    checkpoint_ns,
-                    checkpoint["id"],
-                    configurable.get("checkpoint_id", ""),
-                ),
-            )
-            await cur.executemany(
-                self.UPSERT_CHECKPOINT_BLOBS_SQL,
-                _dump_blobs(
-                    self.serde,
-                    thread_id,
-                    checkpoint_ns,
-                    copy.pop("channel_values"),  # type: ignore[misc]
-                    new_versions,
-                ),
-            )
-            await cur.execute(
-                self.UPSERT_CHECKPOINTS_SQL,
-                (
-                    thread_id,
-                    checkpoint_ns,
-                    Jsonb(self._dump_checkpoint(copy)),
-                    self._dump_metadata(get_checkpoint_metadata(config, metadata)),
-                ),
-            )
-        return next_config
-
-    async def aput_writes(
-        self,
-        config: RunnableConfig,
-        writes: Sequence[tuple[str, Any]],
-        task_id: str,
-        task_path: str = "",
-    ) -> None:
-        """Store intermediate writes linked to a checkpoint asynchronously.
-
-        This method saves intermediate writes associated with a checkpoint to the database.
-
-        Args:
-            config: Configuration of the related checkpoint.
-            writes: List of writes to store, each as (channel, value) pair.
-            task_id: Identifier for the task creating the writes.
-        """
-        query = (
-            self.UPSERT_CHECKPOINT_WRITES_SQL
-            if all(w[0] in WRITES_IDX_MAP for w in writes)
-            else self.INSERT_CHECKPOINT_WRITES_SQL
-        )
-        params = await asyncio.to_thread(
-            self._dump_writes,
-            config["configurable"]["thread_id"],
-            config["configurable"]["checkpoint_ns"],
-            config["configurable"]["checkpoint_id"],
-            task_id,
-            task_path,
-            writes,
-        )
-        async with self._cursor(pipeline=True) as cur:
-            await cur.executemany(query, params)
-
-    @asynccontextmanager
-    async def _cursor(
-        self, *, pipeline: bool = False
-    ) -> AsyncIterator[AsyncCursor[DictRow]]:
-        """Create a database cursor as a context manager.
-
-        Args:
-            pipeline: whether to use pipeline for the DB operations inside the context manager.
-                Will be applied regardless of whether the AsyncShallowPostgresSaver instance was initialized with a pipeline.
-                If pipeline mode is not supported, will fall back to using transaction context manager.
-        """
-        async with _ainternal.get_connection(self.conn) as conn:
-            if self.pipe:
-                # a connection in pipeline mode can be used concurrently
-                # in multiple threads/coroutines, but only one cursor can be
-                # used at a time
-                try:
-                    async with conn.cursor(binary=True, row_factory=dict_row) as cur:
-                        yield cur
-                finally:
-                    if pipeline:
-                        await self.pipe.sync()
-            elif pipeline:
-                # a connection not in pipeline mode can only be used by one
-                # thread/coroutine at a time, so we acquire a lock
-                if self.supports_pipeline:
-                    async with (
-                        self.lock,
-                        conn.pipeline(),
-                        conn.cursor(binary=True, row_factory=dict_row) as cur,
-                    ):
-                        yield cur
-                else:
-                    # Use connection's transaction context manager when pipeline mode not supported
-                    async with (
-                        self.lock,
-                        conn.transaction(),
-                        conn.cursor(binary=True, row_factory=dict_row) as cur,
-                    ):
-                        yield cur
-            else:
-                async with (
-                    self.lock,
-                    conn.cursor(binary=True, row_factory=dict_row) as cur,
-                ):
-                    yield cur
-
-    def list(
-        self,
-        config: Optional[RunnableConfig],
-        *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
-    ) -> Iterator[CheckpointTuple]:
-        """List checkpoints from the database.
-
-        This method retrieves a list of checkpoint tuples from the Postgres database based
-        on the provided config. For ShallowPostgresSaver, this method returns a list with
-        ONLY the most recent checkpoint.
-        """
-        aiter_ = self.alist(config, filter=filter, before=before, limit=limit)
-        while True:
-            try:
-                yield asyncio.run_coroutine_threadsafe(
-                    anext(aiter_),  # noqa: F821
-                    self.loop,
-                ).result()
-            except StopAsyncIteration:
-                break
-
-    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
-        """Get a checkpoint tuple from the database.
-
-        This method retrieves a checkpoint tuple from the Postgres database based on the
-        provided config (matching the thread ID in the config).
-
-        Args:
-            config: The config to use for retrieving the checkpoint.
-
-        Returns:
-            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
-        """
-        try:
-            # check if we are in the main thread, only bg threads can block
-            # we don't check in other methods to avoid the overhead
-            if asyncio.get_running_loop() is self.loop:
-                raise asyncio.InvalidStateError(
-                    "Synchronous calls to AsyncShallowPostgresSaver are only allowed from a "
-                    "different thread. From the main thread, use the async interface."
-                    "For example, use `await checkpointer.aget_tuple(...)` or `await "
-                    "graph.ainvoke(...)`."
-                )
-        except RuntimeError:
-            pass
-        return asyncio.run_coroutine_threadsafe(
-            self.aget_tuple(config), self.loop
-        ).result()
-
-    def put(
-        self,
-        config: RunnableConfig,
-        checkpoint: Checkpoint,
-        metadata: CheckpointMetadata,
-        new_versions: ChannelVersions,
-    ) -> RunnableConfig:
-        """Save a checkpoint to the database.
-
-        This method saves a checkpoint to the Postgres database. The checkpoint is associated
-        with the provided config. For AsyncShallowPostgresSaver, this method saves ONLY the most recent
-        checkpoint and overwrites a previous checkpoint, if it exists.
-
-        Args:
-            config: The config to associate with the checkpoint.
-            checkpoint: The checkpoint to save.
-            metadata: Additional metadata to save with the checkpoint.
-            new_versions: New channel versions as of this write.
-
-        Returns:
-            RunnableConfig: Updated configuration after storing the checkpoint.
-        """
-        return asyncio.run_coroutine_threadsafe(
-            self.aput(config, checkpoint, metadata, new_versions), self.loop
-        ).result()
-
-    def put_writes(
-        self,
-        config: RunnableConfig,
-        writes: Sequence[tuple[str, Any]],
-        task_id: str,
-        task_path: str = "",
-    ) -> None:
-        """Store intermediate writes linked to a checkpoint.
-
-        This method saves intermediate writes associated with a checkpoint to the database.
-
-        Args:
-            config: Configuration of the related checkpoint.
-            writes: List of writes to store, each as (channel, value) pair.
-            task_id: Identifier for the task creating the writes.
-            task_path: Path of the task creating the writes.
-        """
-        return asyncio.run_coroutine_threadsafe(
-            self.aput_writes(config, writes, task_id, task_path), self.loop
-        ).result()

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

@@ -1,9 +1,11 @@
+from __future__ import annotations
+
 import asyncio
 import logging
 from collections.abc import AsyncIterator, Iterable, Sequence
 from contextlib import asynccontextmanager
 from types import TracebackType
-from typing import Any, Callable, Optional, Union, cast
+from typing import Any, Callable, cast
 
 import orjson
 from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
@@ -132,12 +134,10 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
         self,
         conn: _ainternal.Conn,
         *,
-        pipe: Optional[AsyncPipeline] = None,
-        deserializer: Optional[
-            Callable[[Union[bytes, orjson.Fragment]], dict[str, Any]]
-        ] = None,
-        index: Optional[PostgresIndexConfig] = None,
-        ttl: Optional[TTLConfig] = None,
+        pipe: AsyncPipeline | None = None,
+        deserializer: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None = None,
+        index: PostgresIndexConfig | None = None,
+        ttl: TTLConfig | None = None,
     ) -> None:
         if isinstance(conn, AsyncConnectionPool) and pipe is not None:
             raise ValueError(
@@ -157,7 +157,7 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
             self.embeddings = None
 
         self.ttl_config = ttl
-        self._ttl_sweeper_task: Optional[asyncio.Task[None]] = None
+        self._ttl_sweeper_task: asyncio.Task[None] | None = None
         self._ttl_stop_event = asyncio.Event()
 
     async def abatch(self, ops: Iterable[Op]) -> list[Result]:
@@ -180,10 +180,10 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
         conn_string: str,
         *,
         pipeline: bool = False,
-        pool_config: Optional[PoolConfig] = None,
-        index: Optional[PostgresIndexConfig] = None,
-        ttl: Optional[TTLConfig] = None,
-    ) -> AsyncIterator["AsyncPostgresStore"]:
+        pool_config: PoolConfig | None = None,
+        index: PostgresIndexConfig | None = None,
+        ttl: TTLConfig | None = None,
+    ) -> AsyncIterator[AsyncPostgresStore]:
         """Create a new AsyncPostgresStore instance from a connection string.
 
         Args:
@@ -289,7 +289,7 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
             return deleted_count
 
     async def start_ttl_sweeper(
-        self, sweep_interval_minutes: Optional[int] = None
+        self, sweep_interval_minutes: int | None = None
     ) -> asyncio.Task[None]:
         """Periodically delete expired store items based on TTL.
 
@@ -334,7 +334,7 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
         self._ttl_sweeper_task = task
         return task
 
-    async def stop_ttl_sweeper(self, timeout: Optional[float] = None) -> bool:
+    async def stop_ttl_sweeper(self, timeout: float | None = None) -> bool:
         """Stop the TTL sweeper task if it's running.
 
         Args:
@@ -369,14 +369,14 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
 
         return success
 
-    async def __aenter__(self) -> "AsyncPostgresStore":
+    async def __aenter__(self) -> AsyncPostgresStore:
         return self
 
     async def __aexit__(
         self,
-        exc_type: Optional[type[BaseException]],
-        exc_val: Optional[BaseException],
-        exc_tb: Optional["TracebackType"],
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
     ) -> None:
         # Ensure the TTL sweeper task is stopped when exiting the context
         if hasattr(self, "_ttl_sweeper_task") and self._ttl_sweeper_task is not None:

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

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import asyncio
 import concurrent.futures
 import json
@@ -14,7 +16,6 @@ from typing import (
     Generic,
     Literal,
     NamedTuple,
-    Optional,
     TypeVar,
     Union,
     cast,
@@ -56,8 +57,8 @@ class Migration(NamedTuple):
     """A database migration with optional conditions and parameters."""
 
     sql: str
-    params: Optional[dict[str, Any]] = None
-    condition: Optional[Callable[["BasePostgresStore"], bool]] = None
+    params: dict[str, Any] | None = None
+    condition: Callable[[BasePostgresStore], bool] | None = None
 
 
 MIGRATIONS: Sequence[str] = [
@@ -155,7 +156,7 @@ class PoolConfig(TypedDict, total=False):
     min_size: int
     """Minimum number of connections maintained in the pool. Defaults to 1."""
 
-    max_size: Optional[int]
+    max_size: int | None
     """Maximum number of connections allowed in the pool. None means unlimited."""
 
     kwargs: dict
@@ -230,8 +231,8 @@ class BasePostgresStore(Generic[C]):
     MIGRATIONS = MIGRATIONS
     VECTOR_MIGRATIONS = VECTOR_MIGRATIONS
     conn: C
-    _deserializer: Optional[Callable[[Union[bytes, orjson.Fragment]], dict[str, Any]]]
-    index_config: Optional[PostgresIndexConfig]
+    _deserializer: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None
+    index_config: PostgresIndexConfig | None
 
     def _get_batch_GET_ops_queries(
         self,
@@ -293,7 +294,7 @@ class BasePostgresStore(Generic[C]):
         put_ops: Sequence[tuple[int, PutOp]],
     ) -> tuple[
         list[tuple[str, Sequence]],
-        Optional[tuple[str, Sequence[tuple[str, str, str, str]]]],
+        tuple[str, Sequence[tuple[str, str, str, str]]] | None,
     ]:
         dedupped_ops: dict[tuple[tuple[str, ...], str], PutOp] = {}
         for _, op in put_ops:
@@ -320,9 +321,7 @@ class BasePostgresStore(Generic[C]):
                 )
                 params = (_namespace_to_text(namespace), *keys)
                 queries.append((query, params))
-        embedding_request: Optional[tuple[str, Sequence[tuple[str, str, str, str]]]] = (
-            None
-        )
+        embedding_request: tuple[str, Sequence[tuple[str, str, str, str]]] | None = None
         if inserts:
             values = []
             insertion_params = []
@@ -403,7 +402,7 @@ class BasePostgresStore(Generic[C]):
         self,
         search_ops: Sequence[tuple[int, SearchOp]],
     ) -> tuple[
-        list[tuple[str, list[Union[None, str, list[float]]]]],  # queries, params
+        list[tuple[str, list[None | str | list[float]]]],  # queries, params
         list[tuple[int, str]],  # idx, query_text pairs to embed
     ]:
         """
@@ -432,7 +431,7 @@ class BasePostgresStore(Generic[C]):
                         filter_params.extend([key, orjson.dumps(value).decode("utf-8")])
 
             ns_condition = "TRUE"
-            ns_param: Optional[Sequence[Union[str]]] = None
+            ns_param: Sequence[str] | None = None
             if op.namespace_prefix:
                 ns_condition = "store.prefix LIKE %s"
                 ns_param = (f"{_namespace_to_text(op.namespace_prefix)}%",)
@@ -719,12 +718,10 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
         self,
         conn: _pg_internal.Conn,
         *,
-        pipe: Optional[Pipeline] = None,
-        deserializer: Optional[
-            Callable[[Union[bytes, orjson.Fragment]], dict[str, Any]]
-        ] = None,
-        index: Optional[PostgresIndexConfig] = None,
-        ttl: Optional[TTLConfig] = None,
+        pipe: Pipeline | None = None,
+        deserializer: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None = None,
+        index: PostgresIndexConfig | None = None,
+        ttl: TTLConfig | None = None,
     ) -> None:
         super().__init__()
         self._deserializer = deserializer
@@ -738,7 +735,7 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
         else:
             self.embeddings = None
         self.ttl_config = ttl
-        self._ttl_sweeper_thread: Optional[threading.Thread] = None
+        self._ttl_sweeper_thread: threading.Thread | None = None
         self._ttl_stop_event = threading.Event()
 
     @classmethod
@@ -748,10 +745,10 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
         conn_string: str,
         *,
         pipeline: bool = False,
-        pool_config: Optional[PoolConfig] = None,
-        index: Optional[PostgresIndexConfig] = None,
-        ttl: Optional[TTLConfig] = None,
-    ) -> Iterator["PostgresStore"]:
+        pool_config: PoolConfig | None = None,
+        index: PostgresIndexConfig | None = None,
+        ttl: TTLConfig | None = None,
+    ) -> Iterator[PostgresStore]:
         """Create a new PostgresStore instance from a connection string.
 
         Args:
@@ -810,7 +807,7 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
             return deleted_count
 
     def start_ttl_sweeper(
-        self, sweep_interval_minutes: Optional[int] = None
+        self, sweep_interval_minutes: int | None = None
     ) -> concurrent.futures.Future[None]:
         """Periodically delete expired store items based on TTL.
 
@@ -867,7 +864,7 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
         )
         return future
 
-    def stop_ttl_sweeper(self, timeout: Optional[float] = None) -> bool:
+    def stop_ttl_sweeper(self, timeout: float | None = None) -> bool:
         """Stop the TTL sweeper thread if it's running.
 
         Args:
@@ -1196,7 +1193,7 @@ def _row_to_item(
     namespace: tuple[str, ...],
     row: Row,
     *,
-    loader: Optional[Callable[[Union[bytes, orjson.Fragment]], dict[str, Any]]] = None,
+    loader: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None = None,
 ) -> Item:
     """Convert a row from the database into an Item.
 
@@ -1224,7 +1221,7 @@ def _row_to_search_item(
     namespace: tuple[str, ...],
     row: Row,
     *,
-    loader: Optional[Callable[[Union[bytes, orjson.Fragment]], dict[str, Any]]] = None,
+    loader: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None = None,
 ) -> SearchItem:
     """Convert a row from the database into an Item."""
     loader = loader or _json_loads
@@ -1255,7 +1252,7 @@ def _group_ops(ops: Iterable[Op]) -> tuple[dict[type, list[tuple[int, Op]]], int
     return grouped_ops, tot
 
 
-def _json_loads(content: Union[bytes, orjson.Fragment]) -> Any:
+def _json_loads(content: bytes | orjson.Fragment) -> Any:
     if isinstance(content, orjson.Fragment):
         if hasattr(content, "buf"):
             content = content.buf
@@ -1267,7 +1264,7 @@ def _json_loads(content: Union[bytes, orjson.Fragment]) -> Any:
     return orjson.loads(cast(bytes, content))
 
 
-def _decode_ns_bytes(namespace: Union[str, bytes, list]) -> tuple[str, ...]:
+def _decode_ns_bytes(namespace: str | bytes | list) -> tuple[str, ...]:
     if isinstance(namespace, list):
         return tuple(namespace)
     if isinstance(namespace, bytes):
@@ -1316,9 +1313,9 @@ def get_distance_operator(store: Any) -> tuple[str, str]:
 
 def _ensure_index_config(
     index_config: PostgresIndexConfig,
-) -> tuple[Optional["Embeddings"], PostgresIndexConfig]:
+) -> tuple[Embeddings | None, PostgresIndexConfig]:
     index_config = index_config.copy()
-    tokenized: list[tuple[str, Union[Literal["$"], list[str]]]] = []
+    tokenized: list[tuple[str, Literal["$"] | list[str]]] = []
     tot = 0
     text_fields = index_config.get("fields") or ["$"]
     if isinstance(text_fields, str):

libs/checkpoint-postgres/pyproject.toml

@@ -56,7 +56,7 @@ lint.select = [
   "B",  # flake8-bugbear
   "I",  # isort
 ]
-lint.ignore = ["E501", "B008", "UP007", "UP006"]
+lint.ignore = ["E501", "B008"]
 
 [tool.mypy]
 # https://mypy.readthedocs.io/en/stable/config_file.html

libs/checkpoint-postgres/tests/checkpoint_utils.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from datetime import datetime, timezone
+from typing import Any, Protocol
+
+from langgraph.checkpoint.base import Checkpoint, EmptyChannelError
+from langgraph.checkpoint.base.id import uuid6
+
+
+class ChannelProtocol(Protocol):
+    def checkpoint(self) -> Any | None: ...
+
+
+def empty_checkpoint() -> Checkpoint:
+    return Checkpoint(
+        v=1,
+        id=str(uuid6(clock_seq=-2)),
+        ts=datetime.now(timezone.utc).isoformat(),
+        channel_values={},
+        channel_versions={},
+        versions_seen={},
+    )
+
+
+def create_checkpoint(
+    checkpoint: Checkpoint,
+    channels: Mapping[str, ChannelProtocol] | None,
+    step: int,
+    *,
+    id: str | None = None,
+) -> Checkpoint:
+    """Create a checkpoint for the given channels."""
+    ts = datetime.now(timezone.utc).isoformat()
+    if channels is None:
+        values = checkpoint["channel_values"]
+    else:
+        values = {}
+        for k, v in channels.items():
+            if k not in checkpoint["channel_versions"]:
+                continue
+            try:
+                values[k] = v.checkpoint()
+            except EmptyChannelError:
+                pass
+    return Checkpoint(
+        v=1,
+        ts=ts,
+        id=id or str(uuid6(clock_seq=step)),
+        channel_values=values,
+        channel_versions=checkpoint["channel_versions"],
+        versions_seen=checkpoint["versions_seen"],
+    )

libs/checkpoint-postgres/tests/test_async.py

@@ -14,13 +14,10 @@ from langgraph.checkpoint.base import (
     EXCLUDED_METADATA_KEYS,
     Checkpoint,
     CheckpointMetadata,
-    create_checkpoint,
-    empty_checkpoint,
-)
-from langgraph.checkpoint.postgres.aio import (
-    AsyncPostgresSaver,
-    AsyncShallowPostgresSaver,
 )
+from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
+from langgraph.checkpoint.serde.types import TASKS
+from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
 from tests.conftest import DEFAULT_POSTGRES_URI
 
 
@@ -111,41 +108,11 @@ async def _base_saver():
             await conn.execute(f"DROP DATABASE {database}")
 
 
-@asynccontextmanager
-async def _shallow_saver():
-    """Fixture for shallow connection mode testing."""
-    database = f"test_{uuid4().hex[:16]}"
-    # create unique db
-    async with await AsyncConnection.connect(
-        DEFAULT_POSTGRES_URI, autocommit=True
-    ) as conn:
-        await conn.execute(f"CREATE DATABASE {database}")
-    try:
-        async with await AsyncConnection.connect(
-            DEFAULT_POSTGRES_URI + database,
-            autocommit=True,
-            prepare_threshold=0,
-            row_factory=dict_row,
-        ) as conn:
-            checkpointer = AsyncShallowPostgresSaver(conn)
-            await checkpointer.setup()
-            yield checkpointer
-    finally:
-        # drop unique db
-        async with await AsyncConnection.connect(
-            DEFAULT_POSTGRES_URI, autocommit=True
-        ) as conn:
-            await conn.execute(f"DROP DATABASE {database}")
-
-
 @asynccontextmanager
 async def _saver(name: str):
     if name == "base":
         async with _base_saver() as saver:
             yield saver
-    elif name == "shallow":
-        async with _shallow_saver() as saver:
-            yield saver
     elif name == "pool":
         async with _pool_saver() as saver:
             yield saver
@@ -205,7 +172,7 @@ def test_data():
     }
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
 async def test_combined_metadata(saver_name: str, test_data) -> None:
     async with _saver(saver_name) as saver:
         config = {
@@ -232,7 +199,7 @@ async def test_combined_metadata(saver_name: str, test_data) -> None:
         }
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
 async def test_asearch(saver_name: str, test_data) -> None:
     async with _saver(saver_name) as saver:
         configs = test_data["configs"]
@@ -283,7 +250,7 @@ async def test_asearch(saver_name: str, test_data) -> None:
         } == {"", "inner"}
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
 async def test_null_chars(saver_name: str, test_data) -> None:
     async with _saver(saver_name) as saver:
         config = await saver.aput(
@@ -296,3 +263,52 @@ async def test_null_chars(saver_name: str, test_data) -> None:
         assert [c async for c in saver.alist(None, filter={"my_key": "abc"})][
             0
         ].metadata["my_key"] == "abc"
+
+
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+async def test_pending_sends_migration(saver_name: str) -> None:
+    async with _saver(saver_name) as saver:
+        config = {
+            "configurable": {
+                "thread_id": "thread-1",
+                "checkpoint_ns": "",
+            }
+        }
+
+        # create the first checkpoint
+        # and put some pending sends
+        checkpoint_0 = empty_checkpoint()
+        config = await saver.aput(config, checkpoint_0, {}, {})
+        await saver.aput_writes(
+            config, [(TASKS, "send-1"), (TASKS, "send-2")], task_id="task-1"
+        )
+        await saver.aput_writes(config, [(TASKS, "send-3")], task_id="task-2")
+
+        # check that fetching checkpoint_0 doesn't attach pending sends
+        # (they should be attached to the next checkpoint)
+        tuple_0 = await saver.aget_tuple(config)
+        assert tuple_0.checkpoint["channel_values"] == {}
+        assert tuple_0.checkpoint["channel_versions"] == {}
+
+        # create the second checkpoint
+        checkpoint_1 = create_checkpoint(checkpoint_0, {}, 1)
+        config = await saver.aput(config, checkpoint_1, {}, {})
+
+        # check that pending sends are attached to checkpoint_1
+        tuple_1 = await saver.aget_tuple(config)
+        assert tuple_1.checkpoint["channel_values"] == {
+            TASKS: ["send-1", "send-2", "send-3"]
+        }
+        assert TASKS in tuple_1.checkpoint["channel_versions"]
+
+        # check that list also applies the migration
+        search_results = [
+            c async for c in saver.alist({"configurable": {"thread_id": "thread-1"}})
+        ]
+        assert len(search_results) == 2
+        assert search_results[-1].checkpoint["channel_values"] == {}
+        assert search_results[-1].checkpoint["channel_versions"] == {}
+        assert search_results[0].checkpoint["channel_values"] == {
+            TASKS: ["send-1", "send-2", "send-3"]
+        }
+        assert TASKS in search_results[0].checkpoint["channel_versions"]

libs/checkpoint-postgres/tests/test_async_store.py

@@ -1,4 +1,6 @@
 # type: ignore
+from __future__ import annotations
+
 import asyncio
 import itertools
 import sys
@@ -6,7 +8,7 @@ import uuid
 from collections.abc import AsyncIterator
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import asynccontextmanager
-from typing import Any, Optional
+from typing import Any
 
 import pytest
 from langchain_core.embeddings import Embeddings
@@ -353,7 +355,7 @@ async def _create_vector_store(
     vector_type: str,
     distance_type: str,
     fake_embeddings: CharacterEmbeddings,
-    text_fields: Optional[list[str]] = None,
+    text_fields: list[str] | None = None,
 ) -> AsyncIterator[AsyncPostgresStore]:
     """Create a store with vector search enabled."""
     if sys.version_info < (3, 10):

libs/checkpoint-postgres/tests/test_store.py

@@ -1,9 +1,10 @@
 # type: ignore
+from __future__ import annotations
 
 import re
 import time
 from contextlib import contextmanager
-from typing import Any, Optional
+from typing import Any
 from uuid import uuid4
 
 import pytest
@@ -379,7 +380,7 @@ def _create_vector_store(
     vector_type: str,
     distance_type: str,
     fake_embeddings: Embeddings,
-    text_fields: Optional[list[str]] = None,
+    text_fields: list[str] | None = None,
     enable_ttl: bool = True,
 ) -> PostgresStore:
     """Create a store with vector search enabled."""

libs/checkpoint-postgres/tests/test_sync.py

@@ -15,10 +15,10 @@ from langgraph.checkpoint.base import (
     EXCLUDED_METADATA_KEYS,
     Checkpoint,
     CheckpointMetadata,
-    create_checkpoint,
-    empty_checkpoint,
 )
-from langgraph.checkpoint.postgres import PostgresSaver, ShallowPostgresSaver
+from langgraph.checkpoint.postgres import PostgresSaver
+from langgraph.checkpoint.serde.types import TASKS
+from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
 from tests.conftest import DEFAULT_POSTGRES_URI
 
 
@@ -97,37 +97,11 @@ def _base_saver():
             conn.execute(f"DROP DATABASE {database}")
 
 
-@contextmanager
-def _shallow_saver():
-    """Fixture for regular connection mode testing with a shallow checkpointer."""
-    database = f"test_{uuid4().hex[:16]}"
-    # create unique db
-    with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
-        conn.execute(f"CREATE DATABASE {database}")
-    try:
-        with Connection.connect(
-            DEFAULT_POSTGRES_URI + database,
-            autocommit=True,
-            prepare_threshold=0,
-            row_factory=dict_row,
-        ) as conn:
-            checkpointer = ShallowPostgresSaver(conn)
-            checkpointer.setup()
-            yield checkpointer
-    finally:
-        # drop unique db
-        with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
-            conn.execute(f"DROP DATABASE {database}")
-
-
 @contextmanager
 def _saver(name: str):
     if name == "base":
         with _base_saver() as saver:
             yield saver
-    elif name == "shallow":
-        with _shallow_saver() as saver:
-            yield saver
     elif name == "pool":
         with _pool_saver() as saver:
             yield saver
@@ -187,7 +161,7 @@ def test_data():
     }
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
 def test_combined_metadata(saver_name: str, test_data) -> None:
     with _saver(saver_name) as saver:
         config = {
@@ -214,7 +188,7 @@ def test_combined_metadata(saver_name: str, test_data) -> None:
         }
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@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"]
@@ -263,7 +237,7 @@ def test_search(saver_name: str, test_data) -> None:
         } == {"", "inner"}
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@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(
@@ -284,3 +258,52 @@ def test_nonnull_migrations() -> None:
     for migration in PostgresSaver.MIGRATIONS:
         statement = _leading_comment_remover.sub("", migration).split()[0]
         assert statement.strip()
+
+
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+def test_pending_sends_migration(saver_name: str) -> None:
+    with _saver(saver_name) as saver:
+        config = {
+            "configurable": {
+                "thread_id": "thread-1",
+                "checkpoint_ns": "",
+            }
+        }
+
+        # create the first checkpoint
+        # and put some pending sends
+        checkpoint_0 = empty_checkpoint()
+        config = saver.put(config, checkpoint_0, {}, {})
+        saver.put_writes(
+            config, [(TASKS, "send-1"), (TASKS, "send-2")], task_id="task-1"
+        )
+        saver.put_writes(config, [(TASKS, "send-3")], task_id="task-2")
+
+        # check that fetching checkpoint_0 doesn't attach pending sends
+        # (they should be attached to the next checkpoint)
+        tuple_0 = saver.get_tuple(config)
+        assert tuple_0.checkpoint["channel_values"] == {}
+        assert tuple_0.checkpoint["channel_versions"] == {}
+
+        # create the second checkpoint
+        checkpoint_1 = create_checkpoint(checkpoint_0, {}, 1)
+        config = saver.put(config, checkpoint_1, {}, {})
+
+        # check that pending sends are attached to checkpoint_1
+        checkpoint_1 = saver.get_tuple(config)
+        assert checkpoint_1.checkpoint["channel_values"] == {
+            TASKS: ["send-1", "send-2", "send-3"]
+        }
+        assert TASKS in checkpoint_1.checkpoint["channel_versions"]
+
+        # check that list also applies the migration
+        search_results = [
+            c for c in saver.list({"configurable": {"thread_id": "thread-1"}})
+        ]
+        assert len(search_results) == 2
+        assert search_results[-1].checkpoint["channel_values"] == {}
+        assert search_results[-1].checkpoint["channel_versions"] == {}
+        assert search_results[0].checkpoint["channel_values"] == {
+            TASKS: ["send-1", "send-2", "send-3"]
+        }
+        assert TASKS in search_results[0].checkpoint["channel_versions"]

libs/checkpoint-sqlite/README.md

@@ -12,7 +12,7 @@ read_config = {"configurable": {"thread_id": "1"}}
 
 with SqliteSaver.from_conn_string(":memory:") as checkpointer:
     checkpoint = {
-        "v": 2,
+        "v": 4,
         "ts": "2024-07-31T20:14:19.804150+00:00",
         "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
         "channel_values": {
@@ -34,7 +34,6 @@ with SqliteSaver.from_conn_string(":memory:") as checkpointer:
                 "start:node": 2
             }
         },
-        "pending_sends": [],
     }
 
     # store checkpoint
@@ -54,7 +53,7 @@ from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
 
 async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
     checkpoint = {
-        "v": 2,
+        "v": 4,
         "ts": "2024-07-31T20:14:19.804150+00:00",
         "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
         "channel_values": {
@@ -76,7 +75,6 @@ async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
                 "start:node": 2
             }
         },
-        "pending_sends": [],
     }
 
     # store checkpoint

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

@@ -1,9 +1,11 @@
+from __future__ import annotations
+
 import random
 import sqlite3
 import threading
 from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import closing, contextmanager
-from typing import Any, Optional, cast
+from typing import Any, cast
 
 from langchain_core.runnables import RunnableConfig
 
@@ -19,7 +21,6 @@ from langgraph.checkpoint.base import (
     get_checkpoint_metadata,
 )
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
-from langgraph.checkpoint.serde.types import ChannelProtocol
 from langgraph.checkpoint.sqlite.utils import search_where
 
 _AIO_ERROR_MSG = (
@@ -77,7 +78,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
         self,
         conn: sqlite3.Connection,
         *,
-        serde: Optional[SerializerProtocol] = None,
+        serde: SerializerProtocol | None = None,
     ) -> None:
         super().__init__(serde=serde)
         self.jsonplus_serde = JsonPlusSerializer()
@@ -87,7 +88,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
 
     @classmethod
     @contextmanager
-    def from_conn_string(cls, conn_string: str) -> Iterator["SqliteSaver"]:
+    def from_conn_string(cls, conn_string: str) -> Iterator[SqliteSaver]:
         """Create a new SqliteSaver instance from a connection string.
 
         Args:
@@ -179,7 +180,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                     self.conn.commit()
                 cur.close()
 
-    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the database.
 
         This method retrieves a checkpoint tuple from the SQLite database based on the
@@ -287,11 +288,11 @@ class SqliteSaver(BaseCheckpointSaver[str]):
 
     def list(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> Iterator[CheckpointTuple]:
         """List checkpoints from the database.
 
@@ -494,7 +495,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                 (str(thread_id),),
             )
 
-    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the database asynchronously.
 
         Note:
@@ -505,11 +506,11 @@ class SqliteSaver(BaseCheckpointSaver[str]):
 
     async def alist(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> AsyncIterator[CheckpointTuple]:
         """List checkpoints from the database asynchronously.
 
@@ -535,14 +536,13 @@ class SqliteSaver(BaseCheckpointSaver[str]):
         """
         raise NotImplementedError(_AIO_ERROR_MSG)
 
-    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
+    def get_next_version(self, current: str | None) -> str:
         """Generate the next version ID for a channel.
 
         This method creates a new version identifier for a channel based on its current version.
 
         Args:
             current (Optional[str]): The current version identifier of the channel.
-            channel (BaseChannel): The channel being versioned.
 
         Returns:
             str: The next version identifier, which is guaranteed to be monotonically increasing.

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

@@ -1,8 +1,10 @@
+from __future__ import annotations
+
 import asyncio
 import random
 from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import asynccontextmanager
-from typing import Any, Callable, Optional, TypeVar, cast
+from typing import Any, Callable, TypeVar, cast
 
 import aiosqlite
 from langchain_core.runnables import RunnableConfig
@@ -19,7 +21,6 @@ from langgraph.checkpoint.base import (
     get_checkpoint_metadata,
 )
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
-from langgraph.checkpoint.serde.types import ChannelProtocol
 from langgraph.checkpoint.sqlite.utils import search_where
 
 T = TypeVar("T", bound=Callable)
@@ -109,7 +110,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
         self,
         conn: aiosqlite.Connection,
         *,
-        serde: Optional[SerializerProtocol] = None,
+        serde: SerializerProtocol | None = None,
     ):
         super().__init__(serde=serde)
         self.jsonplus_serde = JsonPlusSerializer()
@@ -122,7 +123,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
     @asynccontextmanager
     async def from_conn_string(
         cls, conn_string: str
-    ) -> AsyncIterator["AsyncSqliteSaver"]:
+    ) -> AsyncIterator[AsyncSqliteSaver]:
         """Create a new AsyncSqliteSaver instance from a connection string.
 
         Args:
@@ -134,7 +135,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
         async with aiosqlite.connect(conn_string) as conn:
             yield cls(conn)
 
-    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the database.
 
         This method retrieves a checkpoint tuple from the SQLite database based on the
@@ -166,11 +167,11 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
 
     def list(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> Iterator[CheckpointTuple]:
         """List checkpoints from the database asynchronously.
 
@@ -311,7 +312,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
 
             self.is_setup = True
 
-    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the database asynchronously.
 
         This method retrieves a checkpoint tuple from the SQLite database based on the
@@ -399,11 +400,11 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
 
     async def alist(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> AsyncIterator[CheckpointTuple]:
         """List checkpoints from the database asynchronously.
 
@@ -590,14 +591,13 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
             )
             await self.conn.commit()
 
-    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
+    def get_next_version(self, current: str | None) -> str:
         """Generate the next version ID for a channel.
 
         This method creates a new version identifier for a channel based on its current version.
 
         Args:
             current (Optional[str]): The current version identifier of the channel.
-            channel (BaseChannel): The channel being versioned.
 
         Returns:
             str: The next version identifier, which is guaranteed to be monotonically increasing.

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/utils.py

@@ -1,6 +1,8 @@
+from __future__ import annotations
+
 import json
 from collections.abc import Sequence
-from typing import Any, Optional
+from typing import Any
 
 from langchain_core.runnables import RunnableConfig
 
@@ -52,9 +54,9 @@ def _metadata_predicate(
 
 
 def search_where(
-    config: Optional[RunnableConfig],
-    filter: Optional[dict[str, Any]],
-    before: Optional[RunnableConfig] = None,
+    config: RunnableConfig | None,
+    filter: dict[str, Any] | None,
+    before: RunnableConfig | None = None,
 ) -> tuple[str, Sequence[Any]]:
     """Return WHERE clause predicates for (a)search() given metadata filter
     and `before` config.

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

@@ -1,10 +1,12 @@
+from __future__ import annotations
+
 import asyncio
 import logging
 from collections import defaultdict
 from collections.abc import AsyncIterator, Iterable, Sequence
 from contextlib import asynccontextmanager
 from types import TracebackType
-from typing import Any, Callable, Optional, Union, cast
+from typing import Any, Callable, cast
 
 import aiosqlite
 import orjson
@@ -88,11 +90,10 @@ class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
         self,
         conn: aiosqlite.Connection,
         *,
-        deserializer: Optional[
-            Callable[[Union[bytes, str, orjson.Fragment]], dict[str, Any]]
-        ] = None,
-        index: Optional[SqliteIndexConfig] = None,
-        ttl: Optional[TTLConfig] = None,
+        deserializer: Callable[[bytes | str | orjson.Fragment], dict[str, Any]]
+        | None = None,
+        index: SqliteIndexConfig | None = None,
+        ttl: TTLConfig | None = None,
     ):
         """Initialize the async SQLite store.
 
@@ -114,7 +115,7 @@ class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
         else:
             self.embeddings = None
         self.ttl_config = ttl
-        self._ttl_sweeper_task: Optional[asyncio.Task[None]] = None
+        self._ttl_sweeper_task: asyncio.Task[None] | None = None
         self._ttl_stop_event = asyncio.Event()
 
     @classmethod
@@ -123,9 +124,9 @@ class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
         cls,
         conn_string: str,
         *,
-        index: Optional[SqliteIndexConfig] = None,
-        ttl: Optional[TTLConfig] = None,
-    ) -> AsyncIterator["AsyncSqliteStore"]:
+        index: SqliteIndexConfig | None = None,
+        ttl: TTLConfig | None = None,
+    ) -> AsyncIterator[AsyncSqliteStore]:
         """Create a new AsyncSqliteStore instance from a connection string.
 
         Args:
@@ -253,7 +254,7 @@ class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
             return deleted_count
 
     async def start_ttl_sweeper(
-        self, sweep_interval_minutes: Optional[int] = None
+        self, sweep_interval_minutes: int | None = None
     ) -> asyncio.Task[None]:
         """Periodically delete expired store items based on TTL.
 
@@ -298,7 +299,7 @@ class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
         self._ttl_sweeper_task = task
         return task
 
-    async def stop_ttl_sweeper(self, timeout: Optional[float] = None) -> bool:
+    async def stop_ttl_sweeper(self, timeout: float | None = None) -> bool:
         """Stop the TTL sweeper task if it's running.
 
         Args:
@@ -333,14 +334,14 @@ class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
 
         return success
 
-    async def __aenter__(self) -> "AsyncSqliteStore":
+    async def __aenter__(self) -> AsyncSqliteStore:
         return self
 
     async def __aexit__(
         self,
-        exc_type: Optional[type[BaseException]],
-        exc_val: Optional[BaseException],
-        exc_tb: Optional["TracebackType"],
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
     ) -> None:
         # Ensure the TTL sweeper task is stopped when exiting the context
         if hasattr(self, "_ttl_sweeper_task") and self._ttl_sweeper_task is not None:

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

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import concurrent.futures
 import datetime
 import logging
@@ -6,7 +8,7 @@ import threading
 from collections import defaultdict
 from collections.abc import Iterable, Iterator, Sequence
 from contextlib import contextmanager
-from typing import Any, Callable, Literal, NamedTuple, Optional, Union, cast
+from typing import Any, Callable, Literal, NamedTuple, cast
 
 import orjson
 import sqlite_vec  # type: ignore[import-untyped]
@@ -105,7 +107,7 @@ def _decode_ns_text(namespace: str) -> tuple[str, ...]:
     return tuple(namespace.split("."))
 
 
-def _json_loads(content: Union[bytes, str, orjson.Fragment]) -> Any:
+def _json_loads(content: bytes | str | orjson.Fragment) -> Any:
     if isinstance(content, orjson.Fragment):
         if hasattr(content, "buf"):
             content = content.buf
@@ -125,9 +127,7 @@ def _row_to_item(
     namespace: tuple[str, ...],
     row: dict[str, Any],
     *,
-    loader: Optional[
-        Callable[[Union[bytes, str, orjson.Fragment]], dict[str, Any]]
-    ] = None,
+    loader: Callable[[bytes | str | orjson.Fragment], dict[str, Any]] | None = None,
 ) -> Item:
     """Convert a row from the database into an Item."""
     val = row["value"]
@@ -149,9 +149,7 @@ def _row_to_search_item(
     namespace: tuple[str, ...],
     row: dict[str, Any],
     *,
-    loader: Optional[
-        Callable[[Union[bytes, str, orjson.Fragment]], dict[str, Any]]
-    ] = None,
+    loader: Callable[[bytes | str | orjson.Fragment], dict[str, Any]] | None = None,
 ) -> SearchItem:
     """Convert a row from the database into a SearchItem."""
     loader = loader or _json_loads
@@ -196,8 +194,8 @@ class BaseSqliteStore:
     MIGRATIONS = MIGRATIONS
     VECTOR_MIGRATIONS = VECTOR_MIGRATIONS
     supports_ttl = True
-    index_config: Optional[SqliteIndexConfig] = None
-    ttl_config: Optional[TTLConfig] = None
+    index_config: SqliteIndexConfig | None = None
+    ttl_config: TTLConfig | None = None
 
     def _get_batch_GET_ops_queries(
         self, get_ops: Sequence[tuple[int, GetOp]]
@@ -259,7 +257,7 @@ class BaseSqliteStore:
         self, put_ops: Sequence[tuple[int, PutOp]]
     ) -> tuple[
         list[tuple[str, Sequence]],
-        Optional[tuple[str, Sequence[tuple[str, str, str, str]]]],
+        tuple[str, Sequence[tuple[str, str, str, str]]] | None,
     ]:
         # Last-write wins
         dedupped_ops: dict[tuple[tuple[str, ...], str], PutOp] = {}
@@ -288,9 +286,7 @@ class BaseSqliteStore:
                 params = (_namespace_to_text(namespace), *keys)
                 queries.append((query, params))
 
-        embedding_request: Optional[tuple[str, Sequence[tuple[str, str, str, str]]]] = (
-            None
-        )
+        embedding_request: tuple[str, Sequence[tuple[str, str, str, str]]] | None = None
         if inserts:
             values = []
             insertion_params = []
@@ -358,7 +354,7 @@ class BaseSqliteStore:
     def _prepare_batch_search_queries(
         self, search_ops: Sequence[tuple[int, SearchOp]]
     ) -> tuple[
-        list[tuple[str, list[Union[None, str, list[float]]]]],  # queries, params
+        list[tuple[str, list[None | str | list[float]]]],  # queries, params
         list[tuple[int, str]],  # idx, query_text pairs to embed
     ]:
         """
@@ -785,11 +781,10 @@ class SqliteStore(BaseSqliteStore, BaseStore):
         self,
         conn: sqlite3.Connection,
         *,
-        deserializer: Optional[
-            Callable[[Union[bytes, str, orjson.Fragment]], dict[str, Any]]
-        ] = None,
-        index: Optional[SqliteIndexConfig] = None,
-        ttl: Optional[TTLConfig] = None,
+        deserializer: Callable[[bytes | str | orjson.Fragment], dict[str, Any]]
+        | None = None,
+        index: SqliteIndexConfig | None = None,
+        ttl: TTLConfig | None = None,
     ):
         super().__init__()
         self._deserializer = deserializer
@@ -802,7 +797,7 @@ class SqliteStore(BaseSqliteStore, BaseStore):
         else:
             self.embeddings = None
         self.ttl_config = ttl
-        self._ttl_sweeper_thread: Optional[threading.Thread] = None
+        self._ttl_sweeper_thread: threading.Thread | None = None
         self._ttl_stop_event = threading.Event()
 
     def _get_batch_GET_ops_queries(
@@ -956,9 +951,9 @@ class SqliteStore(BaseSqliteStore, BaseStore):
         cls,
         conn_string: str,
         *,
-        index: Optional[SqliteIndexConfig] = None,
-        ttl: Optional[TTLConfig] = None,
-    ) -> Iterator["SqliteStore"]:
+        index: SqliteIndexConfig | None = None,
+        ttl: TTLConfig | None = None,
+    ) -> Iterator[SqliteStore]:
         """Create a new SqliteStore instance from a connection string.
 
         Args:
@@ -1087,7 +1082,7 @@ class SqliteStore(BaseSqliteStore, BaseStore):
             return deleted_count
 
     def start_ttl_sweeper(
-        self, sweep_interval_minutes: Optional[int] = None
+        self, sweep_interval_minutes: int | None = None
     ) -> concurrent.futures.Future[None]:
         """Periodically delete expired store items based on TTL.
 
@@ -1144,7 +1139,7 @@ class SqliteStore(BaseSqliteStore, BaseStore):
         )
         return future
 
-    def stop_ttl_sweeper(self, timeout: Optional[float] = None) -> bool:
+    def stop_ttl_sweeper(self, timeout: float | None = None) -> bool:
         """Stop the TTL sweeper thread if it's running.
 
         Args:
@@ -1396,7 +1391,7 @@ def _ensure_index_config(
 ) -> tuple[Any, SqliteIndexConfig]:
     """Process and validate index configuration."""
     index_config = index_config.copy()
-    tokenized: list[tuple[str, Union[Literal["$"], list[str]]]] = []
+    tokenized: list[tuple[str, Literal["$"] | list[str]]] = []
     tot = 0
     text_fields = index_config.get("text_fields") or ["$"]
     if isinstance(text_fields, str):

libs/checkpoint-sqlite/pyproject.toml

@@ -54,7 +54,7 @@ lint.select = [
   "B",  # flake8-bugbear
   "I",  # isort
 ]
-lint.ignore = ["E501", "B008", "UP007", "UP006"]
+lint.ignore = ["E501", "B008"]
 
 [tool.pytest-watcher]
 now = true

libs/checkpoint-sqlite/tests/checkpoint_utils.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from datetime import datetime, timezone
+from typing import Any, Protocol
+
+from langgraph.checkpoint.base import Checkpoint, EmptyChannelError
+from langgraph.checkpoint.base.id import uuid6
+
+
+class ChannelProtocol(Protocol):
+    def checkpoint(self) -> Any | None: ...
+
+
+def empty_checkpoint() -> Checkpoint:
+    return Checkpoint(
+        v=1,
+        id=str(uuid6(clock_seq=-2)),
+        ts=datetime.now(timezone.utc).isoformat(),
+        channel_values={},
+        channel_versions={},
+        versions_seen={},
+    )
+
+
+def create_checkpoint(
+    checkpoint: Checkpoint,
+    channels: Mapping[str, ChannelProtocol] | None,
+    step: int,
+    *,
+    id: str | None = None,
+) -> Checkpoint:
+    """Create a checkpoint for the given channels."""
+    ts = datetime.now(timezone.utc).isoformat()
+    if channels is None:
+        values = checkpoint["channel_values"]
+    else:
+        values = {}
+        for k, v in channels.items():
+            if k not in checkpoint["channel_versions"]:
+                continue
+            try:
+                values[k] = v.checkpoint()
+            except EmptyChannelError:
+                pass
+    return Checkpoint(
+        v=1,
+        ts=ts,
+        id=id or str(uuid6(clock_seq=step)),
+        channel_values=values,
+        channel_versions=checkpoint["channel_versions"],
+        versions_seen=checkpoint["versions_seen"],
+    )

libs/checkpoint-sqlite/tests/test_aiosqlite.py

@@ -6,10 +6,9 @@ from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
     Checkpoint,
     CheckpointMetadata,
-    create_checkpoint,
-    empty_checkpoint,
 )
 from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
 
 
 class TestAsyncSqliteSaver:

libs/checkpoint-sqlite/tests/test_sqlite.py

@@ -6,11 +6,10 @@ from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
     Checkpoint,
     CheckpointMetadata,
-    create_checkpoint,
-    empty_checkpoint,
 )
 from langgraph.checkpoint.sqlite import SqliteSaver
 from langgraph.checkpoint.sqlite.utils import _metadata_predicate, search_where
+from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
 
 
 class TestSqliteSaver:

libs/checkpoint/README.md

@@ -51,7 +51,7 @@ read_config = {"configurable": {"thread_id": "1"}}
 
 checkpointer = MemorySaver()
 checkpoint = {
-    "v": 2,
+    "v": 4,
     "ts": "2024-07-31T20:14:19.804150+00:00",
     "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
     "channel_values": {
@@ -73,7 +73,6 @@ checkpoint = {
         "start:node": 2
       }
     },
-    "pending_sends": [],
 }
 
 # store checkpoint

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

@@ -1,22 +1,18 @@
-from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
-from datetime import datetime, timezone
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Iterator, Sequence
 from typing import (  # noqa: UP035
     Any,
-    Dict,
     Generic,
-    List,
     Literal,
     NamedTuple,
-    Optional,
-    Tuple,
     TypedDict,
     TypeVar,
     Union,
 )
 
-from langchain_core.runnables import ConfigurableFieldSpec, RunnableConfig
+from langchain_core.runnables import RunnableConfig
 
-from langgraph.checkpoint.base.id import uuid6
 from langgraph.checkpoint.serde.base import SerializerProtocol, maybe_add_typed_methods
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
 from langgraph.checkpoint.serde.types import (
@@ -24,14 +20,10 @@ from langgraph.checkpoint.serde.types import (
     INTERRUPT,
     RESUME,
     SCHEDULED,
-    ChannelProtocol,
-    SendProtocol,
 )
 
 V = TypeVar("V", int, float, str)
-PendingWrite = Tuple[str, str, Any]
-# Kept for backwards compat, newer versions of LangGraph no longer use this.
-LATEST_VERSION = 2
+PendingWrite = tuple[str, str, Any]
 
 
 # Marked as total=False to allow for future expansion.
@@ -53,11 +45,6 @@ class CheckpointMetadata(TypedDict, total=False):
     0 for the first "loop" checkpoint.
     ... for the nth checkpoint afterwards.
     """
-    writes: dict[str, Any]
-    """The writes that were made between the previous checkpoint and this one.
-
-    Mapping from node name to writes emitted by that node.
-    """
     parents: dict[str, str]
     """The IDs of the parent checkpoints.
 
@@ -65,10 +52,6 @@ class CheckpointMetadata(TypedDict, total=False):
     """
 
 
-class TaskInfo(TypedDict):
-    status: Literal["scheduled", "success", "error"]
-
-
 ChannelVersions = dict[str, Union[str, int, float]]
 
 
@@ -96,22 +79,6 @@ class Checkpoint(TypedDict):
     This keeps track of the versions of the channels that each node has seen.
     Used to determine which nodes to execute next.
     """
-    pending_sends: List[SendProtocol]
-    """List of inputs pushed to nodes but not yet processed.
-    Cleared by the next checkpoint."""
-
-
-# Kept for backwards compat, newer versions of LangGraph no longer use this.
-def empty_checkpoint() -> Checkpoint:
-    return Checkpoint(
-        v=LATEST_VERSION,
-        id=str(uuid6(clock_seq=-2)),
-        ts=datetime.now(timezone.utc).isoformat(),
-        channel_values={},
-        channel_versions={},
-        versions_seen={},
-        pending_sends=[],
-    )
 
 
 def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
@@ -122,39 +89,6 @@ def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
         channel_values=checkpoint["channel_values"].copy(),
         channel_versions=checkpoint["channel_versions"].copy(),
         versions_seen={k: v.copy() for k, v in checkpoint["versions_seen"].items()},
-        pending_sends=checkpoint.get("pending_sends", []).copy(),
-    )
-
-
-# Kept for backwards compat, newer versions of LangGraph no longer use this.
-def create_checkpoint(
-    checkpoint: Checkpoint,
-    channels: Optional[Mapping[str, ChannelProtocol]],
-    step: int,
-    *,
-    id: Optional[str] = None,
-) -> Checkpoint:
-    """Create a checkpoint for the given channels."""
-    ts = datetime.now(timezone.utc).isoformat()
-    if channels is None:
-        values = checkpoint["channel_values"]
-    else:
-        values = {}
-        for k, v in channels.items():
-            if k not in checkpoint["channel_versions"]:
-                continue
-            try:
-                values[k] = v.checkpoint()
-            except EmptyChannelError:
-                pass
-    return Checkpoint(
-        v=LATEST_VERSION,
-        ts=ts,
-        id=id or str(uuid6(clock_seq=step)),
-        channel_values=values,
-        channel_versions=checkpoint["channel_versions"],
-        versions_seen=checkpoint["versions_seen"],
-        pending_sends=checkpoint.get("pending_sends", []),
     )
 
 
@@ -164,36 +98,8 @@ class CheckpointTuple(NamedTuple):
     config: RunnableConfig
     checkpoint: Checkpoint
     metadata: CheckpointMetadata
-    parent_config: Optional[RunnableConfig] = None
-    pending_writes: Optional[List[PendingWrite]] = None
-
-
-CheckpointThreadId = ConfigurableFieldSpec(
-    id="thread_id",
-    annotation=str,
-    name="Thread ID",
-    description=None,
-    default="",
-    is_shared=True,
-)
-
-CheckpointNS = ConfigurableFieldSpec(
-    id="checkpoint_ns",
-    annotation=str,
-    name="Checkpoint NS",
-    description='Checkpoint namespace. Denotes the path to the subgraph node the checkpoint originates from, separated by `|` character, e.g. `"child|grandchild"`. Defaults to "" (root graph).',
-    default="",
-    is_shared=True,
-)
-
-CheckpointId = ConfigurableFieldSpec(
-    id="checkpoint_id",
-    annotation=Optional[str],
-    name="Checkpoint ID",
-    description="Pass to fetch a past checkpoint. If None, fetches the latest checkpoint.",
-    default=None,
-    is_shared=True,
-)
+    parent_config: RunnableConfig | None = None
+    pending_writes: list[PendingWrite] | None = None
 
 
 class BaseCheckpointSaver(Generic[V]):
@@ -215,20 +121,11 @@ class BaseCheckpointSaver(Generic[V]):
     def __init__(
         self,
         *,
-        serde: Optional[SerializerProtocol] = None,
+        serde: SerializerProtocol | None = None,
     ) -> None:
         self.serde = maybe_add_typed_methods(serde or self.serde)
 
-    @property
-    def config_specs(self) -> list[ConfigurableFieldSpec]:
-        """Define the configuration options for the checkpoint saver.
-
-        Returns:
-            list[ConfigurableFieldSpec]: List of configuration field specs.
-        """
-        return [CheckpointThreadId, CheckpointNS, CheckpointId]
-
-    def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
+    def get(self, config: RunnableConfig) -> Checkpoint | None:
         """Fetch a checkpoint using the given configuration.
 
         Args:
@@ -240,7 +137,7 @@ class BaseCheckpointSaver(Generic[V]):
         if value := self.get_tuple(config):
             return value.checkpoint
 
-    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Fetch a checkpoint tuple using the given configuration.
 
         Args:
@@ -256,11 +153,11 @@ class BaseCheckpointSaver(Generic[V]):
 
     def list(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[Dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> Iterator[CheckpointTuple]:
         """List checkpoints that match the given criteria.
 
@@ -304,7 +201,7 @@ class BaseCheckpointSaver(Generic[V]):
     def put_writes(
         self,
         config: RunnableConfig,
-        writes: Sequence[Tuple[str, Any]],
+        writes: Sequence[tuple[str, Any]],
         task_id: str,
         task_path: str = "",
     ) -> None:
@@ -332,7 +229,7 @@ class BaseCheckpointSaver(Generic[V]):
         """
         raise NotImplementedError
 
-    async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
+    async def aget(self, config: RunnableConfig) -> Checkpoint | None:
         """Asynchronously fetch a checkpoint using the given configuration.
 
         Args:
@@ -344,7 +241,7 @@ class BaseCheckpointSaver(Generic[V]):
         if value := await self.aget_tuple(config):
             return value.checkpoint
 
-    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Asynchronously fetch a checkpoint tuple using the given configuration.
 
         Args:
@@ -360,11 +257,11 @@ class BaseCheckpointSaver(Generic[V]):
 
     async def alist(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[Dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> AsyncIterator[CheckpointTuple]:
         """Asynchronously list checkpoints that match the given criteria.
 
@@ -409,7 +306,7 @@ class BaseCheckpointSaver(Generic[V]):
     async def aput_writes(
         self,
         config: RunnableConfig,
-        writes: Sequence[Tuple[str, Any]],
+        writes: Sequence[tuple[str, Any]],
         task_id: str,
         task_path: str = "",
     ) -> None:
@@ -437,7 +334,7 @@ class BaseCheckpointSaver(Generic[V]):
         """
         raise NotImplementedError
 
-    def get_next_version(self, current: Optional[V], channel: ChannelProtocol) -> V:
+    def get_next_version(self, current: V | None) -> V:
         """Generate the next version ID for a channel.
 
         Default is to use integer versions, incrementing by 1. If you override, you can use str/int/float versions,
@@ -445,7 +342,6 @@ class BaseCheckpointSaver(Generic[V]):
 
         Args:
             current: The current version identifier (int, float, or str).
-            channel: The channel being versioned.
 
         Returns:
             V: The next version identifier, which must be increasing.
@@ -465,7 +361,7 @@ class EmptyChannelError(Exception):
     pass
 
 
-def get_checkpoint_id(config: RunnableConfig) -> Optional[str]:
+def get_checkpoint_id(config: RunnableConfig) -> str | None:
     """Get checkpoint ID in a backwards-compatible manner (fallback on thread_ts)."""
     return config["configurable"].get(
         "checkpoint_id", config["configurable"].get("thread_ts")
@@ -476,7 +372,10 @@ def get_checkpoint_metadata(
     config: RunnableConfig, metadata: CheckpointMetadata
 ) -> CheckpointMetadata:
     """Get checkpoint metadata in a backwards-compatible manner."""
-    metadata = metadata.copy()
+    metadata = {
+        k: v.replace("\u0000", "") if isinstance(v, str) else v
+        for k, v in metadata.items()
+    }
     for obj in (config.get("metadata"), config.get("configurable")):
         if not obj:
             continue
@@ -484,8 +383,10 @@ def get_checkpoint_metadata(
             if key in metadata or key in EXCLUDED_METADATA_KEYS or key.startswith("__"):
                 continue
             v = obj[key]
-            if isinstance(v, (str, int, bool, float)):
-                metadata[key] = v  # type: ignore[literal-required]
+            if isinstance(v, str):
+                metadata[key] = v.replace("\u0000", "")
+            elif isinstance(v, (int, bool, float)):
+                metadata[key] = v
     return metadata
 
 

libs/checkpoint/langgraph/checkpoint/base/id.py

@@ -3,10 +3,11 @@ https://github.com/oittaa/uuid6-python/blob/main/src/uuid6/__init__.py#L95
 Bundled in to avoid install issues with uuid6 package
 """
 
+from __future__ import annotations
+
 import random
 import time
 import uuid
-from typing import Optional
 
 _last_v6_timestamp = None
 
@@ -18,12 +19,12 @@ class UUID(uuid.UUID):
 
     def __init__(
         self,
-        hex: Optional[str] = None,
-        bytes: Optional[bytes] = None,
-        bytes_le: Optional[bytes] = None,
-        fields: Optional[tuple[int, int, int, int, int, int]] = None,
-        int: Optional[int] = None,
-        version: Optional[int] = None,
+        hex: str | None = None,
+        bytes: bytes | None = None,
+        bytes_le: bytes | None = None,
+        fields: tuple[int, int, int, int, int, int] | None = None,
+        int: int | None = None,
+        version: int | None = None,
         *,
         is_safe: uuid.SafeUUID = uuid.SafeUUID.unknown,
     ) -> None:
@@ -75,7 +76,7 @@ def _subsec_decode(value: int) -> int:
     return -(-value * 10**6 // 2**20)
 
 
-def uuid6(node: Optional[int] = None, clock_seq: Optional[int] = None) -> UUID:
+def uuid6(node: int | None = None, clock_seq: int | None = None) -> UUID:
     r"""UUID version 6 is a field-compatible version of UUIDv1, reordered for
     improved DB locality. It is expected that UUIDv6 will primarily be
     used in contexts where there are existing v1 UUIDs. Systems that do

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

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 import os
 import pickle
@@ -7,7 +9,7 @@ from collections import defaultdict
 from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import AbstractAsyncContextManager, AbstractContextManager, ExitStack
 from types import TracebackType
-from typing import Any, Optional, Union
+from typing import Any
 
 from langchain_core.runnables import RunnableConfig
 
@@ -22,7 +24,6 @@ from langgraph.checkpoint.base import (
     get_checkpoint_id,
     get_checkpoint_metadata,
 )
-from langgraph.checkpoint.serde.types import TASKS, ChannelProtocol
 
 logger = logging.getLogger(__name__)
 
@@ -64,9 +65,7 @@ class InMemorySaver(
     # thread ID ->  checkpoint NS -> checkpoint ID -> checkpoint mapping
     storage: defaultdict[
         str,
-        dict[
-            str, dict[str, tuple[tuple[str, bytes], tuple[str, bytes], Optional[str]]]
-        ],
+        dict[str, dict[str, tuple[tuple[str, bytes], tuple[str, bytes], str | None]]],
     ]
     # (thread ID, checkpoint NS, checkpoint ID) -> (task ID, write idx)
     writes: defaultdict[
@@ -75,7 +74,7 @@ class InMemorySaver(
     ]
     blobs: dict[
         tuple[
-            str, str, str, Union[str, int, float]
+            str, str, str, str | int | float
         ],  # thread id, checkpoint ns, channel, version
         tuple[str, bytes],
     ]
@@ -83,7 +82,7 @@ class InMemorySaver(
     def __init__(
         self,
         *,
-        serde: Optional[SerializerProtocol] = None,
+        serde: SerializerProtocol | None = None,
         factory: type[defaultdict] = defaultdict,
     ) -> None:
         super().__init__(serde=serde)
@@ -96,26 +95,26 @@ class InMemorySaver(
             self.stack.enter_context(self.writes)  # type: ignore[arg-type]
             self.stack.enter_context(self.blobs)  # type: ignore[arg-type]
 
-    def __enter__(self) -> "InMemorySaver":
+    def __enter__(self) -> InMemorySaver:
         return self.stack.__enter__()
 
     def __exit__(
         self,
-        exc_type: Optional[type[BaseException]],
-        exc_value: Optional[BaseException],
-        traceback: Optional[TracebackType],
-    ) -> Optional[bool]:
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> bool | None:
         return self.stack.__exit__(exc_type, exc_value, traceback)
 
-    async def __aenter__(self) -> "InMemorySaver":
+    async def __aenter__(self) -> InMemorySaver:
         return self.stack.__enter__()
 
     async def __aexit__(
         self,
-        __exc_type: Optional[type[BaseException]],
-        __exc_value: Optional[BaseException],
-        __traceback: Optional[TracebackType],
-    ) -> Optional[bool]:
+        __exc_type: type[BaseException] | None,
+        __exc_value: BaseException | None,
+        __traceback: TracebackType | None,
+    ) -> bool | None:
         return self.stack.__exit__(__exc_type, __exc_value, __traceback)
 
     def _load_blobs(
@@ -130,7 +129,7 @@ class InMemorySaver(
                     channel_values[k] = self.serde.loads_typed(vv)
         return channel_values
 
-    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the in-memory storage.
 
         This method retrieves a checkpoint tuple from the in-memory storage based on the
@@ -150,19 +149,6 @@ class InMemorySaver(
             if saved := self.storage[thread_id][checkpoint_ns].get(checkpoint_id):
                 checkpoint, metadata, parent_checkpoint_id = saved
                 writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
-                if parent_checkpoint_id:
-                    sends = sorted(
-                        (
-                            (*w, k[1])
-                            for k, w in self.writes[
-                                (thread_id, checkpoint_ns, parent_checkpoint_id)
-                            ].items()
-                            if w[1] == TASKS
-                        ),
-                        key=lambda w: (w[3], w[0], w[4]),
-                    )
-                else:
-                    sends = []
                 checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)
                 return CheckpointTuple(
                     config=config,
@@ -171,7 +157,6 @@ class InMemorySaver(
                         "channel_values": self._load_blobs(
                             thread_id, checkpoint_ns, checkpoint_["channel_versions"]
                         ),
-                        "pending_sends": [self.serde.loads_typed(s[2]) for s in sends],
                     },
                     metadata=self.serde.loads_typed(metadata),
                     pending_writes=[
@@ -194,22 +179,7 @@ class InMemorySaver(
                 checkpoint_id = max(checkpoints.keys())
                 checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id]
                 writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
-                if parent_checkpoint_id:
-                    sends = sorted(
-                        (
-                            (*w, k[1])
-                            for k, w in self.writes[
-                                (thread_id, checkpoint_ns, parent_checkpoint_id)
-                            ].items()
-                            if w[1] == TASKS
-                        ),
-                        key=lambda w: (w[3], w[0], w[4]),
-                    )
-                else:
-                    sends = []
-
                 checkpoint_ = self.serde.loads_typed(checkpoint)
-
                 return CheckpointTuple(
                     config={
                         "configurable": {
@@ -223,7 +193,6 @@ class InMemorySaver(
                         "channel_values": self._load_blobs(
                             thread_id, checkpoint_ns, checkpoint_["channel_versions"]
                         ),
-                        "pending_sends": [self.serde.loads_typed(s[2]) for s in sends],
                     },
                     metadata=self.serde.loads_typed(metadata),
                     pending_writes=[
@@ -244,11 +213,11 @@ class InMemorySaver(
 
     def list(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> Iterator[CheckpointTuple]:
         """List checkpoints from the in-memory storage.
 
@@ -316,20 +285,6 @@ class InMemorySaver(
                         (thread_id, checkpoint_ns, checkpoint_id)
                     ].values()
 
-                    if parent_checkpoint_id:
-                        sends = sorted(
-                            (
-                                (*w, k[1])
-                                for k, w in self.writes[
-                                    (thread_id, checkpoint_ns, parent_checkpoint_id)
-                                ].items()
-                                if w[1] == TASKS
-                            ),
-                            key=lambda w: (w[3], w[0], w[4]),
-                        )
-                    else:
-                        sends = []
-
                     checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)
 
                     yield CheckpointTuple(
@@ -347,9 +302,6 @@ class InMemorySaver(
                                 checkpoint_ns,
                                 checkpoint_["channel_versions"],
                             ),
-                            "pending_sends": [
-                                self.serde.loads_typed(s[2]) for s in sends
-                            ],
                         },
                         metadata=metadata,
                         parent_config=(
@@ -390,7 +342,6 @@ class InMemorySaver(
             RunnableConfig: The updated config containing the saved checkpoint's timestamp.
         """
         c = checkpoint.copy()
-        c.pop("pending_sends")  # type: ignore[misc]
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"]["checkpoint_ns"]
         values: dict[str, Any] = c.pop("channel_values")  # type: ignore[misc]
@@ -471,7 +422,7 @@ class InMemorySaver(
             if k[0] == thread_id:
                 del self.blobs[k]
 
-    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Asynchronous version of get_tuple.
 
         This method is an asynchronous wrapper around get_tuple that runs the synchronous
@@ -487,11 +438,11 @@ class InMemorySaver(
 
     async def alist(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[dict[str, Any]] = None,
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> AsyncIterator[CheckpointTuple]:
         """Asynchronous version of list.
 
@@ -561,7 +512,7 @@ class InMemorySaver(
         """
         return self.delete_thread(thread_id)
 
-    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
+    def get_next_version(self, current: str | None) -> str:
         if current is None:
             current_v = 0
         elif isinstance(current, int):
@@ -620,7 +571,7 @@ class PersistentDict(defaultdict):
         self.sync()
         self.clear()
 
-    def __enter__(self) -> "PersistentDict":
+    def __enter__(self) -> PersistentDict:
         return self
 
     def __exit__(self, *exc_info: Any) -> None:

libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import dataclasses
 import decimal
 import importlib
@@ -18,7 +20,7 @@ from ipaddress import (
     IPv6Interface,
     IPv6Network,
 )
-from typing import Any, Callable, Optional, Union, cast
+from typing import Any, Callable, cast
 from uuid import UUID
 from zoneinfo import ZoneInfo
 
@@ -41,7 +43,7 @@ class JsonPlusSerializer(SerializerProtocol):
         self,
         *,
         pickle_fallback: bool = False,
-        __unpack_ext_hook__: Optional[Callable[[int, bytes], Any]] = None,
+        __unpack_ext_hook__: Callable[[int, bytes], Any] | None = None,
     ) -> None:
         self.pickle_fallback = pickle_fallback
         self._unpack_ext_hook = (
@@ -52,11 +54,11 @@ class JsonPlusSerializer(SerializerProtocol):
 
     def _encode_constructor_args(
         self,
-        constructor: Union[Callable, type[Any]],
+        constructor: Callable | type[Any],
         *,
-        method: Union[None, str, Sequence[Union[None, str]]] = None,
-        args: Optional[Sequence[Any]] = None,
-        kwargs: Optional[dict[str, Any]] = None,
+        method: None | str | Sequence[None | str] = None,
+        args: Sequence[Any] | None = None,
+        kwargs: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
         out = {
             "lc": 2,
@@ -71,7 +73,7 @@ class JsonPlusSerializer(SerializerProtocol):
             out["kwargs"] = kwargs
         return out
 
-    def _default(self, obj: Any) -> Union[str, dict[str, Any]]:
+    def _default(self, obj: Any) -> str | dict[str, Any]:
         if isinstance(obj, Serializable):
             return cast(dict[str, Any], obj.to_json())
         elif hasattr(obj, "model_dump") and callable(obj.model_dump):
@@ -251,7 +253,7 @@ EXT_PYDANTIC_V1 = 4
 EXT_PYDANTIC_V2 = 5
 
 
-def _msgpack_default(obj: Any) -> Union[str, ormsgpack.Ext]:
+def _msgpack_default(obj: Any) -> str | ormsgpack.Ext:
     if hasattr(obj, "model_dump") and callable(obj.model_dump):  # pydantic v2
         return ormsgpack.Ext(
             EXT_PYDANTIC_V2,

libs/checkpoint/langgraph/checkpoint/serde/types.py

@@ -1,13 +1,4 @@
-from collections.abc import Sequence
-from typing import (
-    Any,
-    Optional,
-    Protocol,
-    TypeVar,
-    runtime_checkable,
-)
-
-from typing_extensions import Self
+from typing import Any, Protocol, TypeVar, runtime_checkable
 
 ERROR = "__error__"
 SCHEDULED = "__scheduled__"
@@ -20,25 +11,6 @@ Update = TypeVar("Update", contravariant=True)
 C = TypeVar("C")
 
 
-class ChannelProtocol(Protocol[Value, Update, C]):
-    # Mirrors langgraph.channels.base.BaseChannel
-    @property
-    def ValueType(self) -> Any: ...
-
-    @property
-    def UpdateType(self) -> Any: ...
-
-    def checkpoint(self) -> Optional[C]: ...
-
-    def from_checkpoint(self, checkpoint: Optional[C]) -> Self: ...
-
-    def update(self, values: Sequence[Update]) -> bool: ...
-
-    def get(self) -> Value: ...
-
-    def consume(self) -> bool: ...
-
-
 @runtime_checkable
 class SendProtocol(Protocol):
     # Mirrors langgraph.constants.Send

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

@@ -9,6 +9,8 @@ Core types:
     - Op: Get/Put/Search/List operations
 """
 
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
 from collections.abc import Iterable
 from datetime import datetime
@@ -16,7 +18,6 @@ from typing import (
     Any,
     Literal,
     NamedTuple,
-    Optional,
     TypedDict,
     Union,
     cast,
@@ -127,7 +128,7 @@ class SearchItem(Item):
         value: dict[str, Any],
         created_at: datetime,
         updated_at: datetime,
-        score: Optional[float] = None,
+        score: float | None = None,
     ) -> None:
         """Initialize a result item.
 
@@ -242,7 +243,7 @@ class SearchOp(NamedTuple):
         ```
     """
 
-    filter: Optional[dict[str, Any]] = None
+    filter: dict[str, Any] | None = None
     """Key-value pairs for filtering results based on exact matches or comparison operators.
 
     The filter supports both exact matches and operator-based comparisons.
@@ -284,7 +285,7 @@ class SearchOp(NamedTuple):
     offset: int = 0
     """Number of matching items to skip for pagination."""
 
-    query: Optional[str] = None
+    query: str | None = None
     """Natural language search query for semantic search capabilities.
 
     ???+ example "Examples"
@@ -379,7 +380,7 @@ class ListNamespacesOp(NamedTuple):
 
     """
 
-    match_conditions: Optional[tuple[MatchCondition, ...]] = None
+    match_conditions: tuple[MatchCondition, ...] | None = None
     """Optional conditions for filtering namespaces.
 
     ???+ example "Examples"
@@ -397,7 +398,7 @@ class ListNamespacesOp(NamedTuple):
         ```
     """
 
-    max_depth: Optional[int] = None
+    max_depth: int | None = None
     """Maximum depth of namespace hierarchy to return.
 
     Note:
@@ -452,7 +453,7 @@ class PutOp(NamedTuple):
         the full path would effectively be "documents/user123/report1"
     """
 
-    value: Optional[dict[str, Any]]
+    value: dict[str, Any] | None
     """The data to store, or None to mark the item for deletion.
 
     The value must be a dictionary with string keys and JSON-serializable values.
@@ -466,7 +467,7 @@ class PutOp(NamedTuple):
         }
     """
 
-    index: Optional[Union[Literal[False], list[str]]] = None  # type: ignore[assignment]
+    index: Literal[False] | list[str] | None = None  # type: ignore[assignment]
     """Controls how the item's fields are indexed for search operations.
 
     Indexing configuration determines how the item can be found through search:
@@ -501,7 +502,7 @@ class PutOp(NamedTuple):
         ]
         ```
     """
-    ttl: Optional[float] = None
+    ttl: float | None = None
     """Controls the TTL (time-to-live) for the item in minutes.
 
     If provided, and if the store you are using supports this feature, the item
@@ -530,14 +531,14 @@ class TTLConfig(TypedDict, total=False):
     This can be overridden per-operation by explicitly setting refresh_ttl.
     Defaults to True if not configured.
     """
-    default_ttl: Optional[float]
+    default_ttl: float | None
     """Default TTL (time-to-live) in minutes for new items.
     
     If provided, new items will expire after this many minutes after their last access.
     The expiration timer refreshes on both read and write operations.
     Defaults to None (no expiration).
     """
-    sweep_interval_minutes: Optional[int]
+    sweep_interval_minutes: int | None
     """Interval in minutes between TTL sweep operations.
     
     If provided, the store will periodically delete expired items based on TTL.
@@ -565,7 +566,7 @@ class IndexConfig(TypedDict, total=False):
         - cohere:embed-multilingual-light-v3.0: 384
     """
 
-    embed: Union[Embeddings, EmbeddingsFunc, AEmbeddingsFunc, str]
+    embed: Embeddings | EmbeddingsFunc | AEmbeddingsFunc | str
     """Optional function to generate embeddings from text.
     
     Can be specified in three ways:
@@ -633,7 +634,7 @@ class IndexConfig(TypedDict, total=False):
         ```
     """
 
-    fields: Optional[list[str]]
+    fields: list[str] | None
     """Fields to extract text from for embedding generation.
     
     Controls which parts of stored items are embedded for semantic search. Follows JSON path syntax:
@@ -690,7 +691,7 @@ class BaseStore(ABC):
     """
 
     supports_ttl: bool = False
-    ttl_config: Optional[TTLConfig] = None
+    ttl_config: TTLConfig | None = None
 
     __slots__ = ("__weakref__",)
 
@@ -723,8 +724,8 @@ class BaseStore(ABC):
         namespace: tuple[str, ...],
         key: str,
         *,
-        refresh_ttl: Optional[bool] = None,
-    ) -> Optional[Item]:
+        refresh_ttl: bool | None = None,
+    ) -> Item | None:
         """Retrieve a single item.
 
         Args:
@@ -746,11 +747,11 @@ class BaseStore(ABC):
         namespace_prefix: tuple[str, ...],
         /,
         *,
-        query: Optional[str] = None,
-        filter: Optional[dict[str, Any]] = None,
+        query: str | None = None,
+        filter: dict[str, Any] | None = None,
         limit: int = 10,
         offset: int = 0,
-        refresh_ttl: Optional[bool] = None,
+        refresh_ttl: bool | None = None,
     ) -> list[SearchItem]:
         """Search for items within a namespace prefix.
 
@@ -817,9 +818,9 @@ class BaseStore(ABC):
         namespace: tuple[str, ...],
         key: str,
         value: dict[str, Any],
-        index: Optional[Union[Literal[False], list[str]]] = None,
+        index: Literal[False] | list[str] | None = None,
         *,
-        ttl: Union[Optional[float], "NotProvided"] = NOT_PROVIDED,
+        ttl: float | None | NotProvided = NOT_PROVIDED,
     ) -> None:
         """Store or update an item in the store.
 
@@ -901,9 +902,9 @@ class BaseStore(ABC):
     def list_namespaces(
         self,
         *,
-        prefix: Optional[NamespacePath] = None,
-        suffix: Optional[NamespacePath] = None,
-        max_depth: Optional[int] = None,
+        prefix: NamespacePath | None = None,
+        suffix: NamespacePath | None = None,
+        max_depth: int | None = None,
         limit: int = 100,
         offset: int = 0,
     ) -> list[tuple[str, ...]]:
@@ -956,8 +957,8 @@ class BaseStore(ABC):
         namespace: tuple[str, ...],
         key: str,
         *,
-        refresh_ttl: Optional[bool] = None,
-    ) -> Optional[Item]:
+        refresh_ttl: bool | None = None,
+    ) -> Item | None:
         """Asynchronously retrieve a single item.
 
         Args:
@@ -984,11 +985,11 @@ class BaseStore(ABC):
         namespace_prefix: tuple[str, ...],
         /,
         *,
-        query: Optional[str] = None,
-        filter: Optional[dict[str, Any]] = None,
+        query: str | None = None,
+        filter: dict[str, Any] | None = None,
         limit: int = 10,
         offset: int = 0,
-        refresh_ttl: Optional[bool] = None,
+        refresh_ttl: bool | None = None,
     ) -> list[SearchItem]:
         """Asynchronously search for items within a namespace prefix.
 
@@ -1058,9 +1059,9 @@ class BaseStore(ABC):
         namespace: tuple[str, ...],
         key: str,
         value: dict[str, Any],
-        index: Optional[Union[Literal[False], list[str]]] = None,
+        index: Literal[False] | list[str] | None = None,
         *,
-        ttl: Union[Optional[float], "NotProvided"] = NOT_PROVIDED,
+        ttl: float | None | NotProvided = NOT_PROVIDED,
     ) -> None:
         """Asynchronously store or update an item in the store.
 
@@ -1150,9 +1151,9 @@ class BaseStore(ABC):
     async def alist_namespaces(
         self,
         *,
-        prefix: Optional[NamespacePath] = None,
-        suffix: Optional[NamespacePath] = None,
-        max_depth: Optional[int] = None,
+        prefix: NamespacePath | None = None,
+        suffix: NamespacePath | None = None,
+        max_depth: int | None = None,
         limit: int = 100,
         offset: int = 0,
     ) -> list[tuple[str, ...]]:
@@ -1226,7 +1227,7 @@ def _validate_namespace(namespace: tuple[str, ...]) -> None:
 
 
 def _ensure_refresh(
-    ttl_config: Optional[TTLConfig], refresh_ttl: Optional[bool] = None
+    ttl_config: TTLConfig | None, refresh_ttl: bool | None = None
 ) -> bool:
     if refresh_ttl is not None:
         return refresh_ttl
@@ -1236,9 +1237,9 @@ def _ensure_refresh(
 
 
 def _ensure_ttl(
-    ttl_config: Optional[TTLConfig],
-    ttl: Union[Optional[float], "NotProvided"] = NOT_PROVIDED,
-) -> Optional[float]:
+    ttl_config: TTLConfig | None,
+    ttl: float | None | NotProvided = NOT_PROVIDED,
+) -> float | None:
     if ttl is NOT_PROVIDED:
         if ttl_config:
             return ttl_config.get("default_ttl")

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

@@ -1,10 +1,12 @@
 """Utilities for batching operations in a background task."""
 
+from __future__ import annotations
+
 import asyncio
 import functools
 import weakref
 from collections.abc import Iterable
-from typing import Any, Callable, Literal, Optional, TypeVar, Union
+from typing import Any, Callable, Literal, TypeVar
 
 from langgraph.store.base import (
     NOT_PROVIDED,
@@ -30,7 +32,7 @@ F = TypeVar("F", bound=Callable)
 
 def _check_loop(func: F) -> F:
     @functools.wraps(func)
-    def wrapper(store: "AsyncBatchedBaseStore", *args: Any, **kwargs: Any) -> Any:
+    def wrapper(store: AsyncBatchedBaseStore, *args: Any, **kwargs: Any) -> Any:
         method_name: str = func.__name__
         try:
             current_loop = asyncio.get_running_loop()
@@ -75,8 +77,8 @@ class AsyncBatchedBaseStore(BaseStore):
         namespace: tuple[str, ...],
         key: str,
         *,
-        refresh_ttl: Optional[bool] = None,
-    ) -> Optional[Item]:
+        refresh_ttl: bool | None = None,
+    ) -> Item | None:
         assert not self._task.done()
         fut = self._loop.create_future()
         self._aqueue.put_nowait(
@@ -96,11 +98,11 @@ class AsyncBatchedBaseStore(BaseStore):
         namespace_prefix: tuple[str, ...],
         /,
         *,
-        query: Optional[str] = None,
-        filter: Optional[dict[str, Any]] = None,
+        query: str | None = None,
+        filter: dict[str, Any] | None = None,
         limit: int = 10,
         offset: int = 0,
-        refresh_ttl: Optional[bool] = None,
+        refresh_ttl: bool | None = None,
     ) -> list[SearchItem]:
         assert not self._task.done()
         fut = self._loop.create_future()
@@ -124,9 +126,9 @@ class AsyncBatchedBaseStore(BaseStore):
         namespace: tuple[str, ...],
         key: str,
         value: dict[str, Any],
-        index: Optional[Union[Literal[False], list[str]]] = None,
+        index: Literal[False] | list[str] | None = None,
         *,
-        ttl: Union[Optional[float], "NotProvided"] = NOT_PROVIDED,
+        ttl: float | None | NotProvided = NOT_PROVIDED,
     ) -> None:
         assert not self._task.done()
         _validate_namespace(namespace)
@@ -154,9 +156,9 @@ class AsyncBatchedBaseStore(BaseStore):
     async def alist_namespaces(
         self,
         *,
-        prefix: Optional[NamespacePath] = None,
-        suffix: Optional[NamespacePath] = None,
-        max_depth: Optional[int] = None,
+        prefix: NamespacePath | None = None,
+        suffix: NamespacePath | None = None,
+        max_depth: int | None = None,
         limit: int = 100,
         offset: int = 0,
     ) -> list[tuple[str, ...]]:
@@ -187,8 +189,8 @@ class AsyncBatchedBaseStore(BaseStore):
         namespace: tuple[str, ...],
         key: str,
         *,
-        refresh_ttl: Optional[bool] = None,
-    ) -> Optional[Item]:
+        refresh_ttl: bool | None = None,
+    ) -> Item | None:
         return asyncio.run_coroutine_threadsafe(
             self.aget(namespace, key=key, refresh_ttl=refresh_ttl), self._loop
         ).result()
@@ -199,11 +201,11 @@ class AsyncBatchedBaseStore(BaseStore):
         namespace_prefix: tuple[str, ...],
         /,
         *,
-        query: Optional[str] = None,
-        filter: Optional[dict[str, Any]] = None,
+        query: str | None = None,
+        filter: dict[str, Any] | None = None,
         limit: int = 10,
         offset: int = 0,
-        refresh_ttl: Optional[bool] = None,
+        refresh_ttl: bool | None = None,
     ) -> list[SearchItem]:
         return asyncio.run_coroutine_threadsafe(
             self.asearch(
@@ -223,9 +225,9 @@ class AsyncBatchedBaseStore(BaseStore):
         namespace: tuple[str, ...],
         key: str,
         value: dict[str, Any],
-        index: Optional[Union[Literal[False], list[str]]] = None,
+        index: Literal[False] | list[str] | None = None,
         *,
-        ttl: Union[Optional[float], "NotProvided"] = NOT_PROVIDED,
+        ttl: float | None | NotProvided = NOT_PROVIDED,
     ) -> None:
         _validate_namespace(namespace)
         asyncio.run_coroutine_threadsafe(
@@ -253,9 +255,9 @@ class AsyncBatchedBaseStore(BaseStore):
     def list_namespaces(
         self,
         *,
-        prefix: Optional[NamespacePath] = None,
-        suffix: Optional[NamespacePath] = None,
-        max_depth: Optional[int] = None,
+        prefix: NamespacePath | None = None,
+        suffix: NamespacePath | None = None,
+        max_depth: int | None = None,
         limit: int = 100,
         offset: int = 0,
     ) -> list[tuple[str, ...]]:
@@ -271,7 +273,7 @@ class AsyncBatchedBaseStore(BaseStore):
         ).result()
 
 
-def _dedupe_ops(values: list[Op]) -> tuple[Optional[list[int]], list[Op]]:
+def _dedupe_ops(values: list[Op]) -> tuple[list[int] | None, list[Op]]:
     """Dedupe operations while preserving order for results.
 
     Args:

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

@@ -6,11 +6,13 @@ with LangChain-compatible tools while maintaining support for both synchronous a
 asynchronous operations.
 """
 
+from __future__ import annotations
+
 import asyncio
 import functools
 import json
 from collections.abc import Awaitable, Sequence
-from typing import Any, Callable, Optional, Union
+from typing import Any, Callable
 
 from langchain_core.embeddings import Embeddings
 
@@ -30,7 +32,7 @@ Similar to EmbeddingsFunc, but returns an awaitable that resolves to the embeddi
 
 
 def ensure_embeddings(
-    embed: Union[Embeddings, EmbeddingsFunc, AEmbeddingsFunc, str, None],
+    embed: Embeddings | EmbeddingsFunc | AEmbeddingsFunc | str | None,
 ) -> Embeddings:
     """Ensure that an embedding function conforms to LangChain's Embeddings interface.
 
@@ -141,7 +143,7 @@ class EmbeddingsLambda(Embeddings):
 
     def __init__(
         self,
-        func: Union[EmbeddingsFunc, AEmbeddingsFunc],
+        func: EmbeddingsFunc | AEmbeddingsFunc,
     ) -> None:
         if func is None:
             raise ValueError("func must be provided")
@@ -221,7 +223,7 @@ class EmbeddingsLambda(Embeddings):
         return (await afunc([text]))[0]
 
 
-def get_text_at_path(obj: Any, path: Union[str, list[str]]) -> list[str]:
+def get_text_at_path(obj: Any, path: str | list[str]) -> list[str]:
     """Extract text from an object using a path expression or pre-tokenized path.
 
     Args:
@@ -279,7 +281,7 @@ def get_text_at_path(obj: Any, path: Union[str, list[str]]) -> list[str]:
             for field in fields:
                 nested_tokens = tokenize_path(field)
                 if nested_tokens:
-                    current_obj: Optional[dict] = obj
+                    current_obj: dict | None = obj
                     for nested_token in nested_tokens:
                         if (
                             isinstance(current_obj, dict)
@@ -404,7 +406,7 @@ def _is_async_callable(
 
 
 @functools.lru_cache
-def _get_init_embeddings() -> Optional[Callable[[str], Embeddings]]:
+def _get_init_embeddings() -> Callable[[str], Embeddings] | None:
     try:
         from langchain.embeddings import init_embeddings  # type: ignore
 

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

@@ -99,6 +99,8 @@ Tip:
     ```
 """
 
+from __future__ import annotations
+
 import asyncio
 import concurrent.futures as cf
 import functools
@@ -107,7 +109,7 @@ from collections import defaultdict
 from collections.abc import Iterable
 from datetime import datetime, timezone
 from importlib import util
-from typing import Any, Optional
+from typing import Any
 
 from langchain_core.embeddings import Embeddings
 
@@ -178,7 +180,7 @@ class InMemoryStore(BaseStore):
         "embeddings",
     )
 
-    def __init__(self, *, index: Optional[IndexConfig] = None) -> None:
+    def __init__(self, *, index: IndexConfig | None = None) -> None:
         # Both _data and _vectors are wrapped in the In-memory API
         # Do not change their names
         self._data: dict[tuple[str, ...], dict[str, Item]] = defaultdict(dict)
@@ -189,7 +191,7 @@ class InMemoryStore(BaseStore):
         self.index_config = index
         if self.index_config:
             self.index_config = self.index_config.copy()
-            self.embeddings: Optional[Embeddings] = ensure_embeddings(
+            self.embeddings: Embeddings | None = ensure_embeddings(
                 self.index_config.get("embed"),
             )
             self.index_config["__tokenized_fields"] = [
@@ -325,7 +327,7 @@ class InMemoryStore(BaseStore):
                 )
                 # max pooling
                 seen: set[tuple[tuple[str, ...], str]] = set()
-                kept: list[tuple[Optional[float], Item]] = []
+                kept: list[tuple[float | None, Item]] = []
                 for score, item in sorted_results:
                     key = (item.namespace, item.key)
                     if key in seen:

libs/checkpoint/pyproject.toml

@@ -46,7 +46,7 @@ lint.select = [
   "B",  # flake8-bugbear
   "I",  # isort
 ]
-lint.ignore = ["E501", "B008", "UP007", "UP006"]
+lint.ignore = ["E501", "B008"]
 
 [tool.pytest-watcher]
 now = true

libs/checkpoint/tests/checkpoint_utils.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from datetime import datetime, timezone
+from typing import Any, Protocol
+
+from langgraph.checkpoint.base import Checkpoint, EmptyChannelError
+from langgraph.checkpoint.base.id import uuid6
+
+
+class ChannelProtocol(Protocol):
+    def checkpoint(self) -> Any | None: ...
+
+
+def empty_checkpoint() -> Checkpoint:
+    return Checkpoint(
+        v=1,
+        id=str(uuid6(clock_seq=-2)),
+        ts=datetime.now(timezone.utc).isoformat(),
+        channel_values={},
+        channel_versions={},
+        versions_seen={},
+    )
+
+
+def create_checkpoint(
+    checkpoint: Checkpoint,
+    channels: Mapping[str, ChannelProtocol] | None,
+    step: int,
+    *,
+    id: str | None = None,
+) -> Checkpoint:
+    """Create a checkpoint for the given channels."""
+    ts = datetime.now(timezone.utc).isoformat()
+    if channels is None:
+        values = checkpoint["channel_values"]
+    else:
+        values = {}
+        for k, v in channels.items():
+            if k not in checkpoint["channel_versions"]:
+                continue
+            try:
+                values[k] = v.checkpoint()
+            except EmptyChannelError:
+                pass
+    return Checkpoint(
+        v=1,
+        ts=ts,
+        id=id or str(uuid6(clock_seq=step)),
+        channel_values=values,
+        channel_versions=checkpoint["channel_versions"],
+        versions_seen=checkpoint["versions_seen"],
+    )

libs/checkpoint/tests/test_memory.py

@@ -6,10 +6,12 @@ from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
     Checkpoint,
     CheckpointMetadata,
+)
+from langgraph.checkpoint.memory import InMemorySaver
+from tests.checkpoint_utils import (
     create_checkpoint,
     empty_checkpoint,
 )
-from langgraph.checkpoint.memory import InMemorySaver
 
 
 class TestMemorySaver:

libs/cli/generate_schema.py

@@ -165,6 +165,13 @@ def generate_schema():
     if "python_version" in python_schema["properties"]:
         python_schema["properties"]["python_version"]["enum"] = ["3.11", "3.12"]
 
+    # Add enum constraint for image_distro
+    if "image_distro" in python_schema["properties"]:
+        python_schema["properties"]["image_distro"]["anyOf"] = [
+            {"type": "string", "enum": ["debian", "wolfi"]},
+            {"type": "null"},
+        ]
+
     # Create Node.js schema with node_version
     node_schema = {
         "type": "object",
@@ -184,6 +191,13 @@ def generate_schema():
             {"type": "null"},
         ]
 
+    # Add enum constraint for image_distro
+    if "image_distro" in node_schema["properties"]:
+        node_schema["properties"]["image_distro"]["anyOf"] = [
+            {"type": "string", "enum": ["debian", "wolfi"]},
+            {"type": "null"},
+        ]
+
     # Replace the Config schema with a oneOf constraint
     config_schema["oneOf"] = [python_schema, node_schema]
 

libs/cli/langgraph_cli/cli.py

@@ -20,6 +20,7 @@ from langgraph_cli.docker import DockerCapabilities
 from langgraph_cli.exec import Runner, subp_exec
 from langgraph_cli.progress import Progress
 from langgraph_cli.templates import TEMPLATE_HELP_STRING, create_new
+from langgraph_cli.util import warn_non_wolfi_distro
 from langgraph_cli.version import __version__
 
 OPT_DOCKER_COMPOSE = click.option(
@@ -317,10 +318,8 @@ def _build(
     )
     # add additional_contexts
     if additional_contexts:
-        additional_contexts_str = ",".join(
-            f"{k}={v}" for k, v in additional_contexts.items()
-        )
-        args.extend(["--build-context", additional_contexts_str])
+        for k, v in additional_contexts.items():
+            args.extend(["--build-context", f"{k}={v}"])
     # run docker build
     runner.run(
         subp_exec(
@@ -375,6 +374,7 @@ def build(
         if shutil.which("docker") is None:
             raise click.UsageError("Docker not installed") from None
         config_json = langgraph_cli.config.validate_config_file(config)
+        warn_non_wolfi_distro(config_json)
         _build(
             runner, set, config, config_json, base_image, pull, tag, docker_build_args
         )
@@ -466,6 +466,7 @@ def dockerfile(
     save_path = pathlib.Path(save_path).absolute()
     secho(f"🔍 Validating configuration at path: {config}", fg="yellow")
     config_json = langgraph_cli.config.validate_config_file(config)
+    warn_non_wolfi_distro(config_json)
     secho("✅ Configuration validated!", fg="green")
 
     secho(f"📝 Generating Dockerfile at {save_path}", fg="yellow")
@@ -791,6 +792,7 @@ def prepare(
 ) -> tuple[list[str], str]:
     """Prepare the arguments and stdin for running the LangGraph API server."""
     config_json = langgraph_cli.config.validate_config_file(config_path)
+    warn_non_wolfi_distro(config_json)
     # pull latest images
     if pull:
         runner.run(

libs/cli/langgraph_cli/config.py

@@ -1,6 +1,7 @@
 import json
 import os
 import pathlib
+import re
 import textwrap
 from collections import Counter
 from typing import Any, Literal, NamedTuple, Optional, TypedDict, Union
@@ -13,6 +14,8 @@ DEFAULT_NODE_VERSION = "20"
 MIN_PYTHON_VERSION = "3.11"
 DEFAULT_PYTHON_VERSION = "3.11"
 
+DEFAULT_IMAGE_DISTRO = "debian"
+
 
 class TTLConfig(TypedDict, total=False):
     """Configuration for TTL (time-to-live) behavior in the store."""
@@ -367,6 +370,12 @@ class Config(TypedDict, total=False):
     
     Defaults to langchain/langgraph-api or langchain/langgraphjs-api."""
 
+    image_distro: Optional[str]
+    """Optional. Linux distribution for the base image.
+    
+    Must be either 'debian' or 'wolfi'. If omitted, defaults to 'debian'.
+    """
+
     pip_config_file: Optional[str]
     """Optional. Path to a pip config file (e.g., "/etc/pip.conf" or "pip.ini") for controlling
     package installation (custom indices, credentials, etc.).
@@ -453,12 +462,16 @@ class Config(TypedDict, total=False):
 PIP_CLEANUP_LINES = """# -- Ensure user deps didn't inadvertently overwrite langgraph-api
 RUN mkdir -p /api/langgraph_api /api/langgraph_runtime /api/langgraph_license && \
     touch /api/langgraph_api/__init__.py /api/langgraph_runtime/__init__.py /api/langgraph_license/__init__.py
-RUN PYTHONDONTWRITEBYTECODE=1 pip install --no-cache-dir --no-deps -e /api
+RUN PYTHONDONTWRITEBYTECODE=1 {install_cmd} --no-cache-dir --no-deps -e /api
 # -- End of ensuring user deps didn't inadvertently overwrite langgraph-api --
 # -- Removing pip from the final image ~<:===~~~ --
 RUN pip uninstall -y pip setuptools wheel && \
     rm -rf /usr/local/lib/python*/site-packages/pip* /usr/local/lib/python*/site-packages/setuptools* /usr/local/lib/python*/site-packages/wheel* && \
-    find /usr/local/bin -name "pip*" -delete
+    find /usr/local/bin -name "pip*" -delete || true
+# pip removal for wolfi
+RUN rm -rf /usr/lib/python*/site-packages/pip* /usr/lib/python*/site-packages/setuptools* /usr/lib/python*/site-packages/wheel* && \
+    find /usr/bin -name "pip*" -delete || true
+{uv_removal}
 # -- End of pip removal --"""
 
 
@@ -517,12 +530,15 @@ def validate_config(config: Config) -> Config:
         "python_version", DEFAULT_PYTHON_VERSION if some_python else None
     )
 
+    image_distro = config.get("image_distro", DEFAULT_IMAGE_DISTRO)
+
     config = {
         "node_version": node_version,
         "python_version": python_version,
         "pip_config_file": config.get("pip_config_file"),
         "_INTERNAL_docker_tag": config.get("_INTERNAL_docker_tag"),
         "base_image": config.get("base_image"),
+        "image_distro": image_distro,
         "dependencies": config.get("dependencies", []),
         "dockerfile_lines": config.get("dockerfile_lines", []),
         "graphs": config.get("graphs", {}),
@@ -576,6 +592,14 @@ def validate_config(config: Config) -> Config:
             "Add at least one graph to 'graphs' dictionary."
         )
 
+    # Validate image_distro config
+    if image_distro := config.get("image_distro"):
+        if image_distro not in ["debian", "wolfi"]:
+            raise click.UsageError(
+                f"Invalid image_distro: '{image_distro}'. "
+                "Must be either 'debian' or 'wolfi'."
+            )
+
     # Validate auth config
     if auth_conf := config.get("auth"):
         if "path" in auth_conf:
@@ -1067,16 +1091,38 @@ def _get_node_pm_install_cmd(config_path: pathlib.Path, config: Config) -> str:
     return install_cmd
 
 
+semver_pattern = re.compile(r":(\d+(?:\.\d+)?(?:\.\d+)?)(?:-|$)")
+
+
+def _image_supports_uv(base_image: str) -> bool:
+    if base_image == "langchain/langgraph-trial":
+        return False
+    match = semver_pattern.search(base_image)
+    if not match:
+        # Default image (langchain/langgraph-api) supports it.
+        return True
+
+    version_str = match.group(1)
+    version = tuple(map(int, version_str.split(".")))
+    min_uv = (0, 2, 47)
+    return version >= min_uv
+
+
 def python_config_to_docker(
     config_path: pathlib.Path,
     config: Config,
     base_image: str,
 ) -> tuple[str, dict[str, str]]:
     """Generate a Dockerfile from the configuration."""
+    if _image_supports_uv(base_image):
+        install_cmd = "uv pip install --system"
+        uv_removal = "RUN uv pip uninstall --system pip setuptools wheel && rm /usr/bin/uv /usr/bin/uvx"
+    else:
+        install_cmd = "pip install"
+        uv_removal = ""
+
     # configure pip
-    pip_install = (
-        "PYTHONDONTWRITEBYTECODE=1 pip install --no-cache-dir -c /api/constraints.txt"
-    )
+    pip_install = f"PYTHONDONTWRITEBYTECODE=1 {install_cmd} --no-cache-dir -c /api/constraints.txt"
     if config.get("pip_config_file"):
         pip_install = f"PIP_CONFIG_FILE=/pipconfig.txt {pip_install}"
     pip_config_file_str = (
@@ -1085,8 +1131,6 @@ def python_config_to_docker(
         else ""
     )
 
-    docker_tag = config.get("_INTERNAL_docker_tag") or config["python_version"]
-
     # collect dependencies
     pypi_deps = [dep for dep in config["dependencies"] if not dep.startswith(".")]
     local_deps = _assemble_local_deps(config_path, config)
@@ -1131,7 +1175,10 @@ RUN set -ex && \\
                 'name = "{fullpath.name}"' \\
                 'version = "0.1"' \\
                 '[tool.setuptools.package-data]' \\
-                '"*" = ["**/*"]'; do \\
+                '"*" = ["**/*"]' \\
+                '[build-system]' \\
+                'requires = ["setuptools>=61"]' \\
+                'build-backend = "setuptools.build_meta"'; do \\
         echo "$line" >> /deps/__outer_{fullpath.name}/pyproject.toml; \\
     done
 # -- End of non-package dependency {fullpath.name} --"""
@@ -1205,10 +1252,7 @@ ADD {relpath} /deps/{name}
                 "# -- End of JS dependencies install --",
             ]
         )
-    if "/langgraph-server" in base_image:
-        image_str = f"{base_image}-py{docker_tag}"
-    else:
-        image_str = f"{base_image}:{docker_tag}"
+    image_str = docker_tag(config, base_image)
     docker_file_contents = [
         f"FROM {image_str}",
         "",
@@ -1223,7 +1267,8 @@ ADD {relpath} /deps/{name}
         "",
         js_inst_str,
         "",
-        PIP_CLEANUP_LINES,  # Add pip cleanup after all installations are complete
+        # Add pip cleanup after all installations are complete
+        PIP_CLEANUP_LINES.format(install_cmd=install_cmd, uv_removal=uv_removal),
         "",
         f"WORKDIR {local_deps.working_dir}" if local_deps.working_dir else "",
     ]
@@ -1248,7 +1293,7 @@ def node_config_to_docker(
 ) -> tuple[str, dict[str, str]]:
     faux_path = f"/deps/{config_path.parent.name}"
     install_cmd = _get_node_pm_install_cmd(config_path, config)
-    docker_tag = config.get("_INTERNAL_docker_tag") or config["node_version"]
+    image_str = docker_tag(config, base_image)
 
     env_vars: list[str] = []
 
@@ -1275,7 +1320,7 @@ def node_config_to_docker(
     env_vars.append(f"ENV LANGSERVE_GRAPHS='{json.dumps(config['graphs'])}'")
 
     docker_file_contents = [
-        f"FROM {base_image}:{docker_tag}",
+        f"FROM {image_str}",
         "",
         os.linesep.join(config["dockerfile_lines"]),
         "",
@@ -1306,6 +1351,10 @@ def docker_tag(
     base_image: Optional[str] = None,
 ) -> str:
     base_image = base_image or default_base_image(config)
+
+    image_distro = config.get("image_distro")
+    distro_tag = "" if image_distro == DEFAULT_IMAGE_DISTRO else f"-{image_distro}"
+
     if config.get("_INTERNAL_docker_tag"):
         return f"{base_image}:{config['_INTERNAL_docker_tag']}"
 
@@ -1313,8 +1362,8 @@ def docker_tag(
         return f"{base_image}-py{config['python_version']}"
 
     if config.get("node_version") and not config.get("python_version"):
-        return f"{base_image}:{config['node_version']}"
-    return f"{base_image}:{config['python_version']}"
+        return f"{base_image}:{config['node_version']}{distro_tag}"
+    return f"{base_image}:{config['python_version']}{distro_tag}"
 
 
 def config_to_docker(

libs/cli/langgraph_cli/util.py

@@ -1,2 +1,25 @@
+import click
+
+
 def clean_empty_lines(input_str: str):
     return "\n".join(filter(None, input_str.splitlines()))
+
+
+def warn_non_wolfi_distro(config_json: dict) -> None:
+    """Show warning if image_distro is not set to 'wolfi'."""
+    image_distro = config_json.get("image_distro", "debian")  # Default is debian
+    if image_distro != "wolfi":
+        click.secho(
+            "⚠️  Security Recommendation: Consider switching to Wolfi Linux for enhanced security.",
+            fg="yellow",
+            bold=True,
+        )
+        click.secho(
+            "   Wolfi is a security-oriented, minimal Linux distribution designed for containers.",
+            fg="yellow",
+        )
+        click.secho(
+            '   To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.',
+            fg="yellow",
+        )
+        click.secho("")  # Empty line for better readability

libs/cli/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-cli"
-version = "0.2.10"
+version = "0.3.1"
 description = "CLI for interacting with LangGraph API"
 authors = []
 requires-python = ">=3.9"