Structured Data Extraction: Pydantic, Instructor, and Outlines for Strict Type Safety
Early LLM extraction pipelines relied on praying that the model would output clean JSON, followed by brittle regular expressions, json.loads() try-catch blocks, and retry loops. In mission-critical production environments, this non-deterministic output format is completely unacceptable.
Modern structured output architectures solve this at the foundational level through two distinct methodologies:
- Grammar-Constrained Logit Masking (Outlines / llama.cpp / vLLM): Compiling a Pydantic model or JSON schema into a Deterministic Finite Automaton (DFA). During each generation step, the engine masks the vocabulary logits so that only tokens corresponding to valid grammar transitions can receive non-zero probabilities.
- Semantic Retries and Validation (Instructor): Equipping Pydantic schemas with custom field validators (e.g. asserting that a stock ticker exists or that a date is in the future) and feeding validation error traces back to the model for self-healing.
Production Implementation with Instructor and Pydantic v2
import instructor
from openai import OpenAI
from pydantic import BaseModel, Field, field_validator
from typing import List
# Patch OpenAI client with Instructor superpowers
client = instructor.from_openai(OpenAI())
class LineItem(BaseModel):
sku: str = Field(description="Stock Keeping Unit formatted as ABC-1234")
quantity: int = Field(gt=0, description="Quantity ordered, must be strictly positive")
unit_price: float = Field(gt=0.0, description="Unit price in USD")
@field_validator("sku")
@classmethod
def validate_sku_format(cls, v: str) -> str:
if not ("-" in v and len(v) >= 6):
raise ValueError(f"SKU '{v}' must contain a hyphen and be at least 6 characters.")
return v.upper()
class Invoice(BaseModel):
invoice_number: str
customer_email: str
items: List[LineItem]
total_amount: float
@field_validator("total_amount")
@classmethod
def validate_total(cls, v: float, info) -> float:
# Check computed sum against claimed total
items = info.data.get("items", [])
calculated = sum(item.quantity * item.unit_price for item in items)
if abs(v - calculated) > 0.05:
raise ValueError(f"Total {v} does not match sum of line items {calculated:.2f}")
return v
# Deterministic extraction with automated validation feedback loop
invoice = client.chat.completions.create(
model="gpt-4o",
response_model=Invoice,
max_retries=3, # Automatically re-prompts if field_validator fails!
messages=[
{"role": "user", "content": "Purchased 2 units of PRO-9942 at $49.50 and 1 unit of DEV-1100 at $120.00 for client acme@corp.com, Invoice #INV-8821."}
]
)
print(f"Extracted Invoice: {invoice.invoice_number} | Total: ${invoice.total_amount:.2f}")
