Building a Multimodal Document Analysis System with Gemini API — Processing Images, PDFs, and Videos in a Unified Architecture

Setup and context

Business documents come in far more shapes than plain text. Scanned invoices, PDF contracts, recorded meeting videos — extracting the information you need from these diverse formats has traditionally been a time-consuming, fragmented process.

With Gemini API's multimodal capabilities, you can process images, PDFs, and videos through a single API. This guide walks you through the design, implementation, and production deployment of a multimodal document analysis system.

Understanding Gemini API's Multimodal Input

Supported File Formats

Gemini API natively supports the following formats:

Images: JPEG, PNG, GIF, WebP, HEIC, HEIF
PDFs: Multi-page support (up to 3,600 pages)
Videos: MP4, MPEG, AVI, FLV, MKV, MOV, WebM, and more
Audio: MP3, WAV, FLAC, AAC, OGG, and more

What's particularly noteworthy is native PDF understanding. There's no need to run OCR separately — PDFs containing both text and images can be processed in a single API call.

Two File Input Methods

Gemini API offers two input methods depending on file size:

Inline data (under 20MB): Base64-encode the file and include it in the request body. Ideal for small files.

import google.generativeai as genai
import base64
from pathlib import Path
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro")
 
# Send an image inline
image_data = Path("receipt.jpg").read_bytes()
response = model.generate_content([
    "Extract the date, store name, and total amount from this receipt. "
    "Return the result in JSON format.",
    {
        "mime_type": "image/jpeg",
        "data": base64.b64encode(image_data).decode("utf-8")
    }
])
print(response.text)

File API (over 20MB): Upload the file to Google's servers and reference it by URI. Used for large files like PDFs and videos.

import google.generativeai as genai
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro")
 
# Upload a PDF via the File API
uploaded_file = genai.upload_file(
    path="contract.pdf",
    display_name="Contract",
    mime_type="application/pdf"
)
 
# Wait for upload processing to complete
import time
while uploaded_file.state.name == "PROCESSING":
    time.sleep(2)
    uploaded_file = genai.get_file(uploaded_file.name)
 
# Analyze the PDF contents
response = model.generate_content([
    "Extract the following information from this contract in JSON format: "
    "contracting parties, contract date, contract period, "
    "key terms, and termination conditions",
    uploaded_file
])
print(response.text)

Designing the Multimodal Analysis Pipeline

Architecture Overview

A practical document analysis system consists of four stages:

File intake: Format detection and preprocessing of uploaded files
Analysis execution: Requests to Gemini API (with format-appropriate prompt selection)
Structured output: Retrieving structured data using JSON Schema
Post-processing: Validation, database storage, notifications

Implementing the Unified Analyzer

import google.generativeai as genai
from pathlib import Path
from dataclasses import dataclass
from enum import Enum
import json
import mimetypes
 
class FileType(Enum):
    IMAGE = "image"
    PDF = "pdf"
    VIDEO = "video"
    AUDIO = "audio"
 
@dataclass
class AnalysisResult:
    file_name: str
    file_type: FileType
    extracted_data: dict
    confidence: float
    raw_response: str
 
class MultimodalAnalyzer:
    """Multimodal document analysis engine"""
 
    # Format-specific system prompts
    PROMPTS = {
        FileType.IMAGE: (
            "Analyze this image in detail. "
            "Extract all text, numbers, tables, and charts, "
            "and return the results in structured JSON format."
        ),
        FileType.PDF: (
            "Analyze all pages of this PDF document. "
            "Extract the title, section headings, body text, tables, "
            "and figure captions in structured JSON format."
        ),
        FileType.VIDEO: (
            "Analyze the contents of this video. "
            "Return key scenes, text overlays, and speech summaries "
            "in JSON format with timestamps."
        ),
    }
 
    def __init__(self, api_key: str, model_name: str = "gemini-2.5-pro"):
        genai.configure(api_key=api_key)
        self.model = genai.GenerativeModel(
            model_name,
            generation_config={
                "response_mime_type": "application/json",
                "temperature": 0.1,  # Prioritize analysis accuracy
            }
        )
 
    def detect_file_type(self, file_path: str) -> FileType:
        """Auto-detect file format"""
        mime_type, _ = mimetypes.guess_type(file_path)
        if mime_type:
            if mime_type.startswith("image/"):
                return FileType.IMAGE
            elif mime_type == "application/pdf":
                return FileType.PDF
            elif mime_type.startswith("video/"):
                return FileType.VIDEO
            elif mime_type.startswith("audio/"):
                return FileType.AUDIO
        raise ValueError(f"Unsupported file format: {file_path}")
 
    def analyze(
        self,
        file_path: str,
        custom_prompt: str | None = None
    ) -> AnalysisResult:
        """Analyze a file and return structured data"""
        path = Path(file_path)
        file_type = self.detect_file_type(file_path)
        file_size = path.stat().st_size
 
        # Choose upload method based on file size
        if file_size > 20 * 1024 * 1024:  # Over 20MB
            uploaded = genai.upload_file(path=str(path))
            import time
            while uploaded.state.name == "PROCESSING":
                time.sleep(2)
                uploaded = genai.get_file(uploaded.name)
            file_input = uploaded
        else:
            import base64
            mime_type, _ = mimetypes.guess_type(str(path))
            file_input = {
                "mime_type": mime_type,
                "data": base64.b64encode(
                    path.read_bytes()
                ).decode("utf-8")
            }
 
        # Select prompt
        prompt = custom_prompt or self.PROMPTS.get(
            file_type, self.PROMPTS[FileType.IMAGE]
        )
 
        # API call
        response = self.model.generate_content([prompt, file_input])
        raw_text = response.text
 
        # Parse JSON
        try:
            extracted = json.loads(raw_text)
        except json.JSONDecodeError:
            extracted = {"raw_text": raw_text}
 
        return AnalysisResult(
            file_name=path.name,
            file_type=file_type,
            extracted_data=extracted,
            confidence=0.95,  # Example value
            raw_response=raw_text,
        )

Improving Accuracy with Structured Output

Leveraging JSON Schema

By combining response_mime_type and response_schema, you can strictly control the JSON structure of the output.

import google.generativeai as genai
from google.generativeai import types
 
# Schema definition for invoice analysis
invoice_schema = types.Schema(
    type=types.Type.OBJECT,
    properties={
        "invoice_number": types.Schema(type=types.Type.STRING),
        "date": types.Schema(type=types.Type.STRING),
        "vendor": types.Schema(
            type=types.Type.OBJECT,
            properties={
                "name": types.Schema(type=types.Type.STRING),
                "address": types.Schema(type=types.Type.STRING),
            },
        ),
        "items": types.Schema(
            type=types.Type.ARRAY,
            items=types.Schema(
                type=types.Type.OBJECT,
                properties={
                    "description": types.Schema(type=types.Type.STRING),
                    "quantity": types.Schema(type=types.Type.INTEGER),
                    "unit_price": types.Schema(type=types.Type.NUMBER),
                    "amount": types.Schema(type=types.Type.NUMBER),
                },
            ),
        ),
        "subtotal": types.Schema(type=types.Type.NUMBER),
        "tax": types.Schema(type=types.Type.NUMBER),
        "total": types.Schema(type=types.Type.NUMBER),
    },
)
 
model = genai.GenerativeModel(
    "gemini-2.5-pro",
    generation_config={
        "response_mime_type": "application/json",
        "response_schema": invoice_schema,
    },
)
 
# Analyze an invoice image
response = model.generate_content([
    "Extract the contents of this invoice accurately.",
    uploaded_invoice_image,
])
 
# Type-safe JSON is returned
invoice_data = json.loads(response.text)
print(f"Total: ${invoice_data['total']:,.2f}")

Processing Multiple Files Simultaneously

Gemini API can process multiple files in a single request. You can submit related documents together for cross-document analysis.

# Batch process multiple invoices
files = [
    genai.upload_file("invoice_2026_01.pdf"),
    genai.upload_file("invoice_2026_02.pdf"),
    genai.upload_file("invoice_2026_03.pdf"),
]
 
response = model.generate_content([
    "Compare and analyze these 3 invoices. "
    "Show the monthly total trends, common expense categories, "
    "and flag any anomalies.",
    *files,
])

Building a Batch Processing Pipeline

Implementing Async Batch Processing

When processing large volumes of files, async processing with retry mechanisms is essential.

import asyncio
from typing import List
from dataclasses import dataclass
 
@dataclass
class BatchResult:
    total: int
    succeeded: int
    failed: int
    results: List[AnalysisResult]
    errors: List[dict]
 
class BatchProcessor:
    """Batch analysis for large file volumes"""
 
    def __init__(
        self,
        analyzer: MultimodalAnalyzer,
        max_concurrent: int = 5,
        max_retries: int = 3,
    ):
        self.analyzer = analyzer
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.max_retries = max_retries
 
    async def process_file(
        self, file_path: str
    ) -> tuple[AnalysisResult | None, dict | None]:
        """Process a single file with retry logic"""
        async with self.semaphore:
            for attempt in range(self.max_retries):
                try:
                    # Run synchronous API in async context
                    result = await asyncio.to_thread(
                        self.analyzer.analyze, file_path
                    )
                    return result, None
                except Exception as e:
                    if attempt < self.max_retries - 1:
                        # Exponential backoff
                        wait = 2 ** attempt
                        await asyncio.sleep(wait)
                    else:
                        return None, {
                            "file": file_path,
                            "error": str(e),
                            "attempts": self.max_retries,
                        }
 
    async def process_batch(
        self, file_paths: List[str]
    ) -> BatchResult:
        """Process multiple files concurrently"""
        tasks = [
            self.process_file(path) for path in file_paths
        ]
        outcomes = await asyncio.gather(*tasks)
 
        results = []
        errors = []
        for result, error in outcomes:
            if result:
                results.append(result)
            if error:
                errors.append(error)
 
        return BatchResult(
            total=len(file_paths),
            succeeded=len(results),
            failed=len(errors),
            results=results,
            errors=errors,
        )

Usage Example

import asyncio
 
analyzer = MultimodalAnalyzer(api_key="YOUR_GEMINI_API_KEY")
processor = BatchProcessor(analyzer, max_concurrent=3)
 
# Batch process all PDFs in a folder
import glob
pdf_files = glob.glob("invoices/*.pdf")
 
batch_result = asyncio.run(processor.process_batch(pdf_files))
print(f"Complete: {batch_result.succeeded}/{batch_result.total} succeeded")
# Expected output: Complete: 48/50 succeeded
 
if batch_result.errors:
    print(f"Failed: {batch_result.failed}")
    for err in batch_result.errors:
        print(f"  - {err['file']}: {err['error']}")

Cost Optimization Techniques

Choosing the Right Model

Not every task needs Gemini 2.5 Pro. Switching models based on task complexity can dramatically reduce costs.

Gemini 2.5 Flash: Standard-format documents (invoices, receipts, business cards)
Gemini 2.5 Pro: Complex contract clause analysis, cross-document analysis
Gemini 2.5 Flash Lite: Simple text extraction, file format detection

Leveraging Context Caching

When processing many files with the same system prompt and schema, context caching can reduce input token costs by up to 75%.

from google.generativeai import caching
import datetime
 
# Create a cache (valid for 1 hour)
cache = caching.CachedContent.create(
    model="models/gemini-2.5-pro",
    display_name="invoice-analysis-cache",
    system_instruction=(
        "You are an expert invoice analyst. "
        "Extract data accurately from uploaded files "
        "and return results in the specified JSON format."
    ),
    ttl=datetime.timedelta(hours=1),
)
 
# Initialize model with cache
cached_model = genai.GenerativeModel.from_cached_content(cache)
 
# Subsequent requests don't charge for system prompt tokens
for pdf in pdf_files:
    response = cached_model.generate_content([
        "Analyze this invoice.",
        genai.upload_file(pdf)
    ])

Summary

Gemini API's multimodal capabilities provide a powerful foundation for unifying document analysis workflows that previously required separate tools. Processing images, PDFs, and videos through the same API, combined with structured output and context caching, lets you build production-quality document analysis pipelines efficiently.

Start with a single file format — invoices in PDF format, for example — and gradually expand to other formats. Leverage structured output production patterns, and combine them with RAG pipelines and prompt engineering techniques to build even more sophisticated analysis systems.