fastrag.vector_store.milvus#

MilvusVectorStore, via pymilvus’ async client.

A VectorStore backed by Milvus. It needs only what the sink, sync and the read path use — open, ensure the collection, check which ids already exist, upsert, list and delete ids, search, close. Filtering is not ported: MilvusVectorStore.search() takes a backend-native filter string directly, per search()’s own signature.

Module Attributes

ID_MAX_LENGTH

Length caps of the VARCHAR columns, checked locally so one oversized document fails with a message naming the field instead of a Milvus batch rejection.

DEFAULT_COLLECTION

Collection name used when the caller does not name one.

ID_PAGE_SIZE

Ids per query page in MilvusVectorStore.all_ids(), below Milvus's 16384-row cap on one query's results.

Classes

MilvusVectorStore(*, url[, collection, ...])

Vectors in a Milvus or Milvus-Lite server.
fastrag.vector_store.milvus.ID_MAX_LENGTH = 64#

Length caps of the VARCHAR columns, checked locally so one oversized document fails with a message naming the field instead of a Milvus batch rejection.

fastrag.vector_store.milvus.DEFAULT_COLLECTION = 'fastrag_documents'#

Collection name used when the caller does not name one.

fastrag.vector_store.milvus.ID_PAGE_SIZE = 10000#

Ids per query page in MilvusVectorStore.all_ids(), below Milvus’s 16384-row cap on one query’s results.

class fastrag.vector_store.milvus.MilvusVectorStore(*, url, collection='fastrag_documents', user=None, password=None, metric_type='COSINE', enable_bm25=False, timeout=30.0, policy=None)[source]#

Bases: VectorStore, LexicalStore

Vectors in a Milvus or Milvus-Lite server.

Choose this when the corpus outgrows in-process search, or when several processes index or query the same data. Construction never touches the network — the pymilvus client is built in __aenter__ and closed in __aexit__.

async with MilvusVectorStore(url=MILVUS_URI) as vector_store:
    saved_documents = vector_store.save(embedded_documents)
    async for document in saved_documents:
        ...
Parameters:

  • url (str) – Server endpoint, e.g. http://localhost:19530.

  • collection (str) – Name of the Milvus collection documents are upserted into. Created automatically, on demand, if absent.

  • user (str | None) – Username, combined with password into the token Milvus authenticates with. None connects without credentials.

  • password (str | None) – Password paired with user.

  • metric_type (str) – Distance metric for the vector index. COSINE suits embedding models whose vectors are not length-normalized; L2 is the other common choice.

  • enable_bm25 (bool) – Create an analyzed content field, BM25 function and sparse index. Requires Milvus 2.5+. Existing dense-only collections must be reindexed into a new collection to use this option.

  • timeout (float) – Seconds per client operation.

  • policy (Policy | None) – Execution policy for the terminal step save() returns.

async ensure_ready(dimension)[source]#

Create collection for dimension-wide vectors if absent.

Idempotent: safe to call on every run, including one where the collection already exists. The schema mirrors Document: id and provenance are columns so they filter without reading the JSON, and the content and metadata live alongside the vector.

Parameters:

dimension (int)

Return type:

None

async upsert(documents)[source]#

Insert or replace documents by id, in one request.

Raises:

VectorStoreError – A document is missing its embedding (checked again defensively — the terminal step should already have rejected it), its UTF-8 content exceeds the column cap, or its serialized metadata exceeds the JSON cap.

Parameters:

documents (Sequence[Document])

Return type:

None

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

Return at most k documents nearest embedding, best first.

One search call, requesting search_params={"metric_type": self.metric_type} explicitly so the query is scored the same way the collection’s own index was built (ensure_ready()). pymilvus already returns hits best-first for whichever metric that is, so no reordering is needed: score is each hit’s own distance field, and rank its position in that already-correct order.

Queries with consistency_level="Strong" for read-your-writes: pymilvus’s client-wide default, "Bounded", may not yet reflect an upsert() this same process just issued, and a caller retrieving right after indexing must see it.

Parameters:
Return type:

list[Document]

Rank text using Milvus’s server-side BM25 function and sparse index.

Parameters:
Return type:

list[Document]

async existing_ids(ids)[source]#

Query collection once for which of ids are already indexed.

One query call for the whole batch, filtering on the primary key (id in [...]) — never one call per id — so this scales with the number of batches save()’s terminal step flushes, not with the number of documents in them.

Queries with consistency_level="Strong" for read-your-writes: pymilvus’s client-wide default, "Bounded", may not yet reflect an upsert this same process just issued — a staleness window here is not an acceptable trade for the latency it would save.

Parameters:

ids (Sequence[str])

Return type:

set[str]

async all_ids()[source]#

Page through every id in collection, in id order.

AsyncMilvusClient has no query_iterator, so this pages the way pymilvus’s own iterator does: iterator asks the server for each page sorted by primary key, and the next page filters on ids after the last one seen. offset paging would stop at Milvus’s 16384-row result window.

Return type:

set[str]

async delete(ids)[source]#

Delete ids by primary key in one request.

Parameters:

ids (Sequence[str])

Return type:

None

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]