SELFTEST Production Document Ingestion and Structure-Aware Chunking

Build a deterministic ingestion pipeline that preserves document structure and produces retrieval-ready chunks with traceable metadata.

Engineers building document-QA systems who already understand Python, embeddings, and basic vector search.

What you'll be able to do

• Normalize PDF, Markdown, HTML, and plain-text inputs into a common document representation.
• Detect headings and preserve section context during chunking.
• Emit deterministic JSONL records with source, page, section, offsets, and chunk identifiers.
• Validate chunk size, metadata completeness, and deterministic output before indexing.

The problem

A PDF, Markdown file, or HTML page can be split into arbitrary character windows that separate headings from content, duplicate context across chunks, and make citations impossible to trace back to the source.

How it works

Central question

How should an ingestion pipeline transform heterogeneous documents into chunks that retrieval can use without destroying the document structure needed for ranking, citations, and access control?

The answer is to treat chunking as a data-contract problem, not as a string-splitting operation. The pipeline should produce records with stable identifiers, source provenance, section context, normalized text, and explicit size estimates. Later retrieval and generation stages should consume this contract rather than reimplementing parsing assumptions.

The ingestion contract

Each emitted chunk should contain at least:

{
  "chunk_id": "sha256-prefix:ordinal",
  "document_id": "sha256-prefix",
  "source": "relative/path.pdf",
  "source_type": "pdf",
  "page": 3,
  "section": "Installation > Configuration",
  "text": "...",
  "char_start": 1204,
  "char_end": 1890,
  "token_estimate": 172,
  "metadata": {}
}

The important properties are:

• document_id identifies the source content, not its filename alone.
• chunk_id is stable when the source and chunking configuration are unchanged.
• source and page support citations and debugging.
• section preserves hierarchy that a flat embedding cannot recover reliably.
• char_start and char_end make it possible to inspect exactly what was indexed.
• metadata is the place to carry tenant, ACL, retention, and classification attributes. It must be populated before indexing, not added after retrieval.

Parse first, chunk second

Parsing and chunking should be separate stages:

1. Discover supported files.
2. Extract text and structural hints.
3. Normalize whitespace without removing meaningful boundaries.
4. Detect headings and maintain a section stack.
5. Group paragraphs under the current section.
6. Split only oversized groups.
7. Emit records and validate the contract.

This separation makes failures diagnosable. An empty PDF extraction is a parser failure. A chunk that exceeds the configured limit is a chunker failure. A retrieved record without tenant metadata is a metadata-contract failure.

Structure-aware boundaries

Prefer boundaries in this order:

1. Document or page boundary.
2. Heading boundary.
3. Paragraph boundary.
4. Sentence boundary.
5. Character boundary as a final fallback.

A heading should be copied into the chunk's section metadata and, where useful, into the chunk text. Keeping it in both places helps lexical retrieval match terms such as Retries or Authentication while allowing downstream systems to display the hierarchy.

The example implementation uses a conservative paragraph-oriented splitter. It does not attempt to infer tables, code blocks, or semantic boundaries from arbitrary PDFs. Those need format-specific handlers and validation fixtures.

Size is a budget, not a universal constant

A character limit is only a proxy for model tokens. Tokenization varies by language, markup, code, and model vocabulary. Use a conservative character limit during ingestion, then measure the actual tokenizer distribution used by the embedding and generation models.

Record at least p50, p95, and maximum token counts per source type. If p95 is acceptable but the maximum is not, investigate outliers instead of raising the global limit. Large chunks increase embedding input cost and can reduce retrieval precision by combining unrelated facts.

Overlap is selective context duplication

Overlap can help when a fact crosses a paragraph boundary, but it also duplicates vectors, increases index size, and can cause answer generation to see repeated evidence. Start with zero overlap at clean heading and paragraph boundaries. Add a measured overlap only for formats where boundary splitting causes a recall regression.

Determinism and reindexing

Hash normalized source content and the chunking configuration. A changed parser version, normalization rule, or maximum size should produce a new ingestion version. Do not silently overwrite records while retaining old vectors; otherwise a query can retrieve chunks from incompatible generations of the same document.

A production record should therefore also carry fields such as ingestion_version, parser_version, and acl_version. The sample keeps the implementation small but leaves a metadata object for those fields.

flowchart LR
    A[Source files] --> B[Discovery]
    B --> C[Format parser]
    C --> D[Normalized blocks]
    D --> E[Heading and section tracking]
    E --> F[Paragraph-aware chunker]
    F --> G[Schema validation]
    G --> H[JSONL artifact]
    H --> I[Embedding and indexing]
    G --> J[Rejects and diagnostics]

Parsing, structural chunking, validation, and indexing are separate stages so each failure has a distinct diagnostic location.

Implementation

Install

The example uses PyMuPDF for PDFs and Beautiful Soup for HTML. Markdown and text use the standard library.

python -m venv .venv
. .venv/bin/activate
pip install pymupdf==1.25.5 beautifulsoup4==4.13.3

Set the input and output locations without embedding credentials:

export RAG_INPUT_DIR=./documents
export RAG_OUTPUT_FILE=./artifacts/chunks.jsonl

The program walks the input directory, extracts blocks, tracks Markdown and HTML headings, groups paragraphs, splits oversized paragraphs by characters, and writes one JSON object per line. It also computes document and chunk identifiers from content so the output is reproducible.

The output JSONL is intentionally independent of a vector database. An embedding/indexing stage can read it later and attach vectors without changing provenance fields.

Operational extensions

For production, add parser-specific fixtures for:

• PDFs containing tables, headers, footers, scanned pages, and multi-column layouts.
• Markdown code blocks, lists, and nested headings.
• HTML navigation, boilerplate, hidden content, and repeated templates.
• Documents with explicit tenant and access-control metadata.

Reject or quarantine documents when extraction yields no text, when the source type is unsupported, or when required security metadata is absent.

Code 1: Ingest PDF, HTML, Markdown, and text files; preserve section metadata; emit deterministic JSONL chunks.

# Install:
# python -m venv .venv
# . .venv/bin/activate
# pip install pymupdf==1.25.5 beautifulsoup4==4.13.3

from __future__ import annotations

import hashlib
import json
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable

import fitz  # PyMuPDF
from bs4 import BeautifulSoup


MAX_CHARS = 2200
HEADING_RE = re.compile(r"^(#{1,6})\\s+(.+?)\\s*$")


@dataclass
class Block:
    text: str
    page: int | None = None
    heading_level: int | None = None
    is_heading: bool = False


def normalize(text: str) -> str:
    text = text.replace("\\r\\n", "\\n").replace("\\r", "\\n")
    text = re.sub(r"[ \\t]+", " ", text)
    text = re.sub(r"\\n{3,}", "\\n\\n", text)
    return text.strip()


def read_text_blocks(path: Path) -> list[Block]:
    raw = path.read_text(encoding="utf-8", errors="replace")
    suffix = path.suffix.lower()
    if suffix == ".html" or suffix == ".htm":
        soup = BeautifulSoup(raw, "html.parser")
        blocks: list[Block] = []
        for node in soup.find_all(["h1", "h2", "h3", "h4", "h5", "h6", "p", "li"]):
            text = normalize(node.get_text(" ", strip=True))
            if not text:
                continue
            if node.name.startswith("h"):
                blocks.append(Block(text, heading_level=int(node.name[1]), is_heading=True))
            else:
                blocks.append(Block(text))
        return blocks

    blocks = []
    for part in re.split(r"\\n\\s*\\n", raw):
        text = normalize(part)
        if not text:
            continue
        match = HEADING_RE.match(text)
        if match and "\\n" not in text:
            blocks.append(Block(match.group(2), heading_level=len(match.group(1)), is_heading=True))
        else:
            blocks.append(Block(text))
    return blocks


def read_pdf_blocks(path: Path) -> list[Block]:
    blocks: list[Block] = []
    with fitz.open(path) as document:
        for page_number, page in enumerate(document, start=1):
            text = normalize(page.get_text("text"))
            if not text:
                continue
            for paragraph in re.split(r"\\n\\s*\\n", text):
                paragraph = normalize(paragraph)
                if paragraph:
                    blocks.append(Block(paragraph, page=page_number))
    return blocks


def source_blocks(path: Path) -> list[Block]:
    if path.suffix.lower() == ".pdf":
        return read_pdf_blocks(path)
    if path.suffix.lower() in {".txt", ".md", ".markdown", ".html", ".htm"}:
        return read_text_blocks(path)
    return []


def split_oversized(text: str, limit: int) -> Iterable[str]:
    if len(text) <= limit:
        yield text
        return
    start = 0
    while start < len(text):
        end = min(start + limit, len(text))
        if end < len(text):
            cut = text.rfind(" ", start, end)
            if cut > start + limit // 2:
                end = cut
        piece = text[start:end].strip()
        if piece:
            yield piece
        start = end


def digest(value: str, length: int = 16) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()[:length]


def make_chunks(path: Path, root: Path, limit: int = MAX_CHARS) -> list[dict]:
    blocks = source_blocks(path)
    if not blocks:
        raise ValueError(f"no extractable text: {path}")

    source = path.relative_to(root).as_posix()
    document_id = digest("\\n".join(block.text for block in blocks))
    section_stack: list[tuple[int, str]] = []
    output: list[dict] = []
    pending: list[Block] = []

    def flush() -> None:
        if not pending:
            return
        text = "\\n\\n".join(block.text for block in pending).strip()
        if not text:
            pending.clear()
            return
        page_values = [block.page for block in pending if block.page is not None]
        page = page_values[0] if page_values else None
        section = " > ".join(item[1] for item in section_stack)
        ordinal = len(output)
        chunk_key = f"{document_id}|{ordinal}|{section}|{text}"
        output.append({
            "chunk_id": f"{document_id}:{digest(chunk_key, 12)}",
            "document_id": document_id,
            "source": source,
            "source_type": path.suffix.lower().lstrip("."),
            "page": page,
            "section": section,
            "text": text,
            "char_start": None,
            "char_end": None,
            "token_estimate": max(1, round(len(text) / 4)),
            "metadata": {"ingestion_version": "1"},
        })
        pending.clear()

    for block in blocks:
        if block.is_heading:
            flush()
            level = block.heading_level or 1
            section_stack = [(n, title) for n, title in section_stack if n < level]
            section_stack.append((level, block.text))
            continue
        for piece in split_oversized(block.text, limit):
            candidate = "\\n\\n".join(item.text for item in pending + [Block(piece, block.page)])
            if pending and len(candidate) > limit:
                flush()
            pending.append(Block(piece, block.page))
    flush()
    return output


def main() -> None:
    root = Path(os.environ.get("RAG_INPUT_DIR", "./documents")).resolve()
    output = Path(os.environ.get("RAG_OUTPUT_FILE", "./artifacts/chunks.jsonl"))
    output.parent.mkdir(parents=True, exist_ok=True)
    supported = {".pdf", ".txt", ".md", ".markdown", ".html", ".htm"}
    files = sorted(path for path in root.rglob("*") if path.is_file() and path.suffix.lower() in supported)
    if not files:
        raise SystemExit(f"no supported files found below {root}")

    count = 0
    with output.open("w", encoding="utf-8") as handle:
        for path in files:
            for record in make_chunks(path, root):
                if not record["text"] or not record["document_id"] or not record["chunk_id"]:
                    raise ValueError(f"invalid record from {path}")
                handle.write(json.dumps(record, ensure_ascii=False) + "\\n")
                count += 1
    print(f"wrote {count} chunks to {output}")


if __name__ == "__main__":
    main()

Verifying it

Create at least one Markdown file with # Authentication, two paragraphs, and ## Token expiry, then run python ingest.py. Expected output is a line such as wrote N chunks to ..., where N is greater than zero. Inspect the artifact with python -c "import json, pathlib; p=pathlib.Path('artifacts/chunks.jsonl'); rows=[json.loads(x) for x in p.read_text().splitlines()]; assert rows; assert all(r['chunk_id'] and r['document_id'] and r['text'] for r in rows); print(rows[0].keys())". Add a deterministic-output check by running the command twice and comparing normalized JSONL files with sha256sum; identical source files and configuration should produce identical output. For real documents, separately measure empty extraction rate, chunk character and tokenizer-length distributions, chunks per document, and duplicate-content rate.

Performance and cost

Ingestion cost is primarily disk I/O and parser CPU; PDF and HTML parsing do not require model tokens or API dollars in this implementation. Chunking increases downstream cost approximately in proportion to the number of emitted chunks: each chunk generally requires one embedding request or local embedding inference. Measure total characters, actual tokenizer tokens, embedding batch size, indexing latency, and storage bytes. Increasing overlap or reducing the chunk limit increases vector count and embedding cost. The token_estimate field is only a screening estimate; use the exact embedding-model tokenizer before budgeting.

What breaks

Scanned PDFs may produce empty text because OCR is not included; detect this through the no-extractable-text rejection and route those files to an OCR stage. Multi-column PDFs may interleave reading order; detect it with page-level fixtures and visual review. Headers and footers may repeat in every page; detect repeated n-grams across pages and remove them with a format-specific rule. Large tables may be flattened into unusable prose; detect low row or column fidelity and use table extraction. Heading detection can fail when a PDF encodes headings as ordinary text; compare parser output with expected section fixtures. Oversized chunks indicate a splitter or normalization bug; monitor maximum actual tokenizer length rather than only character count.

Security

Treat every source file and extracted string as untrusted input. Do not execute embedded macros, HTML, or code. Store tenant, document, and ACL metadata with every chunk and enforce authorization before retrieval results reach the model; filtering only after generation is too late. Avoid logging full document text by default. Restrict input and output directories, validate file types using parser behavior rather than filename alone, cap file size and page count, and quarantine decompression or parser failures. If documents contain personal or regulated data, define retention and deletion propagation for source files, chunks, embeddings, caches, and evaluation traces. Keep document_id opaque and avoid placing sensitive content in identifiers.

In production

Replace the local directory walk with an object-store or queue-driven ingestion service. Make ingestion idempotent using a source version or content hash, and maintain states such as discovered, parsed, chunked, indexed, failed, and quarantined. Pin parser versions and retain parser and chunker configuration with each batch. Add golden fixtures for each document family and compare extraction, section paths, chunk counts, and token distributions in CI. Carry ACL and tenant metadata into the index, support deletion by document version, and expose metrics for throughput, failure rate, empty extraction, chunk-size percentiles, duplicate rate, and reprocessing lag. Use bounded concurrency and backpressure so a large upload cannot starve query traffic. Keep the JSONL artifact or an equivalent manifest for replay and audit.

Limitations

The sample does not perform OCR, semantic table extraction, language-specific tokenization, sentence-boundary detection, or embedding. Character limits are not token limits. PDF reading order and heading recovery are inherently parser-dependent. The implementation sets page metadata but leaves character offsets null because offsets from heterogeneous parsers are not comparable without a source-specific coordinate model. It also treats HTML list items as paragraphs and does not remove boilerplate. These trade-offs make the example deterministic and inspectable, not universally correct.

Conclusion

1. Define a chunk schema with provenance, section context, stable IDs, and security metadata before choosing a vector store.
2. Parse structure before splitting; prefer heading and paragraph boundaries and use character splitting only as a fallback.
3. Validate extraction and chunk distributions with representative fixtures, including malformed and scanned documents.
4. Version parser and chunking configuration so reindexing is reproducible and deletions can follow a document version.

Further reading

• Partitioning
• Text - PyMuPDF documentation
• Beautiful Soup Documentation
• JSON Lines

In this series

• RAG architecture and the five places it fails
• Building a document ingestion and chunking pipeline
• Hybrid retrieval and reranking
• Evaluating retrieval quality and answer faithfulness separately
• Deploying RAG with monitoring and access control
• Benchmarking RAG architectures under realistic workloads
• D01/01 — RAG architecture and the five places it fails
• D01/03 — Hybrid retrieval and reranking
• Production-Grade Retrieval-Augmented Generation: From Fundamentals to Operations

Try it yourself

Add a fixture-based quality gate for one PDF, one HTML page, and one Markdown document. Assert that expected headings appear in section metadata, no emitted chunk exceeds the configured character limit, every chunk has tenant and ACL metadata, and rerunning ingestion produces byte-identical JSONL. Then add an OCR adapter for scanned PDFs and compare empty-extraction rate before and after it.

---

Originally published in the LLM engineering blog series.