Invoice Processing Agent: OCR, Data Extraction, and Accounting System Integration

The Invoice Processing Bottleneck

Accounts payable teams manually process thousands of invoices monthly. Each invoice arrives in a different format — PDF attachments, scanned images, even photographs of paper documents. A clerk must open each one, find the vendor name, invoice number, dates, line items, and totals, then enter them into the accounting system. This manual process takes 10 to 15 minutes per invoice and introduces errors. An AI agent reduces this to seconds per invoice with higher accuracy.

This guide builds a complete invoice processing agent that handles OCR for scanned documents, extracts structured data using vision-capable LLMs, validates extracted fields, and pushes results to accounting systems.

Extracting Text from Invoice Documents

Invoices arrive as native PDFs (with embedded text) or scanned images. We handle both:

from pathlib import Path
from dataclasses import dataclass, field
import base64

@dataclass
class InvoiceDocument:
    file_path: str
    text_content: str
    images: list[str] = field(default_factory=list)  # base64-encoded page images
    source_type: str = ""  # "native_pdf", "scanned_pdf", "image"

def process_invoice_file(file_path: str) -> InvoiceDocument:
    """Extract text and images from an invoice file."""
    path = Path(file_path)
    ext = path.suffix.lower()

    if ext == ".pdf":
        return _process_pdf(file_path)
    elif ext in (".png", ".jpg", ".jpeg", ".tiff", ".bmp"):
        return _process_image(file_path)
    else:
        raise ValueError(f"Unsupported file type: {ext}")

def _process_pdf(file_path: str) -> InvoiceDocument:
    """Process a PDF invoice — handle both native and scanned."""
    import pymupdf

    doc = pymupdf.open(file_path)
    text = ""
    images = []

    for page in doc:
        page_text = page.get_text()
        text += page_text + "\n"

        # Render page as image for vision model fallback
        pix = page.get_pixmap(dpi=200)
        img_bytes = pix.tobytes("png")
        images.append(base64.b64encode(img_bytes).decode())

    doc.close()
    source_type = "native_pdf" if len(text.strip()) > 50 else "scanned_pdf"
    return InvoiceDocument(
        file_path=file_path,
        text_content=text.strip(),
        images=images,
        source_type=source_type,
    )

def _process_image(file_path: str) -> InvoiceDocument:
    """Process an image invoice."""
    with open(file_path, "rb") as f:
        img_b64 = base64.b64encode(f.read()).decode()
    return InvoiceDocument(
        file_path=file_path,
        text_content="",
        images=[img_b64],
        source_type="image",
    )

Extracting Structured Data with Vision LLMs

For scanned documents or when text extraction produces poor results, we use a vision-capable LLM to read the invoice image directly:

from openai import OpenAI
import json

client = OpenAI()

@dataclass
class InvoiceData:
    vendor_name: str
    vendor_address: str
    invoice_number: str
    invoice_date: str
    due_date: str
    currency: str
    subtotal: float
    tax_amount: float
    total_amount: float
    line_items: list[dict]
    payment_terms: str
    po_number: str
    confidence: float

def extract_invoice_data(doc: InvoiceDocument) -> InvoiceData:
    """Extract structured invoice data using LLM."""
    messages = [
        {
            "role": "system",
            "content": (
                "You extract structured data from invoices. Return JSON with:\n"
                "vendor_name, vendor_address, invoice_number, invoice_date (YYYY-MM-DD), "
                "due_date (YYYY-MM-DD), currency (ISO code), subtotal (float), "
                "tax_amount (float), total_amount (float), "
                "line_items (list of {description, quantity, unit_price, amount}), "
                "payment_terms, po_number, confidence (0-1).\n"
                "Use 0.0 for missing numeric fields. Use empty string for missing text fields."
            ),
        }
    ]

    if doc.text_content and doc.source_type == "native_pdf":
        messages.append({
            "role": "user",
            "content": f"Extract invoice data from this text:\n\n{doc.text_content[:4000]}",
        })
    elif doc.images:
        messages.append({
            "role": "user",
            "content": [
                {"type": "text", "text": "Extract all invoice data from this image."},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/png;base64,{doc.images[0]}",
                        "detail": "high",
                    },
                },
            ],
        })

    response = client.chat.completions.create(
        model="gpt-4o",
        temperature=0,
        response_format={"type": "json_object"},
        messages=messages,
    )
    data = json.loads(response.choices[0].message.content)

    return InvoiceData(
        vendor_name=data.get("vendor_name", ""),
        vendor_address=data.get("vendor_address", ""),
        invoice_number=data.get("invoice_number", ""),
        invoice_date=data.get("invoice_date", ""),
        due_date=data.get("due_date", ""),
        currency=data.get("currency", "USD"),
        subtotal=float(data.get("subtotal", 0)),
        tax_amount=float(data.get("tax_amount", 0)),
        total_amount=float(data.get("total_amount", 0)),
        line_items=data.get("line_items", []),
        payment_terms=data.get("payment_terms", ""),
        po_number=data.get("po_number", ""),
        confidence=float(data.get("confidence", 0)),
    )

Validating Extracted Data

Validation catches extraction errors before they propagate to the accounting system. We check mathematical consistency and required fields:

@dataclass
class ValidationResult:
    is_valid: bool
    errors: list[str]
    warnings: list[str]

def validate_invoice(data: InvoiceData) -> ValidationResult:
    """Validate extracted invoice data for consistency."""
    errors = []
    warnings = []

    # Required fields
    if not data.vendor_name:
        errors.append("Missing vendor name")
    if not data.invoice_number:
        errors.append("Missing invoice number")
    if not data.invoice_date:
        errors.append("Missing invoice date")
    if data.total_amount <= 0:
        errors.append("Total amount must be positive")

    # Line item math validation
    if data.line_items:
        computed_subtotal = sum(
            float(item.get("amount", 0)) for item in data.line_items
        )
        if abs(computed_subtotal - data.subtotal) > 0.02:
            warnings.append(
                f"Line items sum ({computed_subtotal:.2f}) does not match "
                f"subtotal ({data.subtotal:.2f})"
            )

    # Tax + subtotal = total validation
    computed_total = data.subtotal + data.tax_amount
    if abs(computed_total - data.total_amount) > 0.02:
        warnings.append(
            f"Subtotal + tax ({computed_total:.2f}) does not match "
            f"total ({data.total_amount:.2f})"
        )

    # Date validation
    from datetime import datetime
    try:
        inv_date = datetime.strptime(data.invoice_date, "%Y-%m-%d")
        if data.due_date:
            due_date = datetime.strptime(data.due_date, "%Y-%m-%d")
            if due_date < inv_date:
                warnings.append("Due date is before invoice date")
    except ValueError:
        errors.append("Invalid date format")

    return ValidationResult(
        is_valid=len(errors) == 0,
        errors=errors,
        warnings=warnings,
    )

Integrating with Accounting Systems

The agent pushes validated invoices to the accounting system. Here we demonstrate integration with a REST-based accounting API pattern common across QuickBooks, Xero, and similar systems:

import httpx
import logging

logger = logging.getLogger("invoice_agent")

class AccountingClient:
    def __init__(self, base_url: str, api_key: str):
        self.client = httpx.Client(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30,
        )

    def create_bill(self, invoice: InvoiceData) -> str:
        """Create a bill/payable in the accounting system."""
        payload = {
            "vendor_name": invoice.vendor_name,
            "invoice_number": invoice.invoice_number,
            "invoice_date": invoice.invoice_date,
            "due_date": invoice.due_date,
            "currency": invoice.currency,
            "line_items": [
                {
                    "description": item["description"],
                    "quantity": float(item.get("quantity", 1)),
                    "unit_price": float(item.get("unit_price", 0)),
                    "amount": float(item.get("amount", 0)),
                }
                for item in invoice.line_items
            ],
            "tax_amount": invoice.tax_amount,
            "total_amount": invoice.total_amount,
        }
        response = self.client.post("/api/bills", json=payload)
        response.raise_for_status()
        bill_id = response.json()["id"]
        logger.info(f"Created bill {bill_id} for invoice {invoice.invoice_number}")
        return bill_id

Orchestrating the Full Pipeline

The main processing function ties together extraction, validation, and submission:

def process_invoice(file_path: str, accounting: AccountingClient) -> dict:
    """Complete invoice processing pipeline."""
    doc = process_invoice_file(file_path)
    data = extract_invoice_data(doc)
    validation = validate_invoice(data)

    result = {
        "file": file_path,
        "vendor": data.vendor_name,
        "invoice_number": data.invoice_number,
        "total": data.total_amount,
        "validation": {"valid": validation.is_valid, "errors": validation.errors},
    }

    if not validation.is_valid:
        logger.warning(f"Validation failed for {file_path}: {validation.errors}")
        result["status"] = "needs_review"
        return result

    if data.confidence < 0.8:
        logger.warning(f"Low confidence ({data.confidence}) for {file_path}")
        result["status"] = "needs_review"
        return result

    bill_id = accounting.create_bill(data)
    result["status"] = "processed"
    result["bill_id"] = bill_id
    return result

FAQ

How do I handle invoices in languages other than English?

Vision-capable LLMs like GPT-4o handle multilingual invoices well. Specify the expected language in the system prompt, or let the model detect it automatically. For OCR-based approaches, Tesseract supports over 100 languages via language packs. The key is to keep the extraction prompt language-agnostic by asking for structured fields rather than specific text patterns.

What accuracy should I expect from automated invoice extraction?

Modern vision LLMs achieve 90 to 95 percent field-level accuracy on well-formatted invoices. Accuracy drops for handwritten invoices, very low resolution scans, or unconventional layouts. The validation step catches most extraction errors. Set a confidence threshold of 0.85 and route low-confidence invoices to human review.

How do I prevent duplicate invoice processing?

Maintain a processed invoice registry keyed by vendor name plus invoice number. Before processing, check if the invoice already exists in the registry. For partial matches (same vendor, similar amount, close dates), flag the invoice for manual deduplication review rather than rejecting it outright.

---

#InvoiceProcessing #AIAgents #OCR #DataExtraction #Accounting #Python #AgenticAI #LearnAI #AIEngineering