fastrag.ingestion.steps.chunking.structure#

Deterministic normalized-tree packing with exact serialized token budgets.

Functions

build_embedding_text(content, metadata)

Build the exact versioned embedding serialization independently.

Classes

StructureAwareChunker(*[, tokenizer, ...])

Pack normalized sections and structures using a hard embedding budget.

TiktokenTokenizer([encoding])

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

Tokenizer(*args, **kwargs)

Exact counting and lossless Unicode boundaries, with cache identity.

Utf8ByteTokenizer()

Explicit offline byte budget, not an embedding-model token estimate.
class fastrag.ingestion.steps.chunking.structure.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.structure.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]

class fastrag.ingestion.steps.chunking.structure.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]

fastrag.ingestion.steps.chunking.structure.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

class fastrag.ingestion.steps.chunking.structure.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.

options: dict[str, Any]#
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#