langgraph: release 0.2.67 (#3182)

Simple update that adds support for the `update` attribute of the
`Command` class to support Pydantic `BaseModel` type.

LangGraph already supports [Pydantic models for graph
states](https://langchain-ai.github.io/langgraph/how-tos/state-model/).

Extending support to the `update` attribute allows users to pass custom
BaseModel instances. Additionally, updates defined as `BaseModel` types
are type-validated when created.

https://github.com/langchain-ai/langgraph/issues/2804

---------

Co-authored-by: vbarda <vadym@langchain.dev>

.github/workflows/codespell.yml

@@ -21,7 +21,7 @@
     
           - name: Install Dependencies
             run: |
-              pip install toml codespell jupytext
+              pip install toml codespell==2.3.0 jupytext
     
           - name: Extract Ignore Words List
             run: |

docs/_scripts/_patch.py

@@ -31,7 +31,7 @@ def request(self, method, url, body=None, headers=None):
         The result of calling the parent request method.
     """
     # Update the inner socket's timeout value to send the request.
-    # This only triggers if the connection is re-used.
+    # This only triggers if the connection is reused.
     if getattr(self, "sock", None) is not None:
         self.sock.settimeout(self.timeout)
 
@@ -90,4 +90,4 @@ def patch_urllib3():
         return request(self, *args, **kwargs)
 
     connection.HTTPConnection.request = new_request
-    _PATCHED = True
+    _PATCHED = True

docs/docs/concepts/high_level.md

@@ -1,58 +1,26 @@
 # Why LangGraph?
 
-LLMs are extremely powerful, particularly when connected to other systems such as a retriever or APIs. This is why many LLM applications use a control flow of steps before and / or after LLM calls. As an example [RAG](https://github.com/langchain-ai/rag-from-scratch) performs retrieval of relevant documents to a question, and passes those documents to an LLM in order to ground the response. Often a control flow of steps before and / or after an LLM is called a "chain." Chains are a popular paradigm for programming with LLMs and offer a high degree of reliability; the same set of steps runs with each chain invocation.
+## LLM applications
 
-However, we often want LLM systems that can pick their own control flow! This is one definition of an [agent](https://blog.langchain.dev/what-is-an-agent/): an agent is a system that uses an LLM to decide the control flow of an application. Unlike a chain, an agent gives an LLM some degree of control over the sequence of steps in the application. Examples of using an LLM to decide the control of an application:
+LLMs make it possible to embed intelligence into a new class of applications. There are many patterns for building applications that use LLMs. [Workflows](https://www.anthropic.com/research/building-effective-agents) have scaffolding of predefined code paths around LLM calls. LLMs can direct the control flow through these predefined code paths, which some consider to be an "[agentic system](https://www.anthropic.com/research/building-effective-agents)". In other cases, it's possible to remove this scaffolding, creating autonomous agents that can [plan](https://huyenchip.com/2025/01/07/agents.html), take actions via [tool calls](https://python.langchain.com/docs/concepts/tool_calling/), and directly respond [to the feedback from their own actions](https://research.google/blog/react-synergizing-reasoning-and-acting-in-language-models/) with further actions.
 
-- Using an LLM to route between two potential paths
-- Using an LLM to decide which of many tools to call
-- Using an LLM to decide whether the generated answer is sufficient or more work is need
+![Agent Workflow](img/agent_workflow.png)
 
-There are many different types of [agent architectures](https://blog.langchain.dev/what-is-a-cognitive-architecture/) to consider, which give an LLM varying levels of control. On one extreme, a router allows an LLM to select a single step from a specified set of options and, on the other extreme, a fully autonomous long-running agent may have complete freedom to select any sequence of steps that it wants for a given problem. 
+## What LangGraph provides
 
-![Agent Types](img/agent_types.png)
+LangGraph provides low-level supporting infrastructure that sits underneath *any* workflow or agent. It does not abstract prompts or architecture, and provides three central benefits:
 
-Several concepts are utilized in many agent architectures:
+### Persistence
 
-- [Tool calling](agentic_concepts.md#tool-calling): this is often how LLMs make decisions
-- Action taking: often times, the LLMs' outputs are used as the input to an action
-- [Memory](agentic_concepts.md#memory): reliable systems need to have knowledge of things that occurred
-- [Planning](agentic_concepts.md#planning): planning steps (either explicit or implicit) are useful for ensuring that the LLM, when making decisions, makes them in the highest fidelity way.
+LangGraph has a [persistence layer](https://langchain-ai.github.io/langgraph/concepts/persistence/), which offers a number of benefits:
 
-## Challenges
+- [Memory](https://langchain-ai.github.io/langgraph/concepts/memory/): LangGraph persists arbitrary aspects of your application's state, supporting memory of conversations and other updates within and across user interactions;
+- [Human-in-the-loop](https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/): Because state is checkpointed, execution can be interrupted and resumed, allowing for decisions, validation, and corrections via human input.
 
-In practice, there is often a trade-off between control and reliability. As we give LLMs more control, the application often become less reliable. This can be due to factors such as LLM non-determinism and / or errors in selecting tools (or steps) that the agent uses (takes).
+### Streaming
 
-![Agent Challenge](img/challenge.png)
+LangGraph also provides support for [streaming](../how-tos/index.md#streaming) workflow / agent state to the user (or developer) over the course of execution. LangGraph supports streaming of both events ([such as feedback from a tool call](../how-tos/stream-updates.ipynb)) and [tokens from LLM calls](../how-tos/streaming-tokens.ipynb) embedded in an application.
 
-## Core Principles
+### Debugging and Deployment
 
-The motivation of LangGraph is to help bend the curve, preserving higher reliability as we give the agent more control over the application. We'll outline a few specific pillars of LangGraph that make it well suited for building reliable agents. 
-
-![Langgraph](img/langgraph.png)
-
-**Controllability**
-
-LangGraph gives the developer a high degree of [control](../how-tos/index.md#controllability) by expressing the flow of the application as a set of nodes and edges. All nodes can access and modify a common state (memory). The control flow of the application can set using edges that connect nodes, either deterministically or via conditional logic. 
-
-**Persistence**
-
-LangGraph gives the developer many options for [persisting](../how-tos/index.md#persistence) graph state using short-term or long-term (e.g., via a database) memory. 
-
-**Human-in-the-Loop**
-
-The persistence layer enables several different [human-in-the-loop](../how-tos/index.md#human-in-the-loop) interaction patterns with agents; for example, it's possible to pause an agent, review its state, edit it state, and approve a follow-up step. 
-
-**Streaming**
-
-LangGraph comes with first class support for [streaming](../how-tos/index.md#streaming), which can expose state to the user (or developer) over the course of agent execution. LangGraph supports streaming of both events ([like a tool call being taken](../how-tos/stream-updates.ipynb)) as well as of [tokens that an LLM may emit](../how-tos/streaming-tokens.ipynb).
-
-## Debugging
-
-Once you've built a graph, you often want to test and debug it. [LangGraph Studio](https://github.com/langchain-ai/langgraph-studio?tab=readme-ov-file) is a specialized IDE for visualization and debugging of LangGraph applications.
-
-![Langgraph Studio](img/lg_studio.png)
-
-## Deployment
-
-Once you have confidence in your LangGraph application, many developers want an easy path to deployment. [LangGraph Platform](../concepts/index.md#langgraph-platform) offers a range of options for deploying LangGraph graphs.
+LangGraph provides an easy onramp for testing, debugging, and deploying applications via [LangGraph Platform](https://langchain-ai.github.io/langgraph/concepts/langgraph_platform/). This includes [Studio](https://langchain-ai.github.io/langgraph/concepts/langgraph_studio/), an IDE that enables visualization, interaction, and debugging of workflows or agents. This also includes numerous [options](https://langchain-ai.github.io/langgraph/tutorials/deployment/) for deployment. 

docs/docs/how-tos/cross-thread-persistence.ipynb

@@ -64,18 +64,10 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": null,
    "id": "aa2c64a7",
    "metadata": {},
-   "outputs": [
-    {
-     "name": "stdout",
-     "output_type": "stream",
-     "text": [
-      "ANTHROPIC_API_KEY:  ········\n"
-     ]
-    }
-   ],
+   "outputs": [],
    "source": [
     "import getpass\n",
     "import os\n",
@@ -86,7 +78,8 @@
     "        os.environ[var] = getpass.getpass(f\"{var}: \")\n",
     "\n",
     "\n",
-    "_set_env(\"ANTHROPIC_API_KEY\")"
+    "_set_env(\"ANTHROPIC_API_KEY\")\n",
+    "_set_env(\"OPENAI_API_KEY\")"
    ]
   },
   {
@@ -356,7 +349,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.11.9"
+   "version": "3.12.3"
   }
  },
  "nbformat": 4,

docs/docs/how-tos/deploy-self-hosted.md

@@ -25,8 +25,20 @@ If you would like to deploy LangGraph Cloud on Kubernetes, you can use this [Hel
 
 You will eventually need to pass in the following environment variables to the LangGraph Deploy server:
 
-- `REDIS_URI`: Connection details to a Redis instance. Redis will be used as a pub-sub broker to enable streaming real time output from background runs.
-- `DATABASE_URI`: Postgres connection details. Postgres will be used to store assistants, threads, runs, persist thread state and long term memory, and to manage the state of the background task queue with 'exactly once' semantics.
+- `REDIS_URI`: Connection details to a Redis instance. Redis will be used as a pub-sub broker to enable streaming real time output from background runs. The value of `REDIS_URI` must be a valid [Redis connection URI](https://redis-py.readthedocs.io/en/stable/connections.html#redis.Redis.from_url).
+
+    !!! Note "Shared Redis Instance"
+        Multiple self-hosted deployments can share the same Redis instance. For example, for `Deployment A`, `REDIS_URI` can be set to `redis://<hostname_1>:<port>/1` and for `Deployment B`, `REDIS_URI` can be set to `redis://<hostname_1>:<port>/2`.
+
+        `1` and `2` are different database numbers within the same instance, but `<hostname_1>` is shared. **The same database number cannot be used for separate deployments**.
+
+- `DATABASE_URI`: Postgres connection details. Postgres will be used to store assistants, threads, runs, persist thread state and long term memory, and to manage the state of the background task queue with 'exactly once' semantics. The value of `DATABASE_URI` must be a valid [Postgres connection URI](https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING-URIS).
+
+    !!! Note "Shared Postgres Instance"
+        Multiple self-hosted deployments can share the same Postgres instance. For example, for `Deployment A`, `DATABASE_URI` can be set to `postgres://<user>:<password>@/<database_name_1>?host=<hostname_1>` and for `Deployment B`, `DATABASE_URI` can be set to `postgres://<user>:<password>@/<database_name_2>?host=<hostname_1>`.
+
+        `<database_name_1>` and `database_name_2` are different databases within the same instance, but `<hostname_1>` is shared. **The same database cannot be used for separate deployments**.
+
 - `LANGSMITH_API_KEY`: (If using [Self-Hosted Lite](../concepts/deployment_options.md#self-hosted-lite)) LangSmith API key. This will be used to authenticate ONCE at server start up.
 - `LANGGRAPH_CLOUD_LICENSE_KEY`: (If using [Self-Hosted Enterprise](../concepts/deployment_options.md#self-hosted-enterprise)) LangGraph Platform license key. This will be used to authenticate ONCE at server start up.
 - `LANGCHAIN_ENDPOINT`: To send traces to a [self-hosted LangSmith](https://docs.smith.langchain.com/self_hosting) instance, set `LANGCHAIN_ENDPOINT` to the hostname of the self-hosted LangSmith instance.

docs/docs/how-tos/persistence.ipynb

@@ -78,7 +78,7 @@
    "id": "0abe11f4-62ed-4dc4-8875-3db21e260d1d",
    "metadata": {},
    "source": [
-    "Next, we need to set API keys for OpenAI (the LLM we will use) and Tavily (the search tool we will use)"
+    "Next, we need to set API key for Anthropic (the LLM we will use)."
    ]
   },
   {
@@ -378,7 +378,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.11.9"
+   "version": "3.12.3"
   }
  },
  "nbformat": 4,

docs/docs/tutorials/index.md

@@ -9,6 +9,7 @@ New to LangGraph or LLM app development? Read this material to get up and runnin
 ## Get Started 🚀 {#quick-start}
 
 - [LangGraph Quickstart](introduction.ipynb): Build a chatbot that can use tools and keep track of conversation history. Add human-in-the-loop capabilities and explore how time-travel works.
+- [LangGraph Cheatsheet For Common Workflows](workflows.ipynb): Overview of the most common workflows and agent architectures in LangGraph.
 - [LangGraph Server Quickstart](langgraph-platform/local-server.md): Launch a LangGraph server locally and interact with it using REST API and LangGraph Studio Web UI.
 - [LangGraph Template Quickstart](../concepts/template_applications.md): Start building with LangGraph Platform using a template application.
 - [Deploy with LangGraph Cloud Quickstart](../cloud/quick_start.md): Deploy a LangGraph app using LangGraph Cloud.

docs/docs/tutorials/plan-and-execute/plan-and-execute.ipynb

@@ -130,36 +130,19 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 7,
+   "execution_count": null,
    "id": "72d233ca-1dbf-4b43-b680-b3bf39e3691f",
    "metadata": {},
-   "outputs": [
-    {
-     "name": "stdout",
-     "output_type": "stream",
-     "text": [
-      "================================\u001b[1m System Message \u001b[0m================================\n",
-      "\n",
-      "You are a helpful assistant.\n",
-      "\n",
-      "=============================\u001b[1m Messages Placeholder \u001b[0m=============================\n",
-      "\n",
-      "\u001b[33;1m\u001b[1;3m{messages}\u001b[0m\n"
-     ]
-    }
-   ],
+   "outputs": [],
    "source": [
     "from langchain import hub\n",
     "from langchain_openai import ChatOpenAI\n",
     "\n",
     "from langgraph.prebuilt import create_react_agent\n",
     "\n",
-    "# Get the prompt to use - you can modify this!\n",
-    "prompt = hub.pull(\"ih/ih-react-agent-executor\")\n",
-    "prompt.pretty_print()\n",
-    "\n",
     "# Choose the LLM that will drive the agent\n",
     "llm = ChatOpenAI(model=\"gpt-4-turbo-preview\")\n",
+    "prompt = \"You are a helpful assistant.\"\n",
     "agent_executor = create_react_agent(llm, tools, state_modifier=prompt)"
    ]
   },
@@ -546,7 +529,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.11.9"
+   "version": "3.12.3"
   }
  },
  "nbformat": 4,

docs/mkdocs.yml

@@ -296,6 +296,7 @@ nav:
         - Quick Start:
           - Quick Start: tutorials#quick-start
           - tutorials/introduction.ipynb
+          - tutorials/workflows.ipynb
           - tutorials/langgraph-platform/local-server.md
           - cloud/quick_start.md
         - Chatbots:

libs/cli/langgraph_cli/config.py

@@ -330,7 +330,7 @@ def _assemble_local_deps(config_path: pathlib.Path, config: Config) -> LocalDeps
                 rfile = resolved / "requirements.txt"
                 pip_reqs.append(
                     (
-                        rfile.relative_to(config_path.parent),
+                        rfile.relative_to(config_path.parent).as_posix(),
                         f"{container_path}/requirements.txt",
                     )
                 )

libs/langgraph/langgraph/constants.py

@@ -23,6 +23,7 @@ END = sys.intern("__end__")
 """The last (maybe virtual) node in graph-style Pregel."""
 SELF = sys.intern("__self__")
 """The implicit branch that handles each node's Control values."""
+PREVIOUS = sys.intern("__previous__")
 
 # --- Reserved write keys ---
 INPUT = sys.intern("__input__")
@@ -78,7 +79,7 @@ CONFIG_KEY_NODE_FINISHED = sys.intern("__pregel_node_finished")
 # holds a callback to be called when a node is finished
 CONFIG_KEY_SCRATCHPAD = sys.intern("__pregel_scratchpad")
 # holds a mutable dict for temporary storage scoped to the current task
-CONFIG_KEY_END = sys.intern("__pregel_previous")
+CONFIG_KEY_PREVIOUS = sys.intern("__pregel_previous")
 # holds the previous return value from a stateful Pregel graph.
 
 # --- Other constants ---

libs/langgraph/langgraph/func/__init__.py

@@ -4,12 +4,17 @@ import functools
 import inspect
 import types
 from collections.abc import Iterator
+from dataclasses import dataclass
 from typing import (
     Any,
     Awaitable,
     Callable,
+    Generic,
     Optional,
+    TypeVar,
     Union,
+    get_args,
+    get_origin,
     overload,
 )
 
@@ -20,14 +25,14 @@ from langchain_core.runnables.graph import Graph, Node
 from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue
 from langgraph.checkpoint.base import BaseCheckpointSaver
-from langgraph.constants import END, START, TAG_HIDDEN
+from langgraph.constants import END, PREVIOUS, START, TAG_HIDDEN
 from langgraph.pregel import Pregel
 from langgraph.pregel.call import P, T, call, get_runnable_for_entrypoint
 from langgraph.pregel.protocol import PregelProtocol
 from langgraph.pregel.read import PregelNode
 from langgraph.pregel.write import ChannelWrite, ChannelWriteEntry
 from langgraph.store.base import BaseStore
-from langgraph.types import RetryPolicy, StreamMode, StreamWriter
+from langgraph.types import _DC_KWARGS, RetryPolicy, StreamMode, StreamWriter
 
 
 @overload
@@ -140,12 +145,15 @@ def task(
     return decorator
 
 
-def entrypoint(
-    *,
-    checkpointer: Optional[BaseCheckpointSaver] = None,
-    store: Optional[BaseStore] = None,
-    config_schema: Optional[type[Any]] = None,
-) -> Callable[[types.FunctionType], Pregel]:
+R = TypeVar("R")
+S = TypeVar("S")
+
+
+# The decorator was wrapped in a class to support the `final` attribute.
+# In this form, the `final` attribute should play nicely with IDE autocompletion,
+# and type checking tools.
+# In addition, we'll be able to surface this information in the API Reference.
+class entrypoint:
     """Define a LangGraph workflow using the `entrypoint` decorator.
 
     !!! warning "Experimental"
@@ -156,8 +164,10 @@ def entrypoint(
     to the function. This input parameter can be of any type. Use a dictionary
     to pass multiple parameters to the function.
 
-    The decorated function also has access to these optional parameters:
+    The decorated function can request access to additional parameters
+    that will be injected automatically at run time. These parameters include:
 
+    - `store`: An instance of [BaseStore][langgraph.store.base.BaseStore]. Useful for long-term memory.
     - `writer`: A `StreamWriter` instance for writing data to a stream.
     - `config`: A configuration object for accessing workflow settings.
     - `previous`: The previous return value for the given thread (available only when
@@ -178,9 +188,6 @@ def entrypoint(
         config_schema: Specifies the schema for the configuration object that will be
             passed to the workflow.
 
-    Returns:
-        A decorator that converts a function into a Pregel graph.
-
     Example: Using entrypoint and tasks
         ```python
         import time
@@ -250,7 +257,7 @@ def entrypoint(
 
         ```python
         from langgraph.checkpoint.memory import MemorySaver
-        from langgraph.func import entrypoint, task
+        from langgraph.func import entrypoint
 
         @entrypoint(checkpointer=MemorySaver())
         def my_workflow(input_data: str, previous: Optional[str] = None) -> str:
@@ -266,7 +273,57 @@ def entrypoint(
         ```
     """
 
-    def _imp(func: types.FunctionType) -> Pregel:
+    def __init__(
+        self,
+        checkpointer: Optional[BaseCheckpointSaver] = None,
+        store: Optional[BaseStore] = None,
+        config_schema: Optional[type[Any]] = None,
+    ) -> None:
+        """Initialize the entrypoint decorator."""
+        self.checkpointer = checkpointer
+        self.store = store
+        self.config_schema = config_schema
+
+    @dataclass(**_DC_KWARGS)
+    class final(Generic[R, S]):
+        """A primitive that can be returned from an entrypoint.
+
+        This primitive allows to save a value to the checkpointer distinct from the
+        return value from the entrypoint.
+
+        Example: Decoupling the return value and the save value
+            ```python
+            from langgraph.checkpoint.memory import MemorySaver
+            from langgraph.func import entrypoint
+
+            @entrypoint(checkpointer=MemorySaver())
+            def my_workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
+                previous = previous or 0
+                # This will return the previous value to the caller, saving
+                # 2 * number to the checkpoint, which will be used in the next invocation
+                # for the `previous` parameter.
+                return entrypoint.final(value=previous, save=2 * number)
+
+            config = {
+                "configurable": {
+                    "thread_id": "1"
+                }
+            }
+
+            my_workflow.invoke(3, config)  # 0 (previous was None)
+            my_workflow.invoke(1, config)  # 6 (previous was 3 * 2 from the previous invocation)
+            ```
+        """
+
+        value: R
+        """Value to return. A value will always be returned even if it is None."""
+        save: S
+        """The value for the state for the next checkpoint. 
+
+        A value will always be saved even if it is None.
+        """
+
+    def __call__(self, func: types.FunctionType) -> Pregel:
         """Convert a function into a Pregel graph.
 
         Args:
@@ -286,22 +343,53 @@ def entrypoint(
 
                 @functools.wraps(func)
                 def gen_wrapper(*args: Any, writer: StreamWriter, **kwargs: Any) -> Any:
+                    final_: Optional[entrypoint.final] = None
                     chunks = []
                     for chunk in func(*args, writer=writer, **kwargs):
-                        writer(chunk)
-                        chunks.append(chunk)
-                    return chunks
+                        if isinstance(chunk, entrypoint.final):
+                            if final_ is not None:
+                                raise RuntimeError(
+                                    "Yielding multiple entrypoint.final "
+                                    "objects is not allowed."
+                                )
+                            else:
+                                final_ = chunk
+                        else:
+                            if final_ is not None:
+                                raise RuntimeError(
+                                    "Yielding a value after a entrypoint.final "
+                                    "object is not allowed."
+                                )
+                            writer(chunk)
+                            chunks.append(chunk)
+
+                    return final_ if final_ else chunks
             else:
 
                 @functools.wraps(func)
                 def gen_wrapper(*args: Any, writer: StreamWriter, **kwargs: Any) -> Any:
+                    final_: Optional[entrypoint.final] = None
                     chunks = []
                     # Do not pass the writer argument to the wrapped function
                     # as it does not have a matching parameter
                     for chunk in func(*args, **kwargs):
-                        writer(chunk)
-                        chunks.append(chunk)
-                    return chunks
+                        if isinstance(chunk, entrypoint.final):
+                            if final_ is not None:
+                                raise RuntimeError(
+                                    "Yielding multiple entrypoint.final "
+                                    "objects is not allowed."
+                                )
+                            else:
+                                final_ = chunk
+                        else:
+                            if final_ is not None:
+                                raise RuntimeError(
+                                    "Yielding a value after a entrypoint.final "
+                                    "object is not allowed."
+                                )
+                            writer(chunk)
+                            chunks.append(chunk)
+                    return final_ if final_ else chunks
 
                 # Create a new parameter for the writer argument
                 extra_param = inspect.Parameter(
@@ -329,22 +417,50 @@ def entrypoint(
                 async def agen_wrapper(
                     *args: Any, writer: StreamWriter, **kwargs: Any
                 ) -> Any:
+                    final_: Optional[entrypoint.final] = None
                     chunks = []
                     async for chunk in func(*args, writer=writer, **kwargs):
-                        writer(chunk)
-                        chunks.append(chunk)
-                    return chunks
+                        if isinstance(chunk, entrypoint.final):
+                            if final_ is not None:
+                                raise RuntimeError(
+                                    "Yielding multiple entrypoint.final objects is not allowed."
+                                )
+                            else:
+                                final_ = chunk
+                        else:
+                            if final_ is not None:
+                                raise RuntimeError(
+                                    "Yielding a value after a entrypoint.final object is not allowed."
+                                )
+                            writer(chunk)
+                            chunks.append(chunk)
+
+                    return final_ if final_ else chunks
             else:
 
                 @functools.wraps(func)
                 async def agen_wrapper(
                     *args: Any, writer: StreamWriter, **kwargs: Any
                 ) -> Any:
+                    final_: Optional[entrypoint.final] = None
                     chunks = []
                     async for chunk in func(*args, **kwargs):
-                        writer(chunk)
-                        chunks.append(chunk)
-                    return chunks
+                        if isinstance(chunk, entrypoint.final):
+                            if final_ is not None:
+                                raise RuntimeError(
+                                    "Yielding multiple entrypoint.final objects is not allowed."
+                                )
+                            else:
+                                final_ = chunk
+                        else:
+                            if final_ is not None:
+                                raise RuntimeError(
+                                    "Yielding a value after a entrypoint.final object is not allowed."
+                                )
+                            writer(chunk)
+                            chunks.append(chunk)
+
+                    return final_ if final_ else chunks
 
                 # Create a new parameter for the writer argument
                 extra_param = inspect.Parameter(
@@ -376,11 +492,36 @@ def entrypoint(
             is not inspect.Signature.empty
             else Any
         )
-        output_type = (
-            sig.return_annotation
-            if sig.return_annotation is not inspect.Signature.empty
-            else Any
-        )
+
+        def _pluck_return_value(value: Any) -> Any:
+            """Extract the return_ value the entrypoint.final object or passthrough."""
+            return value.value if isinstance(value, entrypoint.final) else value
+
+        def _pluck_save_value(value: Any) -> Any:
+            """Get save value from the entrypoint.final object or passthrough."""
+            return value.save if isinstance(value, entrypoint.final) else value
+
+        output_type, save_type = Any, Any
+        if sig.return_annotation is not inspect.Signature.empty:
+            # User does not parameterize entrypoint.final properly
+            if (
+                sig.return_annotation is entrypoint.final
+            ):  # Un-parameterized entrypoint.final
+                output_type = save_type = Any
+            else:
+                origin = get_origin(sig.return_annotation)
+                if origin is entrypoint.final:
+                    type_annotations = get_args(sig.return_annotation)
+                    if len(type_annotations) != 2:
+                        raise TypeError(
+                            "Please an annotation for both the return_ and "
+                            "the save values."
+                            "For example, `-> entrypoint.final[int, str]` would assign a "
+                            "return_ a type of `int` and save the type `str`."
+                        )
+                    output_type, save_type = get_args(sig.return_annotation)
+                else:
+                    output_type = save_type = sig.return_annotation
 
         return EntrypointPregel(
             nodes={
@@ -388,25 +529,32 @@ def entrypoint(
                     bound=bound,
                     triggers=[START],
                     channels=[START],
-                    writers=[ChannelWrite([ChannelWriteEntry(END)], tags=[TAG_HIDDEN])],
+                    writers=[
+                        ChannelWrite(
+                            [
+                                ChannelWriteEntry(END, mapper=_pluck_return_value),
+                                ChannelWriteEntry(PREVIOUS, mapper=_pluck_save_value),
+                            ],
+                            tags=[TAG_HIDDEN],
+                        )
+                    ],
                 )
             },
             channels={
                 START: EphemeralValue(input_type),
                 END: LastValue(output_type, END),
+                PREVIOUS: LastValue(save_type, PREVIOUS),
             },
             input_channels=START,
             output_channels=END,
             stream_channels=END,
             stream_mode=stream_mode,
             stream_eager=True,
-            checkpointer=checkpointer,
-            store=store,
-            config_type=config_schema,
+            checkpointer=self.checkpointer,
+            store=self.store,
+            config_type=self.config_schema,
         )
 
-    return _imp
-
 
 class EntrypointPregel(Pregel):
     def get_graph(

libs/langgraph/langgraph/graph/state.py

@@ -379,14 +379,27 @@ class StateGraph(Graph):
                     if input_hint := hints.get(first_parameter_name):
                         if isinstance(input_hint, type) and get_type_hints(input_hint):
                             input = input_hint
-                if (
-                    (rtn := hints.get("return"))
-                    and get_origin(rtn) is Command
-                    and (rargs := get_args(rtn))
-                    and get_origin(rargs[0]) is Literal
-                    and (vals := get_args(rargs[0]))
-                ):
-                    ends = vals
+                if rtn := hints.get("return"):
+                    # Handle Union types
+                    rtn_origin = get_origin(rtn)
+                    if rtn_origin is Union:
+                        rtn_args = get_args(rtn)
+                        # Look for Command in the union
+                        for arg in rtn_args:
+                            arg_origin = get_origin(arg)
+                            if arg_origin is Command:
+                                rtn = arg
+                                rtn_origin = arg_origin
+                                break
+
+                    # Check if it's a Command type
+                    if (
+                        rtn_origin is Command
+                        and (rargs := get_args(rtn))
+                        and get_origin(rargs[0]) is Literal
+                        and (vals := get_args(rargs[0]))
+                    ):
+                        ends = vals
         except (TypeError, StopIteration):
             pass
         if input is not None:

libs/langgraph/langgraph/prebuilt/chat_agent_executor.py

@@ -1,3 +1,4 @@
+import inspect
 from typing import (
     Callable,
     Literal,
@@ -91,6 +92,12 @@ def _get_state_modifier_runnable(
             lambda state: [state_modifier] + state["messages"],
             name=STATE_MODIFIER_RUNNABLE_NAME,
         )
+    elif inspect.iscoroutinefunction(state_modifier):
+        state_modifier_runnable = RunnableCallable(
+            None,
+            state_modifier,
+            name=STATE_MODIFIER_RUNNABLE_NAME,
+        )
     elif callable(state_modifier):
         state_modifier_runnable = RunnableCallable(
             state_modifier,

libs/langgraph/langgraph/pregel/algo.py

@@ -1,3 +1,5 @@
+import functools
+import itertools
 import sys
 from collections import defaultdict, deque
 from functools import partial
@@ -37,7 +39,7 @@ from langgraph.constants import (
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_CHECKPOINT_NS,
     CONFIG_KEY_CHECKPOINTER,
-    CONFIG_KEY_END,
+    CONFIG_KEY_PREVIOUS,
     CONFIG_KEY_READ,
     CONFIG_KEY_SCRATCHPAD,
     CONFIG_KEY_SEND,
@@ -46,11 +48,11 @@ from langgraph.constants import (
     EMPTY_SEQ,
     ERROR,
     INTERRUPT,
-    MISSING,
     NO_WRITES,
     NS_END,
     NS_SEP,
     NULL_TASK_ID,
+    PREVIOUS,
     PULL,
     PUSH,
     RESERVED,
@@ -324,7 +326,7 @@ def apply_writes(
 @overload
 def prepare_next_tasks(
     checkpoint: Checkpoint,
-    pending_writes: Sequence[PendingWrite],
+    pending_writes: list[PendingWrite],
     processes: Mapping[str, PregelNode],
     channels: Mapping[str, BaseChannel],
     managed: ManagedValueMapping,
@@ -341,7 +343,7 @@ def prepare_next_tasks(
 @overload
 def prepare_next_tasks(
     checkpoint: Checkpoint,
-    pending_writes: Sequence[PendingWrite],
+    pending_writes: list[PendingWrite],
     processes: Mapping[str, PregelNode],
     channels: Mapping[str, BaseChannel],
     managed: ManagedValueMapping,
@@ -357,7 +359,7 @@ def prepare_next_tasks(
 
 def prepare_next_tasks(
     checkpoint: Checkpoint,
-    pending_writes: Sequence[PendingWrite],
+    pending_writes: list[PendingWrite],
     processes: Mapping[str, PregelNode],
     channels: Mapping[str, BaseChannel],
     managed: ManagedValueMapping,
@@ -418,7 +420,7 @@ def prepare_single_task(
     task_id_checksum: Optional[str],
     *,
     checkpoint: Checkpoint,
-    pending_writes: Sequence[PendingWrite],
+    pending_writes: list[PendingWrite],
     processes: Mapping[str, PregelNode],
     channels: Mapping[str, BaseChannel],
     managed: ManagedValueMapping,
@@ -508,9 +510,6 @@ def prepare_single_task(
                             pending_writes,
                             task_id,
                         ),
-                        CONFIG_KEY_END: checkpoint["channel_values"].get(
-                            "__end__", None
-                        ),
                     },
                 ),
                 triggers,
@@ -620,8 +619,8 @@ def prepare_single_task(
                                 pending_writes,
                                 task_id,
                             ),
-                            CONFIG_KEY_END: checkpoint["channel_values"].get(
-                                "__end__", None
+                            CONFIG_KEY_PREVIOUS: checkpoint["channel_values"].get(
+                                PREVIOUS, None
                             ),
                         },
                     ),
@@ -744,8 +743,8 @@ def prepare_single_task(
                                     pending_writes,
                                     task_id,
                                 ),
-                                CONFIG_KEY_END: checkpoint["channel_values"].get(
-                                    "__end__", None
+                                CONFIG_KEY_PREVIOUS: checkpoint["channel_values"].get(
+                                    PREVIOUS, None
                                 ),
                             },
                         ),
@@ -761,23 +760,27 @@ def prepare_single_task(
 
 
 def _scratchpad(
-    pending_writes: Sequence[PendingWrite],
+    pending_writes: list[PendingWrite],
     task_id: str,
 ) -> PregelScratchpad:
+    null_resume_write = next(
+        (w for w in pending_writes if w[0] == NULL_TASK_ID and w[1] == RESUME), None
+    )
+    # using itertools.count as an atomic counter (+= 1 is not thread-safe)
     return PregelScratchpad(
         # call
-        call_counter=0,
+        call_counter=itertools.count(0).__next__,
         # interrupt
-        interrupt_counter=-1,
+        interrupt_counter=itertools.count(0).__next__,
         resume=next(
             (w[2] for w in pending_writes if w[0] == task_id and w[1] == RESUME), []
         ),
-        null_resume=next(
-            (w[2] for w in pending_writes if w[0] == NULL_TASK_ID and w[1] == RESUME),
-            MISSING,
-        ),
+        null_resume=null_resume_write[2] if null_resume_write is not None else None,
+        _consume_null_resume=functools.partial(pending_writes.remove, null_resume_write)
+        if null_resume_write is not None
+        else lambda: None,
         # subgraph
-        subgraph_counter=0,
+        subgraph_counter=itertools.count(0).__next__,
     )
 
 

libs/langgraph/langgraph/pregel/call.py

@@ -8,6 +8,7 @@ import sys
 import types
 from typing import Any, Callable, Optional, TypeVar, Union
 
+from langchain_core.runnables import Runnable
 from typing_extensions import ParamSpec
 
 from langgraph.constants import CONF, CONFIG_KEY_CALL, RETURN, TAG_HIDDEN
@@ -144,7 +145,9 @@ def get_runnable_for_entrypoint(func: Callable[..., Any]) -> RunnableSeq:
         return CACHE[key]
     else:
         if is_async_callable(func):
-            run = RunnableCallable(None, func, name=func.__name__, trace=False)
+            run = RunnableCallable(
+                None, func, name=func.__name__, trace=False, recurse=False
+            )
         else:
             afunc = functools.update_wrapper(
                 functools.partial(run_in_executor, None, func), func
@@ -154,15 +157,11 @@ def get_runnable_for_entrypoint(func: Callable[..., Any]) -> RunnableSeq:
                 afunc,
                 name=func.__name__,
                 trace=False,
+                recurse=False,
             )
-        seq = RunnableSeq(
-            run,
-            ChannelWrite([ChannelWriteEntry(RETURN)], tags=[TAG_HIDDEN]),
-            name=func.__name__,
-        )
         if not _lookup_module_and_qualname(func):
-            return seq
-        return CACHE.setdefault(key, seq)
+            return run
+        return CACHE.setdefault(key, run)
 
 
 def get_runnable_for_task(func: Callable[..., Any]) -> RunnableSeq:
@@ -172,7 +171,12 @@ def get_runnable_for_task(func: Callable[..., Any]) -> RunnableSeq:
     else:
         if is_async_callable(func):
             run = RunnableCallable(
-                None, func, explode_args=True, name=func.__name__, trace=False
+                None,
+                func,
+                explode_args=True,
+                name=func.__name__,
+                trace=False,
+                recurse=False,
             )
         else:
             run = RunnableCallable(
@@ -181,6 +185,7 @@ def get_runnable_for_task(func: Callable[..., Any]) -> RunnableSeq:
                 explode_args=True,
                 name=func.__name__,
                 trace=False,
+                recurse=False,
             )
         seq = RunnableSeq(
             run,
@@ -195,7 +200,7 @@ def get_runnable_for_task(func: Callable[..., Any]) -> RunnableSeq:
         return CACHE.setdefault(key, seq)
 
 
-CACHE: dict[tuple[Callable[..., Any], bool], RunnableSeq] = {}
+CACHE: dict[tuple[Callable[..., Any], bool], Runnable] = {}
 
 
 P = ParamSpec("P")

libs/langgraph/langgraph/pregel/io.py

@@ -89,7 +89,7 @@ def map_command(
                 raise TypeError(
                     f"In Command.goto, expected Send/str, got {type(send).__name__}"
                 )
-    if cmd.resume:
+    if cmd.resume is not None:
         if isinstance(cmd.resume, dict) and all(is_task_id(k) for k in cmd.resume):
             for tid, resume in cmd.resume.items():
                 existing: list[Any] = next(

libs/langgraph/langgraph/pregel/loop.py

@@ -54,7 +54,6 @@ from langgraph.constants import (
     ERROR,
     INPUT,
     INTERRUPT,
-    MISSING,
     NS_SEP,
     NULL_TASK_ID,
     PUSH,
@@ -229,20 +228,23 @@ class PregelLoop(LoopProtocol):
         if self.stream is not None and CONFIG_KEY_STREAM in config[CONF]:
             self.stream = DuplexStream(self.stream, config[CONF][CONFIG_KEY_STREAM])
         scratchpad: Optional[PregelScratchpad] = config[CONF].get(CONFIG_KEY_SCRATCHPAD)
-        if not self.config[CONF].get(CONFIG_KEY_DELEGATE) and scratchpad is not None:
-            if scratchpad["subgraph_counter"]:
+        if not self.config[CONF].get(CONFIG_KEY_DELEGATE) and isinstance(
+            scratchpad, PregelScratchpad
+        ):
+            # if count is > 0, append to checkpoint_ns
+            # if count is 0, leave as is
+            if cnt := scratchpad.subgraph_counter():
                 self.config = patch_configurable(
                     self.config,
                     {
                         CONFIG_KEY_CHECKPOINT_NS: NS_SEP.join(
                             (
                                 config[CONF][CONFIG_KEY_CHECKPOINT_NS],
-                                str(scratchpad["subgraph_counter"]),
+                                str(cnt),
                             )
                         )
                     },
                 )
-            scratchpad["subgraph_counter"] += 1
         if not self.is_nested and config[CONF].get(CONFIG_KEY_CHECKPOINT_NS):
             self.config = patch_configurable(
                 self.config,
@@ -563,9 +565,14 @@ class PregelLoop(LoopProtocol):
         )
 
         # take resume value from parent
-        if scratchpad := configurable.get(CONFIG_KEY_SCRATCHPAD):
-            if scratchpad["null_resume"] is not MISSING:
-                self.put_writes(NULL_TASK_ID, [(RESUME, scratchpad["null_resume"])])
+        if scratchpad := cast(
+            Optional[PregelScratchpad], configurable.get(CONFIG_KEY_SCRATCHPAD)
+        ):
+            if (
+                isinstance(scratchpad, PregelScratchpad)
+                and scratchpad.null_resume is not None
+            ):
+                self.put_writes(NULL_TASK_ID, [(RESUME, scratchpad.null_resume)])
         # map command to writes
         if isinstance(self.input, Command):
             if self.input.resume is not None and not self.checkpointer:
@@ -1084,6 +1091,6 @@ class AsyncPregelLoop(PregelLoop, AsyncContextManager):
             return await exit_task
         except asyncio.CancelledError as e:
             # Bubble up the exit task upon cancellation to permit the API
-            # consumer to await it before e.g., re-using the DB connection.
+            # consumer to await it before e.g., reusing the DB connection.
             e.args = (*e.args, exit_task)
             raise

libs/langgraph/langgraph/pregel/runner.py

@@ -39,7 +39,7 @@ from langgraph.errors import GraphBubbleUp, GraphInterrupt
 from langgraph.pregel.algo import Call
 from langgraph.pregel.executor import Submit
 from langgraph.pregel.retry import arun_with_retry, run_with_retry
-from langgraph.types import PregelExecutableTask, RetryPolicy
+from langgraph.types import PregelExecutableTask, PregelScratchpad, RetryPolicy
 from langgraph.utils.future import chain_future
 
 F = TypeVar("F", concurrent.futures.Future, asyncio.Future)
@@ -135,8 +135,7 @@ class PregelRunner:
                 return task.config[CONF][CONFIG_KEY_SEND](writes)
 
             # schedule PUSH tasks, collect futures
-            scratchpad = task.config[CONF][CONFIG_KEY_SCRATCHPAD]
-            scratchpad.setdefault("call_counter", 0)
+            scratchpad: PregelScratchpad = task.config[CONF][CONFIG_KEY_SCRATCHPAD]
             rtn: dict[int, Optional[concurrent.futures.Future]] = {}
             for idx, w in enumerate(writes):
                 # bail if not a PUSH write
@@ -144,9 +143,9 @@ class PregelRunner:
                     continue
                 # schedule the next task, if the callback returns one
                 wcall = calls[idx] if calls else None
-                cnt = scratchpad["call_counter"]
-                scratchpad["call_counter"] += 1
-                if next_task := self.schedule_task(task, cnt, wcall):
+                if next_task := self.schedule_task(
+                    task, scratchpad.call_counter(), wcall
+                ):
                     if fut := next(
                         (
                             f
@@ -324,8 +323,7 @@ class PregelRunner:
                 return task.config[CONF][CONFIG_KEY_SEND](writes)
 
             # schedule PUSH tasks, collect futures
-            scratchpad = task.config[CONF][CONFIG_KEY_SCRATCHPAD]
-            scratchpad.setdefault("call_counter", 0)
+            scratchpad: PregelScratchpad = task.config[CONF][CONFIG_KEY_SCRATCHPAD]
             rtn: dict[int, Optional[asyncio.Future]] = {}
             for idx, w in enumerate(writes):
                 # bail if not a PUSH write
@@ -333,9 +331,9 @@ class PregelRunner:
                     continue
                 # schedule the next task, if the callback returns one
                 wcall = calls[idx] if calls is not None else None
-                cnt = scratchpad["call_counter"]
-                scratchpad["call_counter"] += 1
-                if next_task := self.schedule_task(task, cnt, wcall):
+                if next_task := self.schedule_task(
+                    task, scratchpad.call_counter(), wcall
+                ):
                     # if the parent task was retried,
                     # the next task might already be running
                     if fut := next(

libs/langgraph/langgraph/types.py

@@ -16,10 +16,11 @@ from typing import (
     TypeVar,
     Union,
     cast,
+    get_type_hints,
 )
 
 from langchain_core.runnables import Runnable, RunnableConfig
-from typing_extensions import Self, TypedDict
+from typing_extensions import Self
 
 from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata
 
@@ -289,6 +290,8 @@ class Command(Generic[N], ToolOutputMixin):
             for t in self.update
         ):
             return self.update
+        elif hints := get_type_hints(type(self.update)):
+            return [(k, getattr(self.update, k)) for k in hints]
         elif self.update is not None:
             return [("__root__", self.update)]
         else:
@@ -339,15 +342,25 @@ class LoopProtocol:
         self.stop = stop
 
 
-class PregelScratchpad(TypedDict):
+@dataclasses.dataclass(**{**_DC_KWARGS, "frozen": False})
+class PregelScratchpad:
     # call
-    call_counter: int
+    call_counter: Callable[[], int]
     # interrupt
-    interrupt_counter: int
+    interrupt_counter: Callable[[], int]
     resume: list[Any]
-    null_resume: Any
+    null_resume: Optional[Any]
+    _consume_null_resume: Callable[[], None]
     # subgraph
-    subgraph_counter: int
+    subgraph_counter: Callable[[], int]
+
+    def consume_null_resume(self) -> Any:
+        if self.null_resume is not None:
+            value = self.null_resume
+            self._consume_null_resume()
+            self.null_resume = None
+            return value
+        raise ValueError("No null resume to consume")
 
 
 def interrupt(value: Any) -> Any:
@@ -449,7 +462,6 @@ def interrupt(value: Any) -> Any:
         CONFIG_KEY_CHECKPOINT_NS,
         CONFIG_KEY_SCRATCHPAD,
         CONFIG_KEY_SEND,
-        MISSING,
         NS_SEP,
         RESUME,
     )
@@ -459,19 +471,17 @@ def interrupt(value: Any) -> Any:
     conf = get_config()["configurable"]
     # track interrupt index
     scratchpad: PregelScratchpad = conf[CONFIG_KEY_SCRATCHPAD]
-    scratchpad["interrupt_counter"] += 1
-    idx = scratchpad["interrupt_counter"]
+    idx = scratchpad.interrupt_counter()
     # find previous resume values
-    if scratchpad["resume"]:
-        if idx < len(scratchpad["resume"]):
-            return scratchpad["resume"][idx]
+    if scratchpad.resume:
+        if idx < len(scratchpad.resume):
+            return scratchpad.resume[idx]
     # find current resume value
-    if scratchpad["null_resume"] is not MISSING:
-        assert len(scratchpad["resume"]) == idx, (scratchpad["resume"], idx)
-        v = scratchpad["null_resume"]
-        scratchpad["null_resume"] = MISSING
-        scratchpad["resume"].append(v)
-        conf[CONFIG_KEY_SEND]([(RESUME, scratchpad["resume"])])
+    if scratchpad.null_resume is not None:
+        assert len(scratchpad.resume) == idx, (scratchpad.resume, idx)
+        v = scratchpad.consume_null_resume()
+        scratchpad.resume.append(v)
+        conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)])
         return v
     # no resume value found
     raise GraphInterrupt(

libs/langgraph/langgraph/utils/runnable.py

@@ -36,7 +36,7 @@ from typing_extensions import TypeGuard
 
 from langgraph.constants import (
     CONF,
-    CONFIG_KEY_END,
+    CONFIG_KEY_PREVIOUS,
     CONFIG_KEY_STORE,
     CONFIG_KEY_STREAM_WRITER,
 )
@@ -85,7 +85,7 @@ KWARGS_CONFIG_KEYS: tuple[tuple[str, tuple[Any, ...], str, Any], ...] = (
     (
         sys.intern("previous"),
         (ANY_TYPE,),
-        CONFIG_KEY_END,
+        CONFIG_KEY_PREVIOUS,
         inspect.Parameter.empty,
     ),
 )

libs/langgraph/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph"
-version = "0.2.66"
+version = "0.2.67"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 license = "MIT"

libs/langgraph/tests/test_large_cases.py

@@ -2829,9 +2829,9 @@ def test_state_graph_packets(
     # Define decision-making logic
     def should_continue(data: AgentState) -> str:
         assert isinstance(data["session"], httpx.Client)
-        assert data["something_extra"] == "hi there", (
-            "nodes can pass extra data to their cond edges, which isn't saved in state"
-        )
+        assert (
+            data["something_extra"] == "hi there"
+        ), "nodes can pass extra data to their cond edges, which isn't saved in state"
         # Logic to decide whether to continue in the loop or exit
         if tool_calls := data["messages"][-1].tool_calls:
             return [Send("tools", tool_call) for tool_call in tool_calls]

libs/langgraph/tests/test_prebuilt.py

@@ -346,6 +346,50 @@ def test_state_modifier_with_store():
     assert response["messages"][-1].content == "foo-hi"
 
 
+async def test_state_modifier_with_store_async():
+    async def add(a: int, b: int):
+        """Adds a and b"""
+        return a + b
+
+    in_memory_store = InMemoryStore()
+    await in_memory_store.aput(
+        ("memories", "1"), "user_name", {"data": "User name is Alice"}
+    )
+    await in_memory_store.aput(
+        ("memories", "2"), "user_name", {"data": "User name is Bob"}
+    )
+
+    async def modify(state, config, *, store):
+        user_id = config["configurable"]["user_id"]
+        system_str = (await store.aget(("memories", user_id), "user_name")).value[
+            "data"
+        ]
+        return [SystemMessage(system_str)] + state["messages"]
+
+    async def modify_no_store(state, config):
+        return SystemMessage("foo") + state["messages"]
+
+    model = FakeToolCallingModel()
+
+    # test state modifier that uses store works
+    agent = create_react_agent(
+        model, [add], state_modifier=modify, store=in_memory_store
+    )
+    response = await agent.ainvoke(
+        {"messages": [("user", "hi")]}, {"configurable": {"user_id": "1"}}
+    )
+    assert response["messages"][-1].content == "User name is Alice-hi"
+
+    # test state modifier that doesn't use store works
+    agent = create_react_agent(
+        model, [add], state_modifier=modify_no_store, store=in_memory_store
+    )
+    response = await agent.ainvoke(
+        {"messages": [("user", "hi")]}, {"configurable": {"user_id": "2"}}
+    )
+    assert response["messages"][-1].content == "foo-hi"
+
+
 @pytest.mark.parametrize("tool_style", ["openai", "anthropic"])
 def test_model_with_tools(tool_style: str):
     model = FakeToolCallingModel(tool_style=tool_style)

libs/langgraph/tests/test_pregel.py

@@ -9,6 +9,7 @@ import warnings
 from collections import Counter, deque
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
+from dataclasses import dataclass
 from random import randrange
 from typing import (
     Annotated,
@@ -1539,12 +1540,14 @@ def test_imp_nested(
 
     @task
     def submapper(input: int) -> str:
+        time.sleep(input / 100)
         return str(input)
 
     @task()
     def mapper(input: int) -> str:
+        sub = submapper(input)
         time.sleep(input / 100)
-        return submapper(input).result() * 2
+        return sub.result() * 2
 
     @entrypoint(checkpointer=checkpointer)
     def graph(input: list[int]) -> list[str]:
@@ -5082,11 +5085,27 @@ def test_interrupt_task_functional(
 
     config = {"configurable": {"thread_id": "1"}}
     # First run, interrupted at bar
-    graph.invoke({"a": ""}, config)
+    assert not graph.invoke({"a": ""}, config)
     # Resume with an answer
     res = graph.invoke(Command(resume="bar"), config)
     assert res == {"a": "foobar"}
 
+    # Test that we can interrupt the same task multiple times
+    config = {"configurable": {"thread_id": "2"}}
+
+    @entrypoint(checkpointer=checkpointer)
+    def graph(inputs: dict) -> dict:
+        foo_result = foo(inputs).result()
+        bar_result = bar(foo_result).result()
+        baz_result = bar(bar_result).result()
+        return baz_result
+
+    # First run, interrupted at bar
+    assert not graph.invoke({"a": ""}, config)
+    # Provide resumes
+    assert not graph.invoke(Command(resume="bar"), config)
+    assert graph.invoke(Command(resume="baz"), config) == {"a": "foobarbaz"}
+
 
 def test_root_mixed_return() -> None:
     def my_node(state: list[str]):
@@ -5116,6 +5135,35 @@ def test_dict_mixed_return() -> None:
     assert graph.invoke({"foo": ""}) == {"foo": "ab"}
 
 
+def test_command_pydantic_dataclass() -> None:
+    from pydantic import BaseModel
+
+    class PydanticState(BaseModel):
+        foo: str
+
+    @dataclass
+    class DataclassState:
+        foo: str
+
+    for State in (PydanticState, DataclassState):
+
+        def node_a(state) -> Command[Literal["node_b"]]:
+            return Command(
+                update=State(foo="foo"),
+                goto="node_b",
+            )
+
+        def node_b(state):
+            return {"foo": state.foo + "bar"}
+
+        builder = StateGraph(State)
+        builder.add_edge(START, "node_a")
+        builder.add_node(node_a)
+        builder.add_node(node_b)
+        graph = builder.compile()
+        assert graph.invoke(State(foo="")) == {"foo": "foobar"}
+
+
 @pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
 def test_command_with_static_breakpoints(
     request: pytest.FixtureRequest, checkpointer_name: str
@@ -6092,3 +6140,140 @@ def test_multiple_subgraphs_mixed_checkpointer(
         ),
         ((), {"parent_node": {"parent_counter": 7}}),
     ]
+
+
+def test_entrypoint_output_schema_with_return_and_save() -> None:
+    """Test output schema inference with entrypoint.final."""
+
+    # Un-parameterized entrypoint.final is interpreted as entrypoint.final[Any, Any]
+    @entrypoint()
+    def foo2(inputs, *, previous: Any) -> entrypoint.final:
+        return entrypoint.final(value="foo", save=1)
+
+    assert foo2.get_output_schema().model_json_schema() == {
+        "title": "LangGraphOutput",
+    }
+
+    @entrypoint()
+    def foo(inputs, *, previous: Any) -> entrypoint.final[str, int]:
+        return entrypoint.final(value="foo", save=1)
+
+    assert foo.get_output_schema().model_json_schema() == {
+        "title": "LangGraphOutput",
+        "type": "string",
+    }
+
+    with pytest.raises(TypeError):
+        # Raise an exception on an improperly parameterized entrypoint.final
+        # User is attempting to parameterize in this case, so we'll offer
+        # a bit of help if it's not done correctly.
+        @entrypoint()
+        def foo(inputs, *, previous: Any) -> entrypoint.final[int]:
+            return entrypoint.final(value=1, save=1)  # type: ignore
+
+    @entrypoint()
+    def foo(inputs, *, previous: Any) -> Generator[int, None, None]:
+        yield 1
+
+    assert foo.get_output_schema().model_json_schema() == {
+        "items": {
+            "type": "integer",
+        },
+        "title": "LangGraphOutput",
+        "type": "array",
+    }
+
+
+def test_entrypoint_with_return_and_save() -> None:
+    """Test entrypoint with return and save."""
+    previous_ = None
+
+    @entrypoint(checkpointer=MemorySaver())
+    def foo(msg: str, *, previous: Any) -> entrypoint.final[int, list[str]]:
+        nonlocal previous_
+        previous_ = previous
+        previous = previous or []
+        return entrypoint.final(value=len(previous), save=previous + [msg])
+
+    assert foo.get_output_schema().model_json_schema() == {
+        "title": "LangGraphOutput",
+        "type": "integer",
+    }
+
+    config = {"configurable": {"thread_id": "1"}}
+    assert foo.invoke("hello", config) == 0
+    assert previous_ is None
+    assert foo.invoke("goodbye", config) == 1
+    assert previous_ == ["hello"]
+    assert foo.invoke("definitely", config) == 2
+    assert previous_ == ["hello", "goodbye"]
+
+
+def test_entrypoint_generator_with_return_and_save() -> None:
+    """Verify that generators produce expected results."""
+    previous_ = None
+
+    @entrypoint(checkpointer=MemorySaver())
+    def workflow(inputs: dict, *, previous: Any):
+        nonlocal previous_
+        previous_ = previous
+
+        yield "hello"
+        yield "world"
+        yield entrypoint.final(value="!", save="saved value")
+
+    assert list(workflow.stream({}, {"configurable": {"thread_id": "0"}})) == [
+        "hello",
+        "world",
+    ]
+    assert list(
+        workflow.stream({}, {"configurable": {"thread_id": "0"}}, stream_mode="updates")
+    ) == [
+        {
+            "workflow": "!",
+        }
+    ]
+
+    assert workflow.invoke({}, {"configurable": {"thread_id": "1"}}) == "!"
+    assert previous_ is None
+
+    # 2nd time around previous is set
+    assert workflow.invoke({}, {"configurable": {"thread_id": "1"}}) == "!"
+    assert previous_ == "saved value"
+
+    # Test with another thread
+    assert workflow.invoke({}, {"configurable": {"thread_id": "2"}}) == "!"
+    assert previous_ is None
+
+
+async def test_entrypoint_async_generator_with_return_and_save() -> None:
+    """Verify that generators produce expected results."""
+    previous_ = None
+
+    @entrypoint(checkpointer=MemorySaver())
+    async def workflow(inputs: dict, *, previous: Any):
+        nonlocal previous_
+        previous_ = previous
+
+        yield "hello"
+        yield "world"
+        yield entrypoint.final(value="!", save="saved value")
+
+    assert [
+        c async for c in workflow.astream({}, {"configurable": {"thread_id": "0"}})
+    ] == [
+        "hello",
+        "world",
+    ]
+
+    assert await workflow.ainvoke({}, {"configurable": {"thread_id": "1"}}) == "!"
+
+    assert previous_ is None
+
+    # 2nd time around previous is set
+    assert await workflow.ainvoke({}, {"configurable": {"thread_id": "1"}}) == "!"
+    assert previous_ == "saved value"
+
+    # Test with another thread
+    assert await workflow.ainvoke({}, {"configurable": {"thread_id": "2"}}) == "!"
+    assert previous_ is None

libs/scheduler-kafka/langgraph/scheduler/kafka/orchestrator.py

@@ -13,10 +13,8 @@ from typing_extensions import Self
 
 import langgraph.scheduler.kafka.serde as serde
 from langgraph.constants import (
-    CONF,
     CONFIG_KEY_DEDUPE_TASKS,
     CONFIG_KEY_ENSURE_LATEST,
-    CONFIG_KEY_SCRATCHPAD,
     INTERRUPT,
     SCHEDULED,
 )
@@ -178,8 +176,6 @@ class AsyncKafkaOrchestrator(AbstractAsyncContextManager):
                             CONFIG_KEY_ENSURE_LATEST: True,
                         },
                     )
-                    if CONFIG_KEY_SCRATCHPAD in config[CONF]:
-                        config[CONF][CONFIG_KEY_SCRATCHPAD]["subgraph_counter"] = 0
                     # send messages to executor
                     futures = await asyncio.gather(
                         *(
@@ -366,8 +362,6 @@ class KafkaOrchestrator(AbstractContextManager):
                             CONFIG_KEY_ENSURE_LATEST: True,
                         },
                     )
-                    if CONFIG_KEY_SCRATCHPAD in config[CONF]:
-                        config[CONF][CONFIG_KEY_SCRATCHPAD]["subgraph_counter"] = 0
                     # send messages to executor
                     futures = [
                         self.producer.send(

libs/scheduler-kafka/tests/test_subgraph.py

@@ -15,7 +15,7 @@ from langgraph.graph.state import StateGraph
 from langgraph.pregel import Pregel
 from langgraph.scheduler.kafka import serde
 from langgraph.scheduler.kafka.types import MessageToOrchestrator, Topics
-from tests.any import AnyDict, AnyInt
+from tests.any import AnyDict
 from tests.drain import drain_topics_async
 from tests.messages import _AnyIdAIMessage, _AnyIdHumanMessage
 
@@ -199,9 +199,9 @@ async def test_subgraph_w_interrupt(
                             "__pregel_store": None,
                             "__pregel_task_id": history[0].tasks[0].id,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },
@@ -272,9 +272,9 @@ async def test_subgraph_w_interrupt(
                             "__pregel_store": None,
                             "__pregel_task_id": history[0].tasks[0].id,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },
@@ -375,9 +375,9 @@ async def test_subgraph_w_interrupt(
                             "__pregel_store": None,
                             "__pregel_task_id": history[0].tasks[0].id,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },
@@ -488,9 +488,9 @@ async def test_subgraph_w_interrupt(
                             "__pregel_store": None,
                             "__pregel_task_id": history[1].tasks[0].id,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },
@@ -556,9 +556,9 @@ async def test_subgraph_w_interrupt(
                             "__pregel_store": None,
                             "__pregel_task_id": history[1].tasks[0].id,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },
@@ -680,9 +680,9 @@ async def test_subgraph_w_interrupt(
                             "__pregel_store": None,
                             "__pregel_task_id": history[1].tasks[0].id,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },

libs/scheduler-kafka/tests/test_subgraph_sync.py

@@ -15,7 +15,7 @@ from langgraph.pregel import Pregel
 from langgraph.scheduler.kafka import serde
 from langgraph.scheduler.kafka.default_sync import DefaultProducer
 from langgraph.scheduler.kafka.types import MessageToOrchestrator, Topics
-from tests.any import AnyDict, AnyInt
+from tests.any import AnyDict
 from tests.drain import drain_topics
 from tests.messages import _AnyIdAIMessage, _AnyIdHumanMessage
 
@@ -198,9 +198,9 @@ def test_subgraph_w_interrupt(
                             "__pregel_store": None,
                             "__pregel_task_id": history[0].tasks[0].id,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },
@@ -271,9 +271,9 @@ def test_subgraph_w_interrupt(
                             "__pregel_previous": None,
                             "__pregel_task_id": history[0].tasks[0].id,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },
@@ -374,9 +374,9 @@ def test_subgraph_w_interrupt(
                             "__pregel_task_id": history[0].tasks[0].id,
                             "__pregel_previous": None,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },
@@ -486,9 +486,9 @@ def test_subgraph_w_interrupt(
                             "__pregel_previous": None,
                             "__pregel_task_id": history[1].tasks[0].id,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },
@@ -554,9 +554,9 @@ def test_subgraph_w_interrupt(
                             "__pregel_previous": None,
                             "__pregel_task_id": history[1].tasks[0].id,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },
@@ -678,9 +678,9 @@ def test_subgraph_w_interrupt(
                             "__pregel_store": None,
                             "__pregel_task_id": history[1].tasks[0].id,
                             "__pregel_scratchpad": {
-                                "subgraph_counter": AnyInt(),
-                                "call_counter": 0,
-                                "interrupt_counter": -1,
+                                "subgraph_counter": None,
+                                "call_counter": None,
+                                "interrupt_counter": None,
                                 "null_resume": None,
                                 "resume": [],
                             },