Source code for langchain.vectorstores.redis.base
"""Wrapper around Redis vector database."""
from __future__ import annotations
import logging
import os
import uuid
from typing import (
    TYPE_CHECKING,
    Any,
    Callable,
    Dict,
    Iterable,
    List,
    Mapping,
    Optional,
    Tuple,
    Type,
    Union,
    cast,
)
import numpy as np
import yaml
from langchain._api import deprecated
from langchain.callbacks.manager import (
    AsyncCallbackManagerForRetrieverRun,
    CallbackManagerForRetrieverRun,
)
from langchain.docstore.document import Document
from langchain.schema.embeddings import Embeddings
from langchain.schema.vectorstore import VectorStore, VectorStoreRetriever
from langchain.utilities.redis import (
    _array_to_buffer,
    _buffer_to_array,
    check_redis_module_exist,
    get_client,
)
from langchain.utils import get_from_dict_or_env
from langchain.vectorstores.redis.constants import (
    REDIS_REQUIRED_MODULES,
    REDIS_TAG_SEPARATOR,
)
from langchain.vectorstores.utils import maximal_marginal_relevance
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
    from redis.client import Redis as RedisType
    from redis.commands.search.query import Query
    from langchain.vectorstores.redis.filters import RedisFilterExpression
    from langchain.vectorstores.redis.schema import RedisModel
def _redis_key(prefix: str) -> str:
    """Redis key schema for a given prefix."""
    return f"{prefix}:{uuid.uuid4().hex}"
def _redis_prefix(index_name: str) -> str:
    """Redis key prefix for a given index."""
    return f"doc:{index_name}"
def _default_relevance_score(val: float) -> float: