ontocast.tool.chunk.prepare

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

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

    def needs_section_prepare(self) -> bool:
        return self.target_sections is not None or self.summarize_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

PreparedChunk dataclass

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

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

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

    text: str
    headings: list[str] | None
    doc_item_refs: tuple[str, ...] = ()
    section_label: str | None = None

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

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

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

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

    if not options.needs_section_prepare():
        return _simple_prepare(
            docling_doc, document_text, splitter, config, hybrid_chunker
        )

    schema_id = resolve_section_schema_id(
        section_schema_id=options.section_schema_id,
        document_type_hint=options.document_type_hint,
    )
    schema = load_section_label_schema(schema_id)
    spans = detect_section_spans(document_text, schema)

    segments = _hybrid_segments(docling_doc, hybrid_chunker)
    if not segments:
        segments = _semantic_full_doc_segments(document_text, splitter)

    if not segments:
        return []

    segments = coalesce_small_segments_right(
        segments,
        config.section_tag_min_chars,
        schema,
    )
    _tag_segments(segments, document_text, spans, schema)
    await llm_backfill_section_labels(
        segments,
        tools,
        section_schema_id=options.section_schema_id,
        document_type_hint=options.document_type_hint,
        section_tag_min_chars=config.section_tag_min_chars,
    )
    _forward_fill_section_labels(segments, schema)

    unlabeled = sum(1 for s in segments if s.section_label is None)
    if unlabeled:
        logger.warning(
            "%s segment(s) remain without section_label after LLM backfill",
            unlabeled,
        )

    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,
        )
        if before > 0 and not segments:
            logger.warning(
                "Section filter %s removed all segments; check headings or allowlist",
                allowlist,
            )

    return _size_segments(segments, splitter, config)