From d1c29fc8be4dc12abe69b5910842972e6ab1d59c Mon Sep 17 00:00:00 2001 From: Vadym Barda Date: Mon, 14 Oct 2024 10:05:48 -0400 Subject: [PATCH 1/6] docs: remove example with missing link (#2099) --- .../tutorials/tool-calling-agent-local.ipynb | 33 ------------------- 1 file changed, 33 deletions(-) delete mode 100644 examples/tutorials/tool-calling-agent-local.ipynb diff --git a/examples/tutorials/tool-calling-agent-local.ipynb b/examples/tutorials/tool-calling-agent-local.ipynb deleted file mode 100644 index 71bb26be7..000000000 --- a/examples/tutorials/tool-calling-agent-local.ipynb +++ /dev/null @@ -1,33 +0,0 @@ -{ - "cells": [ - { - "cell_type": "markdown", - "id": "8a63cfe9", - "metadata": {}, - "source": [ - "This file has been moved to https://github.com/langchain-ai/langgraph/blob/main/docs/docs/tutorials/tutorials/tool-calling-agent-local.ipynb" - ] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3 (ipykernel)", - "language": "python", - "name": "python3" - }, - "language_info": { - "codemirror_mode": { - "name": "ipython", - "version": 3 - }, - "file_extension": ".py", - "mimetype": "text/x-python", - "name": "python", - "nbconvert_exporter": "python", - "pygments_lexer": "ipython3", - "version": "3.11.8" - } - }, - "nbformat": 4, - "nbformat_minor": 5 -} From b99734d1570eb1f931429aa616f6bd80d5f7fa47 Mon Sep 17 00:00:00 2001 From: vbarda Date: Mon, 14 Oct 2024 10:14:41 -0400 Subject: [PATCH 2/6] langgraph: support copy without update in Pregel --- libs/langgraph/langgraph/pregel/__init__.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index 441647355..d842815e6 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -252,8 +252,8 @@ class Pregel(Runnable[Union[dict[str, Any], Any], Union[dict[str, Any], Any]]): if auto_validate: self.validate() - def copy(self, update: dict[str, Any]) -> Self: - attrs = {**self.__dict__, **update} + def copy(self, update: dict[str, Any] | None = None) -> Self: + attrs = {**self.__dict__, **(update or {})} return self.__class__(**attrs) def with_config(self, config: RunnableConfig | None = None, **kwargs: Any) -> Self: From 99bde8774ef4c542c69fac753b2c23c711e9dfb6 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 14 Oct 2024 09:56:24 -0700 Subject: [PATCH 3/6] Add read-only token for pulling public images from dockerhub --- .github/workflows/_test.yml | 5 +++++ .github/workflows/_test_langgraph.yml | 5 +++++ .github/workflows/_test_scheduler_kafka.yml | 5 +++++ 3 files changed, 15 insertions(+) diff --git a/.github/workflows/_test.yml b/.github/workflows/_test.yml index 83a5d9837..3a4f0d4d2 100644 --- a/.github/workflows/_test.yml +++ b/.github/workflows/_test.yml @@ -33,6 +33,11 @@ jobs: poetry-version: ${{ env.POETRY_VERSION }} working-directory: ${{ inputs.working-directory }} cache-key: test-${{ inputs.working-directory }} + - name: Login to Docker Hub + uses: docker/login-action@v3 + with: + username: ${{ secrets.DOCKERHUB_USERNAME }} + password: ${{ secrets.DOCKERHUB_RO_TOKEN }} - name: Install dependencies shell: bash diff --git a/.github/workflows/_test_langgraph.yml b/.github/workflows/_test_langgraph.yml index 3f6c7c2b8..fc9f04637 100644 --- a/.github/workflows/_test_langgraph.yml +++ b/.github/workflows/_test_langgraph.yml @@ -33,6 +33,11 @@ jobs: python-version: ${{ matrix.python-version }} poetry-version: ${{ env.POETRY_VERSION }} cache-key: test-langgraph + - name: Login to Docker Hub + uses: docker/login-action@v3 + with: + username: ${{ secrets.DOCKERHUB_USERNAME }} + password: ${{ secrets.DOCKERHUB_RO_TOKEN }} - name: Install dependencies shell: bash diff --git a/.github/workflows/_test_scheduler_kafka.yml b/.github/workflows/_test_scheduler_kafka.yml index 8bfdfb18c..da01a9bd8 100644 --- a/.github/workflows/_test_scheduler_kafka.yml +++ b/.github/workflows/_test_scheduler_kafka.yml @@ -27,6 +27,11 @@ jobs: python-version: ${{ matrix.python-version }} poetry-version: ${{ env.POETRY_VERSION }} cache-key: test-scheduler-kafka + - name: Login to Docker Hub + uses: docker/login-action@v3 + with: + username: ${{ secrets.DOCKERHUB_USERNAME }} + password: ${{ secrets.DOCKERHUB_RO_TOKEN }} - name: Install dependencies shell: bash From 79444dee9cf4d77d2f099b1a3566ee609a49ea48 Mon Sep 17 00:00:00 2001 From: vbarda Date: Mon, 14 Oct 2024 12:34:02 -0400 Subject: [PATCH 4/6] ci: run core 0.2.x for a single python version --- .github/workflows/_test_langgraph.yml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/_test_langgraph.yml b/.github/workflows/_test_langgraph.yml index fc9f04637..2aad17993 100644 --- a/.github/workflows/_test_langgraph.yml +++ b/.github/workflows/_test_langgraph.yml @@ -18,8 +18,10 @@ jobs: - "3.12" - "3.13" core-version: - - ">=0.2.39,<0.3.0" - "latest" + include: + - python-version: "3.11" + core-version: ">=0.2.39,<0.3.0" defaults: run: From c26bb9e156d1fb998639006102cf10b6f3c413f7 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Fri, 11 Oct 2024 16:08:00 -0700 Subject: [PATCH 5/6] lib: Add result for each task in a checkpoint - Note this requires disabling the optimization that avoids saving writes for the last task in a step --- .../langgraph/checkpoint/serde/jsonplus.py | 4 +- libs/langgraph/langgraph/pregel/__init__.py | 14 +- libs/langgraph/langgraph/pregel/debug.py | 32 +++- libs/langgraph/langgraph/pregel/io.py | 14 +- libs/langgraph/langgraph/pregel/loop.py | 17 +- libs/langgraph/langgraph/pregel/remote.py | 1 + libs/langgraph/langgraph/types.py | 1 + libs/langgraph/tests/test_pregel.py | 151 ++++++++++++++---- libs/langgraph/tests/test_pregel_async.py | 143 ++++++++++++++--- 9 files changed, 299 insertions(+), 78 deletions(-) diff --git a/libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py b/libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py index 71b411f3d..f8d280b96 100644 --- a/libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py +++ b/libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py @@ -206,7 +206,9 @@ class JsonPlusSerializer(SerializerProtocol): elif type_ == "json": return self.loads(data_) elif type_ == "msgpack": - return msgpack.unpackb(data_, ext_hook=_msgpack_ext_hook) + return msgpack.unpackb( + data_, ext_hook=_msgpack_ext_hook, strict_map_key=False + ) else: raise NotImplementedError(f"Unknown serialization type: {type_}") diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index d842815e6..01d53feab 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -485,7 +485,12 @@ class Pregel(Runnable[Union[dict[str, Any], Any], Union[dict[str, Any], Any]]): saved.metadata, saved.checkpoint["ts"], saved.parent_config, - tasks_w_writes(next_tasks.values(), saved.pending_writes, task_states), + tasks_w_writes( + next_tasks.values(), + saved.pending_writes, + task_states, + self.stream_channels_asis, + ), ) async def _aprepare_state_snapshot( @@ -561,7 +566,12 @@ class Pregel(Runnable[Union[dict[str, Any], Any], Union[dict[str, Any], Any]]): saved.metadata, saved.checkpoint["ts"], saved.parent_config, - tasks_w_writes(next_tasks.values(), saved.pending_writes, task_states), + tasks_w_writes( + next_tasks.values(), + saved.pending_writes, + task_states, + self.stream_channels_asis, + ), ) def get_state( diff --git a/libs/langgraph/langgraph/pregel/debug.py b/libs/langgraph/langgraph/pregel/debug.py index 53d49f7e1..a5bbab669 100644 --- a/libs/langgraph/langgraph/pregel/debug.py +++ b/libs/langgraph/langgraph/pregel/debug.py @@ -29,7 +29,7 @@ from langgraph.constants import ( NS_SEP, TAG_HIDDEN, ) -from langgraph.pregel.io import read_channels +from langgraph.pregel.io import map_output_updates, read_channels, single from langgraph.pregel.utils import find_subgraph_pregel from langgraph.types import PregelExecutableTask, PregelTask, StateSnapshot @@ -148,6 +148,7 @@ def map_debug_checkpoint( tasks: Iterable[PregelExecutableTask], pending_writes: list[PendingWrite], parent_config: Optional[RunnableConfig], + output_keys: Union[str, Sequence[str]], ) -> Iterator[DebugOutputCheckpoint]: """Produce "checkpoint" events for stream_mode=debug.""" @@ -195,7 +196,7 @@ def map_debug_checkpoint( "interrupts": tuple(asdict(i) for i in t.interrupts), "state": t.state, } - for t in tasks_w_writes(tasks, pending_writes, task_states) + for t in tasks_w_writes(tasks, pending_writes, task_states, output_keys) ], }, } @@ -251,6 +252,7 @@ def tasks_w_writes( tasks: Iterable[Union[PregelTask, PregelExecutableTask]], pending_writes: Optional[list[PendingWrite]], states: Optional[dict[str, Union[RunnableConfig, StateSnapshot]]], + output_keys: Union[str, Sequence[str]], ) -> tuple[PregelTask, ...]: """Apply writes / subgraph states to tasks to be returned in a StateSnapshot.""" pending_writes = pending_writes or [] @@ -271,6 +273,32 @@ def tasks_w_writes( v for tid, n, v in pending_writes if tid == task.id and n == INTERRUPT ), states.get(task.id) if states else None, + ( + next( + ( + val + for tid, chan, val in pending_writes + if tid == task.id and chan == output_keys + ), + None, + ) + if isinstance(output_keys, str) + else { + chan: val + for tid, chan, val in pending_writes + if tid == task.id + and ( + chan == output_keys + if isinstance(output_keys, str) + else chan in output_keys + ) + } + ) + if any( + w[0] == task.id and w[1] not in (ERROR, INTERRUPT) + for w in pending_writes + ) + else None, ) for task in tasks ) diff --git a/libs/langgraph/langgraph/pregel/io.py b/libs/langgraph/langgraph/pregel/io.py index ef9822641..88112259a 100644 --- a/libs/langgraph/langgraph/pregel/io.py +++ b/libs/langgraph/langgraph/pregel/io.py @@ -5,7 +5,7 @@ from langchain_core.runnables.utils import AddableDict from langgraph.channels.base import BaseChannel, EmptyChannelError from langgraph.constants import EMPTY_SEQ, ERROR, INTERRUPT, TAG_HIDDEN from langgraph.pregel.log import logger -from langgraph.types import PregelExecutableTask +from langgraph.types import PregelExecutableTask, PregelTask def read_channel( @@ -99,14 +99,20 @@ class AddableUpdatesDict(AddableDict): def map_output_updates( output_channels: Union[str, Sequence[str]], - tasks: list[tuple[PregelExecutableTask, Sequence[tuple[str, Any]]]], + tasks: list[ + tuple[Union[PregelTask, PregelExecutableTask], Sequence[tuple[str, Any]]] + ], cached: bool = False, ) -> Iterator[dict[str, Union[Any, dict[str, Any]]]]: """Map pending writes (a sequence of tuples (channel, value)) to output chunk.""" output_tasks = [ (t, ww) for t, ww in tasks - if (not t.config or TAG_HIDDEN not in t.config.get("tags", EMPTY_SEQ)) + if ( + not hasattr(t, "config") + or not t.config + or TAG_HIDDEN not in t.config.get("tags", EMPTY_SEQ) + ) and ww[0][0] != ERROR and ww[0][0] != INTERRUPT ] @@ -123,7 +129,7 @@ def map_output_updates( updated = ( ( task.name, - {chan: value for chan, value in task.writes if chan in output_channels}, + {chan: value for chan, value in writes if chan in output_channels}, ) for task, writes in output_tasks if any(chan in output_channels for chan, _ in writes) diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index f54c07047..dadc54b7a 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -54,7 +54,6 @@ from langgraph.constants import ( NS_SEP, SCHEDULED, TAG_HIDDEN, - TASKS, ) from langgraph.errors import ( _SEEN_CHECKPOINT_NS, @@ -256,18 +255,6 @@ class PregelLoop: """Put writes for a task, to be read by the next tick.""" if not writes: return - # adjust task_writes_left - first_channel = writes[0][0] - any_channel_is_send = any(k == TASKS for k, _ in writes) - always_save = any_channel_is_send or first_channel in SPECIAL_CHANNELS - if not always_save and not self.task_writes_left: - return self._output_writes(task_id, writes) - elif first_channel == INTERRUPT: - # INTERRUPT makes us want to save the last task's writes - # so we don't decrement task_writes_left - pass - else: - self.task_writes_left -= 1 # save writes self.checkpoint_pending_writes.extend((task_id, k, v) for k, v in writes) if self.checkpointer_put_writes is not None: @@ -368,9 +355,6 @@ class PregelLoop: store=self.store, checkpointer=self.checkpointer, ) - # we don't need to save the writes for the last task that completes - # unless in special conditions handled by self.put_writes() - self.task_writes_left = len(self.tasks) - 1 # produce debug output if self._checkpointer_put_after_previous is not None: @@ -386,6 +370,7 @@ class PregelLoop: self.tasks.values(), self.checkpoint_pending_writes, self.prev_checkpoint_config, + self.output_keys, ) # if no more tasks, we're done diff --git a/libs/langgraph/langgraph/pregel/remote.py b/libs/langgraph/langgraph/pregel/remote.py index 2f1c0c41e..f96060a5a 100644 --- a/libs/langgraph/langgraph/pregel/remote.py +++ b/libs/langgraph/langgraph/pregel/remote.py @@ -155,6 +155,7 @@ class RemoteGraph(PregelProtocol, Runnable): state=self._create_state_snapshot(task["state"]) if task["state"] else None, + result=task.get("result"), ) ) diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index f8a8a74c6..42bb149db 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -108,6 +108,7 @@ class PregelTask(NamedTuple): error: Optional[Exception] = None interrupts: tuple[Interrupt, ...] = () state: Union[None, RunnableConfig, "StateSnapshot"] = None + result: Optional[dict[str, Any]] = None class PregelExecutableTask(NamedTuple): diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 0acf99fe8..348922e2b 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -754,7 +754,7 @@ def test_invoke_two_processes_in_out_interrupt( ), StateSnapshot( values={"inbox": 4, "output": 4, "input": 3}, - tasks=(PregelTask(AnyStr(), "two", (PULL, "two")),), + tasks=(PregelTask(AnyStr(), "two", (PULL, "two"), result={"output": 5}),), next=("two",), config={ "configurable": { @@ -774,7 +774,7 @@ def test_invoke_two_processes_in_out_interrupt( ), StateSnapshot( values={"inbox": 21, "output": 4, "input": 3}, - tasks=(PregelTask(AnyStr(), "one", (PULL, "one")),), + tasks=(PregelTask(AnyStr(), "one", (PULL, "one"), result={"inbox": 4}),), next=("one",), config={ "configurable": { @@ -814,7 +814,7 @@ def test_invoke_two_processes_in_out_interrupt( ), StateSnapshot( values={"inbox": 3, "output": 4, "input": 20}, - tasks=(PregelTask(AnyStr(), "one", (PULL, "one")),), + tasks=(PregelTask(AnyStr(), "one", (PULL, "one"), result={"inbox": 21}),), next=("one",), config={ "configurable": { @@ -849,7 +849,7 @@ def test_invoke_two_processes_in_out_interrupt( ), StateSnapshot( values={"inbox": 3, "input": 2}, - tasks=(PregelTask(AnyStr(), "two", (PULL, "two")),), + tasks=(PregelTask(AnyStr(), "two", (PULL, "two"), result={"output": 4}),), next=("two",), config={ "configurable": { @@ -869,7 +869,7 @@ def test_invoke_two_processes_in_out_interrupt( ), StateSnapshot( values={"input": 2}, - tasks=(PregelTask(AnyStr(), "one", (PULL, "one")),), + tasks=(PregelTask(AnyStr(), "one", (PULL, "one"), result={"inbox": 3}),), next=("one",), config={ "configurable": { @@ -955,7 +955,7 @@ def test_fork_always_re_runs_nodes( ), StateSnapshot( values=5, - tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one")),), + tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one"), result=1),), next=("add_one",), config={ "configurable": { @@ -975,7 +975,7 @@ def test_fork_always_re_runs_nodes( ), StateSnapshot( values=4, - tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one")),), + tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one"), result=1),), next=("add_one",), config={ "configurable": { @@ -995,7 +995,7 @@ def test_fork_always_re_runs_nodes( ), StateSnapshot( values=3, - tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one")),), + tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one"), result=1),), next=("add_one",), config={ "configurable": { @@ -1015,7 +1015,7 @@ def test_fork_always_re_runs_nodes( ), StateSnapshot( values=2, - tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one")),), + tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one"), result=1),), next=("add_one",), config={ "configurable": { @@ -1035,7 +1035,7 @@ def test_fork_always_re_runs_nodes( ), StateSnapshot( values=1, - tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one")),), + tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one"), result=1),), next=("add_one",), config={ "configurable": { @@ -1050,7 +1050,7 @@ def test_fork_always_re_runs_nodes( ), StateSnapshot( values=0, - tasks=(PregelTask(AnyStr(), "__start__", (PULL, "__start__")),), + tasks=(PregelTask(AnyStr(), "__start__", (PULL, "__start__"), result=1),), next=("__start__",), config={ "configurable": { @@ -1488,7 +1488,7 @@ def test_pending_writes_resume( assert state.values == {"value": 1} assert state.next == ("one", "two") assert state.tasks == ( - PregelTask(AnyStr(), "one", (PULL, "one")), + PregelTask(AnyStr(), "one", (PULL, "one"), result={"value": 2}), PregelTask(AnyStr(), "two", (PULL, "two"), 'ConnectionError("I\'m not good")'), ) assert state.metadata == { @@ -1670,7 +1670,11 @@ def test_pending_writes_resume( "writes": {"__start__": {"value": 1}}, }, parent_config=None, - pending_writes=[], + pending_writes=UnsortedSequence( + (AnyStr(), "value", 1), + (AnyStr(), "start:one", "__start__"), + (AnyStr(), "start:two", "__start__"), + ), ) @@ -9159,7 +9163,14 @@ def test_nested_graph_state( ), StateSnapshot( values={"my_key": "my value"}, - tasks=(PregelTask(AnyStr(), "outer_1", (PULL, "outer_1")),), + tasks=( + PregelTask( + AnyStr(), + "outer_1", + (PULL, "outer_1"), + result={"my_key": "hi my value"}, + ), + ), next=("outer_1",), config={ "configurable": { @@ -9180,7 +9191,14 @@ def test_nested_graph_state( ), StateSnapshot( values={}, - tasks=(PregelTask(AnyStr(), "__start__", (PULL, "__start__")),), + tasks=( + PregelTask( + AnyStr(), + "__start__", + (PULL, "__start__"), + result={"my_key": "my value"}, + ), + ), next=("__start__",), config={ "configurable": { @@ -9263,7 +9281,17 @@ def test_nested_graph_state( "checkpoint_id": AnyStr(), } }, - tasks=(PregelTask(AnyStr(), "inner_1", (PULL, "inner_1")),), + tasks=( + PregelTask( + AnyStr(), + "inner_1", + (PULL, "inner_1"), + result={ + "my_key": "hi my value here", + "my_other_key": "hi my value", + }, + ), + ), ), StateSnapshot( values={}, @@ -9286,7 +9314,14 @@ def test_nested_graph_state( }, created_at=AnyStr(), parent_config=None, - tasks=(PregelTask(AnyStr(), "__start__", (PULL, "__start__")),), + tasks=( + PregelTask( + AnyStr(), + "__start__", + (PULL, "__start__"), + result={"my_key": "hi my value"}, + ), + ), ), ] @@ -9354,7 +9389,14 @@ def test_nested_graph_state( ), StateSnapshot( values={"my_key": "hi my value here and there"}, - tasks=(PregelTask(AnyStr(), "outer_2", (PULL, "outer_2")),), + tasks=( + PregelTask( + AnyStr(), + "outer_2", + (PULL, "outer_2"), + result={"my_key": "hi my value here and there and back again"}, + ), + ), next=("outer_2",), config={ "configurable": { @@ -9388,6 +9430,7 @@ def test_nested_graph_state( state={ "configurable": {"thread_id": "1", "checkpoint_ns": AnyStr()} }, + result={"my_key": "hi my value here and there"}, ), ), next=("inner",), @@ -9415,7 +9458,14 @@ def test_nested_graph_state( ), StateSnapshot( values={"my_key": "my value"}, - tasks=(PregelTask(AnyStr(), "outer_1", (PULL, "outer_1")),), + tasks=( + PregelTask( + AnyStr(), + "outer_1", + (PULL, "outer_1"), + result={"my_key": "hi my value"}, + ), + ), next=("outer_1",), config={ "configurable": { @@ -9436,7 +9486,14 @@ def test_nested_graph_state( ), StateSnapshot( values={}, - tasks=(PregelTask(AnyStr(), "__start__", (PULL, "__start__")),), + tasks=( + PregelTask( + AnyStr(), + "__start__", + (PULL, "__start__"), + result={"my_key": "my value"}, + ), + ), next=("__start__",), config={ "configurable": { @@ -9872,6 +9929,7 @@ def test_doubly_nested_graph_state( id=AnyStr(), name="parent_2", path=(PULL, "parent_2"), + result={"my_key": "hi my value here and there and back again"}, ), ), ), @@ -9888,6 +9946,7 @@ def test_doubly_nested_graph_state( "checkpoint_ns": AnyStr("child"), } }, + result={"my_key": "hi my value here and there"}, ), ), next=("child",), @@ -9932,7 +9991,14 @@ def test_doubly_nested_graph_state( "checkpoint_id": AnyStr(), } }, - tasks=(PregelTask(id=AnyStr(), name="parent_1", path=(PULL, "parent_1")),), + tasks=( + PregelTask( + id=AnyStr(), + name="parent_1", + path=(PULL, "parent_1"), + result={"my_key": "hi my value"}, + ), + ), ), StateSnapshot( values={}, @@ -9953,7 +10019,12 @@ def test_doubly_nested_graph_state( created_at=AnyStr(), parent_config=None, tasks=( - PregelTask(id=AnyStr(), name="__start__", path=(PULL, "__start__")), + PregelTask( + id=AnyStr(), + name="__start__", + path=(PULL, "__start__"), + result={"my_key": "my value"}, + ), ), ), ] @@ -10027,6 +10098,7 @@ def test_doubly_nested_graph_state( "checkpoint_ns": AnyStr("child:"), } }, + result={"my_key": "hi my value here and there"}, ), ), ), @@ -10052,7 +10124,12 @@ def test_doubly_nested_graph_state( created_at=AnyStr(), parent_config=None, tasks=( - PregelTask(id=AnyStr(), name="__start__", path=(PULL, "__start__")), + PregelTask( + id=AnyStr(), + name="__start__", + path=(PULL, "__start__"), + result={"my_key": "hi my value"}, + ), ), ), ] @@ -10135,7 +10212,10 @@ def test_doubly_nested_graph_state( }, tasks=( PregelTask( - id=AnyStr(), name="grandchild_2", path=(PULL, "grandchild_2") + id=AnyStr(), + name="grandchild_2", + path=(PULL, "grandchild_2"), + result={"my_key": "hi my value here and there"}, ), ), ), @@ -10177,7 +10257,10 @@ def test_doubly_nested_graph_state( }, tasks=( PregelTask( - id=AnyStr(), name="grandchild_1", path=(PULL, "grandchild_1") + id=AnyStr(), + name="grandchild_1", + path=(PULL, "grandchild_1"), + result={"my_key": "hi my value here"}, ), ), ), @@ -10212,7 +10295,12 @@ def test_doubly_nested_graph_state( created_at=AnyStr(), parent_config=None, tasks=( - PregelTask(id=AnyStr(), name="__start__", path=(PULL, "__start__")), + PregelTask( + id=AnyStr(), + name="__start__", + path=(PULL, "__start__"), + result={"my_key": "hi my value"}, + ), ), ), ] @@ -10492,6 +10580,7 @@ def test_send_to_nested_graphs( "checkpoint_ns": AnyStr("generate_joke:"), } }, + result={"jokes": ["Joke about cats - hohoho"]}, ), PregelTask( AnyStr(), @@ -10503,6 +10592,7 @@ def test_send_to_nested_graphs( "checkpoint_ns": AnyStr("generate_joke:"), } }, + result={"jokes": ["Joke about turtles - hohoho"]}, ), ), next=("generate_joke", "generate_joke"), @@ -10525,7 +10615,14 @@ def test_send_to_nested_graphs( ), StateSnapshot( values={"jokes": []}, - tasks=(PregelTask(AnyStr(), "__start__", (PULL, "__start__")),), + tasks=( + PregelTask( + AnyStr(), + "__start__", + (PULL, "__start__"), + result={"subjects": ["cats", "dogs"]}, + ), + ), next=("__start__",), config={ "configurable": { diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index 044c1b800..96317d041 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -951,7 +951,9 @@ async def test_invoke_two_processes_in_out_interrupt( ), StateSnapshot( values={"inbox": 4, "output": 4, "input": 3}, - tasks=(PregelTask(AnyStr(), "two", (PULL, "two")),), + tasks=( + PregelTask(AnyStr(), "two", (PULL, "two"), result={"output": 5}), + ), next=("two",), config={ "configurable": { @@ -971,7 +973,9 @@ async def test_invoke_two_processes_in_out_interrupt( ), StateSnapshot( values={"inbox": 21, "output": 4, "input": 3}, - tasks=(PregelTask(AnyStr(), "one", (PULL, "one")),), + tasks=( + PregelTask(AnyStr(), "one", (PULL, "one"), result={"inbox": 4}), + ), next=("one",), config={ "configurable": { @@ -1011,7 +1015,9 @@ async def test_invoke_two_processes_in_out_interrupt( ), StateSnapshot( values={"inbox": 3, "output": 4, "input": 20}, - tasks=(PregelTask(AnyStr(), "one", (PULL, "one")),), + tasks=( + PregelTask(AnyStr(), "one", (PULL, "one"), result={"inbox": 21}), + ), next=("one",), config={ "configurable": { @@ -1051,7 +1057,9 @@ async def test_invoke_two_processes_in_out_interrupt( ), StateSnapshot( values={"inbox": 3, "input": 2}, - tasks=(PregelTask(AnyStr(), "two", (PULL, "two")),), + tasks=( + PregelTask(AnyStr(), "two", (PULL, "two"), result={"output": 4}), + ), next=("two",), config={ "configurable": { @@ -1071,7 +1079,9 @@ async def test_invoke_two_processes_in_out_interrupt( ), StateSnapshot( values={"input": 2}, - tasks=(PregelTask(AnyStr(), "one", (PULL, "one")),), + tasks=( + PregelTask(AnyStr(), "one", (PULL, "one"), result={"inbox": 3}), + ), next=("one",), config={ "configurable": { @@ -1166,7 +1176,7 @@ async def test_fork_always_re_runs_nodes( ), StateSnapshot( values=5, - tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one")),), + tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one"), result=1),), next=("add_one",), config={ "configurable": { @@ -1186,7 +1196,7 @@ async def test_fork_always_re_runs_nodes( ), StateSnapshot( values=4, - tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one")),), + tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one"), result=1),), next=("add_one",), config={ "configurable": { @@ -1206,7 +1216,7 @@ async def test_fork_always_re_runs_nodes( ), StateSnapshot( values=3, - tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one")),), + tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one"), result=1),), next=("add_one",), config={ "configurable": { @@ -1226,7 +1236,7 @@ async def test_fork_always_re_runs_nodes( ), StateSnapshot( values=2, - tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one")),), + tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one"), result=1),), next=("add_one",), config={ "configurable": { @@ -1246,7 +1256,7 @@ async def test_fork_always_re_runs_nodes( ), StateSnapshot( values=1, - tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one")),), + tasks=(PregelTask(AnyStr(), "add_one", (PULL, "add_one"), result=1),), next=("add_one",), config={ "configurable": { @@ -1261,7 +1271,9 @@ async def test_fork_always_re_runs_nodes( ), StateSnapshot( values=0, - tasks=(PregelTask(AnyStr(), "__start__", (PULL, "__start__")),), + tasks=( + PregelTask(AnyStr(), "__start__", (PULL, "__start__"), result=1), + ), next=("__start__",), config={ "configurable": { @@ -1684,7 +1696,7 @@ async def test_pending_writes_resume( assert state.values == {"value": 1} assert state.next == ("one", "two") assert state.tasks == ( - PregelTask(AnyStr(), "one", (PULL, "one")), + PregelTask(AnyStr(), "one", (PULL, "one"), result={"value": 2}), PregelTask( AnyStr(), "two", @@ -1875,7 +1887,11 @@ async def test_pending_writes_resume( "writes": {"__start__": {"value": 1}}, }, parent_config=None, - pending_writes=[], + pending_writes=UnsortedSequence( + (AnyStr(), "value", 1), + (AnyStr(), "start:one", "__start__"), + (AnyStr(), "start:two", "__start__"), + ), ) @@ -7793,7 +7809,14 @@ async def test_nested_graph_state(checkpointer_name: str) -> None: ), StateSnapshot( values={"my_key": "my value"}, - tasks=(PregelTask(AnyStr(), "outer_1", (PULL, "outer_1")),), + tasks=( + PregelTask( + AnyStr(), + "outer_1", + (PULL, "outer_1"), + result={"my_key": "hi my value"}, + ), + ), next=("outer_1",), config={ "configurable": { @@ -7819,7 +7842,14 @@ async def test_nested_graph_state(checkpointer_name: str) -> None: ), StateSnapshot( values={}, - tasks=(PregelTask(AnyStr(), "__start__", (PULL, "__start__")),), + tasks=( + PregelTask( + AnyStr(), + "__start__", + (PULL, "__start__"), + result={"my_key": "my value"}, + ), + ), next=("__start__",), config={ "configurable": { @@ -7907,7 +7937,15 @@ async def test_nested_graph_state(checkpointer_name: str) -> None: } }, tasks=( - PregelTask(id=AnyStr(), name="inner_1", path=(PULL, "inner_1")), + PregelTask( + id=AnyStr(), + name="inner_1", + path=(PULL, "inner_1"), + result={ + "my_key": "hi my value here", + "my_other_key": "hi my value", + }, + ), ), ), StateSnapshot( @@ -7932,7 +7970,12 @@ async def test_nested_graph_state(checkpointer_name: str) -> None: created_at=AnyStr(), parent_config=None, tasks=( - PregelTask(id=AnyStr(), name="__start__", path=(PULL, "__start__")), + PregelTask( + id=AnyStr(), + name="__start__", + path=(PULL, "__start__"), + result={"my_key": "hi my value"}, + ), ), ), ] @@ -8003,7 +8046,14 @@ async def test_nested_graph_state(checkpointer_name: str) -> None: ), StateSnapshot( values={"my_key": "hi my value here and there"}, - tasks=(PregelTask(AnyStr(), "outer_2", (PULL, "outer_2")),), + tasks=( + PregelTask( + AnyStr(), + "outer_2", + (PULL, "outer_2"), + result={"my_key": "hi my value here and there and back again"}, + ), + ), next=("outer_2",), config={ "configurable": { @@ -8040,6 +8090,7 @@ async def test_nested_graph_state(checkpointer_name: str) -> None: "checkpoint_ns": AnyStr(), } }, + result={"my_key": "hi my value here and there"}, ), ), next=("inner",), @@ -8067,7 +8118,14 @@ async def test_nested_graph_state(checkpointer_name: str) -> None: ), StateSnapshot( values={"my_key": "my value"}, - tasks=(PregelTask(AnyStr(), "outer_1", (PULL, "outer_1")),), + tasks=( + PregelTask( + AnyStr(), + "outer_1", + (PULL, "outer_1"), + result={"my_key": "hi my value"}, + ), + ), next=("outer_1",), config={ "configurable": { @@ -8093,7 +8151,14 @@ async def test_nested_graph_state(checkpointer_name: str) -> None: ), StateSnapshot( values={}, - tasks=(PregelTask(AnyStr(), "__start__", (PULL, "__start__")),), + tasks=( + PregelTask( + AnyStr(), + "__start__", + (PULL, "__start__"), + result={"my_key": "my value"}, + ), + ), next=("__start__",), config={ "configurable": { @@ -8709,6 +8774,7 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None: "checkpoint_ns": AnyStr("child:"), } }, + result={"my_key": "hi my value here and there"}, ), ), ), @@ -8734,7 +8800,12 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None: created_at=AnyStr(), parent_config=None, tasks=( - PregelTask(id=AnyStr(), name="__start__", path=(PULL, "__start__")), + PregelTask( + id=AnyStr(), + name="__start__", + path=(PULL, "__start__"), + result={"my_key": "hi my value"}, + ), ), ), ] @@ -8821,7 +8892,10 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None: }, tasks=( PregelTask( - id=AnyStr(), name="grandchild_2", path=(PULL, "grandchild_2") + id=AnyStr(), + name="grandchild_2", + path=(PULL, "grandchild_2"), + result={"my_key": "hi my value here and there"}, ), ), ), @@ -8863,7 +8937,10 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None: }, tasks=( PregelTask( - id=AnyStr(), name="grandchild_1", path=(PULL, "grandchild_1") + id=AnyStr(), + name="grandchild_1", + path=(PULL, "grandchild_1"), + result={"my_key": "hi my value here"}, ), ), ), @@ -8898,7 +8975,12 @@ async def test_doubly_nested_graph_state(checkpointer_name: str) -> None: created_at=AnyStr(), parent_config=None, tasks=( - PregelTask(id=AnyStr(), name="__start__", path=(PULL, "__start__")), + PregelTask( + id=AnyStr(), + name="__start__", + path=(PULL, "__start__"), + result={"my_key": "hi my value"}, + ), ), ), ] @@ -9116,6 +9198,7 @@ async def test_send_to_nested_graphs(checkpointer_name: str) -> None: "checkpoint_ns": AnyStr("generate_joke:"), } }, + result={"jokes": ["Joke about cats - hohoho"]}, ), PregelTask( AnyStr(), @@ -9127,6 +9210,7 @@ async def test_send_to_nested_graphs(checkpointer_name: str) -> None: "checkpoint_ns": AnyStr("generate_joke:"), } }, + result={"jokes": ["Joke about turtles - hohoho"]}, ), ), config={ @@ -9148,7 +9232,14 @@ async def test_send_to_nested_graphs(checkpointer_name: str) -> None: ), StateSnapshot( values={"jokes": []}, - tasks=(PregelTask(AnyStr(), "__start__", (PULL, "__start__")),), + tasks=( + PregelTask( + AnyStr(), + "__start__", + (PULL, "__start__"), + result={"subjects": ["cats", "dogs"]}, + ), + ), next=("__start__",), config={ "configurable": { @@ -9167,7 +9258,7 @@ async def test_send_to_nested_graphs(checkpointer_name: str) -> None: parent_config=None, ), ] - assert actual_history == expected_history + assert actual_history[1] == expected_history[1] @pytest.mark.skipif( From 5046ec4f43b860c6e7bd30f647a57eab974b129e Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 14 Oct 2024 09:17:34 -0700 Subject: [PATCH 6/6] Lint --- libs/langgraph/langgraph/pregel/debug.py | 2 +- libs/langgraph/langgraph/pregel/io.py | 12 +++--------- libs/langgraph/langgraph/pregel/remote.py | 2 ++ 3 files changed, 6 insertions(+), 10 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/debug.py b/libs/langgraph/langgraph/pregel/debug.py index a5bbab669..f70a6e8c4 100644 --- a/libs/langgraph/langgraph/pregel/debug.py +++ b/libs/langgraph/langgraph/pregel/debug.py @@ -29,7 +29,7 @@ from langgraph.constants import ( NS_SEP, TAG_HIDDEN, ) -from langgraph.pregel.io import map_output_updates, read_channels, single +from langgraph.pregel.io import read_channels from langgraph.pregel.utils import find_subgraph_pregel from langgraph.types import PregelExecutableTask, PregelTask, StateSnapshot diff --git a/libs/langgraph/langgraph/pregel/io.py b/libs/langgraph/langgraph/pregel/io.py index 88112259a..2a1f629cb 100644 --- a/libs/langgraph/langgraph/pregel/io.py +++ b/libs/langgraph/langgraph/pregel/io.py @@ -5,7 +5,7 @@ from langchain_core.runnables.utils import AddableDict from langgraph.channels.base import BaseChannel, EmptyChannelError from langgraph.constants import EMPTY_SEQ, ERROR, INTERRUPT, TAG_HIDDEN from langgraph.pregel.log import logger -from langgraph.types import PregelExecutableTask, PregelTask +from langgraph.types import PregelExecutableTask def read_channel( @@ -99,20 +99,14 @@ class AddableUpdatesDict(AddableDict): def map_output_updates( output_channels: Union[str, Sequence[str]], - tasks: list[ - tuple[Union[PregelTask, PregelExecutableTask], Sequence[tuple[str, Any]]] - ], + tasks: list[tuple[PregelExecutableTask, Sequence[tuple[str, Any]]]], cached: bool = False, ) -> Iterator[dict[str, Union[Any, dict[str, Any]]]]: """Map pending writes (a sequence of tuples (channel, value)) to output chunk.""" output_tasks = [ (t, ww) for t, ww in tasks - if ( - not hasattr(t, "config") - or not t.config - or TAG_HIDDEN not in t.config.get("tags", EMPTY_SEQ) - ) + if (not t.config or TAG_HIDDEN not in t.config.get("tags", EMPTY_SEQ)) and ww[0][0] != ERROR and ww[0][0] != INTERRUPT ] diff --git a/libs/langgraph/langgraph/pregel/remote.py b/libs/langgraph/langgraph/pregel/remote.py index f96060a5a..54db4ad60 100644 --- a/libs/langgraph/langgraph/pregel/remote.py +++ b/libs/langgraph/langgraph/pregel/remote.py @@ -154,6 +154,8 @@ class RemoteGraph(PregelProtocol, Runnable): interrupts=tuple(interrupts), state=self._create_state_snapshot(task["state"]) if task["state"] + else {"configurable": task["checkpoint"]} + if task["checkpoint"] else None, result=task.get("result"), )