Tair

TairVectorStore

Bases: BasePydanticVectorStore

初始化TairVectorStore。

有两种索引类型可用：FLAT和HNSW。

HNSW的索引参数包括： - ef_construct - M - ef_search

这些参数的详细信息可以在这里找到： https://www.alibabacloud.com/help/en/tair/latest/tairvector#section-c76-ull-5mk

Parameters:

Name	Type	Description	Default
index_name	str	索引的名称。	required
index_type	str	索引的类型。默认为'HNSW'。	'HNSW'
index_args	Dict[str, Any]	索引的参数。默认为None。	None
tair_url	str	Tair实例的URL。	required
overwrite	bool	如果索引已经存在，是否覆盖。默认为False。	False
kwargs	Any	传递给Tair客户端的额外参数。	{}

异常

ValueError: 如果未安装tair-py ValueError: 如果连接到Tair实例失败

示例

pip install llama-index-vector-stores-tair

from llama_index.core.vector_stores.tair import TairVectorStore

# 创建一个TairVectorStore
vector_store = TairVectorStore(
    tair_url="redis://{username}:{password}@r-bp****************.redis.rds.aliyuncs.com:{port}",
    index_name="my_index",
    index_type="HNSW",
    index_args={"M": 16, "ef_construct": 200},
    overwrite=True
)

Source code in llama_index/vector_stores/tair/base.py

class TairVectorStore(BasePydanticVectorStore):
    """初始化TairVectorStore。

    有两种索引类型可用：FLAT和HNSW。

    HNSW的索引参数包括：
        - ef_construct
        - M
        - ef_search

    这些参数的详细信息可以在这里找到：
    https://www.alibabacloud.com/help/en/tair/latest/tairvector#section-c76-ull-5mk

    Args:
        index_name (str): 索引的名称。
        index_type (str): 索引的类型。默认为'HNSW'。
        index_args (Dict[str, Any]): 索引的参数。默认为None。
        tair_url (str): Tair实例的URL。
        overwrite (bool): 如果索引已经存在，是否覆盖。默认为False。
        kwargs (Any): 传递给Tair客户端的额外参数。

    异常:
        ValueError: 如果未安装tair-py
        ValueError: 如果连接到Tair实例失败

    示例:
        `pip install llama-index-vector-stores-tair`

        ```python
        from llama_index.core.vector_stores.tair import TairVectorStore

        # 创建一个TairVectorStore
        vector_store = TairVectorStore(
            tair_url="redis://{username}:{password}@r-bp****************.redis.rds.aliyuncs.com:{port}",
            index_name="my_index",
            index_type="HNSW",
            index_args={"M": 16, "ef_construct": 200},
            overwrite=True
        )
        ```"""

    stores_text = True
    stores_node = True
    flat_metadata = False

    _tair_client: Tair = PrivateAttr()
    _index_name: str = PrivateAttr()
    _index_type: str = PrivateAttr()
    _metric_type: str = PrivateAttr()
    _overwrite: bool = PrivateAttr()
    _index_args: Dict[str, Any] = PrivateAttr()
    _query_args: Dict[str, Any] = PrivateAttr()
    _dim: int = PrivateAttr()

    def __init__(
        self,
        tair_url: str,
        index_name: str,
        index_type: str = "HNSW",
        index_args: Optional[Dict[str, Any]] = None,
        overwrite: bool = False,
        **kwargs: Any,
    ) -> None:
        try:
            self._tair_client = Tair.from_url(tair_url, **kwargs)
        except ValueError as e:
            raise ValueError(f"Tair failed to connect: {e}")

        # index identifiers
        self._index_name = index_name
        self._index_type = index_type
        self._metric_type = "L2"
        self._overwrite = overwrite
        self._index_args = {}
        self._query_args = {}
        if index_type == "HNSW":
            if index_args is not None:
                ef_construct = index_args.get("ef_construct", 500)
                M = index_args.get("M", 24)
                ef_search = index_args.get("ef_search", 400)
            else:
                ef_construct = 500
                M = 24
                ef_search = 400

            self._index_args = {"ef_construct": ef_construct, "M": M}
            self._query_args = {"ef_search": ef_search}

    @classmethod
    def class_name(cls) -> str:
        """类名。"""
        return "TairVectorStore"

    @property
    def client(self) -> "Tair":
        """返回Tair客户端实例。"""
        return self._tair_client

    def add(self, nodes: List[BaseNode], **add_kwargs: Any) -> List[str]:
        """将节点添加到索引中。

Args:
    节点（List[BaseNode]）：带有嵌入的节点列表

Returns:
    List[str]：添加到索引中的文档的id列表。
"""
        # check to see if empty document list was passed
        if len(nodes) == 0:
            return []

        # set vector dim for creation if index doesn't exist
        self._dim = len(nodes[0].get_embedding())

        if self._index_exists():
            if self._overwrite:
                self.delete_index()
                self._create_index()
            else:
                logging.info(f"Adding document to existing index {self._index_name}")
        else:
            self._create_index()

        ids = []
        for node in nodes:
            attributes = {
                "id": node.node_id,
                "doc_id": node.ref_doc_id,
                "text": node.get_content(metadata_mode=MetadataMode.NONE),
            }
            metadata_dict = node_to_metadata_dict(
                node, remove_text=True, flat_metadata=self.flat_metadata
            )
            attributes.update(metadata_dict)

            ids.append(node.node_id)
            self._tair_client.tvs_hset(
                self._index_name,
                f"{node.ref_doc_id}#{node.node_id}",
                vector=node.get_embedding(),
                is_binary=False,
                **attributes,
            )

        _logger.info(f"Added {len(ids)} documents to index {self._index_name}")
        return ids

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """删除一个文档。

Args:
    doc_id (str): 文档id
"""
        iter = self._tair_client.tvs_scan(self._index_name, "%s#*" % ref_doc_id)
        for k in iter:
            self._tair_client.tvs_del(self._index_name, k)

    def delete_index(self) -> None:
        """删除索引和所有文档。"""
        _logger.info(f"Deleting index {self._index_name}")
        self._tair_client.tvs_del_index(self._index_name)

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询索引。

Args:
    query（VectorStoreQuery）：查询对象

Returns:
    VectorStoreQueryResult：查询结果

引发：
    ValueError：如果query.query_embedding为None。
"""
        filter_expr = None
        if query.filters is not None:
            filter_expr = _to_filter_expr(query.filters)

        if not query.query_embedding:
            raise ValueError("Query embedding is required for querying.")

        _logger.info(f"Querying index {self._index_name}")

        query_args = self._query_args
        if self._index_type == "HNSW" and "ef_search" in kwargs:
            query_args["ef_search"] = kwargs["ef_search"]

        results = self._tair_client.tvs_knnsearch(
            self._index_name,
            query.similarity_top_k,
            query.query_embedding,
            False,
            filter_str=filter_expr,
            **query_args,
        )
        results = [(k.decode(), float(s)) for k, s in results]

        ids = []
        nodes = []
        scores = []
        pipe = self._tair_client.pipeline(transaction=False)
        for key, score in results:
            scores.append(score)
            pipe.tvs_hmget(self._index_name, key, "id", "doc_id", "text")
        metadatas = pipe.execute()
        for i, m in enumerate(metadatas):
            # TODO: properly get the _node_conent
            doc_id = m[0].decode()
            node = TextNode(
                text=m[2].decode(),
                id_=doc_id,
                embedding=None,
                relationships={
                    NodeRelationship.SOURCE: RelatedNodeInfo(node_id=m[1].decode())
                },
            )
            ids.append(doc_id)
            nodes.append(node)
        _logger.info(f"Found {len(nodes)} results for query with id {ids}")

        return VectorStoreQueryResult(nodes=nodes, ids=ids, similarities=scores)

    def _create_index(self) -> None:
        _logger.info(f"Creating index {self._index_name}")
        self._tair_client.tvs_create_index(
            self._index_name,
            self._dim,
            distance_type=self._metric_type,
            index_type=self._index_type,
            data_type=tairvector.DataType.Float32,
            **self._index_args,
        )

    def _index_exists(self) -> bool:
        index = self._tair_client.tvs_get_index(self._index_name)
        return index is not None

client property

client: Tair

返回Tair客户端实例。

class_name classmethod

class_name() -> str

类名。

Source code in llama_index/vector_stores/tair/base.py

@classmethod
def class_name(cls) -> str:
    """类名。"""
    return "TairVectorStore"

add

add(nodes: List[BaseNode], **add_kwargs: Any) -> List[str]

将节点添加到索引中。

Returns:

Type	Description
List[str]	List[str]：添加到索引中的文档的id列表。

Source code in llama_index/vector_stores/tair/base.py

    def add(self, nodes: List[BaseNode], **add_kwargs: Any) -> List[str]:
        """将节点添加到索引中。

Args:
    节点（List[BaseNode]）：带有嵌入的节点列表

Returns:
    List[str]：添加到索引中的文档的id列表。
"""
        # check to see if empty document list was passed
        if len(nodes) == 0:
            return []

        # set vector dim for creation if index doesn't exist
        self._dim = len(nodes[0].get_embedding())

        if self._index_exists():
            if self._overwrite:
                self.delete_index()
                self._create_index()
            else:
                logging.info(f"Adding document to existing index {self._index_name}")
        else:
            self._create_index()

        ids = []
        for node in nodes:
            attributes = {
                "id": node.node_id,
                "doc_id": node.ref_doc_id,
                "text": node.get_content(metadata_mode=MetadataMode.NONE),
            }
            metadata_dict = node_to_metadata_dict(
                node, remove_text=True, flat_metadata=self.flat_metadata
            )
            attributes.update(metadata_dict)

            ids.append(node.node_id)
            self._tair_client.tvs_hset(
                self._index_name,
                f"{node.ref_doc_id}#{node.node_id}",
                vector=node.get_embedding(),
                is_binary=False,
                **attributes,
            )

        _logger.info(f"Added {len(ids)} documents to index {self._index_name}")
        return ids

delete

delete(ref_doc_id: str, **delete_kwargs: Any) -> None

删除一个文档。

Parameters:

Name	Type	Description	Default
doc_id	str	文档id	required

Source code in llama_index/vector_stores/tair/base.py

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """删除一个文档。

Args:
    doc_id (str): 文档id
"""
        iter = self._tair_client.tvs_scan(self._index_name, "%s#*" % ref_doc_id)
        for k in iter:
            self._tair_client.tvs_del(self._index_name, k)

delete_index

delete_index() -> None

删除索引和所有文档。

Source code in llama_index/vector_stores/tair/base.py

def delete_index(self) -> None:
    """删除索引和所有文档。"""
    _logger.info(f"Deleting index {self._index_name}")
    self._tair_client.tvs_del_index(self._index_name)

query

query(
    query: VectorStoreQuery, **kwargs: Any
) -> VectorStoreQueryResult

查询索引。

Returns:

Type	Description
VectorStoreQueryResult	VectorStoreQueryResult：查询结果

引发： ValueError：如果query.query_embedding为None。

Source code in llama_index/vector_stores/tair/base.py

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询索引。

Args:
    query（VectorStoreQuery）：查询对象

Returns:
    VectorStoreQueryResult：查询结果

引发：
    ValueError：如果query.query_embedding为None。
"""
        filter_expr = None
        if query.filters is not None:
            filter_expr = _to_filter_expr(query.filters)

        if not query.query_embedding:
            raise ValueError("Query embedding is required for querying.")

        _logger.info(f"Querying index {self._index_name}")

        query_args = self._query_args
        if self._index_type == "HNSW" and "ef_search" in kwargs:
            query_args["ef_search"] = kwargs["ef_search"]

        results = self._tair_client.tvs_knnsearch(
            self._index_name,
            query.similarity_top_k,
            query.query_embedding,
            False,
            filter_str=filter_expr,
            **query_args,
        )
        results = [(k.decode(), float(s)) for k, s in results]

        ids = []
        nodes = []
        scores = []
        pipe = self._tair_client.pipeline(transaction=False)
        for key, score in results:
            scores.append(score)
            pipe.tvs_hmget(self._index_name, key, "id", "doc_id", "text")
        metadatas = pipe.execute()
        for i, m in enumerate(metadatas):
            # TODO: properly get the _node_conent
            doc_id = m[0].decode()
            node = TextNode(
                text=m[2].decode(),
                id_=doc_id,
                embedding=None,
                relationships={
                    NodeRelationship.SOURCE: RelatedNodeInfo(node_id=m[1].decode())
                },
            )
            ids.append(doc_id)
            nodes.append(node)
        _logger.info(f"Found {len(nodes)} results for query with id {ids}")

        return VectorStoreQueryResult(nodes=nodes, ids=ids, similarities=scores)