Epsilla

EpsillaVectorStore

Bases: BasePydanticVectorStore

Epsilla向量存储。

在这个向量存储中，我们将文本、其嵌入和一些元数据存储在Epsilla集合中。这个实现允许使用已经存在的集合。它还支持在集合不存在或overwrite设置为True时创建一个新的集合。

作为先决条件，您需要安装pyepsilla包，并且有一个正在运行的Epsilla向量数据库（例如，通过我们的docker镜像）。请参阅以下文档，了解如何运行Epsilla向量数据库：https://epsilla-inc.gitbook.io/epsilladb/quick-start

Parameters:

Name	Type	Description	Default
client	Any	要连接的Epsilla客户端。	required
collection_name	Optional[str]	要使用的集合名称。默认为"llama_collection"。	'llama_collection'
db_path	Optional[str]	数据库将持久化的路径。默认为"/tmp/langchain-epsilla"。	DEFAULT_PERSIST_DIR
db_name	Optional[str]	给加载的数据库命名。默认为"langchain_store"。	'llama_db'
dimension	Optional[int]	嵌入的维度。如果未提供，将在第一次插入时创建集合。默认为None。	None
overwrite	Optional[bool]	是否覆盖同名的现有集合。默认为False。	False

Returns:

Name	Type	Description
EpsillaVectorStore		支持添加、删除和查询的向量存储。

示例： pip install llama-index-vector-stores-epsilla

```python
from llama_index.vector_stores.epsilla import EpsillaVectorStore
from pyepsilla import vectordb

client = vectordb.Client()
vector_store = EpsillaVectorStore(client=client, db_path="/tmp/llamastore")
```

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

class EpsillaVectorStore(BasePydanticVectorStore):
    """Epsilla向量存储。

在这个向量存储中，我们将文本、其嵌入和一些元数据存储在Epsilla集合中。这个实现允许使用已经存在的集合。它还支持在集合不存在或`overwrite`设置为True时创建一个新的集合。

作为先决条件，您需要安装``pyepsilla``包，并且有一个正在运行的Epsilla向量数据库（例如，通过我们的docker镜像）。请参阅以下文档，了解如何运行Epsilla向量数据库：https://epsilla-inc.gitbook.io/epsilladb/quick-start

Args:
    client (Any): 要连接的Epsilla客户端。
    collection_name (Optional[str]): 要使用的集合名称。默认为"llama_collection"。
    db_path (Optional[str]): 数据库将持久化的路径。默认为"/tmp/langchain-epsilla"。
    db_name (Optional[str]): 给加载的数据库命名。默认为"langchain_store"。
    dimension (Optional[int]): 嵌入的维度。如果未提供，将在第一次插入时创建集合。默认为None。
    overwrite (Optional[bool]): 是否覆盖同名的现有集合。默认为False。

Returns:
    EpsillaVectorStore: 支持添加、删除和查询的向量存储。

示例：
    `pip install llama-index-vector-stores-epsilla`

    ```python
    from llama_index.vector_stores.epsilla import EpsillaVectorStore
    from pyepsilla import vectordb

    client = vectordb.Client()
    vector_store = EpsillaVectorStore(client=client, db_path="/tmp/llamastore")
    ```"""

    stores_text = True
    flat_metadata: bool = False

    _client: vectordb.Client = PrivateAttr()
    _collection_name: str = PrivateAttr()
    _collection_created: bool = PrivateAttr()

    def __init__(
        self,
        client: Any,
        collection_name: str = "llama_collection",
        db_path: Optional[str] = DEFAULT_PERSIST_DIR,  # sub folder
        db_name: Optional[str] = "llama_db",
        dimension: Optional[int] = None,
        overwrite: bool = False,
        **kwargs: Any,
    ) -> None:
        """初始化参数。"""
        super().__init__()

        if not isinstance(client, vectordb.Client):
            raise TypeError(
                f"client should be an instance of pyepsilla.vectordb.Client, "
                f"got {type(client)}"
            )

        self._client: vectordb.Client = client
        self._collection_name = collection_name
        self._client.load_db(db_name, db_path)
        self._client.use_db(db_name)
        self._collection_created = False

        status_code, response = self._client.list_tables()
        if status_code != 200:
            self._handle_error(msg=response["message"])
        table_list = response["result"]

        if self._collection_name in table_list and overwrite is False:
            self._collection_created = True

        if self._collection_name in table_list and overwrite is True:
            status_code, response = self._client.drop_table(
                table_name=self._collection_name
            )
            if status_code != 200:
                self._handle_error(msg=response["message"])
            logger.debug(
                f"Successfully removed old collection: {self._collection_name}"
            )
            if dimension is not None:
                self._create_collection(dimension)

        if self._collection_name not in table_list and dimension is not None:
            self._create_collection(dimension)

    @classmethod
    def class_name(cls) -> str:
        return "EpsillaVectorStore"

    @property
    def client(self) -> Any:
        """返回Epsilla客户端。"""
        return self._client

    def _handle_error(self, msg: str) -> None:
        """处理错误。"""
        logger.error(f"Failed to get records: {msg}")
        raise Exception(f"Error: {msg}.")

    def _create_collection(self, dimension: int) -> None:
        """创建集合。

Args:
    dimension（int）：嵌入的维度。
"""
        fields: List[dict] = [
            {"name": "id", "dataType": "STRING", "primaryKey": True},
            {"name": DEFAULT_DOC_ID_KEY, "dataType": "STRING"},
            {"name": DEFAULT_TEXT_KEY, "dataType": "STRING"},
            {
                "name": DEFAULT_EMBEDDING_KEY,
                "dataType": "VECTOR_FLOAT",
                "dimensions": dimension,
            },
            {"name": "metadata", "dataType": "JSON"},
        ]
        status_code, response = self._client.create_table(
            table_name=self._collection_name, table_fields=fields
        )
        if status_code != 200:
            self._handle_error(msg=response["message"])
        self._collection_created = True
        logger.debug(f"Successfully created collection: {self._collection_name}")

    def add(
        self,
        nodes: List[BaseNode],
        **add_kwargs: Any,
    ) -> List[str]:
        """将节点添加到Epsilla向量存储中。

Args:
    nodes: List[BaseNode]: 带有嵌入的节点列表

Returns:
    List[str]: 插入的id列表。
"""
        # If the collection doesn't exist yet, create the collection
        if not self._collection_created and len(nodes) > 0:
            dimension = len(nodes[0].get_embedding())
            self._create_collection(dimension)

        elif len(nodes) == 0:
            return []

        ids = []
        records = []
        for node in nodes:
            ids.append(node.node_id)
            text = node.get_content(metadata_mode=MetadataMode.NONE)
            metadata_dict = node_to_metadata_dict(node, remove_text=True)
            metadata = metadata_dict["_node_content"]
            record = {
                "id": node.node_id,
                DEFAULT_DOC_ID_KEY: node.ref_doc_id,
                DEFAULT_TEXT_KEY: text,
                DEFAULT_EMBEDDING_KEY: node.get_embedding(),
                "metadata": metadata,
            }
            records.append(record)

        status_code, response = self._client.insert(
            table_name=self._collection_name, records=records
        )
        if status_code != 200:
            self._handle_error(msg=response["message"])

        return ids

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """使用ref_doc_id删除节点。

Args:
    ref_doc_id（str）：要删除的文档的doc_id。
"""
        raise NotImplementedError("Delete with filtering will be coming soon.")

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询前k个最相似节点的索引。

Args:
    query（VectorStoreQuery）：查询。

Returns:
    向量存储查询结果。
"""
        if not self._collection_created:
            raise ValueError("Please initialize a collection first.")

        if query.mode != VectorStoreQueryMode.DEFAULT:
            raise NotImplementedError(f"Epsilla does not support {query.mode} yet.")

        if query.filters is not None:
            raise NotImplementedError("Epsilla does not support Metadata filters yet.")

        if query.doc_ids is not None and len(query.doc_ids) > 0:
            raise NotImplementedError("Epsilla does not support filters yet.")

        status_code, response = self._client.query(
            table_name=self._collection_name,
            query_field=DEFAULT_EMBEDDING_KEY,
            query_vector=query.query_embedding,
            limit=query.similarity_top_k,
            with_distance=True,
        )
        if status_code != 200:
            self._handle_error(msg=response["message"])

        results = response["result"]
        logger.debug(
            f"Successfully searched embedding in collection: {self._collection_name}"
            f" Num Results: {len(results)}"
        )

        nodes = []
        similarities = []
        ids = []
        for res in results:
            try:
                node = metadata_dict_to_node({"_node_content": res["metadata"]})
                node.text = res[DEFAULT_TEXT_KEY]
            except Exception:
                # NOTE: deprecated legacy logic for backward compatibility
                metadata, node_info, relationships = legacy_metadata_dict_to_node(
                    res["metadata"]
                )
                node = TextNode(
                    id=res["id"],
                    text=res[DEFAULT_TEXT_KEY],
                    metadata=metadata,
                    start_char_idx=node_info.get("start", None),
                    end_char_idx=node_info.get("end", None),
                    relationships=relationships,
                )
            nodes.append(node)
            similarities.append(res["@distance"])
            ids.append(res["id"])

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

client property

client: Any

返回Epsilla客户端。

add

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

将节点添加到Epsilla向量存储中。

Parameters:

Name	Type	Description	Default
nodes	List[BaseNode]	List[BaseNode]: 带有嵌入的节点列表	required

Returns:

Type	Description
List[str]	List[str]: 插入的id列表。

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

    def add(
        self,
        nodes: List[BaseNode],
        **add_kwargs: Any,
    ) -> List[str]:
        """将节点添加到Epsilla向量存储中。

Args:
    nodes: List[BaseNode]: 带有嵌入的节点列表

Returns:
    List[str]: 插入的id列表。
"""
        # If the collection doesn't exist yet, create the collection
        if not self._collection_created and len(nodes) > 0:
            dimension = len(nodes[0].get_embedding())
            self._create_collection(dimension)

        elif len(nodes) == 0:
            return []

        ids = []
        records = []
        for node in nodes:
            ids.append(node.node_id)
            text = node.get_content(metadata_mode=MetadataMode.NONE)
            metadata_dict = node_to_metadata_dict(node, remove_text=True)
            metadata = metadata_dict["_node_content"]
            record = {
                "id": node.node_id,
                DEFAULT_DOC_ID_KEY: node.ref_doc_id,
                DEFAULT_TEXT_KEY: text,
                DEFAULT_EMBEDDING_KEY: node.get_embedding(),
                "metadata": metadata,
            }
            records.append(record)

        status_code, response = self._client.insert(
            table_name=self._collection_name, records=records
        )
        if status_code != 200:
            self._handle_error(msg=response["message"])

        return ids

delete

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

使用ref_doc_id删除节点。

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

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """使用ref_doc_id删除节点。

Args:
    ref_doc_id（str）：要删除的文档的doc_id。
"""
        raise NotImplementedError("Delete with filtering will be coming soon.")

query

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

查询前k个最相似节点的索引。

Returns:

Type	Description
VectorStoreQueryResult	向量存储查询结果。

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

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询前k个最相似节点的索引。

Args:
    query（VectorStoreQuery）：查询。

Returns:
    向量存储查询结果。
"""
        if not self._collection_created:
            raise ValueError("Please initialize a collection first.")

        if query.mode != VectorStoreQueryMode.DEFAULT:
            raise NotImplementedError(f"Epsilla does not support {query.mode} yet.")

        if query.filters is not None:
            raise NotImplementedError("Epsilla does not support Metadata filters yet.")

        if query.doc_ids is not None and len(query.doc_ids) > 0:
            raise NotImplementedError("Epsilla does not support filters yet.")

        status_code, response = self._client.query(
            table_name=self._collection_name,
            query_field=DEFAULT_EMBEDDING_KEY,
            query_vector=query.query_embedding,
            limit=query.similarity_top_k,
            with_distance=True,
        )
        if status_code != 200:
            self._handle_error(msg=response["message"])

        results = response["result"]
        logger.debug(
            f"Successfully searched embedding in collection: {self._collection_name}"
            f" Num Results: {len(results)}"
        )

        nodes = []
        similarities = []
        ids = []
        for res in results:
            try:
                node = metadata_dict_to_node({"_node_content": res["metadata"]})
                node.text = res[DEFAULT_TEXT_KEY]
            except Exception:
                # NOTE: deprecated legacy logic for backward compatibility
                metadata, node_info, relationships = legacy_metadata_dict_to_node(
                    res["metadata"]
                )
                node = TextNode(
                    id=res["id"],
                    text=res[DEFAULT_TEXT_KEY],
                    metadata=metadata,
                    start_char_idx=node_info.get("start", None),
                    end_char_idx=node_info.get("end", None),
                    relationships=relationships,
                )
            nodes.append(node)
            similarities.append(res["@distance"])
            ids.append(res["id"])

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