How to Build an AI Agent for Invoice Processing

import base64
from anthropic import Anthropic

client = Anthropic()   # ANTHROPIC_API_KEY

def classify(pdf: bytes) -> str:
    msg = client.messages.create(
        model="claude-opus-4-8", max_tokens=16,
        messages=[{"role": "user", "content": [
            {"type": "document", "source": {"type": "base64",
             "media_type": "application/pdf",
             "data": base64.standard_b64encode(pdf).decode()}},
            {"type": "text",
             "text": "Label: invoice, credit_note, statement, reminder, other."}]}])
    return msg.content[0].text.strip().lower()

import base64
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import HumanMessage

llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash")  # or ChatAnthropic(...)

def classify(pdf: bytes) -> str:
    msg = HumanMessage(content=[
        {"type": "text", "text": "Label: invoice, credit_note, statement, reminder, other."},
        {"type": "media", "mime_type": "application/pdf",
         "data": base64.b64encode(pdf).decode()}])
    return llm.invoke([msg]).content.strip().lower()

import base64
from anthropic import Anthropic

client = Anthropic()
# Invoice is the same Pydantic schema as the Gemini tab.

def extract(pdf: bytes) -> Invoice:
    msg = client.messages.parse(
        model="claude-opus-4-8", max_tokens=4096,
        messages=[{"role": "user", "content": [
            {"type": "document", "source": {"type": "base64",
             "media_type": "application/pdf",
             "data": base64.standard_b64encode(pdf).decode()}},
            {"type": "text", "text": "Extract this invoice into the schema."}]}],
        output_config={"format": Invoice})   # validated, typed output
    return msg.parsed_output                 # a validated Invoice

import base64
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import HumanMessage

# Invoice is the same Pydantic schema as the Gemini tab.
extractor = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash").with_structured_output(Invoice)

def extract(pdf: bytes) -> Invoice:
    return extractor.invoke([HumanMessage(content=[
        {"type": "text", "text": "Extract this invoice into the schema."},
        {"type": "media", "mime_type": "application/pdf",
         "data": base64.b64encode(pdf).decode()}])])

def resolve_exception(pdf: bytes, po, grn, budget: int = 2):
    """Re-extract with a stricter prompt up to the budget, then hand off to a human."""
    for _ in range(budget):
        inv = extract(pdf)
        checks = three_way_match(inv.model_dump(), po, grn)
        if all(checks.values()):
            return inv, checks                 # clean -> done
    review_queue.add(inv, reason=[k for k, ok in checks.items() if not ok])
    return inv, checks                         # out of retries -> human

from langgraph.graph import StateGraph, START, END

def refine(state: dict) -> dict:               # re-extract + re-check
    inv = extract(state["pdf"])
    return {"invoice": inv, "tries": state["tries"] + 1,
            "checks": three_way_match(inv.model_dump(), state["po"], state["grn"])}

def route(state: dict) -> str:                 # loop until clean or out of tries
    return END if all(state["checks"].values()) or state["tries"] >= 2 else "refine"

g = StateGraph(dict)
g.add_node("refine", refine)
g.add_edge(START, "refine")
g.add_conditional_edges("refine", route)
exception_loop = g.compile()

post_bill = {   # JSON tool schema the model can call
    "name": "post_bill_to_erp",
    "description": "Create a bill in the ERP after approval; returns the bill id.",
    "input_schema": {"type": "object", "required": ["invoice", "attachment_id"],
        "properties": {"invoice": {"type": "object"},
                       "attachment_id": {"type": "string"}}}}

def run_post(invoice: dict, attachment_id: str) -> dict:   # your handler
    bill = erp.create_bill(vendor=invoice["vendor"], total=invoice["total"],
                           line_items=invoice["line_items"], attachment=attachment_id)
    return {"bill_id": bill["id"]}

from langchain_core.tools import tool

@tool
def post_bill_to_erp(invoice: dict, attachment_id: str) -> dict:
    """Create a bill in the ERP after approval; returns the new bill id."""
    bill = erp.create_bill(vendor=invoice["vendor"], total=invoice["total"],
                           line_items=invoice["line_items"], attachment=attachment_id)
    return {"bill_id": bill["id"]}

llm_with_tools = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash").bind_tools([post_bill_to_erp])

def run_ap(messages: list) -> "Message":
    """Drive the tool-use loop until the model has no more tools to call."""
    while True:
        resp = client.messages.create(model="claude-opus-4-8", max_tokens=4096,
                                       tools=[post_bill], messages=messages)
        if resp.stop_reason != "tool_use":
            return resp
        messages.append({"role": "assistant", "content": resp.content})
        for blk in resp.content:
            if blk.type == "tool_use":
                out = run_post(**blk.input)     # execute the tool
                messages.append({"role": "user", "content": [
                    {"type": "tool_result", "tool_use_id": blk.id, "content": str(out)}]})

from langgraph.graph import StateGraph, START, END

g = StateGraph(dict)
g.add_node("extract", lambda s: {"invoice": extract(s["pdf"])})
g.add_node("match",   lambda s: {"checks": three_way_match(
                          s["invoice"].model_dump(), s["po"], s["grn"])})
g.add_node("decide",  lambda s: {"decision": decide(s["invoice"].model_dump(), s["checks"])})
g.add_node("post",    lambda s: {"bill": post_bill_to_erp.invoke(
                          {"invoice": s["invoice"].model_dump(), "attachment_id": s["att"]})})
g.add_edge(START, "extract"); g.add_edge("extract", "match"); g.add_edge("match", "decide")
g.add_conditional_edges("decide",
    lambda s: "post" if s["decision"][0] == "auto_approve" else END)
g.add_edge("post", END)
ap_pipeline = g.compile()