Invoice Processing with Vision Language Models

import json
import base64
from openai import OpenAI
from pydantic import BaseModel

client = OpenAI()

# Define structured output schema
class LineItem(BaseModel):
    description: str
    quantity: float
    unit_price: float
    total: float

class Vendor(BaseModel):
    name: str
    address: str | None
    tax_id: str | None

class InvoiceData(BaseModel):
    invoice_number: str
    date: str
    vendor: Vendor
    line_items: list[LineItem]
    subtotal: float
    tax_amount: float
    total: float
    currency: str
    confidence: float

def extract_invoice(image_path: str) -> InvoiceData:
    """Extract structured data from an invoice using GPT-5.2."""
    with open(image_path, 'rb') as f:
        image_data = base64.b64encode(f.read()).decode()

    # GPT-5.2 uses responses.create() with structured outputs
    response = client.responses.parse(
        model="gpt-5.2",
        input=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "input_text",
                        "text": """Extract all invoice data. Verify math:
sum(line_items.total) should equal subtotal.
Set confidence 0-1 based on image quality and extraction certainty."""
                    },
                    {
                        "type": "input_image",
                        "image_url": f"data:image/png;base64,{image_data}"
                    }
                ]
            }
        ],
        text_format=InvoiceData,
        max_output_tokens=2000
    )

    return response.output_parsed

# Usage
invoice_data = extract_invoice("invoice.png")
print(invoice_data.model_dump_json(indent=2))

# Validate extraction
if invoice_data.confidence < 0.9:
    print("Low confidence - manual review recommended")

import anthropic
import base64
import json
from pathlib import Path

client = anthropic.Anthropic()

def extract_invoice(image_path: str) -> dict:
    """Extract structured data from an invoice using Claude 3.5 Sonnet."""
    with open(image_path, 'rb') as f:
        image_data = base64.b64encode(f.read()).decode()

    # Determine media type
    suffix = Path(image_path).suffix.lower()
    media_types = {'.png': 'image/png', '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.pdf': 'application/pdf'}
    media_type = media_types.get(suffix, 'image/png')

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": media_type,
                        "data": image_data
                    }
                },
                {
                    "type": "text",
                    "text": """Analyze this invoice and extract all data as a JSON object.

Required fields:
- invoice_number (string)
- date (ISO 8601 format)
- vendor (object with name, address)
- line_items (array of objects with description, quantity, unit_price, total)
- subtotal (number)
- tax_amount (number)
- total (number)
- currency (3-letter code)

Reasoning process:
1. First identify all text regions
2. Categorize each region (header, line items, totals)
3. Extract and validate numerical values
4. Cross-check: sum of line items should equal subtotal

Return ONLY the JSON object, no explanation."""
                }
            ]
        }]
    )

    return json.loads(response.content[0].text)

# Multi-page invoice handling
def extract_multipage_invoice(pdf_path: str) -> dict:
    """Handle multi-page invoices using Claude's context window."""
    import fitz  # PyMuPDF

    doc = fitz.open(pdf_path)
    images = []

    for page in doc:
        pix = page.get_pixmap(dpi=150)
        img_bytes = pix.tobytes("png")
        images.append(base64.b64encode(img_bytes).decode())

    content = []
    for i, img_data in enumerate(images):
        content.append({
            "type": "image",
            "source": {"type": "base64", "media_type": "image/png", "data": img_data}
        })
        content.append({"type": "text", "text": f"Page {i+1} of {len(images)}"})

    content.append({
        "type": "text",
        "text": "This is a multi-page invoice. Extract all data across all pages into a single JSON object."
    })

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=4000,
        messages=[{"role": "user", "content": content}]
    )

    return json.loads(response.content[0].text)

import json
import google.generativeai as genai
from PIL import Image

genai.configure(api_key="YOUR_API_KEY")

def extract_invoice(image_path: str) -> dict:
    """Extract structured data from an invoice using Gemini 3 Pro."""
    # Gemini 3 Pro - best-in-class OCR (0.115 OmniDocBench)
    model = genai.GenerativeModel("gemini-3-pro")

    image = Image.open(image_path)

    prompt = """Extract invoice data as JSON with this schema:
{
  "invoice_number": string,
  "date": string,
  "vendor_name": string,
  "line_items": [{"description": string, "quantity": number, "unit_price": number, "total": number}],
  "subtotal": number,
  "tax_amount": number,
  "total": number,
  "currency": string
}

Use spatial reasoning to understand table structure. Handle merged headers.
Return ONLY valid JSON, no markdown code blocks."""

    response = model.generate_content([prompt, image])

    # Parse JSON (handle potential markdown wrapping)
    text = response.text.strip()
    if text.startswith('```'):
        text = text.split('\n', 1)[1].rsplit('\n', 1)[0]

    return json.loads(text)

# Batch processing with rate limiting
import time
from pathlib import Path

def process_batch(folder: str, output_file: str):
    """Process a folder of invoices with rate limiting."""
    results = []
    for i, path in enumerate(Path(folder).glob("*.png")):
        try:
            data = extract_invoice(str(path))
            data["_source_file"] = path.name
            results.append(data)
            print(f"Processed {i+1}: {path.name}")
        except Exception as e:
            results.append({"_source_file": path.name, "_error": str(e)})
        time.sleep(0.5)  # Rate limiting

    with open(output_file, 'w') as f:
        json.dump(results, f, indent=2)

process_batch("invoices/", "extracted_data.json")

import torch
from transformers import AutoModelForCausalLM, AutoProcessor
from PIL import Image
import json

# Load Qwen3-VL (highest DocVQA: 96.5%)
# Use Qwen2.5-VL-7B for smaller deployments
model_name = "Qwen/Qwen3-VL-72B-Instruct"
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    trust_remote_code=True
)
processor = AutoProcessor.from_pretrained(model_name, trust_remote_code=True)

def extract_invoice(image_path: str) -> dict:
    """Extract structured data from an invoice using Qwen3-VL."""
    image = Image.open(image_path).convert("RGB")

    prompt = """<|im_start|>user
<|vision_start|><|image_pad|><|vision_end|>
Extract all invoice data as JSON:
{
  "invoice_number": string,
  "date": string,
  "vendor_name": string,
  "line_items": [{"description": string, "quantity": number, "unit_price": number, "total": number}],
  "subtotal": number,
  "tax_amount": number,
  "total": number
}
<|im_end|>
<|im_start|>assistant
"""

    inputs = processor(
        text=prompt,
        images=image,
        return_tensors="pt"
    ).to(model.device)

    with torch.no_grad():
        output_ids = model.generate(
            **inputs,
            max_new_tokens=2000,
            do_sample=False
        )

    output = processor.decode(output_ids[0], skip_special_tokens=True)

    # Extract JSON from response
    json_start = output.find('{')
    json_end = output.rfind('}') + 1
    return json.loads(output[json_start:json_end])

# Production deployment with vLLM for higher throughput
# pip install vllm
from vllm import LLM, SamplingParams

def create_production_pipeline():
    """High-throughput pipeline using vLLM."""
    llm = LLM(
        model="Qwen/Qwen2-VL-7B-Instruct",
        tensor_parallel_size=2,  # Use 2 GPUs
        max_model_len=4096
    )
    sampling_params = SamplingParams(temperature=0, max_tokens=2000)
    return llm, sampling_params

def validate_invoice(data: dict) -> tuple[bool, list[str]]:
    """Validate extracted invoice data."""
    errors = []

    # Check required fields
    required = ['invoice_number', 'date', 'total', 'line_items']
    for field in required:
        if not data.get(field):
            errors.append(f"Missing required field: {field}")

    # Validate line item math
    if data.get('line_items') and data.get('subtotal'):
        calc_subtotal = sum(item.get('total', 0) for item in data['line_items'])
        if abs(calc_subtotal - data['subtotal']) > 0.01:
            errors.append(f"Line items ({calc_subtotal}) don't match subtotal ({data['subtotal']})")

    # Validate tax calculation
    if data.get('subtotal') and data.get('tax_amount') and data.get('total'):
        expected_total = data['subtotal'] + data['tax_amount']
        if abs(expected_total - data['total']) > 0.01:
            errors.append(f"Tax calculation mismatch")

    return len(errors) == 0, errors

def should_flag_for_review(data: dict, thresholds: dict = None) -> tuple[bool, str]:
    """Determine if invoice needs human review."""
    thresholds = thresholds or {
        'min_confidence': 0.85,
        'max_amount': 10000,
        'required_fields_ratio': 0.9,
    }

    # Model-reported confidence
    if data.get('confidence', 1.0) < thresholds['min_confidence']:
        return True, "Low model confidence"

    # High-value invoices always reviewed
    if data.get('total', 0) > thresholds['max_amount']:
        return True, "High value invoice"

    # Missing critical fields
    required = ['invoice_number', 'date', 'vendor', 'total']
    present = sum(1 for f in required if data.get(f))
    if present / len(required) < thresholds['required_fields_ratio']:
        return True, "Missing critical fields"

    return False, ""

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=60))
async def process_invoice_with_retry(image_path: str) -> dict:
    """Process single invoice with exponential backoff retry."""
    return await asyncio.to_thread(extract_invoice, image_path)

async def process_batch(image_paths: list[str], max_concurrent: int = 5) -> list[dict]:
    """Process multiple invoices with controlled concurrency."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def process_with_limit(path):
        async with semaphore:
            try:
                result = await process_invoice_with_retry(path)
                result['_status'] = 'success'
                result['_source'] = path
            except Exception as e:
                result = {'_status': 'failed', '_source': path, '_error': str(e)}
            return result

    tasks = [process_with_limit(p) for p in image_paths]
    return await asyncio.gather(*tasks)

# Usage
results = asyncio.run(process_batch(["inv1.png", "inv2.png", "inv3.png"]))
success = [r for r in results if r['_status'] == 'success']
failed = [r for r in results if r['_status'] == 'failed']