Multimodal LLM Integration: Processing Images, PDFs, and Documents with Vision APIs

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:

• How vision APIs encode and process images, and what this means for your token budget
• How to extract and route content from multi-page PDFs to the right processing strategy
• How to write structured extraction prompts that return consistent, parseable output from messy visual documents
• How to handle scanned documents, tables, charts, and mixed-content PDFs in a unified pipeline
• How to manage latency, cost, and failure modes in production multimodal workflows

---

Prerequisites

You should be comfortable with:

• Calling LLM APIs (OpenAI, Anthropic, or Google) programmatically in Python
• Basic PDF manipulation with libraries like PyMuPDF or pdfplumber
• JSON parsing and Pydantic for structured data validation
• Async Python (helpful but not required)

You'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.

---

How Vision APIs Actually Work

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.

Image Encoding for APIs

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.

---

PDF Processing: The Routing Problem

Real-world PDFs aren't homogeneous. A single 20-page report might contain:

• Native text pages (created digitally — text is already there, selectable)
• Scanned pages (images of physical documents — no underlying text)
• Pages with embedded charts or diagrams
• Tables that look text-based but have complex merged-cell structures
• Pages with both substantial text and critical visual elements

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.

---

Structured Extraction with Vision Prompts

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.

The Extraction Prompt Pattern

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.

Handling Tables: The Hardest Case

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.

---

Processing Multi-Page Documents

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.

The Context Window Strategy

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.

Document-Aware Prompting

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.

---

Building a Complete Document Intelligence Pipeline

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
        }

---

Hands-On Exercise: Automating Expense Report Processing

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:

1. 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).

2. Implement the classification and routing pipeline from this lesson.

3. 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]

4. Add a validation layer that checks business rules:

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

5. Generate a summary CSV of all processed expenses with a flag column for items needing human review.

This exercise exercises every part of the pipeline: classification, vision extraction, structured output parsing, business logic validation, and output formatting.

---

Common Mistakes & Troubleshooting

"The model returns different values on repeated runs"

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.

"The model is hallucinating numbers that aren't in the document"

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.

"JSON parsing fails on the model's response"

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)

"Processing is too slow for our volume"

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.

"The pipeline fails silently on corrupted PDFs"

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

---

Performance Considerations and Choosing Your Approach

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:

• Documents are scanned (no underlying text layer)
• Visual layout is semantically meaningful (financial tables, forms with checkbox fields, spatial form layouts)
• Documents contain charts, graphs, or diagrams you need to interpret
• You need to extract information that requires understanding of visual hierarchy

Use text extraction + text LLM when:

• PDFs are digitally created with selectable text
• You're processing very high volumes where cost matters
• Documents are pure prose without meaningful visual structure
• Latency requirements are tight (< 2 seconds)

Use OCR + text LLM (no vision API) when:

• Your OCR engine handles the document type reliably
• You need extreme throughput (vision APIs are slow at scale)
• Document variety is predictable enough to tune OCR pipeline

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.

---

Summary & Next Steps

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:

• Explore OpenAI's Structured Outputs feature (available in newer API versions) which enforces JSON schema compliance at the API level, eliminating most parsing failures
• Investigate Gemini 1.5 Pro for long documents — its 1M token context window can process entire document collections in a single call
• Build fine-tuning datasets from your human-reviewed extractions to eventually train smaller, cheaper domain-specific vision models
• Add vector storage to your document pipeline so extracted content becomes searchable via RAG — the combination of vision extraction and semantic search is the foundation of serious enterprise document intelligence