Speed up task triggers check (#3890)

- Using a sentinel value is faster than raising-catching an exception

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

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",

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

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()

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

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

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()

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)

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

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)
         ):

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