ontocast.tool.chunk

Document chunking tools for OntoCast.

ChunkerTool

Bases: Tool

Tool for semantic chunking of documents.

Falls back to naive chunking if sentence-transformers is not available. Includes caching to avoid re-chunking the same text with the same parameters.

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

class ChunkerTool(Tool):
    """Tool for semantic chunking of documents.

    Falls back to naive chunking if sentence-transformers is not available.
    Includes caching to avoid re-chunking the same text with the same parameters.
    """

    config: ChunkConfig = Field(
        default_factory=ChunkConfig, description="Chunking configuration parameters"
    )
    chunking_mode: Literal["semantic", "naive"] = Field(
        default="semantic",
        description="Chunking mode: semantic (requires sentence-transformers) or naive (fallback)",
    )
    cache: Any = Field(default=None, exclude=True)

    def __init__(
        self,
        chunk_config: ChunkConfig | None = None,
        cache: Cacher | None = None,
        **kwargs,
    ):
        """Initialize the ChunkerTool.

        Args:
            chunk_config: Chunking configuration. If None, uses default ChunkConfig.
            cache: Optional shared Cacher instance. If None, creates a new one.
            **kwargs: Additional keyword arguments passed to the parent class.
        """
        super().__init__(**kwargs)
        # The model itself is process-shared and its construction is locked by
        # get_shared_encoder; all this holds is the per-tool adapter around it.
        self._embeddings: SharedSentenceTransformerEmbeddings | None = None
        self._embeddings_unavailable = False

        # Initialize cache - use shared cacher or create new one
        if cache is not None:
            self.cache = ToolCacher(cache, CHUNKER_CACHE_SUBDIR)
        else:
            # Standalone use (CLI helpers, direct library use): fall back to a
            # private Cacher on the configured/default directory.
            shared_cache = Cacher()
            self.cache = ToolCacher(shared_cache, CHUNKER_CACHE_SUBDIR)

        # Override config if provided
        if chunk_config is not None:
            self.config = chunk_config

        # Probe heavy deps only when semantic mode is requested
        if self.chunking_mode == "semantic" and not _semantic_chunking_available():
            self.chunking_mode = "naive"
            logger.warning(
                "Semantic chunking not available (needs the 'semantic-chunking' "
                "extra: sentence-transformers, hdbscan, umap-learn). "
                "Falling back to naive chunking."
            )

    def embeddings(self) -> SharedSentenceTransformerEmbeddings | None:
        """Embeddings over the process-shared encoder, or ``None`` if unavailable.

        The encoder is shared with retrieval and entity clustering when their
        model names match, so this loads no weights of its own in that case, and
        its inference is serialised against theirs.
        """
        if self._embeddings is not None or self._embeddings_unavailable:
            return self._embeddings
        if not _embedding_model_available():
            self._embeddings_unavailable = True
            return None
        try:
            self._embeddings = SharedSentenceTransformerEmbeddings(
                get_shared_encoder(
                    self.config.embedding_model,
                    feature=(
                        "Semantic chunking and schema detection. Install the "
                        "'semantic-chunking' extra"
                    ),
                ),
                normalize=False,
            )
        except Exception as exc:
            # Record the failure rather than retrying the load on every call:
            # a missing or broken checkpoint does not become available later in
            # the same process.
            logger.error("Failed to initialize chunker embedding model: %s", exc)
            self._embeddings_unavailable = True
            return None
        return self._embeddings

    def embed_texts(self, texts: list[str]) -> list[list[float]] | None:
        """Embed short texts with the chunker's model, or ``None`` if unavailable.

        Exposed so document-type detection can reuse the model already loaded
        for semantic chunking instead of constructing a second one. Returns
        ``None`` -- rather than raising -- when the semantic extras are absent,
        so callers degrade to their deterministic tiers exactly as chunking
        itself degrades to ``naive``.

        Args:
            texts: Short strings to embed (headings or sampled paragraphs).

        Returns:
            One embedding per input, or ``None`` when no model is available.
        """
        if not texts:
            return []
        embeddings = self.embeddings()
        if embeddings is None:
            return None
        try:
            return embeddings.embed_documents(texts)
        except Exception as exc:  # pragma: no cover - environment dependent
            logger.warning("Embedding failed, skipping semantic tier: %s", exc)
            return None

    def naive_split(self, doc: str) -> list[str]:
        """Split text by paragraph/sentence boundaries up to ``max_size``.

        Unlike :meth:`_naive_chunk`, does not enforce ``min_size`` filtering.
        """
        paragraphs = re.split(r"\n\s*\n", doc.strip())

        chunks: list[str] = []
        current_chunk = ""

        for paragraph in paragraphs:
            paragraph = paragraph.strip()
            if not paragraph:
                continue

            if (
                current_chunk
                and len(current_chunk) + len(paragraph) + 2 > self.config.max_size
            ):
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = paragraph
            else:
                if current_chunk:
                    current_chunk += "\n\n" + paragraph
                else:
                    current_chunk = paragraph

            if len(current_chunk) > self.config.max_size:
                if len(current_chunk) - len(paragraph) - 2 > 0:
                    prev_chunk = current_chunk[
                        : len(current_chunk) - len(paragraph) - 2
                    ].strip()
                    if prev_chunk:
                        chunks.append(prev_chunk)

                sentences = re.split(r"(?<=[.!?])\s+", paragraph)
                temp_chunk = ""

                for sentence in sentences:
                    if len(temp_chunk) + len(sentence) + 1 > self.config.max_size:
                        if temp_chunk:
                            chunks.append(temp_chunk.strip())
                        temp_chunk = sentence
                    else:
                        if temp_chunk:
                            temp_chunk += " " + sentence
                        else:
                            temp_chunk = sentence

                current_chunk = temp_chunk

        if current_chunk:
            chunks.append(current_chunk.strip())

        return chunks

    def size_text(self, doc: str) -> list[str]:
        """Split ``doc`` to respect ``min_size`` / ``max_size`` using naive boundaries."""
        return size_bounded_text(doc, self.config, self.naive_split)

    def _naive_chunk(self, doc: str) -> list[str]:
        """Naive chunking fallback when semantic chunking is not available.

        Args:
            doc: The document text to chunk.

        Returns:
            List of text chunks.
        """
        chunks = self.size_text(doc)

        logger.info(f"Naive chunking produced {len(chunks)} chunks")
        return chunks

    def __call__(self, doc: str) -> list[str]:
        """Chunk a document into semantic segments.

        Args:
            doc: The document text to chunk.

        Returns:
            List of text chunks.
        """
        # Prepare configuration for caching. The "model" key name is kept even
        # though its source moved to ChunkConfig -- the dict is hashed, so
        # renaming it would invalidate every cached chunking for no reason.
        config_dict = {
            "model": self.config.embedding_model,
            "chunking_mode": self.chunking_mode,
            "max_size": self.config.max_size,
            "min_size": self.config.min_size,
        }

        # Check cache first
        cached_result = self.cache.get(doc, config=config_dict)
        if cached_result is not None:
            logger.debug("Cache hit for document chunking")
            return cached_result

        # Perform chunking
        embeddings = None if self.chunking_mode == "naive" else self.embeddings()
        if embeddings is None or not _semantic_chunking_available():
            if self.chunking_mode != "naive":
                logger.warning(
                    "Semantic chunking requested but not available. "
                    "Falling back to naive chunking."
                )
            result = self._naive_chunk(doc)
        else:
            from ontocast.tool.chunk.util import SemanticChunker

            text_splitter = SemanticChunker(
                embeddings=embeddings,
                chunk_config=self.config,
                sentence_split_regex=SENTENCE_SPLIT_REGEX,
            )

            try:
                # SemanticChunker now handles max_size internally
                result_docs = text_splitter.create_documents([doc])
                result = [chunk.page_content for chunk in result_docs]
            except ValueError as exc:
                # Degenerate inputs (too few distinct sentences for the
                # HDBSCAN neighborhood) must not fail chunking outright.
                logger.warning(
                    "Semantic chunking failed (%s); falling back to "
                    "naive chunking for this text.",
                    exc,
                )
                result = self._naive_chunk(doc)

            # Log chunk lengths for debugging
            lens = [len(chunk) for chunk in result]
            logger.info(
                f"Semantic chunking produced {len(result)} chunks with lengths: {lens}"
            )

        # Cache the result
        self.cache.set(doc, result, config=config_dict)
        logger.debug("Cached document chunking result")

        return result

__call__(doc)

Chunk a document into semantic segments.

Parameters:

Name	Type	Description	Default
doc	str	The document text to chunk.	required

Returns:

Type	Description
list[str]	List of text chunks.

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

def __call__(self, doc: str) -> list[str]:
    """Chunk a document into semantic segments.

    Args:
        doc: The document text to chunk.

    Returns:
        List of text chunks.
    """
    # Prepare configuration for caching. The "model" key name is kept even
    # though its source moved to ChunkConfig -- the dict is hashed, so
    # renaming it would invalidate every cached chunking for no reason.
    config_dict = {
        "model": self.config.embedding_model,
        "chunking_mode": self.chunking_mode,
        "max_size": self.config.max_size,
        "min_size": self.config.min_size,
    }

    # Check cache first
    cached_result = self.cache.get(doc, config=config_dict)
    if cached_result is not None:
        logger.debug("Cache hit for document chunking")
        return cached_result

    # Perform chunking
    embeddings = None if self.chunking_mode == "naive" else self.embeddings()
    if embeddings is None or not _semantic_chunking_available():
        if self.chunking_mode != "naive":
            logger.warning(
                "Semantic chunking requested but not available. "
                "Falling back to naive chunking."
            )
        result = self._naive_chunk(doc)
    else:
        from ontocast.tool.chunk.util import SemanticChunker

        text_splitter = SemanticChunker(
            embeddings=embeddings,
            chunk_config=self.config,
            sentence_split_regex=SENTENCE_SPLIT_REGEX,
        )

        try:
            # SemanticChunker now handles max_size internally
            result_docs = text_splitter.create_documents([doc])
            result = [chunk.page_content for chunk in result_docs]
        except ValueError as exc:
            # Degenerate inputs (too few distinct sentences for the
            # HDBSCAN neighborhood) must not fail chunking outright.
            logger.warning(
                "Semantic chunking failed (%s); falling back to "
                "naive chunking for this text.",
                exc,
            )
            result = self._naive_chunk(doc)

        # Log chunk lengths for debugging
        lens = [len(chunk) for chunk in result]
        logger.info(
            f"Semantic chunking produced {len(result)} chunks with lengths: {lens}"
        )

    # Cache the result
    self.cache.set(doc, result, config=config_dict)
    logger.debug("Cached document chunking result")

    return result

__init__(chunk_config=None, cache=None, **kwargs)

Initialize the ChunkerTool.

Parameters:

Name	Type	Description	Default
chunk_config	ChunkConfig | None	Chunking configuration. If None, uses default ChunkConfig.	None
cache	Cacher | None	Optional shared Cacher instance. If None, creates a new one.	None
**kwargs		Additional keyword arguments passed to the parent class.	{}

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

def __init__(
    self,
    chunk_config: ChunkConfig | None = None,
    cache: Cacher | None = None,
    **kwargs,
):
    """Initialize the ChunkerTool.

    Args:
        chunk_config: Chunking configuration. If None, uses default ChunkConfig.
        cache: Optional shared Cacher instance. If None, creates a new one.
        **kwargs: Additional keyword arguments passed to the parent class.
    """
    super().__init__(**kwargs)
    # The model itself is process-shared and its construction is locked by
    # get_shared_encoder; all this holds is the per-tool adapter around it.
    self._embeddings: SharedSentenceTransformerEmbeddings | None = None
    self._embeddings_unavailable = False

    # Initialize cache - use shared cacher or create new one
    if cache is not None:
        self.cache = ToolCacher(cache, CHUNKER_CACHE_SUBDIR)
    else:
        # Standalone use (CLI helpers, direct library use): fall back to a
        # private Cacher on the configured/default directory.
        shared_cache = Cacher()
        self.cache = ToolCacher(shared_cache, CHUNKER_CACHE_SUBDIR)

    # Override config if provided
    if chunk_config is not None:
        self.config = chunk_config

    # Probe heavy deps only when semantic mode is requested
    if self.chunking_mode == "semantic" and not _semantic_chunking_available():
        self.chunking_mode = "naive"
        logger.warning(
            "Semantic chunking not available (needs the 'semantic-chunking' "
            "extra: sentence-transformers, hdbscan, umap-learn). "
            "Falling back to naive chunking."
        )

embed_texts(texts)

Embed short texts with the chunker's model, or None if unavailable.

Exposed so document-type detection can reuse the model already loaded for semantic chunking instead of constructing a second one. Returns None -- rather than raising -- when the semantic extras are absent, so callers degrade to their deterministic tiers exactly as chunking itself degrades to naive.

Parameters:

Name	Type	Description	Default
texts	list[str]	Short strings to embed (headings or sampled paragraphs).	required

Returns:

Type	Description
list[list[float]] | None	One embedding per input, or None when no model is available.

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

def embed_texts(self, texts: list[str]) -> list[list[float]] | None:
    """Embed short texts with the chunker's model, or ``None`` if unavailable.

    Exposed so document-type detection can reuse the model already loaded
    for semantic chunking instead of constructing a second one. Returns
    ``None`` -- rather than raising -- when the semantic extras are absent,
    so callers degrade to their deterministic tiers exactly as chunking
    itself degrades to ``naive``.

    Args:
        texts: Short strings to embed (headings or sampled paragraphs).

    Returns:
        One embedding per input, or ``None`` when no model is available.
    """
    if not texts:
        return []
    embeddings = self.embeddings()
    if embeddings is None:
        return None
    try:
        return embeddings.embed_documents(texts)
    except Exception as exc:  # pragma: no cover - environment dependent
        logger.warning("Embedding failed, skipping semantic tier: %s", exc)
        return None

embeddings()

Embeddings over the process-shared encoder, or None if unavailable.

The encoder is shared with retrieval and entity clustering when their model names match, so this loads no weights of its own in that case, and its inference is serialised against theirs.

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

def embeddings(self) -> SharedSentenceTransformerEmbeddings | None:
    """Embeddings over the process-shared encoder, or ``None`` if unavailable.

    The encoder is shared with retrieval and entity clustering when their
    model names match, so this loads no weights of its own in that case, and
    its inference is serialised against theirs.
    """
    if self._embeddings is not None or self._embeddings_unavailable:
        return self._embeddings
    if not _embedding_model_available():
        self._embeddings_unavailable = True
        return None
    try:
        self._embeddings = SharedSentenceTransformerEmbeddings(
            get_shared_encoder(
                self.config.embedding_model,
                feature=(
                    "Semantic chunking and schema detection. Install the "
                    "'semantic-chunking' extra"
                ),
            ),
            normalize=False,
        )
    except Exception as exc:
        # Record the failure rather than retrying the load on every call:
        # a missing or broken checkpoint does not become available later in
        # the same process.
        logger.error("Failed to initialize chunker embedding model: %s", exc)
        self._embeddings_unavailable = True
        return None
    return self._embeddings

naive_split(doc)

Split text by paragraph/sentence boundaries up to max_size.

Unlike :meth:_naive_chunk, does not enforce min_size filtering.

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

def naive_split(self, doc: str) -> list[str]:
    """Split text by paragraph/sentence boundaries up to ``max_size``.

    Unlike :meth:`_naive_chunk`, does not enforce ``min_size`` filtering.
    """
    paragraphs = re.split(r"\n\s*\n", doc.strip())

    chunks: list[str] = []
    current_chunk = ""

    for paragraph in paragraphs:
        paragraph = paragraph.strip()
        if not paragraph:
            continue

        if (
            current_chunk
            and len(current_chunk) + len(paragraph) + 2 > self.config.max_size
        ):
            if current_chunk:
                chunks.append(current_chunk.strip())
            current_chunk = paragraph
        else:
            if current_chunk:
                current_chunk += "\n\n" + paragraph
            else:
                current_chunk = paragraph

        if len(current_chunk) > self.config.max_size:
            if len(current_chunk) - len(paragraph) - 2 > 0:
                prev_chunk = current_chunk[
                    : len(current_chunk) - len(paragraph) - 2
                ].strip()
                if prev_chunk:
                    chunks.append(prev_chunk)

            sentences = re.split(r"(?<=[.!?])\s+", paragraph)
            temp_chunk = ""

            for sentence in sentences:
                if len(temp_chunk) + len(sentence) + 1 > self.config.max_size:
                    if temp_chunk:
                        chunks.append(temp_chunk.strip())
                    temp_chunk = sentence
                else:
                    if temp_chunk:
                        temp_chunk += " " + sentence
                    else:
                        temp_chunk = sentence

            current_chunk = temp_chunk

    if current_chunk:
        chunks.append(current_chunk.strip())

    return chunks

size_text(doc)

Split doc to respect min_size / max_size using naive boundaries.

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

def size_text(self, doc: str) -> list[str]:
    """Split ``doc`` to respect ``min_size`` / ``max_size`` using naive boundaries."""
    return size_bounded_text(doc, self.config, self.naive_split)

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)

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

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

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)

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

size_bounded_text(text, config, split_fn, *, separator=DEFAULT_PART_SEPARATOR)

Split text when needed, then enforce OntoCast chunk size bounds.

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

def size_bounded_text(
    text: str,
    config: ChunkConfig,
    split_fn: Callable[[str], list[str]],
    *,
    separator: str = DEFAULT_PART_SEPARATOR,
) -> list[str]:
    """Split ``text`` when needed, then enforce OntoCast chunk size bounds."""
    text = text.strip()
    if not text:
        return []

    if len(text) > config.max_size:
        parts = [part.strip() for part in split_fn(text) if part.strip()]
        if not parts:
            parts = [text]
    else:
        parts = [text]

    return size_text_parts(
        parts,
        config.min_size,
        config.max_size,
        separator=separator,
    )