From 2e656d9145191c43627b8b91383cc093ede309cd Mon Sep 17 00:00:00 2001 From: Vadym Barda Date: Mon, 4 Nov 2024 14:26:34 -0500 Subject: [PATCH] langgraph: add config metadata to pregel loop (#2323) --- libs/langgraph/langgraph/pregel/__init__.py | 12 +- libs/langgraph/langgraph/pregel/loop.py | 2 + libs/langgraph/tests/test_prebuilt.py | 2 + libs/langgraph/tests/test_pregel.py | 392 ++++++++++++++++++-- libs/langgraph/tests/test_pregel_async.py | 338 +++++++++++++++-- libs/langgraph/tests/test_remote_graph.py | 8 +- 6 files changed, 695 insertions(+), 59 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index e6a5638d2..2525081d1 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -814,7 +814,7 @@ def update_state( raise ValueError(f"Subgraph {recast_checkpoint_ns} not found") # get last checkpoint - config = merge_configs(self.config, config) if self.config else config + config = ensure_config(self.config, config) saved = checkpointer.get_tuple(config) checkpoint = copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint() checkpoint_previous_versions = ( @@ -826,14 +826,17 @@ def update_state( config, {CONFIG_KEY_CHECKPOINT_NS: config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, "")}, ) + checkpoint_metadata = config["metadata"] if saved: checkpoint_config = patch_configurable(config, saved.config[CONF]) + checkpoint_metadata = {**saved.metadata, **checkpoint_metadata} # find last node that updated the state, if not provided if values is None and as_node is None: next_config = checkpointer.put( checkpoint_config, create_checkpoint(checkpoint, None, step), { + **checkpoint_metadata, "source": "update", "step": step + 1, "writes": {}, @@ -922,6 +925,7 @@ def update_state( checkpoint_config, checkpoint, { + **checkpoint_metadata, "source": "update", "step": step + 1, "writes": {as_node: values}, @@ -966,7 +970,7 @@ async def aupdate_state( raise ValueError(f"Subgraph {recast_checkpoint_ns} not found") # get last checkpoint - config = merge_configs(self.config, config) if self.config else config + config = ensure_config(self.config, config) saved = await checkpointer.aget_tuple(config) checkpoint = copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint() checkpoint_previous_versions = ( @@ -978,14 +982,17 @@ async def aupdate_state( config, {CONFIG_KEY_CHECKPOINT_NS: config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, "")}, ) + checkpoint_metadata = config["metadata"] if saved: checkpoint_config = patch_configurable(config, saved.config[CONF]) + checkpoint_metadata = {**saved.metadata, **checkpoint_metadata} # find last node that updated the state, if not provided if values is None and as_node is None: next_config = await checkpointer.aput( checkpoint_config, create_checkpoint(checkpoint, None, step), { + **checkpoint_metadata, "source": "update", "step": step + 1, "writes": {}, @@ -1072,6 +1079,7 @@ async def aupdate_state( checkpoint_config, checkpoint, { + **checkpoint_metadata, "source": "update", "step": step + 1, "writes": {as_node: values}, diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index 2ba8af3a6..91c513ede 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -491,6 +491,8 @@ def _first(self, *, input_keys: Union[str, Sequence[str]]) -> None: ) def _put_checkpoint(self, metadata: CheckpointMetadata) -> None: + for k, v in self.config["metadata"].items(): + metadata.setdefault(k, v) # type: ignore # assign step and parents metadata["step"] = self.step metadata["parents"] = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {}) diff --git a/libs/langgraph/tests/test_prebuilt.py b/libs/langgraph/tests/test_prebuilt.py index 1ea506381..a6655a451 100644 --- a/libs/langgraph/tests/test_prebuilt.py +++ b/libs/langgraph/tests/test_prebuilt.py @@ -158,6 +158,7 @@ def test_no_modifier(request: pytest.FixtureRequest, checkpointer_name: str) -> "source": "loop", "writes": {"agent": {"messages": [AIMessage(content="hi?", id="0")]}}, "step": 1, + "thread_id": "123", } assert saved.pending_writes == [] @@ -189,6 +190,7 @@ async def test_no_modifier_async(checkpointer_name: str) -> None: "source": "loop", "writes": {"agent": {"messages": [AIMessage(content="hi?", id="0")]}}, "step": 1, + "thread_id": "123", } assert saved.pending_writes == [] diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index f8a5b49a8..563ce295e 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -748,7 +748,13 @@ def test_invoke_two_processes_in_out_interrupt( "checkpoint_id": AnyStr(), } }, - metadata={"parents": {}, "source": "loop", "step": 6, "writes": {"two": 5}}, + metadata={ + "parents": {}, + "source": "loop", + "step": 6, + "writes": {"two": 5}, + "thread_id": "1", + }, created_at=AnyStr(), parent_config=history[1].config, ), @@ -768,6 +774,7 @@ def test_invoke_two_processes_in_out_interrupt( "source": "loop", "step": 5, "writes": {"one": None}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[2].config, @@ -788,6 +795,7 @@ def test_invoke_two_processes_in_out_interrupt( "source": "input", "step": 4, "writes": {"input": 3}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[3].config, @@ -808,6 +816,7 @@ def test_invoke_two_processes_in_out_interrupt( "source": "loop", "step": 3, "writes": {"one": None}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[4].config, @@ -828,6 +837,7 @@ def test_invoke_two_processes_in_out_interrupt( "source": "input", "step": 2, "writes": {"input": 20}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[5].config, @@ -843,7 +853,13 @@ def test_invoke_two_processes_in_out_interrupt( "checkpoint_id": AnyStr(), } }, - metadata={"parents": {}, "source": "loop", "step": 1, "writes": {"two": 4}}, + metadata={ + "parents": {}, + "source": "loop", + "step": 1, + "writes": {"two": 4}, + "thread_id": "1", + }, created_at=AnyStr(), parent_config=history[6].config, ), @@ -863,6 +879,7 @@ def test_invoke_two_processes_in_out_interrupt( "source": "loop", "step": 0, "writes": {"one": None}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[7].config, @@ -883,6 +900,7 @@ def test_invoke_two_processes_in_out_interrupt( "source": "input", "step": -1, "writes": {"input": 2}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -949,6 +967,7 @@ def test_fork_always_re_runs_nodes( "source": "loop", "step": 5, "writes": {"add_one": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[1].config, @@ -969,6 +988,7 @@ def test_fork_always_re_runs_nodes( "source": "loop", "step": 4, "writes": {"add_one": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[2].config, @@ -989,6 +1009,7 @@ def test_fork_always_re_runs_nodes( "source": "loop", "step": 3, "writes": {"add_one": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[3].config, @@ -1009,6 +1030,7 @@ def test_fork_always_re_runs_nodes( "source": "loop", "step": 2, "writes": {"add_one": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[4].config, @@ -1029,6 +1051,7 @@ def test_fork_always_re_runs_nodes( "source": "loop", "step": 1, "writes": {"add_one": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[5].config, @@ -1044,7 +1067,13 @@ def test_fork_always_re_runs_nodes( "checkpoint_id": AnyStr(), } }, - metadata={"parents": {}, "source": "loop", "step": 0, "writes": None}, + metadata={ + "parents": {}, + "source": "loop", + "step": 0, + "writes": None, + "thread_id": "1", + }, created_at=AnyStr(), parent_config=history[6].config, ), @@ -1064,6 +1093,7 @@ def test_fork_always_re_runs_nodes( "source": "input", "step": -1, "writes": {"__start__": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -1496,6 +1526,7 @@ def reset(self): "source": "loop", "step": 0, "writes": None, + "thread_id": "1", } # should contain pending write of "one" checkpoint = checkpointer.get_tuple(thread1) @@ -1584,6 +1615,7 @@ def reset(self): "step": 1, "source": "loop", "writes": {"one": {"value": 2}, "two": {"value": 3}}, + "thread_id": "1", }, parent_config={ "configurable": { @@ -1628,7 +1660,13 @@ def reset(self): "start:two": "__start__", }, }, - metadata={"parents": {}, "step": 0, "source": "loop", "writes": None}, + metadata={ + "parents": {}, + "step": 0, + "source": "loop", + "writes": None, + "thread_id": "1", + }, parent_config={ "configurable": { "thread_id": "1", @@ -1668,6 +1706,7 @@ def reset(self): "step": -1, "source": "input", "writes": {"__start__": {"value": 1}}, + "thread_id": "1", }, parent_config=None, pending_writes=UnsortedSequence( @@ -2117,7 +2156,9 @@ def should_continue(data: dict) -> str: workflow.add_node("agent", agent) workflow.add_node( - "tools", execute_tools, metadata={"parents": {}, "version": 2, "variant": "b"} + "tools", + execute_tools, + metadata={"parents": {}, "version": 2, "variant": "b"}, ) workflow.set_entry_point("agent") @@ -2311,6 +2352,7 @@ def should_continue(data: dict) -> str: } }, }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -2362,6 +2404,7 @@ def should_continue(data: dict) -> str: "input": "what is weather in sf", }, }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -2482,6 +2525,7 @@ def should_continue(data: dict) -> str: ), } }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -2537,6 +2581,7 @@ def should_continue(data: dict) -> str: } } }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -2582,6 +2627,7 @@ def should_continue(data: dict) -> str: "input": "what is weather in sf", } }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -2702,6 +2748,7 @@ def should_continue(data: dict) -> str: ), } }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -2757,6 +2804,7 @@ def should_continue(data: dict) -> str: } } }, + "thread_id": "3", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -3230,6 +3278,7 @@ def should_continue(data: AgentState) -> str: ), } }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -3272,6 +3321,7 @@ def should_continue(data: AgentState) -> str: ) }, }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -3348,6 +3398,7 @@ def should_continue(data: AgentState) -> str: ) } }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -3399,6 +3450,7 @@ def should_continue(data: AgentState) -> str: ), } }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -3440,6 +3492,7 @@ def should_continue(data: AgentState) -> str: ) } }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -3514,6 +3567,7 @@ def should_continue(data: AgentState) -> str: ) } }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -3541,7 +3595,13 @@ def should_continue(data: AgentState) -> str: next=("agent",), config=app_w_interrupt.checkpointer.get_tuple(config).config, created_at=app_w_interrupt.checkpointer.get_tuple(config).checkpoint["ts"], - metadata={"parents": {}, "source": "loop", "step": 0, "writes": None}, + metadata={ + "parents": {}, + "source": "loop", + "step": 0, + "writes": None, + "thread_id": "3", + }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -3580,6 +3640,7 @@ def should_continue(data: AgentState) -> str: ), } }, + "thread_id": "3", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -3640,6 +3701,7 @@ def should_continue(data: AgentState) -> str: ], } }, + "thread_id": "3", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -3702,6 +3764,7 @@ def should_continue(data: AgentState) -> str: ), } }, + "thread_id": "4", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -3762,6 +3825,7 @@ def should_continue(data: AgentState) -> str: ], } }, + "thread_id": "4", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -4103,7 +4167,7 @@ def search_api(query: str) -> str: "langgraph_step": 1, "langgraph_node": "agent", "langgraph_triggers": ["start:agent"], - "langgraph_path": ("__pregel_pull", "agent"), + "langgraph_path": (PULL, "agent"), "langgraph_checkpoint_ns": AnyStr("agent:"), "checkpoint_ns": AnyStr("agent:"), "ls_provider": "fakechatmodel", @@ -4120,7 +4184,7 @@ def search_api(query: str) -> str: "langgraph_step": 2, "langgraph_node": "tools", "langgraph_triggers": ["branch:agent:should_continue:tools"], - "langgraph_path": ("__pregel_pull", "tools"), + "langgraph_path": (PULL, "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, ), @@ -4162,7 +4226,7 @@ def search_api(query: str) -> str: "langgraph_step": 3, "langgraph_node": "agent", "langgraph_triggers": ["tools"], - "langgraph_path": ("__pregel_pull", "agent"), + "langgraph_path": (PULL, "agent"), "langgraph_checkpoint_ns": AnyStr("agent:"), "checkpoint_ns": AnyStr("agent:"), "ls_provider": "fakechatmodel", @@ -4179,7 +4243,7 @@ def search_api(query: str) -> str: "langgraph_step": 4, "langgraph_node": "tools", "langgraph_triggers": ["branch:agent:should_continue:tools"], - "langgraph_path": ("__pregel_pull", "tools"), + "langgraph_path": (PULL, "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, ), @@ -4193,7 +4257,7 @@ def search_api(query: str) -> str: "langgraph_step": 4, "langgraph_node": "tools", "langgraph_triggers": ["branch:agent:should_continue:tools"], - "langgraph_path": ("__pregel_pull", "tools"), + "langgraph_path": (PULL, "tools"), "langgraph_checkpoint_ns": AnyStr("tools:"), }, ), @@ -4205,7 +4269,7 @@ def search_api(query: str) -> str: "langgraph_step": 5, "langgraph_node": "agent", "langgraph_triggers": ["tools"], - "langgraph_path": ("__pregel_pull", "agent"), + "langgraph_path": (PULL, "agent"), "langgraph_checkpoint_ns": AnyStr("agent:"), "checkpoint_ns": AnyStr("agent:"), "ls_provider": "fakechatmodel", @@ -4678,6 +4742,7 @@ def tools_node(input: ToolCall, config: RunnableConfig) -> AgentState: ) } }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -4731,6 +4796,7 @@ def tools_node(input: ToolCall, config: RunnableConfig) -> AgentState: "something_extra": "hi there", } }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -4837,6 +4903,7 @@ def tools_node(input: ToolCall, config: RunnableConfig) -> AgentState: ) }, }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -4887,6 +4954,7 @@ def tools_node(input: ToolCall, config: RunnableConfig) -> AgentState: "something_extra": "hi there", } }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -5176,6 +5244,7 @@ def should_continue(messages): id="ai1", ) }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -5222,6 +5291,7 @@ def should_continue(messages): id="ai1", ) }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -5304,6 +5374,7 @@ def should_continue(messages): id="ai2", ) }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -5344,6 +5415,7 @@ def should_continue(messages): "source": "update", "step": 5, "writes": {"agent": AIMessage(content="answer", id="ai2")}, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -5408,6 +5480,7 @@ def should_continue(messages): id="ai1", ) }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -5454,6 +5527,7 @@ def should_continue(messages): id="ai1", ) }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -5536,6 +5610,7 @@ def should_continue(messages): id="ai2", ) }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -5577,6 +5652,7 @@ def should_continue(messages): "source": "update", "step": 5, "writes": {"agent": AIMessage(content="answer", id="ai2")}, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -5618,6 +5694,7 @@ def should_continue(messages): "source": "update", "step": 6, "writes": {"tools": UnsortedSequence("ai", "an extra message")}, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -5906,6 +5983,7 @@ class State(TypedDict): id="ai1", ) }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -5952,6 +6030,7 @@ class State(TypedDict): id="ai1", ) }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -6035,6 +6114,7 @@ class State(TypedDict): id="ai2", ) }, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -6076,6 +6156,7 @@ class State(TypedDict): "source": "update", "step": 5, "writes": {"agent": AIMessage(content="answer", id="ai2")}, + "thread_id": "1", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -6140,6 +6221,7 @@ class State(TypedDict): id="ai1", ) }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -6186,6 +6268,7 @@ class State(TypedDict): id="ai1", ) }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -6269,6 +6352,7 @@ class State(TypedDict): id="ai2", ) }, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -6309,6 +6393,7 @@ class State(TypedDict): "source": "update", "step": 5, "writes": {"agent": AIMessage(content="answer", id="ai2")}, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -6350,6 +6435,7 @@ class State(TypedDict): "source": "update", "step": 6, "writes": {"tools": UnsortedSequence("ai", "an extra message")}, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -6422,6 +6508,7 @@ class MoreState(TypedDict): "source": "update", "step": 6, "writes": {"tools": UnsortedSequence("ai", "an extra message")}, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -6751,12 +6838,14 @@ def tool_two_node(s: State) -> State: "source": "loop", "step": 0, "writes": None, + "thread_id": "1", }, { "parents": {}, "source": "input", "step": -1, "writes": {"__start__": {"my_key": "value ⛰️", "market": "DE"}}, + "thread_id": "1", }, ] assert tool_two.get_state(thread1) == StateSnapshot( @@ -6772,7 +6861,13 @@ def tool_two_node(s: State) -> State: ), config=tool_two.checkpointer.get_tuple(thread1).config, created_at=tool_two.checkpointer.get_tuple(thread1).checkpoint["ts"], - metadata={"parents": {}, "source": "loop", "step": 0, "writes": None}, + metadata={ + "parents": {}, + "source": "loop", + "step": 0, + "writes": None, + "thread_id": "1", + }, parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config, ) @@ -6849,12 +6944,16 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: "source": "loop", "step": 0, "writes": None, + "assistant_id": "a", + "thread_id": "1", }, { "parents": {}, "source": "input", "step": -1, "writes": {"__start__": {"my_key": "value ⛰️", "market": "DE"}}, + "assistant_id": "a", + "thread_id": "1", }, ] assert tool_two.get_state(thread1) == StateSnapshot( @@ -6863,7 +6962,14 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: next=("tool_two_slow",), config=tool_two.checkpointer.get_tuple(thread1).config, created_at=tool_two.checkpointer.get_tuple(thread1).checkpoint["ts"], - metadata={"parents": {}, "source": "loop", "step": 0, "writes": None}, + metadata={ + "parents": {}, + "source": "loop", + "step": 0, + "writes": None, + "assistant_id": "a", + "thread_id": "1", + }, parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config, ) # resume, for same result as above @@ -6882,6 +6988,8 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: "source": "loop", "step": 1, "writes": {"tool_two_slow": {"my_key": " slow"}}, + "assistant_id": "a", + "thread_id": "1", }, parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config, ) @@ -6898,7 +7006,14 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: next=("tool_two_fast",), config=tool_two.checkpointer.get_tuple(thread2).config, created_at=tool_two.checkpointer.get_tuple(thread2).checkpoint["ts"], - metadata={"parents": {}, "source": "loop", "step": 0, "writes": None}, + metadata={ + "parents": {}, + "source": "loop", + "step": 0, + "writes": None, + "assistant_id": "a", + "thread_id": "2", + }, parent_config=[*tool_two.checkpointer.list(thread2, limit=2)][-1].config, ) # resume, for same result as above @@ -6917,6 +7032,8 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: "source": "loop", "step": 1, "writes": {"tool_two_fast": {"my_key": " fast"}}, + "assistant_id": "a", + "thread_id": "2", }, parent_config=[*tool_two.checkpointer.list(thread2, limit=2)][-1].config, ) @@ -6933,7 +7050,14 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: next=("tool_two_fast",), config=tool_two.checkpointer.get_tuple(thread3).config, created_at=tool_two.checkpointer.get_tuple(thread3).checkpoint["ts"], - metadata={"parents": {}, "source": "loop", "step": 0, "writes": None}, + metadata={ + "parents": {}, + "source": "loop", + "step": 0, + "writes": None, + "assistant_id": "b", + "thread_id": "3", + }, parent_config=[*tool_two.checkpointer.list(thread3, limit=2)][-1].config, ) # update state @@ -6949,6 +7073,8 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: "source": "update", "step": 1, "writes": {START: {"my_key": "key"}}, + "assistant_id": "b", + "thread_id": "3", }, parent_config=[*tool_two.checkpointer.list(thread3, limit=2)][-1].config, ) @@ -6968,6 +7094,8 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: "source": "loop", "step": 2, "writes": {"tool_two_fast": {"my_key": " fast"}}, + "assistant_id": "b", + "thread_id": "3", }, parent_config=[*tool_two.checkpointer.list(thread3, limit=2)][-1].config, ) @@ -7041,6 +7169,7 @@ class State(TypedDict): "source": "input", "step": -1, "writes": {"__start__": {"my_key": "value", "market": "DE"}}, + "thread_id": "10", }, "parent_config": None, "next": ["__start__"], @@ -7079,6 +7208,7 @@ class State(TypedDict): "source": "loop", "step": 0, "writes": None, + "thread_id": "10", }, "parent_config": { "tags": [], @@ -7145,6 +7275,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "10", }, "parent_config": { "tags": [], @@ -7216,6 +7347,7 @@ class State(TypedDict): "source": "loop", "step": 2, "writes": {"tool_two_slow": {"my_key": " slow"}}, + "thread_id": "10", }, "parent_config": { "tags": [], @@ -7282,6 +7414,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "10", }, "parent_config": { "tags": [], @@ -7325,6 +7458,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "1", }, parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config, ) @@ -7344,6 +7478,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "1", }, parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config, ) @@ -7365,6 +7500,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "2", }, parent_config=[*tool_two.checkpointer.list(thread2, limit=2)][-1].config, ) @@ -7384,6 +7520,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "2", }, parent_config=[*tool_two.checkpointer.list(thread2, limit=2)][-1].config, ) @@ -7413,6 +7550,7 @@ class State(TypedDict): "source": "loop", "step": 2, "writes": {"tool_two_slow": {"my_key": " slow"}}, + "thread_id": "11", }, parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config, ) @@ -7433,6 +7571,7 @@ class State(TypedDict): "source": "update", "step": 3, "writes": {"tool_two_slow": {"my_key": "er"}}, + "thread_id": "11", }, parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config, ) @@ -7462,6 +7601,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "21", }, parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config, ) @@ -7481,6 +7621,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "21", }, parent_config=[*tool_two.checkpointer.list(thread1, limit=2)][-1].config, ) @@ -7502,6 +7643,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "22", }, parent_config=[*tool_two.checkpointer.list(thread2, limit=2)][-1].config, ) @@ -7521,6 +7663,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "22", }, parent_config=[*tool_two.checkpointer.list(thread2, limit=2)][-1].config, ) @@ -7540,6 +7683,7 @@ class State(TypedDict): "source": "update", "step": 0, "writes": {START: {"my_key": "key", "market": "DE"}}, + "thread_id": "23", }, parent_config=None, ) @@ -7560,6 +7704,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "23", }, parent_config=uconfig, ) @@ -7579,6 +7724,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "23", }, parent_config=[*tool_two.checkpointer.list(thread3, limit=2)][-1].config, ) @@ -7706,6 +7852,7 @@ def qa(data: State) -> State: "source": "update", "step": 4, "writes": {"retriever_one": {"docs": ["doc5"]}}, + "thread_id": "2", }, parent_config=[*app_w_interrupt.checkpointer.list(config, limit=2)][-1].config, ) @@ -9032,6 +9179,7 @@ def outer_2(state: State): "source": "loop", "writes": {"outer_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9085,6 +9233,13 @@ def outer_2(state: State): } }, "step": 1, + "thread_id": "1", + "checkpoint_ns": AnyStr("inner:"), + "langgraph_node": "inner", + "langgraph_path": [PULL, "inner"], + "langgraph_step": 2, + "langgraph_triggers": ["outer_1"], + "langgraph_checkpoint_ns": AnyStr("inner:"), }, created_at=AnyStr(), parent_config={ @@ -9113,6 +9268,7 @@ def outer_2(state: State): "source": "loop", "writes": {"outer_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9154,6 +9310,7 @@ def outer_2(state: State): "source": "loop", "writes": {"outer_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9182,7 +9339,13 @@ def outer_2(state: State): "checkpoint_id": AnyStr(), } }, - metadata={"parents": {}, "source": "loop", "writes": None, "step": 0}, + metadata={ + "parents": {}, + "source": "loop", + "writes": None, + "step": 0, + "thread_id": "1", + }, created_at=AnyStr(), parent_config={ "configurable": { @@ -9215,6 +9378,7 @@ def outer_2(state: State): "source": "input", "writes": {"__start__": {"my_key": "my value"}}, "step": -1, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -9246,6 +9410,13 @@ def outer_2(state: State): }, "step": 1, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("inner:"), + "langgraph_node": "inner", + "langgraph_path": [PULL, "inner"], + "langgraph_step": 2, + "langgraph_triggers": ["outer_1"], + "langgraph_checkpoint_ns": AnyStr("inner:"), }, created_at=AnyStr(), parent_config={ @@ -9278,6 +9449,13 @@ def outer_2(state: State): "writes": None, "step": 0, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("inner:"), + "langgraph_node": "inner", + "langgraph_path": [PULL, "inner"], + "langgraph_step": 2, + "langgraph_triggers": ["outer_1"], + "langgraph_checkpoint_ns": AnyStr("inner:"), }, created_at=AnyStr(), parent_config={ @@ -9320,6 +9498,13 @@ def outer_2(state: State): "writes": {"__start__": {"my_key": "hi my value"}}, "step": -1, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("inner:"), + "langgraph_node": "inner", + "langgraph_path": [PULL, "inner"], + "langgraph_step": 2, + "langgraph_triggers": ["outer_1"], + "langgraph_checkpoint_ns": AnyStr("inner:"), }, created_at=AnyStr(), parent_config=None, @@ -9355,6 +9540,7 @@ def outer_2(state: State): "outer_2": {"my_key": "hi my value here and there and back again"} }, "step": 3, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9386,6 +9572,7 @@ def outer_2(state: State): "outer_2": {"my_key": "hi my value here and there and back again"} }, "step": 3, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9419,6 +9606,7 @@ def outer_2(state: State): "source": "loop", "writes": {"inner": {"my_key": "hi my value here and there"}}, "step": 2, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9455,6 +9643,7 @@ def outer_2(state: State): "source": "loop", "writes": {"outer_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9483,7 +9672,13 @@ def outer_2(state: State): "checkpoint_id": AnyStr(), } }, - metadata={"parents": {}, "source": "loop", "writes": None, "step": 0}, + metadata={ + "parents": {}, + "source": "loop", + "writes": None, + "step": 0, + "thread_id": "1", + }, created_at=AnyStr(), parent_config={ "configurable": { @@ -9516,6 +9711,7 @@ def outer_2(state: State): "source": "input", "writes": {"__start__": {"my_key": "my value"}}, "step": -1, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -9622,6 +9818,7 @@ def parent_2(state: State): "source": "loop", "writes": {"parent_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9663,6 +9860,7 @@ def parent_2(state: State): "source": "loop", "writes": None, "step": 0, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9709,6 +9907,13 @@ def parent_2(state: State): "source": "loop", "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")], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config={ @@ -9779,6 +9984,16 @@ def parent_2(state: State): "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"), + ], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config={ @@ -9816,6 +10031,13 @@ def parent_2(state: State): "source": "loop", "writes": None, "step": 0, + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child", + "langgraph_path": [PULL, AnyStr("child")], + "langgraph_step": 2, + "langgraph_triggers": [AnyStr("parent_1")], + "langgraph_checkpoint_ns": AnyStr("child:"), }, created_at=AnyStr(), parent_config={ @@ -9844,6 +10066,7 @@ def parent_2(state: State): "source": "loop", "writes": {"parent_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9854,7 +10077,7 @@ def parent_2(state: State): } }, ) - # resume + # # resume assert [c for c in app.stream(None, config, subgraphs=True)] == [ ( (AnyStr("child:"), AnyStr("child_1:")), @@ -9886,6 +10109,7 @@ def parent_2(state: State): "parent_2": {"my_key": "hi my value here and there and back again"} }, "step": 3, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9918,6 +10142,7 @@ def parent_2(state: State): "parent_2": {"my_key": "hi my value here and there and back again"} }, "step": 3, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9943,6 +10168,7 @@ def parent_2(state: State): "writes": {"child": {"my_key": "hi my value here and there"}}, "step": 2, "parents": {}, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9990,6 +10216,7 @@ def parent_2(state: State): "source": "loop", "writes": {"parent_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -10010,7 +10237,13 @@ def parent_2(state: State): "checkpoint_id": AnyStr(), } }, - metadata={"source": "loop", "writes": None, "step": 0, "parents": {}}, + metadata={ + "source": "loop", + "writes": None, + "step": 0, + "parents": {}, + "thread_id": "1", + }, created_at=AnyStr(), parent_config={ "configurable": { @@ -10043,6 +10276,7 @@ def parent_2(state: State): "writes": {"__start__": {"my_key": "my value"}}, "step": -1, "parents": {}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -10077,6 +10311,13 @@ def parent_2(state: State): "writes": {"child_1": {"my_key": "hi my value here and there"}}, "step": 1, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child", + "langgraph_path": [PULL, AnyStr("child")], + "langgraph_step": 2, + "langgraph_triggers": [AnyStr("parent_1")], + "langgraph_checkpoint_ns": AnyStr("child:"), }, created_at=AnyStr(), parent_config={ @@ -10109,6 +10350,13 @@ def parent_2(state: State): "writes": None, "step": 0, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child", + "langgraph_path": [PULL, AnyStr("child")], + "langgraph_step": 2, + "langgraph_triggers": [AnyStr("parent_1")], + "langgraph_checkpoint_ns": AnyStr("child:"), }, created_at=AnyStr(), parent_config={ @@ -10154,6 +10402,13 @@ def parent_2(state: State): "writes": {"__start__": {"my_key": "hi my value"}}, "step": -1, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child", + "langgraph_path": [PULL, AnyStr("child")], + "langgraph_step": 2, + "langgraph_triggers": [AnyStr("parent_1")], + "langgraph_checkpoint_ns": AnyStr("child:"), }, created_at=AnyStr(), parent_config=None, @@ -10197,6 +10452,16 @@ def parent_2(state: State): AnyStr("child:"): AnyStr(), } ), + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child_1", + "langgraph_path": [ + PULL, + AnyStr("child_1"), + ], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config={ @@ -10242,6 +10507,16 @@ def parent_2(state: State): AnyStr("child:"): AnyStr(), } ), + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child_1", + "langgraph_path": [ + PULL, + AnyStr("child_1"), + ], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config={ @@ -10294,6 +10569,16 @@ def parent_2(state: State): AnyStr("child:"): AnyStr(), } ), + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child_1", + "langgraph_path": [ + PULL, + AnyStr("child_1"), + ], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config={ @@ -10346,6 +10631,16 @@ def parent_2(state: State): AnyStr("child:"): AnyStr(), } ), + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child_1", + "langgraph_path": [ + PULL, + AnyStr("child_1"), + ], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config=None, @@ -10460,7 +10755,13 @@ def edit(state: JokeState): "checkpoint_id": AnyStr(), } }, - metadata={"parents": {}, "source": "loop", "writes": None, "step": 0}, + metadata={ + "parents": {}, + "source": "loop", + "writes": None, + "step": 0, + "thread_id": "1", + }, created_at=AnyStr(), parent_config={ "configurable": { @@ -10492,6 +10793,13 @@ def edit(state: JokeState): "source": "loop", "writes": {"edit": None}, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("generate_joke:"), + "langgraph_checkpoint_ns": AnyStr("generate_joke:"), + "langgraph_node": "generate_joke", + "langgraph_path": [PUSH, 0], + "langgraph_step": 1, + "langgraph_triggers": [PUSH], }, created_at=AnyStr(), parent_config={ @@ -10530,6 +10838,13 @@ def edit(state: JokeState): "source": "loop", "writes": {"edit": None}, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("generate_joke:"), + "langgraph_checkpoint_ns": AnyStr("generate_joke:"), + "langgraph_node": "generate_joke", + "langgraph_path": [PUSH, 1], + "langgraph_step": 1, + "langgraph_triggers": [PUSH], }, created_at=AnyStr(), parent_config={ @@ -10583,6 +10898,7 @@ def edit(state: JokeState): ] }, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -10624,6 +10940,7 @@ def edit(state: JokeState): ] }, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -10670,7 +10987,13 @@ def edit(state: JokeState): "checkpoint_id": AnyStr(), } }, - metadata={"parents": {}, "source": "loop", "writes": None, "step": 0}, + metadata={ + "parents": {}, + "source": "loop", + "writes": None, + "step": 0, + "thread_id": "1", + }, created_at=AnyStr(), parent_config={ "configurable": { @@ -10703,6 +11026,7 @@ def edit(state: JokeState): "source": "input", "writes": {"__start__": {"subjects": ["cats", "dogs"]}}, "step": -1, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -10864,6 +11188,7 @@ def weather_graph(state: RouterState): "writes": {"router_node": {"route": "weather"}}, "step": 1, "parents": {}, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -10951,6 +11276,7 @@ def weather_graph(state: RouterState): "writes": {"router_node": {"route": "weather"}}, "step": 1, "parents": {}, + "thread_id": "14", }, created_at=AnyStr(), parent_config={ @@ -10991,6 +11317,15 @@ def weather_graph(state: RouterState): "writes": {"model_node": {"city": "San Francisco"}}, "step": 1, "parents": {"": AnyStr()}, + "thread_id": "14", + "checkpoint_ns": AnyStr("weather_graph:"), + "langgraph_node": "weather_graph", + "langgraph_path": [PULL, "weather_graph"], + "langgraph_step": 2, + "langgraph_triggers": [ + "branch:router_node:route_after_prediction:weather_graph" + ], + "langgraph_checkpoint_ns": AnyStr("weather_graph:"), }, created_at=AnyStr(), parent_config={ @@ -11041,6 +11376,7 @@ def weather_graph(state: RouterState): "writes": {"router_node": {"route": "weather"}}, "step": 1, "parents": {}, + "thread_id": "14", }, created_at=AnyStr(), parent_config={ @@ -11078,14 +11414,24 @@ def weather_graph(state: RouterState): } }, metadata={ - "source": "update", "step": 2, + "source": "update", "writes": { "weather_node": { "messages": [{"role": "assistant", "content": "rainy"}] } }, "parents": {"": AnyStr()}, + "thread_id": "14", + "checkpoint_id": AnyStr(), + "checkpoint_ns": AnyStr("weather_graph:"), + "langgraph_node": "weather_graph", + "langgraph_path": [PULL, "weather_graph"], + "langgraph_step": 2, + "langgraph_triggers": [ + "branch:router_node:route_after_prediction:weather_graph" + ], + "langgraph_checkpoint_ns": AnyStr("weather_graph:"), }, created_at=AnyStr(), parent_config={ diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index bedbcef4d..25072eeaf 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -306,12 +306,14 @@ async def tool_two_node(s: State) -> State: "source": "loop", "step": 0, "writes": None, + "thread_id": "1", }, { "parents": {}, "source": "input", "step": -1, "writes": {"__start__": {"my_key": "value ⛰️", "market": "DE"}}, + "thread_id": "1", }, ] tup = await tool_two.checkpointer.aget_tuple(thread1) @@ -328,7 +330,13 @@ async def tool_two_node(s: State) -> State: ), config=tup.config, created_at=tup.checkpoint["ts"], - metadata={"parents": {}, "source": "loop", "step": 0, "writes": None}, + metadata={ + "parents": {}, + "source": "loop", + "step": 0, + "writes": None, + "thread_id": "1", + }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread1, limit=2) ][-1].config, @@ -487,6 +495,7 @@ async def alittlewhile(input: State) -> None: "source": "loop", "step": 0, "writes": None, + "thread_id": "1", } @@ -563,6 +572,7 @@ async def alittlewhile(input: State) -> None: "source": "loop", "step": 1, "writes": {"alittlewhile": {"value": 2}}, + "thread_id": "2", } @@ -946,6 +956,7 @@ async def test_invoke_two_processes_in_out_interrupt( "source": "loop", "step": 6, "writes": {"two": 5}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[1].config, @@ -968,6 +979,7 @@ async def test_invoke_two_processes_in_out_interrupt( "source": "loop", "step": 5, "writes": {"one": None}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[2].config, @@ -990,6 +1002,7 @@ async def test_invoke_two_processes_in_out_interrupt( "source": "input", "step": 4, "writes": {"input": 3}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[3].config, @@ -1010,6 +1023,7 @@ async def test_invoke_two_processes_in_out_interrupt( "source": "loop", "step": 3, "writes": {"one": None}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[4].config, @@ -1032,6 +1046,7 @@ async def test_invoke_two_processes_in_out_interrupt( "source": "input", "step": 2, "writes": {"input": 20}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[5].config, @@ -1052,6 +1067,7 @@ async def test_invoke_two_processes_in_out_interrupt( "source": "loop", "step": 1, "writes": {"two": 4}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[6].config, @@ -1074,6 +1090,7 @@ async def test_invoke_two_processes_in_out_interrupt( "source": "loop", "step": 0, "writes": {"one": None}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[7].config, @@ -1096,6 +1113,7 @@ async def test_invoke_two_processes_in_out_interrupt( "source": "input", "step": -1, "writes": {"input": 2}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -1171,6 +1189,7 @@ async def test_fork_always_re_runs_nodes( "source": "loop", "step": 5, "writes": {"add_one": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[1].config, @@ -1191,6 +1210,7 @@ async def test_fork_always_re_runs_nodes( "source": "loop", "step": 4, "writes": {"add_one": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[2].config, @@ -1211,6 +1231,7 @@ async def test_fork_always_re_runs_nodes( "source": "loop", "step": 3, "writes": {"add_one": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[3].config, @@ -1231,6 +1252,7 @@ async def test_fork_always_re_runs_nodes( "source": "loop", "step": 2, "writes": {"add_one": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[4].config, @@ -1251,6 +1273,7 @@ async def test_fork_always_re_runs_nodes( "source": "loop", "step": 1, "writes": {"add_one": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=history[5].config, @@ -1266,7 +1289,13 @@ async def test_fork_always_re_runs_nodes( "checkpoint_id": AnyStr(), } }, - metadata={"parents": {}, "source": "loop", "step": 0, "writes": None}, + metadata={ + "parents": {}, + "source": "loop", + "step": 0, + "writes": None, + "thread_id": "1", + }, created_at=AnyStr(), parent_config=history[6].config, ), @@ -1288,6 +1317,7 @@ async def test_fork_always_re_runs_nodes( "source": "input", "step": -1, "writes": {"__start__": 1}, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -1710,6 +1740,7 @@ def reset(self): "source": "loop", "step": 0, "writes": None, + "thread_id": "1", } # should contain pending write of "one" checkpoint = await checkpointer.aget_tuple(thread1) @@ -1798,6 +1829,7 @@ def reset(self): "step": 1, "source": "loop", "writes": {"one": {"value": 2}, "two": {"value": 3}}, + "thread_id": "1", }, parent_config={ "configurable": { @@ -1844,7 +1876,13 @@ def reset(self): "start:two": "__start__", }, }, - metadata={"parents": {}, "step": 0, "source": "loop", "writes": None}, + metadata={ + "parents": {}, + "step": 0, + "source": "loop", + "writes": None, + "thread_id": "1", + }, parent_config={ "configurable": { "thread_id": "1", @@ -1886,6 +1924,7 @@ def reset(self): "step": -1, "source": "input", "writes": {"__start__": {"value": 1}}, + "thread_id": "1", }, parent_config=None, pending_writes=UnsortedSequence( @@ -2732,6 +2771,7 @@ async def should_continue(data: dict, config: RunnableConfig) -> str: } } }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -2781,6 +2821,7 @@ async def should_continue(data: dict, config: RunnableConfig) -> str: "input": "what is weather in sf", } }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -2905,6 +2946,7 @@ async def should_continue(data: dict, config: RunnableConfig) -> str: ), } }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -2971,6 +3013,7 @@ async def should_continue(data: dict, config: RunnableConfig) -> str: } } }, + "thread_id": "2", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -3020,6 +3063,7 @@ async def should_continue(data: dict, config: RunnableConfig) -> str: "input": "what is weather in sf", } }, + "thread_id": "2", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -3144,6 +3188,7 @@ async def should_continue(data: dict, config: RunnableConfig) -> str: ), } }, + "thread_id": "2", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -3210,6 +3255,7 @@ async def should_continue(data: dict, config: RunnableConfig) -> str: } } }, + "thread_id": "3", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -3630,6 +3676,7 @@ def should_continue(data: AgentState) -> str: ), } }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -3676,6 +3723,7 @@ def should_continue(data: AgentState) -> str: ) } }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -3756,6 +3804,7 @@ def should_continue(data: AgentState) -> str: ) } }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -3815,6 +3864,7 @@ def should_continue(data: AgentState) -> str: ), } }, + "thread_id": "2", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -3860,6 +3910,7 @@ def should_continue(data: AgentState) -> str: ) } }, + "thread_id": "2", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -3938,6 +3989,7 @@ def should_continue(data: AgentState) -> str: ) } }, + "thread_id": "2", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -4632,6 +4684,7 @@ async def tools_node(input: ToolCall, config: RunnableConfig) -> AgentState: ) } }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -4685,6 +4738,7 @@ async def tools_node(input: ToolCall, config: RunnableConfig) -> AgentState: ) } }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -4794,6 +4848,7 @@ async def tools_node(input: ToolCall, config: RunnableConfig) -> AgentState: ) }, }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -4843,6 +4898,7 @@ async def tools_node(input: ToolCall, config: RunnableConfig) -> AgentState: "messages": AIMessage(content="answer", id="ai2"), } }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -5103,6 +5159,7 @@ def should_continue(messages): id="ai1", ) }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -5152,6 +5209,7 @@ def should_continue(messages): id="ai1", ) }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -5237,6 +5295,7 @@ def should_continue(messages): id="ai2", ) }, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -5280,6 +5339,7 @@ def should_continue(messages): "source": "update", "step": 5, "writes": {"agent": AIMessage(content="answer", id="ai2")}, + "thread_id": "1", }, parent_config=[ c async for c in app_w_interrupt.checkpointer.alist(config, limit=2) @@ -5582,12 +5642,16 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: "source": "loop", "step": 0, "writes": None, + "assistant_id": "a", + "thread_id": "1", }, { "parents": {}, "source": "input", "step": -1, "writes": {"__start__": {"my_key": "value", "market": "DE"}}, + "assistant_id": "a", + "thread_id": "1", }, ] assert await tool_two.aget_state(thread1) == StateSnapshot( @@ -5598,7 +5662,14 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: created_at=(await tool_two.checkpointer.aget_tuple(thread1)).checkpoint[ "ts" ], - metadata={"parents": {}, "source": "loop", "step": 0, "writes": None}, + metadata={ + "parents": {}, + "source": "loop", + "step": 0, + "writes": None, + "assistant_id": "a", + "thread_id": "1", + }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread1, limit=2) ][-1].config, @@ -5621,6 +5692,8 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: "source": "loop", "step": 1, "writes": {"tool_two_slow": {"my_key": " slow"}}, + "assistant_id": "a", + "thread_id": "1", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread1, limit=2) @@ -5641,7 +5714,14 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: created_at=(await tool_two.checkpointer.aget_tuple(thread2)).checkpoint[ "ts" ], - metadata={"parents": {}, "source": "loop", "step": 0, "writes": None}, + metadata={ + "parents": {}, + "source": "loop", + "step": 0, + "writes": None, + "assistant_id": "a", + "thread_id": "2", + }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread2, limit=2) ][-1].config, @@ -5664,6 +5744,8 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: "source": "loop", "step": 1, "writes": {"tool_two_fast": {"my_key": " fast"}}, + "assistant_id": "a", + "thread_id": "2", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread2, limit=2) @@ -5684,7 +5766,14 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: created_at=(await tool_two.checkpointer.aget_tuple(thread3)).checkpoint[ "ts" ], - metadata={"parents": {}, "source": "loop", "step": 0, "writes": None}, + metadata={ + "parents": {}, + "source": "loop", + "step": 0, + "writes": None, + "assistant_id": "b", + "thread_id": "3", + }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread3, limit=2) ][-1].config, @@ -5704,6 +5793,8 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: "source": "update", "step": 1, "writes": {START: {"my_key": "key"}}, + "assistant_id": "b", + "thread_id": "3", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread3, limit=2) @@ -5727,6 +5818,8 @@ def tool_two_fast(data: State, config: RunnableConfig) -> State: "source": "loop", "step": 2, "writes": {"tool_two_fast": {"my_key": " fast"}}, + "assistant_id": "b", + "thread_id": "3", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread3, limit=2) @@ -5795,6 +5888,7 @@ class State(TypedDict): "source": "input", "step": -1, "writes": {"__start__": {"my_key": "value", "market": "DE"}}, + "thread_id": "10", }, "parent_config": None, "next": ["__start__"], @@ -5833,6 +5927,7 @@ class State(TypedDict): "source": "loop", "step": 0, "writes": None, + "thread_id": "10", }, "parent_config": { "tags": [], @@ -5904,6 +5999,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "10", }, "parent_config": { "tags": [], @@ -5975,6 +6071,7 @@ class State(TypedDict): "source": "loop", "step": 2, "writes": {"tool_two_slow": {"my_key": " slow"}}, + "thread_id": "10", }, "parent_config": { "tags": [], @@ -6046,6 +6143,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "10", }, "parent_config": { "tags": [], @@ -6103,6 +6201,7 @@ class State(TypedDict): "source": "input", "step": -1, "writes": {"__start__": {"my_key": "value", "market": "DE"}}, + "thread_id": "11", }, "parent_config": None, "next": ["__start__"], @@ -6141,6 +6240,7 @@ class State(TypedDict): "source": "loop", "step": 0, "writes": None, + "thread_id": "11", }, "parent_config": { "tags": [], @@ -6212,6 +6312,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "11", }, "parent_config": { "tags": [], @@ -6249,6 +6350,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "11", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread1, limit=2) @@ -6272,6 +6374,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "11", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread1, limit=2) @@ -6297,6 +6400,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "12", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread2, limit=2) @@ -6320,6 +6424,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "12", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread2, limit=2) @@ -6353,6 +6458,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "21", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread1, limit=2) @@ -6376,6 +6482,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "21", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread1, limit=2) @@ -6401,6 +6508,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "22", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread2, limit=2) @@ -6424,6 +6532,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "22", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread2, limit=2) @@ -6447,6 +6556,7 @@ class State(TypedDict): "source": "update", "step": 0, "writes": {START: {"my_key": "key", "market": "DE"}}, + "thread_id": "23", }, parent_config=None, ) @@ -6469,6 +6579,7 @@ class State(TypedDict): "source": "loop", "step": 1, "writes": {"prepare": {"my_key": " prepared"}}, + "thread_id": "23", }, parent_config=uconfig, ) @@ -6490,6 +6601,7 @@ class State(TypedDict): "source": "loop", "step": 3, "writes": {"finish": {"my_key": " finished"}}, + "thread_id": "23", }, parent_config=[ c async for c in tool_two.checkpointer.alist(thread3, limit=2) @@ -6841,6 +6953,7 @@ async def decider(data: State) -> str: "source": "loop", "writes": {"qa": {"answer": "doc1,doc2,doc3,doc4"}}, "step": 4, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -7806,6 +7919,7 @@ def outer_2(state: State): "source": "loop", "writes": {"outer_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -7832,9 +7946,8 @@ def outer_2(state: State): tasks=( PregelTask( AnyStr(), - name="inner_2", - path=(PULL, "inner_2"), - error=None, + "inner_2", + (PULL, "inner_2"), ), ), next=("inner_2",), @@ -7860,6 +7973,13 @@ def outer_2(state: State): } }, "step": 1, + "thread_id": "1", + "checkpoint_ns": AnyStr("inner:"), + "langgraph_node": "inner", + "langgraph_path": [PULL, "inner"], + "langgraph_step": 2, + "langgraph_triggers": ["outer_1"], + "langgraph_checkpoint_ns": AnyStr("inner:"), }, created_at=AnyStr(), parent_config={ @@ -7888,6 +8008,7 @@ def outer_2(state: State): "source": "loop", "writes": {"outer_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -7929,6 +8050,7 @@ def outer_2(state: State): "source": "loop", "writes": {"outer_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -7962,6 +8084,7 @@ def outer_2(state: State): "source": "loop", "writes": None, "step": 0, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -7995,6 +8118,7 @@ def outer_2(state: State): "source": "input", "writes": {"__start__": {"my_key": "my value"}}, "step": -1, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -8014,7 +8138,7 @@ def outer_2(state: State): "checkpoint_ns": AnyStr("inner:"), "checkpoint_id": AnyStr(), "checkpoint_map": AnyDict( - {"": AnyStr(), AnyStr("inner:"): AnyStr()} + {"": AnyStr(), AnyStr("child:"): AnyStr()} ), } }, @@ -8028,6 +8152,13 @@ def outer_2(state: State): }, "step": 1, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("inner:"), + "langgraph_node": "inner", + "langgraph_path": [PULL, "inner"], + "langgraph_step": 2, + "langgraph_triggers": ["outer_1"], + "langgraph_checkpoint_ns": AnyStr("inner:"), }, created_at=AnyStr(), parent_config={ @@ -8036,13 +8167,11 @@ def outer_2(state: State): "checkpoint_ns": AnyStr("inner:"), "checkpoint_id": AnyStr(), "checkpoint_map": AnyDict( - {"": AnyStr(), AnyStr("inner:"): AnyStr()} + {"": AnyStr(), AnyStr("child:"): AnyStr()} ), } }, - tasks=( - PregelTask(id=AnyStr(), name="inner_2", path=(PULL, "inner_2")), - ), + tasks=(PregelTask(AnyStr(), "inner_2", (PULL, "inner_2")),), ), StateSnapshot( values={"my_key": "hi my value"}, @@ -8053,7 +8182,7 @@ def outer_2(state: State): "checkpoint_ns": AnyStr("inner:"), "checkpoint_id": AnyStr(), "checkpoint_map": AnyDict( - {"": AnyStr(), AnyStr("inner:"): AnyStr()} + {"": AnyStr(), AnyStr("child:"): AnyStr()} ), } }, @@ -8062,6 +8191,13 @@ def outer_2(state: State): "writes": None, "step": 0, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("inner:"), + "langgraph_node": "inner", + "langgraph_path": [PULL, "inner"], + "langgraph_step": 2, + "langgraph_triggers": ["outer_1"], + "langgraph_checkpoint_ns": AnyStr("inner:"), }, created_at=AnyStr(), parent_config={ @@ -8070,15 +8206,15 @@ def outer_2(state: State): "checkpoint_ns": AnyStr("inner:"), "checkpoint_id": AnyStr(), "checkpoint_map": AnyDict( - {"": AnyStr(), AnyStr("inner:"): AnyStr()} + {"": AnyStr(), AnyStr("child:"): AnyStr()} ), } }, tasks=( PregelTask( - id=AnyStr(), - name="inner_1", - path=(PULL, "inner_1"), + AnyStr(), + "inner_1", + (PULL, "inner_1"), result={ "my_key": "hi my value here", "my_other_key": "hi my value", @@ -8095,7 +8231,7 @@ def outer_2(state: State): "checkpoint_ns": AnyStr("inner:"), "checkpoint_id": AnyStr(), "checkpoint_map": AnyDict( - {"": AnyStr(), AnyStr("inner:"): AnyStr()} + {"": AnyStr(), AnyStr("child:"): AnyStr()} ), } }, @@ -8104,14 +8240,21 @@ def outer_2(state: State): "writes": {"__start__": {"my_key": "hi my value"}}, "step": -1, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("inner:"), + "langgraph_node": "inner", + "langgraph_path": [PULL, "inner"], + "langgraph_step": 2, + "langgraph_triggers": ["outer_1"], + "langgraph_checkpoint_ns": AnyStr("inner:"), }, created_at=AnyStr(), parent_config=None, tasks=( PregelTask( - id=AnyStr(), - name="__start__", - path=(PULL, "__start__"), + AnyStr(), + "__start__", + (PULL, "__start__"), result={"my_key": "hi my value"}, ), ), @@ -8139,6 +8282,7 @@ def outer_2(state: State): "outer_2": {"my_key": "hi my value here and there and back again"} }, "step": 3, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8172,6 +8316,7 @@ def outer_2(state: State): } }, "step": 3, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8205,6 +8350,7 @@ def outer_2(state: State): "source": "loop", "writes": {"inner": {"my_key": "hi my value here and there"}}, "step": 2, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8244,6 +8390,7 @@ def outer_2(state: State): "source": "loop", "writes": {"outer_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8277,6 +8424,7 @@ def outer_2(state: State): "source": "loop", "writes": None, "step": 0, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8310,6 +8458,7 @@ def outer_2(state: State): "source": "input", "writes": {"__start__": {"my_key": "my value"}}, "step": -1, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -8415,6 +8564,7 @@ def parent_2(state: State): "source": "loop", "writes": {"parent_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8449,9 +8599,6 @@ def parent_2(state: State): "thread_id": "1", "checkpoint_ns": AnyStr("child:"), "checkpoint_id": AnyStr(), - "checkpoint_map": AnyDict( - {"": AnyStr(), AnyStr("child:"): AnyStr()} - ), } }, metadata={ @@ -8459,6 +8606,7 @@ def parent_2(state: State): "source": "loop", "writes": None, "step": 0, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8466,9 +8614,6 @@ def parent_2(state: State): "thread_id": "1", "checkpoint_ns": AnyStr("child:"), "checkpoint_id": AnyStr(), - "checkpoint_map": AnyDict( - {"": AnyStr(), AnyStr("child:"): AnyStr()} - ), } }, ).tasks[0] @@ -8508,6 +8653,13 @@ def parent_2(state: State): "source": "loop", "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")], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config={ @@ -8580,6 +8732,16 @@ def parent_2(state: State): } }, "step": 1, + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child_1", + "langgraph_path": [ + PULL, + AnyStr("child_1"), + ], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config={ @@ -8617,6 +8779,13 @@ def parent_2(state: State): "source": "loop", "writes": None, "step": 0, + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child", + "langgraph_path": [PULL, AnyStr("child")], + "langgraph_step": 2, + "langgraph_triggers": [AnyStr("parent_1")], + "langgraph_checkpoint_ns": AnyStr("child:"), }, created_at=AnyStr(), parent_config={ @@ -8645,6 +8814,7 @@ def parent_2(state: State): "source": "loop", "writes": {"parent_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8692,6 +8862,7 @@ def parent_2(state: State): } }, "step": 3, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8728,6 +8899,7 @@ def parent_2(state: State): } }, "step": 3, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8753,6 +8925,7 @@ def parent_2(state: State): "source": "loop", "writes": {"child": {"my_key": "hi my value here and there"}}, "step": 2, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8796,6 +8969,7 @@ def parent_2(state: State): "source": "loop", "writes": {"parent_1": {"my_key": "hi my value"}}, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8821,6 +8995,7 @@ def parent_2(state: State): "source": "loop", "writes": None, "step": 0, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -8851,6 +9026,7 @@ def parent_2(state: State): "source": "input", "writes": {"my_key": "my value"}, "step": -1, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -8885,6 +9061,13 @@ def parent_2(state: State): "writes": {"child_1": {"my_key": "hi my value here and there"}}, "step": 1, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child", + "langgraph_path": [PULL, AnyStr("child")], + "langgraph_step": 2, + "langgraph_triggers": [AnyStr("parent_1")], + "langgraph_checkpoint_ns": AnyStr("child:"), }, created_at=AnyStr(), parent_config={ @@ -8917,6 +9100,13 @@ def parent_2(state: State): "writes": None, "step": 0, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child", + "langgraph_path": [PULL, AnyStr("child")], + "langgraph_step": 2, + "langgraph_triggers": [AnyStr("parent_1")], + "langgraph_checkpoint_ns": AnyStr("child:"), }, created_at=AnyStr(), parent_config={ @@ -8962,6 +9152,13 @@ def parent_2(state: State): "writes": {"__start__": {"my_key": "hi my value"}}, "step": -1, "parents": {"": AnyStr()}, + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child", + "langgraph_path": [PULL, AnyStr("child")], + "langgraph_step": 2, + "langgraph_triggers": [AnyStr("parent_1")], + "langgraph_checkpoint_ns": AnyStr("child:"), }, created_at=AnyStr(), parent_config=None, @@ -9009,6 +9206,16 @@ def parent_2(state: State): AnyStr("child:"): AnyStr(), } ), + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child_1", + "langgraph_path": [ + PULL, + AnyStr("child_1"), + ], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config={ @@ -9054,6 +9261,16 @@ def parent_2(state: State): AnyStr("child:"): AnyStr(), } ), + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child_1", + "langgraph_path": [ + PULL, + AnyStr("child_1"), + ], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config={ @@ -9106,6 +9323,16 @@ def parent_2(state: State): AnyStr("child:"): AnyStr(), } ), + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child_1", + "langgraph_path": [ + PULL, + AnyStr("child_1"), + ], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config={ @@ -9158,6 +9385,16 @@ def parent_2(state: State): AnyStr("child:"): AnyStr(), } ), + "thread_id": "1", + "checkpoint_ns": AnyStr("child:"), + "langgraph_checkpoint_ns": AnyStr("child:"), + "langgraph_node": "child_1", + "langgraph_path": [ + PULL, + AnyStr("child_1"), + ], + "langgraph_step": 1, + "langgraph_triggers": [AnyStr("start:child_1")], }, created_at=AnyStr(), parent_config=None, @@ -9272,7 +9509,13 @@ async def edit(state: JokeState): "checkpoint_id": AnyStr(), } }, - metadata={"parents": {}, "source": "loop", "writes": None, "step": 0}, + metadata={ + "parents": {}, + "source": "loop", + "writes": None, + "step": 0, + "thread_id": "1", + }, created_at=AnyStr(), parent_config={ "configurable": { @@ -9319,6 +9562,7 @@ async def edit(state: JokeState): ] }, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9361,6 +9605,7 @@ async def edit(state: JokeState): ] }, "step": 1, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9407,7 +9652,13 @@ async def edit(state: JokeState): "checkpoint_id": AnyStr(), } }, - metadata={"parents": {}, "source": "loop", "writes": None, "step": 0}, + metadata={ + "parents": {}, + "source": "loop", + "writes": None, + "step": 0, + "thread_id": "1", + }, created_at=AnyStr(), parent_config={ "configurable": { @@ -9440,6 +9691,7 @@ async def edit(state: JokeState): "source": "input", "writes": {"__start__": {"subjects": ["cats", "dogs"]}}, "step": -1, + "thread_id": "1", }, created_at=AnyStr(), parent_config=None, @@ -9617,6 +9869,7 @@ def get_first_in_list(): "writes": {"router_node": {"route": "weather"}}, "step": 1, "parents": {}, + "thread_id": "1", }, created_at=AnyStr(), parent_config={ @@ -9708,6 +9961,7 @@ def get_first_in_list(): "writes": {"router_node": {"route": "weather"}}, "step": 1, "parents": {}, + "thread_id": "14", }, created_at=AnyStr(), parent_config={ @@ -9748,6 +10002,15 @@ def get_first_in_list(): "writes": {"model_node": {"city": "San Francisco"}}, "step": 1, "parents": {"": AnyStr()}, + "thread_id": "14", + "checkpoint_ns": AnyStr("weather_graph:"), + "langgraph_node": "weather_graph", + "langgraph_path": [PULL, "weather_graph"], + "langgraph_step": 2, + "langgraph_triggers": [ + "branch:router_node:route_after_prediction:weather_graph" + ], + "langgraph_checkpoint_ns": AnyStr("weather_graph:"), }, created_at=AnyStr(), parent_config={ @@ -9798,6 +10061,7 @@ def get_first_in_list(): "writes": {"router_node": {"route": "weather"}}, "step": 1, "parents": {}, + "thread_id": "14", }, created_at=AnyStr(), parent_config={ @@ -9835,8 +10099,8 @@ def get_first_in_list(): } }, metadata={ - "source": "update", "step": 2, + "source": "update", "writes": { "weather_node": { "messages": [ @@ -9845,6 +10109,16 @@ def get_first_in_list(): } }, "parents": {"": AnyStr()}, + "thread_id": "14", + "checkpoint_id": AnyStr(), + "checkpoint_ns": AnyStr("weather_graph:"), + "langgraph_node": "weather_graph", + "langgraph_path": [PULL, "weather_graph"], + "langgraph_step": 2, + "langgraph_triggers": [ + "branch:router_node:route_after_prediction:weather_graph" + ], + "langgraph_checkpoint_ns": AnyStr("weather_graph:"), }, created_at=AnyStr(), parent_config={ diff --git a/libs/langgraph/tests/test_remote_graph.py b/libs/langgraph/tests/test_remote_graph.py index 21fb8278f..70857ed61 100644 --- a/libs/langgraph/tests/test_remote_graph.py +++ b/libs/langgraph/tests/test_remote_graph.py @@ -73,7 +73,9 @@ def test_get_graph(): "__start__": DrawableNode( id="__start__", name="__start__", data="__start__", metadata=None ), - "__end__": DrawableNode(id="__end__", name="__end__", data="__end__", metadata=None), + "__end__": DrawableNode( + id="__end__", name="__end__", data="__end__", metadata=None + ), "agent": DrawableNode( id="agent", name="agent_1", @@ -120,7 +122,9 @@ async def test_aget_graph(): "__start__": DrawableNode( id="__start__", name="__start__", data="__start__", metadata=None ), - "__end__": DrawableNode(id="__end__", name="__end__", data="__end__", metadata=None), + "__end__": DrawableNode( + id="__end__", name="__end__", data="__end__", metadata=None + ), "agent": DrawableNode( id="agent", name="agent_1",