Structured Data Extraction from Text with LLMs
Use Python, Pydantic, and LLMs to extract structured data from unstructured text. Turn invoices and job postings into typed, validated Python objects.
Data pipelines assume structured inputs. Reality delivers PDFs, emails, contracts, and free-form text. The gap between the two is where data engineering projects die.
LLMs change this equation. With the right prompting and schema definition, you can turn any text document into a validated Python object — reliably enough to use in production pipelines.
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 rest of the code is identical regardless of the underlying model.
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: $180,000 – $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: ${job.salary.min_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.
Cost and Latency Considerations
Structured extraction with gpt-4o-mini costs roughly 0.001 per document depending on length. For high-volume pipelines:
- Use
gpt-4o-minifor most extractions (fast, cheap, accurate) - Reserve
gpt-4ofor complex edge cases that mini fails on - Cache responses for identical documents
- Batch requests when possible (reduces per-request overhead)
# Approximate cost estimate
COST_PER_1K_INPUT_TOKENS = 0.00015 # gpt-4o-mini input
COST_PER_1K_OUTPUT_TOKENS = 0.0006 # gpt-4o-mini output
avg_input_tokens = 500 # typical invoice
avg_output_tokens = 200 # JSON output
cost_per_doc = (avg_input_tokens / 1000) * COST_PER_1K_INPUT_TOKENS \
+ (avg_output_tokens / 1000) * COST_PER_1K_OUTPUT_TOKENS
print(f"~${cost_per_doc:.5f} per document")
print(f"~${cost_per_doc * 10_000:.2f} for 10,000 documents")
# ~$0.00020 per document
# ~$1.95 for 10,000 documents
Conclusion
The combination of Pydantic + Instructor transforms LLMs from creative text generators into reliable data extraction engines. The key insight is that schema definition is not just about structure — it is about encoding your domain knowledge (what dates look like, what fields are required, what invariants must hold) in a form the LLM is forced to respect. The result is extraction pipelines you can actually trust in production.
One closing distinction: extraction shines when the output is a fixed schema per document. If you need to answer open questions over a document collection, that is a RAG problem; and if you need multi-step computation over the data once it is structured, you want an agent with tools.
About the author
Rodrigo Arenas is a software architect and machine learning engineer in Medellín, Colombia. He builds products, platforms, and open-source software for AI, Data Engineering, and Machine Learning — creator of Ciaren, sklearn-genetic-opt, and PyWorkforce.
Recommended reading
Rodrigo Arenas — ML Engineer & Open Source Maintainer
Working on something similar? I can help you take it to production.