Skip to content

Text Chunking

Split extracted text into overlapping, structure-aware chunks ready to embed and index for RAG. Four strategies support different document types — text splits on whitespace/punctuation, Markdown preserves structure and code blocks, YAML maintains section hierarchy, and semantic chunking uses embeddings to detect topic shifts.

  • Text — splits on whitespace/punctuation boundaries
  • Markdown — structure-aware; preserves headings, lists, and code blocks
  • YAML — section-aware; preserves YAML document structure
  • Semantic — topic-aware; splits at natural document boundaries

Set chunker_type to "semantic". Uses an embedding model for topic detection when one is configured; otherwise falls back to structural heuristics.

config = ExtractionConfig(
chunking=ChunkingConfig(chunker_type="semantic")
)

Behavior:

  • Without embeddings — Uses structural heuristics: detects headers (ALL CAPS, numbered sections) and paragraph boundaries
  • With embeddings — Compares consecutive paragraphs via embeddings to detect topic shifts, merging paragraphs below the topic_threshold (default: 0.75)

Use topic_threshold to control sensitivity: lower values (0.1–0.3) detect more topic boundaries (more, smaller chunks); higher values (0.7–0.9) detect fewer (fewer, larger chunks). Only applies when an embedding model is configured.

When chunker_type is "markdown", set table_chunking to control how tables that exceed the chunk size are split:

  • split (default) — split at row boundaries; continuation chunks do not repeat the header
  • repeat_header — prepend the header row and separator to every continuation chunk so each chunk is self-contained
Python
import asyncio
from xberg import ExtractInput, ExtractionConfig, ChunkingConfig, extract
async def main() -> None:
config: ExtractionConfig = ExtractionConfig(
chunking=ChunkingConfig(
max_chars=1000,
max_overlap=200,
)
)
result = await extract(ExtractInput.from_uri("document.pdf"), config)
print(f"Chunks: {len(result.chunks or [])}")
for chunk in result.chunks or []:
print(f"Length: {len(chunk.content)}")
asyncio.run(main())
Python - Markdown with Heading Context
import asyncio
from xberg import ExtractInput, ExtractionConfig, ChunkingConfig, extract
async def main() -> None:
config: ExtractionConfig = ExtractionConfig(
chunking=ChunkingConfig(
chunker_type="markdown",
max_chars=500,
max_overlap=50,
sizing_type="tokenizer",
sizing_model="Xenova/gpt-4o",
)
)
result = await extract(ExtractInput.from_uri("document.md"), config)
for chunk in result.chunks or []:
heading_context = chunk.metadata.get("heading_context")
if heading_context:
headings = heading_context.get("headings", [])
for h in headings:
print(f"Heading L{h['level']}: {h['text']}")
print(f"Content: {chunk.content[:100]}...")
asyncio.run(main())
Python - Semantic
import asyncio
from xberg import ExtractInput, ExtractionConfig, ChunkingConfig, extract
async def main() -> None:
config: ExtractionConfig = ExtractionConfig(
chunking=ChunkingConfig(chunker_type="semantic")
)
result = await extract(ExtractInput.from_uri("document.pdf"), config)
for chunk in result.chunks or []:
print(f"Content: {chunk.content[:100]}...")
asyncio.run(main())
Python - Prepend Heading Context
import asyncio
from xberg import ExtractInput, ExtractionConfig, ChunkingConfig, extract
async def main() -> None:
config: ExtractionConfig = ExtractionConfig(
chunking=ChunkingConfig(
chunker_type="markdown",
max_chars=500,
max_overlap=50,
prepend_heading_context=True,
)
)
result = await extract(ExtractInput.from_uri("document.md"), config)
for chunk in result.chunks or []:
# Each chunk's content is prefixed with its heading breadcrumb
print(f"Content: {chunk.content[:100]}...")
asyncio.run(main())

Each chunk in result.chunks contains:

Field Description
content Chunk text
metadata.byte_start / byte_end Byte offsets in the original text
metadata.chunk_index / total_chunks Position in sequence
metadata.token_count Token count (when embeddings enabled)
metadata.heading_context Active heading hierarchy (Markdown chunker only)
metadata.heading_path Flattened RAG-shaped heading breadcrumb (e.g., ["Title", "Section", "Subsection"]) for vector database retrieval and context.
metadata.page_spans Per-page coordinates the chunk covers — see Per-Page Chunk Coordinates below.
metadata.node_ids Ids of the covered document.nodes[] entries — see Linking Chunks to Document Nodes below.
metadata.classifications Multi-label classification result — see Chunk Classification below.
embedding Embedding vector (when configured)

Chunks can be sized by token count instead of characters — enable the chunking-tokenizers feature and set sizing to tokenizer.

metadata.page_spans is a list of { page, bbox? } entries, one per page the chunk overlaps, in page order — the first and last entries’ page fields equal metadata.first_page/metadata.last_page. Use it to drive viewer highlighting (draw a box on the source page for a given chunk) without re-deriving page geometry from content offsets.

  • Populated whenever page-boundary provenance is available — the same condition under which first_page/last_page are populated.
  • bbox on each entry is additionally populated when include_document_structure is enabled, as the union of that page’s body-layer node bounding boxes found within the chunk.
  • Empty (and omitted from JSON output) when page-boundary provenance is unavailable.
chunk with page_spans
{
"content": "...",
"metadata": {
"first_page": 2,
"last_page": 3,
"page_spans": [
{ "page": 2, "bbox": { "x0": 72.0, "y0": 120.5, "x1": 540.0, "y1": 640.2 } },
{ "page": 3 }
]
}
}

Cross-page fragment stitching for page_spans mirrors the same-page-only scope of table identity — each page’s span is reported independently.

metadata.node_ids is a list of the id values of the document.nodes[] (DocumentStructure) entries whose text the chunk covers, deduplicated and in document node-traversal order. Join chunk.metadata.node_ids against document.nodes[].id to trace a chunk back to the structural nodes it came from.

  • Populated only when document structure is available (the PDF extraction path) — plain-text and structure-less documents get an empty node_ids.
  • Membership is a best-effort verbatim-text-containment check (a node’s text found within the chunk), gated on a minimum node-text length, so very short nodes may not be linked. It is not a byte-exact mapping.

Classify each chunk against a caller-supplied, description-backed label set — the chunk-level analogue of Page Classification. Set ExtractionConfig.chunk_classification to a ChunkClassificationConfig:

Field Type Default Description
definitions list[{label, description}] Label taxonomy. Unlike page classification’s bare label list, every label carries a description injected into the prompt, so the model can disambiguate similarly named labels in large taxonomies. Must contain at least one entry.
llm LlmConfig LLM configuration used for classification.
batch_size int 10 Number of chunks grouped into a single classification request. Larger batches amortize the fixed cost of the definitions block (repeated verbatim per request) across more chunks, at the risk of exceeding context limits for large taxonomies.
max_concurrency int 4 Maximum number of in-flight batch requests, bounded to avoid rate-limiting the configured LLM provider.
prompt_template str | None None Minijinja template. Receives {{ definitions }} (rendered label + description list) and {{ chunks }} (numbered chunk texts in the batch). None uses the built-in default.

Unlike page classification, chunk classification is always multi-label — a chunk may match zero, one, or many definitions. Leaving chunk_classification unset (the default) disables the post-processor entirely and makes no LLM calls.

chunk_classification.py
from xberg import (
ChunkClassificationConfig,
ChunkClassificationDefinition,
ChunkingConfig,
ExtractInput,
ExtractionConfig,
LlmConfig,
extract,
)
config = ExtractionConfig(
chunking=ChunkingConfig(chunker_type="markdown"),
chunk_classification=ChunkClassificationConfig(
definitions=[
ChunkClassificationDefinition(
label="pricing",
description="Discusses cost, fees, or subscription tiers.",
),
ChunkClassificationDefinition(
label="liability",
description="Discusses indemnification, warranties, or limitation of liability.",
),
],
llm=LlmConfig(model="openai/gpt-4o-mini"),
batch_size=10,
max_concurrency=4,
),
)
output = await extract(ExtractInput(kind="uri", uri="contract.pdf"), config=config)
result = output.results[0]
for chunk in result.chunks or []:
for label in chunk.metadata.classifications:
print(label.label, label.confidence)

Token Sizing with Your Own Tokenizer (Plugin Variant)

Section titled “Token Sizing with Your Own Tokenizer (Plugin Variant)”

Token budgets only protect the embedder when they are counted with the tokenizer the embedder actually uses. When that tokenizer isn’t available as a HuggingFace tokenizer.json (llama.cpp/GGUF vocabularies, SentencePiece models, custom vocabs), plug it in — Xberg calls back into the registered backend to count tokens instead of loading one from the Hub. The chunking-tokenizers feature is still required (it gates the tokenizer sizing variant itself); language bindings ship with it enabled.

  1. Register the backend once at startup via xberg::plugins::register_tokenizer_backend(Arc::new(MyTokenizer)). The backend implements TokenizerBackend (a Plugin-inheriting trait with a synchronous count_tokens(text) -> usize — it runs inside the splitter’s boundary search, so keep it cheap).
  2. Reference it by name in the chunking config: { "sizing": { "type": "tokenizer", "model": "my-tokenizer" } }. A registered name takes precedence over a HuggingFace id; unregistered names fall back to the Hub as before.
  3. max_characters is then the chunk budget in that backend’s tokens.

Language bindings register through the same API — implement the trait’s methods (name, initialize, shutdown, count_tokens) on a host-language object and pass it to register_tokenizer_backend.

Python
import asyncio
from xberg import (
ExtractInput,
extract,
ExtractionConfig,
ChunkingConfig,
EmbeddingConfig,
EmbeddingModelType,
)
async def main() -> None:
config: ExtractionConfig = ExtractionConfig(
chunking=ChunkingConfig(
max_chars=500,
max_overlap=50,
embedding=EmbeddingConfig(
model=EmbeddingModelType.preset("balanced"),
normalize=True,
batch_size=16
)
)
)
result = await extract(ExtractInput.from_uri("research_paper.pdf"), config)
chunks_with_embeddings: list = []
for chunk in result.chunks or []:
if chunk.embedding:
chunks_with_embeddings.append({
"content": chunk.content[:100],
"embedding_dims": len(chunk.embedding)
})
print(f"Chunks with embeddings: {len(chunks_with_embeddings)}")
asyncio.run(main())