Add fast path to serialize None values (#4103)

- If the value to serialize is None we can use encode it in the string
type, and skip msgpack encoding
- Use None value for edge/branch channels in StateGraph

libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py

@@ -30,6 +30,7 @@ from langgraph.checkpoint.serde.types import SendProtocol
 from langgraph.store.base import Item
 
 LC_REVIVER = Reviver()
+EMPTY_BYTES = b""
 
 
 class JsonPlusSerializer(SerializerProtocol):
@@ -194,7 +195,9 @@ class JsonPlusSerializer(SerializerProtocol):
         )
 
     def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
-        if isinstance(obj, bytes):
+        if obj is None:
+            return "null", EMPTY_BYTES
+        elif isinstance(obj, bytes):
             return "bytes", obj
         elif isinstance(obj, bytearray):
             return "bytearray", obj
@@ -211,7 +214,9 @@ class JsonPlusSerializer(SerializerProtocol):
 
     def loads_typed(self, data: tuple[str, bytes]) -> Any:
         type_, data_ = data
-        if type_ == "bytes":
+        if type_ == "null":
+            return None
+        elif type_ == "bytes":
             return data_
         elif type_ == "bytearray":
             return bytearray(data_)

libs/langgraph/langgraph/channels/any_value.py

@@ -1,4 +1,4 @@
-from typing import Any, Generic, Optional, Sequence, Type
+from typing import Any, Generic, Sequence, Type
 
 from typing_extensions import Self
 
@@ -30,10 +30,10 @@ class AnyValue(Generic[Value], BaseChannel[Value, Value, Value]):
         """The type of the update received by the channel."""
         return self.typ
 
-    def from_checkpoint(self, checkpoint: Optional[Value]) -> Self:
+    def from_checkpoint(self, checkpoint: Value) -> Self:
         empty = self.__class__(self.typ)
         empty.key = self.key
-        if checkpoint is not None:
+        if checkpoint is not MISSING:
             empty.value = checkpoint
         return empty
 

libs/langgraph/langgraph/channels/base.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import Any, Generic, Optional, Sequence, TypeVar
+from typing import Any, Generic, Sequence, TypeVar
 
 from typing_extensions import Self
 
@@ -29,14 +29,14 @@ class BaseChannel(Generic[Value, Update, C], ABC):
 
     # serialize/deserialize methods
 
-    def checkpoint(self) -> Optional[C]:
+    def checkpoint(self) -> C:
         """Return a serializable representation of the channel's current state.
         Raises EmptyChannelError if the channel is empty (never updated yet),
         or doesn't support checkpoints."""
         return self.get()
 
     @abstractmethod
-    def from_checkpoint(self, checkpoint: Optional[C]) -> Self:
+    def from_checkpoint(self, checkpoint: C) -> Self:
         """Return a new identical channel, optionally initialized from a checkpoint.
         If the checkpoint contains complex data structures, they should be copied."""
 

libs/langgraph/langgraph/channels/binop.py

@@ -1,11 +1,5 @@
 import collections.abc
-from typing import (
-    Callable,
-    Generic,
-    Optional,
-    Sequence,
-    Type,
-)
+from typing import Callable, Generic, Sequence, Type
 
 from typing_extensions import NotRequired, Required, Self
 
@@ -72,10 +66,10 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]):
         """The type of the update received by the channel."""
         return self.typ
 
-    def from_checkpoint(self, checkpoint: Optional[Value]) -> Self:
+    def from_checkpoint(self, checkpoint: Value) -> Self:
         empty = self.__class__(self.typ, self.operator)
         empty.key = self.key
-        if checkpoint is not None:
+        if checkpoint is not MISSING:
             empty.value = checkpoint
         return empty
 

libs/langgraph/langgraph/channels/dynamic_barrier_value.py

@@ -3,6 +3,7 @@ from typing import Any, Generic, NamedTuple, Optional, Sequence, Type, Union
 from typing_extensions import Self
 
 from langgraph.channels.base import BaseChannel, Value
+from langgraph.constants import MISSING
 from langgraph.errors import EmptyChannelError, InvalidUpdateError
 
 
@@ -49,12 +50,11 @@ class DynamicBarrierValue(
         return (self.names, self.seen)
 
     def from_checkpoint(
-        self,
-        checkpoint: Optional[tuple[Optional[set[Value]], set[Value]]],
+        self, checkpoint: tuple[Optional[set[Value]], set[Value]]
     ) -> Self:
         empty = self.__class__(self.typ)
         empty.key = self.key
-        if checkpoint is not None:
+        if checkpoint is not MISSING:
             names, seen = checkpoint
             empty.names = names if names is not None else None
             empty.seen = seen

libs/langgraph/langgraph/channels/ephemeral_value.py

@@ -1,4 +1,4 @@
-from typing import Any, Generic, Optional, Sequence, Type
+from typing import Any, Generic, Sequence, Type
 
 from typing_extensions import Self
 
@@ -30,10 +30,10 @@ class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]):
         """The type of the update received by the channel."""
         return self.typ
 
-    def from_checkpoint(self, checkpoint: Optional[Value]) -> Self:
+    def from_checkpoint(self, checkpoint: Value) -> Self:
         empty = self.__class__(self.typ, self.guard)
         empty.key = self.key
-        if checkpoint is not None:
+        if checkpoint is not MISSING:
             empty.value = checkpoint
         return empty
 

libs/langgraph/langgraph/channels/last_value.py

@@ -1,4 +1,4 @@
-from typing import Any, Generic, Optional, Sequence, Type
+from typing import Any, Generic, Sequence, Type
 
 from typing_extensions import Self
 
@@ -34,10 +34,10 @@ class LastValue(Generic[Value], BaseChannel[Value, Value, Value]):
         """The type of the update received by the channel."""
         return self.typ
 
-    def from_checkpoint(self, checkpoint: Optional[Value]) -> Self:
+    def from_checkpoint(self, checkpoint: Value) -> Self:
         empty = self.__class__(self.typ)
         empty.key = self.key
-        if checkpoint is not None:
+        if checkpoint is not MISSING:
             empty.value = checkpoint
         return empty
 

libs/langgraph/langgraph/channels/named_barrier_value.py

@@ -1,8 +1,9 @@
-from typing import Generic, Optional, Sequence, Type
+from typing import Generic, 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
 
 
@@ -35,10 +36,10 @@ class NamedBarrierValue(Generic[Value], BaseChannel[Value, Value, set[Value]]):
     def checkpoint(self) -> set[Value]:
         return self.seen
 
-    def from_checkpoint(self, checkpoint: Optional[set[Value]]) -> Self:
+    def from_checkpoint(self, checkpoint: set[Value]) -> Self:
         empty = self.__class__(self.typ, self.names)
         empty.key = self.key
-        if checkpoint is not None:
+        if checkpoint is not MISSING:
             empty.seen = checkpoint
         return empty
 

libs/langgraph/langgraph/channels/topic.py

@@ -1,8 +1,9 @@
-from typing import Any, Generic, Iterator, Optional, Sequence, Type, Union
+from typing import Any, Generic, Iterator, Sequence, Type, Union
 
 from typing_extensions import Self
 
 from langgraph.channels.base import BaseChannel, Value
+from langgraph.constants import MISSING
 from langgraph.errors import EmptyChannelError
 
 
@@ -16,9 +17,7 @@ def flatten(values: Sequence[Union[Value, list[Value]]]) -> Iterator[Value]:
 
 class Topic(
     Generic[Value],
-    BaseChannel[
-        Sequence[Value], Union[Value, list[Value]], tuple[set[Value], list[Value]]
-    ],
+    BaseChannel[Sequence[Value], Union[Value, list[Value]], list[Value]],
 ):
     """A configurable PubSub Topic.
 
@@ -49,14 +48,15 @@ class Topic(
         """The type of the update received by the channel."""
         return Union[self.typ, list[self.typ]]  # type: ignore[name-defined]
 
-    def checkpoint(self) -> tuple[set[Value], list[Value]]:
+    def checkpoint(self) -> list[Value]:
         return self.values
 
-    def from_checkpoint(self, checkpoint: Optional[list[Value]]) -> Self:
+    def from_checkpoint(self, checkpoint: list[Value]) -> Self:
         empty = self.__class__(self.typ, self.accumulate)
         empty.key = self.key
-        if checkpoint is not None:
+        if checkpoint is not MISSING:
             if isinstance(checkpoint, tuple):
+                # backwards compatibility
                 empty.values = checkpoint[1]
             else:
                 empty.values = checkpoint

libs/langgraph/langgraph/channels/untracked_value.py

@@ -1,4 +1,4 @@
-from typing import Generic, Optional, Sequence, Type
+from typing import Generic, Sequence, Type
 
 from typing_extensions import Self
 
@@ -33,7 +33,7 @@ class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]):
     def checkpoint(self) -> Value:
         raise EmptyChannelError()
 
-    def from_checkpoint(self, checkpoint: Optional[Value]) -> Self:
+    def from_checkpoint(self, checkpoint: Value) -> Self:
         empty = self.__class__(self.typ, self.guard)
         empty.key = self.key
         return empty

libs/langgraph/langgraph/graph/state.py

@@ -845,7 +845,7 @@ class CompiledStateGraph(CompiledGraph):
             if end != END:
                 self.nodes[starts].writers.append(
                     ChannelWrite(
-                        (ChannelWriteEntry(CHANNEL_BRANCH_TO.format(end), starts),)
+                        (ChannelWriteEntry(CHANNEL_BRANCH_TO.format(end), None),)
                     )
                 )
         elif end != END:
@@ -871,7 +871,7 @@ class CompiledStateGraph(CompiledGraph):
             if filtered := [p for p in packets if p != END]:
                 writes = [
                     (
-                        ChannelWriteEntry(CHANNEL_BRANCH_TO.format(p), start)
+                        ChannelWriteEntry(CHANNEL_BRANCH_TO.format(p), None)
                         if not isinstance(p, Send)
                         else p
                     )

libs/langgraph/langgraph/pregel/manager.py

@@ -4,6 +4,7 @@ from typing import AsyncIterator, Iterator, Mapping, Union
 
 from langgraph.channels.base import BaseChannel
 from langgraph.checkpoint.base import Checkpoint
+from langgraph.constants import MISSING
 from langgraph.managed.base import (
     ConfiguredManagedValue,
     ManagedValueMapping,
@@ -36,7 +37,7 @@ def ChannelsManager(
     with ExitStack() as stack:
         yield (
             {
-                k: v.from_checkpoint(checkpoint["channel_values"].get(k))
+                k: v.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
                 for k, v in channel_specs.items()
             },
             ManagedValueMapping(
@@ -90,7 +91,7 @@ async def AsyncChannelsManager(
         yield (
             # channels: enter each channel with checkpoint
             {
-                k: v.from_checkpoint(checkpoint["channel_values"].get(k))
+                k: v.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
                 for k, v in channel_specs.items()
             },
             # managed: build mapping from spec to result

libs/langgraph/tests/test_channels.py

@@ -6,13 +6,14 @@ import pytest
 from langgraph.channels.binop import BinaryOperatorAggregate
 from langgraph.channels.last_value import LastValue
 from langgraph.channels.topic import Topic
+from langgraph.constants import MISSING
 from langgraph.errors import EmptyChannelError, InvalidUpdateError
 
 pytestmark = pytest.mark.anyio
 
 
 def test_last_value() -> None:
-    channel = LastValue(int).from_checkpoint(None)
+    channel = LastValue(int).from_checkpoint(MISSING)
     assert channel.ValueType is int
     assert channel.UpdateType is int
 
@@ -31,7 +32,7 @@ def test_last_value() -> None:
 
 
 def test_topic() -> None:
-    channel = Topic(str).from_checkpoint(None)
+    channel = Topic(str).from_checkpoint(MISSING)
     assert channel.ValueType is Sequence[str]
     assert channel.UpdateType is Union[str, list[str]]
 
@@ -55,7 +56,7 @@ def test_topic() -> None:
 
 
 def test_topic_accumulate() -> None:
-    channel = Topic(str, accumulate=True).from_checkpoint(None)
+    channel = Topic(str, accumulate=True).from_checkpoint(MISSING)
     assert channel.ValueType is Sequence[str]
     assert channel.UpdateType is Union[str, list[str]]
 
@@ -73,7 +74,7 @@ def test_topic_accumulate() -> None:
 
 
 def test_binop() -> None:
-    channel = BinaryOperatorAggregate(int, operator.add).from_checkpoint(None)
+    channel = BinaryOperatorAggregate(int, operator.add).from_checkpoint(MISSING)
     assert channel.ValueType is int
     assert channel.UpdateType is int
 

libs/langgraph/tests/test_pregel.py

@@ -1310,8 +1310,8 @@ def test_pending_writes_resume(
             },
             "channel_values": {
                 "value": 1,
-                "branch:to:one": "__start__",
-                "branch:to:two": "__start__",
+                "branch:to:one": None,
+                "branch:to:two": None,
             },
         },
         metadata={
@@ -1363,8 +1363,8 @@ def test_pending_writes_resume(
         parent_config=None,
         pending_writes=UnsortedSequence(
             (AnyStr(), "value", 1),
-            (AnyStr(), "branch:to:one", "__start__"),
-            (AnyStr(), "branch:to:two", "__start__"),
+            (AnyStr(), "branch:to:one", None),
+            (AnyStr(), "branch:to:two", None),
         ),
     )
 

libs/langgraph/tests/test_pregel_async.py

@@ -2146,8 +2146,8 @@ async def test_pending_writes_resume(
                 },
                 "channel_values": {
                     "value": 1,
-                    "branch:to:one": "__start__",
-                    "branch:to:two": "__start__",
+                    "branch:to:one": None,
+                    "branch:to:two": None,
                 },
             },
             metadata={
@@ -2201,8 +2201,8 @@ async def test_pending_writes_resume(
             parent_config=None,
             pending_writes=UnsortedSequence(
                 (AnyStr(), "value", 1),
-                (AnyStr(), "branch:to:one", "__start__"),
-                (AnyStr(), "branch:to:two", "__start__"),
+                (AnyStr(), "branch:to:one", None),
+                (AnyStr(), "branch:to:two", None),
             ),
         )