windlass.interfaces.chunker

chunker

The chunker interface.

Chunking is the single highest-leverage decision in a RAG system: it determines what the retriever can find and what the model gets to read. Windlass therefore treats it as a first-class, swappable strategy rather than a helper function.

Implementers override one method, :meth:Chunker.split_text, which works on a plain string. The base class handles document iteration, metadata propagation, offset tracking and id assignment — so a custom chunker is genuinely a dozen lines.

Example

from windlass.providers.chunkers.recursive import RecursiveChunker chunker = RecursiveChunker(chunk_size=60, overlap=0, min_chunk_size=0) len(chunker.chunk_text("word " * 60)) >= 2 True

Chunker

Chunker(
    *,
    chunk_size: int = 1000,
    overlap: int | None = None,
    min_chunk_size: int = 50,
    keep_separator: bool = True,
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract text splitter.

Parameters:

Name	Type	Description	Default
chunk_size	int	Target size of each chunk, measured in :attr:unit.	1000
overlap	int | None	How much each chunk repeats from the previous one. Overlap keeps a fact that straddles a boundary retrievable from either side. None scales it to :data:DEFAULT_OVERLAP_RATIO of chunk_size, so shrinking chunk_size alone always produces a working chunker.	None
min_chunk_size	int	Chunks shorter than this are merged into their neighbour rather than emitted — stray one-line fragments pollute retrieval.	50
keep_separator	bool	Whether the split separator stays in the output.	True
name	str | None	Component name for traces.	None
**config	Any	Strategy-specific options.	{}

Attributes:

Name	Type	Description
unit	str	"char" or "token" — how chunk_size is measured.

Raises:

Type	Description
ValueError	If an explicit overlap is negative or not smaller than chunk_size. A derived default is never invalid.

Example

Implementing a chunker takes one method::

class LineChunker(Chunker):
    provider_name = "line"

    def split_text(self, text: str) -> list[str]:
        return [ln for ln in text.splitlines() if ln.strip()]

Source code in src\windlass\interfaces\chunker.py

def __init__(
    self,
    *,
    chunk_size: int = 1000,
    overlap: int | None = None,
    min_chunk_size: int = 50,
    keep_separator: bool = True,
    name: str | None = None,
    **config: Any,
) -> None:
    if chunk_size <= 0:
        raise ValueError("chunk_size must be positive")

    if overlap is None:
        # Scale with chunk_size. A fixed default would make the perfectly
        # reasonable `chunker(chunk_size=20)` raise, which is a poor trade:
        # the user changed one obvious knob and got an error about another
        # they never set.
        overlap = int(chunk_size * DEFAULT_OVERLAP_RATIO)
    elif overlap < 0:
        raise ValueError("overlap must not be negative")
    elif overlap >= chunk_size:
        raise ValueError(
            f"overlap ({overlap}) must be smaller than chunk_size ({chunk_size}); "
            "otherwise chunking cannot make progress."
        )
    super().__init__(
        name=name or self.provider_name,
        chunk_size=chunk_size,
        overlap=overlap,
        min_chunk_size=min_chunk_size,
        keep_separator=keep_separator,
        **config,
    )
    self.chunk_size = chunk_size
    self.overlap = overlap
    self.min_chunk_size = min_chunk_size
    self.keep_separator = keep_separator

split_text abstractmethod

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

Split raw text into chunk strings.

The only method a strategy must implement. Return the pieces in reading order; the base class turns them into :class:Chunk objects with ids, offsets and inherited metadata.

Parameters:

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

Returns:

Type	Description
list[str]	Chunk strings in document order.

Source code in src\windlass\interfaces\chunker.py

@abc.abstractmethod
def split_text(self, text: str) -> list[str]:
    """Split raw text into chunk strings.

    The only method a strategy must implement. Return the pieces in reading
    order; the base class turns them into :class:`Chunk` objects with ids,
    offsets and inherited metadata.

    Args:
        text: The text to split.

    Returns:
        Chunk strings in document order.
    """

asplit_text async

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

Async :meth:split_text.

Override this instead when your strategy needs to await something — the semantic chunker calls an embedding model here.

Parameters:

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

Returns:

Type	Description
list[str]	Chunk strings in document order.

Source code in src\windlass\interfaces\chunker.py

async def asplit_text(self, text: str) -> list[str]:
    """Async :meth:`split_text`.

    Override this instead when your strategy needs to await something — the
    semantic chunker calls an embedding model here.

    Args:
        text: The text to split.

    Returns:
        Chunk strings in document order.
    """
    return self.split_text(text)

achunk_document async

achunk_document(document: Document) -> list[Chunk]

Split one document into chunks.

Parameters:

Name	Type	Description	Default
document	Document	The document to split.	required

Returns:

Type	Description
list[Chunk]	Chunks carrying the document's metadata plus chunk_index,
list[Chunk]	chunk_total and character offsets.

Source code in src\windlass\interfaces\chunker.py

async def achunk_document(self, document: Document) -> list[Chunk]:
    """Split one document into chunks.

    Args:
        document: The document to split.

    Returns:
        Chunks carrying the document's metadata plus ``chunk_index``,
        ``chunk_total`` and character offsets.
    """
    pieces = await self.asplit_text(document.content)
    pieces = self._merge_undersized(pieces)
    chunks: list[Chunk] = []
    cursor = 0
    for index, piece in enumerate(pieces):
        start = document.content.find(piece, cursor)
        if start < 0:  # strategy rewrote the text (e.g. added a heading path)
            start = cursor
        else:
            cursor = start + len(piece)
        chunks.append(
            Chunk(
                content=piece,
                document_id=document.id,
                index=index,
                start_char=start,
                end_char=start + len(piece),
                metadata={
                    **document.metadata,
                    "document_id": document.id,
                    "chunk_index": index,
                    "chunk_total": len(pieces),
                    "chunker": self.name,
                    **({"source": document.source} if document.source else {}),
                },
            )
        )
    return chunks

achunk async

achunk(
    documents: Sequence[Document] | Document, *, concurrency: int | None = None
) -> list[Chunk]

Split many documents concurrently.

Parameters:

Name	Type	Description	Default
documents	Sequence[Document] | Document	One document or a sequence of them.	required
concurrency	int | None	Maximum simultaneous splits. Defaults to the global max_concurrency setting.	None

Returns:

Type	Description
list[Chunk]	All chunks, grouped by source document in input order.

Performance

CPU-bound strategies see little benefit from concurrency; strategies that await a model (semantic, contextual) see a lot.

Source code in src\windlass\interfaces\chunker.py

async def achunk(
    self, documents: Sequence[Document] | Document, *, concurrency: int | None = None
) -> list[Chunk]:
    """Split many documents concurrently.

    Args:
        documents: One document or a sequence of them.
        concurrency: Maximum simultaneous splits. Defaults to the global
            ``max_concurrency`` setting.

    Returns:
        All chunks, grouped by source document in input order.

    Performance:
        CPU-bound strategies see little benefit from concurrency; strategies
        that await a model (``semantic``, ``contextual``) see a lot.
    """
    items = [documents] if isinstance(documents, Document) else list(documents)
    if not items:
        return []
    limit = concurrency or settings().max_concurrency
    grouped = await gather_bounded([self.achunk_document(doc) for doc in items], limit=limit)
    return [chunk for group in grouped for chunk in group]

chunk

chunk(documents: Sequence[Document] | Document) -> list[Chunk]

Blocking :meth:achunk.

Source code in src\windlass\interfaces\chunker.py

def chunk(self, documents: Sequence[Document] | Document) -> list[Chunk]:
    """Blocking :meth:`achunk`."""
    return run_sync(self.achunk(documents))

chunk_text

chunk_text(text: str, *, metadata: dict[str, Any] | None = None) -> list[Chunk]

Split a bare string, without constructing a Document first.

Parameters:

Name	Type	Description	Default
text	str	The text to split.	required
metadata	dict[str, Any] | None	Metadata copied onto every chunk.	None

Returns:

Type	Description
list[Chunk]	The resulting chunks.

Example

from windlass.providers.chunkers.recursive import RecursiveChunker RecursiveChunker(chunk_size=100, overlap=10).chunk_text("hello")[0].content 'hello'

Source code in src\windlass\interfaces\chunker.py

def chunk_text(self, text: str, *, metadata: dict[str, Any] | None = None) -> list[Chunk]:
    """Split a bare string, without constructing a Document first.

    Args:
        text: The text to split.
        metadata: Metadata copied onto every chunk.

    Returns:
        The resulting chunks.

    Example:
        >>> from windlass.providers.chunkers.recursive import RecursiveChunker
        >>> RecursiveChunker(chunk_size=100, overlap=10).chunk_text("hello")[0].content
        'hello'
    """
    return run_sync(self.achunk(Document(content=text, metadata=dict(metadata or {}))))