fastrag.ingestion.steps.embedding

Embedding: documents in, the same documents with vectors attached out.

class fastrag.ingestion.steps.embedding.Embedder(*, cache=None, policy=None)

Bases: DocumentStep, ABC

Attaches a dense vector to each incoming document.

Binding an upstream (embedder(chunked_documents), via the inherited __call__) and evaluating it (async for document in embedder / run()) never embeds eagerly — only consuming the resulting stream does.

Caching (per document id and this embedder’s own configuration) and retries are entirely DocumentStep’s job, applied around transform(); nothing here re-implements either. Only the underlying model call is batched, up to policy.batch_size texts per call — see the module docstring.

Writing one is a class plus three members:

from collections.abc import Sequence

from fastrag.ingestion.steps.embedding import Embedder


class EchoEmbedder(Embedder):
    def __init__(self, *, width: int = 8, cache=False, policy=None):
        super().__init__(cache=cache, policy=policy)
        self.width = width

    def config(self):
        return {"width": self.width}

    @property
    def model_id(self) -> str:
        return f"echo@{self.width}"

    @property
    def dimension(self) -> int:
        return self.width

    async def embed(self, texts: Sequence[str]) -> list[list[float]]:
        return [[float(len(text))] * self.width for text in texts]

Every event an embedder emits is labelled "Embedding", inherited by every concrete embedder (Model2VecEmbedder, OpenAIEmbedder, …) unless it declares an event_name of its own.

Parameters:

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

• policy (Policy | None)

event_name: str = '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 config_hash: str

Invalidate embeddings when the shared input serialization changes.

abstract property model_id: str

Stable identifier of the model configuration.

Changing it — a different model, or a pinned dimension — must change a subclass’s config() (and therefore its config_hash), so vectors of different models or widths never share a document-cache entry.

abstract property dimension: int

The width of every returned vector, known at construction.

Resolved eagerly — never behind a network call — so a caller can name a vector-store collection from it before anything is embedded.

abstractmethod async embed(texts)

Embed a batch of texts, in order.

Called with at most policy.batch_size texts at a time — the batcher in this base class is what enforces that ceiling, not the caller. Must return exactly one vector per text, in the same order.

Parameters:

texts (Sequence[str])

Return type:

list[list[float]]

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:

• item (Document)

• context (ExecutionContext)

Return type:

AsyncIterator[Document]

property cache: CacheStore | Literal[False] | None

Return the cache explicitly configured for this step.

abstractmethod config()

Return this step’s own configuration, for cache-identity hashing.

Every concrete step declares this explicitly — steps are plain classes, not dataclasses, so there is no automatic field introspection. A step that forgets it cannot be instantiated: config is abstract, so construction itself fails loudly rather than config_hash silently collapsing to a constant.

Return type:

Mapping[str, Any]

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:

• items (AsyncIterator[Any])

• context (ExecutionContext)

Return type:

AsyncIterator[Document]

property policy: Policy | None

Return the policy explicitly configured for this step.

replay_needs_content: bool = 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]

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: str

class fastrag.ingestion.steps.embedding.Model2VecEmbedder(*, model='minishlab/potion-base-8M', dimensions=None, normalize=None, batch_size=1024, token=None, cache=None, policy=None)

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()

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)

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:

• items (AsyncIterator[Any])

• context (ExecutionContext)

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:

• item (Document)

• context (ExecutionContext)

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

class fastrag.ingestion.steps.embedding.OpenAIEmbedder(*, model, api_key, base_url=None, dimensions=None, batch_size=256, timeout=60.0, extra_body=None, cache=None, policy=None)

Bases: Embedder

Embed with any OpenAI-compatible endpoint.

Choose this for OpenAI’s models or for a self-hosted server that speaks the OpenAI protocol (vLLM, Ollama, text-generation-inference). Point base_url at the self-hosted endpoint; the only things that change are the model name and, for a model Fastrag has never heard of, dimensions.

from fastrag.ingestion.steps.embedding import OpenAIEmbedder

embedder = OpenAIEmbedder(
    model="text-embedding-3-small",
    api_key=api_key,
    cache=cache,
)

Parameters:

• model (str) – Model name, e.g. text-embedding-3-small or qwen3-embedding:8b.

• api_key (str) – API key. Self-hosted endpoints usually accept any non-empty string.

• base_url (str | None) – Endpoint base URL. None uses the OpenAI API. When set, context-length checking is disabled: langchain then posts strings rather than tiktoken token-id arrays, which is what an Ollama/vLLM-style server expects.

• dimensions (int | None) – Pinned output width. None infers the width of a known OpenAI model and raises for anything else — a self-hosted model must pin it explicitly. Pinning also forces a model that can shorten its output (text-embedding-3-*) to one stable width, which is what keeps a collection’s vectors comparable. Explicit values are sent to the provider and require provider/model support; inferred widths do not add a dimensions request field. Returned counts and widths are validated before indexing or querying.

• batch_size (int) – Texts per provider request — langchain calls this chunk_size. Distinct from policy.batch_size (how many documents this step’s own batcher groups into one call to embed()); this instead bounds a single such call, exactly as in the old service.

• extra_body (Mapping[str, Any] | None) – Additional JSON request fields; matching fields override defaults.

• timeout (float) – Seconds per provider call.

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

• policy (Policy | None)

Raises:

fastrag.exceptions.ConfigurationError – The “openai” extra is not installed, dimensions is not a positive integer, batch_size < 1, or the dimension of model cannot be inferred and dimensions was not given.

config()

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, resolved at construction.

async embed(texts)

Embed a batch of texts.

No provider exception escapes: everything is translated into the Fastrag hierarchy, with timeouts, rate limits, and server errors marked transient so Policy can retry them.

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:

• items (AsyncIterator[Any])

• context (ExecutionContext)

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:

• item (Document)

• context (ExecutionContext)

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

Modules

base	The Embedder family base.
model2vec	Model2VecEmbedder: local static embeddings, computed in this process.
openai	OpenAIEmbedder: OpenAI-compatible embeddings, via langchain-openai.