4. Vision Analysis

Description

The VisionAnalyzer is the intelligent component of the Document Analyzer: it receives the images extracted by the DocumentProcessor and turns them into useful information. It classifies the document type (invoice, contract, manual), extracts structured data according to the type, and generates image descriptions to index in RAG. It uses GPT-4o as the primary provider with fallback to Claude and Gemini.

Why it matters: A scanned PDF without Vision is just a collection of pixels. The VisionAnalyzer gives it meaning: "this is a $1,740 invoice from Tech Solutions". Without this component, scanned documents would be useless to the system. And with multi-provider fallback, the system keeps working even if one provider fails.

Connection with the module: In Module 2 you learned to send images to GPT-4 Vision and get analysis. Here you integrate that capability into a class with three features: classification, structured extraction, and description for RAG. In addition, you implement the fallback pattern from Module 7: if OpenAI fails or is slow, the system automatically tries Anthropic or Google.

VisionAnalyzer Architecture

Responsibilities

VisionAnalyzer
├── classify()              → Determine the document type
├── extract_structured()    → Extract data by type (invoice, contract, etc.)
├── describe_for_rag()      → Generate text descriptions for indexing
└── _call_with_fallback()   → Run with multi-provider fallback

Decision flow

ProcessedDocument
    │
    ├── Has images? ───────── Yes ──→ classify() with the first image
    │                                    │
    │                                    ▼
    │                          extract_structured() with the type's schema
    │                                    │
    │                                    ▼
    │                          describe_for_rag() for each image
    │
    └── Text only? ────────── Yes ──→ classify() with text (no Vision)
                                         │
                                         ▼
                               extract_structured() with a text LLM

Pydantic Models for Extraction

Base models

from pydantic import BaseModel, Field
from typing import Optional


class ExtractionResult(BaseModel):
    document_type: str
    fields: dict
    confidence: Optional[float] = Field(None, ge=0, le=1)
    raw_response: Optional[str] = None


class InvoiceData(BaseModel):
    date: Optional[str] = None
    invoice_number: Optional[str] = None
    vendor: Optional[str] = None
    recipient: Optional[str] = None
    subtotal: Optional[float] = None
    tax: Optional[float] = None
    total: Optional[float] = None
    currency: str = "MXN"
    items: list[dict] = []


class ContractData(BaseModel):
    contract_type: Optional[str] = None
    parties: list[str] = []
    signing_date: Optional[str] = None
    effective_date: Optional[str] = None
    subject: Optional[str] = None
    amount: Optional[float] = None
    key_clauses: list[str] = []


class ReportData(BaseModel):
    title: Optional[str] = None
    author: Optional[str] = None
    date: Optional[str] = None
    sections: list[str] = []
    executive_summary: Optional[str] = None

Schema registry

EXTRACTION_SCHEMAS: dict[str, type[BaseModel]] = {
    "invoice": InvoiceData,
    "contract": ContractData,
    "manual": ReportData,
    "report": ReportData,
}

SCHEMA_PROMPTS: dict[str, str] = {
    "invoice": "date, invoice_number, vendor, recipient, subtotal, tax, total, currency, items (description, quantity, unit_price, line_total)",
    "contract": "contract_type, parties, signing_date, effective_date, subject, amount, key_clauses",
    "manual": "title, author, date, sections, executive_summary",
    "report": "title, author, date, sections, executive_summary",
}

DEFAULT_SCHEMA_PROMPT = "title, main_content, key_points, date (if present)"

Implementation: VisionAnalyzer

Complete class

import json
import logging
import os
from typing import Optional

from openai import OpenAI

logger = logging.getLogger(__name__)


class VisionAnalyzer:
    def __init__(self):
        self.openai_client = OpenAI()
        self.anthropic_client = None
        self.google_model = None
        self._init_fallback_providers()

    def _init_fallback_providers(self):
        try:
            import anthropic
            if os.getenv("ANTHROPIC_API_KEY"):
                self.anthropic_client = anthropic.Anthropic()
                logger.info("Anthropic available as fallback")
        except ImportError:
            logger.info("Anthropic not installed — no fallback")

        try:
            import google.generativeai as genai
            if os.getenv("GOOGLE_API_KEY"):
                genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
                self.google_model = genai.GenerativeModel("gemini-1.5-flash")
                logger.info("Google Gemini available as fallback")
        except ImportError:
            logger.info("Google GenAI not installed — no fallback")

    def classify(self, content) -> str:
        if content.has_image_pages:
            images = content.get_images_for_vision()
            return self._classify_from_image(images[0]["base64"])

        if content.full_text:
            return self._classify_from_text(content.full_text[:2000])

        return "other"

    def _classify_from_image(self, image_base64: str) -> str:
        prompt = (
            "Classify this document into one of these categories: "
            "invoice, contract, manual, report, other. "
            "Respond ONLY with the category name, no explanation."
        )

        def openai_call():
            r = self.openai_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "text", "text": prompt},
                        {"type": "image_url", "image_url": {
                            "url": f"data:image/png;base64,{image_base64}"
                        }}
                    ]
                }],
                max_tokens=20,
                temperature=0
            )
            return r.choices[0].message.content.strip().lower()

        def anthropic_call():
            if not self.anthropic_client:
                raise RuntimeError("Anthropic not available")
            r = self.anthropic_client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=20,
                messages=[{
                    "role": "user",
                    "content": [
                        {"type": "image", "source": {
                            "type": "base64", "media_type": "image/png",
                            "data": image_base64
                        }},
                        {"type": "text", "text": prompt}
                    ]
                }]
            )
            return r.content[0].text.strip().lower()

        return self._call_with_fallback([openai_call, anthropic_call], "classification")

    def _classify_from_text(self, text: str) -> str:
        r = self.openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": (
                    "Classify this document: invoice, contract, manual, report, other. "
                    f"Just the name.\n\n{text}"
                )
            }],
            max_tokens=20,
            temperature=0
        )
        return r.choices[0].message.content.strip().lower()

    def extract_structured(self, content, doc_type: str) -> ExtractionResult:
        schema_prompt = SCHEMA_PROMPTS.get(doc_type, DEFAULT_SCHEMA_PROMPT)

        if content.has_image_pages:
            images = content.get_images_for_vision()
            data = self._extract_from_images(images[:5], schema_prompt)
        elif content.full_text:
            data = self._extract_from_text(content.full_text[:4000], schema_prompt)
        else:
            return ExtractionResult(
                document_type=doc_type, fields={},
                confidence=0, raw_response="No content to extract"
            )

        return ExtractionResult(
            document_type=doc_type,
            fields=data,
            confidence=self._estimate_confidence(data, doc_type)
        )

    def _extract_from_images(self, images: list[dict], schema_prompt: str) -> dict:
        prompt = (
            f"Extract the following fields from this document: {schema_prompt}\n\n"
            "Respond ONLY with valid JSON. "
            "Use null for fields not found. "
            "For empty lists use []."
        )

        content_parts = [{"type": "text", "text": prompt}]
        for img in images:
            content_parts.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/png;base64,{img['base64']}"}
            })

        def openai_call():
            r = self.openai_client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": content_parts}],
                response_format={"type": "json_object"},
                temperature=0,
                max_tokens=2000
            )
            return json.loads(r.choices[0].message.content)

        def anthropic_call():
            if not self.anthropic_client:
                raise RuntimeError("Anthropic not available")
            ant_content = []
            for img in images:
                ant_content.append({
                    "type": "image", "source": {
                        "type": "base64", "media_type": "image/png",
                        "data": img["base64"]
                    }
                })
            ant_content.append({"type": "text", "text": prompt + "\nRespond only JSON."})

            r = self.anthropic_client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=2000,
                messages=[{"role": "user", "content": ant_content}]
            )
            return json.loads(r.content[0].text)

        return self._call_with_fallback([openai_call, anthropic_call], "structured extraction")

    def _extract_from_text(self, text: str, schema_prompt: str) -> dict:
        r = self.openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": (
                    f"Extract the following fields: {schema_prompt}\n\n"
                    "Respond ONLY with valid JSON. Use null for not found.\n\n"
                    f"{text}"
                )
            }],
            response_format={"type": "json_object"},
            temperature=0,
            max_tokens=2000
        )
        return json.loads(r.choices[0].message.content)

    def describe_for_rag(self, images: list[dict]) -> list[str]:
        descriptions = []
        for img in images[:10]:
            try:
                desc = self._describe_single_image(img["base64"])
                descriptions.append(f"[Page {img.get('page', '?')}] {desc}")
            except Exception as e:
                logger.warning(f"Error describing image on page {img.get('page', '?')}: {e}")
                descriptions.append(f"[Page {img.get('page', '?')}] Image not described due to an error.")
        return descriptions

    def _describe_single_image(self, image_base64: str) -> str:
        r = self.openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": (
                        "Describe the content of this document image in 2-3 sentences. "
                        "Include: document type, visible data, structure."
                    )},
                    {"type": "image_url", "image_url": {
                        "url": f"data:image/png;base64,{image_base64}"
                    }}
                ]
            }],
            max_tokens=200,
            temperature=0
        )
        return r.choices[0].message.content

    def _estimate_confidence(self, data: dict, doc_type: str) -> float:
        if not data:
            return 0.0

        expected_fields = {
            "invoice": ["date", "total", "vendor"],
            "contract": ["parties", "signing_date", "subject"],
            "manual": ["title", "sections"],
            "report": ["title", "sections"],
        }

        required = expected_fields.get(doc_type, [])
        if not required:
            return 0.5

        found = sum(1 for f in required if data.get(f) is not None)
        return round(found / len(required), 2)

    def _call_with_fallback(self, providers: list, operation: str):
        last_error = None
        for i, provider_fn in enumerate(providers):
            try:
                result = provider_fn()
                if i > 0:
                    logger.info(f"{operation}: success with fallback provider #{i}")
                return result
            except Exception as e:
                last_error = e
                logger.warning(f"{operation}: provider #{i} failed: {e}")
                continue

        raise RuntimeError(
            f"{operation}: all providers failed. Last error: {last_error}"
        )

Multi-Page Extraction

The problem

A 10-page document can have information distributed: the date is on page 1, the items on pages 2-8, and the total on page 9. Sending only the first image loses information.

Strategy

For documents with multiple images, we send up to 5 images in a single Vision call. If there are more than 5, we prioritize:

1. First page (header, general data)
2. Last page (totals, signatures)
3. Selected intermediate pages

def _select_pages_for_extraction(self, images: list[dict], max_pages: int = 5) -> list[dict]:
    if len(images) <= max_pages:
        return images

    selected = [images[0], images[-1]]

    remaining = images[1:-1]
    step = max(1, len(remaining) // (max_pages - 2))
    for i in range(0, len(remaining), step):
        if len(selected) >= max_pages:
            break
        selected.append(remaining[i])

    selected.sort(key=lambda x: x.get("page", 0))
    return selected

Multi-page cost

Multi-Provider Fallback in Detail

Why you need fallback

Order of preference

1. OpenAI GPT-4o       → Best general quality, most expensive
2. Anthropic Claude 3.5 → Comparable quality, good fallback
3. Google Gemini Flash  → Cheapest, good for classification

Fallback configuration

PROVIDER_CONFIG = {
    "openai": {
        "timeout": 30,
        "max_retries": 1,
        "models": {
            "vision": "gpt-4o",
            "classify": "gpt-4o-mini",
            "describe": "gpt-4o-mini"
        }
    },
    "anthropic": {
        "timeout": 30,
        "max_retries": 1,
        "models": {
            "vision": "claude-3-5-sonnet-20241022",
            "classify": "claude-3-5-sonnet-20241022"
        }
    },
    "google": {
        "timeout": 20,
        "max_retries": 1,
        "models": {
            "vision": "gemini-1.5-flash",
            "classify": "gemini-1.5-flash"
        }
    }
}

Troubleshooting

"Extraction returns empty or null fields"

Probable cause: The image is low resolution or the document has small text.

Solution: Increase the rendering resolution:

processor = DocumentProcessor(dpi_scale=300/72)  # 300 DPI instead of 150

Or use a more capable model:

# Switch from gpt-4o-mini to gpt-4o for extraction
r = self.openai_client.chat.completions.create(
    model="gpt-4o",  # more capable for complex documents
    ...
)

"The classifier returns 'other' for common documents"

Probable cause: The classification prompt needs more context.

Solution: Improve the prompt with examples:

CLASSIFICATION_PROMPT = """Classify this document into a category:
- invoice: billing documents with amounts, items, tax
- contract: legal agreements between parties with clauses
- manual: technical documentation with instructions
- report: reports with data, charts, conclusions
- other: any document that doesn't fit the previous ones

Respond ONLY with the category name."""

"Fallback to Anthropic fails with an image format error"

Probable cause: The image is JPEG but is sent as image/png.

Solution: Detect the real format:

import imghdr

def detect_media_type(image_base64: str) -> str:
    raw = base64.b64decode(image_base64[:100])
    img_type = imghdr.what(None, h=raw)
    media_types = {
        "jpeg": "image/jpeg",
        "png": "image/png",
        "webp": "image/webp",
        "gif": "image/gif"
    }
    return media_types.get(img_type, "image/png")

"Rate limit (429) when extracting from multiple documents"

Solution: Add client-side rate limiting:

import time

def extract_batch(self, documents: list, delay: float = 1.0) -> list[ExtractionResult]:
    results = []
    for doc in documents:
        result = self.extract_structured(doc, doc.get("type", "other"))
        results.append(result)
        time.sleep(delay)
    return results

Using the VisionAnalyzer

Complete example

processor = DocumentProcessor()
analyzer = VisionAnalyzer()

doc = processor.process("invoice_march.pdf")

doc_type = analyzer.classify(doc)
print(f"Document type: {doc_type}")

extraction = analyzer.extract_structured(doc, doc_type)
print(f"Confidence: {extraction.confidence}")
print(f"Extracted data:")
for key, value in extraction.fields.items():
    print(f"  {key}: {value}")

if doc.has_image_pages:
    descriptions = analyzer.describe_for_rag(doc.get_images_for_vision())
    print(f"\nDescriptions for RAG:")
    for desc in descriptions:
        print(f"  {desc}")

Expected output

Document type: invoice
Confidence: 1.0
Extracted data:
  date: 2025-03-15
  invoice_number: FAC-2025-0042
  vendor: Tech Solutions S.A.
  recipient: Company ABC
  subtotal: 1500.0
  tax: 240.0
  total: 1740.0
  currency: MXN
  items: [{'description': 'Software license', 'quantity': 1, ...}]

Descriptions for RAG:
  [Page 1] Commercial invoice from Tech Solutions S.A. with number FAC-2025-0042...

Exercises

Exercise 1: Dynamic schema by classification

Implement a complete flow that: (1) classifies the document, (2) selects the correct schema, (3) extracts data, (4) validates with the corresponding Pydantic model. If the classification is unknown, use a generic schema.

def analyze_document(content, analyzer: VisionAnalyzer) -> ExtractionResult:
    doc_type = analyzer.classify(content)
    logger.info(f"Document classified as: {doc_type}")

    extraction = analyzer.extract_structured(content, doc_type)

    model_class = EXTRACTION_SCHEMAS.get(doc_type)
    if model_class:
        try:
            validated = model_class(**extraction.fields)
            extraction.fields = validated.model_dump()
            logger.info(f"Data validated with {model_class.__name__}")
        except Exception as e:
            logger.warning(f"Validation failed: {e}. Using raw data.")
    else:
        logger.info(f"No specific schema for '{doc_type}', data without extra validation")

    return extraction


processor = DocumentProcessor()
analyzer = VisionAnalyzer()

doc = processor.process("document.pdf")
result = analyze_document(doc, analyzer)
print(f"Type: {result.document_type}")
print(f"Fields: {json.dumps(result.fields, indent=2, ensure_ascii=False)}")
print(f"Confidence: {result.confidence}")

Exercise 2: Complete fallback with Google Gemini

Extend _extract_from_images to include Google Gemini as a third fallback provider. Gemini uses a different API: it accepts images as PIL.Image or as bytes with upload_file. Implement the Gemini function and add it to the fallback chain.

import base64
from PIL import Image
import io


def gemini_extract(self, images: list[dict], schema_prompt: str) -> dict:
    if not self.google_model:
        raise RuntimeError("Google Gemini not available")

    pil_images = []
    for img in images[:5]:
        raw = base64.b64decode(img["base64"])
        pil_images.append(Image.open(io.BytesIO(raw)))

    prompt = (
        f"Extract the following fields from this document: {schema_prompt}\n"
        "Respond ONLY with valid JSON. Use null for not found."
    )

    content_parts = pil_images + [prompt]

    response = self.google_model.generate_content(
        content_parts,
        generation_config={"temperature": 0, "max_output_tokens": 2000}
    )

    text = response.text
    if text.startswith("```"):
        text = text.split("\n", 1)[1].rsplit("```", 1)[0]

    return json.loads(text)


def _extract_from_images_with_gemini(self, images, schema_prompt):
    def openai_call():
        return self._extract_from_images_openai(images, schema_prompt)

    def anthropic_call():
        return self._extract_from_images_anthropic(images, schema_prompt)

    def google_call():
        return gemini_extract(self, images, schema_prompt)

    return self._call_with_fallback(
        [openai_call, anthropic_call, google_call],
        "structured extraction"
    )

Summary

• The VisionAnalyzer turns document images into structured data.
• Three main functions: classify the type, extract data by schema, describe for RAG.
• Multi-provider fallback: OpenAI → Anthropic → Google, automatic and transparent.
• The extraction schemas vary by document type (invoice, contract, manual).
• Multi-page extraction selects the most informative pages to optimize costs.
• Confidence is estimated by comparing found fields vs expected fields.
• Troubleshooting: image resolution, classification prompts, image format, rate limits.

Additional Resources

1. OpenAI Vision Guide — GPT-4 Vision
2. Anthropic Vision Docs — Claude Vision
3. Google Gemini Vision — Gemini multimodal
4. Module 2 of this guide — Vision foundation