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__ = ("EphemeralValue",)13 14 15class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]):16    """Stores the value received in the step immediately preceding, clears after."""17 18    __slots__ = ("value", "guard")19 20    value: Value | Any21    guard: bool22 23    def __init__(self, typ: Any, 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, EphemeralValue) 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 from_checkpoint(self, checkpoint: Value) -> Self:49        empty = self.__class__(self.typ, self.guard)50        empty.key = self.key51        if checkpoint is not MISSING:52            empty.value = checkpoint53        return empty54 55    def update(self, values: Sequence[Value]) -> bool:56        if len(values) == 0:57            if self.value is not MISSING:58                self.value = MISSING59                return True60            else:61                return False62        if len(values) != 1 and self.guard:63            raise InvalidUpdateError(64                f"At key '{self.key}': EphemeralValue(guard=True) can receive only one value per step. Use guard=False if you want to store any one of multiple values."65            )66 67        self.value = values[-1]68        return True69 70    def get(self) -> Value:71        if self.value is MISSING:72            raise EmptyChannelError()73        return self.value74 75    def is_available(self) -> bool:76        return self.value is not MISSING77 78    def checkpoint(self) -> Value:79        return self.value80