windlass.providers.chunkers.semantic

semantic

Semantic chunking — split where the meaning changes, not where the count runs out.

Fixed-size chunking cuts through the middle of an argument as often as not. Semantic chunking embeds each sentence, walks the document measuring the similarity between consecutive sentences, and starts a new chunk wherever that similarity drops sharply. Boundaries land at topic shifts.

The cost is one embedding call per sentence at ingestion time. That is real, but it is paid once and it improves every retrieval afterwards.

Example

from windlass.providers.embeddings.hash import HashEmbedder text = ( ... "Cats are mammals. Cats purr when content. Cats groom themselves often. " ... "Rockets burn fuel. Rockets reach orbit. Rockets carry satellites." ... ) chunker = SemanticChunker(embedder=HashEmbedder(dimensions=256), threshold=0.5) len(chunker.split_text(text)) >= 1 True

SemanticChunker

SemanticChunker(
    *,
    embedder: Embedder | None = None,
    chunk_size: int = 2000,
    overlap: int = 0,
    threshold: float = 0.5,
    percentile: float | None = 20.0,
    buffer_size: int = 2,
    min_sentences: int = 2,
    **config: Any
)

Bases: Chunker

Embedding-driven topic-boundary splitter.

Parameters:

Name	Type	Description	Default
embedder	Embedder | None	Embedding model used to score sentence similarity. Required — the RAG builder injects the pipeline's embedder automatically.	None
chunk_size	int	Upper bound in characters. A topic that runs longer than this is still split, so no chunk can blow past your context budget.	2000
overlap	int	Characters repeated between chunks.	0
threshold	float	Similarity below which a boundary is placed. Ignored when percentile is set.	0.5
percentile	float | None	Adaptive alternative to threshold: place a boundary wherever similarity falls into the lowest percentile of the document's own distribution. More robust across mixed corpora, since "low similarity" means different things in a legal contract and a chat log.	20.0
buffer_size	int	How many neighbouring sentences to average into each comparison. 1 compares single sentences and is noisy; 2-3 smooths it out.	2
min_sentences	int	Minimum sentences per chunk, to avoid one-line chunks.	2
**config	Any	Forwarded to :class:~windlass.interfaces.chunker.Chunker.	{}

Raises:

Type	Description
ConfigurationError	When no embedder is supplied.

Performance

Embeds every sentence once per document. Sentence embeddings are batched into a single call, and the embedder's cache (when configured) makes re-ingestion nearly free.

Source code in src\windlass\providers\chunkers\semantic.py

def __init__(
    self,
    *,
    embedder: Embedder | None = None,
    chunk_size: int = 2000,
    overlap: int = 0,
    threshold: float = 0.5,
    percentile: float | None = 20.0,
    buffer_size: int = 2,
    min_sentences: int = 2,
    **config: Any,
) -> None:
    super().__init__(chunk_size=chunk_size, overlap=overlap, **config)
    if embedder is None:
        raise ConfigurationError(
            "Semantic chunking needs an embedding model.",
            hint="Pass one explicitly — Windlass.chunker('semantic', "
            "embedder=Windlass.embedding('huggingface')) — or use the RAG "
            "builder, which injects the pipeline's embedder for you.",
        )
    self.embedder = embedder
    self.threshold = threshold
    self.percentile = percentile
    self.buffer_size = max(1, buffer_size)
    self.min_sentences = max(1, min_sentences)

split_text

split_text(text: str) -> list[str]

Blocking :meth:asplit_text.

Source code in src\windlass\providers\chunkers\semantic.py

def split_text(self, text: str) -> list[str]:
    """Blocking :meth:`asplit_text`."""
    return run_sync(self.asplit_text(text))

asplit_text async

asplit_text(text: str) -> list[str]

Split text at detected topic boundaries.

Parameters:

Name	Type	Description	Default
text	str	The text to split.	required

Returns:

Type	Description
list[str]	Chunk strings. Short inputs (fewer than 2 * min_sentences
list[str]	sentences) are returned whole rather than force-split.

Source code in src\windlass\providers\chunkers\semantic.py

async def asplit_text(self, text: str) -> list[str]:
    """Split ``text`` at detected topic boundaries.

    Args:
        text: The text to split.

    Returns:
        Chunk strings. Short inputs (fewer than ``2 * min_sentences``
        sentences) are returned whole rather than force-split.
    """
    sentences = split_sentences(text)
    if len(sentences) < max(2, self.min_sentences * 2):
        stripped = text.strip()
        return [stripped] if stripped else []

    windows = self._windows(sentences)
    vectors = await self.embedder.aembed(windows)
    similarities = [
        cosine_similarity(vectors[i], vectors[i + 1]) for i in range(len(vectors) - 1)
    ]
    cutoff = self._cutoff(similarities)
    boundaries = self._boundaries(similarities, cutoff)
    return self._assemble(sentences, boundaries)