fastrag.ingestion.steps.embedding.base#

The Embedder family base.

docs/specification/subclasses.md: Embedder is the DocumentStep every concrete embedder (Model2VecEmbedder, OpenAIEmbedder) extends. Its one job beyond DocumentStep itself is attaching a dense vector to each incoming document while coalescing the underlying model calls into batches of up to policy.batch_size texts, rather than issuing one call per document.

DocumentStep (task 003) drives caching, retries and per-item concurrency around a single-item transform hook — there is no batch-shaped hook on the base to plug into, and adding one would mean changing a foundation three already-finished tasks (parsing, chunking) build on. So batching is a detail private to this family: Embedder.transform() still satisfies the single-item hook (each cache decision, retry and ItemFailed stays exactly what DocumentStep already gives every other step), but internally it hands its text to a small per-instance batcher that coalesces concurrently pending calls into one Embedder.embed() call once policy.batch_size texts have queued, or after a short window lets slower siblings catch up. Nothing here re-implements caching or retries — a batch that fails fails every text in it, and each of those documents’ own DocumentStep-level retry (if any) resubmits into a fresh batch.

Classes

Embedder(*[, cache, policy])

Attaches a dense vector to each incoming document.
class fastrag.ingestion.steps.embedding.base.Embedder(*, cache=None, policy=None)[source]#

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

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

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