diff --git a/docs/docs/concepts/durable_execution.md b/docs/docs/concepts/durable_execution.md index f49b989fe..6e038b302 100644 --- a/docs/docs/concepts/durable_execution.md +++ b/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. diff --git a/docs/uv.lock b/docs/uv.lock index 95ae39d0e..171f9df70 100644 --- a/docs/uv.lock +++ b/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" }, diff --git a/libs/langgraph/langgraph/pregel/main.py b/libs/langgraph/langgraph/pregel/main.py index 2a65122ed..87f64bebc 100644 --- a/libs/langgraph/langgraph/pregel/main.py +++ b/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": diff --git a/libs/langgraph/tests/test_deprecation.py b/libs/langgraph/tests/test_deprecation.py index 5c43ef1db..01ab10d2e 100644 --- a/libs/langgraph/tests/test_deprecation.py +++ b/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"