Document Processing Agent
A shoebox of invoices, a folder of scanned receipts, a decade of PDFs someone needs in a spreadsheet by Friday. This used to be a data-entry job. A multimodal model does it in minutes, and does it well, right up until it reads a smudged 7 as a 1 and your books are off by six hundred dollars with total confidence. The model does not know when it's wrong. That's the entire design problem.
So the design has three parts. A schema, decided before you extract anything, because "pull out the important stuff" produces a different shape for every document and §5.1 already told you why that ruins everything downstream. A confidence score on every field, reported by the model itself. And a review queue: extractions below your threshold go to a human, and until you've audited the pipeline against documents you've checked by hand, the threshold is effectively "everything." Auto-accept is a privilege the pipeline earns, not a default it ships with.
This one is tagged advanced because it calls the model API directly instead of hiding behind a framework, and because the failure mode is quiet. A broken website looks broken. A wrong number in a spreadsheet looks exactly like a right number.
Prerequisites
- An Anthropic API key exported as `ANTHROPIC_API_KEY`. In the env, never in the file.
- `pip install anthropic`.
- An `inbox/` folder of PDFs, PNGs, or JPGs, and a handful you've already keyed in by hand for auditing.
"""
Document Processing Agent
- Extracts structured data from PDFs/images in inbox/ against a fixed
schema, using structured outputs so the JSON always parses.
- The model scores its own confidence per field.
- Low-confidence extractions land in a review queue for a human.
Start with the threshold high enough that EVERYTHING is reviewed.
"""
import base64
import json
import os
from pathlib import Path
from anthropic import Anthropic
INBOX = Path("inbox")
ACCEPTED = Path("out/accepted.jsonl")
REVIEW_QUEUE = Path("out/review_queue.jsonl")
# Until you have audited runs against hand-checked documents, keep this
# above 1.0 so every extraction is reviewed. Lower it deliberately.
AUTO_ACCEPT_THRESHOLD = 1.01
MEDIA_TYPES = {".pdf": "application/pdf", ".png": "image/png",
".jpg": "image/jpeg", ".jpeg": "image/jpeg"}
# The schema IS the product. Decide it first; change it rarely.
# Structured outputs require additionalProperties: false and required
# on every object. Every field pairs a value with a confidence, and
# value is nullable: a null beats a guess.
FIELD = {
"type": "object",
"properties": {
"value": {"type": ["string", "null"]},
"confidence": {
"type": "number",
"description": "0 to 1. Below 0.9 means a human should look.",
},
},
"required": ["value", "confidence"],
"additionalProperties": False,
}
SCHEMA = {
"type": "object",
"properties": {
"vendor_name": FIELD,
"document_date": FIELD, # ISO 8601 as a string
"total_amount": FIELD, # string, exactly as printed
"currency": FIELD,
"invoice_number": FIELD,
"notes": {
"type": ["string", "null"],
"description": "Anything odd: handwriting, stamps, damage.",
},
},
"required": ["vendor_name", "document_date", "total_amount",
"currency", "invoice_number", "notes"],
"additionalProperties": False,
}
PROMPT = """Extract the fields defined by the schema from this document.
Rules:
- Transcribe values exactly as printed. Do not normalize amounts.
- If a field is absent or illegible, set value to null and confidence
to 0. Never infer a value from context or from typical documents.
- Confidence reflects legibility and ambiguity, not your general
optimism. A smudged digit caps the field at 0.5."""
client = Anthropic()
def extract(path: Path) -> dict:
media_type = MEDIA_TYPES[path.suffix.lower()]
data = base64.standard_b64encode(path.read_bytes()).decode()
block_type = "document" if media_type == "application/pdf" else "image"
response = client.messages.create(
model="claude-opus-5",
max_tokens=4096, # extraction output is small and bounded
output_config={"format": {"type": "json_schema", "schema": SCHEMA}},
messages=[{
"role": "user",
"content": [
{"type": block_type,
"source": {"type": "base64", "media_type": media_type,
"data": data}},
{"type": "text", "text": PROMPT},
],
}],
)
if response.stop_reason != "end_turn":
# refusal / max_tokens: do not trust partial JSON. Queue it.
return {"_extraction_failed": response.stop_reason}
return json.loads(response.content[0].text)
def route(filename: str, record: dict) -> None:
row = {"file": filename, **record}
fields = [v for v in record.values()
if isinstance(v, dict) and "confidence" in v]
worst = min((f["confidence"] for f in fields), default=0.0)
ok = "_extraction_failed" not in record and worst >= AUTO_ACCEPT_THRESHOLD
dest = ACCEPTED if ok else REVIEW_QUEUE
dest.parent.mkdir(parents=True, exist_ok=True)
with open(dest, "a") as f:
f.write(json.dumps(row) + "\n")
print(f"{filename}: worst confidence {worst:.2f} -> {dest.name}")
if __name__ == "__main__":
for path in sorted(INBOX.iterdir()):
if path.suffix.lower() in MEDIA_TYPES:
route(path.name, extract(path))Adaptation notes:
- Change the schema to your documents: purchase orders, lab reports, timesheets. Keep the value-plus-confidence shape per field and keep values nullable. The schema is the contract everything downstream depends on; treat changes to it like migrations.
- The routing key is the worst field, not the average. One unreadable total on an otherwise clean invoice is exactly the document a human must see.
- Build the review step as a loop through
review_queue.jsonlnext to the original file. Corrected rows are gold: they tell you which fields the model actually struggles with before you ever lower the threshold. - Financial figures stay strings, exactly as printed. Parse to numbers in your own code, where a failure throws an error instead of silently rounding.
- These documents contain vendor names, amounts, and sometimes personal data. Check your API provider's data retention terms before feeding it real records, and keep
inbox/andout/out of the repo (§6.5). - The mistake people make: lowering the threshold after ten clean documents. Ten is an anecdote. Audit a real sample against hand-keyed truth, per field, and lower the threshold only for the fields that earned it.