cache.py — langchain Source File

Source Code

"""Module contains code for a cache backed embedder.

The cache backed embedder is a wrapper around an embedder that caches
embeddings in a key-value store. The cache is used to avoid recomputing
embeddings for the same text.

The text is hashed and the hash is used as the key in the cache.
"""

from __future__ import annotations

import hashlib
import json
import uuid
import warnings
from collections.abc import Callable, Sequence
from typing import Literal, cast

from langchain_core.embeddings import Embeddings
from langchain_core.stores import BaseStore, ByteStore
from langchain_core.utils.iter import batch_iterate

from langchain_classic.storage.encoder_backed import EncoderBackedStore

NAMESPACE_UUID = uuid.UUID(int=1985)


def _sha1_hash_to_uuid(text: str) -> uuid.UUID:
    """Return a UUID derived from *text* using SHA-1 (deterministic).

    Deterministic and fast, **but not collision-resistant**.

    A malicious attacker could try to create two different texts that hash to the same
    UUID. This may not necessarily be an issue in the context of caching embeddings,
    but new applications should swap this out for a stronger hash function like
    xxHash, BLAKE2 or SHA-256, which are collision-resistant.
    """
    sha1_hex = hashlib.sha1(text.encode("utf-8"), usedforsecurity=False).hexdigest()
    # Embed the hex string in `uuid5` to obtain a valid UUID.
    return uuid.uuid5(NAMESPACE_UUID, sha1_hex)


def _make_default_key_encoder(namespace: str, algorithm: str) -> Callable[[str], str]:
    """Create a default key encoder function.

    Args:
        namespace: Prefix that segregates keys from different embedding models.
        algorithm:
           * `'sha1'` - fast but not collision-resistant
           * `'blake2b'` - cryptographically strong, faster than SHA-1
           * `'sha256'` - cryptographically strong, slower than SHA-1
           * `'sha512'` - cryptographically strong, slower than SHA-1

    Returns:
        A function that encodes a key using the specified algorithm.
    """
    if algorithm == "sha1":
        _warn_about_sha1_encoder()

    def _key_encoder(key: str) -> str:
// ... (311 more lines)