ontocast.tool.chunk.prepare

Prepare content units: segment, tag, filter, and size within section boundaries.

Attributes

NormalizedChunk = PreparedChunk module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

PrepareOptions dataclass

Options for the chunk preparation pipeline.

Source code in ontocast/tool/chunk/prepare.py

@dataclass
class PrepareOptions:
    """Options for the chunk preparation pipeline."""

    section_schema_id: str | None = None
    document_type_hint: str | None = None
    target_sections: list[str] | None = None
    summarize_sections: list[str] | None = None
    exclude_sections: list[str] | None = None

    def needs_section_prepare(self) -> bool:
        """True when a request option explicitly requires section labels.

        Section tagging itself is default-on (see ``CHUNK_SECTION_CLASSIFIER``);
        this only reports whether the request carries section-dependent options.
        """
        return (
            self.target_sections is not None
            or self.summarize_sections is not None
            or self.exclude_sections is not None
        )

    def filter_allowlist(self) -> list[str] | None:
        if self.target_sections is not None:
            return self.target_sections
        if (
            self.summarize_sections is not None
            and self.summarize_sections
            and "*" not in self.summarize_sections
        ):
            return self.summarize_sections
        return None

    def filter_allowlist_param(self) -> str:
        """Name of the request option that produced :meth:`filter_allowlist`."""
        return (
            "target_sections"
            if self.target_sections is not None
            else "summarize_sections"
        )

    def filter_denylist(self, schema: SectionLabelSchema) -> list[str]:
        """Effective exclusion denylist.

        ``None`` means "use the resolved schema's default_exclude"; an explicit
        ``[]`` opts out of exclusion entirely; a non-empty list is used as-is.
        """
        if self.exclude_sections is not None:
            return list(self.exclude_sections)
        return list(schema.default_exclude)

Attributes

document_type_hint = None class-attribute instance-attribute

exclude_sections = None class-attribute instance-attribute

section_schema_id = None class-attribute instance-attribute

summarize_sections = None class-attribute instance-attribute

target_sections = None class-attribute instance-attribute

Methods:

__init__(section_schema_id=None, document_type_hint=None, target_sections=None, summarize_sections=None, exclude_sections=None)

filter_allowlist()

Source code in ontocast/tool/chunk/prepare.py

def filter_allowlist(self) -> list[str] | None:
    if self.target_sections is not None:
        return self.target_sections
    if (
        self.summarize_sections is not None
        and self.summarize_sections
        and "*" not in self.summarize_sections
    ):
        return self.summarize_sections
    return None

filter_allowlist_param()

Name of the request option that produced :meth:filter_allowlist.

Source code in ontocast/tool/chunk/prepare.py

def filter_allowlist_param(self) -> str:
    """Name of the request option that produced :meth:`filter_allowlist`."""
    return (
        "target_sections"
        if self.target_sections is not None
        else "summarize_sections"
    )

filter_denylist(schema)

Effective exclusion denylist.

None means "use the resolved schema's default_exclude"; an explicit [] opts out of exclusion entirely; a non-empty list is used as-is.

Source code in ontocast/tool/chunk/prepare.py

def filter_denylist(self, schema: SectionLabelSchema) -> list[str]:
    """Effective exclusion denylist.

    ``None`` means "use the resolved schema's default_exclude"; an explicit
    ``[]`` opts out of exclusion entirely; a non-empty list is used as-is.
    """
    if self.exclude_sections is not None:
        return list(self.exclude_sections)
    return list(schema.default_exclude)

needs_section_prepare()

True when a request option explicitly requires section labels.

Section tagging itself is default-on (see CHUNK_SECTION_CLASSIFIER); this only reports whether the request carries section-dependent options.

Source code in ontocast/tool/chunk/prepare.py

def needs_section_prepare(self) -> bool:
    """True when a request option explicitly requires section labels.

    Section tagging itself is default-on (see ``CHUNK_SECTION_CLASSIFIER``);
    this only reports whether the request carries section-dependent options.
    """
    return (
        self.target_sections is not None
        or self.summarize_sections is not None
        or self.exclude_sections is not None
    )

PreparedChunk dataclass

A prepared text chunk with optional structural metadata and section label.

section_label_source and section_label_confidence record which tier of the classification cascade decided the label, so a run can be audited and weak labels can be told from strong ones.

Source code in ontocast/tool/chunk/prepare.py

@dataclass(frozen=True)
class PreparedChunk:
    """A prepared text chunk with optional structural metadata and section label.

    ``section_label_source`` and ``section_label_confidence`` record which tier
    of the classification cascade decided the label, so a run can be audited
    and weak labels can be told from strong ones.
    """

    text: str
    headings: list[str] | None
    doc_item_refs: tuple[str, ...] = ()
    section_label: str | None = None
    section_label_source: SectionLabelSource | None = None
    section_label_confidence: float = 0.0

Attributes

doc_item_refs = () class-attribute instance-attribute

headings instance-attribute

section_label = None class-attribute instance-attribute

section_label_confidence = 0.0 class-attribute instance-attribute

section_label_source = None class-attribute instance-attribute

text instance-attribute

Methods:

__init__(text, headings, doc_item_refs=(), section_label=None, section_label_source=None, section_label_confidence=0.0)

SchemaDecision dataclass

Which label schema a document is prepared against, and why.

Recorded rather than recomputed: schema selection now depends on document text, and it used to be resolved independently in three places. If those disagreed, the deterministic tiers would tag against one schema while the LLM backfill validated against another, and normalise_llm_label drops labels absent from its schema -- silent label loss, not an error.

Source code in ontocast/tool/chunk/prepare.py

@dataclass(frozen=True)
class SchemaDecision:
    """Which label schema a document is prepared against, and why.

    Recorded rather than recomputed: schema selection now depends on document
    text, and it used to be resolved independently in three places. If those
    disagreed, the deterministic tiers would tag against one schema while the
    LLM backfill validated against another, and ``normalise_llm_label`` drops
    labels absent from its schema -- silent label loss, not an error.
    """

    schema: SectionLabelSchema
    source: str
    detection: SchemaDetection | None = None

    @property
    def schema_id(self) -> str:
        return self.schema.id

Attributes

detection = None class-attribute instance-attribute

schema instance-attribute

schema_id property

source instance-attribute

Methods:

__init__(schema, source, detection=None)

SectionSelectionEmptyError

Bases: ValueError

A section selection matched no segment in this document.

Distinct from a malformed parameter: target_sections=["reslts"] is syntactically fine and only turns out to be wrong once this document has been classified. Under the default CHUNK_SECTION_FILTER_ON_EMPTY=warn this is a log line and the run continues to an empty graph, which reads exactly like a document that genuinely had nothing to extract -- telling those two apart is the whole point of the error mode.

Subclasses :class:ValueError so existing parameter guards keep their shape. Deliberately not an api-layer error: nothing under tool/ imports ontocast.api, and the parameter here is well-formed.

Source code in ontocast/tool/chunk/prepare.py

class SectionSelectionEmptyError(ValueError):
    """A section selection matched no segment in this document.

    Distinct from a malformed parameter: ``target_sections=["reslts"]`` is
    syntactically fine and only turns out to be wrong once *this* document has
    been classified. Under the default ``CHUNK_SECTION_FILTER_ON_EMPTY=warn``
    this is a log line and the run continues to an empty graph, which reads
    exactly like a document that genuinely had nothing to extract -- telling
    those two apart is the whole point of the ``error`` mode.

    Subclasses :class:`ValueError` so existing parameter guards keep their
    shape. Deliberately *not* an ``api``-layer error: nothing under ``tool/``
    imports ``ontocast.api``, and the parameter here is well-formed.
    """

    def __init__(self, param: str, message: str) -> None:
        super().__init__(message)
        self.param = param

Attributes

param = param instance-attribute

Methods:

__init__(param, message)

Source code in ontocast/tool/chunk/prepare.py

def __init__(self, param: str, message: str) -> None:
    super().__init__(message)
    self.param = param

Functions:

prepare_content_units(docling_doc, splitter, config, options, tools=None) async

Segment, tag, filter, and size document text into prepared chunks.

Section tagging is default-on: the sections-first flow runs unless CHUNK_SECTION_CLASSIFIER=off (which also disables section filters and schema default exclusions; explicit section options are ignored with a warning in that case).

Parameters:

Name	Type	Description	Default
docling_doc	DoclingDocument	Converted source document.	required
splitter	ChunkerTool	Chunker used to size oversized sections.	required
config	ChunkConfig	Chunk configuration, including the classifier tier.	required
options	PrepareOptions	Per-request section schema and filters.	required
tools	'ToolBox | None'	ToolBox providing the LLM. Required only when config.section_classifier == "llm"; the deterministic tiers need no LLM, so callers that only inspect sections may omit it.	None

Raises:

Type	Description
ValueError	section_classifier is "llm" but no ToolBox was supplied.
SectionSelectionEmptyError	A section allowlist or denylist removed every segment and CHUNK_SECTION_FILTER_ON_EMPTY=error.

Source code in ontocast/tool/chunk/prepare.py

async def prepare_content_units(
    docling_doc: DoclingDocument,
    splitter: ChunkerTool,
    config: ChunkConfig,
    options: PrepareOptions,
    tools: "ToolBox | None" = None,
) -> list[PreparedChunk]:
    """Segment, tag, filter, and size document text into prepared chunks.

    Section tagging is default-on: the sections-first flow runs unless
    ``CHUNK_SECTION_CLASSIFIER=off`` (which also disables section filters and
    schema default exclusions; explicit section options are ignored with a
    warning in that case).

    Args:
        docling_doc: Converted source document.
        splitter: Chunker used to size oversized sections.
        config: Chunk configuration, including the classifier tier.
        options: Per-request section schema and filters.
        tools: ToolBox providing the LLM. Required only when
            ``config.section_classifier == "llm"``; the deterministic tiers
            need no LLM, so callers that only inspect sections may omit it.

    Raises:
        ValueError: ``section_classifier`` is ``"llm"`` but no ToolBox was
            supplied.
        SectionSelectionEmptyError: A section allowlist or denylist removed
            every segment and ``CHUNK_SECTION_FILTER_ON_EMPTY=error``.
    """
    if config.section_classifier == "llm" and tools is None:
        raise ValueError(
            "CHUNK_SECTION_CLASSIFIER=llm requires a ToolBox providing an LLM"
        )
    # Segmentation is CPU-bound and runs local embedding models, but this
    # coroutine is awaited on the event loop -- inline, it would freeze every
    # concurrent document's in-flight provider sockets for its whole duration.
    document_text = await asyncio.to_thread(
        document_text_for_section_tagging, docling_doc
    )

    if config.section_classifier == "off":
        if options.needs_section_prepare():
            logger.warning(
                "Section options requested but CHUNK_SECTION_CLASSIFIER=off; "
                "section filters are ignored"
            )
        return await asyncio.to_thread(
            _simple_prepare, docling_doc, document_text, splitter, config
        )

    decision = await asyncio.to_thread(
        resolve_prepare_schema, document_text, config, options, splitter
    )
    schema = decision.schema
    spans = await asyncio.to_thread(
        detect_section_spans,
        document_text,
        schema,
        include_text_headings=config.section_text_headings,
    )

    segments = await asyncio.to_thread(
        _primary_segments, docling_doc, document_text, spans, splitter, config
    )
    if not segments:
        return []

    segments = await asyncio.to_thread(
        coalesce_small_segments_right,
        segments,
        config.section_tag_min_chars,
        schema,
    )
    await asyncio.to_thread(_tag_segments, segments, document_text, spans, schema)
    if config.section_classifier in ("heuristic", "llm"):
        await asyncio.to_thread(_density_label_segments, segments, schema, config)
    if config.section_classifier == "llm" and tools is not None:
        await llm_backfill_section_labels(
            segments,
            tools,
            # The resolved schema is threaded, not re-derived: re-resolving from
            # the raw request would ignore a text-based detection and validate
            # LLM labels against a different schema, silently dropping them.
            schema=schema,
            section_schema_id=options.section_schema_id,
            document_type_hint=options.document_type_hint,
            section_tag_min_chars=config.section_tag_min_chars,
            batch_size=config.section_llm_batch_size,
        )
    _forward_fill_section_labels(segments, schema)
    segments = _attach_bare_headings(segments)

    unlabeled = sum(1 for s in segments if s.section_label is None)
    if unlabeled:
        # Normal for text with no headings (plain-text and JSON inputs); the
        # manifest's section_label_histogram is the record.
        logger.info(
            "%s segment(s) remain without section_label after classification",
            unlabeled,
        )

    on_empty = config.section_filter_on_empty

    allowlist = options.filter_allowlist()
    if allowlist is not None:
        before = len(segments)
        segments = _filter_segments(segments, allowlist)
        logger.info(
            "Section filter %s: kept %s/%s segments before sizing",
            allowlist,
            len(segments),
            before,
        )
        _guard_empty_selection(
            param=options.filter_allowlist_param(),
            selection=allowlist,
            before=before,
            after=len(segments),
            on_empty=on_empty,
        )

    denylist = options.filter_denylist(schema)
    if denylist:
        before = len(segments)
        segments = _filter_segments_excluding(segments, denylist)
        # The denylist can come entirely from the resolved schema's
        # default_exclude with no caller involvement, so name the source: this
        # path had no empty guard at all and could blank the document silently.
        _guard_empty_selection(
            param=(
                "exclude_sections"
                if options.exclude_sections is not None
                else f"exclude_sections (schema '{schema.id}' default)"
            ),
            selection=denylist,
            before=before,
            after=len(segments),
            on_empty=on_empty,
        )

    return await asyncio.to_thread(_size_segments, segments, splitter, config)

resolve_prepare_schema(document_text, config, options, splitter=None)

Choose the section-label schema for one document.

Precedence: an explicit section_schema_id, then a document_type_hint that maps to a schema, then automatic detection, then the manifest default. Caller-supplied intent is never overridden -- detection only fills the gap where the request said nothing.

Parameters:

Name	Type	Description	Default
document_text	str	Markdown export used for heading detection.	required
config	ChunkConfig	Chunk configuration, including the detection tier.	required
options	PrepareOptions	Per-request schema id and document type hint.	required
splitter	ChunkerTool | None	Chunker, used only to reach the embedding model already loaded for semantic chunking. None restricts detection to the lexical tier.	None

Returns:

Type	Description
SchemaDecision	The chosen schema and how it was chosen.

Source code in ontocast/tool/chunk/prepare.py

def resolve_prepare_schema(
    document_text: str,
    config: ChunkConfig,
    options: PrepareOptions,
    splitter: ChunkerTool | None = None,
) -> SchemaDecision:
    """Choose the section-label schema for one document.

    Precedence: an explicit ``section_schema_id``, then a ``document_type_hint``
    that maps to a schema, then automatic detection, then the manifest default.
    Caller-supplied intent is never overridden -- detection only fills the gap
    where the request said nothing.

    Args:
        document_text: Markdown export used for heading detection.
        config: Chunk configuration, including the detection tier.
        options: Per-request schema id and document type hint.
        splitter: Chunker, used only to reach the embedding model already loaded
            for semantic chunking. ``None`` restricts detection to the lexical
            tier.

    Returns:
        The chosen schema and how it was chosen.
    """
    if options.section_schema_id and options.section_schema_id.strip():
        schema_id = resolve_section_schema_id(
            section_schema_id=options.section_schema_id
        )
        return SchemaDecision(load_section_label_schema(schema_id), "explicit")

    from_hint = schema_id_from_hint(options.document_type_hint)
    if from_hint is not None:
        return SchemaDecision(load_section_label_schema(from_hint), "hint")

    detection = _detect_schema(document_text, config, splitter)
    if detection is not None:
        logger.info(
            "Detected document schema %r via %s tier (score %.1f, margin %.1fx): %s",
            detection.schema_id,
            detection.tier,
            detection.score,
            detection.margin,
            ", ".join(detection.evidence[0].examples[:3]),
        )
        return SchemaDecision(
            load_section_label_schema(detection.schema_id), "detected", detection
        )

    default_id = resolve_section_schema_id()
    logger.info("No schema evidence; falling back to default schema %r", default_id)
    return SchemaDecision(load_section_label_schema(default_id), "default")