TL;DR:
- Document processing is one of the highest-ROI use cases for AI agents — invoices, contracts, and forms are everywhere and manual handling is expensive
- A well-designed pipeline separates extraction, classification, and routing into distinct agent steps, each with clear inputs and outputs
- Validation against a schema and confidence thresholds before routing prevents downstream errors from propagating through your systems
Every organisation processes documents — invoices, purchase orders, contracts, onboarding forms, compliance submissions, insurance claims. Before AI agents, this meant either manual data entry, brittle OCR pipelines, or expensive custom ML models trained on proprietary data.
LLMs have fundamentally changed this. A well-prompted agent can extract structured fields from a document, classify it by type, validate the output, and route it to the right system — all without training a custom model for each document format.
Here’s how to build a production-grade document processing pipeline.
The Three-Stage Pipeline
The most reliable architecture separates document processing into three distinct stages, each handled by an agent step with a clearly defined schema:
Stage 1 — Extraction: Pull structured fields from the raw document.
Stage 2 — Classification: Determine the document type, urgency, and destination.
Stage 3 — Routing: Send the extracted data to the appropriate downstream system with any required transformation.
This separation matters. A single agent prompt that tries to extract, classify, and route simultaneously produces worse results and is harder to debug when something goes wrong. Staging lets you inspect each step independently and add validation between them.
Stage 1: Extraction
The extraction agent receives the raw document (PDF text, image, or OCR output) and returns a structured JSON object.
from anthropic import Anthropic
import json
client = Anthropic()
INVOICE_SCHEMA = {
"vendor_name": "string",
"vendor_address": "string | null",
"invoice_number": "string",
"invoice_date": "ISO 8601 date string",
"due_date": "ISO 8601 date string | null",
"line_items": [{"description": "string", "quantity": "number", "unit_price": "number", "total": "number"}],
"subtotal": "number",
"tax_amount": "number | null",
"total_amount": "number",
"currency": "3-letter ISO 4217 code",
"payment_terms": "string | null"
}
def extract_invoice(document_text: str) -> dict:
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
messages=[{
"role": "user",
"content": f"""Extract all invoice fields from the document below.
Return a JSON object matching this schema exactly:
{json.dumps(INVOICE_SCHEMA, indent=2)}
For fields not present in the document, use null. Do not invent values.
Return only valid JSON, no explanation.
DOCUMENT:
{document_text}"""
}]
)
return json.loads(response.content[0].text)Two principles matter here: explicit schema definition in the prompt, and a strict instruction not to invent values. Hallucinated fields in financial documents cause real problems downstream.
Stage 2: Classification
After extraction, a classification step determines how to handle the document:
def classify_document(extracted_data: dict, raw_text: str) -> dict:
response = client.messages.create(
model="claude-haiku-4-5-20251001", # Faster/cheaper for classification
max_tokens=256,
messages=[{
"role": "user",
"content": f"""Classify this document and return JSON:
{{
"document_type": "invoice | purchase_order | credit_note | statement | other",
"priority": "high | normal | low",
"requires_approval": true | false,
"approval_threshold_exceeded": true | false,
"routing_destination": "accounts_payable | accounts_receivable | legal_review | manual_review"
}}
Rules:
- requires_approval = true if total_amount > 5000
- routing_destination = legal_review if payment_terms mentions unusual conditions
- routing_destination = manual_review if confidence is low
Extracted data: {json.dumps(extracted_data)}
Raw text excerpt: {raw_text[:500]}"""
}]
)
return json.loads(response.content[0].text)Use a faster, cheaper model for classification — it’s a simpler task and you’ll run it on every document. Save the more capable model for extraction where accuracy is critical.
Stage 3: Validation and Routing
Before routing, validate the extracted data against business rules:
def validate_and_route(extracted: dict, classification: dict) -> dict:
errors = []
# Basic sanity checks
if extracted.get("total_amount") != round(
extracted.get("subtotal", 0) + (extracted.get("tax_amount") or 0), 2
):
errors.append("Total does not match subtotal + tax")
if not extracted.get("invoice_number"):
errors.append("Missing invoice number")
if errors:
return {
"status": "validation_failed",
"errors": errors,
"destination": "manual_review"
}
# Route based on classification
destination = classification["routing_destination"]
routing_map = {
"accounts_payable": post_to_accounts_payable,
"legal_review": queue_for_legal_review,
"manual_review": flag_for_manual_processing,
}
routing_map[destination](extracted, classification)
return {"status": "routed", "destination": destination}This pattern — extract, validate, then route — prevents bad data from propagating into your accounting or ERP systems where it causes downstream failures that are expensive to unpick.
Handling Multi-Format Documents
Real document pipelines deal with PDFs, images, and Word documents. Use a preprocessing layer to normalise inputs before the extraction agent:
import anthropic
import base64
from pathlib import Path
def process_document(file_path: str) -> str:
path = Path(file_path)
if path.suffix.lower() in [".jpg", ".jpeg", ".png", ".webp"]:
# Use vision for images
with open(file_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode()
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=4096,
messages=[{
"role": "user",
"content": [
{
"type": "image",
"source": {"type": "base64", "media_type": "image/jpeg", "data": image_data}
},
{"type": "text", "text": "Extract all text from this document image, preserving structure and layout."}
]
}]
)
return response.content[0].text
elif path.suffix.lower() == ".pdf":
# Use a PDF library to extract text, then pass to LLM
import pypdf
reader = pypdf.PdfReader(file_path)
return "\n".join(page.extract_text() for page in reader.pages)
elif path.suffix.lower() in [".txt", ".md"]:
return path.read_text()
else:
raise ValueError(f"Unsupported file type: {path.suffix}")Confidence Scoring
For high-volume pipelines, add a confidence scoring step so you only send high-confidence extractions to automated routing:
def score_extraction_confidence(extracted: dict, raw_text: str) -> float:
"""
Simple heuristic: penalise null fields and cross-check amounts.
In production, use a trained classifier or a second LLM call.
"""
required_fields = ["vendor_name", "invoice_number", "invoice_date", "total_amount"]
null_penalty = sum(0.15 for f in required_fields if not extracted.get(f))
# Spot-check: does vendor name appear in raw text?
name_present = (extracted.get("vendor_name") or "").lower() in raw_text.lower()
name_penalty = 0.2 if not name_present else 0
return max(0.0, 1.0 - null_penalty - name_penalty)Route anything below 0.7 to manual review. It’s better to have a human handle 15% of documents than to let low-quality extractions pollute your data.
Where This Works Well
Document processing agents deliver the most value when:
- Document formats vary: You receive invoices from hundreds of different vendors with different layouts — a rule-based extractor would need hundreds of templates.
- Volume is high but not massive: For thousands of documents per day, LLM processing costs are justified. For millions, consider training a fine-tuned model on your LLM-generated extractions.
- Errors have meaningful costs: Manual correction time or downstream system errors make validation worth the effort.
The pattern scales. Start with invoices, prove ROI, then extend to contracts, onboarding forms, and compliance documents using the same three-stage architecture with document-specific schemas.