fastrag.ingestion.steps.chunking.router#

ChunkerRouter: dispatch each document to the chunker registered for it.

Classes

ChunkerRouter(*[, cache, policy])

Route each document to the first matching .route chunker, or .default.
class fastrag.ingestion.steps.chunking.router.ChunkerRouter(*, cache=None, policy=None)[source]#

Bases: RoutedStep[Document], Chunker

Route each document to the first matching .route chunker, or .default.

Holds no chunking logic of its own: a document matching no route with no default registered is skipped with an ItemFailed.

from fastrag.ingestion.filters import ContentFormatFilter
from fastrag.ingestion.steps.chunking import (
    ChunkerRouter,
    DoclingChunker,
    RecursiveCharacterTextSplitter,
)

chunker = (
    ChunkerRouter(cache=cache)
    .route(ContentFormatFilter(include_formats=["docling"]), DoclingChunker())
    .default(RecursiveCharacterTextSplitter(chunk_size=500))
)
Parameters:
routing_requires_content: bool = False#

Unknown routing overrides conservatively require complete inputs. Opt out only when built-in selection uses identity/metadata alone.

async validate_output(item, produced, context)[source]#

Forward resolved output validation to the selected child chunker.

Return type:

None

split(document)[source]#

Never called: RoutedStep.transform shadows Chunker.transform.

Defined only because Chunker declares split abstract.

Raises:

NotImplementedError – Always.

Parameters:

document (Document)

Return type:

Iterator[str]

accepted_content_formats: frozenset[str] | None = None#

None preserves permissive custom subclasses; sets may include custom strings.

async asplit(document)#

Yield split pieces asynchronously; override for model-backed chunking.

The default delegates to the synchronous split hook. Both hooks share chunk identity, metadata, caching and events through transform.

Parameters:

document (Document)

Return type:

AsyncIterator[str | tuple[str, Mapping[str, Any]]]

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

Return the cache explicitly configured for this step.

config()#

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.

content_format: ContentFormat | None = None#

Overrides the parent document’s content_format in every chunk this chunker produces. None — the default — changes nothing: metadata inheritance in _build_chunk already carries the parent’s content_format (and everything else in its metadata) forward unchanged. Set this on a chunker whose output format differs from its input’s (e.g. one that flattens a DOCLING document into Markdown chunks) so the chunks it produces describe themselves truthfully.

default(step)#

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

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: str = 'Chunking'#

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

route(filter_rule, step)#

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

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

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:
Return type:

AsyncIterator[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)#

Reject incompatible representations before splitting or cache replay.

Parameters:

item (Document)

Return type:

None

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