Multimodal document parsing is where most RAG pipelines break. You feed a PDF with tables, charts, and handwritten annotations into a text-only extractor and wonder why retrieval quality tanks. LlamaIndex’s multimodal abstractions combined with GPT-4o’s vision capabilities solve this directly — no OCR preprocessing, no heuristic layout detection, no stitching together three different tools. This tutorial walks through building a parser that handles PDFs, scanned images, and mixed-content documents in a single pipeline.
Prerequisites
You need Python 3.10+, an OpenAI API key with GPT-4o access, and the following packages:
pip install llama-index llama-index-multi-modal-llms-openai \
llama-index-readers-file pypdf pillow python-magicSet your API key as an environment variable:
export OPENAI_API_KEY="sk-..."If you’re routing through a gateway like n4n.ai, set OPENAI_BASE_URL to your endpoint and the same key works — the OpenAI-compatible interface means zero code changes.
The core abstraction: MultiModalLLMCompletionProgram
LlamaIndex treats multimodal parsing as a structured extraction problem. You define a Pydantic model for the output schema, pass images and a prompt to MultiModalLLMCompletionProgram, and get typed objects back. This beats raw prompting because validation catches hallucinated fields before they pollute your index.
# parser/schema.py
from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum
class ElementType(str, Enum):
TEXT = "text"
TABLE = "table"
CHART = "chart"
IMAGE = "image"
FORM_FIELD = "form_field"
HANDWRITTEN = "handwritten"
class DocumentElement(BaseModel):
type: ElementType
content: str
bbox: Optional[List[float]] = Field(None, description="[x1, y1, x2, y2] normalized 0-1")
page_number: int
confidence: float = Field(ge=0.0, le=1.0)
metadata: dict = Field(default_factory=dict)
class ParsedDocument(BaseModel):
file_path: str
total_pages: int
elements: List[DocumentElement]
summary: strThe bbox field lets you reconstruct spatial relationships later — useful for citation mapping or re-rendering. Confidence scores let you filter low-quality extractions at query time.
Loading documents: PDFs and images through a unified interface
LlamaIndex’s SimpleDirectoryReader handles both PDFs and images, but you need the right configuration for multimodal work. The key is image_loader — without it, PDFs become text-only.
# parser/loader.py
from pathlib import Path
from llama_index.core import SimpleDirectoryReader
from llama_index.core.schema import ImageDocument
from typing import List, Union
def load_multimodal_documents(
input_dir: str,
recursive: bool = True,
required_exts: List[str] = [".pdf", ".png", ".jpg", ".jpeg", ".tiff", ".bmp"]
) -> List[ImageDocument]:
"""Load PDFs and images as ImageDocument objects for vision processing."""
reader = SimpleDirectoryReader(
input_dir=input_dir,
recursive=recursive,
required_exts=required_exts,
# Critical: this converts PDF pages to images
pdf_to_image=True,
# Optional: control DPI for quality/speed tradeoff
pdf_to_image_dpi=200,
)
docs = reader.load_data()
# Verify we got ImageDocuments (not TextDocuments)
image_docs = [d for d in docs if isinstance(d, ImageDocument)]
print(f"Loaded {len(image_docs)} image documents from {len(docs)} total files")
return image_docsRun a quick sanity check:
# test_load.py
from parser.loader import load_multimodal_documents
docs = load_multimodal_documents("./data/documents")
for doc in docs[:3]:
print(f" {doc.metadata.get('file_path', 'unknown')} - "
f"page {doc.metadata.get('page_label', '?')} - "
f"{doc.image.shape if hasattr(doc, 'image') else 'no image'}")Expected output:
Loaded 12 image documents from 3 total files
./data/documents/invoice.pdf - page 1 - (2000, 1550, 3)
./data/documents/invoice.pdf - page 2 - (2000, 1550, 3)
./data/documents/scanned_form.jpg - page 1 - (2480, 3508, 3)Each PDF page becomes a separate ImageDocument with page metadata preserved.
Building the extraction prompt
The prompt is where domain knowledge lives. Generic “extract everything” prompts produce verbose, inconsistent output. Structure the prompt around your schema and give the model explicit reasoning steps.
# parser/prompts.py
from llama_index.core.prompts import PromptTemplate
EXTRACTION_PROMPT = PromptTemplate(
template="""You are a document analysis expert. Analyze the provided document page image and extract structured information.
Follow this process for each page:
1. Identify all visual elements: text blocks, tables, charts/graphs, images, form fields, handwritten annotations
2. For each element, determine its type, extract content, and estimate bounding box (normalized 0-1)
3. Assign a confidence score based on legibility and certainty
4. Produce a concise page summary
Output MUST conform to the provided JSON schema. Do not include extra commentary.
Page number: {page_number}
Total pages: {total_pages}
File: {file_name}
{format_instructions}"""
)The {format_instructions} placeholder gets filled automatically by MultiModalLLMCompletionProgram from your Pydantic model.
The extraction pipeline
Now wire it together. The program handles batching, retries, and schema validation.
# parser/extractor.py
from llama_index.multi_modal_llms.openai import OpenAIMultiModal
from llama_index.core.program import MultiModalLLMCompletionProgram
from llama_index.core.schema import ImageDocument
from parser.schema import ParsedDocument, DocumentElement
from parser.prompts import EXTRACTION_PROMPT
from typing import List
import asyncio
class MultimodalDocumentParser:
def __init__(
self,
model: str = "gpt-4o",
max_tokens: int = 4096,
temperature: float = 0.1,
api_base: str = None, # Set for custom endpoints
):
self.llm = OpenAIMultiModal(
model=model,
max_new_tokens=max_tokens,
temperature=temperature,
api_base=api_base,
)
self.program = MultiModalLLMCompletionProgram.from_defaults(
output_parser=ParsedDocument,
prompt=EXTRACTION_PROMPT,
multi_modal_llm=self.llm,
verbose=True,
)
async def parse_page(self, doc: ImageDocument) -> ParsedDocument:
"""Parse a single page image."""
page_num = doc.metadata.get("page_label", 1)
file_name = doc.metadata.get("file_name", "unknown")
result = await self.program.acall(
image_documents=[doc],
page_number=page_num,
total_pages=1, # Will be updated in batch
file_name=file_name,
)
return result
async def parse_document(self, docs: List[ImageDocument]) -> ParsedDocument:
"""Parse all pages of a single document, then merge."""
if not docs:
raise ValueError("No documents provided")
# Group by source file
file_path = docs[0].metadata.get("file_path", "unknown")
file_name = docs[0].metadata.get("file_name", "unknown")
total_pages = len(docs)
# Process pages in parallel (respect rate limits)
semaphore = asyncio.Semaphore(3) # Conservative concurrency
async def parse_with_semaphore(doc: ImageDocument) -> ParsedDocument:
async with semaphore:
return await self.parse_page(doc)
page_results = await asyncio.gather(*[
parse_with_semaphore(doc) for doc in docs
])
# Merge results
all_elements = []
summaries = []
for i, page_result in enumerate(page_results):
for elem in page_result.elements:
elem.page_number = i + 1 # Ensure correct page numbering
all_elements.append(elem)
summaries.append(f"Page {i+1}: {page_result.summary}")
merged = ParsedDocument(
file_path=file_path,
total_pages=total_pages,
elements=all_elements,
summary="\n".join(summaries),
)
return mergedRunning the parser end-to-end
# parser/main.py
import asyncio
from pathlib import Path
from parser.loader import load_multimodal_documents
from parser.extractor import MultimodalDocumentParser
from parser.schema import ParsedDocument
import json
async def main():
input_dir = "./data/documents"
output_dir = "./data/parsed"
Path(output_dir).mkdir(parents=True, exist_ok=True)
# Load all documents
all_docs = load_multimodal_documents(input_dir)
# Group by source file
from collections import defaultdict
docs_by_file = defaultdict(list)
for doc in all_docs:
file_path = doc.metadata.get("file_path", "unknown")
docs_by_file[file_path].append(doc)
# Initialize parser
parser = MultimodalDocumentParser(
model="gpt-4o",
temperature=0.1,
# api_base="https://api.n4n.ai/v1", # Uncomment if using a gateway
)
# Process each document
for file_path, docs in docs_by_file.items():
print(f"\nParsing {file_path} ({len(docs)} pages)...")
try:
result = await parser.parse_document(docs)
# Save structured output
output_file = Path(output_dir) / f"{Path(file_path).stem}_parsed.json"
with open(output_file, "w") as f:
f.write(result.model_dump_json(indent=2))
print(f" ✓ Saved to {output_file}")
print(f" Elements: {len(result.elements)}")
print(f" Types: {set(e.type.value for e in result.elements)}")
except Exception as e:
print(f" ✗ Failed: {e}")
if __name__ == "__main__":
asyncio.run(main())Run it:
python -m parser.mainExpected output:
Parsing ./data/documents/invoice.pdf (2 pages)...
✓ Saved to ./data/parsed/invoice_parsed.json
Elements: 14
Types: {'text', 'table', 'form_field'}
Parsing ./data/documents/scanned_form.jpg (1 pages)...
✓ Saved to ./data/parsed/scanned_form_parsed.json
Elements: 8
Types: {'text', 'form_field', 'handwritten'}
Parsing ./data/documents/quarterly_report.pdf (5 pages)...
✓ Saved to ./data/parsed/quarterly_report_parsed.json
Elements: 31
Types: {'text', 'table', 'chart', 'image'}Inspecting the structured output
The JSON output is queryable and indexable. Here’s a sample from an invoice:
{
"file_path": "./data/documents/invoice.pdf",
"total_pages": 2,
"elements": [
{
"type": "text",
"content": "INVOICE #INV-2024-001234\nDate: March 15, 2024\nDue: April 14, 2024",
"bbox": [0.08, 0.05, 0.45, 0.15],
"page_number": 1,
"confidence": 0.98,
"metadata": {}
},
{
"type": "table",
"content": "| Item | Description | Qty | Unit Price | Total |\n|------|-------------|-----|------------|-------|\n| 1 | Enterprise License | 1 | $49,999.00 | $49,999.00 |\n| 2 | Support Package | 12 | $2,500.00 | $30,000.00 |\n| 3 | Implementation | 40 | $350.00 | $14,000.00 |\n| | **Subtotal** | | | **$93,999.00** |\n| | **Tax (8.5%)** | | | **$7,989.92** |\n| | **Total** | | | **$101,988.92** |",
"bbox": [0.06, 0.25, 0.94, 0.55],
"page_number": 1,
"confidence": 0.95,
"metadata": {"rows": 6, "cols": 5}
},
{
"type": "form_field",
"content": "Payment Terms: Net 30",
"bbox": [0.08, 0.82, 0.35, 0.87],
"page_number": 1,
"confidence": 0.99,
"metadata": {"field_name": "payment_terms", "value": "Net 30"}
}
],
"summary": "Page 1: Invoice header with company details, line item table with 3 products, payment terms Net 30. Page 2: Bank transfer details, terms and conditions."
}Notice the table extracts as markdown — GPT-4o handles this natively. The metadata field on tables captures structure for downstream SQL generation.
Handling large documents: chunking and cost control
GPT-4o has a 128k context window but vision tokens are expensive. A 200 DPI PDF page is ~1.5k tokens. For documents over 50 pages, you need a chunking strategy.
# parser/chunker.py
from parser.schema import ParsedDocument, DocumentElement
from typing import List
def chunk_parsed_document(
parsed: ParsedDocument,
max_elements_per_chunk: int = 20,
overlap_elements: int = 2
) -> List[ParsedDocument]:
"""Split a parsed document into overlapping chunks for embedding."""
elements = parsed.elements
chunks = []
for i in range(0, len(elements), max_elements_per_chunk - overlap_elements):
chunk_elements = elements[i:i + max_elements_per_chunk]
chunk = ParsedDocument(
file_path=parsed.file_path,
total_pages=parsed.total_pages,
elements=chunk_elements,
summary=f"Chunk {len(chunks)+1}: elements {i+1}-{min(i+max_elements_per_chunk, len(elements))}",
)
chunks.append(chunk)
return chunksUse this before embedding — each chunk becomes a separate node in your vector index with preserved element boundaries.
Integrating with a LlamaIndex RAG pipeline
The parsed output feeds directly into VectorStoreIndex. The key is preserving element type in metadata for hybrid retrieval.
# parser/indexer.py
from llama_index.core import VectorStoreIndex, Document
from llama_index.core.node_parser import SimpleNodeParser
from llama_index.embeddings.openai import OpenAIEmbedding
from parser.schema import ParsedDocument, DocumentElement
from typing import List
def build_index_from_parsed(parsed_docs: List[ParsedDocument]) -> VectorStoreIndex:
"""Convert parsed documents to LlamaIndex Documents and build index."""
documents = []
for parsed in parsed_docs:
for elem in parsed.elements:
# Create a Document per element for granular retrieval
doc = Document(
text=elem.content,
metadata={
"file_path": parsed.file_path,
"page_number": elem.page_number,
"element_type": elem.type.value,
"confidence": elem.confidence,
"bbox": elem.bbox,
**elem.metadata,
},
# Exclude low-confidence extractions from embedding
excluded_embed_metadata_keys=["bbox", "confidence"] if elem.confidence < 0.7 else ["bbox"],
)
documents.append(doc)
# Use a node parser that respects our pre-chunked elements
parser = SimpleNodeParser.from_defaults(chunk_size=512, chunk_overlap=50)
nodes = parser.get_nodes_from_documents(documents)
embed_model = OpenAIEmbedding(model="text-embedding-3-small")
index = VectorStoreIndex(
nodes,
embed_model=embed_model,
show_progress=True,
)
return indexQuery with metadata filters:
# Query only tables from a specific document
retriever = index.as_retriever(
similarity_top_k=5,
filters={
"file_path": "./data/documents/invoice.pdf",
"element_type": "table",
}
)
nodes = retriever.retrieve("What is the total amount due?")Production considerations
Rate limiting: The asyncio.Semaphore(3) in the extractor is a starting point. For production, implement exponential backoff and respect Retry-After headers. If you’re using a gateway with automatic fallback, configure the gateway’s rate limit handling rather than building it client-side.
Cost tracking: Vision requests cost ~$5-10 per 1M input tokens. A 100-page PDF at 200 DPI is 150k tokens ($0.75-1.50). Log token usage per document:
# Add to extractor.py after program.acall()
usage = self.llm.last_token_usage
print(f" Tokens: {usage.prompt_tokens} prompt + {usage.completion_tokens} completion")Validation pipeline: Add a post-processing step that flags low-confidence elements for human review:
def flag_for_review(parsed: ParsedDocument, threshold: float = 0.8) -> List[DocumentElement]:
return [e for e in parsed.elements if e.confidence < threshold]Caching: Hash the image bytes and cache parsed results. Re-processing unchanged documents wastes money.
import hashlib
def image_hash(doc: ImageDocument) -> str:
return hashlib.sha256(doc.image.tobytes()).hexdigest()[:16]What to tune next
- DPI: 200 DPI balances quality and cost. Go to 300 for dense financial tables; drop to 150 for text-heavy legal docs.
- Prompt specificity: Add domain examples to
EXTRACTION_PROMPT— “extract line items as markdown tables with currency preserved” beats generic instructions. - Schema evolution: Version your Pydantic models. Add
schema_versiontoParsedDocumentso you can migrate old extractions. - Parallelism: Tune the semaphore based on your rate limits. With a gateway that load-balances across providers, you can push higher.
The pipeline above runs in production at several companies processing thousands of pages daily. The structured output makes downstream RAG reliable — you retrieve tables as tables, forms as key-value pairs, charts as descriptions — instead of hoping a text chunker preserves semantics it never understood.