fastrag.ingestion.steps.chunking.base#

The Chunker family base.

docs/specification/subclasses.md (a historical design record — see that file’s header): Chunker is the DocumentStep every concrete chunker (RecursiveCharacterTextSplitter, CharacterTextSplitter, DoclingChunker, ChunkerRouter, and SemanticChunker) extends. Its one job beyond DocumentStep itself is owning chunk identity — the same property the Chunker contract demands: a subclass never stamps id or metadata["chunk_index"] itself. It implements Chunker.split(), a pure generator of chunk texts, and this base class turns that into fully identified Document chunks, whatever split() does. Synchronous chunkers implement split; model-backed chunkers override asplit for asynchronous calls. Both use this base class for chunk identity, metadata and events.

Functions

structural_chunk_id(parent_id, index, ...)

Versioned structural identity; runtime diagnostics are excluded.

Classes

Chunker(*[, cache, policy])

Splits one parsed document into the units that get embedded.

fastrag.ingestion.steps.chunking.base.structural_chunk_id(parent_id, index, content, metadata, config)[source]#: Versioned structural identity; runtime diagnostics are excluded.

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

Bases: DocumentStep, ABC

Splits one parsed document into the units that get embedded.

Chunking is where chunk identity is decided, and this base class owns that decision so no implementation can get it wrong: every piece split() yields leaves transform() as a Document carrying a deterministic id (a stable function of the parent document’s own id, the chunk’s index, and its text) and metadata["chunk_index"], whatever split() does. Binding an upstream (chunker(parsed_documents), via the inherited __call__) and evaluating it (async for document in chunker / run()) never splits eagerly — only consuming the resulting stream does.

Caching (per parent document id and this chunker’s own configuration) and retries are entirely DocumentStep’s job, applied around transform(); nothing here re-implements either.

Writing one is a class and a generator:

from collections.abc import Iterator

from fastrag.models import Document
from fastrag.ingestion.steps.chunking import Chunker


class ParagraphChunker(Chunker):
    def __init__(self, *, max_chars: int = 400, cache=False, policy=None):
        super().__init__(cache=cache, policy=policy)
        self.max_chars = max_chars

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

    def split(self, document: Document) -> Iterator[str]:
        for paragraph in document.content.split("\\n\\n"):
            yield paragraph[: self.max_chars]

Every event a chunker emits is labelled "Chunking", inherited by every concrete chunker (RecursiveCharacterTextSplitter, CharacterTextSplitter, ChunkerRouter, …) unless it declares an event_name of its own.

Parameters:

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

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.

accepted_content_formats: frozenset[str] | None = None#: None preserves permissive custom subclasses; sets may include custom strings.

validate_input(item)[source]#

Reject incompatible representations before splitting or cache replay.

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

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.

abstractmethod split(document)[source]#

Split document.content, yielding one piece per chunk.

Pure computation, so a plain generator, never async. A whitespace-only piece is dropped by the base class before it is ever numbered, so an implementation need not filter those itself. Yield nothing for a document with no usable text.

Each yielded piece is either a plain str — the text of the chunk, as every chunker before this one yielded — or a (text, metadata) tuple, for a chunker whose structure carries information plain text can’t (e.g. DoclingChunker’s headings/page_no). metadata is merged into the built chunk’s metadata on top of the parent document’s own — so it may override an inherited key like title — but parent_id and chunk_index are set afterwards and always win, and content_format (when set) is applied last and always wins too: neither is ever something a chunker’s own split can shadow.

Raises:: fastrag.exceptions.ChunkingError – The document could not be split.
Parameters:: document (Document)
Return type:: Iterator[str | tuple[str, Mapping[str, Any]]]

async asplit(document)[source]#

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

async transform(item, context)[source]#

Build every chunk of item, numbering them as they are produced.

Numbering lives on this one call’s stack, not on anything shared by the chunker instance, so two documents chunked concurrently by the same chunker can never share a counter — each call to transform() starts its own counter from zero.

This method only runs on a cache miss — a cached parent document’s chunks are replayed by DocumentStep without calling transform() at all. Emitting DocumentChunked 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 split.

Parameters:

item (Document)
context (ExecutionContext)

Return type:

AsyncIterator[Document]

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

Validate resolved output, including replay; routers forward this hook.

Return type:: None

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

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.

name: str#