From ce1077da40bc10c95d3bd6b80f8a052ef0ba5913 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Mon, 17 Mar 2025 21:04:05 -0700 Subject: [PATCH 1/2] Speed up task triggers check - Using a sentinel value is faster than raising-catching an exception --- libs/langgraph/langgraph/channels/base.py | 9 +++++++++ .../langgraph/channels/ephemeral_value.py | 16 ++++++++++------ libs/langgraph/langgraph/channels/last_value.py | 15 +++++++++++---- .../langgraph/channels/untracked_value.py | 10 +++++++--- libs/langgraph/langgraph/pregel/algo.py | 5 ++--- libs/langgraph/langgraph/pregel/io.py | 7 ++----- 6 files changed, 41 insertions(+), 21 deletions(-) diff --git a/libs/langgraph/langgraph/channels/base.py b/libs/langgraph/langgraph/channels/base.py index 4aaeb5681..c4b49c650 100644 --- a/libs/langgraph/langgraph/channels/base.py +++ b/libs/langgraph/langgraph/channels/base.py @@ -3,6 +3,7 @@ from typing import Any, Generic, Optional, Sequence, TypeVar from typing_extensions import Self +from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError, InvalidUpdateError Value = TypeVar("Value") @@ -64,6 +65,14 @@ class BaseChannel(Generic[Value, Update, C], ABC): """ return False + def get_catch(self) -> Value: + """Return the current value of the channel, or MISSING if the channel + is empty. Subclasses can override to skip the EmptyChannelError check.""" + try: + return self.get() + except EmptyChannelError: + return MISSING + __all__ = [ "BaseChannel", diff --git a/libs/langgraph/langgraph/channels/ephemeral_value.py b/libs/langgraph/langgraph/channels/ephemeral_value.py index 537a8763c..4a64c6b32 100644 --- a/libs/langgraph/langgraph/channels/ephemeral_value.py +++ b/libs/langgraph/langgraph/channels/ephemeral_value.py @@ -3,6 +3,7 @@ from typing import Any, Generic, Optional, Sequence, Type from typing_extensions import Self from langgraph.channels.base import BaseChannel, Value +from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError, InvalidUpdateError @@ -14,6 +15,7 @@ class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]): def __init__(self, typ: Any, guard: bool = True) -> None: super().__init__(typ) self.guard = guard + self.value = MISSING def __eq__(self, value: object) -> bool: return isinstance(value, EphemeralValue) and value.guard == self.guard @@ -37,10 +39,10 @@ class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]): def update(self, values: Sequence[Value]) -> bool: if len(values) == 0: - try: - del self.value + if self.value is not MISSING: + self.value = MISSING return True - except AttributeError: + else: return False if len(values) != 1 and self.guard: raise InvalidUpdateError( @@ -51,7 +53,9 @@ class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]): return True def get(self) -> Value: - try: - return self.value - except AttributeError: + if self.value is MISSING: raise EmptyChannelError() + return self.value + + def get_catch(self) -> Value: + return self.value diff --git a/libs/langgraph/langgraph/channels/last_value.py b/libs/langgraph/langgraph/channels/last_value.py index 5065f4fc5..13fac983e 100644 --- a/libs/langgraph/langgraph/channels/last_value.py +++ b/libs/langgraph/langgraph/channels/last_value.py @@ -1,8 +1,9 @@ -from typing import Generic, Optional, Sequence, Type +from typing import Any, Generic, Optional, Sequence, Type from typing_extensions import Self from langgraph.channels.base import BaseChannel, Value +from langgraph.constants import MISSING from langgraph.errors import ( EmptyChannelError, ErrorCode, @@ -16,6 +17,10 @@ class LastValue(Generic[Value], BaseChannel[Value, Value, Value]): __slots__ = ("value",) + def __init__(self, typ: Any, key: str = "") -> None: + super().__init__(typ, key) + self.value = MISSING + def __eq__(self, value: object) -> bool: return isinstance(value, LastValue) @@ -50,7 +55,9 @@ class LastValue(Generic[Value], BaseChannel[Value, Value, Value]): return True def get(self) -> Value: - try: - return self.value - except AttributeError: + if self.value is MISSING: raise EmptyChannelError() + return self.value + + def get_catch(self) -> Value: + return self.value diff --git a/libs/langgraph/langgraph/channels/untracked_value.py b/libs/langgraph/langgraph/channels/untracked_value.py index 9b1020710..2560d61a3 100644 --- a/libs/langgraph/langgraph/channels/untracked_value.py +++ b/libs/langgraph/langgraph/channels/untracked_value.py @@ -3,6 +3,7 @@ from typing import Generic, Optional, Sequence, Type from typing_extensions import Self from langgraph.channels.base import BaseChannel, Value +from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError, InvalidUpdateError @@ -14,6 +15,7 @@ class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]): def __init__(self, typ: Type[Value], guard: bool = True) -> None: super().__init__(typ) self.guard = guard + self.value = MISSING def __eq__(self, value: object) -> bool: return isinstance(value, UntrackedValue) and value.guard == self.guard @@ -48,7 +50,9 @@ class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]): return True def get(self) -> Value: - try: - return self.value - except AttributeError: + if self.value is MISSING: raise EmptyChannelError() + return self.value + + def get_catch(self) -> Value: + return self.value diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index 03b2af6f4..b2bcdcedf 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -48,6 +48,7 @@ from langgraph.constants import ( EMPTY_SEQ, ERROR, INTERRUPT, + MISSING, NO_WRITES, NS_END, NS_SEP, @@ -649,9 +650,7 @@ def prepare_single_task( if triggers := sorted( chan for chan in proc.triggers - if not isinstance( - read_channel(channels, chan, return_exception=True), EmptyChannelError - ) + if channels[chan].get_catch() is not MISSING and checkpoint["channel_versions"].get(chan, null_version) # type: ignore[operator] > seen.get(chan, null_version) ): diff --git a/libs/langgraph/langgraph/pregel/io.py b/libs/langgraph/langgraph/pregel/io.py index e9963f5c8..30e976d99 100644 --- a/libs/langgraph/langgraph/pregel/io.py +++ b/libs/langgraph/langgraph/pregel/io.py @@ -38,14 +38,11 @@ def read_channel( chan: str, *, catch: bool = True, - return_exception: bool = False, ) -> Any: try: return channels[chan].get() - except EmptyChannelError as exc: - if return_exception: - return exc - elif catch: + except EmptyChannelError: + if catch: return None else: raise From 47d38a302283acf676a2ad4ea9ee541e3978acf6 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 18 Mar 2025 08:19:22 -0700 Subject: [PATCH 2/2] Replace get_catch w is_available --- .../langgraph/langgraph/channels/any_value.py | 23 ++++++++++++------- libs/langgraph/langgraph/channels/base.py | 14 ++++++----- libs/langgraph/langgraph/channels/binop.py | 13 +++++++---- .../channels/dynamic_barrier_value.py | 3 +++ .../langgraph/channels/ephemeral_value.py | 4 ++-- .../langgraph/channels/last_value.py | 4 ++-- .../langgraph/channels/named_barrier_value.py | 3 +++ libs/langgraph/langgraph/channels/topic.py | 3 +++ .../langgraph/channels/untracked_value.py | 4 ++-- libs/langgraph/langgraph/pregel/algo.py | 3 +-- 10 files changed, 47 insertions(+), 27 deletions(-) diff --git a/libs/langgraph/langgraph/channels/any_value.py b/libs/langgraph/langgraph/channels/any_value.py index e9dfb77d6..35452084f 100644 --- a/libs/langgraph/langgraph/channels/any_value.py +++ b/libs/langgraph/langgraph/channels/any_value.py @@ -1,8 +1,9 @@ -from typing import Generic, Optional, Sequence, Type +from typing import Any, Generic, Optional, Sequence, Type from typing_extensions import Self from langgraph.channels.base import BaseChannel, Value +from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError @@ -12,6 +13,10 @@ class AnyValue(Generic[Value], BaseChannel[Value, Value, Value]): __slots__ = ("typ", "value") + def __init__(self, typ: Any, key: str = "") -> None: + super().__init__(typ, key) + self.value = MISSING + def __eq__(self, value: object) -> bool: return isinstance(value, AnyValue) @@ -34,17 +39,19 @@ class AnyValue(Generic[Value], BaseChannel[Value, Value, Value]): def update(self, values: Sequence[Value]) -> bool: if len(values) == 0: - try: - del self.value - return True - except AttributeError: + if self.value is MISSING: return False + else: + self.value = MISSING + return True self.value = values[-1] return True def get(self) -> Value: - try: - return self.value - except AttributeError: + if self.value is MISSING: raise EmptyChannelError() + return self.value + + def is_available(self) -> bool: + return self.value is not MISSING diff --git a/libs/langgraph/langgraph/channels/base.py b/libs/langgraph/langgraph/channels/base.py index c4b49c650..b9239be7a 100644 --- a/libs/langgraph/langgraph/channels/base.py +++ b/libs/langgraph/langgraph/channels/base.py @@ -3,7 +3,6 @@ from typing import Any, Generic, Optional, Sequence, TypeVar from typing_extensions import Self -from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError, InvalidUpdateError Value = TypeVar("Value") @@ -65,13 +64,16 @@ class BaseChannel(Generic[Value, Update, C], ABC): """ return False - def get_catch(self) -> Value: - """Return the current value of the channel, or MISSING if the channel - is empty. Subclasses can override to skip the EmptyChannelError check.""" + def is_available(self) -> bool: + """Return True if the channel is available (not empty), False otherwise. + Subclasses should override this method to provide a more efficient + implementation than calling get() and catching EmptyChannelError. + """ try: - return self.get() + self.get() + return True except EmptyChannelError: - return MISSING + return False __all__ = [ diff --git a/libs/langgraph/langgraph/channels/binop.py b/libs/langgraph/langgraph/channels/binop.py index a2360142b..413e0b91a 100644 --- a/libs/langgraph/langgraph/channels/binop.py +++ b/libs/langgraph/langgraph/channels/binop.py @@ -10,6 +10,7 @@ from typing import ( from typing_extensions import NotRequired, Required, Self from langgraph.channels.base import BaseChannel, Value +from langgraph.constants import MISSING from langgraph.errors import EmptyChannelError @@ -51,7 +52,7 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]): try: self.value = typ() except Exception: - pass + self.value = MISSING def __eq__(self, value: object) -> bool: return isinstance(value, BinaryOperatorAggregate) and ( @@ -81,7 +82,7 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]): def update(self, values: Sequence[Value]) -> bool: if not values: return False - if not hasattr(self, "value"): + if self.value is MISSING: self.value = values[0] values = values[1:] for value in values: @@ -89,7 +90,9 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]): return True def get(self) -> Value: - try: - return self.value - except AttributeError: + if self.value is MISSING: raise EmptyChannelError() + return self.value + + def is_available(self) -> bool: + return self.value is not MISSING diff --git a/libs/langgraph/langgraph/channels/dynamic_barrier_value.py b/libs/langgraph/langgraph/channels/dynamic_barrier_value.py index f64191e86..155c65446 100644 --- a/libs/langgraph/langgraph/channels/dynamic_barrier_value.py +++ b/libs/langgraph/langgraph/channels/dynamic_barrier_value.py @@ -85,6 +85,9 @@ class DynamicBarrierValue( raise EmptyChannelError() return None + def is_available(self) -> bool: + return self.seen == self.names + def consume(self) -> bool: if self.seen == self.names: self.seen = set() diff --git a/libs/langgraph/langgraph/channels/ephemeral_value.py b/libs/langgraph/langgraph/channels/ephemeral_value.py index 4a64c6b32..29a9a698c 100644 --- a/libs/langgraph/langgraph/channels/ephemeral_value.py +++ b/libs/langgraph/langgraph/channels/ephemeral_value.py @@ -57,5 +57,5 @@ class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]): raise EmptyChannelError() return self.value - def get_catch(self) -> Value: - return self.value + def is_available(self) -> bool: + return self.value is not MISSING diff --git a/libs/langgraph/langgraph/channels/last_value.py b/libs/langgraph/langgraph/channels/last_value.py index 13fac983e..61669d390 100644 --- a/libs/langgraph/langgraph/channels/last_value.py +++ b/libs/langgraph/langgraph/channels/last_value.py @@ -59,5 +59,5 @@ class LastValue(Generic[Value], BaseChannel[Value, Value, Value]): raise EmptyChannelError() return self.value - def get_catch(self) -> Value: - return self.value + def is_available(self) -> bool: + return self.value is not MISSING diff --git a/libs/langgraph/langgraph/channels/named_barrier_value.py b/libs/langgraph/langgraph/channels/named_barrier_value.py index 4a1d990ca..553316e19 100644 --- a/libs/langgraph/langgraph/channels/named_barrier_value.py +++ b/libs/langgraph/langgraph/channels/named_barrier_value.py @@ -60,6 +60,9 @@ class NamedBarrierValue(Generic[Value], BaseChannel[Value, Value, set[Value]]): raise EmptyChannelError() return None + def is_available(self) -> bool: + return self.seen == self.names + def consume(self) -> bool: if self.seen == self.names: self.seen = set() diff --git a/libs/langgraph/langgraph/channels/topic.py b/libs/langgraph/langgraph/channels/topic.py index 0430343dc..91e7027f9 100644 --- a/libs/langgraph/langgraph/channels/topic.py +++ b/libs/langgraph/langgraph/channels/topic.py @@ -75,3 +75,6 @@ class Topic( return list(self.values) else: raise EmptyChannelError + + def is_available(self) -> bool: + return bool(self.values) diff --git a/libs/langgraph/langgraph/channels/untracked_value.py b/libs/langgraph/langgraph/channels/untracked_value.py index 2560d61a3..f9168131e 100644 --- a/libs/langgraph/langgraph/channels/untracked_value.py +++ b/libs/langgraph/langgraph/channels/untracked_value.py @@ -54,5 +54,5 @@ class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]): raise EmptyChannelError() return self.value - def get_catch(self) -> Value: - return self.value + def is_available(self) -> bool: + return self.value is not MISSING diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index b2bcdcedf..e8fd76a1e 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -48,7 +48,6 @@ from langgraph.constants import ( EMPTY_SEQ, ERROR, INTERRUPT, - MISSING, NO_WRITES, NS_END, NS_SEP, @@ -650,7 +649,7 @@ def prepare_single_task( if triggers := sorted( chan for chan in proc.triggers - if channels[chan].get_catch() is not MISSING + if channels[chan].is_available() and checkpoint["channel_versions"].get(chan, null_version) # type: ignore[operator] > seen.get(chan, null_version) ):