Complex Document RAG
RAG pipelines for documents that mix text, tables, images, and layout structure.
When to Activate
- Building a RAG pipeline that ingests PDFs, Excel, CSV, or image files
- Debugging silent data loss from document extraction (empty chunks, missing tables, dropped figures)
- Designing chunking for documents with tables, figures, or hierarchical headings
- Choosing between OCR libraries or managed document intelligence services
- Optimizing cost when processing large batches of mixed-format documents
- Handling edge cases: scanned PDFs, merged cells, embedded charts, photographed tables
- Deciding how to index multimodal content (charts, diagrams, infographics)
Extraction Tool Decision Matrix
| Document Type | First-Choice Tool | Fallback / Managed | |
|---|
| PDF — native text | pdfplumber or PyMuPDF | — | |
| PDF — tables | pdfplumber.extract_tables() or camelot-py | Azure Document Intelligence prebuilt-layout | |
| PDF — scanned pages | pdf2image + pytesseract | AWS Textract, Azure Document Intelligence Read, Google Document AI | |
| PDF — layout-aware (multi-column) | unstructured.io or Surya | AWS Textract AnalyzeDocument | |
| PDF — embedded images | PyMuPDF page.get_images() → vision model | — | |
| Excel — structure-aware | openpyxl | — | |
| Excel — formulas as values | openpyxl(data_only=True) or xlrd | — | |
| CSV — dialect/encoding | csv.Sniffer + chardet | — | |
| Image — printed text | pytesseract (≥150 DPI) or PaddleOCR | Google Vision API, Azure AI Vision Read | |
| Image — handwriting | — | Azure AI Vision Read, Google Vision API | |
| Image — tables | TableTransformer (HuggingFace) | AWS Textract AnalyzeDocument TABLES | |
| Image — visual/diagrams | — | GPT-4o vision, Claude Sonnet vision, Gemini 1.5 Pro | |
| Math formulas | pix2tex | Mathpix | |
| Multilingual | PaddleOCR or EasyOCR | Google Vision, Azure AI Vision | |
Tiered Processing Strategy
Never call a vision model on content that can be extracted structurally. Escalate only when lower tiers fail.
Tier 1 — free, instant
├─ Native PDF: pdfplumber/PyMuPDF text extraction
├─ Excel: openpyxl data_only read
└─ CSV: pandas read with dialect + encoding detection
Tier 2 — cheap, moderate latency
├─ Scanned pages: pdf2image + pytesseract / PaddleOCR
└─ Table structure in images: TableTransformer
Tier 3 — expensive, use sparingly
└─ Vision model call: only when Tier 1+2 produce
< confidence_threshold OR content is purely visual
python
def extraction_tier(page_text: str, ocr_confidence: float) -> str:
if len(page_text.strip()) > 100: # Tier 1 succeeded
return "native_text"
if ocr_confidence >= 0.70: # Tier 2 acceptable
return "ocr"
return "vision_model" # Tier 3 required
Cache by content hash — never re-call a vision model for a page you've already processed.
python
import hashlib, json
from pathlib import Path
def vision_with_cache(image_bytes: bytes, prompt: str, cache_dir: Path) -> str:
key = hashlib.sha256(image_bytes + prompt.encode()).hexdigest()
cache_file = cache_dir / f"{key}.json"
if cache_file.exists():
return json.loads(cache_file.read_text())["description"]
description = call_vision_model(image_bytes, prompt)
cache_file.write_text(json.dumps({"description": description}))
return description
PDF Processing
Detect Native vs Scanned vs Mixed
python
import fitz # PyMuPDF
def classify_pages(pdf_path: str) -> list[dict]:
doc = fitz.open(pdf_path)
pages = []
for i, page in enumerate(doc):
text = page.get_text().strip()
images = page.get_images(full=True)
pages.append({
"page": i + 1,
"type": "native" if len(text) > 100 else ("scanned" if images else "blank"),
"char_count": len(text),
"image_count": len(images),
})
return pages
Native Text Extraction
python
import pdfplumber
def extract_native_text(pdf_path: str) -> list[dict]:
chunks = []
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
# Strip headers/footers: exclude top 5% and bottom 7% of page
h = page.height
crop = page.crop((0, h * 0.05, page.width, h * 0.93))
text = crop.extract_text(layout=True) or ""
if text.strip():
chunks.append({"page": page.page_number, "text": text, "type": "text"})
return chunks
Scanned Page OCR
python
from pdf2image import convert_from_path
import pytesseract
from PIL import Image
def ocr_page(page_image: Image.Image, lang: str = "eng") -> dict:
data = pytesseract.image_to_data(
page_image, lang=lang, output_type=pytesseract.Output.DICT
)
# Filter low-confidence words
words = [
w for w, conf in zip(data["text"], data["conf"])
if int(conf) > 40 and w.strip()
]
text = " ".join(words)
avg_conf = sum(c for c in data["conf"] if int(c) > 0) / max(
sum(1 for c in data["conf"] if int(c) > 0), 1
)
return {"text": text, "ocr_confidence": avg_conf / 100}
def ocr_pdf(pdf_path: str, dpi: int = 300) -> list[dict]:
images = convert_from_path(pdf_path, dpi=dpi)
return [{"page": i + 1, **ocr_page(img)} for i, img in enumerate(images)]
Multi-Column Layout
python
# BAD: top-to-bottom reading interleaves columns
text = page.extract_text()
# GOOD: cluster by x-coordinate, sort each column independently
def extract_columns(page) -> str:
words = page.extract_words()
if not words:
return ""
# Detect column boundaries by gap in x-coordinates
x_positions = sorted({int(w["x0"] // 50) * 50 for w in words})
mid_x = (x_positions[0] + x_positions[-1]) / 2 if len(x_positions) > 1 else float("inf")
left = sorted([w for w in words if w["x0"] < mid_x], key=lambda w: (w["top"], w["x0"]))
right = sorted([w for w in words if w["x0"] >= mid_x], key=lambda w: (w["top"], w["x0"]))
def words_to_text(ws):
return " ".join(w["text"] for w in ws)
return words_to_text(left) + "\n\n" + words_to_text(right)
Table Extraction from PDFs
python
def extract_tables_from_page(page) -> list[dict]:
tables = page.extract_tables()
result = []
for table in tables:
if not table or not table[0]:
continue
headers = [str(h or "").strip() for h in table[0]]
rows = []
for row in table[1:]:
if any(cell for cell in row if cell): # skip all-empty rows
rows.append({headers[i]: str(cell or "").strip()
for i, cell in enumerate(row)})
# Serialize as Markdown — preserves structure in embedding
md = "| " + " | ".join(headers) + " |\n"
md += "| " + " | ".join("---" for _ in headers) + " |\n"
for row in rows:
md += "| " + " | ".join(row.get(h, "") for h in headers) + " |\n"
result.append({"type": "table", "markdown": md, "row_count": len(rows)})
return result
Cross-Page Tables
python
# BAD: Each page's table fragment is an independent chunk
# GOOD: Carry headers forward onto continuation pages
def merge_cross_page_tables(page_tables: list[list[dict]]) -> list[dict]:
merged = []
pending_header = None
for page_idx, tables in enumerate(page_tables):
for table in tables:
headers = table["headers"]
# If this table's column count matches the last seen header,
# and it has no header row of its own, it's a continuation
if (pending_header and
len(headers) == len(pending_header) and
headers == pending_header):
# Prepend headers to continuation chunk
table["continuation"] = True
table["headers_prepended"] = pending_header
else:
pending_header = headers
merged.append(table)
return merged
Embedded Images in PDFs
python
import fitz
import base64
import anthropic
client = anthropic.Anthropic()
def extract_and_describe_images(pdf_path: str, cache_dir) -> list[dict]:
doc = fitz.open(pdf_path)
figures = []
for page_num, page in enumerate(doc, 1):
for img_index, img in enumerate(page.get_images(full=True)):
xref = img[0]
base_image = doc.extract_image(xref)
image_bytes = base_image["image"]
description = vision_with_cache(
image_bytes,
"Describe this image. If it is a chart or graph, extract: "
"chart type, title, axis labels, approximate values, and key trends. "
"If it contains text, extract all visible text verbatim.",
cache_dir,
)
# Find adjacent caption text (heuristic: text within 50pt below image bbox)
figures.append({
"page": page_num,
"type": "figure",
"description": description,
"image_index": img_index,
})
return figures
Heading Hierarchy Detection
python
def detect_headings(page) -> list[dict]:
"""Extract text with font size metadata for hierarchy inference."""
blocks = []
for block in page.get_text("dict")["blocks"]:
if block.get("type") != 0: # skip image blocks
continue
for line in block.get("lines", []):
for span in line.get("spans", []):
blocks.append({
"text": span["text"].strip(),
"size": round(span["size"]),
"bold": "bold" in span.get("font", "").lower(),
"bbox": span["bbox"],
})
# Infer heading level from font size rank
sizes = sorted({b["size"] for b in blocks}, reverse=True)
size_to_level = {s: i + 1 for i, s in enumerate(sizes[:4])}
return [
{**b, "heading_level": size_to_level.get(b["size"])}
for b in blocks
if b["bold"] or b["size"] in size_to_level
]
Excel Processing
Multi-Sheet with Visibility Check
python
import openpyxl
def load_workbook_sheets(path: str, include_hidden: bool = False) -> dict:
wb = openpyxl.load_workbook(path, data_only=True)
sheets = {}
for name in wb.sheetnames:
ws = wb[name]
if not include_hidden and ws.sheet_state != "visible":
continue # never silently include hidden data in public-facing RAG
sheets[name] = ws
return sheets
Merged Cell Normalization
python
def normalize_merged_cells(ws) -> list[list]:
"""Fill merged cell siblings with the top-left value before DataFrame conversion."""
for merge_range in ws.merged_cells.ranges:
top_left = ws.cell(merge_range.min_row, merge_range.min_col).value
for row in ws.iter_rows(
min_row=merge_range.min_row, max_row=merge_range.max_row,
min_col=merge_range.min_col, max_col=merge_range.max_col,
):
for cell in row:
cell.value = top_left # fill siblings
return [[cell.value for cell in row] for row in ws.iter_rows()]
Formula vs Value Handling
python
# BAD: reads formula strings, not computed values
wb = openpyxl.load_workbook(path) # data_only defaults to False
# GOOD: reads cached computed values
wb = openpyxl.load_workbook(path, data_only=True)
# If a cell returns None under data_only=True, the file was never calculated
# → warn and flag for manual review; do NOT embed None as "missing"