Opensearch

OpensearchVectorStore

Bases: BasePydanticVectorStore

Elasticsearch/Opensearch vector store.

Parameters:

Name	Type	Description	Default
client	OpensearchVectorClient	Vector index client to use for data insertion/querying.	required

Examples:

pip install llama-index-vector-stores-opensearch

from llama_index.vector_stores.opensearch import (
    OpensearchVectorStore,
    OpensearchVectorClient,
)

# http endpoint for your cluster (opensearch required for vector index usage)
endpoint = "http://localhost:9200"
# index to demonstrate the VectorStore impl
idx = "gpt-index-demo"

# OpensearchVectorClient stores text in this field by default
text_field = "content"
# OpensearchVectorClient stores embeddings in this field by default
embedding_field = "embedding"

# OpensearchVectorClient encapsulates logic for a
# single opensearch index with vector search enabled
client = OpensearchVectorClient(
    endpoint, idx, 1536, embedding_field=embedding_field, text_field=text_field
)

# initialize vector store
vector_store = OpensearchVectorStore(client)

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

class OpensearchVectorStore(BasePydanticVectorStore):
    """
    Elasticsearch/Opensearch vector store.

    Args:
        client (OpensearchVectorClient): Vector index client to use
            for data insertion/querying.

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

        ```python
        from llama_index.vector_stores.opensearch import (
            OpensearchVectorStore,
            OpensearchVectorClient,
        )

        # http endpoint for your cluster (opensearch required for vector index usage)
        endpoint = "http://localhost:9200"
        # index to demonstrate the VectorStore impl
        idx = "gpt-index-demo"

        # OpensearchVectorClient stores text in this field by default
        text_field = "content"
        # OpensearchVectorClient stores embeddings in this field by default
        embedding_field = "embedding"

        # OpensearchVectorClient encapsulates logic for a
        # single opensearch index with vector search enabled
        client = OpensearchVectorClient(
            endpoint, idx, 1536, embedding_field=embedding_field, text_field=text_field
        )

        # initialize vector store
        vector_store = OpensearchVectorStore(client)
        ```

    """

    stores_text: bool = True
    _client: OpensearchVectorClient = PrivateAttr(default=None)

    def __init__(
        self,
        client: OpensearchVectorClient,
    ) -> None:
        """Initialize params."""
        super().__init__()
        self._client = client

    @property
    def client(self) -> Any:
        """Get client."""
        return self._client

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

        Args:
            nodes: List[BaseNode]: list of nodes with embeddings.

        """
        self._client.index_results(nodes)
        return [result.node_id for result in nodes]

    async def async_add(
        self,
        nodes: List[BaseNode],
        **add_kwargs: Any,
    ) -> List[str]:
        """
        Async add nodes to index.

        Args:
            nodes: List[BaseNode]: list of nodes with embeddings.

        """
        await self._client.aindex_results(nodes)
        return [result.node_id for result in nodes]

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """
        Delete nodes using with ref_doc_id.

        Args:
            ref_doc_id (str): The doc_id of the document to delete.

        """
        self._client.delete_by_doc_id(ref_doc_id)

    async def adelete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """
        Async delete nodes using with ref_doc_id.

        Args:
            ref_doc_id (str): The doc_id of the document to delete.

        """
        await self._client.adelete_by_doc_id(ref_doc_id)

    def delete_nodes(
        self,
        node_ids: Optional[List[str]] = None,
        filters: Optional[MetadataFilters] = None,
        **delete_kwargs: Any,
    ) -> None:
        """
        Deletes nodes async.

        Args:
            node_ids (Optional[List[str]], optional): IDs of nodes to delete. Defaults to None.
            filters (Optional[MetadataFilters], optional): Metadata filters. Defaults to None.

        """
        self._client.delete_nodes(node_ids, filters, **delete_kwargs)

    async def adelete_nodes(
        self,
        node_ids: Optional[List[str]] = None,
        filters: Optional[MetadataFilters] = None,
        **delete_kwargs: Any,
    ) -> None:
        """
        Async deletes nodes async.

        Args:
            node_ids (Optional[List[str]], optional): IDs of nodes to delete. Defaults to None.
            filters (Optional[MetadataFilters], optional): Metadata filters. Defaults to None.

        """
        await self._client.adelete_nodes(node_ids, filters, **delete_kwargs)

    def clear(self) -> None:
        """Clears index."""
        self._client.clear()

    async def aclear(self) -> None:
        """Async clears index."""
        await self._client.aclear()

    def close(self) -> None:
        """Close the vector store and release resources."""
        self._client.close()

    async def aclose(self) -> None:
        """Asynchronously close the vector store and release resources."""
        await self._client.aclose()

    def __del__(self) -> None:
        """Clean up resources during garbage collection."""
        try:
            self.close()
        except Exception as exc:
            logger.debug(
                "Failed to close OpenSearch vector store during garbage collection, "
                "type=%s err='%s'",
                type(exc),
                exc,
            )

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """
        Query index for top k most similar nodes.

        Args:
            query (VectorStoreQuery): Store query object.

        """
        query_embedding = cast(List[float], query.query_embedding)

        return self._client.query(
            query.mode,
            query.query_str,
            query_embedding,
            query.similarity_top_k,
            filters=query.filters,
        )

    async def aquery(
        self, query: VectorStoreQuery, **kwargs: Any
    ) -> VectorStoreQueryResult:
        """
        Async query index for top k most similar nodes.

        Args:
            query (VectorStoreQuery): Store query object.

        """
        query_embedding = cast(List[float], query.query_embedding)

        return await self._client.aquery(
            query.mode,
            query.query_str,
            query_embedding,
            query.similarity_top_k,
            filters=query.filters,
        )

client property

client: Any

Get client.

add

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

Add nodes to index.

Parameters:

Name	Type	Description	Default
nodes	List[BaseNode]	List[BaseNode]: list of nodes with embeddings.	required

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

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

    Args:
        nodes: List[BaseNode]: list of nodes with embeddings.

    """
    self._client.index_results(nodes)
    return [result.node_id for result in nodes]

async_add async

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

Async add nodes to index.

Parameters:

Name	Type	Description	Default
nodes	List[BaseNode]	List[BaseNode]: list of nodes with embeddings.	required

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

async def async_add(
    self,
    nodes: List[BaseNode],
    **add_kwargs: Any,
) -> List[str]:
    """
    Async add nodes to index.

    Args:
        nodes: List[BaseNode]: list of nodes with embeddings.

    """
    await self._client.aindex_results(nodes)
    return [result.node_id for result in nodes]

delete

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

Delete nodes using with ref_doc_id.

Parameters:

Name	Type	Description	Default
ref_doc_id	str	The doc_id of the document to delete.	required

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

def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
    """
    Delete nodes using with ref_doc_id.

    Args:
        ref_doc_id (str): The doc_id of the document to delete.

    """
    self._client.delete_by_doc_id(ref_doc_id)

adelete async

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

Async delete nodes using with ref_doc_id.

Parameters:

Name	Type	Description	Default
ref_doc_id	str	The doc_id of the document to delete.	required

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

async def adelete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
    """
    Async delete nodes using with ref_doc_id.

    Args:
        ref_doc_id (str): The doc_id of the document to delete.

    """
    await self._client.adelete_by_doc_id(ref_doc_id)

delete_nodes

delete_nodes(
    node_ids: Optional[List[str]] = None,
    filters: Optional[MetadataFilters] = None,
    **delete_kwargs: Any
) -> None

Deletes nodes async.

Parameters:

Name	Type	Description	Default
node_ids	Optional[List[str]]	IDs of nodes to delete. Defaults to None.	None
filters	Optional[MetadataFilters]	Metadata filters. Defaults to None.	None

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

def delete_nodes(
    self,
    node_ids: Optional[List[str]] = None,
    filters: Optional[MetadataFilters] = None,
    **delete_kwargs: Any,
) -> None:
    """
    Deletes nodes async.

    Args:
        node_ids (Optional[List[str]], optional): IDs of nodes to delete. Defaults to None.
        filters (Optional[MetadataFilters], optional): Metadata filters. Defaults to None.

    """
    self._client.delete_nodes(node_ids, filters, **delete_kwargs)

adelete_nodes async

adelete_nodes(
    node_ids: Optional[List[str]] = None,
    filters: Optional[MetadataFilters] = None,
    **delete_kwargs: Any
) -> None

Async deletes nodes async.

Parameters:

Name	Type	Description	Default
node_ids	Optional[List[str]]	IDs of nodes to delete. Defaults to None.	None
filters	Optional[MetadataFilters]	Metadata filters. Defaults to None.	None

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

async def adelete_nodes(
    self,
    node_ids: Optional[List[str]] = None,
    filters: Optional[MetadataFilters] = None,
    **delete_kwargs: Any,
) -> None:
    """
    Async deletes nodes async.

    Args:
        node_ids (Optional[List[str]], optional): IDs of nodes to delete. Defaults to None.
        filters (Optional[MetadataFilters], optional): Metadata filters. Defaults to None.

    """
    await self._client.adelete_nodes(node_ids, filters, **delete_kwargs)

clear

clear() -> None

Clears index.

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

def clear(self) -> None:
    """Clears index."""
    self._client.clear()

aclear async

aclear() -> None

Async clears index.

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

async def aclear(self) -> None:
    """Async clears index."""
    await self._client.aclear()

close

close() -> None

Close the vector store and release resources.

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

def close(self) -> None:
    """Close the vector store and release resources."""
    self._client.close()

aclose async

aclose() -> None

Asynchronously close the vector store and release resources.

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

async def aclose(self) -> None:
    """Asynchronously close the vector store and release resources."""
    await self._client.aclose()

query

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

Query index for top k most similar nodes.

Parameters:

Name	Type	Description	Default
query	VectorStoreQuery	Store query object.	required

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

def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
    """
    Query index for top k most similar nodes.

    Args:
        query (VectorStoreQuery): Store query object.

    """
    query_embedding = cast(List[float], query.query_embedding)

    return self._client.query(
        query.mode,
        query.query_str,
        query_embedding,
        query.similarity_top_k,
        filters=query.filters,
    )

aquery async

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

Async query index for top k most similar nodes.

Parameters:

Name	Type	Description	Default
query	VectorStoreQuery	Store query object.	required

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

async def aquery(
    self, query: VectorStoreQuery, **kwargs: Any
) -> VectorStoreQueryResult:
    """
    Async query index for top k most similar nodes.

    Args:
        query (VectorStoreQuery): Store query object.

    """
    query_embedding = cast(List[float], query.query_embedding)

    return await self._client.aquery(
        query.mode,
        query.query_str,
        query_embedding,
        query.similarity_top_k,
        filters=query.filters,
    )

OpensearchVectorClient

Object encapsulating an Opensearch index that has vector search enabled.

Index creation is deferred until first use (e.g., query, add, delete) rather than during construction. This avoids making network calls in init, which prevents RuntimeError: This event loop is already running in environments like Jupyter notebooks and FastAPI that already have a running event loop.

Parameters:

Name	Type	Description	Default
endpoint	str	URL (http/https) of elasticsearch endpoint	required
index	str	Name of the elasticsearch index	required
dim	int	Dimension of the vector	required
embedding_field	str	Name of the field in the index to store embedding array in.	'embedding'
text_field	str	Name of the field to grab text from	'content'
method	Optional[dict]	Opensearch "method" JSON obj for configuring the KNN index. This includes engine, metric, and other config params. Defaults to: {"name": "hnsw", "space_type": "l2", "engine": "faiss", "parameters": {"ef_construction": 256, "m": 48}}	None
settings	Optional[dict]	Optional[dict]: Settings for the Opensearch index creation. Defaults to: {"index": {"knn": True, "knn.algo_param.ef_search": 100}}	None
space_type	Optional[str]	space type for distance metric calculation. Defaults to: l2	'l2'
idx_conf	Optional[dict]	Optional[dict]: Custom index configuration for Opensearch. Defaults to None and will overwrite settings above.	None
os_client	Optional[Client]	Custom synchronous client (see OpenSearch from opensearch-py)	None
os_async_client	Optional[Client]	Custom asynchronous client (see AsyncOpenSearch from opensearch-py)	None
excluded_source_fields	Optional[List[str]]	Optional list of document "source" fields to exclude from OpenSearch responses.	None
**kwargs	Any	Optional arguments passed to the OpenSearch client from opensearch-py.	{}

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

class OpensearchVectorClient:
    """
    Object encapsulating an Opensearch index that has vector search enabled.

    Index creation is deferred until first use (e.g., query, add, delete) rather
    than during construction. This avoids making network calls in __init__, which
    prevents ``RuntimeError: This event loop is already running`` in environments
    like Jupyter notebooks and FastAPI that already have a running event loop.

    Args:
        endpoint (str): URL (http/https) of elasticsearch endpoint
        index (str): Name of the elasticsearch index
        dim (int): Dimension of the vector
        embedding_field (str): Name of the field in the index to store
            embedding array in.
        text_field (str): Name of the field to grab text from
        method (Optional[dict]): Opensearch "method" JSON obj for configuring
            the KNN index.
            This includes engine, metric, and other config params. Defaults to:
            {"name": "hnsw", "space_type": "l2", "engine": "faiss",
            "parameters": {"ef_construction": 256, "m": 48}}
        settings: Optional[dict]: Settings for the Opensearch index creation. Defaults to:
            {"index": {"knn": True, "knn.algo_param.ef_search": 100}}
        space_type (Optional[str]): space type for distance metric calculation. Defaults to: l2
        idx_conf: Optional[dict]: Custom index configuration for Opensearch. Defaults to None and will overwrite settings above.
        os_client (Optional[OSClient]): Custom synchronous client (see OpenSearch from opensearch-py)
        os_async_client (Optional[OSClient]): Custom asynchronous client (see AsyncOpenSearch from opensearch-py)
        excluded_source_fields (Optional[List[str]]): Optional list of document "source" fields to exclude from OpenSearch responses.
        **kwargs: Optional arguments passed to the OpenSearch client from opensearch-py.

    """

    def __init__(
        self,
        endpoint: str,
        index: str,
        dim: int,
        embedding_field: str = "embedding",
        text_field: str = "content",
        method: Optional[dict] = None,
        settings: Optional[dict] = None,
        engine: Optional[str] = "faiss",
        space_type: Optional[str] = "l2",
        idx_conf: Optional[dict] = None,
        max_chunk_bytes: int = 1 * 1024 * 1024,
        search_pipeline: Optional[str] = None,
        os_client: Optional[OSClient] = None,
        os_async_client: Optional[OSClient] = None,
        excluded_source_fields: Optional[List[str]] = None,
        **kwargs: Any,
    ):
        """Init params."""
        if engine == "nmslib":
            warnings.warn(
                "nmslib engine is deprecated in OpenSearch starting from version 3.0.0, consider using faiss or lucene instead.",
                FutureWarning,
            )
        if method is None:
            method = {
                "name": "hnsw",
                "space_type": "l2",
                "engine": engine,
                "parameters": {"ef_construction": 256, "m": 48},
            }
        if settings is None:
            settings = {"index": {"knn": True, "knn.algo_param.ef_search": 100}}
        if embedding_field is None:
            embedding_field = "embedding"

        self._method = method
        self._embedding_field = embedding_field
        self._endpoint = endpoint
        self._dim = dim
        self._index = index
        self._text_field = text_field
        self._max_chunk_bytes = max_chunk_bytes
        self._excluded_source_fields = excluded_source_fields

        self._search_pipeline = search_pipeline
        http_auth = kwargs.get("http_auth")
        self.space_type = space_type
        self.is_aoss = self._is_aoss_enabled(http_auth=http_auth)
        # initialize mapping
        if idx_conf is None:
            idx_conf = {
                "settings": settings,
                "mappings": {
                    "properties": {
                        embedding_field: {
                            "type": "knn_vector",
                            "dimension": dim,
                            "method": method,
                        },
                    }
                },
            }
        self._idx_conf = idx_conf
        self._owns_os_client = os_client is None
        self._owns_os_async_client = os_async_client is None
        self._os_client = os_client or self._get_opensearch_client(
            self._endpoint, **kwargs
        )
        self._os_async_client = os_async_client or self._get_async_opensearch_client(
            self._endpoint, **kwargs
        )
        self._initialized = False
        self._efficient_filtering_enabled = False

    def _import_opensearch(self) -> Any:
        """Import OpenSearch if available, otherwise raise error."""
        try:
            from opensearchpy import OpenSearch
        except ImportError:
            raise ImportError(IMPORT_OPENSEARCH_PY_ERROR)
        return OpenSearch

    def _import_async_opensearch(self) -> Any:
        """Import AsyncOpenSearch if available, otherwise raise error."""
        try:
            from opensearchpy import AsyncOpenSearch
        except ImportError:
            raise ImportError(IMPORT_ASYNC_OPENSEARCH_PY_ERROR)
        return AsyncOpenSearch

    def _import_bulk(self) -> Any:
        """Import bulk if available, otherwise raise error."""
        try:
            from opensearchpy.helpers import bulk
        except ImportError:
            raise ImportError(IMPORT_OPENSEARCH_PY_ERROR)
        return bulk

    def _import_async_bulk(self) -> Any:
        """Import async_bulk if available, otherwise raise error."""
        try:
            from opensearchpy.helpers import async_bulk
        except ImportError:
            raise ImportError(IMPORT_ASYNC_OPENSEARCH_PY_ERROR)
        return async_bulk

    def _import_not_found_error(self) -> Any:
        """Import not found error if available, otherwise raise error."""
        try:
            from opensearchpy.exceptions import NotFoundError
        except ImportError:
            raise ImportError(IMPORT_OPENSEARCH_PY_ERROR)
        return NotFoundError

    def _get_opensearch_client(self, opensearch_url: str, **kwargs: Any) -> Any:
        """Get OpenSearch client from the opensearch_url, otherwise raise error."""
        try:
            opensearch = self._import_opensearch()
            client = opensearch(opensearch_url, **kwargs)
        except ValueError as e:
            raise ImportError(
                f"OpenSearch client string provided is not in proper format. "
                f"Got error: {e} "
            )
        return client

    def _get_async_opensearch_client(self, opensearch_url: str, **kwargs: Any) -> Any:
        """Get AsyncOpenSearch client from the opensearch_url, otherwise raise error."""
        try:
            opensearch = self._import_async_opensearch()
            client = opensearch(opensearch_url, **kwargs)

        except ValueError as e:
            raise ValueError(
                f"AsyncOpenSearch client string provided is not in proper format. "
                f"Got error: {e} "
            )
        return client

    def _get_opensearch_version(self) -> str:
        info = self._os_client.info()
        return info["version"]["number"]

    async def _aget_opensearch_version(self) -> str:
        info = await self._os_async_client.info()
        return info["version"]["number"]

    def _ensure_initialized(self) -> None:
        """Lazily initialize the index on first use (sync)."""
        if self._initialized:
            return
        self._efficient_filtering_enabled = self._is_efficient_filtering_enabled()
        not_found_error = self._import_not_found_error()
        try:
            self._os_client.indices.get(index=self._index)
        except not_found_error:
            self._os_client.indices.create(index=self._index, body=self._idx_conf)
            if self.is_aoss:
                self._os_client.indices.exists(index=self._index)
            else:
                self._os_client.indices.refresh(index=self._index)
        self._initialized = True

    async def _async_ensure_initialized(self) -> None:
        """Lazily initialize the index on first use (async)."""
        if self._initialized:
            return
        self._efficient_filtering_enabled = (
            await self._async_is_efficient_filtering_enabled()
        )
        not_found_error = self._import_not_found_error()
        try:
            await self._os_async_client.indices.get(index=self._index)
        except not_found_error:
            await self._os_async_client.indices.create(
                index=self._index, body=self._idx_conf
            )
            if self.is_aoss:
                await self._os_async_client.indices.exists(index=self._index)
            else:
                await self._os_async_client.indices.refresh(index=self._index)
        self._initialized = True

    def _bulk_ingest_embeddings(
        self,
        client: Any,
        index_name: str,
        embeddings: List[List[float]],
        texts: Iterable[str],
        metadatas: Optional[List[dict]] = None,
        ids: Optional[List[str]] = None,
        vector_field: str = "embedding",
        text_field: str = "content",
        mapping: Optional[Dict] = None,
        max_chunk_bytes: Optional[int] = 1 * 1024 * 1024,
        is_aoss: bool = False,
    ) -> List[str]:
        """Bulk Ingest Embeddings into given index."""
        if not mapping:
            mapping = {}

        bulk = self._import_bulk()
        not_found_error = self._import_not_found_error()
        requests = []
        return_ids = []

        try:
            client.indices.get(index=index_name)
        except not_found_error:
            client.indices.create(index=index_name, body=mapping)

        for i, text in enumerate(texts):
            metadata = metadatas[i] if metadatas else {}
            _id = ids[i] if ids else str(uuid.uuid4())
            request = {
                "_op_type": "index",
                "_index": index_name,
                vector_field: embeddings[i],
                text_field: text,
                "metadata": metadata,
            }
            if is_aoss:
                request["id"] = _id
            else:
                request["_id"] = _id
            requests.append(request)
            return_ids.append(_id)

        bulk(client, requests, max_chunk_bytes=max_chunk_bytes)
        if not is_aoss:
            client.indices.refresh(index=index_name)

        return return_ids

    async def _abulk_ingest_embeddings(
        self,
        client: Any,
        index_name: str,
        embeddings: List[List[float]],
        texts: Iterable[str],
        metadatas: Optional[List[dict]] = None,
        ids: Optional[List[str]] = None,
        vector_field: str = "embedding",
        text_field: str = "content",
        mapping: Optional[Dict] = None,
        max_chunk_bytes: Optional[int] = 1 * 1024 * 1024,
        is_aoss: bool = False,
    ) -> List[str]:
        """Async Bulk Ingest Embeddings into given index."""
        if not mapping:
            mapping = {}

        async_bulk = self._import_async_bulk()
        not_found_error = self._import_not_found_error()
        requests = []
        return_ids = []

        try:
            await client.indices.get(index=index_name)
        except not_found_error:
            await client.indices.create(index=index_name, body=mapping)

        for i, text in enumerate(texts):
            metadata = metadatas[i] if metadatas else {}
            _id = ids[i] if ids else str(uuid.uuid4())
            request = {
                "_op_type": "index",
                "_index": index_name,
                vector_field: embeddings[i],
                text_field: text,
                "metadata": metadata,
            }
            if is_aoss:
                request["id"] = _id
            else:
                request["_id"] = _id
            requests.append(request)
            return_ids.append(_id)

        await async_bulk(client, requests, max_chunk_bytes=max_chunk_bytes)
        if not is_aoss:
            await client.indices.refresh(index=index_name)

        return return_ids

    def _default_approximate_search_query(
        self,
        query_vector: List[float],
        k: int = 4,
        filters: Optional[Union[Dict, List]] = None,
        vector_field: str = "embedding",
        excluded_source_fields: Optional[List[str]] = None,
    ) -> Dict:
        """For Approximate k-NN Search, this is the default query."""
        query = {
            "size": k,
            "query": {
                "knn": {
                    vector_field: {
                        "vector": query_vector,
                        "k": k,
                    }
                }
            },
        }

        if filters:
            # filter key must be added only when filtering to avoid "filter doesn't support values of type: START_ARRAY" exception
            query["query"]["knn"][vector_field]["filter"] = filters
        if excluded_source_fields:
            query["_source"] = {"exclude": excluded_source_fields}
        return query

    def _is_text_field(self, value: Any) -> bool:
        """
        Check if value is a string and keyword filtering needs to be performed.

        Not applied to datetime strings.
        """
        if isinstance(value, str):
            try:
                datetime.fromisoformat(value)
                return False
            except ValueError as e:
                return True
        else:
            return False

    def _parse_filter(self, filter: MetadataFilter) -> dict:
        """
        Parse a single MetadataFilter to equivalent OpenSearch expression.

        As Opensearch does not differentiate between scalar/array keyword fields, IN and ANY are equivalent.
        """
        key = f"metadata.{filter.key}"
        op = filter.operator

        equality_postfix = ".keyword" if self._is_text_field(value=filter.value) else ""

        if op == FilterOperator.EQ:
            return {"term": {f"{key}{equality_postfix}": filter.value}}
        elif op in [
            FilterOperator.GT,
            FilterOperator.GTE,
            FilterOperator.LT,
            FilterOperator.LTE,
        ]:
            return {"range": {key: {filter.operator.name.lower(): filter.value}}}
        elif op == FilterOperator.NE:
            return {
                "bool": {
                    "must_not": {"term": {f"{key}{equality_postfix}": filter.value}}
                }
            }
        elif op in [FilterOperator.IN, FilterOperator.ANY]:
            if isinstance(filter.value, list) and all(
                self._is_text_field(val) for val in filter.value
            ):
                return {"terms": {f"{key}.keyword": filter.value}}
            else:
                return {"terms": {key: filter.value}}
        elif op == FilterOperator.NIN:
            return {"bool": {"must_not": {"terms": {key: filter.value}}}}
        elif op == FilterOperator.ALL:
            return {
                "terms_set": {
                    key: {
                        "terms": filter.value,
                        "minimum_should_match_script": {"source": "params.num_terms"},
                    }
                }
            }
        elif op in (FilterOperator.TEXT_MATCH, FilterOperator.TEXT_MATCH_INSENSITIVE):
            return {"match": {key: {"query": filter.value, "fuzziness": "AUTO"}}}
        elif op == FilterOperator.CONTAINS:
            return {"wildcard": {key: f"*{filter.value}*"}}
        elif op == FilterOperator.IS_EMPTY:
            return {"bool": {"must_not": {"exists": {"field": key}}}}
        else:
            raise ValueError(f"Unsupported filter operator: {filter.operator}")

    def _parse_filters_recursively(self, filters: MetadataFilters) -> dict:
        """Parse (possibly nested) MetadataFilters to equivalent OpenSearch expression."""
        condition_map = {FilterCondition.AND: "must", FilterCondition.OR: "should"}

        bool_clause = condition_map[filters.condition]
        bool_query: dict[str, dict[str, list[dict]]] = {"bool": {bool_clause: []}}

        for filter_item in filters.filters:
            if isinstance(filter_item, MetadataFilter):
                bool_query["bool"][bool_clause].append(self._parse_filter(filter_item))
            elif isinstance(filter_item, MetadataFilters):
                bool_query["bool"][bool_clause].append(
                    self._parse_filters_recursively(filter_item)
                )
            else:
                raise ValueError(f"Unsupported filter type: {type(filter_item)}")

        return bool_query

    def _parse_filters(self, filters: Optional[MetadataFilters]) -> List[dict]:
        """Parse MetadataFilters to equivalent OpenSearch expression."""
        if filters is None:
            return []
        return [self._parse_filters_recursively(filters=filters)]

    def _knn_search_query(
        self,
        embedding_field: str,
        query_embedding: List[float],
        k: int,
        filters: Optional[MetadataFilters] = None,
        search_method="approximate",
        excluded_source_fields: Optional[List[str]] = None,
    ) -> Dict:
        """
        Perform a k-Nearest Neighbors (kNN) search.

        If the search method is "approximate" and the engine is "lucene" or "faiss", use efficient kNN filtering.
        Otherwise, perform an exhaustive exact kNN search using "painless scripting" if the version of
        OpenSearch supports it. If the OpenSearch version does not support it, use scoring script search.

        Note:
            - AWS OpenSearch Serverless does not support the painless scripting functionality at this time according to AWS.
            - Approximate kNN search does not support pre-filtering.

        Args:
            query_embedding (List[float]): Vector embedding to query.
            k (int): Maximum number of results.
            filters (Optional[MetadataFilters]): Optional filters to apply for the search.
                Supports filter-context queries documented at
                https://opensearch.org/docs/latest/query-dsl/query-filter-context/
            excluded_source_fields: Optional list of document "source" fields to exclude from the response.

        Returns:
            Dict: Up to k documents closest to query_embedding.

        """
        filters = self._parse_filters(filters)

        if not filters:
            search_query = self._default_approximate_search_query(
                query_embedding,
                k,
                vector_field=embedding_field,
                excluded_source_fields=excluded_source_fields,
            )
        elif (
            search_method == "approximate"
            and self._method["engine"]
            in [
                "lucene",
                "faiss",
            ]
            and self._efficient_filtering_enabled
        ):
            # if engine is lucene or faiss, opensearch recommends efficient-kNN filtering.
            search_query = self._default_approximate_search_query(
                query_embedding,
                k,
                filters={"bool": {"filter": filters}},
                vector_field=embedding_field,
                excluded_source_fields=excluded_source_fields,
            )
        else:
            if self.is_aoss:
                # if is_aoss is set we are using Opensearch Serverless AWS offering which cannot use
                # painless scripting so default scoring script returned will be just normal knn_score script
                search_query = self._default_scoring_script_query(
                    query_embedding,
                    k,
                    space_type=self.space_type,
                    pre_filter={"bool": {"filter": filters}},
                    vector_field=embedding_field,
                    excluded_source_fields=excluded_source_fields,
                )
            else:
                # https://opensearch.org/docs/latest/search-plugins/knn/painless-functions/
                search_query = self._default_scoring_script_query(
                    query_embedding,
                    k,
                    space_type="l2Squared",
                    pre_filter={"bool": {"filter": filters}},
                    vector_field=embedding_field,
                    excluded_source_fields=excluded_source_fields,
                )
        return search_query

    def _hybrid_search_query(
        self,
        text_field: str,
        query_str: str,
        embedding_field: str,
        query_embedding: List[float],
        k: int,
        filters: Optional[MetadataFilters] = None,
        excluded_source_fields: Optional[List[str]] = None,
    ) -> Dict:
        knn_query = self._knn_search_query(embedding_field, query_embedding, k, filters)
        lexical_query = self._lexical_search_query(text_field, query_str, k, filters)

        query = {
            "size": k,
            "query": {
                "hybrid": {"queries": [lexical_query["query"], knn_query["query"]]}
            },
        }
        if excluded_source_fields:
            query["_source"] = {"exclude": excluded_source_fields}
        return query

    def _lexical_search_query(
        self,
        text_field: str,
        query_str: str,
        k: int,
        filters: Optional[MetadataFilters] = None,
        excluded_source_fields: Optional[List[str]] = None,
    ) -> Dict:
        lexical_query = {
            "bool": {"must": {"match": {text_field: {"query": query_str}}}}
        }

        parsed_filters = self._parse_filters(filters)
        if len(parsed_filters) > 0:
            lexical_query["bool"]["filter"] = parsed_filters

        query = {
            "size": k,
            "query": lexical_query,
        }
        if excluded_source_fields:
            query["_source"] = {"exclude": excluded_source_fields}
        return query

    def __get_painless_scripting_source(
        self, space_type: str, vector_field: str = "embedding"
    ) -> str:
        """
        For Painless Scripting, it returns the script source based on space type.
        This does not work with Opensearch Serverless currently.
        """
        source_value = (
            f"(1.0 + {space_type}(params.query_value, doc['{vector_field}']))"
        )
        if space_type == "cosineSimilarity":
            return source_value
        else:
            return f"1/{source_value}"

    def _get_knn_scoring_script(self, space_type, vector_field, query_vector):
        """Default scoring script that will work with AWS Opensearch Serverless."""
        return {
            "source": "knn_score",
            "lang": "knn",
            "params": {
                "field": vector_field,
                "query_value": query_vector,
                "space_type": space_type,
            },
        }

    def _get_painless_scoring_script(self, space_type, vector_field, query_vector):
        source = self.__get_painless_scripting_source(space_type, vector_field)
        return {
            "source": source,
            "params": {
                "field": vector_field,
                "query_value": query_vector,
            },
        }

    def _default_scoring_script_query(
        self,
        query_vector: List[float],
        k: int = 4,
        space_type: str = "l2Squared",
        pre_filter: Optional[Union[Dict, List]] = None,
        vector_field: str = "embedding",
        excluded_source_fields: Optional[List[str]] = None,
    ) -> Dict:
        """
        For Scoring Script Search, this is the default query. Has to account for Opensearch Service
        Serverless which does not support painless scripting functions so defaults to knn_score.
        """
        if not pre_filter:
            pre_filter = MATCH_ALL_QUERY

        # check if we can use painless scripting or have to use default knn_score script
        if self.is_aoss:
            if space_type == "l2Squared":
                raise ValueError(
                    "Unsupported space type for aoss. Can only use l1, l2, cosinesimil."
                )
            script = self._get_knn_scoring_script(
                space_type, vector_field, query_vector
            )
        else:
            script = self._get_painless_scoring_script(
                space_type, vector_field, query_vector
            )
        query = {
            "size": k,
            "query": {
                "script_score": {
                    "query": pre_filter,
                    "script": script,
                }
            },
        }
        if excluded_source_fields:
            query["_source"] = {"exclude": excluded_source_fields}
        return query

    def _is_aoss_enabled(self, http_auth: Any) -> bool:
        """Check if the service is http_auth is set as `aoss`."""
        return (
            http_auth is not None
            and hasattr(http_auth, "service")
            and http_auth.service == "aoss"
        )

    @staticmethod
    def _version_supports_efficient_filtering(version: str) -> bool:
        """Return True if the OpenSearch version supports efficient kNN filtering."""
        major, minor, _patch = version.split(".")
        return int(major) > 2 or (int(major) == 2 and int(minor) >= 9)

    def _is_efficient_filtering_enabled(self) -> bool:
        """Check if kNN with efficient filtering is enabled."""
        # Technically, AOSS supports efficient filtering,
        # but we can't check the version number using .info(); AOSS doesn't support 'GET /'
        #  so we must skip and disable by default.
        if self.is_aoss:
            return False
        self._os_version = self._get_opensearch_version()
        return self._version_supports_efficient_filtering(self._os_version)

    async def _async_is_efficient_filtering_enabled(self) -> bool:
        """Async check if kNN with efficient filtering is enabled."""
        if self.is_aoss:
            return False
        self._os_version = await self._aget_opensearch_version()
        return self._version_supports_efficient_filtering(self._os_version)

    def index_results(self, nodes: List[BaseNode], **kwargs: Any) -> List[str]:
        """Store results in the index."""
        self._ensure_initialized()
        embeddings: List[List[float]] = []
        texts: List[str] = []
        metadatas: List[dict] = []
        ids: List[str] = []
        for node in nodes:
            ids.append(node.node_id)
            embeddings.append(node.get_embedding())
            texts.append(node.get_content(metadata_mode=MetadataMode.NONE))
            metadatas.append(node_to_metadata_dict(node, remove_text=True))

        return self._bulk_ingest_embeddings(
            self._os_client,
            self._index,
            embeddings,
            texts,
            metadatas=metadatas,
            ids=ids,
            vector_field=self._embedding_field,
            text_field=self._text_field,
            mapping=None,
            max_chunk_bytes=self._max_chunk_bytes,
            is_aoss=self.is_aoss,
        )

    async def aindex_results(self, nodes: List[BaseNode], **kwargs: Any) -> List[str]:
        """Store results in the index."""
        await self._async_ensure_initialized()
        embeddings: List[List[float]] = []
        texts: List[str] = []
        metadatas: List[dict] = []
        ids: List[str] = []
        for node in nodes:
            ids.append(node.node_id)
            embeddings.append(node.get_embedding())
            texts.append(node.get_content(metadata_mode=MetadataMode.NONE))
            metadatas.append(node_to_metadata_dict(node, remove_text=True))

        return await self._abulk_ingest_embeddings(
            self._os_async_client,
            self._index,
            embeddings,
            texts,
            metadatas=metadatas,
            ids=ids,
            vector_field=self._embedding_field,
            text_field=self._text_field,
            mapping=None,
            max_chunk_bytes=self._max_chunk_bytes,
            is_aoss=self.is_aoss,
        )

    def delete_by_doc_id(self, doc_id: str) -> None:
        """
        Deletes all OpenSearch documents corresponding to the given LlamaIndex `Document` ID.

        Args:
            doc_id (str): a LlamaIndex `Document` id

        """
        self._ensure_initialized()
        search_query = {
            "query": {"term": {"metadata.doc_id.keyword": {"value": doc_id}}}
        }
        self._os_client.delete_by_query(
            index=self._index, body=search_query, refresh=True
        )

    async def adelete_by_doc_id(self, doc_id: str) -> None:
        """
        Deletes all OpenSearch documents corresponding to the given LlamaIndex `Document` ID.

        Args:
            doc_id (str): a LlamaIndex `Document` id

        """
        await self._async_ensure_initialized()
        search_query = {
            "query": {"term": {"metadata.doc_id.keyword": {"value": doc_id}}}
        }
        await self._os_async_client.delete_by_query(
            index=self._index, body=search_query, refresh=True
        )

    def delete_nodes(
        self,
        node_ids: Optional[List[str]] = None,
        filters: Optional[MetadataFilters] = None,
        **delete_kwargs: Any,
    ) -> None:
        """
        Deletes nodes.

        Args:
            node_ids (Optional[List[str]], optional): IDs of nodes to delete. Defaults to None.
            filters (Optional[MetadataFilters], optional): Metadata filters. Defaults to None.

        """
        self._ensure_initialized()
        if not node_ids and not filters:
            return

        query = {"query": {"bool": {"filter": []}}}
        if node_ids:
            query["query"]["bool"]["filter"].append({"terms": {"_id": node_ids or []}})

        if filters:
            query["query"]["bool"]["filter"].extend(self._parse_filters(filters))

        self._os_client.delete_by_query(index=self._index, body=query, refresh=True)

    async def adelete_nodes(
        self,
        node_ids: Optional[List[str]] = None,
        filters: Optional[MetadataFilters] = None,
        **delete_kwargs: Any,
    ) -> None:
        """
        Deletes nodes.

        Args:
            node_ids (Optional[List[str]], optional): IDs of nodes to delete. Defaults to None.
            filters (Optional[MetadataFilters], optional): Metadata filters. Defaults to None.

        """
        await self._async_ensure_initialized()
        if not node_ids and not filters:
            return

        query = {"query": {"bool": {"filter": []}}}
        if node_ids:
            query["query"]["bool"]["filter"].append({"terms": {"_id": node_ids or []}})

        if filters:
            query["query"]["bool"]["filter"].extend(self._parse_filters(filters))

        await self._os_async_client.delete_by_query(
            index=self._index, body=query, refresh=True
        )

    def clear(self) -> None:
        """Clears index."""
        self._ensure_initialized()
        query = {"query": {"bool": {"filter": []}}}
        self._os_client.delete_by_query(index=self._index, body=query, refresh=True)

    async def aclear(self) -> None:
        """Clears index."""
        await self._async_ensure_initialized()
        query = {"query": {"bool": {"filter": []}}}
        await self._os_async_client.delete_by_query(
            index=self._index, body=query, refresh=True
        )

    def close(self) -> None:
        """
        Close the OpenSearch clients and release resources.

        Only closes clients that were created internally by this class.
        Clients passed in by the user are not closed.
        """
        if self._owns_os_client and self._os_client is not None:
            self._os_client.close()
        if self._owns_os_async_client and self._os_async_client is not None:
            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                asyncio.run(self._os_async_client.close())
            else:
                loop.create_task(self._os_async_client.close())

    async def aclose(self) -> None:
        """
        Asynchronously close the OpenSearch clients and release resources.

        Only closes clients that were created internally by this class.
        Clients passed in by the user are not closed.
        """
        if self._owns_os_client and self._os_client is not None:
            self._os_client.close()
        if self._owns_os_async_client and self._os_async_client is not None:
            await self._os_async_client.close()

    def __del__(self) -> None:
        """Clean up OpenSearch clients during garbage collection."""
        try:
            self.close()
        except Exception as exc:
            logger.debug(
                "Failed to close OpenSearch clients during garbage collection, "
                "type=%s err='%s'",
                type(exc),
                exc,
            )

    def query(
        self,
        query_mode: VectorStoreQueryMode,
        query_str: Optional[str],
        query_embedding: List[float],
        k: int,
        filters: Optional[MetadataFilters] = None,
    ) -> VectorStoreQueryResult:
        self._ensure_initialized()
        if query_mode == VectorStoreQueryMode.HYBRID:
            if query_str is None or self._search_pipeline is None:
                raise ValueError(INVALID_HYBRID_QUERY_ERROR)
            search_query = self._hybrid_search_query(
                self._text_field,
                query_str,
                self._embedding_field,
                query_embedding,
                k,
                filters=filters,
                excluded_source_fields=self._excluded_source_fields,
            )
            params = {
                "search_pipeline": self._search_pipeline,
            }
        elif query_mode == VectorStoreQueryMode.TEXT_SEARCH:
            search_query = self._lexical_search_query(
                self._text_field,
                query_str,
                k,
                filters=filters,
                excluded_source_fields=self._excluded_source_fields,
            )
            params = None
        else:
            search_query = self._knn_search_query(
                self._embedding_field,
                query_embedding,
                k,
                filters=filters,
                excluded_source_fields=self._excluded_source_fields,
            )
            params = None

        res = self._os_client.search(
            index=self._index, body=search_query, params=params
        )

        return self._to_query_result(res)

    async def aquery(
        self,
        query_mode: VectorStoreQueryMode,
        query_str: Optional[str],
        query_embedding: List[float],
        k: int,
        filters: Optional[MetadataFilters] = None,
    ) -> VectorStoreQueryResult:
        await self._async_ensure_initialized()
        if query_mode == VectorStoreQueryMode.HYBRID:
            if query_str is None or self._search_pipeline is None:
                raise ValueError(INVALID_HYBRID_QUERY_ERROR)
            search_query = self._hybrid_search_query(
                self._text_field,
                query_str,
                self._embedding_field,
                query_embedding,
                k,
                filters=filters,
                excluded_source_fields=self._excluded_source_fields,
            )
            params = {
                "search_pipeline": self._search_pipeline,
            }
        elif query_mode == VectorStoreQueryMode.TEXT_SEARCH:
            search_query = self._lexical_search_query(
                self._text_field,
                query_str,
                k,
                filters=filters,
                excluded_source_fields=self._excluded_source_fields,
            )
            params = None
        else:
            search_query = self._knn_search_query(
                self._embedding_field,
                query_embedding,
                k,
                filters=filters,
                excluded_source_fields=self._excluded_source_fields,
            )
            params = None

        res = await self._os_async_client.search(
            index=self._index, body=search_query, params=params
        )

        return self._to_query_result(res)

    def _to_query_result(self, res) -> VectorStoreQueryResult:
        nodes = []
        ids = []
        scores = []
        for hit in res["hits"]["hits"]:
            source = hit["_source"]
            node_id = hit["_id"]
            text = source[self._text_field]
            metadata = source.get("metadata", None)

            try:
                node = metadata_dict_to_node(metadata)
                node.text = text
            except Exception:
                # TODO: Legacy support for old nodes
                node_info = source.get("node_info")
                relationships = source.get("relationships") or {}
                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,
                )
            ids.append(node_id)
            nodes.append(node)
            scores.append(hit["_score"])

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

index_results

index_results(
    nodes: List[BaseNode], **kwargs: Any
) -> List[str]

Store results in the index.

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

def index_results(self, nodes: List[BaseNode], **kwargs: Any) -> List[str]:
    """Store results in the index."""
    self._ensure_initialized()
    embeddings: List[List[float]] = []
    texts: List[str] = []
    metadatas: List[dict] = []
    ids: List[str] = []
    for node in nodes:
        ids.append(node.node_id)
        embeddings.append(node.get_embedding())
        texts.append(node.get_content(metadata_mode=MetadataMode.NONE))
        metadatas.append(node_to_metadata_dict(node, remove_text=True))

    return self._bulk_ingest_embeddings(
        self._os_client,
        self._index,
        embeddings,
        texts,
        metadatas=metadatas,
        ids=ids,
        vector_field=self._embedding_field,
        text_field=self._text_field,
        mapping=None,
        max_chunk_bytes=self._max_chunk_bytes,
        is_aoss=self.is_aoss,
    )

aindex_results async

aindex_results(
    nodes: List[BaseNode], **kwargs: Any
) -> List[str]

Store results in the index.

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

async def aindex_results(self, nodes: List[BaseNode], **kwargs: Any) -> List[str]:
    """Store results in the index."""
    await self._async_ensure_initialized()
    embeddings: List[List[float]] = []
    texts: List[str] = []
    metadatas: List[dict] = []
    ids: List[str] = []
    for node in nodes:
        ids.append(node.node_id)
        embeddings.append(node.get_embedding())
        texts.append(node.get_content(metadata_mode=MetadataMode.NONE))
        metadatas.append(node_to_metadata_dict(node, remove_text=True))

    return await self._abulk_ingest_embeddings(
        self._os_async_client,
        self._index,
        embeddings,
        texts,
        metadatas=metadatas,
        ids=ids,
        vector_field=self._embedding_field,
        text_field=self._text_field,
        mapping=None,
        max_chunk_bytes=self._max_chunk_bytes,
        is_aoss=self.is_aoss,
    )

delete_by_doc_id

delete_by_doc_id(doc_id: str) -> None

Deletes all OpenSearch documents corresponding to the given LlamaIndex Document ID.

Parameters:

Name	Type	Description	Default
doc_id	str	a LlamaIndex Document id	required

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

def delete_by_doc_id(self, doc_id: str) -> None:
    """
    Deletes all OpenSearch documents corresponding to the given LlamaIndex `Document` ID.

    Args:
        doc_id (str): a LlamaIndex `Document` id

    """
    self._ensure_initialized()
    search_query = {
        "query": {"term": {"metadata.doc_id.keyword": {"value": doc_id}}}
    }
    self._os_client.delete_by_query(
        index=self._index, body=search_query, refresh=True
    )

adelete_by_doc_id async

adelete_by_doc_id(doc_id: str) -> None

Deletes all OpenSearch documents corresponding to the given LlamaIndex Document ID.

Parameters:

Name	Type	Description	Default
doc_id	str	a LlamaIndex Document id	required

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

async def adelete_by_doc_id(self, doc_id: str) -> None:
    """
    Deletes all OpenSearch documents corresponding to the given LlamaIndex `Document` ID.

    Args:
        doc_id (str): a LlamaIndex `Document` id

    """
    await self._async_ensure_initialized()
    search_query = {
        "query": {"term": {"metadata.doc_id.keyword": {"value": doc_id}}}
    }
    await self._os_async_client.delete_by_query(
        index=self._index, body=search_query, refresh=True
    )

delete_nodes

delete_nodes(
    node_ids: Optional[List[str]] = None,
    filters: Optional[MetadataFilters] = None,
    **delete_kwargs: Any
) -> None

Deletes nodes.

Parameters:

Name	Type	Description	Default
node_ids	Optional[List[str]]	IDs of nodes to delete. Defaults to None.	None
filters	Optional[MetadataFilters]	Metadata filters. Defaults to None.	None

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

def delete_nodes(
    self,
    node_ids: Optional[List[str]] = None,
    filters: Optional[MetadataFilters] = None,
    **delete_kwargs: Any,
) -> None:
    """
    Deletes nodes.

    Args:
        node_ids (Optional[List[str]], optional): IDs of nodes to delete. Defaults to None.
        filters (Optional[MetadataFilters], optional): Metadata filters. Defaults to None.

    """
    self._ensure_initialized()
    if not node_ids and not filters:
        return

    query = {"query": {"bool": {"filter": []}}}
    if node_ids:
        query["query"]["bool"]["filter"].append({"terms": {"_id": node_ids or []}})

    if filters:
        query["query"]["bool"]["filter"].extend(self._parse_filters(filters))

    self._os_client.delete_by_query(index=self._index, body=query, refresh=True)

adelete_nodes async

adelete_nodes(
    node_ids: Optional[List[str]] = None,
    filters: Optional[MetadataFilters] = None,
    **delete_kwargs: Any
) -> None

Deletes nodes.

Parameters:

Name	Type	Description	Default
node_ids	Optional[List[str]]	IDs of nodes to delete. Defaults to None.	None
filters	Optional[MetadataFilters]	Metadata filters. Defaults to None.	None

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

async def adelete_nodes(
    self,
    node_ids: Optional[List[str]] = None,
    filters: Optional[MetadataFilters] = None,
    **delete_kwargs: Any,
) -> None:
    """
    Deletes nodes.

    Args:
        node_ids (Optional[List[str]], optional): IDs of nodes to delete. Defaults to None.
        filters (Optional[MetadataFilters], optional): Metadata filters. Defaults to None.

    """
    await self._async_ensure_initialized()
    if not node_ids and not filters:
        return

    query = {"query": {"bool": {"filter": []}}}
    if node_ids:
        query["query"]["bool"]["filter"].append({"terms": {"_id": node_ids or []}})

    if filters:
        query["query"]["bool"]["filter"].extend(self._parse_filters(filters))

    await self._os_async_client.delete_by_query(
        index=self._index, body=query, refresh=True
    )

clear

clear() -> None

Clears index.

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

def clear(self) -> None:
    """Clears index."""
    self._ensure_initialized()
    query = {"query": {"bool": {"filter": []}}}
    self._os_client.delete_by_query(index=self._index, body=query, refresh=True)

aclear async

aclear() -> None

Clears index.

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

async def aclear(self) -> None:
    """Clears index."""
    await self._async_ensure_initialized()
    query = {"query": {"bool": {"filter": []}}}
    await self._os_async_client.delete_by_query(
        index=self._index, body=query, refresh=True
    )

close

close() -> None

Close the OpenSearch clients and release resources.

Only closes clients that were created internally by this class. Clients passed in by the user are not closed.

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

def close(self) -> None:
    """
    Close the OpenSearch clients and release resources.

    Only closes clients that were created internally by this class.
    Clients passed in by the user are not closed.
    """
    if self._owns_os_client and self._os_client is not None:
        self._os_client.close()
    if self._owns_os_async_client and self._os_async_client is not None:
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            asyncio.run(self._os_async_client.close())
        else:
            loop.create_task(self._os_async_client.close())

aclose async

aclose() -> None

Asynchronously close the OpenSearch clients and release resources.

Only closes clients that were created internally by this class. Clients passed in by the user are not closed.

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

async def aclose(self) -> None:
    """
    Asynchronously close the OpenSearch clients and release resources.

    Only closes clients that were created internally by this class.
    Clients passed in by the user are not closed.
    """
    if self._owns_os_client and self._os_client is not None:
        self._os_client.close()
    if self._owns_os_async_client and self._os_async_client is not None:
        await self._os_async_client.close()

options: members: - OpensearchVectorStore