WIP

langgraph/channels/any_value.py

@@ -0,0 +1,55 @@
+from contextlib import contextmanager
+from typing import Generator, Generic, Optional, Sequence, Type
+
+from typing_extensions import Self
+
+from langgraph.channels.base import BaseChannel, EmptyChannelError, Value
+
+
+class AnyValue(Generic[Value], BaseChannel[Value, Value, Value]):
+    """Stores the last value received, assumes that if multiple values are
+    received, they are all equal."""
+
+    def __init__(self, typ: Type[Value]) -> None:
+        self.typ = typ
+
+    @property
+    def ValueType(self) -> Type[Value]:
+        """The type of the value stored in the channel."""
+        return self.typ
+
+    @property
+    def UpdateType(self) -> Type[Value]:
+        """The type of the update received by the channel."""
+        return self.typ
+
+    @contextmanager
+    def empty(self, checkpoint: Optional[Value] = None) -> Generator[Self, None, None]:
+        empty = self.__class__(self.typ)
+        if checkpoint is not None:
+            empty.value = checkpoint
+        try:
+            yield empty
+        finally:
+            try:
+                del empty.value
+            except AttributeError:
+                pass
+
+    def update(self, values: Sequence[Value]) -> None:
+        if len(values) == 0:
+            return
+
+        self.value = values[-1]
+
+    def get(self) -> Value:
+        try:
+            return self.value
+        except AttributeError:
+            raise EmptyChannelError()
+
+    def checkpoint(self) -> Value:
+        try:
+            return self.value
+        except AttributeError:
+            raise EmptyChannelError()

langgraph/channels/ephemeral_value.py

@@ -0,0 +1,67 @@
+from contextlib import contextmanager
+from typing import Generator, Generic, Optional, Sequence, Type
+
+from typing_extensions import Self
+
+from langgraph.channels.base import (
+    BaseChannel,
+    EmptyChannelError,
+    InvalidUpdateError,
+    Value,
+)
+
+
+class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]):
+    """Stores the value received in the step immediately preceding, clears after."""
+
+    def __init__(self, typ: Type[Value], guard: bool = True) -> None:
+        self.typ = typ
+        self.guard = guard
+
+    @property
+    def ValueType(self) -> Type[Value]:
+        """The type of the value stored in the channel."""
+        return self.typ
+
+    @property
+    def UpdateType(self) -> Type[Value]:
+        """The type of the update received by the channel."""
+        return self.typ
+
+    @contextmanager
+    def empty(self, checkpoint: Optional[Value] = None) -> Generator[Self, None, None]:
+        empty = self.__class__(self.typ, self.guard)
+        if checkpoint is not None:
+            empty.value = checkpoint
+        try:
+            yield empty
+        finally:
+            try:
+                del empty.value
+            except AttributeError:
+                pass
+
+    def update(self, values: Sequence[Value]) -> None:
+        if len(values) == 0:
+            try:
+                del self.value
+            except AttributeError:
+                pass
+            finally:
+                return
+        if len(values) != 1 and self.guard:
+            raise InvalidUpdateError("LastValue can only receive one value per step.")
+
+        self.value = values[-1]
+
+    def get(self) -> Value:
+        try:
+            return self.value
+        except AttributeError:
+            raise EmptyChannelError()
+
+    def checkpoint(self) -> Value:
+        try:
+            return self.value
+        except AttributeError:
+            raise EmptyChannelError()

langgraph/channels/last_value.py

@@ -14,9 +14,8 @@ from langgraph.channels.base import (
 class LastValue(Generic[Value], BaseChannel[Value, Value, Value]):
     """Stores the last value received, can receive at most one value per step."""
 
-    def __init__(self, typ: Type[Value], guard: bool = True) -> None:
+    def __init__(self, typ: Type[Value]) -> None:
         self.typ = typ
-        self.guard = guard
 
     @property
     def ValueType(self) -> Type[Value]:
@@ -30,7 +29,7 @@ class LastValue(Generic[Value], BaseChannel[Value, Value, Value]):
 
     @contextmanager
     def empty(self, checkpoint: Optional[Value] = None) -> Generator[Self, None, None]:
-        empty = self.__class__(self.typ, self.guard)
+        empty = self.__class__(self.typ)
         if checkpoint is not None:
             empty.value = checkpoint
         try:
@@ -44,7 +43,7 @@ class LastValue(Generic[Value], BaseChannel[Value, Value, Value]):
     def update(self, values: Sequence[Value]) -> None:
         if len(values) == 0:
             return
-        if len(values) != 1 and self.guard:
+        if len(values) != 1:
             raise InvalidUpdateError("LastValue can only receive one value per step.")
 
         self.value = values[-1]

langgraph/graph/state.py

@@ -6,8 +6,10 @@ from typing import Any, Optional, Sequence, Type
 from langchain_core.runnables import RunnableLambda, RunnablePassthrough
 from langchain_core.runnables.base import RunnableLike
 
+from langgraph.channels.any_value import AnyValue
 from langgraph.channels.base import BaseChannel, InvalidUpdateError
 from langgraph.channels.binop import BinaryOperatorAggregate
+from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue
 from langgraph.checkpoint import BaseCheckpointSaver
 from langgraph.graph.graph import END, Graph
@@ -71,11 +73,12 @@ class StateGraph(Graph):
             for key, node in self.nodes.items()
         }
         node_inboxes = {
-            # we can take any value written to channel because all writers
-            # write the entire state as of that step, which is equal for all writers
-            f"{key}:inbox": LastValue(Any, guard=False)
+            # we take any value written to channel because all writers
+            # write the entire state as of that step, which is equal for all
+            f"{key}:inbox": AnyValue(Any)
             for key in self.nodes
         }
+        node_outboxes = {key: EphemeralValue(Any) for key in self.nodes}
 
         for key in self.nodes:
             outgoing = outgoing_edges[key]
@@ -103,7 +106,7 @@ class StateGraph(Graph):
 
         return Pregel(
             nodes=nodes,
-            channels={**self.channels, **node_inboxes},
+            channels={**self.channels, **node_inboxes, **node_outboxes},
             input=f"{START}:inbox",
             output=END,
             hidden=[f"{node}:inbox" for node in self.nodes] + [START] + state_keys,