Skip to main content

agrag.vectordb.milvus.MilvusVectorStore

MilvusVectorStore(*, settings:MilvusSettings | None = None, client:Any | None = None, tracer:Tracer | None = None) -> None

Bases: VectorStore

A VectorStore backed by Milvus, including native hybrid search.

The client connects lazily on first use, so constructing the store does not open a network connection. Milvus performs BM25 server-side, so hybrid search needs no client-side sparse embedder; the sparse vector is computed by a Milvus Function from the text field on write and at query time.

Functions:

  • close – Release the backend connection.
  • collection_exists – Report whether a collection exists.
  • commit_pending – Promote one job's staged records to committed records.
  • count – Count records in a collection.
  • delete – Delete records by id.
  • delete_collection – Delete a collection and all its entities.
  • delete_pending – Delete one job's staged records, leaving committed records alone.
  • ensure_collection – Create the collection if it does not exist.
  • hybrid_search – Search by dense vector and keyword text in one fused call.
  • initialize – Check connectivity and authentication.
  • invalidate_collection – Drop cached distance-metric knowledge of a collection.
  • retrieve – Fetch records by id.
  • scroll – Iterate records in a collection, in batches.
  • search – Search by dense vector only.
  • upsert – Write or overwrite records in a collection.

Parameters:

  • settings (MilvusSettings | None) – Milvus connection settings. Defaults to MilvusSettings().
  • client (Any | None) – A pre-built AsyncMilvusClient, for tests. When set, __init__ imports nothing and the store calls this object directly instead of building one.
  • tracer (Tracer | None) – Opens this store's spans.

close​

close() -> None

Release the backend connection.

collection_exists​

collection_exists(name:str) -> bool

Report whether a collection exists.

Parameters:

  • name (str) – The collection name.

Returns:

  • bool – True if the collection exists.

commit_pending​

commit_pending(collection:str, *, job_id:UUID) -> None

Promote one job's staged records to committed records.

Each staged record is written under its real id, which replaces any committed record with that id, and the staged copy is deleted. The method reads the first page of staged records again after each delete, so it needs no offset. Re-running after a failure finishes the remaining records.

Parameters:

  • collection (str) – The collection the job wrote to.
  • job_id (UUID) – The committed job whose records become visible.

Raises:

  • RuntimeError – A delete left the same staged records in place.

count​

count(collection:str, *, filters:dict[str, Any] | None = None, pending_job_id:UUID | None = None) -> int

Count records in a collection.

Parameters:

  • collection (str) – The collection to count.
  • filters (dict[str, Any] | None) – A flat-dict filter on scalar fields.
  • pending_job_id (UUID | None) – Count only this job's staged records.

Returns:

  • int – The number of matching records.

delete​

delete(collection:str, ids:Sequence[UUID]) -> None

Delete records by id.

Parameters:

  • collection (str) – The collection to delete from.
  • ids (Sequence[UUID]) – The ids to delete.

delete_collection​

delete_collection(name:str) -> None

Delete a collection and all its entities.

Parameters:

  • name (str) – The collection name.

delete_pending​

delete_pending(collection:str, *, job_id:UUID) -> None

Delete one job's staged records, leaving committed records alone.

Parameters:

  • collection (str) – The collection the job wrote to.
  • job_id (UUID) – The rolled-back job whose records are deleted.

Raises:

  • RuntimeError – A delete left the same staged records in place.

ensure_collection​

ensure_collection(name:str, *, dimensions:int, distance:Distance, hybrid:bool = False) -> None

Create the collection if it does not exist.

Milvus performs BM25 server-side, so the sparse field and its Function are always provisioned; the hybrid flag is accepted for interface parity but is a no-op here. An existing collection must already carry this same fixed schema, since upsert and hybrid_search always read and write every field regardless of hybrid.

Parameters:

  • name (str) – The collection name.
  • dimensions (int) – The embedding dimension.
  • distance (Distance) – The distance metric new collections use.
  • hybrid (bool) – Accepted for interface parity; ignored by Milvus.

Raises:

hybrid_search(collection:str, query_vector:Sequence[float], query_text:str, *, limit:int = 10, filters:dict[str, Any] | None = None, alpha:float = 0.5) -> list[VectorHit]

Search by dense vector and keyword text in one fused call.

Fusion uses Milvus's native weighted reranker, which normalizes each request's scores before applying alpha.

Parameters:

  • collection (str) – The collection to search.
  • query_vector (Sequence[float]) – The dense query embedding.
  • query_text (str) – The query text, matched by BM25.
  • limit (int) – The maximum number of hits to return.
  • filters (dict[str, Any] | None) – A flat-dict filter on scalar fields.
  • alpha (float) – The dense/keyword balance. 1.0 is pure dense, 0.0 is pure keyword.

Returns:

  • list[VectorHit] – The fused hits, highest score first.

Raises:

  • ValueError – limit is not positive, or exceeds MAX_SEARCH_LIMIT, or alpha is outside [0.0, 1.0].

initialize​

initialize() -> None

Check connectivity and authentication.

invalidate_collection​

invalidate_collection(name:str) -> None

Drop cached distance-metric knowledge of a collection.

This store caches a collection's distance metric after the first call that resolves it, on the assumption that it alone (via ensure_collection/delete_collection) owns the collection's lifecycle for as long as this instance is in use. If something outside this instance deletes and recreates a collection under the same name with a different metric, call this first so the next call re-resolves that collection's metric from the backend instead of trusting the stale cache.

Parameters:

  • name (str) – The collection name.

retrieve​

retrieve(collection:str, ids:Sequence[UUID]) -> list[VectorRecord]

Fetch records by id.

Requests at most MAX_RESPONSE_LIMIT ids per call, so a large ids list cannot exceed Milvus's response-size ceiling in one request the way sending every id at once would.

Parameters:

  • collection (str) – The collection to read.
  • ids (Sequence[UUID]) – The ids to fetch.

Returns:

scroll​

scroll(collection:str, *, limit:int = 100, page_offset:str | None = None, filters:dict[str, Any] | None = None, with_vectors:bool = False, pending_job_id:UUID | None = None) -> tuple[list[VectorRecord], str | None]

Iterate records in a collection, in batches.

Milvus rejects a query whose offset + limit exceeds MAX_RESPONSE_LIMIT, so a numeric offset cannot page past that many total records. Pages instead cursor on the id primary key: each page filters on id > page_offset and orders by id ascending, which needs no offset at all and so never hits that window regardless of collection size. The explicit order is load bearing: without it, an unordered query result could omit rows at or below the next cursor, permanently skipping them on the next page.

Parameters:

  • collection (str) – The collection to read.
  • limit (int) – The maximum number of records per page.
  • page_offset (str | None) – The id cursor from a previous scroll call.
  • filters (dict[str, Any] | None) – A flat-dict filter on scalar fields.
  • with_vectors (bool) – Whether to return each record's vector.
  • pending_job_id (UUID | None) – Read only this job's staged records.

Returns:

  • list[VectorRecord] – The page of records and the next page cursor, or None at the
  • str | None – end.
search(collection:str, query_vector:Sequence[float], *, limit:int = 10, filters:dict[str, Any] | None = None) -> list[VectorHit]

Search by dense vector only.

Parameters:

  • collection (str) – The collection to search.
  • query_vector (Sequence[float]) – The dense query embedding.
  • limit (int) – The maximum number of hits to return.
  • filters (dict[str, Any] | None) – A flat-dict filter on scalar fields.

Returns:

  • list[VectorHit] – The matched hits, highest score first.

Raises:

  • ValueError – limit is not positive, or exceeds MAX_SEARCH_LIMIT.

upsert​

upsert(collection:str, records:Sequence[VectorRecord], *, batch_size:int = 256, pending_job_id:UUID | None = None) -> None

Write or overwrite records in a collection.

Parameters:

  • collection (str) – The collection to write to.
  • records (Sequence[VectorRecord]) – The records to upsert, in order.
  • batch_size (int) – The number of records per backend write call. Must be positive.
  • pending_job_id (UUID | None) – The in-flight job staging these records.

Raises:

  • ValueError – batch_size is not positive, or a payload uses a key the store reserves for pending records (_pending, _pending_job_id, _target_id).