fastrag.ingestion.steps.chunking#

Chunking: parsed documents in, embeddable chunks out.

DoclingChunker is an opt-in chunker for a DOCLING-format document (task 011’s DoclingParser(output_format="docling")): it chunks the real DoclingDocument structure with one of Docling’s own chunkers, instead of re-splitting flattened text — never a ChunkerRouter default, the same reasoning DoclingParser documents for itself.

class fastrag.ingestion.steps.chunking.ChatBoundaryRefiner(*, llm, identity, allow_external_transmission=False, timeout=30.0, retries=0, backoff=0.1, concurrency=2, max_input_chars=12000, max_response_chars=16000, cache_size=0)[source]#

Bases: object

Reuse an injected ChatModel with explicit permission and nonsecret identity.

identity must distinguish endpoint/model/generation settings without containing secrets. The strategy never examines provider credentials. Successful proposals optionally cache in an instance-local bounded LRU. Configure the provider’s own output-token limit to bound response generation.

Parameters:
config()[source]#

Nonsecret complete strategy identity for pipeline/response caches.

Return type:

dict[str, Any]

property cache_entries: int#

Current bounded response-cache occupancy, without source exposure.

async refine(*, node_id, text, allowed)[source]#

Select validated offsets; all ordinary failures become local fallback.

Parameters:
Return type:

RefinementResult

class fastrag.ingestion.steps.chunking.RefinementResult(boundaries=(), calls=0, cache_hit=False, fallback=None)[source]#

Bases: object

Validated proposal and runtime-only observations for a single request.

Parameters:
class fastrag.ingestion.steps.chunking.ChunkValidation(chunk_count, retained_characters, embedding_tokens, classification, validation_seconds=0.0, token_counts=(), merged_chunks=0, split_chunks=0, fallback_chunks=0, oversized_chunks=0, excluded_characters=0, normalization_diagnostics=(), cached_chunks=0, small_chunks=0, model_calls=0, model_cache_hits=0, model_fallbacks=(), operation_seconds=0.0)[source]#

Bases: object

Runtime diagnostics kept outside chunk metadata and identity.

Parameters:

  • chunk_count (int)

  • retained_characters (int)

  • embedding_tokens (int)

  • classification (str)

  • validation_seconds (float)

  • token_counts (tuple[int, ...])

  • merged_chunks (int)

  • split_chunks (int)

  • fallback_chunks (int)

  • oversized_chunks (int)

  • excluded_characters (int)

  • normalization_diagnostics (tuple[str, ...])

  • cached_chunks (int)

  • small_chunks (int)

  • model_calls (int)

  • model_cache_hits (int)

  • model_fallbacks (tuple[str, ...])

  • operation_seconds (float)

fastrag.ingestion.steps.chunking.validate_structure_chunks(normalized, pieces, tokenizer, max_tokens, max_overlap_tokens=50, min_preferred_tokens=100)[source]#

Validate split tuples or cached Documents against authoritative source.

Raises ChunkingError for invalid persisted metadata. Counts are recomputed; neither recorded budgets nor node IDs substitute for interval coverage.

Parameters:
Return type:

ChunkValidation

class fastrag.ingestion.steps.chunking.CharacterTextSplitter(*, chunk_size=1000, chunk_overlap=200, separator='\n\n', cache=None, policy=None)[source]#

Bases: Chunker

Split on a single literal separator, with no separator hierarchy.

Unlike RecursiveCharacterTextSplitter, this never descends to a smaller separator when a piece is still too large after splitting on separator — the right choice when a source’s structure already guarantees short-enough pieces at one boundary (e.g. a page’s own section breaks), and the wrong one for prose of unpredictable length.

from fastrag.ingestion.steps.chunking import CharacterTextSplitter

chunker = CharacterTextSplitter(chunk_size=500, chunk_overlap=50, cache=cache)
Parameters:
accepted_content_formats: frozenset[str] | None = frozenset({'html', 'markdown', 'rst', 'text'})#

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

config()[source]#

This splitter’s configuration, for cache-identity hashing.

Return type:

Mapping[str, Any]

split(document)[source]#

Split document.content on self.separator only.

Parameters:

document (Document)

Return type:

Iterator[str]

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.

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.

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.

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:
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#
class fastrag.ingestion.steps.chunking.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:
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:
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:
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#
class fastrag.ingestion.steps.chunking.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#
class fastrag.ingestion.steps.chunking.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:
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:
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#
class fastrag.ingestion.steps.chunking.HeadingChunker(*, cache=None, policy=None)[source]#

Bases: Chunker

Split Markdown at headings without splitting section bodies.

Each chunk includes its ancestor headings in text and a headings metadata list ordered from outermost to innermost. ATX and Setext headings are recognized; headings inside fenced or indented code are ignored. Consecutive ancestor headings travel with their first subsection instead of becoming isolated title chunks. Text before the first heading is kept with an empty heading path. Sections have no size cap or overlap, so code, lists and tables stay intact even when a section is long.

Inherits Chunker’s cache and policy constructor arguments.

Parameters:
accepted_content_formats: frozenset[str] | None = frozenset({'markdown', 'text'})#

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

config()[source]#

Return the section splitting configuration for cache identity.

Return type:

Mapping[str, Any]

split(document)[source]#

Yield whole sections with their heading paths and original Markdown.

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.

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.

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.

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:
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#
class fastrag.ingestion.steps.chunking.LangchainSemanticChunker(*, embeddings, buffer_size=1, breakpoint_threshold_type='percentile', breakpoint_threshold_amount=None, number_of_chunks=None, sentence_split_regex='(?<=[.?!])\\s+', min_chunk_size=None, add_start_index=False, batch_size=64, cache=None, policy=None)[source]#

Bases: Chunker

Split with LangChain Experimental’s SemanticChunker, embedding with an SDK Embedder.

A thin wrapper: sentence grouping, distance thresholds and chunk assembly are all LangChain’s, so output matches SemanticChunker exactly (it trims and rejoins sentences rather than preserving source whitespace — use EmbeddingSemanticChunker when exact source slices matter). Only the embedding model is ours.

from fastrag.ingestion.steps.chunking import LangchainSemanticChunker
from fastrag.ingestion.steps.embedding import OpenAIEmbedder

chunker = LangchainSemanticChunker(
    embeddings=OpenAIEmbedder(model="text-embedding-3-small", api_key=key)
)
Parameters:

  • embeddings (Embedder)

  • buffer_size (int)

  • breakpoint_threshold_type (Literal['percentile', 'standard_deviation', 'interquartile', 'gradient'])

  • breakpoint_threshold_amount (float | None)

  • number_of_chunks (int | None)

  • sentence_split_regex (str)

  • min_chunk_size (int | None)

  • add_start_index (bool)

  • batch_size (int)

  • cache (CacheStore | Literal[False] | None)

  • policy (Policy | None)

accepted_content_formats: frozenset[str] | None = frozenset({'markdown', 'text'})#

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

config()[source]#

Return splitter settings and the embedder’s nonsecret identity.

Return type:

Mapping[str, Any]

split(document)[source]#

Require asynchronous model evaluation before yielding any pieces.

Parameters:

document (Document)

Return type:

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

async asplit(document)[source]#

Run LangChain’s splitter in a thread, embedding on this event loop.

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.

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.

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.

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:
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#
class fastrag.ingestion.steps.chunking.EmbeddingSemanticChunker(*, embeddings, buffer_size=1, breakpoint_threshold_type='percentile', breakpoint_threshold_amount=None, number_of_chunks=None, sentence_split_regex='(?<=[.?!])\\s+', min_chunk_size=None, add_start_index=False, batch_size=64, cache=None, policy=None)[source]#

Bases: Chunker

Select sentence cuts with an existing asynchronous SDK Embedder.

Threshold modes mirror LangChain Experimental statistics. Requested counts instead select the strongest cuts exactly, with earlier offsets winning ties. Minimum size is in source characters and may reduce that count. This is a sentence splitter, with no syntax awareness or hard token budget. Both direct and pipeline outputs preserve all original whitespace and source order.

Parameters:

  • embeddings (Embedder)

  • buffer_size (int)

  • breakpoint_threshold_type (Literal['percentile', 'standard_deviation', 'interquartile', 'gradient'])

  • breakpoint_threshold_amount (float | None)

  • number_of_chunks (int | None)

  • sentence_split_regex (str)

  • min_chunk_size (int | None)

  • add_start_index (bool)

  • batch_size (int)

  • cache (CacheStore | Literal[False] | None)

  • policy (Policy | None)

accepted_content_formats: frozenset[str] | None = frozenset({'markdown', 'text'})#

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

config()[source]#

Return strategy settings and opaque nonsecret provider identity.

Return type:

Mapping[str, Any]

split(document)[source]#

Require asynchronous model evaluation before yielding any pieces.

Parameters:

document (Document)

Return type:

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

async asplit(document)[source]#

Embed neighboring source sentences, then slice at validated offsets.

Parameters:

document (Document)

Return type:

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

async transform(item, context)[source]#

Use base-owned metadata/identity while retaining exact source whitespace.

Parameters:
Return type:

AsyncIterator[Document]

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

Restore source order after stores replay chunks in arbitrary ID order.

Return type:

None

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.

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.

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.

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.

validate_input(item)#

Reject incompatible representations before splitting or cache replay.

Parameters:

item (Document)

Return type:

None

name: str#
class fastrag.ingestion.steps.chunking.RecursiveCharacterTextSplitter(*, chunk_size=1000, chunk_overlap=200, separators=None, cache=None, policy=None)[source]#

Bases: Chunker

Split on paragraphs, then lines, then words, then characters.

The default chunker: it descends the separator hierarchy only when a piece is still too large, so a chunk breaks at a blank line whenever it can and mid-word only when it must. That makes it the right choice for prose regardless of source format, including the Markdown HtmlParser produces.

from fastrag.ingestion.steps.chunking import RecursiveCharacterTextSplitter

chunker = RecursiveCharacterTextSplitter(
    chunk_size=500, chunk_overlap=50, cache=cache
)
Parameters:

  • chunk_size (int)

  • chunk_overlap (int)

  • separators (Sequence[str] | None)

  • cache (CacheStore | Literal[False] | None)

  • policy (Policy | None)

accepted_content_formats: frozenset[str] | None = frozenset({'html', 'markdown', 'rst', 'text'})#

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

config()[source]#

This splitter’s configuration, for cache-identity hashing.

Return type:

Mapping[str, Any]

split(document)[source]#

Split document.content on the paragraph/line/word/character hierarchy.

Parameters:

document (Document)

Return type:

Iterator[str]

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.

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.

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.

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:
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#
class fastrag.ingestion.steps.chunking.SemanticChunker(*, llm, chunk_size=1000, max_input_chars=12000, fallback_on_transient_error=False, cache=None, policy=None)[source]#

Bases: Chunker

Ask a chat model to group text blocks by topic, preserving source text.

chunk_size is a target in characters, not a hard output limit. Long paragraphs are divided into blocks of that size before grouping. max_input_chars bounds the source text sent in each model call; documents exceeding it are processed in consecutive windows. Use asplit or bind this chunker into a pipeline for asynchronous calls. With fallback_on_transient_error, temporary model failures switch this instance to local splitting at the original block boundaries. Invalid model responses and permanent failures still raise.

Parameters:
accepted_content_formats: frozenset[str] | None = frozenset({'markdown', 'text'})#

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

config()[source]#

Describe semantic grouping without credentials.

Return type:

Mapping[str, Any]

split(document)[source]#

Require async evaluation for model-backed splitting.

Parameters:

document (Document)

Return type:

Iterator[str]

async asplit(document)[source]#

Select boundaries, validate them, and yield original contiguous text.

Parameters:

document (Document)

Return type:

AsyncIterator[str]

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.

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.

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.

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:
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#
class fastrag.ingestion.steps.chunking.StructureAwareChunker(*, tokenizer=None, normalizers=None, target_tokens=400, max_tokens=600, min_preferred_tokens=100, overlap_tokens=0, max_overlap_tokens=50, merge_small_sections=True, allow_cross_top_level_merge=False, preserve_heading_hierarchy=True, preserve_code_blocks=True, preserve_table_rows=True, preserve_list_hierarchy=True, include_heading_path_in_embedding=True, semantic_mode='heuristic', semantic_refiner=None, oversized_block_policy='split_with_context', cache=None, policy=None, diagnostics=None)[source]#

Bases: Chunker

Pack normalized sections and structures using a hard embedding budget.

Defaults never invoke a model. Exact budgets guarantee only the injected tokenizer; select the embedding model’s tokenizer explicitly when needed. See the living specification for serialization and overflow contracts.

Parameters:

  • tokenizer (Tokenizer | None)

  • normalizers (NormalizerRegistry | None)

  • target_tokens (int)

  • max_tokens (int)

  • min_preferred_tokens (int)

  • overlap_tokens (int)

  • max_overlap_tokens (int)

  • merge_small_sections (bool)

  • allow_cross_top_level_merge (bool)

  • preserve_heading_hierarchy (bool)

  • preserve_code_blocks (bool)

  • preserve_table_rows (bool)

  • preserve_list_hierarchy (bool)

  • include_heading_path_in_embedding (bool)

  • semantic_mode (str)

  • semantic_refiner (ChatBoundaryRefiner | None)

  • oversized_block_policy (str)

content_format: ContentFormat | None = 'markdown'#

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

Identify all output choices for existing pipeline/cache machinery.

Return type:

Mapping[str, Any]

property accepted_content_formats: frozenset[str]#

Capabilities follow the supplied instance-local normalizer registry.

split(document)[source]#

Normalize once, then use the independently callable tree engine.

Parameters:

document (Document)

Return type:

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

async asplit(document)[source]#

Normalize, optionally refine safe prose, then deterministically pack.

Parameters:

document (Document)

Return type:

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

async asplit_normalized(normalized)[source]#

Async tree API; per-operation state isolates concurrent documents.

Parameters:

normalized (NormalizedDocument)

Return type:

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

split_normalized(normalized, *, _refined=False)[source]#

Consume any valid registered normalized tree without format routing.

Parameters:
Return type:

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

replay_needs_content: bool = True#

validate_output hydrates every replayed chunk.

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

Independently validate fresh/cached output, hydrating projections.

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

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

name: str#
class fastrag.ingestion.steps.chunking.Tokenizer(*args, **kwargs)[source]#

Bases: Protocol

Exact counting and lossless Unicode boundaries, with cache identity.

identity: str#
count(text)[source]#

Count complete serialized text; counts need not be additive.

Parameters:

text (str)

Return type:

int

boundaries(text)[source]#

Return ordered safe offsets including zero and the final offset.

Parameters:

text (str)

Return type:

Sequence[int]

class fastrag.ingestion.steps.chunking.TiktokenTokenizer(encoding='cl100k_base')[source]#

Bases: object

Exact ordinary-text BPE count for an explicitly named encoding.

Parameters:

encoding (str)

count(text)[source]#

Treat literal special-token spellings as ordinary source text.

Parameters:

text (str)

Return type:

int

boundaries(text)[source]#

Use original Unicode offsets rather than decoding partial BPE bytes.

Parameters:

text (str)

Return type:

Sequence[int]

class fastrag.ingestion.steps.chunking.Utf8ByteTokenizer[source]#

Bases: object

Explicit offline byte budget, not an embedding-model token estimate.

identity = 'utf8-bytes:v1'#
count(text)[source]#

Count bytes, conservatively including multibyte Unicode.

Parameters:

text (str)

Return type:

int

boundaries(text)[source]#

Never split a Unicode code point.

Parameters:

text (str)

Return type:

Sequence[int]

fastrag.ingestion.steps.chunking.build_embedding_text(content, metadata)[source]#

Build the exact versioned embedding serialization independently.

The engine records which heading IDs are already displayed. Checking IDs rather than title strings handles duplicate titles and headings in code. Generic documents without this version marker retain content-only input.

Parameters:
Return type:

str

Modules

base

The Chunker family base.

character

CharacterTextSplitter: split on one separator only, no fallback hierarchy.

docling

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

embedding_semantic

Embedding-distance sentence boundaries over exact source slices.

heading

Markdown sections kept whole, with their heading ancestry for retrieval.

langchain_semantic

LangchainSemanticChunker: LangChain Experimental's SemanticChunker over an SDK Embedder.

recursive

RecursiveCharacterTextSplitter: split on a separator hierarchy, largest first.

refinement

Opt-in, bounded chat boundary selection; source text is never rewritten.

router

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

semantic

LLM-selected semantic boundaries with verbatim source text.

structure

Deterministic normalized-tree packing with exact serialized token budgets.

validation

Independent validation of persisted structural output, without a database.
On this page
Edit on GitHub
Show Source