Text-only LLMs can't handle scanned receipts, chart-heavy reports, or complex PDF tables — but multimodal models can. Learn how to build a production document intelligence pipeline that routes, processes, and extracts structured data from any document type using vision APIs, with full cost and accuracy control.
Your expense reporting pipeline breaks every time someone uploads a photo of a handwritten receipt. Your document extraction workflow fails on PDFs with embedded charts. Your contract review tool chokes on scanned agreements that never went through OCR. These aren't edge cases — they're the majority of real-world documents, and text-only LLMs simply can't handle them.
Multimodal LLMs change that equation entirely. Models like GPT-4o, Claude 3.5 Sonnet, and Gemini 1.5 Pro can reason about images, diagrams, tables, and document layouts natively — without a separate OCR pipeline, without converting everything to plain text, without losing the spatial relationships that make a balance sheet a balance sheet rather than a list of numbers. When a model can see that two numbers are in the same column, under the same header, in the same table — that context changes everything about how it interprets them.
By the end of this lesson, you'll have built a production-grade document intelligence pipeline that handles real-world document chaos: scanned PDFs, mixed-content files with both text and images, structured forms, and visual reports. You'll understand not just how to call vision APIs, but how to architect the preprocessing, prompt engineering, and error handling that makes these systems actually reliable.
What you'll learn:
You should be comfortable with:
PyMuPDF or pdfplumberYou'll need API keys for at least one of: OpenAI (GPT-4o), Anthropic (Claude 3.5 Sonnet), or Google (Gemini 1.5 Pro). Code examples use OpenAI's API but the patterns transfer directly.
Before writing a single line of document processing code, you need to understand the mechanism underneath — because it directly affects your architecture decisions.
When you send an image to GPT-4o or Claude, it doesn't receive pixels. The image goes through a vision encoder that converts it into a sequence of tokens, just like text. Those vision tokens get concatenated with your text prompt tokens and fed into the transformer together. The model doesn't have a separate "image understanding" mode — it reasons about images and text simultaneously in the same attention space.
This has three practical implications:
Images cost tokens, and the cost scales with resolution. OpenAI's tile-based pricing for GPT-4o charges 85 base tokens plus 170 tokens per 512×512 tile. A 1024×1024 image costs 765 tokens. A full-page scanned document at 2000×2600 pixels costs somewhere around 1500–2000 tokens just for the image. Multiply that by 50 pages and you've burned 75,000–100,000 tokens before writing a single prompt word.
Resolution affects accuracy. Send a blurry thumbnail of a financial table and the model will hallucinate values. Send the full-resolution scan and it reads accurately. There's a sweet spot — usually around 1500–2000px on the long edge for document pages — where you get accuracy without unnecessary token bloat.
The model sees spatial layout. This is what makes vision APIs genuinely different from OCR + text. A model can understand that a number is in the "Q3" column of a revenue table, that a signature appears at the bottom of a contract page, or that a warning label uses red text. Text extraction destroys this structure; vision preserves it.
OpenAI and Anthropic both accept images as either base64-encoded strings or URLs. For production pipelines, base64 is almost always the right choice — it doesn't require your images to be publicly accessible, eliminates network round-trips for image fetching, and makes your requests self-contained.
import base64
from pathlib import Path
def encode_image_to_base64(image_path: str) -> str:
"""Encode a local image file to base64 string."""
with open(image_path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")
def build_vision_message(
prompt: str,
image_path: str,
detail: str = "high" # "low", "high", or "auto"
) -> dict:
"""
Build an OpenAI-format message with an embedded image.
detail="low" forces 512x512 processing (cheaper, worse accuracy)
detail="high" uses full tile-based processing (accurate, more tokens)
detail="auto" lets the API decide based on image size
"""
image_data = encode_image_to_base64(image_path)
return {
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_data}",
"detail": detail
}
},
{
"type": "text",
"text": prompt
}
]
}
Notice that the content is a list, not a string. When you include images, the message content becomes an array of content blocks — some image, some text. You can include multiple images in a single message by adding more image blocks to that array. We'll use this heavily when processing multi-page documents.
Real-world PDFs aren't homogeneous. A single 20-page report might contain:
The worst approach is treating all of these the same way. Sending every page to the vision API is wasteful and slow. Sending every page through text extraction misses the visual content. You need a routing layer that classifies pages and handles them appropriately.
import fitz # PyMuPDF
from dataclasses import dataclass
from enum import Enum
from typing import Optional
import io
from PIL import Image
class PageType(Enum):
TEXT_NATIVE = "text_native" # Extract text directly, skip vision
SCANNED = "scanned" # Image-only, must use vision
MIXED = "mixed" # Text + significant visual elements
VISUAL_HEAVY = "visual_heavy" # Charts, diagrams, tables
@dataclass
class PageAnalysis:
page_num: int
page_type: PageType
text_content: Optional[str]
image_path: Optional[str]
text_coverage: float # fraction of page area covered by text blocks
has_images: bool
image_count: int
def analyze_pdf_page(page: fitz.Page, page_num: int, output_dir: str) -> PageAnalysis:
"""
Classify a PDF page and extract or render it appropriately.
The key heuristic: if a page has fewer than 50 characters of
extractable text but contains image objects, it's almost certainly
a scanned page. If it has substantial text AND images, it's mixed.
"""
# Attempt text extraction
text_blocks = page.get_text("blocks")
extracted_text = page.get_text("text").strip()
# Count images embedded in the page
image_list = page.get_images(full=True)
has_images = len(image_list) > 0
image_count = len(image_list)
# Calculate text coverage (rough proxy for how "text-rich" the page is)
page_area = page.rect.width * page.rect.height
text_area = sum(
(b[2] - b[0]) * (b[3] - b[1])
for b in text_blocks
if len(b) > 4
)
text_coverage = text_area / page_area if page_area > 0 else 0
# Classify the page
char_count = len(extracted_text)
if char_count < 50 and has_images:
page_type = PageType.SCANNED
elif char_count > 200 and not has_images:
page_type = PageType.TEXT_NATIVE
elif has_images and image_count > 2:
page_type = PageType.VISUAL_HEAVY
elif has_images:
page_type = PageType.MIXED
else:
page_type = PageType.TEXT_NATIVE
# Render to image if we'll need vision processing
image_path = None
if page_type in (PageType.SCANNED, PageType.MIXED, PageType.VISUAL_HEAVY):
# 2x zoom gives ~150 DPI equivalent — good balance of quality vs size
mat = fitz.Matrix(2.0, 2.0)
pix = page.get_pixmap(matrix=mat, alpha=False)
image_path = f"{output_dir}/page_{page_num:04d}.jpg"
pix.save(image_path)
return PageAnalysis(
page_num=page_num,
page_type=page_type,
text_content=extracted_text if char_count > 50 else None,
image_path=image_path,
text_coverage=text_coverage,
has_images=has_images,
image_count=image_count
)
def analyze_pdf(pdf_path: str, output_dir: str = "/tmp/pdf_pages") -> list[PageAnalysis]:
"""Analyze all pages of a PDF and return their classifications."""
import os
os.makedirs(output_dir, exist_ok=True)
doc = fitz.open(pdf_path)
analyses = []
for page_num in range(len(doc)):
page = doc[page_num]
analysis = analyze_pdf_page(page, page_num, output_dir)
analyses.append(analysis)
print(f"Page {page_num + 1}: {analysis.page_type.value} "
f"(chars: {len(analysis.text_content or '')}, "
f"images: {analysis.image_count})")
doc.close()
return analyses
The 50-character threshold for detecting scanned pages might seem arbitrary, but it handles the common case where PDF "scanning" leaves behind a few metadata characters or whitespace that trips up naive character-count checks. Adjust it based on your document domain — legal documents might have headers that push legitimate scanned pages past that threshold.
Getting a model to describe an image is easy. Getting it to return consistent, structured data that you can actually use in a pipeline is the real skill. The difference comes down to prompt engineering that treats the model like a precise extraction tool, not a creative writer.
For document intelligence tasks, your prompts need to do three things: specify exactly what to extract, define the output format with no ambiguity, and handle the case where the data isn't present.
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Optional
import json
client = OpenAI()
class InvoiceLineItem(BaseModel):
description: str
quantity: Optional[float] = None
unit_price: Optional[float] = None
total: float
class ExtractedInvoice(BaseModel):
invoice_number: Optional[str] = None
invoice_date: Optional[str] = None
vendor_name: Optional[str] = None
vendor_address: Optional[str] = None
bill_to_name: Optional[str] = None
line_items: list[InvoiceLineItem] = Field(default_factory=list)
subtotal: Optional[float] = None
tax_amount: Optional[float] = None
total_amount: Optional[float] = None
payment_terms: Optional[str] = None
extraction_confidence: str = Field(
description="high/medium/low based on image quality and clarity"
)
extraction_notes: str = Field(
description="Any ambiguities, unclear values, or important caveats"
)
INVOICE_EXTRACTION_PROMPT = """
You are a precise document extraction system. Extract all invoice data from this image.
Return a JSON object with exactly this structure — no additional commentary, no markdown formatting, just the raw JSON:
{
"invoice_number": "string or null",
"invoice_date": "YYYY-MM-DD format or null if unclear",
"vendor_name": "string or null",
"vendor_address": "full address as single string or null",
"bill_to_name": "string or null",
"line_items": [
{
"description": "string",
"quantity": number_or_null,
"unit_price": number_or_null,
"total": number
}
],
"subtotal": number_or_null,
"tax_amount": number_or_null,
"total_amount": number_or_null,
"payment_terms": "string or null",
"extraction_confidence": "high|medium|low",
"extraction_notes": "describe any ambiguities here"
}
Rules:
- All monetary values should be numbers (no currency symbols)
- Dates must be in YYYY-MM-DD format
- If a field is not present in the document, use null
- If you can see a value but cannot read it clearly, note it in extraction_notes and use null
- Do not infer or calculate values that aren't explicitly shown
"""
def extract_invoice(image_path: str) -> ExtractedInvoice:
"""Extract structured invoice data from an image using GPT-4o."""
image_data = encode_image_to_base64(image_path)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_data}",
"detail": "high"
}
},
{
"type": "text",
"text": INVOICE_EXTRACTION_PROMPT
}
]
}
],
max_tokens=2000,
temperature=0 # Zero temperature for extraction tasks — you want determinism
)
raw_response = response.choices[0].message.content.strip()
# Models sometimes wrap JSON in markdown code blocks despite instructions
# This is a persistent behavior worth defending against
if raw_response.startswith("```"):
raw_response = raw_response.split("```")[1]
if raw_response.startswith("json"):
raw_response = raw_response[4:]
data = json.loads(raw_response)
return ExtractedInvoice(**data)
Why temperature=0 matters here. Extraction is not a creative task. You want the model to read what's there and report it faithfully, not explore the distribution of possible interpretations. Temperature=0 dramatically reduces hallucination on structured extraction tasks. Some practitioners use 0.1 or 0.2 to avoid rare edge cases where greedy decoding can get stuck, but for most extraction workflows, 0 is the right starting point.
Tables are where text extraction genuinely fails and vision genuinely excels — but also where vision models most frequently hallucinate. A complex merged-cell table in a financial report can fool even GPT-4o into swapping values between columns.
The mitigation strategy is to make the model show its work:
TABLE_EXTRACTION_PROMPT = """
Extract the data from the table in this image.
First, describe the table structure:
- How many columns are there?
- What are the column headers?
- Are there any merged cells or grouped rows?
- What does each row represent?
Then, output the table as a JSON array of objects, where each object represents one data row.
Use the column headers as keys. Preserve the exact values as shown — don't round numbers,
don't expand abbreviations.
If you see any cell values that are unclear or could be misread, include them with a
"?" prefix (e.g., "?42.3") so they can be flagged for review.
Output format:
{
"table_description": "brief description of what this table contains",
"column_headers": ["header1", "header2", ...],
"has_merged_cells": true/false,
"rows": [
{"header1": "value", "header2": "value", ...},
...
],
"questionable_values": ["list any cell values you marked with ?"]
}
"""
The key technique here is the chain-of-thought structure: ask the model to reason about the table structure first before extracting values. When the model explicitly identifies "there are 4 columns: Quarter, Revenue, COGS, Gross Margin," it's much less likely to mix up which number belongs in which column.
Single-image extraction is the easy case. Real document workflows involve multi-page PDFs where context from page 3 is essential for interpreting page 7. A contract's defined terms section matters when reading the obligations section. A financial report's note disclosures matter when reading the summary figures.
For documents under ~50 pages, you can often send all relevant pages in a single API call, using the multi-content-block format. The model gets the full document context and can cross-reference pages naturally.
def process_multipage_document(
page_analyses: list[PageAnalysis],
extraction_prompt: str,
max_pages_per_call: int = 20
) -> list[dict]:
"""
Process a multi-page document by batching vision pages.
For text-native pages, we extract text and include it as text blocks.
For visual pages, we include the rendered image.
This hybrid approach minimizes token cost while preserving visual fidelity.
"""
results = []
# Process in batches if document is large
for batch_start in range(0, len(page_analyses), max_pages_per_call):
batch = page_analyses[batch_start:batch_start + max_pages_per_call]
content_blocks = []
for analysis in batch:
# Always include a page marker for context
content_blocks.append({
"type": "text",
"text": f"\n--- PAGE {analysis.page_num + 1} ---\n"
})
if analysis.page_type == PageType.TEXT_NATIVE and analysis.text_content:
# Include extracted text — much cheaper than vision
content_blocks.append({
"type": "text",
"text": analysis.text_content
})
elif analysis.image_path:
# Include rendered image for visual pages
image_data = encode_image_to_base64(analysis.image_path)
content_blocks.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_data}",
"detail": "high"
}
})
# If the page also has some text, include it too
if analysis.text_content:
content_blocks.append({
"type": "text",
"text": f"[Extracted text from this page: {analysis.text_content}]"
})
# Add the extraction instruction at the end
content_blocks.append({
"type": "text",
"text": extraction_prompt
})
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": content_blocks}],
max_tokens=4000,
temperature=0
)
results.append({
"batch_start": batch_start,
"batch_end": batch_start + len(batch),
"content": response.choices[0].message.content,
"usage": {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens
}
})
return results
Watch your context window limits. GPT-4o supports 128K tokens. Twenty high-resolution document pages at ~1500 tokens each is 30K tokens for images alone — well within limits. But 60 pages starts pushing you toward 90K tokens, leaving little room for your prompt and the output. For very long documents, consider a two-pass strategy: first pass extracts a summary and table of contents, second pass processes only the sections relevant to your task.
When you send multiple pages together, your prompt needs to tell the model how to handle the multi-page context:
CONTRACT_EXTRACTION_PROMPT = """
You are reviewing a multi-page contract document. The pages are presented in order,
marked with --- PAGE N --- headers.
Extract the following information, noting which page number each item appears on:
1. PARTIES: Full legal names of all contracting parties
2. EFFECTIVE_DATE: When the contract becomes effective
3. TERM: Duration of the agreement
4. KEY_OBLIGATIONS: What each party is required to do (summarize, don't quote)
5. PAYMENT_TERMS: All payment amounts, schedules, and conditions
6. TERMINATION_CONDITIONS: What triggers termination and notice requirements
7. GOVERNING_LAW: Which jurisdiction's law governs the contract
8. DEFINED_TERMS: Any capitalized terms that are explicitly defined
For each item, include:
- The extracted value
- The page number where it appears
- A confidence score (high/medium/low)
- The verbatim quote from the document that supports your extraction
If information spans multiple pages or references earlier sections, note all relevant pages.
Return as a structured JSON object.
"""
Asking the model to cite page numbers and verbatim quotes serves two purposes: it makes the output auditable, and it forces the model to ground its responses in the actual document content rather than making inferences.
Let's put all of this together into a pipeline that a real team could deploy. We'll build a system that processes financial report PDFs — the kind that might contain a mix of executive summary text, embedded charts, financial tables, and auditor notes.
import asyncio
import hashlib
import json
import os
from pathlib import Path
from typing import Callable
from dataclasses import asdict
class DocumentIntelligencePipeline:
"""
Production document intelligence pipeline with:
- Automatic page classification and routing
- Hybrid text/vision processing
- Response caching to avoid re-processing
- Cost tracking
- Structured output validation
"""
def __init__(
self,
openai_client: OpenAI,
cache_dir: str = "/tmp/doc_cache",
temp_dir: str = "/tmp/doc_pages"
):
self.client = openai_client
self.cache_dir = Path(cache_dir)
self.temp_dir = Path(temp_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.temp_dir.mkdir(parents=True, exist_ok=True)
self.total_tokens_used = 0
self.total_api_calls = 0
def _get_cache_key(self, content: str, prompt: str) -> str:
"""Generate a cache key based on content and prompt."""
combined = f"{content}||{prompt}"
return hashlib.sha256(combined.encode()).hexdigest()
def _check_cache(self, cache_key: str) -> Optional[str]:
cache_path = self.cache_dir / f"{cache_key}.json"
if cache_path.exists():
with open(cache_path) as f:
return json.load(f)["response"]
return None
def _write_cache(self, cache_key: str, response: str) -> None:
cache_path = self.cache_dir / f"{cache_key}.json"
with open(cache_path, "w") as f:
json.dump({"response": response}, f)
def _call_vision_api(
self,
content_blocks: list[dict],
extraction_prompt: str,
max_tokens: int = 3000
) -> str:
"""Make a vision API call with caching and cost tracking."""
# Create a cache key from a text representation of the request
request_repr = json.dumps(
[b for b in content_blocks if b["type"] == "text"],
sort_keys=True
)
cache_key = self._get_cache_key(request_repr, extraction_prompt)
cached = self._check_cache(cache_key)
if cached:
print(" [Cache hit — skipping API call]")
return cached
all_content = content_blocks + [
{"type": "text", "text": extraction_prompt}
]
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": all_content}],
max_tokens=max_tokens,
temperature=0
)
result = response.choices[0].message.content
self.total_tokens_used += response.usage.total_tokens
self.total_api_calls += 1
self._write_cache(cache_key, result)
return result
def process_financial_report(
self,
pdf_path: str,
extract_sections: list[str] = None
) -> dict:
"""
Process a financial report PDF end-to-end.
Returns structured extraction results with metadata.
"""
print(f"\nProcessing: {pdf_path}")
print("=" * 50)
# Step 1: Analyze and classify all pages
print("\n[Step 1] Analyzing page types...")
page_analyses = analyze_pdf(pdf_path, str(self.temp_dir))
type_counts = {}
for analysis in page_analyses:
t = analysis.page_type.value
type_counts[t] = type_counts.get(t, 0) + 1
print(f" Page breakdown: {type_counts}")
# Step 2: Extract text from text-native pages
print("\n[Step 2] Extracting text from native pages...")
text_content = {}
for analysis in page_analyses:
if analysis.page_type == PageType.TEXT_NATIVE and analysis.text_content:
text_content[analysis.page_num] = analysis.text_content
print(f" Extracted text from {len(text_content)} pages")
# Step 3: Process visual pages with vision API
print("\n[Step 3] Processing visual pages with vision API...")
visual_pages = [
a for a in page_analyses
if a.page_type in (PageType.SCANNED, PageType.MIXED, PageType.VISUAL_HEAVY)
and a.image_path
]
visual_extractions = {}
for analysis in visual_pages:
print(f" Processing page {analysis.page_num + 1} ({analysis.page_type.value})...")
image_data = encode_image_to_base64(analysis.image_path)
content_blocks = [{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_data}",
"detail": "high"
}
}]
# Choose prompt based on visual content
if analysis.image_count > 2 or analysis.page_type == PageType.VISUAL_HEAVY:
prompt = self._get_chart_extraction_prompt()
else:
prompt = self._get_general_visual_extraction_prompt()
result = self._call_vision_api(content_blocks, prompt)
visual_extractions[analysis.page_num] = result
# Step 4: Synthesize results
print("\n[Step 4] Synthesizing document analysis...")
synthesis = self._synthesize_results(
text_content,
visual_extractions,
page_analyses,
pdf_path
)
print(f"\n[Complete] Total API calls: {self.total_api_calls}, "
f"Tokens used: {self.total_tokens_used:,}")
return synthesis
def _get_chart_extraction_prompt(self) -> str:
return """
This page contains charts or graphs. For each visual element:
1. Identify the chart type (bar, line, pie, etc.)
2. Extract the title and axis labels
3. Extract the data values shown (approximate if exact values aren't labeled)
4. Describe the key trend or insight the chart conveys
Return as JSON with a "charts" array, each containing:
{
"chart_type": "...",
"title": "...",
"x_axis": "...",
"y_axis": "...",
"data_points": {"label": value, ...},
"key_insight": "..."
}
"""
def _get_general_visual_extraction_prompt(self) -> str:
return """
Extract all meaningful content from this document page.
Include:
- Any text content (verbatim where important, summarized where lengthy)
- Any tables (as structured JSON)
- Any key figures, metrics, or values
- Page section/heading if visible
Return as JSON:
{
"section_heading": "...",
"text_content": "...",
"tables": [...],
"key_metrics": {"metric_name": value, ...},
"other_visual_elements": "description of any other important visual content"
}
"""
def _synthesize_results(
self,
text_content: dict,
visual_extractions: dict,
page_analyses: list[PageAnalysis],
source_path: str
) -> dict:
"""Combine all extracted content into a unified document model."""
pages = []
for analysis in page_analyses:
page_data = {
"page_num": analysis.page_num + 1,
"page_type": analysis.page_type.value,
"content": None
}
if analysis.page_num in text_content:
page_data["content"] = {
"type": "text",
"data": text_content[analysis.page_num]
}
elif analysis.page_num in visual_extractions:
raw = visual_extractions[analysis.page_num]
# Attempt to parse as JSON, fall back to raw string
try:
if "```" in raw:
raw = raw.split("```")[1].lstrip("json").strip()
page_data["content"] = {
"type": "visual_extraction",
"data": json.loads(raw)
}
except json.JSONDecodeError:
page_data["content"] = {
"type": "visual_extraction_raw",
"data": raw
}
pages.append(page_data)
return {
"source_file": source_path,
"total_pages": len(page_analyses),
"processing_summary": {
"text_native_pages": sum(
1 for a in page_analyses
if a.page_type == PageType.TEXT_NATIVE
),
"visual_pages_processed": len(visual_extractions),
"total_tokens_used": self.total_tokens_used,
"api_calls_made": self.total_api_calls
},
"pages": pages
}
Here's your real-world project: build an expense report processor that handles the mess that finance teams actually face — a folder full of expense submissions that might be photographed receipts, scanned paper forms, or PDF exports from expense management tools.
Your task:
Create a test dataset by collecting 5–10 receipts or expense documents (photograph physical receipts with your phone, download a few from free stock document sites, or create simple ones in a PDF tool).
Implement the classification and routing pipeline from this lesson.
Extend the ExtractedInvoice model with expense-specific fields:
class ExpenseRecord(BaseModel):
vendor_name: str
transaction_date: Optional[str]
amount: float
currency: str = "USD"
expense_category: str # Travel, Meals, Software, Office, etc.
payment_method: Optional[str] # Credit card last 4 digits if visible
tax_amount: Optional[float]
tip_amount: Optional[float]
business_purpose_hint: Optional[str] # Any memo or description visible
extraction_confidence: str
needs_review: bool # Flag for low-confidence extractions
review_reason: Optional[str]
def validate_expense(record: ExpenseRecord) -> tuple[bool, list[str]]:
"""
Apply business rules to flag suspicious or incomplete expense records.
Returns (is_valid, list_of_issues).
"""
issues = []
if record.extraction_confidence == "low":
issues.append("Low extraction confidence — manual review required")
if record.amount > 500 and not record.business_purpose_hint:
issues.append("Expenses over $500 require visible business purpose")
if record.expense_category == "Meals" and record.amount > 150:
issues.append("Meal expense exceeds per-person policy limit — check attendee count")
if record.transaction_date is None:
issues.append("Transaction date not found — required for reimbursement")
return len(issues) == 0, issues
This exercise exercises every part of the pipeline: classification, vision extraction, structured output parsing, business logic validation, and output formatting.
If you're seeing inconsistent extractions on the same document, the first thing to check is temperature. Anything above 0 introduces randomness. But even at temperature=0, minor variations in prompt wording or API model versions can shift outputs. Add the extraction_confidence and extraction_notes fields to all your schemas — low-confidence extractions are a signal to send the document for human review rather than trusting the automated result.
This typically happens when image quality is poor (the model fills in what it expects to see rather than what's there) or when your prompt doesn't explicitly say "use null if you cannot clearly read the value." Add the "null if unclear" instruction to every extraction prompt. Also increase rendering resolution — try fitz.Matrix(3.0, 3.0) for 225 DPI equivalent if you're seeing accuracy issues on fine-print financial documents.
Three common causes:
def robust_json_parse(raw_response: str) -> dict:
"""
Handle the three most common JSON extraction failure modes.
"""
text = raw_response.strip()
# 1. Response wrapped in markdown code block
if "```json" in text:
text = text.split("```json")[1].split("```")[0].strip()
elif "```" in text:
text = text.split("```")[1].split("```")[0].strip()
# 2. Response has leading explanation text before the JSON
if not text.startswith("{") and "{" in text:
text = text[text.index("{"):]
# 3. Response has trailing text after the JSON
# Find the last closing brace that balances the first opening brace
brace_count = 0
end_pos = 0
for i, char in enumerate(text):
if char == "{":
brace_count += 1
elif char == "}":
brace_count -= 1
if brace_count == 0:
end_pos = i + 1
break
if end_pos > 0:
text = text[:end_pos]
return json.loads(text)
For high-volume workflows, two strategies help most:
Async batching: Use asyncio with aiohttp to fire multiple API calls concurrently. OpenAI's rate limits are per-minute, so you can often run 5–10 requests in parallel before hitting them.
Tiered processing: Not every document needs GPT-4o's full vision capabilities. Use detail="low" for a fast first pass to classify document type and route to specialized prompts, then only use detail="high" for documents where precision matters.
Always wrap PDF processing in proper exception handling and validate the PDF before processing:
def safe_process_pdf(pdf_path: str) -> Optional[list[PageAnalysis]]:
try:
doc = fitz.open(pdf_path)
if doc.is_encrypted:
print(f"WARNING: {pdf_path} is encrypted — cannot process")
return None
if len(doc) == 0:
print(f"WARNING: {pdf_path} has no pages")
return None
doc.close()
return analyze_pdf(pdf_path)
except fitz.FileDataError as e:
print(f"ERROR: Corrupted PDF {pdf_path}: {e}")
return None
except Exception as e:
print(f"ERROR: Unexpected failure on {pdf_path}: {e}")
return None
Vision APIs are powerful but expensive and slow compared to pure text processing. Here's a decision framework for when to use them:
Use vision APIs when:
Use text extraction + text LLM when:
Use OCR + text LLM (no vision API) when:
Hybrid approach (what this lesson built) is usually the right answer for real-world pipelines: text extraction for native PDF pages, vision for visual content, unified output model across both.
On cost: a 20-page financial report with 5 visual pages will typically consume 15,000–25,000 tokens on GPT-4o, costing roughly $0.10–$0.20 at current pricing. That's fine for dozens of documents per day, expensive for tens of thousands.
You've built a production document intelligence pipeline that handles the real-world diversity of document formats. The key architectural insights to carry forward:
Classify before processing. Don't treat all pages or all documents the same. A routing layer that identifies page types pays for itself immediately in cost and accuracy.
Prompt structure determines output reliability. Ask the model to reason about structure before extracting values. Demand specific output formats. Require uncertainty signals. The difference between a brittle demo and a production system is almost entirely in the prompt.
Build for failure, not just success. Caching prevents re-spending on API calls during debugging. Validation layers catch low-confidence extractions before they corrupt downstream data. Graceful JSON parsing handles the model's inevitable formatting inconsistencies.
Vision tokens are expensive. Track them, cache aggressively, and use lower resolution where accuracy allows.
Where to go from here: