MD ConverterUsing MarkItDown in a RAG Pipeline: The Complete Ingestion Guide (2026)
Retrieval-augmented generation lives or dies on the quality of what you feed it. Garbage in, hallucinations out. markitdown — Microsoft's Python tool that converts PDF, DOCX, PPTX, images, and audio into clean Markdown — has quietly become one of the most popular ingestion layers for teams building RAG on internal documents.
This guide is a working blueprint. You'll see how to wire markitdown into a real ingestion pipeline (source files → Markdown → chunks → embeddings → vector database), the gotchas that bite people in production, and a complete FastAPI + pgvector reference implementation you can copy.
If you're new to markitdown, start with our beginner's tutorial. If you're deciding between markitdown and Pandoc for a different use case, see the comparison.
Why markitdown for RAG?
Three specific properties matter for RAG:
- Structure preservation. Headings become
#, tables become GFM tables, lists become-/*. Chunking algorithms can then split on semantic boundaries instead of arbitrary character counts. - Format coverage. One tool handles PDF, DOCX, PPTX, XLSX, HTML, images (OCR), audio (STT), and ZIP archives. Fewer moving parts, fewer failure modes.
- Token efficiency. markitdown's output is denser than typical PDF text extraction — less filler, more signal. Lower embedding cost, tighter retrieval hits.
The tradeoff: markitdown doesn't try to reconstruct exact visual layout. If your use case is "give the LLM a faithful copy of a legal form with checkbox positions preserved," markitdown is the wrong tool — you want Azure Document Intelligence or Unstructured directly. For everything else (knowledge bases, tech docs, meeting notes, research papers, wiki dumps), markitdown is a strong default.
The Ingestion Pipeline in Five Stages
[Source Files] → [markitdown] → [Chunker] → [Embedder] → [Vector DB]
PDF/DOCX text_content Markdown vectors + pgvector /
PPTX/HTML chunks metadata Pinecone /
images/audio Weaviate
We'll build each stage as a composable function so you can swap out any piece later (different embedder, different DB) without rewriting the pipeline.
Stage 1: Extract to Markdown
from pathlib import Path
from markitdown import MarkItDown
md = MarkItDown()
def extract(path: Path) -> str:
"""Convert any supported file to Markdown."""
result = md.convert(str(path))
return result.text_content
Production tip: wrap this in a retry with a timeout. convert() can hang on malformed PDFs.
import signal
from contextlib import contextmanager
@contextmanager
def timeout(seconds):
def _handler(signum, frame):
raise TimeoutError()
signal.signal(signal.SIGALRM, _handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
def extract_safe(path: Path, seconds: int = 60) -> str | None:
try:
with timeout(seconds):
return md.convert(str(path)).text_content
except (TimeoutError, Exception) as e:
print(f"skip {path.name}: {e}")
return None
(Note: signal.alarm is Unix-only. On Windows, use concurrent.futures with a per-call ProcessPoolExecutor.)
Stage 2: Chunk on Heading Boundaries
The single most important RAG decision is how you chunk. Fixed-size chunks (say, 800 chars) will slice through the middle of a table or bullet list, and retrieval quality collapses. Heading-aware chunking respects semantic boundaries:
import re
HEADING_RE = re.compile(r"^(#{1,6})\s+(.+?)$", flags=re.MULTILINE)
def chunk_by_headings(markdown: str, max_chars: int = 2000) -> list[dict]:
"""
Split markdown into chunks at heading boundaries. Each chunk carries
the heading path (e.g., ['Chapter 3', 'Section 3.1']) as metadata.
"""
sections = re.split(r"(?=^#+\s)", markdown, flags=re.MULTILINE)
chunks = []
heading_stack: list[tuple[int, str]] = []
for section in sections:
m = HEADING_RE.match(section)
if m:
level = len(m.group(1))
title = m.group(2).strip()
# Pop deeper-or-equal levels off the stack
heading_stack = [(lvl, t) for lvl, t in heading_stack if lvl < level]
heading_stack.append((level, title))
path = [t for _, t in heading_stack]
# Split oversized sections at paragraph boundaries
for piece in _split_by_paragraphs(section, max_chars):
chunks.append({"text": piece, "heading_path": path})
return chunks
def _split_by_paragraphs(text: str, max_chars: int) -> list[str]:
if len(text) <= max_chars:
return [text]
parts, current = [], ""
for para in text.split("\n\n"):
if len(current) + len(para) + 2 > max_chars and current:
parts.append(current)
current = para
else:
current = f"{current}\n\n{para}" if current else para
if current:
parts.append(current)
return parts
Why the heading path matters: at retrieval time, you can surface "Chapter 3 > Section 3.1 > Payment terms" alongside the chunk text. The LLM answers get noticeably more accurate when it knows where a passage came from.
Stage 3: Embed
Any embedding model works. OpenAI's text-embedding-3-small is a solid default (cheap, fast, 1536 dims); text-embedding-3-large gets you higher retrieval accuracy at ~7× the cost.
from openai import OpenAI
client = OpenAI()
def embed(texts: list[str]) -> list[list[float]]:
"""Batch-embed texts. Returns one 1536-dim vector per input."""
response = client.embeddings.create(
model="text-embedding-3-small",
input=texts,
)
return [d.embedding for d in response.data]
Batch calls (up to 2048 inputs per request). One-at-a-time embedding is 20–50× slower and burns through your rate limit.
Stage 4: Store with Metadata
Metadata is where cheap RAG becomes good RAG. Store at minimum:
source_path— the original filenamesource_type— pdf, docx, mp3, etc.heading_path— the semantic breadcrumb from Stage 2chunk_index— position in the doc, for context expansion at retrieval timeingested_at— timestamp, useful for cache invalidation
import psycopg
from psycopg.rows import dict_row
DB = psycopg.connect("postgresql://localhost/ragdb", row_factory=dict_row)
def store(chunks: list[dict], vectors: list[list[float]], source: Path):
with DB.cursor() as cur:
for i, (chunk, vec) in enumerate(zip(chunks, vectors)):
cur.execute("""
INSERT INTO documents
(source_path, source_type, heading_path, chunk_index, content, embedding)
VALUES (%s, %s, %s, %s, %s, %s)
""", (
str(source),
source.suffix.lstrip("."),
chunk["heading_path"],
i,
chunk["text"],
vec,
))
DB.commit()
Schema:
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE documents (
id BIGSERIAL PRIMARY KEY,
source_path TEXT NOT NULL,
source_type TEXT NOT NULL,
heading_path TEXT[],
chunk_index INT,
content TEXT NOT NULL,
embedding vector(1536) NOT NULL,
ingested_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);
CREATE INDEX ON documents (source_path);
Stage 5: Retrieve
Given a user query, embed it, run a nearest-neighbor search, and return the top-k chunks with their metadata:
def retrieve(query: str, k: int = 5) -> list[dict]:
query_vec = embed([query])[0]
with DB.cursor() as cur:
cur.execute("""
SELECT
content,
heading_path,
source_path,
1 - (embedding <=> %s::vector) AS similarity
FROM documents
ORDER BY embedding <=> %s::vector
LIMIT %s
""", (query_vec, query_vec, k))
return cur.fetchall()
Handing the results to the LLM:
def answer(question: str) -> str:
hits = retrieve(question, k=5)
context = "\n\n---\n\n".join(
f"Source: {h['source_path']}\nSection: {' > '.join(h['heading_path'])}\n\n{h['content']}"
for h in hits
)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "Answer only from the provided sources. Cite section names."},
{"role": "user", "content": f"Sources:\n\n{context}\n\nQuestion: {question}"},
],
)
return response.choices[0].message.content
Full FastAPI Reference Implementation
Bringing it all together — an ingestion API you can POST files to:
from fastapi import FastAPI, UploadFile
from pathlib import Path
import tempfile
app = FastAPI()
@app.post("/ingest")
async def ingest(file: UploadFile):
suffix = Path(file.filename).suffix
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
tmp.write(await file.read())
tmp_path = Path(tmp.name)
markdown = extract_safe(tmp_path)
if markdown is None:
return {"status": "skipped", "reason": "extraction failed"}
chunks = chunk_by_headings(markdown)
vectors = embed([c["text"] for c in chunks])
store(chunks, vectors, Path(file.filename))
return {"status": "ok", "chunks": len(chunks)}
@app.get("/ask")
def ask(q: str):
return {"answer": answer(q)}
Run it:
uvicorn main:app --reload
curl -F "file=@handbook.pdf" http://localhost:8000/ingest
curl "http://localhost:8000/ask?q=What+is+the+refund+policy?"
Production Gotchas
1. markitdown is not always deterministic
The Word (.docx) parser occasionally re-orders footnotes or produces slightly different whitespace between runs. If you hash Markdown output to detect "already ingested" files, hash the source file bytes, not the Markdown.
2. Large PDFs blow up memory
markitdown loads the full document. For a 2000-page PDF, this can spike a worker's RAM. Options:
- Split the PDF upstream with
pdftkorqpdfinto 100-page chunks. - Run ingestion in a subprocess so the memory is released on completion.
- Move to Azure Document Intelligence for anything > 500 pages.
3. Audio transcription is slow
.mp3 and .wav are handled via Whisper. A 30-minute podcast can take several minutes on CPU. For production audio pipelines, transcribe out-of-band with a hosted API (Deepgram, AssemblyAI, or Whisper API) and feed the text to a separate ingestion path — don't try to do it inside your synchronous FastAPI handler.
4. Tables in complex PDFs still lose alignment
markitdown's default PDF parser handles most tables well, but multi-page tables or tables with merged cells often come out ragged. If tables are load-bearing for your use case (financial reports, medical records), enable Azure Document Intelligence:
md = MarkItDown(docintel_endpoint="https://YOUR-RESOURCE.cognitiveservices.azure.com/")
The cost is real (~$1.50 per 1000 pages at time of writing) but table quality improves dramatically.
5. HTML-to-Markdown noise
When you feed HTML pages, expect residual whitespace and occasional stray <span> markers. Post-process:
import re
def clean(md: str) -> str:
md = re.sub(r"<span[^>]*>|</span>", "", md)
md = re.sub(r"\n{3,}", "\n\n", md)
return md.strip()
6. Duplicate detection
Users re-upload the same handbook every quarter. Without dedup, your DB bloats and retrieval quality degrades from redundant hits.
import hashlib
def file_hash(path: Path) -> str:
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(65536), b""):
h.update(chunk)
return h.hexdigest()
# Skip if hash already exists in a `source_hashes` table
When markitdown Is Not Enough
For the following, consider alternatives or hybrid pipelines:
- OCR-heavy scanned PDFs — pair markitdown with
ocrmypdffirst, or use Azure Document Intelligence. - Preserving exact visual layout — use Unstructured's
hi_resstrategy or Docling. - Real-time streaming ingestion — markitdown is batch-oriented. For streams (Slack, email), buffer and batch.
- Very structured data (invoices, forms) — use a dedicated document AI (Textract, Azure Form Recognizer) that returns typed fields, not Markdown.
For the reverse direction — turning your Markdown answers back into shareable PDFs or Word docs — pair this pipeline with MD2PDF Online (browser) or Pandoc (scripted). The full picture:
Source docs → markitdown → Markdown → LLM
↓
Answer Markdown → Pandoc / MD2PDF → PDF / DOCX
Frequently Asked Questions
How does markitdown compare to Unstructured for RAG?
Unstructured has more sophisticated layout parsing (better for research papers with figures, captions, footnotes) but is heavier to install and slower on large batches. markitdown is faster and covers more input formats (audio, YouTube) but produces less layout-aware output. For most knowledge-base use cases, markitdown is enough; for scientific literature ingestion, prefer Unstructured or Docling.
Should I chunk by tokens or by characters?
Characters are fine for most models. Tokens matter if you're aiming for a specific embedding context (e.g., exactly 512 tokens). Use tiktoken:
import tiktoken
enc = tiktoken.encoding_for_model("text-embedding-3-small")
token_count = len(enc.encode(text))
What chunk size works best?
Start with 1500–2500 characters (~350–600 tokens). Smaller = more precise but loses context; larger = more context but noisier retrieval. Run an eval set with a few sizes; pick what maximizes answer accuracy on your own questions, not on generic benchmarks.
Can I re-ingest incrementally when a source doc changes?
Yes. Store source_hash alongside each chunk. When a file is re-uploaded: compare hashes; if different, delete all chunks for that source_path and re-ingest. Adds ~10 lines of code.
Does markitdown work with Claude / Gemini / local LLMs?
Yes — markitdown output is just text. The examples above use OpenAI's SDK, but swap in anthropic, google-generativeai, or an Ollama client and everything else stays the same.
Wrapping Up
markitdown covers the messy first mile of RAG — turning heterogeneous binary formats into clean, chunk-friendly Markdown — with one dependency and one line of Python. That's why it's showing up in more and more production pipelines: the alternative is stitching together five different parsers per format.
The pipeline above is roughly what a team-sized internal-docs RAG looks like: one API endpoint for ingestion, one for querying, a Postgres+pgvector store, a few hundred lines of code total. Scale from there as needed.
Related reading: