fastrag.vector_store.base

The VectorStore sink family base.

A VectorStore is the pipeline’s storage backend, not a PipelineStep itself: a caller opens it once (async with MilvusVectorStore(url=...) as vector_store:) and hands an upstream DocumentStep to save(), which returns the terminal PipelineStep[Document] that indexes when evaluated. Constructing that step performs no I/O.

Batching mirrors fastrag.ingestion.steps.embedding.base.Embedder: each document is queued into a private micro-batcher that coalesces pending documents into one existence check plus (at most) one upsert() call once policy.batch_size documents have queued, or after a short window lets slower siblings catch up. The store is the source of truth for “already indexed”: existing_ids() is asked once per batch, present ids are skipped and reported cached=True, the rest are upserted and reported cached=False. Under a policy with recreate_cache, nothing counts as already indexed: every document is upserted, overwriting what the store holds for its id, as a rebuild expects.

sync() wraps that same step and, once the upstream is exhausted, deletes every stored document the run did not yield, so the store ends up holding exactly the run’s documents. It deletes only after a clean run: a failed item’s documents are missing from the run without being gone from the source.

VectorStore and its concrete stores live at the top level, in fastrag.vector_store, rather than under fastrag.ingestion: a vector store is queried at read time as well as written at ingest time, so it is a storage backend the whole SDK depends on.

Classes

VectorStore(*[, policy])

Sink backend for persisting processed and embedded documents.

class fastrag.vector_store.base.VectorStore(*, policy=None)

Bases: AbstractAsyncContextManager[VectorStore], ABC

Sink backend for persisting processed and embedded documents.

Opened once per run (async with MilvusVectorStore(url=...) as vector_store:) and handed the final upstream step via save(), which returns a lazy terminal PipelineStep — nothing is indexed until that step is evaluated.

Parameters:

policy (Policy | None)

abstractmethod async ensure_ready(dimension)

Create the backing collection for dimension-wide vectors if absent.

Idempotent and cheap to call again on a collection that already exists; called once, lazily, before the first upsert, with the vector width taken from the first document flowing through the terminal step.

Parameters:

dimension (int) – Width of the vectors this run’s documents carry.

Return type:

None

abstractmethod async upsert(documents)

Insert or replace documents by id, in one call.

Idempotent: rerunning with the same ids converges rather than duplicating rows. Called with at most self.policy.batch_size documents at a time — the batcher in save()’s returned step enforces that ceiling, not the caller — and possibly fewer, since documents already reported by existing_ids() are excluded.

Parameters:

documents (Sequence[Document]) – The batch to upsert. Every document carries a non-None embedding — the terminal step never lets one through otherwise.

Return type:

None

abstractmethod async existing_ids(ids)

Return the subset of ids already present in the store.

Called once per pending batch, never once per document, so the round trips it adds scale with the number of batches. An id returned here is treated as already indexed (skipped, reported cached=True); an absent id is upserted and reported cached=False. Skipping a present document is an optimisation, not a correctness requirement — a store unable to answer cheaply may upsert everything, but must then report which ids were genuinely present rather than guessing.

Parameters:

ids (Sequence[str]) – Candidate document ids to check, at most self.policy.batch_size at a time.

Return type:

set[str]

abstractmethod async search(embedding, *, k, filter=None)

Return at most k documents nearest embedding, best first.

The read-path counterpart to upsert(), queried by VectorRetriever and, through it, RerankingRetriever. Never opens the store — the caller (a retriever) holds an already-open one.

Stamps two reserved metadata keys on every returned document, alongside its stored content/metadata (including source_uri):

• score: the backend’s own relevance score, on the backend’s own scale — never comparable across backends or metrics.

• rank: 0-based position in the returned (best-first) list.

The list is best-first regardless of the configured metric’s raw sort direction, so position 0 is always the best match.

Parameters:

• embedding (Sequence[float]) – Query vector, the same width the store’s collection was created with.

• k (int) – Maximum number of documents to return. Fewer come back if the collection holds fewer than k matching documents — this is not an error.

• filter (str | None) – Backend-native filter expression, applied server-side. None returns the k nearest documents unfiltered.

Return type:

list[Document]

async all_ids()

Return the id of every document in the store.

Needed only by sync(), so a store that does not support it may leave this unimplemented. Returns an empty set when the backing collection does not exist yet.

Return type:

set[str]

async delete(ids)

Remove the documents with ids; ids the store does not hold are ignored.

Needed only by sync(), which calls it with at most 1000 ids at a time.

Parameters:

ids (Sequence[str])

Return type:

None

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:

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

• skip_existing (bool) – Passed through to save().

Return type:

PipelineStep[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.