stores.py — langchain Source File

Source Code

"""**Store** implements the key-value stores and storage helpers.

Module provides implementations of various key-value stores that conform
to a simple key-value interface.

The primary goal of these storages is to support implementation of caching.
"""

from abc import ABC, abstractmethod
from collections.abc import AsyncIterator, Iterator, Sequence
from typing import (
    Any,
    Generic,
    TypeVar,
)

from typing_extensions import override

from langchain_core.exceptions import LangChainException
from langchain_core.runnables import run_in_executor

K = TypeVar("K")
V = TypeVar("V")


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]

// ... (232 more lines)