feat(langgraph): add durability mode for invoke and ainvoke (#5771)

Fixes https://github.com/langchain-ai/langgraph/issues/5741

Follow up to https://github.com/langchain-ai/langgraph/pull/5432

Plus clean up deprecation logic for `checkpoint_during` and add tests.

---------

Co-authored-by: Eugene Yurtsev <eyurtsev@gmail.com>
Co-authored-by: Lauren Hirata Singh <lauren@langchain.dev>

docs/docs/concepts/durable_execution.md

@@ -51,6 +51,51 @@ For some examples of pitfalls to avoid, see the [Common Pitfalls](./functional_a
 how to structure your code using **tasks** to avoid these issues. The same principles apply to the @[StateGraph (Graph API)][StateGraph].
 :::
 
+## Durability modes
+
+LangGraph supports three durability modes that allow you to balance performance and data consistency based on your application's requirements. The durability modes, from least to most durable, are as follows:
+
+- [`"exit"`](#exit)
+- [`"async"`](#async)
+- [`"sync"`](#sync)
+
+A higher durability mode add more overhead to the workflow execution.
+
+!!! version-added "Added in v0.6.0"
+
+    Use the `durability` parameter instead of `checkpoint_during` (deprecated in v0.6.0) for persistence policy management:
+    
+    * `durability="async"` replaces `checkpoint_during=True`
+    * `durability="exit"` replaces `checkpoint_during=False`
+    
+    for persistence policy management, with the following mapping:
+
+    * `checkpoint_during=True` -> `durability="async"`
+    * `checkpoint_during=False` -> `durability="exit"`
+
+
+### `"exit"`
+Changes are persisted only when graph execution completes (either successfully or with an error). This provides the best performance for long-running graphs but means intermediate state is not saved, so you cannot recover from mid-execution failures or interrupt the graph execution.
+
+### `"async"`
+Changes are persisted asynchronously while the next step executes. This provides good performance and durability, but there's a small risk that checkpoints might not be written if the process crashes during execution.
+
+### `"sync"`
+Changes are persisted synchronously before the next step starts. This ensures that every checkpoint is written before continuing execution, providing high durability at the cost of some performance overhead.
+
+You can specify the durability mode when calling any graph execution method:
+
+:::python
+
+```python
+graph.stream(
+    {"input": "test"}, 
+    durability="sync"
+)
+```
+
+:::
+
 ## Using tasks in nodes
 
 If a [node](./low_level.md#nodes) contains multiple operations, you may find it easier to convert each operation into a **task** rather than refactor the operations into individual nodes.

docs/uv.lock

@@ -2337,7 +2337,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "0.6.1"
+version = "0.6.2"
 source = { editable = "../libs/langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -2641,7 +2641,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "0.6.1"
+version = "0.6.2"
 source = { editable = "../libs/prebuilt" }
 dependencies = [
     { name = "langchain-core" },

libs/langgraph/langgraph/pregel/main.py

@@ -11,7 +11,7 @@ from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
 from dataclasses import is_dataclass
 from functools import partial
 from inspect import isclass
-from typing import Any, Callable, Generic, Optional, Union, cast, get_type_hints
+from typing import Any, Callable, Generic, Union, cast, get_type_hints
 from uuid import UUID, uuid5
 
 from langchain_core.globals import get_debug
@@ -2351,7 +2351,6 @@ class Pregel(
         interrupt_before: All | Sequence[str] | None,
         interrupt_after: All | Sequence[str] | None,
         durability: Durability | None = None,
-        checkpoint_during: bool | None = None,
     ) -> tuple[
         set[StreamMode],
         str | Sequence[str],
@@ -2399,15 +2398,6 @@ class Pregel(
             cache: BaseCache | None = config[CONF][CONFIG_KEY_CACHE]
         else:
             cache = self.cache
-        if checkpoint_during is not None:
-            if durability is not None:
-                raise ValueError(
-                    "Cannot use both `checkpoint_during` and `durability` parameters."
-                )
-            elif checkpoint_during:
-                durability = "async"
-            else:
-                durability = "exit"
         if durability is None:
             durability = config.get(CONF, {}).get(CONFIG_KEY_DURABILITY, "async")
         return (
@@ -2480,6 +2470,17 @@ class Pregel(
         Yields:
             The output of each step in the graph. The output shape depends on the stream_mode.
         """
+        if (checkpoint_during := kwargs.get("checkpoint_during")) is not None:
+            warnings.warn(
+                "`checkpoint_during` is deprecated and will be removed. Please use `durability` instead.",
+                category=LangGraphDeprecatedSinceV10,
+                stacklevel=2,
+            )
+            if durability is not None:
+                raise ValueError(
+                    "Cannot use both `checkpoint_during` and `durability` parameters. Please use `durability` instead."
+                )
+            durability = "async" if checkpoint_during else "exit"
 
         if stream_mode is None:
             # if being called as a node in another graph, default to values mode
@@ -2503,14 +2504,6 @@ class Pregel(
             run_id=config.get("run_id"),
         )
         try:
-            deprecated_checkpoint_during = cast(
-                Optional[bool], kwargs.get("checkpoint_during")
-            )
-            if deprecated_checkpoint_during is not None:
-                warnings.warn(
-                    "`checkpoint_during` is deprecated and will be removed. Please use `durability` instead.",
-                    category=LangGraphDeprecatedSinceV10,
-                )
             # assign defaults
             (
                 stream_modes,
@@ -2529,11 +2522,8 @@ class Pregel(
                 interrupt_before=interrupt_before,
                 interrupt_after=interrupt_after,
                 durability=durability,
-                checkpoint_during=deprecated_checkpoint_during,
             )
-            if checkpointer is None and (
-                durability is not None or deprecated_checkpoint_during is not None
-            ):
+            if checkpointer is None and durability is not None:
                 warnings.warn(
                     "`durability` has no effect when no checkpointer is present.",
                 )
@@ -2570,7 +2560,7 @@ class Pregel(
                     pass
 
             # set durability mode for subgraphs
-            if durability is not None or deprecated_checkpoint_during is not None:
+            if durability is not None:
                 config[CONF][CONFIG_KEY_DURABILITY] = durability_
 
             runtime = Runtime(
@@ -2741,6 +2731,17 @@ class Pregel(
         Yields:
             The output of each step in the graph. The output shape depends on the stream_mode.
         """
+        if (checkpoint_during := kwargs.get("checkpoint_during")) is not None:
+            warnings.warn(
+                "`checkpoint_during` is deprecated and will be removed. Please use `durability` instead.",
+                category=LangGraphDeprecatedSinceV10,
+                stacklevel=2,
+            )
+            if durability is not None:
+                raise ValueError(
+                    "Cannot use both `checkpoint_during` and `durability` parameters. Please use `durability` instead."
+                )
+            durability = "async" if checkpoint_during else "exit"
 
         if stream_mode is None:
             # if being called as a node in another graph, default to values mode
@@ -2783,14 +2784,6 @@ class Pregel(
             else False
         )
         try:
-            deprecated_checkpoint_during = cast(
-                Optional[bool], kwargs.get("checkpoint_during")
-            )
-            if deprecated_checkpoint_during is not None:
-                warnings.warn(
-                    "`checkpoint_during` is deprecated and will be removed. Please use `durability` instead.",
-                    category=LangGraphDeprecatedSinceV10,
-                )
             # assign defaults
             (
                 stream_modes,
@@ -2809,11 +2802,8 @@ class Pregel(
                 interrupt_before=interrupt_before,
                 interrupt_after=interrupt_after,
                 durability=durability,
-                checkpoint_during=deprecated_checkpoint_during,
             )
-            if checkpointer is None and (
-                durability is not None or deprecated_checkpoint_during is not None
-            ):
+            if checkpointer is None and durability is not None:
                 warnings.warn(
                     "`durability` has no effect when no checkpointer is present.",
                 )
@@ -2865,7 +2855,7 @@ class Pregel(
                     pass
 
             # set durability mode for subgraphs
-            if durability is not None or deprecated_checkpoint_during is not None:
+            if durability is not None:
                 config[CONF][CONFIG_KEY_DURABILITY] = durability_
 
             runtime = Runtime(
@@ -2990,6 +2980,7 @@ class Pregel(
         output_keys: str | Sequence[str] | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
+        durability: Durability | None = None,
         **kwargs: Any,
     ) -> dict[str, Any] | Any:
         """Run the graph with a single input and config.
@@ -3004,6 +2995,10 @@ class Pregel(
             output_keys: Optional. The output keys to retrieve from the graph run.
             interrupt_before: Optional. The nodes to interrupt the graph run before.
             interrupt_after: Optional. The nodes to interrupt the graph run after.
+            durability: The durability mode for the graph execution, defaults to "async". Options are:
+                - `"sync"`: Changes are persisted synchronously before the next step starts.
+                - `"async"`: Changes are persisted asynchronously while the next step executes.
+                - `"exit"`: Changes are persisted only when the graph exits.
             **kwargs: Additional keyword arguments to pass to the graph run.
 
         Returns:
@@ -3027,6 +3022,7 @@ class Pregel(
             output_keys=output_keys,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
+            durability=durability,
             **kwargs,
         ):
             if stream_mode == "values":
@@ -3069,6 +3065,7 @@ class Pregel(
         output_keys: str | Sequence[str] | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
+        durability: Durability | None = None,
         **kwargs: Any,
     ) -> dict[str, Any] | Any:
         """Asynchronously invoke the graph on a single input.
@@ -3083,6 +3080,10 @@ class Pregel(
             output_keys: Optional. The output keys to include in the result. Default is None.
             interrupt_before: Optional. The nodes to interrupt before. Default is None.
             interrupt_after: Optional. The nodes to interrupt after. Default is None.
+            durability: The durability mode for the graph execution, defaults to "async". Options are:
+                - `"sync"`: Changes are persisted synchronously before the next step starts.
+                - `"async"`: Changes are persisted asynchronously while the next step executes.
+                - `"exit"`: Changes are persisted only when the graph exits.
             **kwargs: Additional keyword arguments.
 
         Returns:
@@ -3107,6 +3108,7 @@ class Pregel(
             output_keys=output_keys,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
+            durability=durability,
             **kwargs,
         ):
             if stream_mode == "values":

libs/langgraph/tests/test_deprecation.py

@@ -1,6 +1,7 @@
 import pytest
+from langchain_core.runnables import RunnableConfig
 from pytest_mock import MockerFixture
-from typing_extensions import TypedDict
+from typing_extensions import NotRequired, TypedDict
 
 from langgraph.channels.last_value import LastValue
 from langgraph.errors import NodeInterrupt
@@ -136,6 +137,7 @@ def test_config_schema_deprecation_on_entrypoint() -> None:
         assert my_entrypoint.config_schema() is not None
 
 
+@pytest.mark.filterwarnings("ignore:`config_type` is deprecated")
 def test_config_type_deprecation_pregel(mocker: MockerFixture) -> None:
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     chain = NodeBuilder().subscribe_only("input").do(add_one).write_to("output")
@@ -185,3 +187,45 @@ def test_deprecated_import() -> None:
         match="Importing PREVIOUS from langgraph.constants is deprecated. This constant is now private and should not be used directly.",
     ):
         from langgraph.constants import PREVIOUS  # noqa: F401
+
+
+@pytest.mark.filterwarnings("ignore:`checkpoint_during` is deprecated")
+def test_checkpoint_during_deprecation_state_graph() -> None:
+    class CheckDurability(TypedDict):
+        durability: NotRequired[str]
+
+    def plain_node(state: CheckDurability, config: RunnableConfig) -> CheckDurability:
+        return {"durability": config["configurable"]["__pregel_durability"]}
+
+    builder = StateGraph(CheckDurability)
+    builder.add_node("plain_node", plain_node)
+    builder.set_entry_point("plain_node")
+    graph = builder.compile()
+
+    with pytest.warns(
+        LangGraphDeprecatedSinceV10,
+        match="`checkpoint_during` is deprecated and will be removed. Please use `durability` instead.",
+    ):
+        result = graph.invoke({}, checkpoint_during=True)
+        assert result["durability"] == "async"
+
+    with pytest.warns(
+        LangGraphDeprecatedSinceV10,
+        match="`checkpoint_during` is deprecated and will be removed. Please use `durability` instead.",
+    ):
+        result = graph.invoke({}, checkpoint_during=False)
+        assert result["durability"] == "exit"
+
+    with pytest.warns(
+        LangGraphDeprecatedSinceV10,
+        match="`checkpoint_during` is deprecated and will be removed. Please use `durability` instead.",
+    ):
+        for chunk in graph.stream({}, checkpoint_during=True):  # type: ignore[arg-type]
+            assert chunk["plain_node"]["durability"] == "async"
+
+    with pytest.warns(
+        LangGraphDeprecatedSinceV10,
+        match="`checkpoint_during` is deprecated and will be removed. Please use `durability` instead.",
+    ):
+        for chunk in graph.stream({}, checkpoint_during=False):  # type: ignore[arg-type]
+            assert chunk["plain_node"]["durability"] == "exit"