From a6381c32b09079b6e590dcfe83f35cb466046a94 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 23 Jun 2025 18:05:46 -0700 Subject: [PATCH] Revert change to default value of checkpoint_during arg --- libs/langgraph/langgraph/pregel/__init__.py | 4 +- libs/langgraph/tests/test_large_cases.py | 103 ++++++++++++++---- .../langgraph/tests/test_large_cases_async.py | 44 ++++++-- libs/langgraph/tests/test_pregel.py | 16 ++- libs/langgraph/tests/test_pregel_async.py | 40 +++++-- 5 files changed, 159 insertions(+), 48 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index e8fa34ade..2f9ca1425 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -2412,7 +2412,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou debug=debug, checkpoint_during=checkpoint_during if checkpoint_during is not None - else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, False), + else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, True), trigger_to_nodes=self.trigger_to_nodes, migrate_checkpoint=self._migrate_checkpoint, retry_policy=self.retry_policy, @@ -2656,7 +2656,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou debug=debug, checkpoint_during=checkpoint_during if checkpoint_during is not None - else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, False), + else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, True), trigger_to_nodes=self.trigger_to_nodes, migrate_checkpoint=self._migrate_checkpoint, retry_policy=self.retry_policy, diff --git a/libs/langgraph/tests/test_large_cases.py b/libs/langgraph/tests/test_large_cases.py index c50e2d1ed..9fd2f40f9 100644 --- a/libs/langgraph/tests/test_large_cases.py +++ b/libs/langgraph/tests/test_large_cases.py @@ -664,7 +664,10 @@ def test_conditional_state_graph( config = {"configurable": {"thread_id": "1"}} assert [ - c for c in app_w_interrupt.stream({"input": "what is weather in sf"}, config) + c + for c in app_w_interrupt.stream( + {"input": "what is weather in sf"}, config, checkpoint_during=False + ) ] == [ { "agent": { @@ -831,7 +834,10 @@ def test_conditional_state_graph( llm.i = 0 # reset the llm assert [ - c for c in app_w_interrupt.stream({"input": "what is weather in sf"}, config) + c + for c in app_w_interrupt.stream( + {"input": "what is weather in sf"}, config, checkpoint_during=False + ) ] == [ { "agent": { @@ -995,7 +1001,10 @@ def test_conditional_state_graph( llm.i = 0 # reset the llm assert [ - c for c in app_w_interrupt.stream({"input": "what is weather in sf"}, config) + c + for c in app_w_interrupt.stream( + {"input": "what is weather in sf"}, config, checkpoint_during=False + ) ] == [ {"__interrupt__": ()}, ] @@ -1139,7 +1148,10 @@ def test_conditional_state_graph( llm.i = 0 # reset the llm assert [ - c for c in app_w_interrupt.stream({"input": "what is weather in sf"}, config) + c + for c in app_w_interrupt.stream( + {"input": "what is weather in sf"}, config, checkpoint_during=False + ) ] == [ { "agent": { @@ -1837,7 +1849,9 @@ def test_state_graph_packets( assert [ c for c in app_w_interrupt.stream( - {"messages": HumanMessage(content="what is weather in sf")}, config + {"messages": HumanMessage(content="what is weather in sf")}, + config, + checkpoint_during=False, ) ] == [ { @@ -2100,7 +2114,9 @@ def test_state_graph_packets( assert [ c for c in app_w_interrupt.stream( - {"messages": HumanMessage(content="what is weather in sf")}, config + {"messages": HumanMessage(content="what is weather in sf")}, + config, + checkpoint_during=False, ) ] == [ { @@ -2566,7 +2582,10 @@ def test_message_graph( config = {"configurable": {"thread_id": "1"}} assert [ - c for c in app_w_interrupt.stream(("human", "what is weather in sf"), config) + c + for c in app_w_interrupt.stream( + ("human", "what is weather in sf"), config, checkpoint_during=False + ) ] == [ { "agent": AIMessage( @@ -2787,7 +2806,12 @@ def test_message_graph( config = {"configurable": {"thread_id": "2"}} model.i = 0 # reset the llm - assert [c for c in app_w_interrupt.stream("what is weather in sf", config)] == [ + assert [ + c + for c in app_w_interrupt.stream( + "what is weather in sf", config, checkpoint_during=False + ) + ] == [ { "agent": AIMessage( content="", @@ -3280,7 +3304,10 @@ def test_root_graph( config = {"configurable": {"thread_id": "1"}} assert [ - c for c in app_w_interrupt.stream(("human", "what is weather in sf"), config) + c + for c in app_w_interrupt.stream( + ("human", "what is weather in sf"), config, checkpoint_during=False + ) ] == [ { "agent": AIMessage( @@ -3503,7 +3530,12 @@ def test_root_graph( config = {"configurable": {"thread_id": "2"}} model.i = 0 # reset the llm - assert [c for c in app_w_interrupt.stream("what is weather in sf", config)] == [ + assert [ + c + for c in app_w_interrupt.stream( + "what is weather in sf", config, checkpoint_during=False + ) + ] == [ { "agent": AIMessage( content="", @@ -4187,7 +4219,9 @@ def test_dynamic_interrupt(sync_checkpointer: BaseCheckpointSaver) -> None: # flow: interrupt -> clear tasks thread1 = {"configurable": {"thread_id": "1"}} # stop when about to enter node - assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == { + assert tool_two.invoke( + {"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False + ) == { "my_key": "value ⛰️", "market": "DE", "__interrupt__": [ @@ -4352,7 +4386,9 @@ def test_copy_checkpoint(sync_checkpointer: BaseCheckpointSaver) -> None: # flow: interrupt -> clear tasks thread1 = {"configurable": {"thread_id": "1"}} # stop when about to enter node - assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == { + assert tool_two.invoke( + {"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False + ) == { "my_key": "value ⛰️ one", "market": "DE", "__interrupt__": [ @@ -4538,7 +4574,9 @@ def test_dynamic_interrupt_subgraph(sync_checkpointer: BaseCheckpointSaver) -> N # flow: interrupt -> clear tasks thread1 = {"configurable": {"thread_id": "1"}} # stop when about to enter node - assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == { + assert tool_two.invoke( + {"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False + ) == { "my_key": "value ⛰️", "market": "DE", "__interrupt__": [ @@ -5055,7 +5093,7 @@ def test_nested_graph_state(sync_checkpointer: BaseCheckpointSaver) -> None: app = graph.compile(checkpointer=sync_checkpointer) config = {"configurable": {"thread_id": "1"}} - app.invoke({"my_key": "my value"}, config, debug=True) + app.invoke({"my_key": "my value"}, config, checkpoint_during=False) # test state w/ nested subgraph state (right after interrupt) # first get_state without subgraph state expected = StateSnapshot( @@ -5179,7 +5217,7 @@ def test_nested_graph_state(sync_checkpointer: BaseCheckpointSaver) -> None: assert child_history == expected_child_history # resume - app.invoke(None, config, debug=True) + app.invoke(None, config, checkpoint_during=False) # test state w/ nested subgraph state (after resuming from interrupt) assert app.get_state(config) == StateSnapshot( values={"my_key": "hi my value here and there and back again"}, @@ -5332,7 +5370,12 @@ def test_doubly_nested_graph_state( # test invoke w/ nested interrupt config = {"configurable": {"thread_id": "1"}} - assert [c for c in app.stream({"my_key": "my value"}, config, subgraphs=True)] == [ + assert [ + c + for c in app.stream( + {"my_key": "my value"}, config, subgraphs=True, checkpoint_during=False + ) + ] == [ ((), {"parent_1": {"my_key": "hi my value"}}), ( (AnyStr("child:"), AnyStr("child_1:")), @@ -5549,7 +5592,9 @@ def test_doubly_nested_graph_state( interrupts=(), ) # # resume - assert [c for c in app.stream(None, config, subgraphs=True)] == [ + assert [ + c for c in app.stream(None, config, subgraphs=True, checkpoint_during=False) + ] == [ ( (AnyStr("child:"), AnyStr("child_1:")), {"grandchild_2": {"my_key": "hi my value here and there"}}, @@ -5906,7 +5951,9 @@ def test_send_react_interrupt( foo_called = 0 graph = builder.compile(checkpointer=sync_checkpointer, interrupt_before=["foo"]) thread1 = {"configurable": {"thread_id": "2"}} - assert graph.invoke({"messages": [HumanMessage("hello")]}, thread1) == { + assert graph.invoke( + {"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False + ) == { "messages": [ _AnyIdHumanMessage(content="hello"), _AnyIdAIMessage( @@ -6028,7 +6075,9 @@ def test_send_react_interrupt( foo_called = 0 graph = builder.compile(checkpointer=sync_checkpointer, interrupt_before=["foo"]) thread1 = {"configurable": {"thread_id": "3"}} - assert graph.invoke({"messages": [HumanMessage("hello")]}, thread1) == { + assert graph.invoke( + {"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False + ) == { "messages": [ _AnyIdHumanMessage(content="hello"), _AnyIdAIMessage( @@ -6292,7 +6341,9 @@ def test_send_react_interrupt_control( foo_called = 0 graph = builder.compile(checkpointer=sync_checkpointer, interrupt_before=["foo"]) thread1 = {"configurable": {"thread_id": "2"}} - assert graph.invoke({"messages": [HumanMessage("hello")]}, thread1) == { + assert graph.invoke( + {"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False + ) == { "messages": [ _AnyIdHumanMessage(content="hello"), _AnyIdAIMessage( @@ -6546,7 +6597,11 @@ def test_weather_subgraph( assert [ c for c in graph.stream( - inputs, config=config, stream_mode="updates", subgraphs=True + inputs, + config=config, + stream_mode="updates", + subgraphs=True, + checkpoint_during=False, ) ] == [ ((), {"router_node": {"route": "weather"}}), @@ -6629,7 +6684,11 @@ def test_weather_subgraph( assert [ c for c in graph.stream( - inputs, config=config, stream_mode="updates", subgraphs=True + inputs, + config=config, + stream_mode="updates", + subgraphs=True, + checkpoint_during=False, ) ] == [ ((), {"router_node": {"route": "weather"}}), diff --git a/libs/langgraph/tests/test_large_cases_async.py b/libs/langgraph/tests/test_large_cases_async.py index 624ab0f49..824624680 100644 --- a/libs/langgraph/tests/test_large_cases_async.py +++ b/libs/langgraph/tests/test_large_cases_async.py @@ -683,7 +683,7 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver) assert [ c async for c in app_w_interrupt.astream( - {"input": "what is weather in sf"}, config + {"input": "what is weather in sf"}, config, checkpoint_during=False ) ] == [ { @@ -858,7 +858,7 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver) assert [ c async for c in app_w_interrupt.astream( - {"input": "what is weather in sf"}, config + {"input": "what is weather in sf"}, config, checkpoint_during=False ) ] == [ { @@ -1574,7 +1574,9 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N assert [ c async for c in app_w_interrupt.astream( - {"messages": HumanMessage(content="what is weather in sf")}, config + {"messages": HumanMessage(content="what is weather in sf")}, + config, + checkpoint_during=False, ) ] == [ { @@ -1823,7 +1825,9 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N assert [ c async for c in app_w_interrupt.astream( - {"messages": HumanMessage(content="what is weather in sf")}, config + {"messages": HumanMessage(content="what is weather in sf")}, + config, + checkpoint_during=False, ) ] == [ { @@ -2250,7 +2254,9 @@ async def test_message_graph(async_checkpointer: BaseCheckpointSaver) -> None: assert [ c async for c in app_w_interrupt.astream( - HumanMessage(content="what is weather in sf"), config + HumanMessage(content="what is weather in sf"), + config, + checkpoint_during=False, ) ] == [ { @@ -2733,7 +2739,7 @@ async def test_nested_graph_state(async_checkpointer: BaseCheckpointSaver) -> No app = graph.compile(checkpointer=async_checkpointer) config = {"configurable": {"thread_id": "1"}} - await app.ainvoke({"my_key": "my value"}, config, debug=True) + await app.ainvoke({"my_key": "my value"}, config, checkpoint_during=False) # test state w/ nested subgraph state (right after interrupt) # first get_state without subgraph state expected = StateSnapshot( @@ -2864,7 +2870,7 @@ async def test_nested_graph_state(async_checkpointer: BaseCheckpointSaver) -> No assert child_history == expected_child_history # resume - await app.ainvoke(None, config, debug=True) + await app.ainvoke(None, config, checkpoint_during=False) # test state w/ nested subgraph state (after resuming from interrupt) assert await app.aget_state(config) == StateSnapshot( values={"my_key": "hi my value here and there and back again"}, @@ -3021,7 +3027,10 @@ async def test_doubly_nested_graph_state( # test invoke w/ nested interrupt config = {"configurable": {"thread_id": "1"}} assert [ - c async for c in app.astream({"my_key": "my value"}, config, subgraphs=True) + c + async for c in app.astream( + {"my_key": "my value"}, config, subgraphs=True, checkpoint_during=False + ) ] == [ ((), {"parent_1": {"my_key": "hi my value"}}), ( @@ -3239,7 +3248,12 @@ async def test_doubly_nested_graph_state( interrupts=(), ) # resume - assert [c async for c in app.astream(None, config, subgraphs=True)] == [ + assert [ + c + async for c in app.astream( + None, config, subgraphs=True, checkpoint_during=False + ) + ] == [ ( (AnyStr("child:"), AnyStr("child_1:")), {"grandchild_2": {"my_key": "hi my value here and there"}}, @@ -3643,7 +3657,11 @@ async def test_weather_subgraph( assert [ c async for c in graph.astream( - inputs, config=config, stream_mode="updates", subgraphs=True + inputs, + config=config, + stream_mode="updates", + subgraphs=True, + checkpoint_during=False, ) ] == [ ((), {"router_node": {"route": "weather"}}), @@ -3728,7 +3746,11 @@ async def test_weather_subgraph( assert [ c async for c in graph.astream( - inputs, config=config, stream_mode="updates", subgraphs=True + inputs, + config=config, + stream_mode="updates", + subgraphs=True, + checkpoint_during=False, ) ] == [ ((), {"router_node": {"route": "weather"}}), diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 1ecc18466..c5a892ea2 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -4811,7 +4811,9 @@ def test_parent_command( config = {"configurable": {"thread_id": "1"}} - assert graph.invoke({"messages": [("user", "get user name")]}, config) == { + assert graph.invoke( + {"messages": [("user", "get user name")]}, config, checkpoint_during=False + ) == { "messages": [ _AnyIdHumanMessage( content="get user name", additional_kwargs={}, response_metadata={} @@ -5533,7 +5535,10 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver): configurable = {"configurable": {"thread_id": uuid.uuid4()}} assert [ - chunk for chunk in graph.stream({"a": 5}, configurable, stream_mode="debug") + chunk + for chunk in graph.stream( + {"a": 5}, configurable, stream_mode="debug", checkpoint_during=False + ) ] == [ { "payload": { @@ -5635,7 +5640,12 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver): ] assert [ c - for c in graph.stream(Command(resume="123"), configurable, stream_mode="debug") + for c in graph.stream( + Command(resume="123"), + configurable, + stream_mode="debug", + checkpoint_during=False, + ) ] == [ { "payload": { diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index c09c2d1be..b088eb7c5 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -274,7 +274,9 @@ async def test_checkpoint_put_after_cancellation() -> None: thread1 = {"configurable": {"thread_id": "1"}} # start the task - t = asyncio.create_task(graph.ainvoke({"hello": "world"}, thread1)) + t = asyncio.create_task( + graph.ainvoke({"hello": "world"}, thread1, checkpoint_during=False) + ) # cancel after 0.2 seconds await asyncio.sleep(0.2) t.cancel() @@ -340,7 +342,7 @@ async def test_checkpoint_put_after_cancellation_stream_anext() -> None: thread1 = {"configurable": {"thread_id": "1"}} # start the task - s = graph.astream({"hello": "world"}, thread1) + s = graph.astream({"hello": "world"}, thread1, checkpoint_during=False) t = asyncio.create_task(s.__anext__()) # cancel after 0.2 seconds await asyncio.sleep(0.2) @@ -408,7 +410,11 @@ async def test_checkpoint_put_after_cancellation_stream_events_anext() -> None: # start the task s = graph.astream_events( - {"hello": "world"}, thread1, version="v2", include_names=["LangGraph"] + {"hello": "world"}, + thread1, + version="v2", + include_names=["LangGraph"], + checkpoint_during=False, ) # skip first event (happens right away) await s.__anext__() @@ -595,7 +601,9 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non # stop when about to enter node assert [ c - async for c in tool_two.astream({"my_key": "value ⛰️", "market": "DE"}, thread1) + async for c in tool_two.astream( + {"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False + ) ] == [ { "__interrupt__": ( @@ -765,7 +773,9 @@ async def test_dynamic_interrupt_subgraph( # stop when about to enter node assert [ c - async for c in tool_two.astream({"my_key": "value ⛰️", "market": "DE"}, thread1) + async for c in tool_two.astream( + {"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False + ) ] == [ { "__interrupt__": ( @@ -940,7 +950,9 @@ async def test_copy_checkpoint(async_checkpointer: BaseCheckpointSaver) -> None: # flow: interrupt -> clear tasks thread1 = {"configurable": {"thread_id": "1"}} # stop when about to enter node - assert await tool_two.ainvoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == { + assert await tool_two.ainvoke( + {"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False + ) == { "my_key": "value ⛰️ one", "market": "DE", "__interrupt__": [ @@ -2936,7 +2948,9 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) -> foo_called = 0 graph = builder.compile(checkpointer=async_checkpointer, interrupt_before=["foo"]) thread1 = {"configurable": {"thread_id": "2"}} - assert await graph.ainvoke({"messages": [HumanMessage("hello")]}, thread1) == { + assert await graph.ainvoke( + {"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False + ) == { "messages": [ _AnyIdHumanMessage(content="hello"), _AnyIdAIMessage( @@ -3058,7 +3072,9 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) -> foo_called = 0 graph = builder.compile(checkpointer=async_checkpointer, interrupt_before=["foo"]) thread1 = {"configurable": {"thread_id": "3"}} - assert await graph.ainvoke({"messages": [HumanMessage("hello")]}, thread1) == { + assert await graph.ainvoke( + {"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False + ) == { "messages": [ _AnyIdHumanMessage(content="hello"), _AnyIdAIMessage( @@ -3321,7 +3337,9 @@ async def test_send_react_interrupt_control( foo_called = 0 graph = builder.compile(checkpointer=async_checkpointer, interrupt_before=["foo"]) thread1 = {"configurable": {"thread_id": "2"}} - assert await graph.ainvoke({"messages": [HumanMessage("hello")]}, thread1) == { + assert await graph.ainvoke( + {"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False + ) == { "messages": [ _AnyIdHumanMessage(content="hello"), _AnyIdAIMessage( @@ -6077,7 +6095,9 @@ async def test_parent_command( config = {"configurable": {"thread_id": "1"}} - assert await graph.ainvoke({"messages": [("user", "get user name")]}, config) == { + assert await graph.ainvoke( + {"messages": [("user", "get user name")]}, config, checkpoint_during=False + ) == { "messages": [ _AnyIdHumanMessage( content="get user name", additional_kwargs={}, response_metadata={}