Weaviate

WeaviateVectorStore

Bases: BasePydanticVectorStore

Weaviate向量存储。

在这个向量存储中，嵌入和文档被存储在Weaviate集合中。

在查询时，索引使用Weaviate查询前k个最相似的节点。

Parameters:

Name	Type	Description	Default
weaviate_client	Client	weaviate-client包中的WeaviateClient实例	None
index_name	Optional[str]	Weaviate类的名称	None

示例

pip install llama-index-vector-stores-weaviate

import weaviate

resource_owner_config = weaviate.AuthClientPassword(
    username="<username>",
    password="<password>",
)
client = weaviate.Client(
    "https://llama-test-ezjahb4m.weaviate.network",
    auth_client_secret=resource_owner_config,
)

vector_store = WeaviateVectorStore(
    weaviate_client=client, index_name="LlamaIndex"
)

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

class WeaviateVectorStore(BasePydanticVectorStore):
    """Weaviate向量存储。

在这个向量存储中，嵌入和文档被存储在Weaviate集合中。

在查询时，索引使用Weaviate查询前k个最相似的节点。

Args:
    weaviate_client (weaviate.Client): `weaviate-client`包中的WeaviateClient实例
    index_name (Optional[str]): Weaviate类的名称

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

    ```python
    import weaviate

    resource_owner_config = weaviate.AuthClientPassword(
        username="<username>",
        password="<password>",
    )
    client = weaviate.Client(
        "https://llama-test-ezjahb4m.weaviate.network",
        auth_client_secret=resource_owner_config,
    )

    vector_store = WeaviateVectorStore(
        weaviate_client=client, index_name="LlamaIndex"
    )
    ```"""

    stores_text: bool = True

    index_name: str
    url: Optional[str]
    text_key: str
    auth_config: Dict[str, Any] = Field(default_factory=dict)
    client_kwargs: Dict[str, Any] = Field(default_factory=dict)

    _client = PrivateAttr()

    def __init__(
        self,
        weaviate_client: Optional[Any] = None,
        class_prefix: Optional[str] = None,
        index_name: Optional[str] = None,
        text_key: str = DEFAULT_TEXT_KEY,
        auth_config: Optional[Any] = None,
        client_kwargs: Optional[Dict[str, Any]] = None,
        url: Optional[str] = None,
        **kwargs: Any,
    ) -> None:
        """初始化参数。"""
        if weaviate_client is None:
            if isinstance(auth_config, dict):
                auth_config = weaviate.auth.AuthApiKey(auth_config)

            client_kwargs = client_kwargs or {}
            self._client = weaviate.WeaviateClient(
                auth_client_secret=auth_config, **client_kwargs
            )
        else:
            self._client = cast(weaviate.WeaviateClient, weaviate_client)

        # validate class prefix starts with a capital letter
        if class_prefix is not None:
            _logger.warning("class_prefix is deprecated, please use index_name")
            # legacy, kept for backward compatibility
            index_name = f"{class_prefix}_Node"

        index_name = index_name or f"LlamaIndex_{uuid4().hex}"
        if not index_name[0].isupper():
            raise ValueError(
                "Index name must start with a capital letter, e.g. 'LlamaIndex'"
            )

        # create default schema if does not exist
        if not class_schema_exists(self._client, index_name):
            create_default_schema(self._client, index_name)

        super().__init__(
            url=url,
            index_name=index_name,
            text_key=text_key,
            auth_config=auth_config.__dict__ if auth_config else {},
            client_kwargs=client_kwargs or {},
        )

    @classmethod
    def from_params(
        cls,
        url: str,
        auth_config: Any,
        index_name: Optional[str] = None,
        text_key: str = DEFAULT_TEXT_KEY,
        client_kwargs: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> "WeaviateVectorStore":
        """从配置中创建WeaviateVectorStore。"""
        client_kwargs = client_kwargs or {}
        weaviate_client = Client(
            url=url, auth_client_secret=auth_config, **client_kwargs
        )
        return cls(
            weaviate_client=weaviate_client,
            url=url,
            auth_config=auth_config.__dict__,
            client_kwargs=client_kwargs,
            index_name=index_name,
            text_key=text_key,
            **kwargs,
        )

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

    @property
    def client(self) -> Any:
        """获取客户端。"""
        return self._client

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

Args:
    节点: List[BaseNode]: 带有嵌入的节点列表
"""
        ids = [r.node_id for r in nodes]

        with self._client.batch.dynamic() as batch:
            for node in nodes:
                add_node(
                    self._client,
                    node,
                    self.index_name,
                    batch=batch,
                    text_key=self.text_key,
                )
        return ids

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

Args:
    ref_doc_id（str）：要删除的文档的doc_id。
"""
        collection = self._client.collections.get(self.index_name)

        where_filter = wvc.query.Filter.by_property("ref_doc_id").equal(ref_doc_id)

        if "filter" in delete_kwargs and delete_kwargs["filter"] is not None:
            where_filter = where_filter & _to_weaviate_filter(delete_kwargs["filter"])

        collection.data.delete_many(where=where_filter)

    def delete_index(self) -> None:
        """删除与客户端关联的索引。

引发:
- Exception: 如果由于某种原因删除失败。
"""
        if not class_schema_exists(self._client, self.index_name):
            _logger.warning(
                f"Index '{self.index_name}' does not exist. No action taken."
            )
            return
        try:
            self._client.collections.delete(self.index_name)
            _logger.info(f"Successfully deleted index '{self.index_name}'.")
        except Exception as e:
            _logger.error(f"Failed to delete index '{self.index_name}': {e}")
            raise Exception(f"Failed to delete index '{self.index_name}': {e}")

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询前k个最相似节点的索引。"""
        all_properties = get_all_properties(self._client, self.index_name)
        collection = self._client.collections.get(self.index_name)
        filters = None

        # list of documents to constrain search
        if query.doc_ids:
            filters = wvc.query.Filter.by_property("doc_id").contains_any(query.doc_ids)

        if query.node_ids:
            filters = wvc.query.Filter.by_property("id").contains_any(query.node_ids)

        return_metatada = wvc.query.MetadataQuery(distance=True, score=True)

        vector = query.query_embedding
        similarity_key = "distance"
        if query.mode == VectorStoreQueryMode.DEFAULT:
            _logger.debug("Using vector search")
            if vector is not None:
                alpha = 1
        elif query.mode == VectorStoreQueryMode.HYBRID:
            _logger.debug(f"Using hybrid search with alpha {query.alpha}")
            similarity_key = "score"
            if vector is not None and query.query_str:
                alpha = query.alpha

        if query.filters is not None:
            filters = _to_weaviate_filter(query.filters)
        elif "filter" in kwargs and kwargs["filter"] is not None:
            filters = kwargs["filter"]

        limit = query.similarity_top_k
        _logger.debug(f"Using limit of {query.similarity_top_k}")

        # execute query
        try:
            query_result = collection.query.hybrid(
                query=query.query_str,
                vector=vector,
                alpha=alpha,
                limit=limit,
                filters=filters,
                return_metadata=return_metatada,
                return_properties=all_properties,
                include_vector=True,
            )
        except weaviate.exceptions.WeaviateQueryError as e:
            raise ValueError(f"Invalid query, got errors: {e.message}")

        # parse results

        entries = query_result.objects

        similarities = []
        nodes: List[BaseNode] = []
        node_ids = []

        for i, entry in enumerate(entries):
            if i < query.similarity_top_k:
                entry_as_dict = entry.__dict__
                similarities.append(get_node_similarity(entry_as_dict, similarity_key))
                nodes.append(to_node(entry_as_dict, text_key=self.text_key))
                node_ids.append(nodes[-1].node_id)
            else:
                break

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

client property

client: Any

获取客户端。

from_params classmethod

from_params(
    url: str,
    auth_config: Any,
    index_name: Optional[str] = None,
    text_key: str = DEFAULT_TEXT_KEY,
    client_kwargs: Optional[Dict[str, Any]] = None,
    **kwargs: Any
) -> WeaviateVectorStore

从配置中创建WeaviateVectorStore。

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

@classmethod
def from_params(
    cls,
    url: str,
    auth_config: Any,
    index_name: Optional[str] = None,
    text_key: str = DEFAULT_TEXT_KEY,
    client_kwargs: Optional[Dict[str, Any]] = None,
    **kwargs: Any,
) -> "WeaviateVectorStore":
    """从配置中创建WeaviateVectorStore。"""
    client_kwargs = client_kwargs or {}
    weaviate_client = Client(
        url=url, auth_client_secret=auth_config, **client_kwargs
    )
    return cls(
        weaviate_client=weaviate_client,
        url=url,
        auth_config=auth_config.__dict__,
        client_kwargs=client_kwargs,
        index_name=index_name,
        text_key=text_key,
        **kwargs,
    )

add

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

将节点添加到索引中。

Parameters:

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

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

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

Args:
    节点: List[BaseNode]: 带有嵌入的节点列表
"""
        ids = [r.node_id for r in nodes]

        with self._client.batch.dynamic() as batch:
            for node in nodes:
                add_node(
                    self._client,
                    node,
                    self.index_name,
                    batch=batch,
                    text_key=self.text_key,
                )
        return ids

delete

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

使用ref_doc_id删除节点。

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

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

Args:
    ref_doc_id（str）：要删除的文档的doc_id。
"""
        collection = self._client.collections.get(self.index_name)

        where_filter = wvc.query.Filter.by_property("ref_doc_id").equal(ref_doc_id)

        if "filter" in delete_kwargs and delete_kwargs["filter"] is not None:
            where_filter = where_filter & _to_weaviate_filter(delete_kwargs["filter"])

        collection.data.delete_many(where=where_filter)

delete_index

delete_index() -> None

删除与客户端关联的索引。

引发: - Exception: 如果由于某种原因删除失败。

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

    def delete_index(self) -> None:
        """删除与客户端关联的索引。

引发:
- Exception: 如果由于某种原因删除失败。
"""
        if not class_schema_exists(self._client, self.index_name):
            _logger.warning(
                f"Index '{self.index_name}' does not exist. No action taken."
            )
            return
        try:
            self._client.collections.delete(self.index_name)
            _logger.info(f"Successfully deleted index '{self.index_name}'.")
        except Exception as e:
            _logger.error(f"Failed to delete index '{self.index_name}': {e}")
            raise Exception(f"Failed to delete index '{self.index_name}': {e}")

query

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

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

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

def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
    """查询前k个最相似节点的索引。"""
    all_properties = get_all_properties(self._client, self.index_name)
    collection = self._client.collections.get(self.index_name)
    filters = None

    # list of documents to constrain search
    if query.doc_ids:
        filters = wvc.query.Filter.by_property("doc_id").contains_any(query.doc_ids)

    if query.node_ids:
        filters = wvc.query.Filter.by_property("id").contains_any(query.node_ids)

    return_metatada = wvc.query.MetadataQuery(distance=True, score=True)

    vector = query.query_embedding
    similarity_key = "distance"
    if query.mode == VectorStoreQueryMode.DEFAULT:
        _logger.debug("Using vector search")
        if vector is not None:
            alpha = 1
    elif query.mode == VectorStoreQueryMode.HYBRID:
        _logger.debug(f"Using hybrid search with alpha {query.alpha}")
        similarity_key = "score"
        if vector is not None and query.query_str:
            alpha = query.alpha

    if query.filters is not None:
        filters = _to_weaviate_filter(query.filters)
    elif "filter" in kwargs and kwargs["filter"] is not None:
        filters = kwargs["filter"]

    limit = query.similarity_top_k
    _logger.debug(f"Using limit of {query.similarity_top_k}")

    # execute query
    try:
        query_result = collection.query.hybrid(
            query=query.query_str,
            vector=vector,
            alpha=alpha,
            limit=limit,
            filters=filters,
            return_metadata=return_metatada,
            return_properties=all_properties,
            include_vector=True,
        )
    except weaviate.exceptions.WeaviateQueryError as e:
        raise ValueError(f"Invalid query, got errors: {e.message}")

    # parse results

    entries = query_result.objects

    similarities = []
    nodes: List[BaseNode] = []
    node_ids = []

    for i, entry in enumerate(entries):
        if i < query.similarity_top_k:
            entry_as_dict = entry.__dict__
            similarities.append(get_node_similarity(entry_as_dict, similarity_key))
            nodes.append(to_node(entry_as_dict, text_key=self.text_key))
            node_ids.append(nodes[-1].node_id)
        else:
            break

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