fastrag.vector_store.opensearch#

Async OpenSearch storage with BM25 and dense k-NN search over one index.

Uses OpenSearch’s REST API through the SDK’s existing httpx dependency. Lucene HNSW requires the k-NN plugin; OpenSearch 2.12+ supports the 4096-dimensional embeddings used by the chatbot example. Fusion happens in the SDK, so no server search pipeline or fusion plugin is required.

Module Attributes

ID_PAGE_SIZE

Hits per _search page in OpenSearchVectorStore.all_ids(), the default index.max_result_window.

Classes

OpenSearchVectorStore(*, url[, index, user, ...])

Persist and search documents in an OpenSearch index.
fastrag.vector_store.opensearch.ID_PAGE_SIZE = 10000#

Hits per _search page in OpenSearchVectorStore.all_ids(), the default index.max_result_window.

class fastrag.vector_store.opensearch.OpenSearchVectorStore(*, url, index='fastrag_documents', user=None, password=None, timeout=30.0, policy=None)[source]#

Bases: VectorStore, LexicalStore

Persist and search documents in an OpenSearch index.

filter is a JSON-encoded Query DSL object, e.g. '{"term": {"source_uri": "https://example.org/docs"}}'. Metadata is retained in _source without dynamic indexing. HTTPS uses username and password authentication with TLS certificate verification disabled.

Parameters:
async ensure_ready(dimension)[source]#

Create the index or validate its existing vector and text mappings.

Parameters:

dimension (int)

Return type:

None

async existing_ids(ids)[source]#

Check the entire batch in one real-time multi-get request.

Parameters:

ids (Sequence[str])

Return type:

set[str]

async all_ids()[source]#

Page through every id sorted on the id keyword, with search_after.

Returns an empty set when the index does not exist yet.

Return type:

set[str]

async delete(ids)[source]#

Bulk delete by id and refresh so searches immediately stop seeing them.

Parameters:

ids (Sequence[str])

Return type:

None

async upsert(documents)[source]#

Bulk index by id and refresh so searches immediately see writes.

Parameters:

documents (Sequence[Document])

Return type:

None

async search(embedding, *, k, filter=None)[source]#

Search dense vectors with cosine similarity and optional native filter.

Parameters:
Return type:

list[Document]

Search the analyzed content field using BM25.

Parameters:
Return type:

list[Document]

save(upstream, *, skip_existing=False)#

Return a lazy terminal step that indexes upstream’s documents.

Constructing the returned step performs no I/O — evaluating it (async for document in vector_store.save(upstream): or .run()) is what upserts. Each document is queued into a batcher that checks existing_ids() and flushes to upsert() once self.policy.batch_size documents have queued, or after a short window; every document is yielded back once its batch is resolved, with cached set to whether the store already held it, and DocumentSaved emitted for it.

Parameters:

  • upstream (PipelineStep[Document]) – Final upstream step yielding embedded documents.

  • skip_existing (bool) – For an identity-preserving Embedder, check destination ids before embedding. Indexed items bypass vector retrieval and produce cached DocumentEmbedded and DocumentSaved events. Cached events and results may contain only identity/metadata with empty content and no vectors. Custom content filters disable header projection. Use a separate destination for each embedding configuration.

Return type:

PipelineStep[Document]

Either way, recreate_cache (on the store’s policy or the pipeline’s) skips the existence check and upserts every document.

sync(upstream, *, skip_existing=False)#

Like save(), then delete every stored document upstream did not yield.

After a clean run the store holds exactly the documents the run produced: documents whose source disappeared or changed (a changed chunk gets a new id) are removed. Deletion happens once the upstream is exhausted, so a run that raises or is stopped early deletes nothing. A run with any ItemFailed deletes nothing either: a failed item’s documents are missing from the run, not from the source. Either way a DocumentsPruned event reports the outcome.

The run must cover everything the store should keep — use one store (collection/index) per synced pipeline.

Parameters:
Return type:

PipelineStep[Document]