fastrag.ingestion.steps.base#

PipelineStep, ExecutionContext, and the policy-aware evaluator.

This module is the canonical home for the lazy step contract every concrete step (fetcher, parser, chunker, embedder, vector store) builds on:

Constructing a step, and composing one with step(upstream), does no I/O and emits no event — only evaluating a step does work.
PipelineStep.evaluate stays abstract: a SourceStep subclass (a leaf producer, e.g. a fetcher) implements it directly with its own fetch logic. DocumentStep instead implements evaluate concretely here, driving a per-item transform hook (which subclasses implement) over whatever upstream is bound to it — because a document transform is always “one upstream item in, zero or more items out”, while a source has no such upstream shape to hang caching off of.
The module-level _run_step is “the evaluator”: the single function both PipelineStep.__aiter__ and every DocumentStep/RoutedStep use to pull from an upstream, so lifecycle events (StepStarted/StepProgress/ StepCompleted/StepFailed) and the SourceStep cache wrap apply uniformly at every level of a composed graph, not just at the outermost step a caller happens to call run() on.
run_concurrently is the shared concurrency + retry + ItemFailed primitive. DocumentStep.evaluate is built on it, and it is exported so a later SourceStep (a fetcher with several targets to reach concurrently) can reuse it too, instead of writing its own retry loop.

Caching itself is intentionally split by step family: by default, SourceStep caches its entire evaluate() stream as one bucket keyed by source_cache_id(name, config_hash, config_hash). A step that declares an optional targets hook (duck-typed like is_source_modified) opts into per-target caching instead — one bucket per target (e.g. one per crawled page), via evaluate_target(target, context). DocumentStep caches per upstream item, keyed by document_cache_id(name, item.id, config_hash), because document transforms are pure and each input independently determines its output. RoutedStep (e.g. ParserRouter, ChunkerRouter) is the one exception to “a step caches by its own identity”: it forces its own cache to False so DocumentStep’s generic per-item wrap never runs for it, and instead caches each item by whichever child step the router dispatched it to (document_cache_id(target.name, item.id, target.config_hash), in _dispatch_route) — otherwise every route would share one cache bucket keyed by the router’s own config, and switching a route would silently keep serving the old route’s cached output.

Module Attributes

StepConfigHash

A stable, short hex digest identifying one step's effective configuration.

Functions

`event_sink_scope`(sink)	Install `sink` as the ambient event sink for the duration of the block.
`run_concurrently`(items, worker, *, context, ...)	The shared concurrency, retry, and `ItemFailed` primitive.
`run_source_step`(step, context)	Evaluate `step` through the same wrap `_run_step` applies internally.
`run_step`(step, context)	Evaluate any `step` through the same wrap `_run_step` applies internally.
`stream_batches`(items, size)	Collect a bounded stream window, flushing the final partial batch.

Classes

`DocumentStep`(*[, cache, policy])	Base step transforming or generating `Document` instances.
`ExecutionContext`(pipeline_id, policy[, ...])	Runtime context passed to step evaluators.
`PipelineStep`(*[, cache, policy])	Base class for every executable pipeline step.
`RoutedStep`(*[, cache, policy])	Step that conditionally delegates to child steps based on filter rules.
`SourceStep`(*[, cache, policy, ...])	Base step responsible for generating raw `Source` entities.

fastrag.ingestion.steps.base.StepConfigHash#: A stable, short hex digest identifying one step’s effective configuration.

fastrag.ingestion.steps.base.event_sink_scope(sink)[source]#

Install sink as the ambient event sink for the duration of the block.

A step’s run()/__aiter__ builds its own ExecutionContext when none is threaded to it explicitly, and reads the sink installed here to populate it — this is how “a step nested three levels deep reaches the sink without any step passing it along” holds even when a step is evaluated deep inside another step’s own logic. A Pipeline (task 009) is expected to call this around the whole run it drives.

Parameters:: sink (Callable[[PipelineEvent], Awaitable[None]] | None) – The event sink to install, or None to explicitly install “no sink” for the scope (distinct from simply not calling this at all, which leaves whatever ambient sink — if any — already set).
Return type:: Iterator[None]

class fastrag.ingestion.steps.base.ExecutionContext(pipeline_id, policy, cache=None, event_sink=None, cache_headers_only=False, document_cache_entries=None)[source]#

Bases: object

Runtime context passed to step evaluators.

Parameters:

pipeline_id (str)
policy (Policy)
cache (CacheStore | None)
event_sink (Callable[[PipelineEvent], Awaitable[None]] | None)
cache_headers_only (bool)
document_cache_entries (Mapping[tuple[int, str], list[Document] | None] | None)

pipeline_id#

Global identifier for the executing pipeline run.

Type:: str

policy#

Active execution policy — a step without its own falls back to this.

Type:: fastrag.ingestion.policy.Policy

cache#

Active cache store backend — a step whose own cache is None inherits this one; False disables caching regardless.

Type:: fastrag.ingestion.cache.base.CacheStore | None

event_sink#

Where emitted events go. None means “no sink installed”: emit() becomes a no-op rather than raising, so a step run on its own works without a Pipeline.

Type:: collections.abc.Callable[[fastrag.ingestion.events.PipelineEvent], collections.abc.Awaitable[None]] | None

async emit(event)[source]#

Dispatch event to the registered sink, or do nothing.

Parameters:: event (PipelineEvent) – Event instance to propagate.
Return type:: None

class fastrag.ingestion.steps.base.PipelineStep(*, cache=None, policy=None)[source]#

Bases: ABC, Generic[T]

Base class for every executable pipeline step.

Constructing a step, and composing it with another via __call__, must never execute work. A step can be evaluated independently with run() / async for, or composed into a larger graph and evaluated once the graph’s terminal step is run.

Every step carries two names, and every event it emits carries both: name, identifying this instance (its own class name by default), and event_name, identifying the stage its work belongs to. See event_name for how the latter is resolved.

Parameters:

cache (CacheStore | Literal[False] | None)
policy (Policy | None)

name: str#

event_name: str#

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 cache: CacheStore | Literal[False] | None#: Return the cache explicitly configured for this step.

property policy: Policy | None#: Return the policy explicitly configured for this step.

effective_policy(context)[source]#

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

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.

abstractmethod config()[source]#

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]

property config_hash: str#

A stable hash of this step’s class name and config().

Two identically configured steps of the same class hash equal; changing any declared config value changes the hash; the hash is stable across processes. Used as part of every cache id this step’s evaluation reads or writes.

abstractmethod evaluate(context)[source]#

Lazily evaluate the step within an execution context.

Parameters:: context (ExecutionContext) – Execution-scoped state and services.
Yields:: Values produced by the step.
Return type:: AsyncIterator[T]

async run()[source]#

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 fastrag.ingestion.steps.base.run_concurrently(items, worker, *, context, policy, item_id=<function _item_identifier>, step='', event_name='', policy_for=None)[source]#

The shared concurrency, retry, and ItemFailed primitive.

Applies worker to each item from items, bounded by policy.concurrency calls in flight at once, retrying a call that raises TransientError or TimeoutError per policy.retries/backoff/max_backoff/timeout. A call that still fails after retries emits ItemFailed(uri=item_id(item), error=..., step=step) through context and is skipped (policy.on_error in {"skip", "retry"}) or re-raised (policy.on_error == "raise", propagating after the ItemFailed). Output order follows completion order, not input order.

This is the primitive DocumentStep’s evaluator is built on, and the one a concrete step with several independent targets to reach at once (e.g. a fetcher crawling many URLs) should call instead of writing its own retry loop.

worker must not itself return None on success: None is used internally to mean “this call failed” (see _bounded_map).

Parameters:

items (AsyncIterator[T]) – The upstream items to process.
worker (Callable[[T], Awaitable[R]]) – Called once per item; its result is yielded on success.
context (ExecutionContext) – Execution-scoped state and services, used to emit ItemFailed.
policy (Policy) – Governs concurrency, retries, backoff, timeout, and on_error disposition.
item_id (Callable[[T], str]) – Builds the human-readable identifier recorded on ItemFailed for a given item. Defaults to _item_identifier.
step (str) – Name of the step this call is driving, recorded on any ItemFailed this raises so a consumer (e.g. a ConsoleLogger) can attribute the failure to the right stage. Optional and empty by default — callers that predate this parameter keep working, just with an unattributed ItemFailed.step == "".
event_name (str) – Stage label of that step (PipelineStep.event_name), recorded on any ItemFailed this raises. step names the instance; this names the column a consumer files the failure under. Optional and empty by default, for the same reason step is.
policy_for (Callable[[T], Policy] | None) – Resolves the policy governing one item, for a step whose items are not all handled the same way — a RoutedStep dispatching each item to a child that may declare its own policy. Only the per-item knobs come from it (retries, backoff, max_backoff, timeout, on_error); concurrency is a property of the stream, not of an item, and always comes from policy. None (the default) uses policy for every item.

Return type:

AsyncIterator[R]

class fastrag.ingestion.steps.base.SourceStep(*, cache=None, policy=None, cache_validation_interval=None)[source]#

Bases: PipelineStep[Source], ABC

Base step responsible for generating raw Source entities.

Source steps (fetchers, crawlers, and aggregators like docs/specification/subclasses.md’s Sources) implement evaluate() directly — sourcing has no upstream item to hang a per-item transform off of, unlike DocumentStep. Caching is still applied uniformly: _run_step wraps every SourceStep’s evaluation in the source cache before calling evaluate at all.

Declares event_name = "Fetching" for the whole family: sourcing is the fetching stage, whether the step is a leaf fetcher or a Sources aggregate of several — the aggregate must land in the same column as the fetchers it wraps, not one of its own. DocumentStep deliberately declares none: “a document transform” is not itself a stage, so each transform family (Parser, Chunker, Embedder, …) declares its own and anything else falls back to its own class name.

Parameters:

cache (CacheStore | Literal[False] | None)
policy (Policy | None)
cache_validation_interval (float | None)

event_name: str = 'Fetching'#

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 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]

property config_hash: str#

A stable hash of this step’s class name and config().

Two identically configured steps of the same class hash equal; changing any declared config value changes the hash; the hash is stable across processes. Used as part of every cache id this step’s evaluation reads or writes.

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

abstractmethod evaluate(context)#

Lazily evaluate the step within an execution context.

Parameters:: context (ExecutionContext) – Execution-scoped state and services.
Yields:: Values produced by the step.
Return type:: AsyncIterator[T]

property policy: Policy | None#: Return the policy explicitly configured for this step.

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.

name: str#

async fastrag.ingestion.steps.base.run_source_step(step, context)[source]#

Evaluate step through the same wrap _run_step applies internally.

PipelineStep.__aiter__/run() always build a fresh default context (their own policy, no cache) — fine for running a step standalone, but wrong for an aggregator (e.g. Sources) driving several child SourceSteps that should inherit its resolved cache and policy when they declare none of their own. This exposes the same evaluator with an explicit, already-scoped ExecutionContext.

Parameters:

step (SourceStep)
context (ExecutionContext)

Return type:

AsyncIterator[Source]

async fastrag.ingestion.steps.base.run_step(step, context)[source]#

Evaluate any step through the same wrap _run_step applies internally.

The generic counterpart to run_source_step(): where PipelineStep.__aiter__/run() always build a fresh default context, this lets a caller supply its own ExecutionContext for the whole call — what Pipeline needs to drive its root step (and, transitively, every step in its upstream chain) under one shared context carrying the pipeline’s own default cache, policy, and event sink.

Parameters:

step (PipelineStep[T])
context (ExecutionContext)

Return type:

AsyncIterator[T]

class fastrag.ingestion.steps.base.DocumentStep(*, cache=None, policy=None)[source]#

Bases: PipelineStep[Document], ABC

Base step transforming or generating Document instances.

Document transformations are pure and deterministic, so a cache hit needs no source-freshness revalidation — unlike SourceStep. This class provides a concrete evaluate(): it pulls from self’s upstream (bound via __call__) through the evaluator, and for each upstream item, either serves a cached result or calls transform() — concrete subclasses implement only transform, the per-item hook, never evaluate itself.

Parameters:

cache (CacheStore | Literal[False] | None)
policy (Policy | None)

abstractmethod transform(item, context)[source]#

Transform one upstream item into zero or more output documents.

Parameters:

item (Any) – One item from this step’s upstream (a Source for a parser, or a Document for a chunker/embedder).
context (ExecutionContext) – Execution-scoped state and services.

Yields:

The documents this one upstream item produces.

Return type:

AsyncIterator[Document]

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.

validate_input(item)[source]#

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

Parameters:: item (Any)
Return type:: None

async evaluate(context)[source]#

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

Transform a bounded cache window independently of worker concurrency.

Parameters:

items (AsyncIterator[Any])
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]

property config_hash: str#

A stable hash of this step’s class name and config().

Two identically configured steps of the same class hash equal; changing any declared config value changes the hash; the hash is stable across processes. Used as part of every cache id this step’s evaluation reads or writes.

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

property policy: Policy | None#: Return the policy explicitly configured for this step.

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.

name: str#

event_name: str#

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.

async fastrag.ingestion.steps.base.stream_batches(items, size)[source]#

Collect a bounded stream window, flushing the final partial batch.

Parameters:

items (AsyncIterator)
size (int)

Return type:

AsyncIterator[list[T]]

class fastrag.ingestion.steps.base.RoutedStep(*, cache=None, policy=None)[source]#

Bases: PipelineStep[T], ABC

Step that conditionally delegates to child steps based on filter rules.

A mixin, not independently instantiable: it defines route(), default(), and the per-item dispatch (transform(), satisfying the hook DocumentStep expects), but not evaluate() — a concrete routed step combines this with DocumentStep (routing first in the MRO, e.g. class ChunkerRouter(RoutedStep[Document], Chunker)) so evaluate resolves to DocumentStep’s concrete per-item driver, which calls back into this class’s transform to pick a route.

Deliberately declares no event_name, even though it sits first in a routed step’s MRO: routing is not a stage, it is a way of choosing one. Leaving it undeclared is what lets class ChunkerRouter(RoutedStep[Document], Chunker) resolve to Chunker’s "Chunking" instead of shadowing it.

Parameters:

cache (CacheStore | Literal[False] | None)
policy (Policy | None)

routing_requires_content: bool = True#: Unknown routing overrides conservatively require complete inputs. Opt out only when built-in selection uses identity/metadata alone.

routes: list[tuple[Filter[T], PipelineStep[T]]]#

route(filter_rule, step)[source]#

Register a conditional branch step matching a filter predicate.

Parameters:

filter_rule (Filter[T]) – Filter predicate to match against incoming items.
step (PipelineStep[T]) – Child step to run on match.

Returns:

Self, for fluent chaining.

Return type:

Self

default(step)[source]#

Register a fallback step for items that match no route filters.

Parameters:: step (PipelineStep[T]) – Default child step.
Returns:: Self, for fluent chaining.
Return type:: Self

property cache: CacheStore | Literal[False] | None#: Return the cache explicitly configured for this step.

config()[source]#

No identity-affecting configuration of its own.

A routed item’s cache identity is keyed by whichever child matched it (see _dispatch_route), never by the router itself.

Return type:: Mapping[str, Any]

property config_hash: str#

A stable hash of this step’s class name and config().

Two identically configured steps of the same class hash equal; changing any declared config value changes the hash; the hash is stable across processes. Used as part of every cache id this step’s evaluation reads or writes.

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

abstractmethod evaluate(context)#

Lazily evaluate the step within an execution context.

Parameters:: context (ExecutionContext) – Execution-scoped state and services.
Yields:: Values produced by the step.
Return type:: AsyncIterator[T]

property policy: Policy | None#: Return the policy explicitly configured for this step.

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.

name: str#

event_name: str#

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.

async transform(item, context)[source]#

Dispatch item to the first matching route, or the default.

An item matching no route with no default registered is skipped with an ItemFailed naming it — never silently dropped. A matched child that declares no explicit cache/policy of its own (cache is None) inherits this routed step’s effective cache and policy rather than the outer pipeline’s, since a routed child is conceptually scoped to its router. A child that does declare a policy has it applied to its own items, retries and timeout included — see _policy_for_item, which is where that half happens, since the retry wrap sits above this method rather than inside it.

Parameters:

item (T)
context (ExecutionContext)

Return type:

AsyncIterator[T]