WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-12 12:47:53 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add checkpoint migrations for state graph internal channels (#4125)

Browse Source
This commit is contained in:
Nuno Campos
2025-04-02 15:42:28 -07:00
committed by GitHub
parent 5e0a843423 ffb8400b19
commit 9226b42150
10 changed files with 2072 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/graph/state.py
+115 -1
View File
@@ -2,6 +2,7 @@ import inspect
import logging
import typing
import warnings
from collections import defaultdict
from functools import partial
from inspect import isclass, isfunction, ismethod, signature
from types import FunctionType
@@ -35,7 +36,16 @@ from langgraph.channels.dynamic_barrier_value import DynamicBarrierValue, WaitFo
from langgraph.channels.ephemeral_value import EphemeralValue
from langgraph.channels.last_value import LastValue
from langgraph.channels.named_barrier_value import NamedBarrierValue
from langgraph.constants import EMPTY_SEQ, MISSING, NS_END, NS_SEP, SELF, TAG_HIDDEN
from langgraph.checkpoint.base import Checkpoint
from langgraph.constants import (
EMPTY_SEQ,
INTERRUPT,
MISSING,
NS_END,
NS_SEP,
SELF,
TAG_HIDDEN,
)
from langgraph.errors import (
ErrorCode,
InvalidUpdateError,
@@ -922,6 +932,110 @@ class CompiledStateGraph(CompiledGraph):
)
)
def _migrate_checkpoint(self, checkpoint: Checkpoint) -> None:
"""Migrate a checkpoint to new channel layout."""
values = checkpoint["channel_values"]
versions = checkpoint["channel_versions"]
seen = checkpoint["versions_seen"]
# empty checkpoints do not need migration
if not versions:
return
# current version
if checkpoint["v"] >= 3:
return
# Migrate from start:node to branch:to:node
for k in list(versions):
if k.startswith("start:"):
# confirm node is present
node = k.split(":")[1]
if node not in self.nodes:
continue
# get next version
new_k = f"branch:to:{node}"
new_v = (
max(versions[new_k], versions.pop(k))
if new_k in versions
else versions.pop(k)
)
# update seen
for ss in (seen.get(node, {}), seen.get(INTERRUPT, {})):
if k in ss:
s = ss.pop(k)
if new_k in ss:
ss[new_k] = max(s, ss[new_k])
else:
ss[new_k] = s
# update value
if new_k not in values and k in values:
values[new_k] = values.pop(k)
# update version
versions[new_k] = new_v
# Migrate from branch:source:condition:node to branch:to:node
for k in list(versions):
if k.startswith("branch:") and k.count(":") == 3:
# confirm node is present
node = k.split(":")[-1]
if node not in self.nodes:
continue
# get next version
new_k = f"branch:to:{node}"
new_v = (
max(versions[new_k], versions.pop(k))
if new_k in versions
else versions.pop(k)
)
# update seen
for ss in (seen.get(node, {}), seen.get(INTERRUPT, {})):
if k in ss:
s = ss.pop(k)
if new_k in ss:
ss[new_k] = max(s, ss[new_k])
else:
ss[new_k] = s
# update value
if new_k not in values and k in values:
values[new_k] = values.pop(k)
# update version
versions[new_k] = new_v
if not set(self.nodes).isdisjoint(versions):
# Migrate from "node" to "branch:to:node"
source_to_target = defaultdict(list)
for start, end in self.builder.edges:
if start != START and end != END:
source_to_target[start].append(end)
for k in list(versions):
if k == START:
continue
if k in self.nodes:
v = versions.pop(k)
c = values.pop(k, MISSING)
for end in source_to_target[k]:
# get next version
new_k = f"branch:to:{end}"
new_v = max(versions[new_k], v) if new_k in versions else v
# update seen
for ss in (seen.get(end, {}), seen.get(INTERRUPT, {})):
if k in ss:
s = ss.pop(k)
if new_k in ss:
ss[new_k] = max(s, ss[new_k])
else:
ss[new_k] = s
# update value
if new_k not in values and c is not MISSING:
values[new_k] = c
# update version
versions[new_k] = new_v
# pop interrupt seen
if INTERRUPT in seen:
seen[INTERRUPT].pop(k, MISSING)
def _get_state_reader(
builder: StateGraph, schema: Type[Any]
libs/langgraph/langgraph/pregel/__init__.py
+18 -6
View File
@@ -48,6 +48,7 @@ from langgraph.channels.base import (
)
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
Checkpoint,
CheckpointTuple,
copy_checkpoint,
)
@@ -766,6 +767,10 @@ class Pregel(PregelProtocol):
for name, node in self.get_subgraphs(namespace=namespace, recurse=recurse):
yield name, node
def _migrate_checkpoint(self, checkpoint: Checkpoint) -> None:
"""Migrate a saved checkpoint to new channel layout."""
pass
def _prepare_state_snapshot(
self,
config: RunnableConfig,
@@ -784,6 +789,9 @@ class Pregel(PregelProtocol):
tasks=(),
)
# migrate checkpoint if needed
self._migrate_checkpoint(saved.checkpoint)
with ChannelsManager(
self.channels,
saved.checkpoint,
@@ -897,6 +905,9 @@ class Pregel(PregelProtocol):
tasks=(),
)
# migrate checkpoint if needed
self._migrate_checkpoint(saved.checkpoint)
async with AsyncChannelsManager(
self.channels,
saved.checkpoint,
@@ -1222,6 +1233,8 @@ class Pregel(PregelProtocol):
# get last checkpoint
config = ensure_config(self.config, input_config)
saved = checkpointer.get_tuple(config)
if saved is not None:
self._migrate_checkpoint(saved.checkpoint)
checkpoint = (
copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint()
)
@@ -1632,6 +1645,8 @@ class Pregel(PregelProtocol):
# get last checkpoint
config = ensure_config(self.config, input_config)
saved = await checkpointer.aget_tuple(config)
if saved is not None:
self._migrate_checkpoint(saved.checkpoint)
checkpoint = (
copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint()
)
@@ -2277,6 +2292,7 @@ class Pregel(PregelProtocol):
manager=run_manager,
debug=debug,
trigger_to_nodes=self.trigger_to_nodes,
migrate_checkpoint=self._migrate_checkpoint,
) as loop:
# create runner
runner = PregelRunner(
@@ -2570,12 +2586,8 @@ class Pregel(PregelProtocol):
interrupt_after=interrupt_after_,
manager=run_manager,
debug=debug,
# `self.nodes` can be modified after creation of `Pregel`. For example,
# that's how StateGraph compilation currently works.
# For now, we recompute the trigger_to_nodes mapping every time the
# loop is created. We could potentially memoize this if it becomes a
# performance issue.
trigger_to_nodes=_trigger_to_nodes(self.nodes),
trigger_to_nodes=self.trigger_to_nodes,
migrate_checkpoint=self._migrate_checkpoint,
) as loop:
# create runner
runner = PregelRunner(
libs/langgraph/langgraph/pregel/checkpoint.py
+1 -1
View File
@@ -6,7 +6,7 @@ from langgraph.checkpoint.base import Checkpoint
from langgraph.checkpoint.base.id import uuid6
from langgraph.constants import MISSING
LATEST_VERSION = 2
LATEST_VERSION = 3
def empty_checkpoint() -> Checkpoint:
libs/langgraph/langgraph/pregel/loop.py
+14
View File
@@ -30,6 +30,7 @@ from typing_extensions import ParamSpec, Self
from langgraph.channels.base import BaseChannel
from langgraph.checkpoint.base import (
EXCLUDED_METADATA_KEYS,
WRITES_IDX_MAP,
BaseCheckpointSaver,
ChannelVersions,
@@ -174,6 +175,7 @@ class PregelLoop(LoopProtocol):
Any,
]
]
_migrate_checkpoint: Optional[Callable[[Checkpoint], None]]
submit: Submit
channels: Mapping[str, BaseChannel]
managed: ManagedValueMapping
@@ -211,6 +213,7 @@ class PregelLoop(LoopProtocol):
manager: Union[None, AsyncParentRunManager, ParentRunManager] = None,
input_model: Optional[Type[BaseModel]] = None,
debug: bool = False,
migrate_checkpoint: Optional[Callable[[Checkpoint], None]] = None,
trigger_to_nodes: Optional[Mapping[str, Sequence[str]]] = None,
checkpoint_every_step: bool = True,
) -> None:
@@ -236,6 +239,7 @@ class PregelLoop(LoopProtocol):
CONFIG_KEY_CHECKPOINT_ID not in config[CONF]
or CONFIG_KEY_DEDUPE_TASKS in config[CONF]
)
self._migrate_checkpoint = migrate_checkpoint
self.trigger_to_nodes = trigger_to_nodes
self.checkpoint_every_step = checkpoint_every_step
self.debug = debug
@@ -723,6 +727,8 @@ class PregelLoop(LoopProtocol):
# bail if no checkpointer
if self._checkpointer_put_after_previous is not None:
for k, v in self.config["metadata"].items():
if k in EXCLUDED_METADATA_KEYS:
continue
metadata.setdefault(k, v) # type: ignore
# create new checkpoint
@@ -899,6 +905,7 @@ class SyncPregelLoop(PregelLoop, ContextManager):
stream_keys: Union[str, Sequence[str]] = EMPTY_SEQ,
input_model: Optional[Type[BaseModel]] = None,
debug: bool = False,
migrate_checkpoint: Optional[Callable[[Checkpoint], None]] = None,
trigger_to_nodes: Optional[Mapping[str, Sequence[str]]] = None,
) -> None:
super().__init__(
@@ -916,6 +923,7 @@ class SyncPregelLoop(PregelLoop, ContextManager):
interrupt_before=interrupt_before,
manager=manager,
debug=debug,
migrate_checkpoint=migrate_checkpoint,
trigger_to_nodes=trigger_to_nodes,
)
self.stack = ExitStack()
@@ -984,6 +992,8 @@ class SyncPregelLoop(PregelLoop, ContextManager):
saved = CheckpointTuple(
self.config, empty_checkpoint(), {"step": -2}, None, []
)
elif self._migrate_checkpoint is not None:
self._migrate_checkpoint(saved.checkpoint)
self.checkpoint_config = {
**self.config,
**saved.config,
@@ -1042,6 +1052,7 @@ class AsyncPregelLoop(PregelLoop, AsyncContextManager):
stream_keys: Union[str, Sequence[str]] = EMPTY_SEQ,
input_model: Optional[Type[BaseModel]] = None,
debug: bool = False,
migrate_checkpoint: Optional[Callable[[Checkpoint], None]] = None,
trigger_to_nodes: Optional[Mapping[str, Sequence[str]]] = None,
) -> None:
super().__init__(
@@ -1059,6 +1070,7 @@ class AsyncPregelLoop(PregelLoop, AsyncContextManager):
interrupt_before=interrupt_before,
manager=manager,
debug=debug,
migrate_checkpoint=migrate_checkpoint,
trigger_to_nodes=trigger_to_nodes,
)
self.stack = AsyncExitStack()
@@ -1127,6 +1139,8 @@ class AsyncPregelLoop(PregelLoop, AsyncContextManager):
saved = CheckpointTuple(
self.config, empty_checkpoint(), {"step": -2}, None, []
)
elif self._migrate_checkpoint is not None:
self._migrate_checkpoint(saved.checkpoint)
self.checkpoint_config = {
**self.config,
**saved.config,
libs/langgraph/tests/any_str.py
+5
View File
@@ -4,6 +4,11 @@ from typing import Any, Sequence, Union
from typing_extensions import Self
class AnyObject:
def __eq__(self, value):
return True
class FloatBetween(float):
def __new__(cls, min_value: float, max_value: float) -> Self:
return super().__new__(cls, min_value)
libs/langgraph/tests/test_checkpoint_migration.py
+1807
View File
File diff suppressed because it is too large Load Diff
libs/langgraph/tests/test_large_cases.py
+53 -54
View File
@@ -7782,7 +7782,6 @@ def test_nested_graph_state(
},
"step": 1,
"thread_id": "1",
"checkpoint_ns": AnyStr("inner:"),
"langgraph_node": "inner",
"langgraph_path": [PULL, "inner"],
"langgraph_step": 2,
@@ -7977,7 +7976,6 @@ def test_nested_graph_state(
"step": 1,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("inner:"),
"langgraph_node": "inner",
"langgraph_path": [PULL, "inner"],
"langgraph_step": 2,
@@ -8020,7 +8018,6 @@ def test_nested_graph_state(
"step": 0,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("inner:"),
"langgraph_node": "inner",
"langgraph_path": [PULL, "inner"],
"langgraph_step": 2,
@@ -8069,7 +8066,6 @@ def test_nested_graph_state(
"step": -1,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("inner:"),
"langgraph_node": "inner",
"langgraph_path": [PULL, "inner"],
"langgraph_step": 2,
@@ -8420,51 +8416,65 @@ def test_doubly_nested_graph_state(
),
)
child_state = app.get_state(outer_state.tasks[0].state)
assert (
child_state.tasks[0]
== StateSnapshot(
values={"my_key": "hi my value"},
tasks=(
PregelTask(
AnyStr(),
"child_1",
(PULL, "child_1"),
state={
"configurable": {
"thread_id": "1",
"checkpoint_ns": AnyStr(),
}
},
),
assert child_state == StateSnapshot(
values={"my_key": "hi my value"},
tasks=(
PregelTask(
AnyStr(),
"child_1",
(PULL, "child_1"),
state={
"configurable": {
"thread_id": "1",
"checkpoint_ns": AnyStr(),
}
},
),
next=("child_1",),
config={
),
next=("child_1",),
config={
"configurable": {
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"checkpoint_id": AnyStr(),
"checkpoint_map": AnyDict(
{
"": AnyStr(),
AnyStr("child:"): AnyStr(),
}
),
}
},
metadata={
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child",
"langgraph_path": ["__pregel_pull", "child"],
"langgraph_step": 2,
"langgraph_triggers": ["branch:to:child"],
"parents": {"": AnyStr()},
"source": "loop",
"writes": None,
"step": 0,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=(
None
if "shallow" in checkpointer_name
else {
"configurable": {
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"checkpoint_id": AnyStr(),
"checkpoint_map": AnyDict(
{
"": AnyStr(),
AnyStr("child:"): AnyStr(),
}
),
}
},
metadata={
"parents": {"": AnyStr()},
"source": "loop",
"writes": None,
"step": 0,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=(
None
if "shallow" in checkpointer_name
else {
"configurable": {
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"checkpoint_id": AnyStr(),
}
}
),
).tasks[0]
}
),
)
grandchild_state = app.get_state(child_state.tasks[0].state)
assert grandchild_state == StateSnapshot(
@@ -8502,7 +8512,6 @@ def test_doubly_nested_graph_state(
"writes": {"grandchild_1": {"my_key": "hi my value here"}},
"step": 1,
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [PULL, AnyStr("child_1")],
@@ -8583,7 +8592,6 @@ def test_doubly_nested_graph_state(
},
"step": 1,
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
@@ -8636,7 +8644,6 @@ def test_doubly_nested_graph_state(
"writes": None,
"step": 0,
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child",
"langgraph_path": [PULL, AnyStr("child")],
"langgraph_step": 2,
@@ -8932,7 +8939,6 @@ def test_doubly_nested_graph_state(
"step": 1,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child",
"langgraph_path": [PULL, AnyStr("child")],
"langgraph_step": 2,
@@ -8971,7 +8977,6 @@ def test_doubly_nested_graph_state(
"step": 0,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child",
"langgraph_path": [PULL, AnyStr("child")],
"langgraph_step": 2,
@@ -9023,7 +9028,6 @@ def test_doubly_nested_graph_state(
"step": -1,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child",
"langgraph_path": [PULL, AnyStr("child")],
"langgraph_step": 2,
@@ -9073,7 +9077,6 @@ def test_doubly_nested_graph_state(
}
),
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
@@ -9128,7 +9131,6 @@ def test_doubly_nested_graph_state(
}
),
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
@@ -9190,7 +9192,6 @@ def test_doubly_nested_graph_state(
}
),
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
@@ -9252,7 +9253,6 @@ def test_doubly_nested_graph_state(
}
),
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
@@ -10379,7 +10379,6 @@ def test_weather_subgraph(
"step": 1,
"parents": {"": AnyStr()},
"thread_id": "14",
"checkpoint_ns": AnyStr("weather_graph:"),
"langgraph_node": "weather_graph",
"langgraph_path": [PULL, "weather_graph"],
"langgraph_step": 2,
libs/langgraph/tests/test_large_cases_async.py
+53 -54
View File
@@ -5332,7 +5332,6 @@ async def test_nested_graph_state(checkpointer_name: str) -> None:
},
"step": 1,
"thread_id": "1",
"checkpoint_ns": AnyStr("inner:"),
"langgraph_node": "inner",
"langgraph_path": [PULL, "inner"],
"langgraph_step": 2,
@@ -5529,7 +5528,6 @@ async def test_nested_graph_state(checkpointer_name: str) -> None:
"step": 1,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("inner:"),
"langgraph_node": "inner",
"langgraph_path": [PULL, "inner"],
"langgraph_step": 2,
@@ -5572,7 +5570,6 @@ async def test_nested_graph_state(checkpointer_name: str) -> None:
"step": 0,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("inner:"),
"langgraph_node": "inner",
"langgraph_path": [PULL, "inner"],
"langgraph_step": 2,
@@ -5621,7 +5618,6 @@ async def test_nested_graph_state(checkpointer_name: str) -> None:
"step": -1,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("inner:"),
"langgraph_node": "inner",
"langgraph_path": [PULL, "inner"],
"langgraph_step": 2,
@@ -5976,51 +5972,65 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None:
),
)
child_state = await app.aget_state(outer_state.tasks[0].state)
assert (
child_state.tasks[0]
== StateSnapshot(
values={"my_key": "hi my value"},
tasks=(
PregelTask(
AnyStr(),
"child_1",
(PULL, "child_1"),
state={
"configurable": {
"thread_id": "1",
"checkpoint_ns": AnyStr(),
}
},
),
assert child_state == StateSnapshot(
values={"my_key": "hi my value"},
tasks=(
PregelTask(
AnyStr(),
"child_1",
(PULL, "child_1"),
state={
"configurable": {
"thread_id": "1",
"checkpoint_ns": AnyStr(),
}
},
),
next=("child_1",),
config={
),
next=("child_1",),
config={
"configurable": {
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"checkpoint_id": AnyStr(),
"checkpoint_map": AnyDict(
{
"": AnyStr(),
AnyStr("child:"): AnyStr(),
}
),
}
},
metadata={
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child",
"langgraph_path": ["__pregel_pull", "child"],
"langgraph_step": 2,
"langgraph_triggers": ["branch:to:child"],
"parents": {"": AnyStr()},
"source": "loop",
"writes": None,
"step": 0,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=(
None
if "shallow" in checkpointer_name
else {
"configurable": {
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"checkpoint_id": AnyStr(),
"checkpoint_map": AnyDict(
{
"": AnyStr(),
AnyStr("child:"): AnyStr(),
}
),
}
},
metadata={
"parents": {"": AnyStr()},
"source": "loop",
"writes": None,
"step": 0,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=(
None
if "shallow" in checkpointer_name
else {
"configurable": {
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"checkpoint_id": AnyStr(),
}
}
),
).tasks[0]
}
),
)
grandchild_state = await app.aget_state(child_state.tasks[0].state)
assert grandchild_state == StateSnapshot(
@@ -6058,7 +6068,6 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None:
"writes": {"grandchild_1": {"my_key": "hi my value here"}},
"step": 1,
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [PULL, AnyStr("child_1")],
@@ -6143,7 +6152,6 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None:
},
"step": 1,
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
@@ -6198,7 +6206,6 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None:
"writes": None,
"step": 0,
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child",
"langgraph_path": [PULL, AnyStr("child")],
"langgraph_step": 2,
@@ -6498,7 +6505,6 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None:
"step": 1,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child",
"langgraph_path": [PULL, AnyStr("child")],
"langgraph_step": 2,
@@ -6537,7 +6543,6 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None:
"step": 0,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child",
"langgraph_path": [PULL, AnyStr("child")],
"langgraph_step": 2,
@@ -6589,7 +6594,6 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None:
"step": -1,
"parents": {"": AnyStr()},
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child",
"langgraph_path": [PULL, AnyStr("child")],
"langgraph_step": 2,
@@ -6643,7 +6647,6 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None:
}
),
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
@@ -6700,7 +6703,6 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None:
}
),
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
@@ -6764,7 +6766,6 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None:
}
),
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
@@ -6828,7 +6829,6 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None:
}
),
"thread_id": "1",
"checkpoint_ns": AnyStr("child:"),
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
@@ -7244,7 +7244,6 @@ async def test_weather_subgraph(
"step": 1,
"parents": {"": AnyStr()},
"thread_id": "14",
"checkpoint_ns": AnyStr("weather_graph:"),
"langgraph_node": "weather_graph",
"langgraph_path": [PULL, "weather_graph"],
"langgraph_step": 2,
libs/langgraph/tests/test_pregel.py
+3 -3
View File
@@ -1234,7 +1234,7 @@ def test_pending_writes_resume(
}
},
checkpoint={
"v": 2,
"v": 3,
"id": AnyStr(),
"ts": AnyStr(),
"pending_sends": [],
@@ -1292,7 +1292,7 @@ def test_pending_writes_resume(
}
},
checkpoint={
"v": 2,
"v": 3,
"id": AnyStr(),
"ts": AnyStr(),
"pending_sends": [],
@@ -1343,7 +1343,7 @@ def test_pending_writes_resume(
}
},
checkpoint={
"v": 2,
"v": 3,
"id": AnyStr(),
"ts": AnyStr(),
"pending_sends": [],
libs/langgraph/tests/test_pregel_async.py
+3 -3
View File
@@ -2068,7 +2068,7 @@ async def test_pending_writes_resume(
}
},
checkpoint={
"v": 2,
"v": 3,
"id": AnyStr(),
"ts": AnyStr(),
"pending_sends": [],
@@ -2128,7 +2128,7 @@ async def test_pending_writes_resume(
}
},
checkpoint={
"v": 2,
"v": 3,
"id": AnyStr(),
"ts": AnyStr(),
"pending_sends": [],
@@ -2181,7 +2181,7 @@ async def test_pending_writes_resume(
}
},
checkpoint={
"v": 2,
"v": 3,
"id": AnyStr(),
"ts": AnyStr(),
"pending_sends": [],