fastrag.ingestion.steps.embedding.openai#

OpenAIEmbedder: OpenAI-compatible embeddings, via langchain-openai.

An Embedder that delegates to OpenAI’s embeddings API (or any OpenAI-compatible endpoint) — options and transient-error classification as usual for an Embedder.

Module Attributes

KNOWN_DIMENSIONS

Output widths known without a network call.

Classes

OpenAIEmbedder(*, model, api_key[, ...])

Embed with any OpenAI-compatible endpoint.
fastrag.ingestion.steps.embedding.openai.KNOWN_DIMENSIONS: dict[str, int] = {'text-embedding-3-large': 3072, 'text-embedding-3-small': 1536, 'text-embedding-ada-002': 1536}#

Output widths known without a network call. dimension resolves eagerly at construction because a vector store names its collection from it; any other model must pass dimensions= explicitly.

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

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

async embed(texts)[source]#

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