diff --git a/libs/langgraph/langgraph/channels/any_value.py b/libs/langgraph/langgraph/channels/any_value.py index 345daa7c7..e9bd85572 100644 --- a/libs/langgraph/langgraph/channels/any_value.py +++ b/libs/langgraph/langgraph/channels/any_value.py @@ -28,12 +28,6 @@ class AnyValue(Generic[Value], BaseChannel[Value, Value, Value]): """The type of the update received by the channel.""" return self.typ - def checkpoint(self) -> Value: - try: - return self.value - except AttributeError: - raise EmptyChannelError() - @contextmanager def from_checkpoint( self, checkpoint: Optional[Value], config: RunnableConfig diff --git a/libs/langgraph/langgraph/channels/base.py b/libs/langgraph/langgraph/channels/base.py index 885698743..955b6ab76 100644 --- a/libs/langgraph/langgraph/channels/base.py +++ b/libs/langgraph/langgraph/channels/base.py @@ -35,11 +35,11 @@ class BaseChannel(Generic[Value, Update, C], ABC): # serialize/deserialize methods - @abstractmethod def checkpoint(self) -> Optional[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() @contextmanager @abstractmethod diff --git a/libs/langgraph/langgraph/channels/binop.py b/libs/langgraph/langgraph/channels/binop.py index 72e76c0a7..02e33bd2e 100644 --- a/libs/langgraph/langgraph/channels/binop.py +++ b/libs/langgraph/langgraph/channels/binop.py @@ -73,12 +73,6 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]): """The type of the update received by the channel.""" return self.typ - def checkpoint(self) -> Value: - try: - return self.value - except AttributeError: - raise EmptyChannelError() - @contextmanager def from_checkpoint( self, checkpoint: Optional[Value], config: RunnableConfig diff --git a/libs/langgraph/langgraph/channels/ephemeral_value.py b/libs/langgraph/langgraph/channels/ephemeral_value.py index 4e7f2ed63..59e34a5f0 100644 --- a/libs/langgraph/langgraph/channels/ephemeral_value.py +++ b/libs/langgraph/langgraph/channels/ephemeral_value.py @@ -28,12 +28,6 @@ class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]): """The type of the update received by the channel.""" return self.typ - def checkpoint(self) -> Value: - try: - return self.value - except AttributeError: - raise EmptyChannelError() - @contextmanager def from_checkpoint( self, checkpoint: Optional[Value], config: RunnableConfig diff --git a/libs/langgraph/langgraph/channels/last_value.py b/libs/langgraph/langgraph/channels/last_value.py index e74580d6a..e5e59d111 100644 --- a/libs/langgraph/langgraph/channels/last_value.py +++ b/libs/langgraph/langgraph/channels/last_value.py @@ -27,12 +27,6 @@ class LastValue(Generic[Value], BaseChannel[Value, Value, Value]): """The type of the update received by the channel.""" return self.typ - def checkpoint(self) -> Value: - try: - return self.value - except AttributeError: - raise EmptyChannelError() - @contextmanager def from_checkpoint( self, checkpoint: Optional[Value], config: RunnableConfig diff --git a/libs/langgraph/langgraph/channels/topic.py b/libs/langgraph/langgraph/channels/topic.py index 5e7af6a26..7b4b0b27d 100644 --- a/libs/langgraph/langgraph/channels/topic.py +++ b/libs/langgraph/langgraph/channels/topic.py @@ -30,23 +30,15 @@ class Topic( accumulate: Whether to accumulate values across steps. If False, the channel will be emptied after each step. """ - def __init__( - self, typ: Type[Value], unique: bool = False, accumulate: bool = False - ) -> None: + def __init__(self, typ: Type[Value], accumulate: bool = False) -> None: # attrs self.typ = typ - self.unique = unique self.accumulate = accumulate # state - self.seen = set[Value]() self.values = list[Value]() def __eq__(self, value: object) -> bool: - return ( - isinstance(value, Topic) - and value.unique == self.unique - and value.accumulate == self.accumulate - ) + return isinstance(value, Topic) and value.accumulate == self.accumulate @property def ValueType(self) -> Any: @@ -59,18 +51,20 @@ class Topic( return Union[self.typ, list[self.typ]] # type: ignore[name-defined] def checkpoint(self) -> tuple[set[Value], list[Value]]: - return (self.seen, self.values) + return self.values @contextmanager def from_checkpoint( self, - checkpoint: Optional[tuple[set[Value], list[Value]]], + checkpoint: Optional[list[Value]], config: RunnableConfig, ) -> Generator[Self, None, None]: - empty = self.__class__(self.typ, self.unique, self.accumulate) + empty = self.__class__(self.typ, self.accumulate) if checkpoint is not None: - empty.seen = checkpoint[0].copy() - empty.values = checkpoint[1].copy() + if isinstance(checkpoint, tuple): + empty.values = checkpoint[1].copy() + else: + empty.values = checkpoint.copy() try: yield empty finally: @@ -81,13 +75,7 @@ class Topic( if not self.accumulate: self.values = list[Value]() if flat_values := flatten(values): - if self.unique: - for value in flat_values: - if value not in self.seen: - self.seen.add(value) - self.values.append(value) - else: - self.values.extend(flat_values) + self.values.extend(flat_values) return self.values != current def get(self) -> Sequence[Value]: diff --git a/libs/langgraph/tests/test_channels.py b/libs/langgraph/tests/test_channels.py index 3e0c924ae..ff393577e 100644 --- a/libs/langgraph/tests/test_channels.py +++ b/libs/langgraph/tests/test_channels.py @@ -91,50 +91,6 @@ async def test_topic_async() -> None: assert channel.get() == ["e"] -def test_topic_unique() -> None: - with Topic(str, unique=True).from_checkpoint(None, {}) as channel: - assert channel.ValueType is Sequence[str] - assert channel.UpdateType is Union[str, list[str]] - - assert channel.update(["a", "b"]) - assert channel.get() == ["a", "b"] - assert channel.update(["b", ["c", "d"], "d"]) - assert channel.get() == ["c", "d"], "de-dupes from current and previous steps" - assert channel.update([]) - with pytest.raises(EmptyChannelError): - channel.get() - assert not channel.update([]), "channel already empty" - assert channel.update(["e"]) - assert channel.get() == ["e"] - checkpoint = channel.checkpoint() - with Topic(str, unique=True).from_checkpoint(checkpoint, {}) as channel: - assert channel.get() == ["e"] - assert channel.update(["d", "f"]) - assert channel.get() == ["f"], "de-dupes from checkpoint" - - -async def test_topic_unique_async() -> None: - async with Topic(str, unique=True).afrom_checkpoint(None, {}) as channel: - assert channel.ValueType is Sequence[str] - assert channel.UpdateType is Union[str, list[str]] - - assert channel.update(["a", "b"]) - assert channel.get() == ["a", "b"] - assert channel.update(["b", ["c", "d"], "d"]) - assert channel.get() == ["c", "d"], "de-dupes from current and previous steps" - assert channel.update([]) - with pytest.raises(EmptyChannelError): - channel.get() - assert not channel.update([]), "channel already empty" - assert channel.update(["e"]) - assert channel.get() == ["e"] - checkpoint = channel.checkpoint() - async with Topic(str, unique=True).afrom_checkpoint(checkpoint, {}) as channel: - assert channel.get() == ["e"] - assert channel.update(["d", "f"]) - assert channel.get() == ["f"], "de-dupes from checkpoint" - - def test_topic_accumulate() -> None: with Topic(str, accumulate=True).from_checkpoint(None, {}) as channel: assert channel.ValueType is Sequence[str] @@ -171,49 +127,6 @@ async def test_topic_accumulate_async() -> None: assert channel.get() == ["a", "b", "b", "c", "d", "d", "e"] -def test_topic_unique_accumulate() -> None: - with Topic(str, unique=True, accumulate=True).from_checkpoint(None, {}) as channel: - assert channel.ValueType is Sequence[str] - assert channel.UpdateType is Union[str, list[str]] - - assert channel.update(["a", "b"]) - assert channel.get() == ["a", "b"] - assert channel.update(["b", ["c", "d"], "d"]) - assert channel.get() == ["a", "b", "c", "d"] - assert not channel.update(["c"]), "no new values" - assert not channel.update([]) - assert channel.get() == ["a", "b", "c", "d"] - checkpoint = channel.checkpoint() - with Topic(str, unique=True, accumulate=True).from_checkpoint( - checkpoint, {} - ) as channel: - assert channel.get() == ["a", "b", "c", "d"] - assert channel.update(["d", "e"]) - assert channel.get() == ["a", "b", "c", "d", "e"] - - -async def test_topic_unique_accumulate_async() -> None: - async with Topic(str, unique=True, accumulate=True).afrom_checkpoint( - None, {} - ) as channel: - assert channel.ValueType is Sequence[str] - assert channel.UpdateType is Union[str, list[str]] - - channel.update(["a", "b"]) - assert channel.get() == ["a", "b"] - channel.update(["b", ["c", "d"], "d"]) - assert channel.get() == ["a", "b", "c", "d"] - channel.update([]) - assert channel.get() == ["a", "b", "c", "d"] - checkpoint = channel.checkpoint() - async with Topic(str, unique=True, accumulate=True).afrom_checkpoint( - checkpoint, {} - ) as channel: - assert channel.get() == ["a", "b", "c", "d"] - channel.update(["d", "e"]) - assert channel.get() == ["a", "b", "c", "d", "e"] - - def test_binop() -> None: with BinaryOperatorAggregate(int, operator.add).from_checkpoint( None, {}