codekingpro/portable-devtools

1from __future__ import annotations2 3from collections.abc import Sequence4from typing import Any, Generic5 6from typing_extensions import Self7 8from langgraph._internal._typing import MISSING9from langgraph.channels.base import BaseChannel, Value10from langgraph.errors import EmptyChannelError, InvalidUpdateError11 12__all__ = ("UntrackedValue",)13 14 15class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]):16    """Stores the last value received, never checkpointed."""17 18    __slots__ = ("value", "guard")19 20    guard: bool21    value: Value | Any22 23    def __init__(self, typ: type[Value], guard: bool = True) -> None:24        super().__init__(typ)25        self.guard = guard26        self.value = MISSING27 28    def __eq__(self, value: object) -> bool:29        return isinstance(value, UntrackedValue) and value.guard == self.guard30 31    @property32    def ValueType(self) -> type[Value]:33        """The type of the value stored in the channel."""34        return self.typ35 36    @property37    def UpdateType(self) -> type[Value]:38        """The type of the update received by the channel."""39        return self.typ40 41    def copy(self) -> Self:42        """Return a copy of the channel."""43        empty = self.__class__(self.typ, self.guard)44        empty.key = self.key45        empty.value = self.value46        return empty47 48    def checkpoint(self) -> Value | Any:49        return MISSING50 51    def from_checkpoint(self, checkpoint: Value) -> Self:52        empty = self.__class__(self.typ, self.guard)53        empty.key = self.key54        return empty55 56    def update(self, values: Sequence[Value]) -> bool:57        if len(values) == 0:58            return False59        if len(values) != 1 and self.guard:60            raise InvalidUpdateError(61                f"At key '{self.key}': UntrackedValue(guard=True) can receive only one value per step. Use guard=False if you want to store any one of multiple values."62            )63 64        self.value = values[-1]65        return True66 67    def get(self) -> Value:68        if self.value is MISSING:69            raise EmptyChannelError()70        return self.value71 72    def is_available(self) -> bool:73        return self.value is not MISSING74