Vespa

VespaVectorStore

Bases: BasePydanticVectorStore

    # Vespa向量存储。

    # 可以通过几种方式进行初始化：
    # 1. （默认）使用默认的混合模板和本地（docker）部署初始化Vespa向量存储。
    # 2. 通过提供在pyvespa中创建的应用程序包进行初始化（可以在本地部署或部署到Vespa云）。
    # 3. 通过提供URL从先前部署的Vespa应用程序进行初始化。（本地或云部署）。

    # 应用程序必须设置以下字段：
    # - id：文档id
    # - text：文本字段
    # - embedding：用于存储嵌入向量的字段。
    # - metadata：元数据字段（所有元数据将存储在此处）

    # 应用程序必须设置以下排名配置文件：
    # - bm25：用于文本搜索
    # - semantic：用于语义搜索
    # - fusion：用于语义混合搜索

    # 当从VespaVectorStore创建VectorStoreIndex时，索引将向Vespa应用程序添加文档。
    # 请注意，如果在部署之间未删除Vespa容器，则将重用Vespa容器，以避免数据重复。
    # 在查询时，索引会查询Vespa应用程序以获取前k个最相关的结果。

    # Args:
    #         application_package（ApplicationPackage）：应用程序包
    #         deployment_target（str）：部署目标，可以是`local`或`cloud`
    #         port（int）：Vespa应用程序运行的端口。仅在deployment_target为`local`时适用
    #         default_schema_name（str）：Vespa应用程序中的模式名称
    #         namespace（str）：Vespa应用程序中的命名空间。请参阅https://docs.vespa.ai/en/documents.html#namespace。默认为`default`。
    #         embeddings_outside_vespa（bool）：嵌入是否在Vespa之外创建。
    #         url（Optional[str]）：部署的Vespa应用程序的URL。
    #         groupname（Optional[str]）：Vespa应用程序中的组名，仅适用于`streaming`模式，请参阅https://pyvespa.readthedocs.io/en/latest/examples/scaling-personal-ai-assistants-with-streaming-mode-cloud.html#A-summary-of-Vespa-streaming-mode
    #         tenant（Optional[str]）：Vespa应用程序的租户。仅在deployment_target为`cloud`时适用
    #         key_location（Optional[str]）：用于签署发送到Vespa Cloud的HTTP请求的控制平面密钥的位置。
    #         key_content（Optional[str]）：用于签署发送到Vespa Cloud的HTTP请求的控制平面密钥的内容。仅在密钥文件不可用时使用。
    #         auth_client_token_id（Optional[str]）：使用基于令牌的数据平面身份验证。这是在Vespa Cloud控制台中配置的令牌名称。这用于配置Vespa services.xml。该令牌被赋予读取和写入权限。
    #         kwargs（Any）：Vespa应用程序的其他kwargs

    # 示例：
    #     `pip install llama-index-vector-stores-vespa`

    #     ```python
    #     from llama_index.core import VectorStoreIndex
    #     from llama_index.vector_stores.vespa import VespaVectorStore

    #     vector_store = VespaVectorStore()
    #     storage_context = StorageContext.from_defaults(vector_store=vector_store)
    #     index = VectorStoreIndex(nodes, storage_context=storage_context)
    #     retriever = index.as_retriever()
    #     retriever.retrieve("Who directed inception?")

    #     ```

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

class VespaVectorStore(BasePydanticVectorStore):
    """```python
    # Vespa向量存储。

    # 可以通过几种方式进行初始化：
    # 1. （默认）使用默认的混合模板和本地（docker）部署初始化Vespa向量存储。
    # 2. 通过提供在pyvespa中创建的应用程序包进行初始化（可以在本地部署或部署到Vespa云）。
    # 3. 通过提供URL从先前部署的Vespa应用程序进行初始化。（本地或云部署）。

    # 应用程序必须设置以下字段：
    # - id：文档id
    # - text：文本字段
    # - embedding：用于存储嵌入向量的字段。
    # - metadata：元数据字段（所有元数据将存储在此处）

    # 应用程序必须设置以下排名配置文件：
    # - bm25：用于文本搜索
    # - semantic：用于语义搜索
    # - fusion：用于语义混合搜索

    # 当从VespaVectorStore创建VectorStoreIndex时，索引将向Vespa应用程序添加文档。
    # 请注意，如果在部署之间未删除Vespa容器，则将重用Vespa容器，以避免数据重复。
    # 在查询时，索引会查询Vespa应用程序以获取前k个最相关的结果。

    # Args:
    #         application_package（ApplicationPackage）：应用程序包
    #         deployment_target（str）：部署目标，可以是`local`或`cloud`
    #         port（int）：Vespa应用程序运行的端口。仅在deployment_target为`local`时适用
    #         default_schema_name（str）：Vespa应用程序中的模式名称
    #         namespace（str）：Vespa应用程序中的命名空间。请参阅https://docs.vespa.ai/en/documents.html#namespace。默认为`default`。
    #         embeddings_outside_vespa（bool）：嵌入是否在Vespa之外创建。
    #         url（Optional[str]）：部署的Vespa应用程序的URL。
    #         groupname（Optional[str]）：Vespa应用程序中的组名，仅适用于`streaming`模式，请参阅https://pyvespa.readthedocs.io/en/latest/examples/scaling-personal-ai-assistants-with-streaming-mode-cloud.html#A-summary-of-Vespa-streaming-mode
    #         tenant（Optional[str]）：Vespa应用程序的租户。仅在deployment_target为`cloud`时适用
    #         key_location（Optional[str]）：用于签署发送到Vespa Cloud的HTTP请求的控制平面密钥的位置。
    #         key_content（Optional[str]）：用于签署发送到Vespa Cloud的HTTP请求的控制平面密钥的内容。仅在密钥文件不可用时使用。
    #         auth_client_token_id（Optional[str]）：使用基于令牌的数据平面身份验证。这是在Vespa Cloud控制台中配置的令牌名称。这用于配置Vespa services.xml。该令牌被赋予读取和写入权限。
    #         kwargs（Any）：Vespa应用程序的其他kwargs

    # 示例：
    #     `pip install llama-index-vector-stores-vespa`

    #     ```python
    #     from llama_index.core import VectorStoreIndex
    #     from llama_index.vector_stores.vespa import VespaVectorStore

    #     vector_store = VespaVectorStore()
    #     storage_context = StorageContext.from_defaults(vector_store=vector_store)
    #     index = VectorStoreIndex(nodes, storage_context=storage_context)
    #     retriever = index.as_retriever()
    #     retriever.retrieve("Who directed inception?")

    #     ```
```"""

    stores_text: bool = True
    is_embedding_query: bool = False
    flat_metadata: bool = True

    application_package: ApplicationPackage
    deployment_target: str
    default_schema_name: str
    namespace: str
    embeddings_outside_vespa: bool
    port: int
    url: Optional[str]
    groupname: Optional[str]
    tenant: Optional[str]
    application: Optional[str]
    key_location: Optional[str]
    key_content: Optional[str]
    auth_client_token_id: Optional[str]
    kwargs: dict

    _app: Vespa = PrivateAttr()

    def __init__(
        self,
        application_package: ApplicationPackage = hybrid_template,
        namespace: str = "default",
        default_schema_name: str = "doc",
        deployment_target: str = "local",  # "local" or "cloud"
        port: int = 8080,
        embeddings_outside_vespa: bool = False,
        url: Optional[str] = None,
        groupname: Optional[str] = None,
        tenant: Optional[str] = None,
        application: Optional[str] = "hybridsearch",
        key_location: Optional[str] = None,
        key_content: Optional[str] = None,
        auth_client_token_id: Optional[str] = None,
        **kwargs: Any,
    ) -> None:
        # Verify that application_package is an instance of ApplicationPackage
        if not isinstance(application_package, ApplicationPackage):
            raise ValueError(
                "application_package must be an instance of vespa.package.ApplicationPackage"
            )
        if application_package == hybrid_template:
            logger.info(
                "Using default hybrid template. Please make sure that the Vespa application is set up with the correct schema and rank profile."
            )
        # Initialize all parameters
        super().__init__(
            application_package=application_package,
            namespace=namespace,
            default_schema_name=default_schema_name,
            deployment_target=deployment_target,
            port=port,
            embeddings_outside_vespa=embeddings_outside_vespa,
            url=url,
            groupname=groupname,
            tenant=tenant,
            application=application,
            key_location=key_location,
            key_content=key_content,
            auth_client_token_id=auth_client_token_id,
            kwargs=kwargs,
        )

        if self.url is None:
            self._app = self._deploy()
        else:
            self._app = self._try_get_running_app()

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

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

    def _try_get_running_app(self) -> Vespa:
        app = Vespa(url=f"{self.url}:{self.port}")
        status = app.get_application_status()
        if status.status_code == 200:
            return app
        else:
            raise ConnectionError(
                f"Vespa application not running on url {self.url} and port {self.port}. Please start Vespa application first."
            )

    def _deploy(self) -> Vespa:
        if self.deployment_target == "cloud":
            app = self._deploy_app_cloud()
        elif self.deployment_target == "local":
            app = self._deploy_app_local()
        else:
            raise ValueError(
                f"Deployment target {self.deployment_target} not supported. Please choose either `local` or `cloud`."
            )
        return app

    def _deploy_app_local(self) -> Vespa:
        logger.info(f"Deploying Vespa application {self.application} to Vespa Docker.")
        return VespaDocker(port=8080).deploy(self.application_package)

    def _deploy_app_cloud(self) -> Vespa:
        logger.info(f"Deploying Vespa application {self.application} to Vespa Cloud.")
        return VespaCloud(
            tenant=self.tenant,
            application=self.application,
            application_package=self.application_package,
            key_location=self.key_location,
            key_content=self.key_content,
            auth_client_token_id=self.auth_client_token_id,
            **self.kwargs,
        ).deploy()

    def add(
        self,
        nodes: List[BaseNode],
        schema: Optional[str] = None,
        callback: Optional[Callable[[VespaResponse, str], None]] = callback,
    ) -> List[str]:
        """将节点添加到向量存储中。

Args:
    nodes（List[BaseNode]）：要添加的节点列表
    schema（Optional[str]）：要将节点添加到的Vespa应用程序中的模式名称。默认为`default_schema_name`。
"""
        # Create vespa iterable from nodes
        ids = []
        data_to_insert = []
        for node in nodes:
            metadata = node_to_metadata_dict(
                node, remove_text=False, flat_metadata=self.flat_metadata
            )
            logger.debug(f"Metadata: {metadata}")
            entry = {
                "id": node.node_id,
                "fields": {
                    "id": node.node_id,
                    "text": node.get_content(metadata_mode=MetadataMode.NONE) or "",
                    "metadata": json.dumps(metadata),
                },
            }
            if self.embeddings_outside_vespa:
                entry["fields"]["embedding"] = node.get_embedding()
            data_to_insert.append(entry)
            ids.append(node.node_id)

        self._app.feed_iterable(
            data_to_insert,
            schema=schema or self.default_schema_name,
            namespace=self.namespace,
            operation_type="feed",
            callback=callback,
        )
        return ids

    async def async_add(
        self,
        nodes: List[BaseNode],
        schema: Optional[str] = None,
        callback: Optional[Callable[[VespaResponse, str], None]] = callback,
        max_connections: int = 10,
        num_concurrent_requests: int = 1000,
        total_timeout: int = 60,
        **kwargs: Any,
    ) -> List[str]:
        """将节点异步添加到向量存储中。

Args:
    nodes（List[BaseNode]）：要添加的节点列表
    schema（Optional[str]）：要将节点添加到的Vespa应用程序中的模式名称。默认为`default_schema_name`。
    max_connections（int）：与Vespa应用程序的最大连接数
    num_concurrent_requests（int）：最大并发请求数
    total_timeout（int）：所有请求的总超时时间
    kwargs（Any）：Vespa应用程序的其他kwargs
"""
        semaphore = asyncio.Semaphore(num_concurrent_requests)
        ids = []
        data_to_insert = []
        for node in nodes:
            metadata = node_to_metadata_dict(
                node, remove_text=False, flat_metadata=self.flat_metadata
            )
            logger.debug(f"Metadata: {metadata}")
            entry = {
                "id": node.node_id,
                "fields": {
                    "id": node.node_id,
                    "text": node.get_content(metadata_mode=MetadataMode.NONE) or "",
                    "metadata": json.dumps(metadata),
                },
            }
            if self.embeddings_outside_vespa:
                entry["fields"]["embedding"] = node.get_embedding()
            data_to_insert.append(entry)
            ids.append(node.node_id)

        async with self._app.asyncio(
            connections=max_connections, total_timeout=total_timeout
        ) as async_app:
            tasks = []
            for doc in data_to_insert:
                async with semaphore:
                    task = asyncio.create_task(
                        async_app.feed_data_point(
                            data_id=doc["id"],
                            fields=doc["fields"],
                            schema=schema or self.default_schema_name,
                            namespace=self.namespace,
                            timeout=10,
                        )
                    )
                    tasks.append(task)

            results = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
            for result in results:
                if result.exception():
                    raise result.exception
        return ids

    def delete(
        self,
        ref_doc_id: str,
        namespace: Optional[str] = None,
        **delete_kwargs: Any,
    ) -> None:
        """
        使用ref_doc_id删除节点。
        """
        response: VespaResponse = self._app.delete_data(
            schema=self.default_schema_name,
            namespace=namespace or self.namespace,
            data_id=ref_doc_id,
            kwargs=delete_kwargs,
        )
        if not response.is_successful():
            raise ValueError(
                f"Delete request failed: {response.status_code}, response payload: {response.json}"
            )
        logger.info(f"Deleted node with id {ref_doc_id}")

    async def adelete(
        self,
        ref_doc_id: str,
        namespace: Optional[str] = None,
        **delete_kwargs: Any,
    ) -> None:
        """使用ref_doc_id删除节点。
注意：并非所有向量存储都实现了此功能。如果未实现，它将仅同步调用删除。
"""
        logger.info("Async delete not implemented. Will call delete synchronously.")
        self.delete(ref_doc_id, **delete_kwargs)

    def _create_query_body(
        self,
        query: VectorStoreQuery,
        sources_str: str,
        rank_profile: Optional[str] = None,
        create_embedding: bool = True,
        vector_top_k: int = 10,
    ) -> dict:
        """为Vespa创建查询参数。

Args:
    query (VectorStoreQuery): VectorStoreQuery对象
    sources_str (str): 源字符串
    rank_profile (Optional[str]): 要使用的排名配置文件。如果未提供，则使用默认的排名配置文件。
    create_embedding (bool): 是否创建嵌入
    vector_top_k (int): 要返回的前k个向量的数量

Returns:
    dict: 查询参数
"""
        logger.info(f"Query: {query}")
        if query.filters:
            logger.warning("Filter support not implemented yet. Will be ignored.")
        if query.alpha:
            logger.warning(
                "Alpha support not implemented. Must be defined in Vespa rank profile. "
                "See for example https://pyvespa.readthedocs.io/en/latest/examples/evaluating-with-snowflake-arctic-embed.html"
            )

        if query.query_embedding is None and not create_embedding:
            raise ValueError(
                "Input embedding must be provided if embeddings are not created outside Vespa"
            )

        base_params = {
            "hits": query.similarity_top_k,
            "ranking.profile": rank_profile
            or self._get_default_rank_profile(query.mode),
            "query": query.query_str,
            "tracelevel": 9,
        }
        logger.debug(query.mode)
        if query.mode in [
            VectorStoreQueryMode.TEXT_SEARCH,
            VectorStoreQueryMode.DEFAULT,
        ]:
            query_params = {"yql": f"select * from {sources_str} where userQuery()"}
        elif query.mode in [
            VectorStoreQueryMode.SEMANTIC_HYBRID,
            VectorStoreQueryMode.HYBRID,
        ]:
            if not query.embedding_field:
                embedding_field = "embedding"
                logger.warning(
                    f"Embedding field not provided. Using default embedding field {embedding_field}"
                )
            query_params = {
                "yql": f"select * from {sources_str} where {self._build_query_filter(query.mode, embedding_field, vector_top_k, query.similarity_top_k)}",
                "input.query(q)": (
                    f"embed({query.query_str})"
                    if create_embedding
                    else query.query_embedding
                ),
            }
        else:
            raise NotImplementedError(
                f"Query mode {query.mode} not implemented for Vespa yet. Contributions are welcome!"
            )

        return {**base_params, **query_params}

    def _get_default_rank_profile(self, mode):
        return {
            VectorStoreQueryMode.TEXT_SEARCH: "bm25",
            VectorStoreQueryMode.SEMANTIC_HYBRID: "fusion",
            VectorStoreQueryMode.HYBRID: "fusion",
            VectorStoreQueryMode.DEFAULT: "bm25",
        }.get(mode)

    def _build_query_filter(
        self, mode, embedding_field, vector_top_k, similarity_top_k
    ):
        """构建Vespa查询的查询过滤器。
查询中“select * from {sources_str} where”之后的部分。
"""
        if mode in [
            VectorStoreQueryMode.SEMANTIC_HYBRID,
            VectorStoreQueryMode.HYBRID,
        ]:
            return f"rank({{targetHits:{vector_top_k}}}nearestNeighbor({embedding_field},q), userQuery()) limit {similarity_top_k}"
        else:
            raise ValueError(f"Query mode {mode} not supported.")

    def query(
        self,
        query: VectorStoreQuery,
        sources: Optional[List[str]] = None,
        rank_profile: Optional[str] = None,
        vector_top_k: int = 10,
        **kwargs: Any,
    ) -> VectorStoreQueryResult:
        """查询向量存储。"""
        logger.debug(f"Query: {query}")
        sources_str = ",".join(sources) if sources else "sources *"
        mode = query.mode
        body = self._create_query_body(
            query=query,
            sources_str=sources_str,
            rank_profile=rank_profile,
            create_embedding=not self.embeddings_outside_vespa,
            vector_top_k=vector_top_k,
        )
        logger.info(f"Vespa Query body:\n {body}")
        with self._app.syncio() as session:
            response = session.query(
                body=body,
            )
        if not response.is_successful():
            raise ValueError(
                f"Query request failed: {response.status_code}, response payload: {response.get_json()}"
            )
        logger.debug("Response:")
        logger.debug(response.json)
        logger.debug("Hits:")
        logger.debug(response.hits)
        nodes = []
        ids: List[str] = []
        similarities: List[float] = []
        for hit in response.hits:
            response_fields: dict = hit.get("fields", {})
            metadata = response_fields.get("metadata", {})
            metadata = json.loads(metadata)
            logger.debug(f"Metadata: {metadata}")
            node = metadata_dict_to_node(metadata)
            text = response_fields.get("body", "")
            node.set_content(text)
            nodes.append(node)
            ids.append(response_fields.get("id"))
            similarities.append(hit["relevance"])
        return VectorStoreQueryResult(nodes=nodes, ids=ids, similarities=similarities)

    async def aquery(
        self,
        query: VectorStoreQuery,
        sources: Optional[List[str]] = None,
        rank_profile: Optional[str] = None,
        vector_top_k: int = 10,
        **kwargs: Any,
    ) -> VectorStoreQueryResult:
        """异步查询向量存储。
注意：并非所有向量存储都实现了这个功能。如果没有实现，它将会同步调用查询。
"""
        logger.info("Async query not implemented. Will call query synchronously.")
        return self.query(
            query=query,
            sources=sources,
            rank_profile=rank_profile,
            vector_top_k=vector_top_k,
            **kwargs,
        )

    def persist(
        self,
    ) -> None:
        return NotImplemented("Persist is not implemented for VespaVectorStore")

client property

client: Vespa

获取客户端。

add

add(
    nodes: List[BaseNode],
    schema: Optional[str] = None,
    callback: Optional[
        Callable[[VespaResponse, str], None]
    ] = callback,
) -> List[str]

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

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

    def add(
        self,
        nodes: List[BaseNode],
        schema: Optional[str] = None,
        callback: Optional[Callable[[VespaResponse, str], None]] = callback,
    ) -> List[str]:
        """将节点添加到向量存储中。

Args:
    nodes（List[BaseNode]）：要添加的节点列表
    schema（Optional[str]）：要将节点添加到的Vespa应用程序中的模式名称。默认为`default_schema_name`。
"""
        # Create vespa iterable from nodes
        ids = []
        data_to_insert = []
        for node in nodes:
            metadata = node_to_metadata_dict(
                node, remove_text=False, flat_metadata=self.flat_metadata
            )
            logger.debug(f"Metadata: {metadata}")
            entry = {
                "id": node.node_id,
                "fields": {
                    "id": node.node_id,
                    "text": node.get_content(metadata_mode=MetadataMode.NONE) or "",
                    "metadata": json.dumps(metadata),
                },
            }
            if self.embeddings_outside_vespa:
                entry["fields"]["embedding"] = node.get_embedding()
            data_to_insert.append(entry)
            ids.append(node.node_id)

        self._app.feed_iterable(
            data_to_insert,
            schema=schema or self.default_schema_name,
            namespace=self.namespace,
            operation_type="feed",
            callback=callback,
        )
        return ids

async_add async

async_add(
    nodes: List[BaseNode],
    schema: Optional[str] = None,
    callback: Optional[
        Callable[[VespaResponse, str], None]
    ] = callback,
    max_connections: int = 10,
    num_concurrent_requests: int = 1000,
    total_timeout: int = 60,
    **kwargs: Any
) -> List[str]

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

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

    async def async_add(
        self,
        nodes: List[BaseNode],
        schema: Optional[str] = None,
        callback: Optional[Callable[[VespaResponse, str], None]] = callback,
        max_connections: int = 10,
        num_concurrent_requests: int = 1000,
        total_timeout: int = 60,
        **kwargs: Any,
    ) -> List[str]:
        """将节点异步添加到向量存储中。

Args:
    nodes（List[BaseNode]）：要添加的节点列表
    schema（Optional[str]）：要将节点添加到的Vespa应用程序中的模式名称。默认为`default_schema_name`。
    max_connections（int）：与Vespa应用程序的最大连接数
    num_concurrent_requests（int）：最大并发请求数
    total_timeout（int）：所有请求的总超时时间
    kwargs（Any）：Vespa应用程序的其他kwargs
"""
        semaphore = asyncio.Semaphore(num_concurrent_requests)
        ids = []
        data_to_insert = []
        for node in nodes:
            metadata = node_to_metadata_dict(
                node, remove_text=False, flat_metadata=self.flat_metadata
            )
            logger.debug(f"Metadata: {metadata}")
            entry = {
                "id": node.node_id,
                "fields": {
                    "id": node.node_id,
                    "text": node.get_content(metadata_mode=MetadataMode.NONE) or "",
                    "metadata": json.dumps(metadata),
                },
            }
            if self.embeddings_outside_vespa:
                entry["fields"]["embedding"] = node.get_embedding()
            data_to_insert.append(entry)
            ids.append(node.node_id)

        async with self._app.asyncio(
            connections=max_connections, total_timeout=total_timeout
        ) as async_app:
            tasks = []
            for doc in data_to_insert:
                async with semaphore:
                    task = asyncio.create_task(
                        async_app.feed_data_point(
                            data_id=doc["id"],
                            fields=doc["fields"],
                            schema=schema or self.default_schema_name,
                            namespace=self.namespace,
                            timeout=10,
                        )
                    )
                    tasks.append(task)

            results = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
            for result in results:
                if result.exception():
                    raise result.exception
        return ids

delete

delete(
    ref_doc_id: str,
    namespace: Optional[str] = None,
    **delete_kwargs: Any
) -> None

使用ref_doc_id删除节点。

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

def delete(
    self,
    ref_doc_id: str,
    namespace: Optional[str] = None,
    **delete_kwargs: Any,
) -> None:
    """
    使用ref_doc_id删除节点。
    """
    response: VespaResponse = self._app.delete_data(
        schema=self.default_schema_name,
        namespace=namespace or self.namespace,
        data_id=ref_doc_id,
        kwargs=delete_kwargs,
    )
    if not response.is_successful():
        raise ValueError(
            f"Delete request failed: {response.status_code}, response payload: {response.json}"
        )
    logger.info(f"Deleted node with id {ref_doc_id}")

adelete async

adelete(
    ref_doc_id: str,
    namespace: Optional[str] = None,
    **delete_kwargs: Any
) -> None

使用ref_doc_id删除节点。 注意：并非所有向量存储都实现了此功能。如果未实现，它将仅同步调用删除。

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

    async def adelete(
        self,
        ref_doc_id: str,
        namespace: Optional[str] = None,
        **delete_kwargs: Any,
    ) -> None:
        """使用ref_doc_id删除节点。
注意：并非所有向量存储都实现了此功能。如果未实现，它将仅同步调用删除。
"""
        logger.info("Async delete not implemented. Will call delete synchronously.")
        self.delete(ref_doc_id, **delete_kwargs)

query

query(
    query: VectorStoreQuery,
    sources: Optional[List[str]] = None,
    rank_profile: Optional[str] = None,
    vector_top_k: int = 10,
    **kwargs: Any
) -> VectorStoreQueryResult

查询向量存储。

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

def query(
    self,
    query: VectorStoreQuery,
    sources: Optional[List[str]] = None,
    rank_profile: Optional[str] = None,
    vector_top_k: int = 10,
    **kwargs: Any,
) -> VectorStoreQueryResult:
    """查询向量存储。"""
    logger.debug(f"Query: {query}")
    sources_str = ",".join(sources) if sources else "sources *"
    mode = query.mode
    body = self._create_query_body(
        query=query,
        sources_str=sources_str,
        rank_profile=rank_profile,
        create_embedding=not self.embeddings_outside_vespa,
        vector_top_k=vector_top_k,
    )
    logger.info(f"Vespa Query body:\n {body}")
    with self._app.syncio() as session:
        response = session.query(
            body=body,
        )
    if not response.is_successful():
        raise ValueError(
            f"Query request failed: {response.status_code}, response payload: {response.get_json()}"
        )
    logger.debug("Response:")
    logger.debug(response.json)
    logger.debug("Hits:")
    logger.debug(response.hits)
    nodes = []
    ids: List[str] = []
    similarities: List[float] = []
    for hit in response.hits:
        response_fields: dict = hit.get("fields", {})
        metadata = response_fields.get("metadata", {})
        metadata = json.loads(metadata)
        logger.debug(f"Metadata: {metadata}")
        node = metadata_dict_to_node(metadata)
        text = response_fields.get("body", "")
        node.set_content(text)
        nodes.append(node)
        ids.append(response_fields.get("id"))
        similarities.append(hit["relevance"])
    return VectorStoreQueryResult(nodes=nodes, ids=ids, similarities=similarities)

aquery async

aquery(
    query: VectorStoreQuery,
    sources: Optional[List[str]] = None,
    rank_profile: Optional[str] = None,
    vector_top_k: int = 10,
    **kwargs: Any
) -> VectorStoreQueryResult

异步查询向量存储。 注意：并非所有向量存储都实现了这个功能。如果没有实现，它将会同步调用查询。

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

    async def aquery(
        self,
        query: VectorStoreQuery,
        sources: Optional[List[str]] = None,
        rank_profile: Optional[str] = None,
        vector_top_k: int = 10,
        **kwargs: Any,
    ) -> VectorStoreQueryResult:
        """异步查询向量存储。
注意：并非所有向量存储都实现了这个功能。如果没有实现，它将会同步调用查询。
"""
        logger.info("Async query not implemented. Will call query synchronously.")
        return self.query(
            query=query,
            sources=sources,
            rank_profile=rank_profile,
            vector_top_k=vector_top_k,
            **kwargs,
        )