fastrag.ingestion.steps.chunking.docling#

Native chunking of a serialized DoclingDocument, via Docling’s own chunkers.

DoclingParser (task 011) can put the structured document itself — a serialized DoclingDocument — in Document.content rather than flattened Markdown. DoclingChunker is the chunker that knows what to do with it: it deserializes content back into a DoclingDocument and splits on the real structure (headings, tables, list items) via one of Docling’s own chunkers, instead of re-splitting text that structure was already flattened out of.

Which Docling chunker drives the split is a constructor choice (docling_core.transforms.chunker.BaseChunker, any of HierarchicalChunker, HybridChunker, LineBasedTokenChunker, or a caller’s own subclass). The default needs neither a tokenizer nor a model download; the other two do — see DoclingChunker.__init__.

Every emitted chunk is plain text again (content_format is TEXT — see the class docstring for why), embeddable and storable by every existing downstream step with no further translation, carrying headings and (when Docling reports one) page_no in its metadata alongside the parent_id/chunk_index the Chunker base already owns.

Classes

DoclingChunker(*[, chunker, cache, policy])

Split a serialized DoclingDocument with one of Docling's own chunkers.

class fastrag.ingestion.steps.chunking.docling.DoclingChunker(*, chunker=None, cache=None, policy=None)[source]#

Bases: Chunker

Split a serialized DoclingDocument with one of Docling’s own chunkers.

Handed anything whose metadata["content_format"] is not DOCLING, it refuses rather than chunking the wrong thing — route with a ContentFormatFilter (see the module example below).

Each chunk’s text is Docling’s own contextualized serialization (chunker.contextualize(chunk)) — the chunk’s heading path travels with its text, rather than being lost the moment the chunk is embedded on its own. That serialization is a delimiter-joined concatenation of the heading path and the chunk’s own (sometimes list-flavoured, never table-flavoured — a table row comes out as "Alice, Value = 42", not a Markdown table) text, not guaranteed-valid Markdown, so every chunk this class produces carries content_format = ~fastrag.models.ContentFormat.TEXT, not .MARKDOWN.

from fastrag.ingestion.filters import ContentFormatFilter
from fastrag.ingestion.steps.chunking import ChunkerRouter, DoclingChunker
from fastrag.ingestion.steps.parsing import ParserRouter, DoclingParser, office_filter
from fastrag.models import ContentFormat

parser = ParserRouter(cache=cache).route(
    office_filter, DoclingParser(output_format="docling")
)
chunker = ChunkerRouter(cache=cache).route(
    ContentFormatFilter(include_formats=[ContentFormat.DOCLING]),
    DoclingChunker(),
)

Parameters:

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

accepted_content_formats: frozenset[str] | None = frozenset({'docling'})#: None preserves permissive custom subclasses; sets may include custom strings.

content_format: ContentFormat | None = 'text'#: 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.

config()[source]#

This chunker’s cache-identity configuration.

Only the concrete Docling chunker’s class name — switching from HierarchicalChunker to a caller-supplied HybridChunker (or vice versa) invalidates the cache. A docling_core BaseChunker is a Pydantic model that (for HybridChunker/ LineBasedTokenChunker) carries a tokenizer field and a serializer_provider field, neither of which is JSON-serializable in general (confirmed: HierarchicalChunker().model_dump(mode="json") raises PydanticSerializationError on its own default serializer_provider), so this cannot fingerprint the chunker’s full configuration the way e.g. config() does. A caller who reconfigures a chunker of the same class between runs against the same cache (e.g. a different max_tokens on a HybridChunker) should pass cache=False or a fresh cache rather than rely on this to notice.

Return type:: Mapping[str, Any]

split(document)[source]#

Deserialize document.content and chunk it with the configured Docling chunker.

Each yielded piece is (chunker.contextualize(chunk), metadata), where metadata carries headings (list[str], possibly empty — Docling reports None for a chunk with no heading ancestor, normalized here to []) and, when at least one of the chunk’s source items carries Docling provenance (paginated formats — confirmed for XLSX; not reported for a plain Markdown/text source), page_no (the first one found).

Raises:: fastrag.exceptions.ChunkingError – document is not DOCLING — route with ContentFormatFilter(include_formats=[ContentFormat.DOCLING]) instead of handing this chunker anything else.
Parameters:: document (Document)
Return type:: Iterator[str | tuple[str, Mapping[str, Any]]]

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.

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]

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.

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

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]

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

async validate_output(item, produced, context)#

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

Return type:: None

name: str#