Source code for langchain.vectorstores.tair
from __future__ import annotations
import json
import logging
import uuid
from typing import Any, Iterable, List, Optional, Type
from langchain.docstore.document import Document
from langchain.schema.embeddings import Embeddings
from langchain.schema.vectorstore import VectorStore
from langchain.utils import get_from_dict_or_env
logger = logging.getLogger(__name__)
def _uuid_key() -> str:
    return uuid.uuid4().hex
[docs]class Tair(VectorStore):
    """`Tair` vector store."""
[docs]    def __init__(
        self,
        embedding_function: Embeddings,
        url: str,
        index_name: str,
        content_key: str = "content",
        metadata_key: str = "metadata",
        search_params: Optional[dict] = None,
        **kwargs: Any,
    ):
        self.embedding_function = embedding_function
        self.index_name = index_name
        try:
            from tair import Tair as TairClient
        except ImportError:
            raise ImportError(
                "Could not import tair python package. "
                "Please install it with `pip install tair`."
            )
        try:
            # connect to tair from url
            client = TairClient.from_url(url, **kwargs)
        except ValueError as e:
            raise ValueError(f"Tair failed to connect: {e}")
        self.client = client
        self.content_key = content_key
        self.metadata_key = metadata_key
        self.search_params = search_params
    @property
    def embeddings(self) -> Embeddings:
        return self.embedding_function
[docs]    def create_index_if_not_exist(
        self,
        dim: int,
        distance_type: str,