Prepare documents for RAG: convert to Markdown, then chunk by heading

Convert every source file to Markdown first, then chunk the Markdown at headings. One format means one chunker for PDF, Word, PowerPoint, Excel and HTML.

Word, PowerPoint, Excel and HTML keep their headings in a direct call. PDFs get headings only from the queue, so the chunker below falls back to splitting by size.

Pipeline

import pathlib

import httpx

TYPES = {
    ".pdf": "application/pdf",
    ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
    ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    ".html": "text/html",
    ".txt": "text/plain",
}
MAX_CHARS = 4000


def to_markdown(path: pathlib.Path) -> str:
    resp = httpx.post(
        "https://api.flatmark.dev/v1/convert",
        files={"file": (path.name, path.read_bytes(), TYPES[path.suffix.lower()])},
        headers={"X-API-Key": "YOUR_KEY"},
        timeout=120,
    )
    resp.raise_for_status()
    return resp.json()["markdown"]


def chunk(markdown: str) -> list[str]:
    sections, current = [], []
    for line in markdown.splitlines():
        if line.startswith("#") and current:
            sections.append("\n".join(current))
            current = []
        current.append(line)
    sections.append("\n".join(current))
    chunks = []
    for section in sections:
        while len(section) > MAX_CHARS:
            cut = section.rfind("\n\n", 0, MAX_CHARS)
            cut = cut if cut > 0 else MAX_CHARS
            chunks.append(section[:cut])
            section = section[cut:]
        chunks.append(section)
    return [c.strip() for c in chunks if c.strip()]


for doc in sorted(pathlib.Path("corpus").iterdir()):
    if doc.suffix.lower() not in TYPES:
        continue
    markdown = to_markdown(doc)
    if not markdown.strip():
        print(doc.name, "has no text layer, send it to the queue for OCR")
        continue
    for piece in chunk(markdown):
        pass  # embed and index each piece here

Large corpora

For files over 8 MB, scans, or PDFs that need headings, submit each file to the queue with a webhook_url. Index a document when its webhook arrives.

Next step

Sign in with a magic link and create the API key this pipeline sends.

FAQ

Should I chunk by heading or by token count?

By heading first, then split long sections by size. A fixed window alone can cut a table in half.

What if a PDF has no text layer?

The direct call runs no OCR, so a scanned PDF returns little or no text. Send it to the queue, which runs OCR.