How to Build an AI Agent for Invoice Processing

Subscribe to the accounts-payable inbox and fire the pipeline on every new email that has an attachment. The Gmail and Outlook push APIs give you real-time events; a poller that checks every few minutes is the simpler fallback. Normalise everything to a job the moment it arrives so the handler returns fast and the heavy work happens in the queue. Uploads and vendor-portal exports feed the exact same queue, so there is one path to maintain. Capture the raw file and a content hash up front: the hash is what later stops the same invoice being processed twice.

# Watch the AP inbox; enqueue each attachment, return fast
@gmail.on_message(label="ap", has_attachment=True)
def on_invoice(msg):
    for att in msg.attachments:           # pdf, scan, or image
        raw = att.download()
        jobs.enqueue("process_invoice", {
            "file": raw,
            "hash": sha256(raw),          # used for dedup later
            "source": msg.sender,
        })

Not everything in an AP inbox is an invoice. Before extraction, a quick classification step sorts invoices from credit notes, statements, reminders, and spam, and routes non-invoices out. This one cheap step keeps your extraction accurate and your costs down, because you are not running a full extraction on a marketing email. Tag the vendor here too, so later steps can apply vendor-specific rules and memory.

kind = llm.classify(ocr_preview, labels=[
    "invoice", "credit_note", "statement", "reminder", "other",
])
if kind != "invoice":
    route_elsewhere(kind)                 # not our job, stop here

A PDF or photo is just an image, so OCR first converts it to text and layout (Document AI, Textract, or LlamaParse). Then an LLM maps that text into a typed schema, not free prose. Forcing structured output against a schema is what makes the result usable downstream and is the single biggest quality lever. Every field carries a confidence; anything low-confidence is flagged rather than trusted. Define the schema once and reuse it everywhere.

from pydantic import BaseModel, Field
from datetime import date

class LineItem(BaseModel):
    description: str
    quantity: float
    unit_price: float
    amount: float

class Invoice(BaseModel):
    vendor: str
    invoice_no: str
    invoice_date: date
    po_number: str | None = Field(None, description="PO number if present")
    currency: str = "INR"
    tax: float = 0
    total: float
    line_items: list[LineItem]

# Force the model to fill exactly this shape
def extract(ocr_text: str) -> Invoice:
    return llm.parse(ocr_text, response_format=Invoice)

Now check what you read against your own records. Re-add the line items and compare to the stated total to catch arithmetic and OCR errors. Look up the PO and the goods receipt by PO number through your ERP API (or a nightly export) and compare totals and quantities within a tolerance you set. Verify the vendor against your master, and confirm this invoice number has not already been paid. When invoice, PO, and receipt agree, that is the three-way match, and it is deterministic comparison code, so it is testable and auditable, unlike asking the model to decide.

def three_way_match(inv: Invoice, po, grn, tolerance=100) -> dict:
    return {
        "math":     abs(sum(li.amount for li in inv.line_items) - inv.total) < 1,
        "po_total": abs(inv.total - po.total) <= tolerance,
        "receipt":  all(grn.qty[li.description] >= li.quantity
                        for li in inv.line_items),
        "vendor":   vendor_master.is_approved(inv.vendor),
        "not_dup":  not ledger.exists(inv.vendor, inv.invoice_no),
    }

Turn business rules into explicit conditions you own. Clean, in-policy invoices auto-approve; everything else routes to a human with the exact failing check named, so the reviewer knows why in one glance. The model never decides on its own here, it applies your policy, and every decision is logged with the rule that fired. Keep the thresholds in config, not code, so finance can change them without a deploy.

def decide(inv: Invoice, checks: dict) -> tuple[str, list]:
    failed = [name for name, ok in checks.items() if not ok]
    if not failed and inv.total < policy.auto_approve_limit:
        return "auto_approve", []
    return "exception", failed            # e.g. ["po_total"] -> PO mismatch

Exceptions are a first-class feature, not an afterthought, because they are where money is saved or lost. A flagged invoice lands on a review surface (a web console, a sheet, or a Slack message with buttons) showing the invoice, the extracted fields, and the specific failing check. The reviewer approves or corrects in seconds. Two things make this strong: a retry budget that re-extracts with a stricter prompt before bothering a human, and storing every correction keyed to the vendor so the agent stops repeating that mistake. Low-confidence fields and unseen vendors should default to review until they have a track record.

def on_exception(inv, failed):
    if inv.retries < 2 and "math" in failed:
        return reextract(inv, stricter=True)   # try again before a human
    review_queue.add(inv, reason=failed,
                     fields=inv.dict(), confidence=inv.confidence)

# Learn from the fix so it does not recur for this vendor
on_correction(lambda fix: memory.save(inv.vendor, fix))

On approval, write the bill through your accounting tool's API and attach the source PDF and the decision log, so finance can defend every entry. Wrap the agent's actions as typed tools (here, a LangChain StructuredTool) so the orchestrator can call them safely with validated arguments. Where a tool has no API, emit a validated import file instead of re-keying. After posting, reconcile the payment back against the invoice so the loop closes.

from langchain_core.tools import StructuredTool

post_bill = StructuredTool.from_function(
    name="post_bill_to_erp",
    description="Create a bill in the ERP after approval; returns bill id.",
    args_schema=Invoice,
    func=lambda inv: erp.create_bill(            # Tally / Zoho / QuickBooks
        vendor=inv.vendor, total=inv.total,
        line_items=inv.line_items, attachment=inv.source_file,
    ),
)

Production is the hard part. Block duplicates on vendor plus invoice number plus the content hash before anything posts. Keep an immutable, append-only audit log of every step, who or what did it, and when, so finance and auditors can trust it. Track extraction accuracy, auto-approval rate, and exception rate on a dashboard, and alert when accuracy slips. Watch for fraud signals such as a vendor's bank details changing between invoices. Feed corrections back so accuracy climbs over time. This is the difference between a weekend demo and something a finance team will sign off on.

if ledger.exists(inv.vendor, inv.invoice_no) or ledger.seen(inv.hash):
    return flag("duplicate")              # never pay the same bill twice
if vendor_master.bank_changed(inv.vendor, inv.bank_account):
    return flag("bank_detail_change")     # classic fraud vector -> hold

audit.append(run)                          # tamper-proof, who/what/when
metrics.track(extraction_accuracy, auto_approve_rate, exception_rate)
alert_if(extraction_accuracy < 0.95)