Source code for langchain_community.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_core._api import deprecated
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from langchain_core.documents import Document
from langchain_core.embeddings import Embeddings
from langchain_core.utils import get_from_dict_or_env
from langchain_core.vectorstores import VectorStore, VectorStoreRetriever
from langchain_community.utilities.redis import (
    _array_to_buffer,
    _buffer_to_array,
    check_redis_module_exist,
    get_client,
)
from langchain_community.vectorstores.redis.constants import (
    REDIS_REQUIRED_MODULES,
    REDIS_TAG_SEPARATOR,
)
from langchain_community.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_community.vectorstores.redis.filters import RedisFilterExpression
    from langchain_community.vectorstores.redis.schema import RedisModel
def _default_relevance_score(val: float) -> float:
    return 1 - val
[docs]def check_index_exists(client: RedisType, index_name: str) -> bool:
    """Check if Redis index exists."""
    try:
        client.ft(index_name).info()
    except:  # noqa: E722
        logger.debug("Index does not exist")
        return False