fastrag.vector_store#

fastrag.vector_store — the pipeline’s storage backend, read and written.

Re-exports VectorStore, the sink-family base every concrete vector store implements, and ChromaVectorStore, MilvusVectorStore and OpenSearchVectorStore. Milvus and OpenSearch also implement LexicalStore for BM25 search; Chroma does not. Lives at the top level rather than under fastrag.ingestion: a vector store is queried at read time as well as written at ingest time, so it is not an ingestion-only concern.

class fastrag.vector_store.ChromaVectorStore(*, url=None, path=None, collection='fastrag_documents', headers=None, tenant='default_tenant', database='default_database', metric_type='cosine', policy=None)[source]#

Bases: VectorStore

Vectors in a Chroma server, or in an embedded on-disk Chroma database.

Give url to talk to a Chroma server through its async HTTP client, or path to run Chroma in-process against a local directory (its client is synchronous, so calls run in a worker thread). Construction never touches the network or the disk — the client is built in __aenter__.

async with ChromaVectorStore(url="http://localhost:8000") as vector_store:
    async for document in vector_store.save(embedded_documents):
        ...

filter is a JSON-encoded Chroma where object, e.g. '{"source_uri": "https://example.org/docs"}'. score is Chroma’s own distance, so lower is better; results are still best first.

Parameters:

url (str | None) – Server endpoint, e.g. http://localhost:8000. Exclusive with path.
path (str | None) – Directory of an embedded persistent database. Exclusive with url.
collection (str) – Name of the Chroma collection. Created on demand.
headers (Mapping[str, str] | None) – Extra HTTP headers, e.g. an auth token, sent with url.
tenant (str) – Chroma tenant the collection lives in.
database (str) – Chroma database the collection lives in.
metric_type (str) – HNSW distance for a new collection: cosine, l2 or ip. An existing collection must already use it.
policy (Policy | None) – Execution policy for the terminal step save() returns.

async ensure_ready(dimension)[source]#

Create collection if absent, or validate its metric and width.

Chroma fixes a collection’s width on its first insert rather than at creation, so an existing collection is checked against one stored vector; an empty one accepts any width.

Parameters:: dimension (int)
Return type:: None

async upsert(documents)[source]#

Insert or replace documents by id, in one call.

Parameters:: documents (Sequence[Document])
Return type:: None

async existing_ids(ids)[source]#

Fetch the whole batch by id in one get, returning ids only.

Parameters:: ids (Sequence[str])
Return type:: set[str]

async all_ids()[source]#

Page through every id with get; empty if the collection does not exist.

Return type:: set[str]

async delete(ids)[source]#

Delete ids in one call; Chroma ignores ids it does not hold.

Parameters:: ids (Sequence[str])
Return type:: None

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

Return at most k documents nearest embedding, best first.

score is Chroma’s distance for the collection’s metric, where lower means closer; rank is the position in Chroma’s already nearest-first order.

Parameters:

embedding (Sequence[float])
k (int)
filter (str | None)

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:

upstream (PipelineStep[Document]) – Final upstream step yielding embedded documents.
skip_existing (bool) – Passed through to save().

Return type:

PipelineStep[Document]

class fastrag.vector_store.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:

embedding (Sequence[float])
k (int)
filter (str | None)

Return type:

list[Document]

async lexical_search(query, *, k, filter=None)[source]#

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

Parameters:

query (str)
k (int)
filter (str | None)

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:

upstream (PipelineStep[Document]) – Final upstream step yielding embedded documents.
skip_existing (bool) – Passed through to save().

Return type:

PipelineStep[Document]

class fastrag.vector_store.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:

url (str)
index (str)
user (str | None)
password (str | None)
timeout (float)
policy (Policy | None)

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:

embedding (Sequence[float])
k (int)
filter (str | None)

Return type:

list[Document]

async lexical_search(query, *, k, filter=None)[source]#

Search the analyzed content field using BM25.

Parameters:

query (str)
k (int)
filter (str | None)

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:

upstream (PipelineStep[Document]) – Final upstream step yielding embedded documents.
skip_existing (bool) – Passed through to save().

Return type:

PipelineStep[Document]

class fastrag.vector_store.VectorStore(*, policy=None)[source]#

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)[source]#

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)[source]#

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)[source]#

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)[source]#

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()[source]#

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)[source]#

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)[source]#

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)[source]#

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.

Modules

`base`	The `VectorStore` sink family base.
`chroma`	`ChromaVectorStore`, via chromadb's async HTTP client or an embedded store.
`milvus`	`MilvusVectorStore`, via pymilvus' async client.
`opensearch`	Async OpenSearch storage with BM25 and dense k-NN search over one index.