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 4aaeb5681..b9239be7a 100644 --- a/libs/langgraph/langgraph/channels/base.py +++ b/libs/langgraph/langgraph/channels/base.py @@ -64,6 +64,17 @@ class BaseChannel(Generic[Value, Update, C], ABC): """ return False + 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: + self.get() + return True + except EmptyChannelError: + return False + __all__ = [ "BaseChannel", 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 537a8763c..29a9a698c 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 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 5065f4fc5..61669d390 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 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 9b1020710..f9168131e 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 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 84de26c05..ba028d7bd 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -650,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].is_available() 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 f07745f1b..1bee765db 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