Articles / LLMs & Agents
Structured Data Extraction with LLMs, Python and Pydantic
Extract invoices and job postings with Python, Pydantic and Instructor. Validate types, review failures and measure field-level accuracy.
An invoice in plain text needs fields your application can process: vendor, date, currency and amounts. This tutorial uses Python, Pydantic and Instructor to define those fields and validate an LLM response.
Validation checks structure and the rules you write. It does not prove that an amount or date matches the document; that requires source evidence and reviewed examples.
The Problem with Naive Extraction
The obvious first attempt is to ask an LLM to "extract the invoice date and total amount" and parse its response. This breaks in three ways:
- The LLM returns slightly different field names each time
- Dates come back in inconsistent formats (
Jan 5,2025-01-05,05/01/25) - There is no schema validation: a field can be a string when you expected a float
The solution is to force the LLM to return JSON that matches a Pydantic schema, and validate it before it ever enters your pipeline.
Setup
We will use the instructor library, which patches any OpenAI-compatible client to return validated Pydantic objects:
pip install instructor openai pydantic
import instructor
from openai import OpenAI
client = instructor.from_openai(OpenAI())
If you prefer to use Anthropic's Claude:
pip install instructor anthropic
import instructor
import anthropic
client = instructor.from_anthropic(anthropic.Anthropic())
The examples below use the OpenAI client. For Anthropic, adapt the model, creation method and parameters using the Instructor integration guide.
Example 1: Extracting Invoices
from pydantic import BaseModel, Field
from typing import Optional
from datetime import date
from decimal import Decimal
class LineItem(BaseModel):
description: str
quantity: int
unit_price: Decimal
total: Decimal
class Invoice(BaseModel):
invoice_number: str
vendor_name: str
issue_date: date
due_date: Optional[date] = None
line_items: list[LineItem]
subtotal: Decimal
tax_rate: Optional[float] = Field(None, ge=0, le=1)
total_amount: Decimal
currency: str = 'USD'
INVOICE_TEXT = """
INVOICE #INV-2025-0847
Vendor: Acme Software Solutions
Date: March 12, 2025
Due: April 11, 2025
Services:
- Data Pipeline Development (40 hrs @ $150/hr): $6,000.00
- Cloud Infrastructure Setup (1 unit @ $800): $800.00
- Support & Documentation (8 hrs @ $100/hr): $800.00
Subtotal: $7,600.00
Tax (8%): $608.00
TOTAL DUE: $8,208.00
"""
invoice = client.chat.completions.create(
model='gpt-4o-mini',
response_model=Invoice,
messages=[
{
'role': 'user',
'content': f'Extract all invoice data from the following text:\n\n{INVOICE_TEXT}',
}
],
)
print(f"Invoice: {invoice.invoice_number}")
print(f"Vendor: {invoice.vendor_name}")
print(f"Date: {invoice.issue_date}") # datetime.date(2025, 3, 12): always a date object
print(f"Total: {invoice.total_amount}") # Decimal('8208.00')
print(f"Items: {len(invoice.line_items)}")
for item in invoice.line_items:
print(f" - {item.description}: ${item.total}")
Pydantic validates every field: issue_date is always a Python date object, total_amount is always a Decimal, and if the LLM hallucinates a field that does not match the schema, Instructor retries automatically.
Example 2: Job Postings
from enum import Enum
from typing import Optional
class ExperienceLevel(str, Enum):
JUNIOR = 'junior'
MID = 'mid'
SENIOR = 'senior'
LEAD = 'lead'
PRINCIPAL = 'principal'
class SalaryRange(BaseModel):
min_usd: Optional[int] = None
max_usd: Optional[int] = None
is_disclosed: bool
class JobPosting(BaseModel):
title: str
company: str
location: str
remote_policy: str # 'remote', 'hybrid', 'on-site'
experience_level: ExperienceLevel
required_skills: list[str]
preferred_skills: list[str]
salary: SalaryRange
responsibilities: list[str] = Field(max_length=8)
JOB_TEXT = """
Senior ML Engineer: Stealth AI Startup (Remote / NYC)
We're looking for a Senior Machine Learning Engineer to join our small but mighty team.
You'll own the full ML lifecycle: data collection, model training, evaluation, and deployment.
What we need:
5+ years of experience in ML/Data Science
Strong Python skills (scikit-learn, PyTorch or TensorFlow)
Production experience with model serving (FastAPI, Docker, Kubernetes)
Familiarity with MLflow or similar experiment tracking
Nice to have:
LLM fine-tuning experience
Rust (our inference engine is written in Rust)
Responsibilities:
Design and train models for our core recommendation system
Set up and maintain our ML infrastructure on AWS
Collaborate with product to define ML-driven features
Mentor junior engineers
Compensation: USD 180,000 to USD 240,000 + equity. We don't believe in hiding salaries.
"""
job = client.chat.completions.create(
model='gpt-4o-mini',
response_model=JobPosting,
messages=[{'role': 'user', 'content': f'Extract job posting data:\n\n{JOB_TEXT}'}],
)
print(f"Role: {job.title} ({job.experience_level.value})")
print(f"Company: {job.company} | {job.remote_policy}")
print(f"Salary: USD {job.salary.min_usd:,} to USD {job.salary.max_usd:,}")
print(f"Required: {', '.join(job.required_skills)}")
Example 3: Batch Processing with Validation Stats
In production you process hundreds of documents, usually as one step inside an orchestrated pipeline (I cover that setup in orchestrating ML pipelines with Prefect). At that scale you need to know your extraction success rate:
from pydantic import ValidationError
class ExtractionResult(BaseModel):
success: bool
data: Optional[Invoice] = None
error: Optional[str] = None
raw_text_preview: str
def extract_invoice_safe(text: str) -> ExtractionResult:
try:
invoice = client.chat.completions.create(
model='gpt-4o-mini',
response_model=Invoice,
max_retries=2,
messages=[{
'role': 'user',
'content': f'Extract invoice data:\n\n{text}',
}],
)
return ExtractionResult(success=True, data=invoice, raw_text_preview=text[:100])
except Exception as e:
return ExtractionResult(success=False, error=str(e), raw_text_preview=text[:100])
# Process a batch
invoice_texts = [INVOICE_TEXT, "Not an invoice at all", INVOICE_TEXT]
results = [extract_invoice_safe(t) for t in invoice_texts]
successes = [r for r in results if r.success]
failures = [r for r in results if not r.success]
print(f"Success rate: {len(successes)}/{len(results)} ({len(successes)/len(results)*100:.0f}%)")
for f in failures:
print(f" Failed: {f.error}")
Handling Complex Nested Structures
For documents with conditional or complex structures, add validation logic directly in Pydantic:
from pydantic import field_validator, model_validator
from datetime import datetime
class Invoice(BaseModel):
# ... fields as before ...
@field_validator('issue_date', 'due_date', mode='before')
@classmethod
def parse_flexible_date(cls, v):
if v is None:
return v
if isinstance(v, date):
return v
# Handle common LLM date formats
for fmt in ('%Y-%m-%d', '%B %d, %Y', '%m/%d/%Y', '%d/%m/%Y'):
try:
return datetime.strptime(str(v), fmt).date()
except ValueError:
continue
raise ValueError(f"Cannot parse date: {v}")
@model_validator(mode='after')
def validate_totals(self) -> 'Invoice':
computed = sum(item.total for item in self.line_items)
if abs(computed - self.subtotal) > Decimal('0.10'):
raise ValueError(
f"Line items sum ({computed}) does not match subtotal ({self.subtotal})"
)
return self
The model_validator catches cases where the LLM extracts individual line items that do not add up to the stated subtotal: a common source of errors in real invoice data.
Measure cost and quality before automating
Calculate cost from input and output tokens for every attempt and the provider's current rates. Include retries, OCR, storage and human review. Cost per document depends on length, model choice and failure rate; a fixed estimate without those conditions is not a useful budget.
Compare models on documents with reviewed field labels. Measure field-level accuracy, missing values, invented values and the share sent for review. Keep the evaluation set fixed when changing the prompt or schema. Version the model and dependencies as well.
Handle missing information explicitly
Distinguish an absent field from a real zero or an empty string. If the document does not contain a date, allow a missing value in the schema and retain the reason. Do not ask the model to fill gaps with plausible data. For amounts, compare currency, units and totals with the original text before writing to the destination system.
These examples demonstrate integration patterns, not guaranteed accuracy or a pipeline for every invoice format. Calls need a provider key and incur usage charges. Start with synthetic documents or remove sensitive data before sending them.
For questions over a document collection, continue with RAG with LangChain. To execute tools over extracted data, see agents with LangGraph.
References
Are you building an AI or data platform?
If you need to adapt this approach to your system, tell me about your data and the technical decision you need to resolve.