Building an AI agent expense report auditing system cuts manual review time from hours to minutes, but only if you architect it around strict policy enforcement and verifiable extraction. This guide walks through deploying a practical AI agent expense report auditing pipeline that ingests receipts, flags policy violations, and routes edge cases to humans.
Step 1: Define audit policy as executable rules
Before calling any model, codify what “non-compliant” means. Finance teams already have a policy PDF; translate it into a version-controlled schema. The model should never invent limits.
from dataclasses import dataclass
from enum import Enum
from typing import FrozenSet
class Category(str, Enum):
MEALS = "meals"
TRAVEL = "travel"
SOFTWARE = "software"
OTHER = "other"
@dataclass
class Policy:
max_meal_per_day: float = 75.0
allowed_categories: FrozenSet[Category] = frozenset({
Category.MEALS, Category.TRAVEL, Category.SOFTWARE
})
require_itemized_receipt: bool = True
duplicate_window_days: int = 30Store this as policy.py and import it everywhere. The AI agent expense report auditing logic will call these values through tools, not guess them from the prompt. If policy changes, you update one file and redeploy—no prompt engineering drift.
Why not prompt-only rules
Prompts are unstable across model versions. A $75 limit buried in system text will be ignored when the model is distracted by a vague receipt. Encode numbers in code and expose them via function calls.
Step 2: Ingest and normalize documents
Most expense reports arrive as PDFs, scanned images, or CSV exports from a card provider. Use a local parser to pull raw text; don’t trust the LLM to OCR at scale.
import pdfplumber
import hashlib
import json
import os
def extract_text(path: str) -> str:
with pdfplumber.open(path) as pdf:
return "\n".join(page.extract_text() or "" for page in pdf.pages)
def normalize(report_path: str, employee_id: str) -> dict:
text = extract_text(report_path)
return {
"employee_id": employee_id,
"doc_hash": hashlib.sha256(open(report_path,'rb').read()).hexdigest(),
"raw_text": text,
"received_at": os.path.getmtime(report_path)
}Write each normalized report to a staging directory as JSON. For a prototype, a filesystem queue is enough; in production use S3 + SQS or a database outbox.
If you also receive photos, run them through tesseract or a vision model first, then feed the same raw_text shape. Keep the ingestion layer dumb.
Step 3: Extract line items with structured generation
Call an OpenAI-compatible chat endpoint with JSON response format. Set response_format={"type": "json_object"} and enforce a schema via system prompt.
import json
from openai import OpenAI
# Point at any OpenAI-compatible gateway. n4n.ai exposes one endpoint
# covering 240+ models with automatic fallback when a provider is degraded.
client = OpenAI(base_url="https://api.n4n.ai/v1", api_key="YOUR_KEY")
SYSTEM = """Extract expense line items from the report text. Return JSON:
{"items": [{"date": "YYYY-MM-DD", "vendor": str, "amount": float,
"category": "meals|travel|software|other", "has_receipt": bool}]}
If a field is missing, set has_receipt false and amount to 0.0."""
def extract_items(text: str) -> dict:
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role":"system","content":SYSTEM},
{"role":"user","content":text}],
response_format={"type":"json_object"},
temperature=0.0,
)
data = json.loads(resp.choices[0].message.content)
assert "items" in data, "schema violation"
return data["items"]Keep temperature at zero. Extraction is not a creative task. Validate the result with a pydantic model in production to catch silent schema drift.
Step 4: Build the auditing agent with tool calls
The AI agent expense report auditing core is a loop where the model decides whether to call policy tools. Define two functions: check_policy and flag_duplicate.
tools = [
{
"type": "function",
"function": {
"name": "check_policy",
"description": "Validate a line item against company policy",
"parameters": {
"type": "object",
"properties": {
"amount": {"type":"number"},
"category": {"type":"string"},
"has_receipt": {"type":"boolean"}
},
"required": ["amount","category","has_receipt"]
}
}
},
{
"type": "function",
"function": {
"name": "flag_duplicate",
"description": "Mark item as potential duplicate of earlier report",
"parameters": {
"type":"object",
"properties": {"vendor":{"type":"string"},"date":{"type":"string"}},
"required":["vendor","date"]
}
}
}
]Run a minimal agent loop with a hard cap on iterations:
def is_dup(vendor: str, date: str, seen: dict) -> bool:
return (vendor, date) in seen
def audit(items: list[dict], policy: Policy, seen: dict):
messages = [{"role":"system","content":"You are an expense audit agent. Use tools."},
{"role":"user","content":json.dumps(items)}]
for _ in range(5):
resp = client.chat.completions.create(
model="gpt-4o-mini", messages=messages, tools=tools, tool_choice="auto")
msg = resp.choices[0].message
if not msg.tool_calls:
return msg.content
messages.append(msg)
for call in msg.tool_calls:
args = json.loads(call.function.arguments)
if call.function.name == "check_policy":
ok = (args["category"] in policy.allowed_categories
and args["amount"] <= policy.max_meal_per_day
and (not policy.require_itemized_receipt or args["has_receipt"]))
result = "compliant" if ok else "violation"
else:
result = "duplicate_found" if is_dup(args["vendor"], args["date"], seen) else "unique"
messages.append({"role":"tool","tool_call_id":call.id,"content":result})
return "agent_loop_exhausted"The loop terminates when the model returns a final summary instead of a tool call. The five-iteration cap prevents runaway token spend on malformed reports.
Binding policy correctly
Note the check_policy tool compares against the Policy instance, not the prompt. This is the linchpin of reliable AI agent expense report auditing: the model proposes, the code disposes.
Step 5: Route exceptions to a human queue
Never auto-reject an expense. Write violations to an append-only log and a review bucket.
import json, datetime
def queue_for_review(employee_id: str, item: dict, reason: str):
record = {"ts": datetime.datetime.utcnow().isoformat(),
"employee": employee_id, "item": item, "reason": reason,
"status": "pending"}
with open("review_queue.jsonl","a") as f:
f.write(json.dumps(record)+"\n")A reviewer loads review_queue.jsonl in a simple UI or even jq. The AI agent expense report auditing system reduces their workload to confirming flags, not reading every receipt. Make the queue idempotent: key on doc_hash + item index so a re-run doesn’t create duplicate review tasks.
Step 6: Meter usage and log decisions
Every audit call consumes tokens. Capture resp.usage and ship it to your cost dashboard. Gateways like n4n.ai provide per-token usage metering so finance can attribute cost per audit run.
def log_usage(resp, run_id: str):
u = resp.usage
with open("token_metrics.jsonl","a") as f:
f.write(json.dumps({"run": run_id,
"prompt_tokens": u.prompt_tokens,
"completion_tokens": u.completion_tokens})+'\n')Store metrics in a time-series DB or CSV. Without metering, the AI agent expense report auditing project will blow up the LLM line item in the FinOps spreadsheet. Correlate run_id with the employee and report hash to answer “how much did auditing cost per department” later.
Verify success
Run the pipeline against a labeled set of 50 historical reports where you already know the violations. Expect the agent to flag at least the same violations a human flagged, with zero silent passes on require_itemized_receipt breaches. Check review_queue.jsonl length equals expected exception count. If the agent calls check_policy but returns “compliant” on a $200 meal, your policy binding is wrong—fix the tool, not the model.
After that, wire a daily cron to scan the expenses bucket. The audit agent runs, humans clear the queue in minutes, and finance gets a clean log with per-token cost attached. That is a defensible, auditable AI agent expense report auditing deployment—not a black box.