Chunking¶

The strategy registry and its five members: recursive_character (the v1 default) plus the v2 additions token_based, markdown_aware, docling_hybrid, and semantic. Mind that CHUNK_SIZE's unit is per-strategy (characters vs tokens — see Configuration).

`varagity.chunking.base` ¶

Chunking-strategy protocol and registry (the spec §5.1 sketch, typed).

Each strategy module defines one implementation decorated with @register("name"); callers resolve the configured strategy with get_chunker(settings.CHUNKING_STRATEGY).

`ChunkingStrategy` ¶

Bases: Protocol

Interface every chunking strategy implements.

Source code in varagity/chunking/base.py

class ChunkingStrategy(Protocol):
    """Interface every chunking strategy implements."""

    def split(
        self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
    ) -> list[Document]:
        """Split a document's text into chunks.

        Args:
            text: The full document text.
            source_meta: Provenance seeded into every chunk's metadata
                (``source``, ``file_name``, ``file_type``, ``page``).
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The chunks, each with metadata seeded plus its ``chunk_index``.
        """
        ...

`split(text, *, source_meta, verbose=None)` ¶

Split a document's text into chunks.

Parameters:

Name	Type	Description	Default
`text`	`str`	The full document text.	required
`source_meta`	`dict[str, Any]`	Provenance seeded into every chunk's metadata (`source`, `file_name`, `file_type`, `page`).	required
`verbose`	`int \| None`	Console verbosity (0–2); defaults to `settings.DEFAULT_VERBOSE`.	`None`

Returns:

Type	Description
`list[Document]`	The chunks, each with metadata seeded plus its `chunk_index`.

Source code in varagity/chunking/base.py

def split(
    self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
) -> list[Document]:
    """Split a document's text into chunks.

    Args:
        text: The full document text.
        source_meta: Provenance seeded into every chunk's metadata
            (``source``, ``file_name``, ``file_type``, ``page``).
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The chunks, each with metadata seeded plus its ``chunk_index``.
    """
    ...

`get_chunker(name)` ¶

Look up a registered chunking strategy by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Registry key (e.g. `"recursive_character"`).	required

Returns:

Type	Description
`ChunkingStrategy`	The registered strategy instance.

Raises:

Type	Description
`KeyError`	If no strategy is registered under `name` (message lists the available ones).

Source code in varagity/chunking/base.py

def get_chunker(name: str) -> ChunkingStrategy:
    """Look up a registered chunking strategy by name.

    Args:
        name: Registry key (e.g. ``"recursive_character"``).

    Returns:
        The registered strategy instance.

    Raises:
        KeyError: If no strategy is registered under ``name`` (message lists
            the available ones).
    """
    if name not in CHUNKER_REGISTRY:
        raise KeyError(f"Unknown chunking strategy {name!r}. Available: {list(CHUNKER_REGISTRY)}")
    return CHUNKER_REGISTRY[name]

`register(name)` ¶

Class decorator registering a chunking strategy instance under name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Registry key (the `CHUNKING_STRATEGY` env value).	required

Returns:

Type	Description
`Callable[[T], T]`	The decorator, which instantiates and registers the class unchanged.

Source code in varagity/chunking/base.py

def register[T: type[Any]](name: str) -> Callable[[T], T]:
    """Class decorator registering a chunking strategy instance under ``name``.

    Args:
        name: Registry key (the ``CHUNKING_STRATEGY`` env value).

    Returns:
        The decorator, which instantiates and registers the class unchanged.
    """

    def deco(cls: T) -> T:
        CHUNKER_REGISTRY[name] = cls()
        return cls

    return deco

`warn_near_token_ceiling(chunks, *, strategy)` ¶

Warn for chunks whose content nears e5's 512-token ceiling (research Q6).

The ingest-time warning :class:~varagity.models.embeddings.EmbeddingsClient fires per contextualized passage, carried forward to chunking time: a chunk already ≥ the threshold before its situating blurb is prepended is guaranteed to truncate at embedding time, and the strategy's sizing knobs — not the blurb length — are what to fix.

Parameters:

Name	Type	Description	Default
`chunks`	`Sequence[Document]`	The chunks produced for one document.	required
`strategy`	`str`	Registry name of the producing strategy (names the knob in the log line).	required

Source code in varagity/chunking/base.py

def warn_near_token_ceiling(chunks: Sequence[Document], *, strategy: str) -> None:
    """Warn for chunks whose content nears e5's 512-token ceiling (research Q6).

    The ingest-time warning :class:`~varagity.models.embeddings.EmbeddingsClient`
    fires per *contextualized* passage, carried forward to chunking time:
    a chunk already ≥ the threshold **before** its situating
    blurb is prepended is guaranteed to truncate at embedding time, and the
    strategy's sizing knobs — not the blurb length — are what to fix.

    Args:
        chunks: The chunks produced for one document.
        strategy: Registry name of the producing strategy (names the knob
            in the log line).
    """
    from varagity.models.embeddings import TOKEN_WARN_THRESHOLD

    for chunk in chunks:
        n_tokens = count_tokens(chunk.page_content)
        if n_tokens >= TOKEN_WARN_THRESHOLD:
            logger.warning(
                "chunk %s is ~%d tokens (≥%d) before contextualization: e5 truncates "
                "at 512 — check the %r sizing (CHUNK_SIZE / CHUNK_OVERLAP)",
                chunk.metadata.get("chunk_index", "?"),
                n_tokens,
                TOKEN_WARN_THRESHOLD,
                strategy,
            )

`varagity.chunking.recursive_character` ¶

Default v1 chunking strategy: recursive character splitting (spec §9.3).

Wraps langchain's RecursiveCharacterTextSplitter with the configured CHUNK_SIZE / CHUNK_OVERLAP. Both parameters count characters, not tokens (the splitter's length_function defaults to len); a token-based strategy would be a new registry file, not an edit here.

`RecursiveCharacterStrategy` ¶

Splits on a separator hierarchy (paragraph → line → word → char).

Source code in varagity/chunking/recursive_character.py

@register("recursive_character")
class RecursiveCharacterStrategy:
    """Splits on a separator hierarchy (paragraph → line → word → char)."""

    def split(
        self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
    ) -> list[Document]:
        """Split a document's text into overlapping character chunks.

        Args:
            text: The full document text.
            source_meta: Provenance copied into every chunk's metadata (the
                splitter deep-copies per chunk, so later per-chunk fields
                don't bleed across siblings).
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The chunks in document order, each carrying ``source_meta`` plus
            its ``chunk_index``.

        Raises:
            ValueError: If ``verbose`` is invalid.
        """
        settings = get_settings()
        verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=settings.CHUNK_SIZE,  # characters, not tokens (spec §9.3)
            chunk_overlap=settings.CHUNK_OVERLAP,
        )
        chunks = splitter.create_documents([text], metadatas=[dict(source_meta)])
        for chunk_index, chunk in enumerate(chunks):
            chunk.metadata["chunk_index"] = chunk_index
        v_chunk(chunks, verbose=verbose)
        return chunks

`split(text, *, source_meta, verbose=None)` ¶

Split a document's text into overlapping character chunks.

Parameters:

Name	Type	Description	Default
`text`	`str`	The full document text.	required
`source_meta`	`dict[str, Any]`	Provenance copied into every chunk's metadata (the splitter deep-copies per chunk, so later per-chunk fields don't bleed across siblings).	required
`verbose`	`int \| None`	Console verbosity (0–2); defaults to `settings.DEFAULT_VERBOSE`.	`None`

Returns:

Type	Description
`list[Document]`	The chunks in document order, each carrying `source_meta` plus
`list[Document]`	its `chunk_index`.

Raises:

Type	Description
`ValueError`	If `verbose` is invalid.

Source code in varagity/chunking/recursive_character.py

def split(
    self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
) -> list[Document]:
    """Split a document's text into overlapping character chunks.

    Args:
        text: The full document text.
        source_meta: Provenance copied into every chunk's metadata (the
            splitter deep-copies per chunk, so later per-chunk fields
            don't bleed across siblings).
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The chunks in document order, each carrying ``source_meta`` plus
        its ``chunk_index``.

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    settings = get_settings()
    verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=settings.CHUNK_SIZE,  # characters, not tokens (spec §9.3)
        chunk_overlap=settings.CHUNK_OVERLAP,
    )
    chunks = splitter.create_documents([text], metadatas=[dict(source_meta)])
    for chunk_index, chunk in enumerate(chunks):
        chunk.metadata["chunk_index"] = chunk_index
    v_chunk(chunks, verbose=verbose)
    return chunks

`varagity.chunking.token_based` ¶

Token-denominated recursive splitting (spec_v2 §7, v2 plan decision #12).

Wraps langchain's RecursiveCharacterTextSplitter.from_tiktoken_encoder: the same separator hierarchy as recursive_character (paragraph → line → word → char), but CHUNK_SIZE / CHUNK_OVERLAP count tokens, not characters — aligning chunk budgets to the e5 embedder's 512-token ceiling instead of the character approximation the v1 strategy documents.

Token counting uses tiktoken's cl100k_base — the codebase-wide documented approximation of e5's actual XLM-RoBERTa vocabulary (see varagity.tokens). An exact counter (the e5 HF tokenizer as a custom length_function) plugs into the __init__ seam without touching the registry; it was evaluated against the approximation in the chunker sweep rather than shipped (an extra transformers hot path + first-run tokenizer download for a small counting delta — ADR-008).

Note the langchain merge nuance: the splitter sizes chunks by summing piece lengths, and re-tokenizing joined text can differ by a token or two at the seams, so real-tokenizer chunks may exceed the budget by a whisker. The ≥480 warning (not a hard 512 assert) is the guard that matters.

`TokenBasedStrategy` ¶

Recursive separator splitting with token-denominated sizes.

Source code in varagity/chunking/token_based.py

@register("token_based")
class TokenBasedStrategy:
    """Recursive separator splitting with token-denominated sizes."""

    def __init__(self, *, length_function: Callable[[str], int] | None = None) -> None:
        """Create the strategy.

        Args:
            length_function: Alternative token counter (e.g. an exact e5 HF
                tokenizer ``lambda``, or a fake in tests). Defaults to the
                tiktoken ``cl100k_base`` counter via
                ``from_tiktoken_encoder``.
        """
        self._length_function = length_function

    def _splitter(self, chunk_size: int, chunk_overlap: int) -> RecursiveCharacterTextSplitter:
        """Build the token-sized splitter for the configured budgets.

        Args:
            chunk_size: Chunk budget, in tokens.
            chunk_overlap: Overlap between consecutive chunks, in tokens.

        Returns:
            The configured splitter.
        """
        if self._length_function is not None:
            return RecursiveCharacterTextSplitter(
                chunk_size=chunk_size,
                chunk_overlap=chunk_overlap,
                length_function=self._length_function,
            )
        return RecursiveCharacterTextSplitter.from_tiktoken_encoder(
            encoding_name=_ENCODING_NAME,
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
        )

    def split(
        self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
    ) -> list[Document]:
        """Split a document's text into overlapping token-budgeted chunks.

        Args:
            text: The full document text.
            source_meta: Provenance copied into every chunk's metadata (the
                splitter deep-copies per chunk).
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The chunks in document order, each carrying ``source_meta`` plus
            its ``chunk_index``.

        Raises:
            ValueError: If ``verbose`` is invalid.
        """
        settings = get_settings()
        verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
        splitter = self._splitter(settings.CHUNK_SIZE, settings.CHUNK_OVERLAP)  # tokens, not chars
        chunks = splitter.create_documents([text], metadatas=[dict(source_meta)])
        for chunk_index, chunk in enumerate(chunks):
            chunk.metadata["chunk_index"] = chunk_index
        warn_near_token_ceiling(chunks, strategy="token_based")
        v_chunk(chunks, verbose=verbose)
        return chunks

`init(*, length_function=None)` ¶

Create the strategy.

Parameters:

Name	Type	Description	Default
`length_function`	`Callable[[str], int] \| None`	Alternative token counter (e.g. an exact e5 HF tokenizer `lambda`, or a fake in tests). Defaults to the tiktoken `cl100k_base` counter via `from_tiktoken_encoder`.	`None`

Source code in varagity/chunking/token_based.py

def __init__(self, *, length_function: Callable[[str], int] | None = None) -> None:
    """Create the strategy.

    Args:
        length_function: Alternative token counter (e.g. an exact e5 HF
            tokenizer ``lambda``, or a fake in tests). Defaults to the
            tiktoken ``cl100k_base`` counter via
            ``from_tiktoken_encoder``.
    """
    self._length_function = length_function

`split(text, *, source_meta, verbose=None)` ¶

Split a document's text into overlapping token-budgeted chunks.

Parameters:

Name	Type	Description	Default
`text`	`str`	The full document text.	required
`source_meta`	`dict[str, Any]`	Provenance copied into every chunk's metadata (the splitter deep-copies per chunk).	required
`verbose`	`int \| None`	Console verbosity (0–2); defaults to `settings.DEFAULT_VERBOSE`.	`None`

Returns:

Type	Description
`list[Document]`	The chunks in document order, each carrying `source_meta` plus
`list[Document]`	its `chunk_index`.

Raises:

Type	Description
`ValueError`	If `verbose` is invalid.

Source code in varagity/chunking/token_based.py

def split(
    self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
) -> list[Document]:
    """Split a document's text into overlapping token-budgeted chunks.

    Args:
        text: The full document text.
        source_meta: Provenance copied into every chunk's metadata (the
            splitter deep-copies per chunk).
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The chunks in document order, each carrying ``source_meta`` plus
        its ``chunk_index``.

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    settings = get_settings()
    verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
    splitter = self._splitter(settings.CHUNK_SIZE, settings.CHUNK_OVERLAP)  # tokens, not chars
    chunks = splitter.create_documents([text], metadatas=[dict(source_meta)])
    for chunk_index, chunk in enumerate(chunks):
        chunk.metadata["chunk_index"] = chunk_index
    warn_near_token_ceiling(chunks, strategy="token_based")
    v_chunk(chunks, verbose=verbose)
    return chunks

`varagity.chunking.markdown_aware` ¶

Heading-aware markdown splitting (spec_v2 §7).

Every rich-format parser (PDF, office, web) exports structure-aware markdown through the shared Docling core — this strategy finally exploits that structure: split at ATX headings (# … ######), keep each section's text together, and carry the section's heading breadcrumb into chunk metadata as heading_path (e.g. "Harbor Operations > Dredging > Channel depth"), where the JSONB metadata column absorbs it with no schema migration and the evidence panel can surface it as breadcrumb context.

This is the from-scratch header splitter the v2 plan benchmarks against Docling's HybridChunker (shipped alongside as docling_hybrid; the chunker sweep decides the default — ADR-008). Behavior notes:

Headings inside fenced code blocks (three-plus backticks or tildes) are content, not structure.
A section's own heading line stays in its chunk text (retrieval-visible); ancestors live only in heading_path.
Sections are not merged across headings — coherence over packing; HybridChunker makes the opposite trade, which is exactly what the sweep measures.
A section longer than CHUNK_SIZE (characters, the v1 unit — this strategy sizes like recursive_character) is re-split recursively; every sub-chunk inherits the section's heading_path.
Text before the first heading forms a section with no heading_path.

`MarkdownAwareStrategy` ¶

Splits on ATX headings, carrying the heading breadcrumb as metadata.

Source code in varagity/chunking/markdown_aware.py

@register("markdown_aware")
class MarkdownAwareStrategy:
    """Splits on ATX headings, carrying the heading breadcrumb as metadata."""

    def split(
        self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
    ) -> list[Document]:
        """Split a document's markdown into heading-coherent chunks.

        Args:
            text: The full document text (markdown from the rich-format
                parsers; plain text degrades to one section, then the
                recursive re-split).
            source_meta: Provenance copied into every chunk's metadata.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The chunks in document order, each carrying ``source_meta``,
            its ``chunk_index``, and — for sections under a heading — the
            ``heading_path`` breadcrumb.

        Raises:
            ValueError: If ``verbose`` is invalid.
        """
        settings = get_settings()
        verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
        resplitter = RecursiveCharacterTextSplitter(
            chunk_size=settings.CHUNK_SIZE,  # characters, like recursive_character
            chunk_overlap=settings.CHUNK_OVERLAP,
        )
        chunks: list[Document] = []
        for path, section_text in _sections(text):
            texts = (
                resplitter.split_text(section_text)
                if len(section_text) > settings.CHUNK_SIZE
                else [section_text]
            )
            for piece in texts:
                metadata: dict[str, Any] = dict(source_meta)
                if path:
                    metadata["heading_path"] = _PATH_SEPARATOR.join(path)
                chunks.append(Document(page_content=piece, metadata=metadata))
        for chunk_index, chunk in enumerate(chunks):
            chunk.metadata["chunk_index"] = chunk_index
        warn_near_token_ceiling(chunks, strategy="markdown_aware")
        v_chunk(chunks, verbose=verbose)
        return chunks

`split(text, *, source_meta, verbose=None)` ¶

Split a document's markdown into heading-coherent chunks.

Parameters:

Name	Type	Description	Default
`text`	`str`	The full document text (markdown from the rich-format parsers; plain text degrades to one section, then the recursive re-split).	required
`source_meta`	`dict[str, Any]`	Provenance copied into every chunk's metadata.	required
`verbose`	`int \| None`	Console verbosity (0–2); defaults to `settings.DEFAULT_VERBOSE`.	`None`

Returns:

Type	Description
`list[Document]`	The chunks in document order, each carrying `source_meta`,
`list[Document]`	its `chunk_index`, and — for sections under a heading — the
`list[Document]`	`heading_path` breadcrumb.

Raises:

Type	Description
`ValueError`	If `verbose` is invalid.

Source code in varagity/chunking/markdown_aware.py

def split(
    self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
) -> list[Document]:
    """Split a document's markdown into heading-coherent chunks.

    Args:
        text: The full document text (markdown from the rich-format
            parsers; plain text degrades to one section, then the
            recursive re-split).
        source_meta: Provenance copied into every chunk's metadata.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The chunks in document order, each carrying ``source_meta``,
        its ``chunk_index``, and — for sections under a heading — the
        ``heading_path`` breadcrumb.

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    settings = get_settings()
    verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
    resplitter = RecursiveCharacterTextSplitter(
        chunk_size=settings.CHUNK_SIZE,  # characters, like recursive_character
        chunk_overlap=settings.CHUNK_OVERLAP,
    )
    chunks: list[Document] = []
    for path, section_text in _sections(text):
        texts = (
            resplitter.split_text(section_text)
            if len(section_text) > settings.CHUNK_SIZE
            else [section_text]
        )
        for piece in texts:
            metadata: dict[str, Any] = dict(source_meta)
            if path:
                metadata["heading_path"] = _PATH_SEPARATOR.join(path)
            chunks.append(Document(page_content=piece, metadata=metadata))
    for chunk_index, chunk in enumerate(chunks):
        chunk.metadata["chunk_index"] = chunk_index
    warn_near_token_ceiling(chunks, strategy="markdown_aware")
    v_chunk(chunks, verbose=verbose)
    return chunks

`varagity.chunking.docling_hybrid` ¶

Docling HybridChunker as a registered strategy (spec_v2 §7).

The alternative markdown_aware implementation the v2 plan benchmarks: Docling's tokenization- and structure-aware chunker, shipped as its own registry entry so the chunker eval sweep compares the two head-to-head and the default is benchmark-decided (ADR-008), not hand-picked.

The strategy re-parses the parser's markdown into a DoclingDocument (in-memory DocumentStream, markdown backend — declarative, no layout models) and runs HybridChunker over the document object model: one chunk per document item, oversized items token-split, undersized peers under the same headings merged (merge_peers — the packing trade markdown_aware declines). Heading provenance lands in heading_path exactly like markdown_aware. Sizing: CHUNK_SIZE is the tokenizer's max_tokens budget (tokens, tiktoken cl100k_base — the codebase-wide documented approximation); CHUNK_OVERLAP has no HybridChunker equivalent and is ignored.

Chunk text is the item text (chunk.text), not the chunker's heading-prefixed contextualize() rendering: Contextual Retrieval already prepends an LLM situating blurb at ingest (spec §9.4), and stacking two context prefixes would double-count headings and eat the 512-token budget.

Docling imports are deferred to call time (the parser-module convention): importing this module on CLI start must not pay for Docling's machinery.

`DoclingHybridStrategy` ¶

Docling HybridChunker over the re-parsed document structure.

Source code in varagity/chunking/docling_hybrid.py

@register("docling_hybrid")
class DoclingHybridStrategy:
    """Docling ``HybridChunker`` over the re-parsed document structure."""

    def __init__(self) -> None:
        """Initialize the converter/chunker caches (nothing heavy here)."""
        self._converter: DocumentConverter | None = None
        self._chunker: HybridChunker | None = None
        self._chunker_max_tokens: int | None = None

    def split(
        self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
    ) -> list[Document]:
        """Split a document's markdown with Docling's ``HybridChunker``.

        Args:
            text: The full document text (markdown from the rich-format
                parsers; plain text parses as markdown paragraphs).
            source_meta: Provenance copied into every chunk's metadata.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The chunks in document order, each carrying ``source_meta``,
            its ``chunk_index``, and — when Docling attributes headings —
            the ``heading_path`` breadcrumb.

        Raises:
            ValueError: If ``verbose`` is invalid.
        """
        from io import BytesIO

        from docling_core.types.io import DocumentStream

        settings = get_settings()
        verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
        converted = self._convert(
            DocumentStream(name=_STREAM_NAME, stream=BytesIO(text.encode("utf-8")))
        )
        chunker = self._hybrid_chunker(settings.CHUNK_SIZE)  # CHUNK_SIZE as tokens
        chunks: list[Document] = []
        for chunk_index, item in enumerate(chunker.chunk(converted)):
            metadata: dict[str, Any] = dict(source_meta)
            headings = getattr(item.meta, "headings", None)
            if headings:
                metadata["heading_path"] = _PATH_SEPARATOR.join(headings)
            metadata["chunk_index"] = chunk_index
            chunks.append(Document(page_content=item.text, metadata=metadata))
        warn_near_token_ceiling(chunks, strategy="docling_hybrid")
        v_chunk(chunks, verbose=verbose)
        return chunks

    def _convert(self, stream: Any) -> Any:
        """Re-parse markdown text into a ``DoclingDocument`` (cached converter).

        Args:
            stream: The in-memory ``DocumentStream`` to convert.

        Returns:
            The converted Docling document.
        """
        if self._converter is None:
            from docling.datamodel.base_models import InputFormat
            from docling.document_converter import DocumentConverter

            # Markdown-only: the input is always the parsers' markdown/plain
            # text, and restricting formats skips every heavyweight pipeline.
            self._converter = DocumentConverter(allowed_formats=[InputFormat.MD])
        return self._converter.convert(stream).document

    def _hybrid_chunker(self, max_tokens: int) -> "HybridChunker":
        """Build (or reuse) the ``HybridChunker`` for the configured budget.

        Args:
            max_tokens: Token budget per chunk (``CHUNK_SIZE``, as tokens).

        Returns:
            The configured chunker (rebuilt only when the budget changes,
            e.g. under the eval harness's pinned settings).
        """
        if self._chunker is None or self._chunker_max_tokens != max_tokens:
            import tiktoken
            from docling.chunking import HybridChunker
            from docling_core.transforms.chunker.tokenizer.openai import OpenAITokenizer

            tokenizer = OpenAITokenizer(
                tokenizer=tiktoken.get_encoding("cl100k_base"), max_tokens=max_tokens
            )
            self._chunker = HybridChunker(tokenizer=tokenizer, merge_peers=True)
            self._chunker_max_tokens = max_tokens
        return self._chunker

`init()` ¶

Initialize the converter/chunker caches (nothing heavy here).

Source code in varagity/chunking/docling_hybrid.py

def __init__(self) -> None:
    """Initialize the converter/chunker caches (nothing heavy here)."""
    self._converter: DocumentConverter | None = None
    self._chunker: HybridChunker | None = None
    self._chunker_max_tokens: int | None = None

`split(text, *, source_meta, verbose=None)` ¶

Split a document's markdown with Docling's HybridChunker.

Parameters:

Name	Type	Description	Default
`text`	`str`	The full document text (markdown from the rich-format parsers; plain text parses as markdown paragraphs).	required
`source_meta`	`dict[str, Any]`	Provenance copied into every chunk's metadata.	required
`verbose`	`int \| None`	Console verbosity (0–2); defaults to `settings.DEFAULT_VERBOSE`.	`None`

Returns:

Type	Description
`list[Document]`	The chunks in document order, each carrying `source_meta`,
`list[Document]`	its `chunk_index`, and — when Docling attributes headings —
`list[Document]`	the `heading_path` breadcrumb.

Raises:

Type	Description
`ValueError`	If `verbose` is invalid.

Source code in varagity/chunking/docling_hybrid.py

def split(
    self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
) -> list[Document]:
    """Split a document's markdown with Docling's ``HybridChunker``.

    Args:
        text: The full document text (markdown from the rich-format
            parsers; plain text parses as markdown paragraphs).
        source_meta: Provenance copied into every chunk's metadata.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The chunks in document order, each carrying ``source_meta``,
        its ``chunk_index``, and — when Docling attributes headings —
        the ``heading_path`` breadcrumb.

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    from io import BytesIO

    from docling_core.types.io import DocumentStream

    settings = get_settings()
    verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
    converted = self._convert(
        DocumentStream(name=_STREAM_NAME, stream=BytesIO(text.encode("utf-8")))
    )
    chunker = self._hybrid_chunker(settings.CHUNK_SIZE)  # CHUNK_SIZE as tokens
    chunks: list[Document] = []
    for chunk_index, item in enumerate(chunker.chunk(converted)):
        metadata: dict[str, Any] = dict(source_meta)
        headings = getattr(item.meta, "headings", None)
        if headings:
            metadata["heading_path"] = _PATH_SEPARATOR.join(headings)
        metadata["chunk_index"] = chunk_index
        chunks.append(Document(page_content=item.text, metadata=metadata))
    warn_near_token_ceiling(chunks, strategy="docling_hybrid")
    v_chunk(chunks, verbose=verbose)
    return chunks

`varagity.chunking.semantic` ¶

Embedding-similarity boundary detection (spec_v2 §7, plan decision #12).

Groups adjacent sentences and splits where the topic shifts: each segment is embedded (with one neighbor of context on each side, smoothing single-sentence noise) through the existing infinity embeddings client in passage mode, consecutive-segment cosine distances are computed, and a boundary is drawn wherever the distance is a statistical outlier — above the 95th percentile of the document's distances (LangChain's default percentile breakpoint; the gradient variant is avoided per its open KeyError bug — plan decision #12).

Two guarantees bound the output:

Exact reconstruction below the ceiling — segments are sliced from the original text with their separators attached, so a chunk is a verbatim substring and un-resplit chunks concatenate back to the document.
The 512-token ceiling — similarity groups are meaning-sized, not budget-sized, so any group exceeding CHUNK_SIZE (tokens, like token_based) is re-split with the token-denominated recursive splitter (CHUNK_OVERLAP tokens of overlap applies only there).

This strategy embeds at ingest time — one passage request per segment window, batched by EMBEDDING_BATCH_SIZE — an added ingest cost the chunker sweep measures before any default changes (ADR-008). The client resolves lazily via get_model("embedding") per split, so the registry's import-time instantiation performs no I/O; tests inject a deterministic fake instead.

`SemanticStrategy` ¶

Splits where consecutive-segment embedding similarity drops.

Source code in varagity/chunking/semantic.py

@register("semantic")
class SemanticStrategy:
    """Splits where consecutive-segment embedding similarity drops."""

    def __init__(self, *, embeddings: EmbeddingsClient | None = None) -> None:
        """Create the strategy.

        Args:
            embeddings: Embeddings client for boundary detection; resolved
                via the model registry (``get_model("embedding")``) per call
                when omitted. Tests inject a deterministic fake.
        """
        self._embeddings = embeddings

    def split(
        self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
    ) -> list[Document]:
        """Split a document's text at embedding-similarity boundaries.

        Args:
            text: The full document text.
            source_meta: Provenance copied into every chunk's metadata.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The chunks in document order, each carrying ``source_meta`` plus
            its ``chunk_index``.

        Raises:
            ValueError: If ``verbose`` is invalid.
            openai.APIError: If segment embedding still fails after the
                client's retries.
        """
        settings = get_settings()
        verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
        groups = self._similarity_groups(_segments(text))
        resplitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
            encoding_name="cl100k_base",
            chunk_size=settings.CHUNK_SIZE,  # tokens, like token_based
            chunk_overlap=settings.CHUNK_OVERLAP,
        )
        chunks: list[Document] = []
        for group in groups:
            group_text = group.strip()
            if not group_text:
                continue
            texts = (
                resplitter.split_text(group_text)
                if count_tokens(group_text) > settings.CHUNK_SIZE
                else [group_text]
            )
            chunks.extend(
                Document(page_content=piece, metadata=dict(source_meta)) for piece in texts
            )
        for chunk_index, chunk in enumerate(chunks):
            chunk.metadata["chunk_index"] = chunk_index
        warn_near_token_ceiling(chunks, strategy="semantic")
        v_chunk(chunks, verbose=verbose)
        return chunks

    def _similarity_groups(self, segments: list[str]) -> list[str]:
        """Group segments between similarity-outlier boundaries.

        Fewer than three segments yield a single group without an
        embeddings call (one consecutive distance can never be its own
        outlier under a strict-greater percentile cut).

        Args:
            segments: The document's segments, in order.

        Returns:
            The grouped texts, in order (verbatim concatenations of their
            segments).
        """
        if not segments:
            return []
        if len(segments) < 3:
            return ["".join(segments)]
        client = self._embeddings if self._embeddings is not None else get_model("embedding")
        windows = [
            "".join(segments[max(0, i - _BUFFER_SIZE) : i + _BUFFER_SIZE + 1]).strip()
            for i in range(len(segments))
        ]
        vectors = client.embed_passages(windows, verbose=0)
        distances = [
            _cosine_distance(left, right) for left, right in zip(vectors, vectors[1:], strict=False)
        ]
        threshold = _percentile(distances, _BREAKPOINT_PERCENTILE)
        groups: list[str] = []
        current: list[str] = []
        for index, segment in enumerate(segments):
            current.append(segment)
            if index < len(distances) and distances[index] > threshold:
                groups.append("".join(current))
                current = []
        if current:
            groups.append("".join(current))
        return groups

`init(*, embeddings=None)` ¶

Create the strategy.

Parameters:

Name	Type	Description	Default
`embeddings`	`EmbeddingsClient \| None`	Embeddings client for boundary detection; resolved via the model registry (`get_model("embedding")`) per call when omitted. Tests inject a deterministic fake.	`None`

Source code in varagity/chunking/semantic.py

def __init__(self, *, embeddings: EmbeddingsClient | None = None) -> None:
    """Create the strategy.

    Args:
        embeddings: Embeddings client for boundary detection; resolved
            via the model registry (``get_model("embedding")``) per call
            when omitted. Tests inject a deterministic fake.
    """
    self._embeddings = embeddings

`split(text, *, source_meta, verbose=None)` ¶

Split a document's text at embedding-similarity boundaries.

Parameters:

Name	Type	Description	Default
`text`	`str`	The full document text.	required
`source_meta`	`dict[str, Any]`	Provenance copied into every chunk's metadata.	required
`verbose`	`int \| None`	Console verbosity (0–2); defaults to `settings.DEFAULT_VERBOSE`.	`None`

Returns:

Type	Description
`list[Document]`	The chunks in document order, each carrying `source_meta` plus
`list[Document]`	its `chunk_index`.

Raises:

Type	Description
`ValueError`	If `verbose` is invalid.
`APIError`	If segment embedding still fails after the client's retries.

Source code in varagity/chunking/semantic.py

def split(
    self, text: str, *, source_meta: dict[str, Any], verbose: int | None = None
) -> list[Document]:
    """Split a document's text at embedding-similarity boundaries.

    Args:
        text: The full document text.
        source_meta: Provenance copied into every chunk's metadata.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The chunks in document order, each carrying ``source_meta`` plus
        its ``chunk_index``.

    Raises:
        ValueError: If ``verbose`` is invalid.
        openai.APIError: If segment embedding still fails after the
            client's retries.
    """
    settings = get_settings()
    verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
    groups = self._similarity_groups(_segments(text))
    resplitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
        encoding_name="cl100k_base",
        chunk_size=settings.CHUNK_SIZE,  # tokens, like token_based
        chunk_overlap=settings.CHUNK_OVERLAP,
    )
    chunks: list[Document] = []
    for group in groups:
        group_text = group.strip()
        if not group_text:
            continue
        texts = (
            resplitter.split_text(group_text)
            if count_tokens(group_text) > settings.CHUNK_SIZE
            else [group_text]
        )
        chunks.extend(
            Document(page_content=piece, metadata=dict(source_meta)) for piece in texts
        )
    for chunk_index, chunk in enumerate(chunks):
        chunk.metadata["chunk_index"] = chunk_index
    warn_near_token_ceiling(chunks, strategy="semantic")
    v_chunk(chunks, verbose=verbose)
    return chunks

Chunking¶

varagity.chunking.base ¶

ChunkingStrategy ¶

split(text, *, source_meta, verbose=None) ¶

get_chunker(name) ¶

register(name) ¶

warn_near_token_ceiling(chunks, *, strategy) ¶

varagity.chunking.recursive_character ¶

RecursiveCharacterStrategy ¶

split(text, *, source_meta, verbose=None) ¶

varagity.chunking.token_based ¶

TokenBasedStrategy ¶

__init__(*, length_function=None) ¶

split(text, *, source_meta, verbose=None) ¶

varagity.chunking.markdown_aware ¶

MarkdownAwareStrategy ¶

split(text, *, source_meta, verbose=None) ¶

varagity.chunking.docling_hybrid ¶

DoclingHybridStrategy ¶

__init__() ¶

split(text, *, source_meta, verbose=None) ¶

varagity.chunking.semantic ¶

SemanticStrategy ¶

__init__(*, embeddings=None) ¶

split(text, *, source_meta, verbose=None) ¶

`varagity.chunking.base` ¶

`ChunkingStrategy` ¶

`split(text, *, source_meta, verbose=None)` ¶

`get_chunker(name)` ¶

`register(name)` ¶

`warn_near_token_ceiling(chunks, *, strategy)` ¶

`varagity.chunking.recursive_character` ¶

`RecursiveCharacterStrategy` ¶

`split(text, *, source_meta, verbose=None)` ¶

`varagity.chunking.token_based` ¶

`TokenBasedStrategy` ¶

`init(*, length_function=None)` ¶

`split(text, *, source_meta, verbose=None)` ¶

`varagity.chunking.markdown_aware` ¶

`MarkdownAwareStrategy` ¶

`split(text, *, source_meta, verbose=None)` ¶

`varagity.chunking.docling_hybrid` ¶

`DoclingHybridStrategy` ¶

`init()` ¶

`split(text, *, source_meta, verbose=None)` ¶

`varagity.chunking.semantic` ¶

`SemanticStrategy` ¶

`init(*, embeddings=None)` ¶

`split(text, *, source_meta, verbose=None)` ¶