LLM-powered data extraction from PDFs and scanned documents
Why traditional parsers fall short for document extraction
PDF parsing libraries like pdf-parse, pdfplumber, or Apache PDFBox work well when documents follow a predictable structure. Financial filings from a single provider, export files from a known SaaS tool, or programmatically-generated invoices all fall into this category. But the moment you deal with scanned paper documents, inconsistently formatted contracts, multi-column layouts, or forms where the field positions vary between versions, rule-based parsers break down quickly.
The core problem is that traditional extraction relies on positional assumptions. A parser that looks for a dollar amount 20 pixels to the right of the word "Total" fails when a different template moves that label. A regex that grabs dates after "Invoice Date:" fails when someone types "Date of Invoice:" instead. Maintaining a fleet of regex rules and positional heuristics for dozens of document types is expensive and fragile.
LLMs change the calculus here. Rather than describing precisely where a field lives on the page, you describe what the field means semantically, and the model does the rest. This approach is far more robust to layout variation, OCR noise, language differences, and novel document types it hasn't seen during any kind of training. The tradeoff is latency and cost, which this article addresses directly.
This guide walks through a complete pipeline: getting readable text from PDFs (including scanned ones), structuring LLM prompts for reliable extraction, parsing and validating the output, and deploying this in a way that doesn't burn your API budget.
Step 1: Getting text out of PDFs
Before an LLM can reason about a document, you need text. The approach differs depending on whether the PDF contains embedded text or is a scanned image.
Programmatic PDFs (text layer present)
When a PDF was generated by software rather than scanned, it typically contains an embedded text layer. You can extract this directly without OCR.
import pdfplumber
def extract_text_from_pdf(path: str) -> list[dict]:
"""Extract text per page with layout preservation."""
pages = []
with pdfplumber.open(path) as pdf:
for i, page in enumerate(pdf.pages):
text = page.extract_text(layout=True)
tables = page.extract_tables()
pages.append({
"page": i + 1,
"text": text or "",
"tables": tables or [],
})
return pages
The layout=True flag in pdfplumber attempts to preserve spatial relationships — columns stay as columns, tables are approximated. This matters because LLMs can use spatial structure to reason about relationships between labels and values.
For table-heavy documents, use extract_tables() to get structured grid data. Convert tables to Markdown or CSV strings before passing to the LLM — models handle tabular text much better than raw concatenated cell values.
def table_to_markdown(table: list[list[str | None]]) -> str:
if not table:
return ""
rows = []
for i, row in enumerate(table):
cells = [str(c or "").strip() for c in row]
rows.append("| " + " | ".join(cells) + " |")
if i == 0:
rows.append("|" + "---|" * len(row))
return "\n".join(rows)
Scanned PDFs (image-only, requires OCR)
Scanned documents have no embedded text — the PDF is just a container for images. You must run OCR before anything else.
Option A: Tesseract (open source, runs locally)
import pytesseract
from pdf2image import convert_from_path
from PIL import Image
def ocr_pdf(path: str, dpi: int = 300) -> list[dict]:
"""Convert PDF pages to images and OCR each one."""
images = convert_from_path(path, dpi=dpi)
pages = []
for i, img in enumerate(images):
text = pytesseract.image_to_string(img, config="--oem 3 --psm 6")
pages.append({"page": i + 1, "text": text})
return pages
DPI of 300 is the practical minimum for reliable OCR. 600 DPI gives better results on degraded documents but doubles processing time and memory.
Option B: Pass images directly to a vision-capable LLM
Modern LLMs including Claude support image inputs. For scanned documents with complex layouts where OCR would scramble the reading order, passing the page as an image directly can produce dramatically better results.
import anthropic
import base64
from pathlib import Path
from pdf2image import convert_from_path
import io
def encode_page_as_base64(img) -> str:
buffer = io.BytesIO()
img.save(buffer, format="PNG")
return base64.standard_b64encode(buffer.getvalue()).decode()
def extract_with_vision(pdf_path: str, schema: dict) -> dict:
client = anthropic.Anthropic()
images = convert_from_path(pdf_path, dpi=200)
content = []
for img in images:
content.append({
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": encode_page_as_base64(img),
}
})
content.append({
"type": "text",
"text": f"""Extract the following fields from this document and return valid JSON matching this schema:
{json.dumps(schema, indent=2)}
Return only the JSON object, no additional text."""
})
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=2048,
messages=[{"role": "user", "content": content}]
)
return json.loads(response.content[0].text)
The vision approach is particularly effective for:
- Forms with checkboxes, stamps, or handwritten fields
- Documents with non-standard reading order (right-to-left columns, footnotes mixed with body)
- Low-quality scans where Tesseract produces garbage
- Documents in languages with complex scripts
The cost tradeoff: image inputs consume more tokens than text. A single page image might cost 500–1500 input tokens versus 200–800 tokens for clean OCR text. For high-volume processing, prefer OCR where quality is acceptable and reserve vision for fallback cases.
Step 2: Designing extraction prompts
The quality of your extraction is mostly determined by your prompt design. LLMs are capable of impressive extraction, but they need clear instructions about what to extract, what to do when fields are missing, and how to format the output.
Define a schema and pass it in the prompt
Always give the model a concrete schema to fill in. This anchors the output format and makes the model's job easier than open-ended extraction.
const invoiceSchema = {
invoice_number: "string | null",
invoice_date: "ISO 8601 date string | null",
due_date: "ISO 8601 date string | null",
vendor: {
name: "string | null",
address: "string | null",
tax_id: "string | null",
},
line_items: [
{
description: "string",
quantity: "number | null",
unit_price: "number | null",
total: "number | null",
}
],
subtotal: "number | null",
tax: "number | null",
total_due: "number | null",
currency: "ISO 4217 currency code | null",
};
const systemPrompt = `You are a document data extraction assistant.
Extract structured data from the provided document text and return valid JSON.
Rules:
- Return ONLY valid JSON matching the provided schema. No markdown, no explanation.
- Use null for any field that is not present or cannot be determined with confidence.
- Normalize dates to ISO 8601 format (YYYY-MM-DD).
- Normalize monetary amounts to numbers (not strings). Remove currency symbols.
- For line items, include all items found. An empty array is valid if none are present.
- Do not infer values — only extract what is explicitly stated in the document.`;
const userPrompt = `Extract data from this invoice according to this schema:
${JSON.stringify(invoiceSchema, null, 2)}
Document text:
${documentText}`;
Handle multi-page documents
For long documents, you have two strategies:
Strategy 1: Concatenate all pages and extract once. Simpler but risks hitting context limits. Works well for documents under ~20 pages.
full_text = "\n\n--- PAGE BREAK ---\n\n".join(
page["text"] for page in pages
)
Strategy 2: Extract per page, then merge. Better for long documents. Extract independently per page, then run a merge pass.
def merge_extractions(results: list[dict]) -> dict:
"""Merge per-page extractions, preferring non-null values."""
merged = {}
for result in results:
for key, value in result.items():
if key not in merged or merged[key] is None:
merged[key] = value
elif isinstance(value, list):
merged[key] = merged.get(key, []) + value
return merged
The merge pass can also be done by the LLM itself, which handles conflicts more intelligently:
def llm_merge(partial_results: list[dict], schema: dict) -> dict:
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=2048,
system="Merge multiple partial extraction results into a single coherent result. "
"Resolve conflicts by choosing the most complete or specific value. "
"Return valid JSON only.",
messages=[{
"role": "user",
"content": f"Schema:\n{json.dumps(schema)}\n\n"
f"Partial results:\n{json.dumps(partial_results, indent=2)}"
}]
)
return json.loads(response.content[0].text)
Step 3: Parsing and validating model output
LLMs occasionally produce malformed JSON, especially when the document is complex or unusual. Never use bare json.loads() in production without validation.
Robust JSON parsing
import Anthropic from "@anthropic-ai/sdk";
import { z } from "zod";
const invoiceLineItemSchema = z.object({
description: z.string(),
quantity: z.number().nullable(),
unit_price: z.number().nullable(),
total: z.number().nullable(),
});
const invoiceSchema = z.object({
invoice_number: z.string().nullable(),
invoice_date: z.string().nullable(),
due_date: z.string().nullable(),
vendor: z.object({
name: z.string().nullable(),
address: z.string().nullable(),
tax_id: z.string().nullable(),
}),
line_items: z.array(invoiceLineItemSchema),
subtotal: z.number().nullable(),
tax: z.number().nullable(),
total_due: z.number().nullable(),
currency: z.string().nullable(),
});
type Invoice = z.infer<typeof invoiceSchema>;
function parseExtractionResult(raw: string): Invoice {
// Strip markdown code fences if the model wrapped the JSON
const cleaned = raw
.replace(/^```(?:json)?\s*/m, "")
.replace(/\s*```$/m, "")
.trim();
const parsed = JSON.parse(cleaned);
return invoiceSchema.parse(parsed);
}
Retry on parse failure with error feedback
When parsing fails, feed the error back to the model. This is surprisingly effective.
async function extractWithRetry(
text: string,
maxAttempts = 3
): Promise<Invoice> {
const client = new Anthropic();
let lastError: Error | null = null;
let lastRaw = "";
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
const messages: Anthropic.MessageParam[] = [
{
role: "user",
content: buildExtractionPrompt(text),
},
];
if (attempt > 1 && lastRaw) {
messages.push({ role: "assistant", content: lastRaw });
messages.push({
role: "user",
content: `Your previous response failed validation with this error: ${lastError?.message}.
Please return corrected, valid JSON only.`,
});
}
const response = await client.messages.create({
model: "claude-haiku-4-5",
max_tokens: 2048,
system: SYSTEM_PROMPT,
messages,
});
lastRaw = response.content[0].type === "text" ? response.content[0].text : "";
try {
return parseExtractionResult(lastRaw);
} catch (err) {
lastError = err as Error;
if (attempt === maxAttempts) throw lastError;
}
}
throw lastError!;
}
Step 4: Model selection and cost management
Extraction tasks span a wide quality/cost curve. Not every document needs the most capable model.
Route by document complexity
type DocumentComplexity = "simple" | "standard" | "complex";
function classifyComplexity(text: string, pageCount: number): DocumentComplexity {
if (pageCount > 10) return "complex";
if (text.length < 1000) return "simple";
// Heuristic: many different date formats, unusual characters, non-English = complex
const hasNonAscii = /[^\x00-\x7F]/.test(text);
const hasMultipleDateFormats = /\d{1,2}[\/\-\.]\d{1,2}[\/\-\.]\d{2,4}/.test(text);
if (hasNonAscii || hasMultipleDateFormats) return "complex";
return "standard";
}
function selectModel(complexity: DocumentComplexity): string {
switch (complexity) {
case "simple":
return "claude-haiku-4-5"; // cheapest, fast
case "standard":
return "claude-sonnet-4-5"; // balanced
case "complex":
return "claude-opus-4-5"; // most capable
}
}
Cache repeated extractions
If you process the same document more than once (e.g., reprocessing with updated schemas), cache the raw LLM output keyed on document hash + schema hash:
import hashlib
import json
from functools import lru_cache
def document_cache_key(text: str, schema: dict) -> str:
content = json.dumps({"text": text, "schema": schema}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()
Use prompt caching (available on Claude API with cache_control) when the schema and system prompt are long and repeated across many documents. This can reduce costs by 80–90% on the cached portion of the prompt.
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=2048,
system=[
{
"type": "text",
"text": LONG_SYSTEM_PROMPT_WITH_SCHEMA,
"cache_control": {"type": "ephemeral"},
}
],
messages=[{"role": "user", "content": document_text}]
)
Step 5: Production pipeline architecture
A complete production pipeline for document extraction at scale needs more than just API calls. Here's a practical architecture:
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ Document upload │────▶│ Preprocessing │────▶│ Text/image │
│ (S3/GCS/R2) │ │ (classify, OCR) │ │ extraction │
└─────────────────┘ └──────────────────┘ └────────┬────────┘
│
▼
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ Structured │◀────│ Validation & │◀────│ LLM extraction │
│ storage output │ │ schema check │ │ with retry │
└─────────────────┘ └──────────────────┘ └─────────────────┘
// Worker function for async document processing
async function processDocument(job: {
documentId: string;
storagePath: string;
documentType: "invoice" | "contract" | "receipt";
outputWebhook?: string;
}): Promise<ExtractionResult> {
// 1. Download document
const buffer = await downloadFromStorage(job.storagePath);
// 2. Detect if scanned or programmatic
const hasTextLayer = await detectTextLayer(buffer);
// 3. Extract text (or images if scanned)
const pages = hasTextLayer
? await extractPdfText(buffer)
: await ocrPdf(buffer);
// 4. Select extraction strategy based on document type
const schema = getSchemaForType(job.documentType);
const complexity = classifyComplexity(
pages.map((p) => p.text).join(" "),
pages.length
);
// 5. Run LLM extraction with retry
const extracted = await extractWithRetry(
pages.map((p) => p.text).join("\n\n--- PAGE BREAK ---\n\n"),
selectModel(complexity),
schema
);
// 6. Store results
await storeResult(job.documentId, extracted);
// 7. Notify via webhook if configured
if (job.outputWebhook) {
await fetch(job.outputWebhook, {
method: "POST",
body: JSON.stringify({ documentId: job.documentId, data: extracted }),
});
}
return { success: true, data: extracted };
}
Handling extraction failures gracefully
Not every document will extract cleanly. Build explicit handling for low-confidence results:
def assess_extraction_confidence(extracted: dict, schema: dict) -> float:
"""Return a 0-1 confidence score based on field coverage."""
required_fields = [k for k, v in schema.items() if "null" not in str(v)]
filled = sum(1 for f in required_fields if extracted.get(f) is not None)
return filled / len(required_fields) if required_fields else 1.0
async def process_with_confidence_check(
document_text: str,
schema: dict,
confidence_threshold: float = 0.7,
) -> tuple[dict, float]:
result = await extract_with_retry(document_text, schema)
confidence = assess_extraction_confidence(result, schema)
if confidence < confidence_threshold:
# Flag for human review rather than returning silently incomplete data
await queue_for_human_review(document_text, result, confidence)
return result, confidence
Practical tips and common pitfalls
Normalize before extraction. Fix obvious OCR errors before passing to the LLM. Common patterns: 0 vs O in IDs, 1 vs l in amounts, | characters from column separators. A quick regex cleanup step saves tokens and reduces hallucination risk.
Don't ask the model to compute values. If you need totals, subtotals, or derived fields, extract the raw values and compute them yourself. Models occasionally make arithmetic errors. Extract line_items with individual amounts, then compute total in code and cross-check against the extracted total.
Use page markers in concatenated text. When joining pages with --- PAGE BREAK ---, models handle cross-page fields (like signatures on page 3 that reference a date on page 1) better than if you naively concatenate.
Test against your worst documents. Your extraction quality is determined by the worst documents in your corpus, not the average ones. Collect edge cases — water-damaged scans, multi-language documents, forms with unusual layouts — and test against them explicitly.
Log raw LLM outputs permanently. Even after you've parsed and validated the result, log the raw model response. When a document extracts incorrectly and a human corrects it later, the raw output becomes training signal and a diagnostic tool.
Consider confidence thresholds by field, not document. A document might extract 95% of fields with high confidence but miss the tax_id because it wasn't visible. Flag individual null fields for targeted review rather than rejecting the entire extraction.
The combination of reliable text extraction, carefully designed prompts, output validation, and thoughtful cost routing produces a document extraction system that handles the variation found in real-world document corpora — far beyond what any rule-based parser can achieve.