Elasticsearch

ElasticsearchStore

Bases: BasePydanticVectorStore

Elasticsearch vector store.

Parameters:

Name	Type	Description	Default
index_name	str	Name of the Elasticsearch index.	required
es_client	Optional[Any]	Optional. Pre-existing AsyncElasticsearch client.	None
es_url	Optional[str]	Optional. Elasticsearch URL.	None
es_cloud_id	Optional[str]	Optional. Elasticsearch cloud ID.	None
es_api_key	Optional[str]	Optional. Elasticsearch API key.	None
es_user	Optional[str]	Optional. Elasticsearch username.	None
es_password	Optional[str]	Optional. Elasticsearch password.	None
text_field	str	Optional. Name of the Elasticsearch field that stores the text.	'content'
vector_field	str	Optional. Name of the Elasticsearch field that stores the embedding.	'embedding'
batch_size	int	Optional. Batch size for bulk indexing. Defaults to 200.	200
distance_strategy	Optional[DISTANCE_STRATEGIES]	Optional. Distance strategy to use for similarity search. Defaults to "COSINE".	'COSINE'
retrieval_strategy	Optional[AsyncRetrievalStrategy]	Retrieval strategy to use. AsyncBM25Strategy / AsyncSparseVectorStrategy / AsyncDenseVectorStrategy / AsyncRetrievalStrategy. Defaults to AsyncDenseVectorStrategy.	None

Raises:

Type	Description
ConnectionError	If AsyncElasticsearch client cannot connect to Elasticsearch.
ValueError	If neither es_client nor es_url nor es_cloud_id is provided.

Examples:

pip install llama-index-vector-stores-elasticsearch

from llama_index.vector_stores import ElasticsearchStore

# Additional setup for ElasticsearchStore class
index_name = "my_index"
es_url = "http://localhost:9200"
es_cloud_id = "<cloud-id>"  # Found within the deployment page
es_user = "elastic"
es_password = "<password>"  # Provided when creating deployment or can be reset
es_api_key = "<api-key>"  # Create an API key within Kibana (Security -> API Keys)

# Connecting to ElasticsearchStore locally
es_local = ElasticsearchStore(
    index_name=index_name,
    es_url=es_url,
)

# Connecting to Elastic Cloud with username and password
es_cloud_user_pass = ElasticsearchStore(
    index_name=index_name,
    es_cloud_id=es_cloud_id,
    es_user=es_user,
    es_password=es_password,
)

# Connecting to Elastic Cloud with API Key
es_cloud_api_key = ElasticsearchStore(
    index_name=index_name,
    es_cloud_id=es_cloud_id,
    es_api_key=es_api_key,
)

Source code in llama-index-integrations/vector_stores/llama-index-vector-stores-elasticsearch/llama_index/vector_stores/elasticsearch/base.py

class ElasticsearchStore(BasePydanticVectorStore):
    """
    Elasticsearch vector store.

    Args:
        index_name: Name of the Elasticsearch index.
        es_client: Optional. Pre-existing AsyncElasticsearch client.
        es_url: Optional. Elasticsearch URL.
        es_cloud_id: Optional. Elasticsearch cloud ID.
        es_api_key: Optional. Elasticsearch API key.
        es_user: Optional. Elasticsearch username.
        es_password: Optional. Elasticsearch password.
        text_field: Optional. Name of the Elasticsearch field that stores the text.
        vector_field: Optional. Name of the Elasticsearch field that stores the
                    embedding.
        batch_size: Optional. Batch size for bulk indexing. Defaults to 200.
        distance_strategy: Optional. Distance strategy to use for similarity search.
                        Defaults to "COSINE".
        retrieval_strategy: Retrieval strategy to use. AsyncBM25Strategy /
            AsyncSparseVectorStrategy / AsyncDenseVectorStrategy / AsyncRetrievalStrategy.
            Defaults to AsyncDenseVectorStrategy.

    Raises:
        ConnectionError: If AsyncElasticsearch client cannot connect to Elasticsearch.
        ValueError: If neither es_client nor es_url nor es_cloud_id is provided.

    Examples:
        `pip install llama-index-vector-stores-elasticsearch`

        ```python
        from llama_index.vector_stores import ElasticsearchStore

        # Additional setup for ElasticsearchStore class
        index_name = "my_index"
        es_url = "http://localhost:9200"
        es_cloud_id = "<cloud-id>"  # Found within the deployment page
        es_user = "elastic"
        es_password = "<password>"  # Provided when creating deployment or can be reset
        es_api_key = "<api-key>"  # Create an API key within Kibana (Security -> API Keys)

        # Connecting to ElasticsearchStore locally
        es_local = ElasticsearchStore(
            index_name=index_name,
            es_url=es_url,
        )

        # Connecting to Elastic Cloud with username and password
        es_cloud_user_pass = ElasticsearchStore(
            index_name=index_name,
            es_cloud_id=es_cloud_id,
            es_user=es_user,
            es_password=es_password,
        )

        # Connecting to Elastic Cloud with API Key
        es_cloud_api_key = ElasticsearchStore(
            index_name=index_name,
            es_cloud_id=es_cloud_id,
            es_api_key=es_api_key,
        )
        ```

    """

    class Config:
        # allow pydantic to tolarate its inability to validate AsyncRetrievalStrategy
        arbitrary_types_allowed = True

    stores_text: bool = True
    index_name: str
    es_client: Optional[Any]
    es_url: Optional[str]
    es_cloud_id: Optional[str]
    es_api_key: Optional[str]
    es_user: Optional[str]
    es_password: Optional[str]
    text_field: str = "content"
    vector_field: str = "embedding"
    batch_size: int = 200
    distance_strategy: Optional[DISTANCE_STRATEGIES] = "COSINE"
    retrieval_strategy: AsyncRetrievalStrategy

    _store = PrivateAttr()

    def __init__(
        self,
        index_name: str,
        es_client: Optional[Any] = None,
        es_url: Optional[str] = None,
        es_cloud_id: Optional[str] = None,
        es_api_key: Optional[str] = None,
        es_user: Optional[str] = None,
        es_password: Optional[str] = None,
        text_field: str = "content",
        vector_field: str = "embedding",
        batch_size: int = 200,
        distance_strategy: Optional[DISTANCE_STRATEGIES] = "COSINE",
        retrieval_strategy: Optional[AsyncRetrievalStrategy] = None,
        metadata_mappings: Optional[Dict[str, Any]] = None,
        **kwargs,
    ) -> None:
        nest_asyncio.apply()

        if not es_client:
            es_client = get_elasticsearch_client(
                url=es_url,
                cloud_id=es_cloud_id,
                api_key=es_api_key,
                username=es_user,
                password=es_password,
            )

        if retrieval_strategy is None:
            retrieval_strategy = AsyncDenseVectorStrategy(
                distance=DistanceMetric[distance_strategy]
            )

        base_metadata_mappings = {
            "document_id": {"type": "keyword"},
            "doc_id": {"type": "keyword"},
            "ref_doc_id": {"type": "keyword"},
        }

        metadata_mappings = metadata_mappings or {}
        metadata_mappings.update(base_metadata_mappings)

        super().__init__(
            index_name=index_name,
            es_client=es_client,
            es_url=es_url,
            es_cloud_id=es_cloud_id,
            es_api_key=es_api_key,
            es_user=es_user,
            es_password=es_password,
            text_field=text_field,
            vector_field=vector_field,
            batch_size=batch_size,
            distance_strategy=distance_strategy,
            retrieval_strategy=retrieval_strategy,
        )

        self._store = AsyncVectorStore(
            user_agent=get_user_agent(),
            client=es_client,
            index=index_name,
            retrieval_strategy=retrieval_strategy,
            text_field=text_field,
            vector_field=vector_field,
            metadata_mappings=metadata_mappings,
        )

        # Disable query embeddings when using Sparse vectors or BM25.
        # ELSER generates its own embeddings server-side
        if not isinstance(retrieval_strategy, AsyncDenseVectorStrategy):
            self.is_embedding_query = False

    @property
    def client(self) -> Any:
        """Get async elasticsearch client."""
        return self._store.client

    def close(self) -> None:
        return asyncio.get_event_loop().run_until_complete(self._store.close())

    def add(
        self,
        nodes: List[BaseNode],
        *,
        create_index_if_not_exists: bool = True,
        **add_kwargs: Any,
    ) -> List[str]:
        """
        Add nodes to Elasticsearch index.

        Args:
            nodes: List of nodes with embeddings.
            create_index_if_not_exists: Optional. Whether to create
                                        the Elasticsearch index if it
                                        doesn't already exist.
                                        Defaults to True.

        Returns:
            List of node IDs that were added to the index.

        Raises:
            ImportError: If elasticsearch['async'] python package is not installed.
            BulkIndexError: If AsyncElasticsearch async_bulk indexing fails.
        """
        return asyncio.get_event_loop().run_until_complete(
            self.async_add(
                nodes,
                create_index_if_not_exists=create_index_if_not_exists,
                **add_kwargs,
            )
        )

    async def async_add(
        self,
        nodes: List[BaseNode],
        *,
        create_index_if_not_exists: bool = True,
        **add_kwargs: Any,
    ) -> List[str]:
        """
        Asynchronous method to add nodes to Elasticsearch index.

        Args:
            nodes: List of nodes with embeddings.
            create_index_if_not_exists: Optional. Whether to create
                                        the AsyncElasticsearch index if it
                                        doesn't already exist.
                                        Defaults to True.

        Returns:
            List of node IDs that were added to the index.

        Raises:
            ImportError: If elasticsearch python package is not installed.
            BulkIndexError: If AsyncElasticsearch async_bulk indexing fails.
        """
        if len(nodes) == 0:
            return []

        embeddings: Optional[List[List[float]]] = None
        texts: List[str] = []
        metadatas: List[dict] = []
        ids: List[str] = []
        for node in nodes:
            ids.append(node.node_id)
            texts.append(node.get_content(metadata_mode=MetadataMode.NONE))
            metadatas.append(node_to_metadata_dict(node, remove_text=True))

        # Generate embeddings when using dense vectors. They are not needed
        # for other strategies.
        if isinstance(self.retrieval_strategy, AsyncDenseVectorStrategy):
            embeddings = []
            for node in nodes:
                embeddings.append(node.get_embedding())

            if not self._store.num_dimensions:
                self._store.num_dimensions = len(embeddings[0])

        return await self._store.add_texts(
            texts=texts,
            metadatas=metadatas,
            vectors=embeddings,
            ids=ids,
            create_index_if_not_exists=create_index_if_not_exists,
            bulk_kwargs=add_kwargs,
        )

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """
        Delete node from Elasticsearch index.

        Args:
            ref_doc_id: ID of the node to delete.
            delete_kwargs: Optional. Additional arguments to
                        pass to Elasticsearch delete_by_query.

        Raises:
            Exception: If Elasticsearch delete_by_query fails.
        """
        return asyncio.get_event_loop().run_until_complete(
            self.adelete(ref_doc_id, **delete_kwargs)
        )

    async def adelete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """
        Async delete node from Elasticsearch index.

        Args:
            ref_doc_id: ID of the node to delete.
            delete_kwargs: Optional. Additional arguments to
                        pass to AsyncElasticsearch delete_by_query.

        Raises:
            Exception: If AsyncElasticsearch delete_by_query fails.
        """
        await self._store.delete(
            query={"term": {"metadata.ref_doc_id": ref_doc_id}}, **delete_kwargs
        )

    def query(
        self,
        query: VectorStoreQuery,
        custom_query: Optional[
            Callable[[Dict, Union[VectorStoreQuery, None]], Dict]
        ] = None,
        es_filter: Optional[List[Dict]] = None,
        **kwargs: Any,
    ) -> VectorStoreQueryResult:
        """
        Query index for top k most similar nodes.

        Args:
            query_embedding (List[float]): query embedding
            custom_query: Optional. custom query function that takes in the es query
                        body and returns a modified query body.
                        This can be used to add additional query
                        parameters to the Elasticsearch query.
            es_filter: Optional. Elasticsearch filter to apply to the
                        query. If filter is provided in the query,
                        this filter will be ignored.

        Returns:
            VectorStoreQueryResult: Result of the query.

        Raises:
            Exception: If Elasticsearch query fails.

        """
        return asyncio.get_event_loop().run_until_complete(
            self.aquery(query, custom_query, es_filter, **kwargs)
        )

    async def aquery(
        self,
        query: VectorStoreQuery,
        custom_query: Optional[
            Callable[[Dict, Union[VectorStoreQuery, None]], Dict]
        ] = None,
        es_filter: Optional[List[Dict]] = None,
        **kwargs: Any,
    ) -> VectorStoreQueryResult:
        """
        Asynchronous query index for top k most similar nodes.

        Args:
            query_embedding (VectorStoreQuery): query embedding
            custom_query: Optional. custom query function that takes in the es query
                        body and returns a modified query body.
                        This can be used to add additional query
                        parameters to the AsyncElasticsearch query.
            es_filter: Optional. AsyncElasticsearch filter to apply to the
                        query. If filter is provided in the query,
                        this filter will be ignored.

        Returns:
            VectorStoreQueryResult: Result of the query.

        Raises:
            Exception: If AsyncElasticsearch query fails.

        """
        _mode_must_match_retrieval_strategy(query.mode, self.retrieval_strategy)

        if query.filters is not None and len(query.filters.legacy_filters()) > 0:
            filter = [_to_elasticsearch_filter(query.filters)]
        else:
            filter = es_filter or []

        hits = await self._store.search(
            query=query.query_str,
            query_vector=query.query_embedding,
            k=query.similarity_top_k,
            num_candidates=query.similarity_top_k * 10,
            filter=filter,
            custom_query=custom_query,
        )

        top_k_nodes = []
        top_k_ids = []
        top_k_scores = []
        for hit in hits:
            source = hit["_source"]
            metadata = source.get("metadata", {})
            text = source.get(self.text_field, None)
            node_id = hit["_id"]

            try:
                node = metadata_dict_to_node(metadata)
                node.text = text
            except Exception:
                # Legacy support for old metadata format
                logger.warning(
                    f"Could not parse metadata from hit {hit['_source'].get('metadata')}"
                )
                node_info = source.get("node_info")
                relationships = source.get("relationships", {})
                start_char_idx = None
                end_char_idx = None
                if isinstance(node_info, dict):
                    start_char_idx = node_info.get("start", None)
                    end_char_idx = node_info.get("end", None)

                node = TextNode(
                    text=text,
                    metadata=metadata,
                    id_=node_id,
                    start_char_idx=start_char_idx,
                    end_char_idx=end_char_idx,
                    relationships=relationships,
                )
            top_k_nodes.append(node)
            top_k_ids.append(node_id)
            top_k_scores.append(hit.get("_rank", hit["_score"]))

        if (
            isinstance(self.retrieval_strategy, AsyncDenseVectorStrategy)
            and self.retrieval_strategy.hybrid
        ):
            total_rank = sum(top_k_scores)
            top_k_scores = [total_rank - rank / total_rank for rank in top_k_scores]

        return VectorStoreQueryResult(
            nodes=top_k_nodes,
            ids=top_k_ids,
            similarities=_to_llama_similarities(top_k_scores),
        )

client property

client: Any

Get async elasticsearch client.

add

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

Add nodes to Elasticsearch index.

Parameters:

Name	Type	Description	Default
nodes	List[BaseNode]	List of nodes with embeddings.	required
create_index_if_not_exists	bool	Optional. Whether to create the Elasticsearch index if it doesn't already exist. Defaults to True.	True

Returns:

Type	Description
List[str]	List of node IDs that were added to the index.

Raises:

Type	Description
ImportError	If elasticsearch['async'] python package is not installed.
BulkIndexError	If AsyncElasticsearch async_bulk indexing fails.

Source code in llama-index-integrations/vector_stores/llama-index-vector-stores-elasticsearch/llama_index/vector_stores/elasticsearch/base.py

def add(
    self,
    nodes: List[BaseNode],
    *,
    create_index_if_not_exists: bool = True,
    **add_kwargs: Any,
) -> List[str]:
    """
    Add nodes to Elasticsearch index.

    Args:
        nodes: List of nodes with embeddings.
        create_index_if_not_exists: Optional. Whether to create
                                    the Elasticsearch index if it
                                    doesn't already exist.
                                    Defaults to True.

    Returns:
        List of node IDs that were added to the index.

    Raises:
        ImportError: If elasticsearch['async'] python package is not installed.
        BulkIndexError: If AsyncElasticsearch async_bulk indexing fails.
    """
    return asyncio.get_event_loop().run_until_complete(
        self.async_add(
            nodes,
            create_index_if_not_exists=create_index_if_not_exists,
            **add_kwargs,
        )
    )

async_add async

async_add(nodes: List[BaseNode], *, create_index_if_not_exists: bool = True, **add_kwargs: Any) -> List[str]

Asynchronous method to add nodes to Elasticsearch index.

Parameters:

Name	Type	Description	Default
nodes	List[BaseNode]	List of nodes with embeddings.	required
create_index_if_not_exists	bool	Optional. Whether to create the AsyncElasticsearch index if it doesn't already exist. Defaults to True.	True

Returns:

Type	Description
List[str]	List of node IDs that were added to the index.

Raises:

Type	Description
ImportError	If elasticsearch python package is not installed.
BulkIndexError	If AsyncElasticsearch async_bulk indexing fails.

Source code in llama-index-integrations/vector_stores/llama-index-vector-stores-elasticsearch/llama_index/vector_stores/elasticsearch/base.py

async def async_add(
    self,
    nodes: List[BaseNode],
    *,
    create_index_if_not_exists: bool = True,
    **add_kwargs: Any,
) -> List[str]:
    """
    Asynchronous method to add nodes to Elasticsearch index.

    Args:
        nodes: List of nodes with embeddings.
        create_index_if_not_exists: Optional. Whether to create
                                    the AsyncElasticsearch index if it
                                    doesn't already exist.
                                    Defaults to True.

    Returns:
        List of node IDs that were added to the index.

    Raises:
        ImportError: If elasticsearch python package is not installed.
        BulkIndexError: If AsyncElasticsearch async_bulk indexing fails.
    """
    if len(nodes) == 0:
        return []

    embeddings: Optional[List[List[float]]] = None
    texts: List[str] = []
    metadatas: List[dict] = []
    ids: List[str] = []
    for node in nodes:
        ids.append(node.node_id)
        texts.append(node.get_content(metadata_mode=MetadataMode.NONE))
        metadatas.append(node_to_metadata_dict(node, remove_text=True))

    # Generate embeddings when using dense vectors. They are not needed
    # for other strategies.
    if isinstance(self.retrieval_strategy, AsyncDenseVectorStrategy):
        embeddings = []
        for node in nodes:
            embeddings.append(node.get_embedding())

        if not self._store.num_dimensions:
            self._store.num_dimensions = len(embeddings[0])

    return await self._store.add_texts(
        texts=texts,
        metadatas=metadatas,
        vectors=embeddings,
        ids=ids,
        create_index_if_not_exists=create_index_if_not_exists,
        bulk_kwargs=add_kwargs,
    )

delete

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

Delete node from Elasticsearch index.

Parameters:

Name	Type	Description	Default
ref_doc_id	str	ID of the node to delete.	required
delete_kwargs	Any	Optional. Additional arguments to pass to Elasticsearch delete_by_query.	{}

Raises:

Type	Description
Exception	If Elasticsearch delete_by_query fails.

Source code in llama-index-integrations/vector_stores/llama-index-vector-stores-elasticsearch/llama_index/vector_stores/elasticsearch/base.py

def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
    """
    Delete node from Elasticsearch index.

    Args:
        ref_doc_id: ID of the node to delete.
        delete_kwargs: Optional. Additional arguments to
                    pass to Elasticsearch delete_by_query.

    Raises:
        Exception: If Elasticsearch delete_by_query fails.
    """
    return asyncio.get_event_loop().run_until_complete(
        self.adelete(ref_doc_id, **delete_kwargs)
    )

adelete async

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

Async delete node from Elasticsearch index.

Parameters:

Name	Type	Description	Default
ref_doc_id	str	ID of the node to delete.	required
delete_kwargs	Any	Optional. Additional arguments to pass to AsyncElasticsearch delete_by_query.	{}

Raises:

Type	Description
Exception	If AsyncElasticsearch delete_by_query fails.

Source code in llama-index-integrations/vector_stores/llama-index-vector-stores-elasticsearch/llama_index/vector_stores/elasticsearch/base.py

async def adelete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
    """
    Async delete node from Elasticsearch index.

    Args:
        ref_doc_id: ID of the node to delete.
        delete_kwargs: Optional. Additional arguments to
                    pass to AsyncElasticsearch delete_by_query.

    Raises:
        Exception: If AsyncElasticsearch delete_by_query fails.
    """
    await self._store.delete(
        query={"term": {"metadata.ref_doc_id": ref_doc_id}}, **delete_kwargs
    )

query

query(query: VectorStoreQuery, custom_query: Optional[Callable[[Dict, Union[VectorStoreQuery, None]], Dict]] = None, es_filter: Optional[List[Dict]] = None, **kwargs: Any) -> VectorStoreQueryResult

Query index for top k most similar nodes.

Parameters:

Name	Type	Description	Default
query_embedding	List[float]	query embedding	required
custom_query	Optional[Callable[[Dict, Union[VectorStoreQuery, None]], Dict]]	Optional. custom query function that takes in the es query body and returns a modified query body. This can be used to add additional query parameters to the Elasticsearch query.	None
es_filter	Optional[List[Dict]]	Optional. Elasticsearch filter to apply to the query. If filter is provided in the query, this filter will be ignored.	None

Returns:

Name	Type	Description
VectorStoreQueryResult	VectorStoreQueryResult	Result of the query.

Raises:

Type	Description
Exception	If Elasticsearch query fails.

Source code in llama-index-integrations/vector_stores/llama-index-vector-stores-elasticsearch/llama_index/vector_stores/elasticsearch/base.py

def query(
    self,
    query: VectorStoreQuery,
    custom_query: Optional[
        Callable[[Dict, Union[VectorStoreQuery, None]], Dict]
    ] = None,
    es_filter: Optional[List[Dict]] = None,
    **kwargs: Any,
) -> VectorStoreQueryResult:
    """
    Query index for top k most similar nodes.

    Args:
        query_embedding (List[float]): query embedding
        custom_query: Optional. custom query function that takes in the es query
                    body and returns a modified query body.
                    This can be used to add additional query
                    parameters to the Elasticsearch query.
        es_filter: Optional. Elasticsearch filter to apply to the
                    query. If filter is provided in the query,
                    this filter will be ignored.

    Returns:
        VectorStoreQueryResult: Result of the query.

    Raises:
        Exception: If Elasticsearch query fails.

    """
    return asyncio.get_event_loop().run_until_complete(
        self.aquery(query, custom_query, es_filter, **kwargs)
    )

aquery async

aquery(query: VectorStoreQuery, custom_query: Optional[Callable[[Dict, Union[VectorStoreQuery, None]], Dict]] = None, es_filter: Optional[List[Dict]] = None, **kwargs: Any) -> VectorStoreQueryResult

Asynchronous query index for top k most similar nodes.

Parameters:

Name	Type	Description	Default
query_embedding	VectorStoreQuery	query embedding	required
custom_query	Optional[Callable[[Dict, Union[VectorStoreQuery, None]], Dict]]	Optional. custom query function that takes in the es query body and returns a modified query body. This can be used to add additional query parameters to the AsyncElasticsearch query.	None
es_filter	Optional[List[Dict]]	Optional. AsyncElasticsearch filter to apply to the query. If filter is provided in the query, this filter will be ignored.	None

Returns:

Name	Type	Description
VectorStoreQueryResult	VectorStoreQueryResult	Result of the query.

Raises:

Type	Description
Exception	If AsyncElasticsearch query fails.

Source code in llama-index-integrations/vector_stores/llama-index-vector-stores-elasticsearch/llama_index/vector_stores/elasticsearch/base.py

async def aquery(
    self,
    query: VectorStoreQuery,
    custom_query: Optional[
        Callable[[Dict, Union[VectorStoreQuery, None]], Dict]
    ] = None,
    es_filter: Optional[List[Dict]] = None,
    **kwargs: Any,
) -> VectorStoreQueryResult:
    """
    Asynchronous query index for top k most similar nodes.

    Args:
        query_embedding (VectorStoreQuery): query embedding
        custom_query: Optional. custom query function that takes in the es query
                    body and returns a modified query body.
                    This can be used to add additional query
                    parameters to the AsyncElasticsearch query.
        es_filter: Optional. AsyncElasticsearch filter to apply to the
                    query. If filter is provided in the query,
                    this filter will be ignored.

    Returns:
        VectorStoreQueryResult: Result of the query.

    Raises:
        Exception: If AsyncElasticsearch query fails.

    """
    _mode_must_match_retrieval_strategy(query.mode, self.retrieval_strategy)

    if query.filters is not None and len(query.filters.legacy_filters()) > 0:
        filter = [_to_elasticsearch_filter(query.filters)]
    else:
        filter = es_filter or []

    hits = await self._store.search(
        query=query.query_str,
        query_vector=query.query_embedding,
        k=query.similarity_top_k,
        num_candidates=query.similarity_top_k * 10,
        filter=filter,
        custom_query=custom_query,
    )

    top_k_nodes = []
    top_k_ids = []
    top_k_scores = []
    for hit in hits:
        source = hit["_source"]
        metadata = source.get("metadata", {})
        text = source.get(self.text_field, None)
        node_id = hit["_id"]

        try:
            node = metadata_dict_to_node(metadata)
            node.text = text
        except Exception:
            # Legacy support for old metadata format
            logger.warning(
                f"Could not parse metadata from hit {hit['_source'].get('metadata')}"
            )
            node_info = source.get("node_info")
            relationships = source.get("relationships", {})
            start_char_idx = None
            end_char_idx = None
            if isinstance(node_info, dict):
                start_char_idx = node_info.get("start", None)
                end_char_idx = node_info.get("end", None)

            node = TextNode(
                text=text,
                metadata=metadata,
                id_=node_id,
                start_char_idx=start_char_idx,
                end_char_idx=end_char_idx,
                relationships=relationships,
            )
        top_k_nodes.append(node)
        top_k_ids.append(node_id)
        top_k_scores.append(hit.get("_rank", hit["_score"]))

    if (
        isinstance(self.retrieval_strategy, AsyncDenseVectorStrategy)
        and self.retrieval_strategy.hybrid
    ):
        total_rank = sum(top_k_scores)
        top_k_scores = [total_rank - rank / total_rank for rank in top_k_scores]

    return VectorStoreQueryResult(
        nodes=top_k_nodes,
        ids=top_k_ids,
        similarities=_to_llama_similarities(top_k_scores),
    )