BaseStore Class — langchain Architecture

Source Code

libs/core/langchain_core/stores.py lines 26–170

class BaseStore(ABC, Generic[K, V]):
    """Abstract interface for a key-value store.

    This is an interface that's meant to abstract away the details of different
    key-value stores. It provides a simple interface for getting, setting, and deleting
    key-value pairs.

    The basic methods are `mget`, `mset`, and `mdelete` for getting, setting, and
    deleting multiple key-value pairs at once. The `yield_keys` method is used to
    iterate over keys that match a given prefix.

    The async versions of these methods are also provided, which are meant to be used in
    async contexts. The async methods are named with an `a` prefix, e.g., `amget`,
    `amset`, `amdelete`, and `ayield_keys`.

    By default, the `amget`, `amset`, `amdelete`, and `ayield_keys` methods are
    implemented using the synchronous methods. If the store can natively support async
    operations, it should override these methods.

    By design the methods only accept batches of keys and values, and not single keys or
    values. This is done to force user code to work with batches which will usually be
    more efficient by saving on round trips to the store.

    Examples:
        ```python
        from langchain.storage import BaseStore


        class MyInMemoryStore(BaseStore[str, int]):
            def __init__(self) -> None:
                self.store: dict[str, int] = {}

            def mget(self, keys: Sequence[str]) -> list[int | None]:
                return [self.store.get(key) for key in keys]

            def mset(self, key_value_pairs: Sequence[tuple[str, int]]) -> None:
                for key, value in key_value_pairs:
                    self.store[key] = value

            def mdelete(self, keys: Sequence[str]) -> None:
                for key in keys:
                    if key in self.store:
                        del self.store[key]

            def yield_keys(self, prefix: str | None = None) -> Iterator[str]:
                if prefix is None:
                    yield from self.store.keys()
                else:
                    for key in self.store.keys():
                        if key.startswith(prefix):
                            yield key
        ```
    """

    @abstractmethod
    def mget(self, keys: Sequence[K]) -> list[V | None]:
        """Get the values associated with the given keys.

        Args:
            keys: A sequence of keys.

        Returns:
            A sequence of optional values associated with the keys.
                If a key is not found, the corresponding value will be `None`.
        """

    async def amget(self, keys: Sequence[K]) -> list[V | None]:
        """Async get the values associated with the given keys.

        Args:
            keys: A sequence of keys.

        Returns:
            A sequence of optional values associated with the keys.
                If a key is not found, the corresponding value will be `None`.
        """
        return await run_in_executor(None, self.mget, keys)

    @abstractmethod
    def mset(self, key_value_pairs: Sequence[tuple[K, V]]) -> None:
        """Set the values for the given keys.