Skip to content

ontocast.tool.chunk.chunker

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 
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
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 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
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 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
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 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
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 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
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 
222
223
224
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)