fastrag.ingestion.steps.embedding.model2vec#

Model2VecEmbedder: local static embeddings, computed in this process.

A local, CPU-backed Embedder: loads a small static model into memory and encodes in-process, so indexing needs no embedding API key.

Classes

Model2VecEmbedder(*[, model, dimensions, ...])

Embed with a static model, computed in this process.
class fastrag.ingestion.steps.embedding.model2vec.Model2VecEmbedder(*, model='minishlab/potion-base-8M', dimensions=None, normalize=None, batch_size=1024, token=None, cache=None, policy=None)[source]#

Bases: Embedder

Embed with a static model, computed in this process.

Choose this over OpenAIEmbedder when the run must not depend on a server: a CI pipeline, a laptop with no endpoint, a demo. The model is a lookup table of token vectors averaged per text — no attention, no GPU, no server — which is why it is fast and why it is weaker than a transformer encoder on paraphrase.

from fastrag.ingestion.steps.embedding import Model2VecEmbedder

embedder = Model2VecEmbedder(model="minishlab/potion-base-8M", cache=cache)
Parameters:

  • model (str) – HuggingFace model id or a local directory of weights. Defaults to minishlab/potion-base-8M: ~30 MB, 256 dimensions.

  • dimensions (int | None) – Truncate the model’s output to this width. None keeps the model’s full width. Changing it also changes model_id (and therefore the document-cache identity), so a truncated model never shares a cached vector with the full one.

  • normalize (bool | None) – Whether to L2-normalize vectors. None keeps the model’s own default (potion models ship normalized).

  • batch_size (int) – Texts per model2vec encode call. Distinct from policy.batch_size (how many documents this step’s batcher groups into one call to embed()) — this instead bounds memory inside one such call, exactly as in the old service.

  • token (str | None) – HuggingFace access token, for gated models.

  • cache (CacheStore | Literal[False] | None)

  • policy (Policy | None)

The first construction downloads ~30 MB into ~/.cache/huggingface; every later one reads the cache, and HF_HUB_OFFLINE=1 forbids the network outright. Pass a local directory as model for a fully air-gapped install.

config()[source]#

This embedder’s configuration, for cache-identity hashing.

Return type:

Mapping[str, Any]

property model_id: str#

The model, suffixed with the pinned dimension when there is one.

property dimension: int#

The width of every returned vector, read from the loaded model.

async embed(texts)[source]#

Embed a batch of texts.

The work happens in a worker thread: encode is synchronous and CPU-bound, and offloading it keeps other batches’ concurrent work (cache lookups, other steps) running while this one encodes.

Parameters:

texts (Sequence[str])

Return type:

list[list[float]]

property cache: CacheStore | Literal[False] | None#

Return the cache explicitly configured for this step.

property config_hash: str#

Invalidate embeddings when the shared input serialization changes.

effective_policy(context)#

Resolve this step’s policy over the one inherited from context.

The step’s own explicitly-set fields take priority; everything else falls through to the pipeline (or enclosing router/aggregator) policy.

Parameters:

context (ExecutionContext)

Return type:

Policy

async evaluate(context)#

Drive transform() over each upstream item, cached per item.

Raises:

fastrag.exceptions.ConfigurationError – No upstream is bound — call step(upstream) before evaluating.

Parameters:

context (ExecutionContext)

Return type:

AsyncIterator[Document]

async evaluate_items(items, context)#

Transform a bounded cache window independently of worker concurrency.

Parameters:
Return type:

AsyncIterator[Document]

event_name = 'Embedding'#

The stage this step’s work belongs to, e.g. "Chunking".

Declared as a plain class attribute by a family base class, and inherited by every concrete step in that family:

class Chunker(DocumentStep, ABC):
    event_name = "Chunking"


class ChunkerRouter(RoutedStep[Document], Chunker):
    ...  # -> "Chunking", inherited from `Chunker`

A concrete step overrides it to opt out of its family’s label:

class ChunkerRouter(RoutedStep[Document], Chunker):
    event_name = "ChunkerRouter"  # -> "ChunkerRouter"

Resolution walks the MRO for the first class that declares it in its own class body (so a mixin listed first, e.g. RoutedStep, contributes nothing unless it declares one itself), and falls back to the class’s own name when nothing in the MRO does — a step family that never declares an event_name still labels each of its steps distinctly rather than collapsing them under one blank heading. That fallback is computed per instance rather than assigned onto the class: assigning it would make it look “declared” to every subclass, which would then inherit their base’s name instead of getting their own.

property policy: Policy | None#

Return the policy explicitly configured for this step.

replay_needs_content = False#

Set by a step whose cached replay reads the full documents anyway (e.g. to re-validate them): its cache reads then skip the header-only projection rather than fetching every entry twice.

async run()#

Execute this step independently and collect every item.

Builds a default ExecutionContext when none is installed (this step’s own policy, or a fresh Policy). Intended primarily for testing and standalone use — pipeline-wide event observation is Pipeline’s job.

Return type:

list[T]

async transform(item, context)#

Embed the versioned input, preserving document display content.

The actual model call this document’s text takes part in may be shared with other documents pending at the same time — see the module docstring — but from the caller’s perspective this awaits one vector for one document, exactly like any other DocumentStep transform.

This method only runs on a cache miss for this embedder — a cached document is replayed by DocumentStep without calling it at all — so the yielded document’s own cached is always reset to False here, regardless of whatever the upstream item (e.g. a chunk already served from the chunker’s own cache) carried in. Emitting DocumentEmbedded is DocumentStep._on_produced’s job, not this method’s, precisely so the event fires the same way for a cache hit as for a fresh embed.

Raises:

fastrag.exceptions.EmbeddingError – item.metadata["content_format"] is DOCLING — a serialized DoclingDocument, not text. A document with no recorded content_format is treated as TEXT (the default for every document that predates content_format — see ContentFormatFilter), so nothing that embedded before this check existed starts failing now.

Parameters:
Return type:

AsyncIterator[Document]

property upstream: PipelineStep[Any] | None#

Return the upstream step bound via __call__, or None.

Read-only counterpart to __call__’s self._upstream assignment — lets external code (e.g. Pipeline, walking a graph from its root to classify each step by position) introspect composition without reaching into a private attribute.

validate_input(item)#

Validate input before work or cache replay; legacy steps accept all.

Parameters:

item (Any)

Return type:

None

name#