Structured Data Extraction with LLMs

Picture the pilot every legal-ops team ran in 2024: a paralegal manually extracts 22 fields from 1,000 contract pages while a GPT-4-class model with a hand-written system prompt processes the same stack. The paralegal takes four weeks. The model takes about 40 minutes and lands within a few points of the paralegal's field-level accuracy — close enough that the partners approve a production rollout.

Six months later, the system is silently inventing renewal dates for contracts that don't have them, because the model was instructed to "extract the renewal date" without being told what to do when none existed. Nobody notices for three billing cycles.

That gap — between "it works in the demo" and "it's trustworthy in production" — is what this article is about.

Why not regex or a fine-tuned NER model

The alternatives are worth taking seriously before defaulting to LLMs. Regex is deterministic and free. A fine-tuned NER model (BERT-based, SpaCy) is cheap to run, and it will never hallucinate a field it was trained to leave blank.

Both break on the same thing: variation. A regex that extracts "Total Amount: $X,XXX.XX" fails on "Grand Total ....... $X,XXX.XX" and on the Spanish version of the same form. A NER model trained on US invoices fails on UK purchase orders. Every new format variant requires a new rule or a retraining run.

LLMs generalize across formats because they've seen all of them. You describe what you want, not how to find it. That generalization is the core value proposition, and it's real — in practice, a well-prompted GPT-4o-class model handles format variation that would require dozens of regex branches.

The failure modes are just different. Regex never invents data. LLMs do, especially on fields with low information density in the document. The extraction pipeline needs to be designed around that.

The schema-first design pattern

The most important decision in an extraction pipeline happens before you write any API code: what is your target schema, and what types does each field actually need?

Define it as a Pydantic model. The model class serves three roles simultaneously: it's the contract you give to the LLM (through instructor or Pydantic AI), the runtime validator that checks what comes back, and the type annotation that downstream code can rely on.

from pydantic import BaseModel, Field
from typing import Optional
from datetime import date
from enum import Enum

class ContractType(str, Enum):
    MSA = "master_services_agreement"
    SOW = "statement_of_work"
    NDA = "non_disclosure_agreement"
    OTHER = "other"

class Party(BaseModel):
    name: str = Field(description="Legal entity name, exactly as written in the document")
    role: str = Field(description="Role of this party: 'client', 'vendor', 'licensor', 'licensee', etc.")
    jurisdiction: Optional[str] = Field(
        default=None,
        description="Governing jurisdiction if stated; null if not present in the document"
    )

class ContractExtraction(BaseModel):
    contract_type: ContractType = Field(
        description="Type of contract. Use 'other' if none of the standard types apply."
    )
    parties: list[Party] = Field(
        description="All named parties to the agreement, minimum two"
    )
    effective_date: Optional[date] = Field(
        default=None,
        description="The date the contract takes effect. ISO 8601 format. Null if not stated."
    )
    expiry_date: Optional[date] = Field(
        default=None,
        description="Expiry or termination date. Null if the contract has no fixed term."
    )
    total_value_usd: Optional[float] = Field(
        default=None,
        description="Total contract value in USD. Convert from other currencies if an exchange rate is stated; null if no value is mentioned."
    )
    auto_renewal: bool = Field(
        description="True only if the document explicitly states automatic renewal. Default false."
    )
    governing_law: Optional[str] = Field(
        default=None,
        description="Governing law clause (e.g., 'State of New York'). Null if not present."
    )

Three things to notice. First, every field has a description that reads like an instruction, not documentation — because the model uses these descriptions as its primary signal for what to do. Second, uncertain fields are Optional[T] with a default of None and an explicit "null if not present" instruction. Third, types are specific: date not str for dates, float not str for money, an Enum for the contract type. Pydantic will validate these at runtime.

The auto_renewal field deserves special attention. An earlier draft of this schema used Optional[bool] with a note to set it True if renewal is automatic. The model would default to True on ambiguous renewal clauses. Changing it to bool with a default of False and "True only if explicitly stated" sharply cut false positives.

Wiring instructor into the extraction call

Instructor patches the OpenAI (and Anthropic, Gemini, and others) SDK client to accept a response_model argument. You pass a Pydantic class; instructor handles the schema serialization, API call, response parsing, and retry loop.

import instructor
from openai import OpenAI
from pathlib import Path

client = instructor.from_openai(OpenAI())

SYSTEM_PROMPT = """You are a contract analysis assistant. Extract the requested fields
from the contract text provided. Follow the field descriptions precisely.

CRITICAL: If a field's information is not present in the document, return null.
Do NOT infer, estimate, or guess values. A null is always correct when information
is absent; an invented value causes downstream errors."""

def extract_contract(document_text: str) -> ContractExtraction:
    return client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=ContractExtraction,
        max_retries=3,  # instructor retries on validation failure
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"<contract>\n{document_text}\n</contract>"},
        ],
    )

# Usage
contract_text = Path("contract.txt").read_text()
result = extract_contract(contract_text)
print(result.model_dump())
# {'contract_type': 'master_services_agreement',
#  'parties': [{'name': 'Acme Corp', 'role': 'client', 'jurisdiction': None}, ...],
#  'effective_date': datetime.date(2024, 3, 1),
#  'expiry_date': None,
#  'total_value_usd': 250000.0,
#  'auto_renewal': False,
#  'governing_law': 'State of California'}

The max_retries=3 line is the key production feature. When the model returns a response that fails Pydantic validation — wrong type, enum value not in the set, a required field missing — instructor appends the validation error to the conversation as a user message and calls the API again. The model sees what it got wrong and corrects it. This loop handles the majority of transient validation failures without any manual intervention.

What does the retry message look like? Something like: "Please correct the following validation errors: effective_date: Input should be a valid date [type=date_from_datetime_inexact, input_value='March 1st, 2024']". The model understands this and usually returns "2024-03-01" on the next attempt.

The retry mechanism is not a substitute for schema design. If your schema is impossible to satisfy — conflicting requirements, enum values that don't cover the real space — you'll retry to exhaustion and still fail. Fix the schema first.

sequenceDiagram
    participant APP as Application
    participant INS as instructor
    participant API as LLM API
    participant VAL as Pydantic validator

    APP->>INS: create(response_model=ContractExtraction, max_retries=3)
    INS->>API: call with schema as tool definition
    API-->>INS: JSON response
    INS->>VAL: validate response
    VAL-->>INS: ValidationError "effective_date: invalid date format"
    INS->>API: retry — append error as user message
    API-->>INS: corrected JSON
    INS->>VAL: validate again
    VAL-->>INS: pass
    INS-->>APP: ContractExtraction(effective_date=date(2024,3,1), ...)

The Anthropic equivalent with Pydantic AI

For multi-step workflows where extraction is one step in an agentic pipeline, Pydantic AI is worth the overhead. It provides dependency injection (useful for mocking the LLM in tests), a structured result type system, and hooks for streaming validation.

from pydantic_ai import Agent
from pydantic_ai.models.anthropic import AnthropicModel

model = AnthropicModel("claude-sonnet-4-5")  # or claude-opus-4

extraction_agent = Agent(
    model,
    output_type=ContractExtraction,
    system_prompt="""You are a contract analysis assistant. Extract the requested fields
    from the contract text. Return null for any field not explicitly present in the document.""",
)

async def extract_with_pydantic_ai(document_text: str) -> ContractExtraction:
    result = await extraction_agent.run(
        f"<contract>\n{document_text}\n</contract>"
    )
    return result.output  # typed as ContractExtraction

# In an agentic pipeline, pass result.output to the next agent

Pydantic AI uses Anthropic's tool-calling format under the hood to enforce the schema — the extraction result becomes a tool call that must match ContractExtraction. This is covered in more depth at Structured Outputs and Constrained Decoding; the short version is that the model can't return freeform text, only a JSON object matching the schema.

The dependency injection feature matters for teams that want proper test coverage:

from pydantic_ai.models.test import TestModel

# In tests, swap in TestModel — it returns canned data without an API call
canned = TestModel(custom_output_args={
    "contract_type": "master_services_agreement",
    "parties": [{"name": "Acme Corp", "role": "client"}],
    "auto_renewal": False,
})

with extraction_agent.override(model=canned):
    result = await extract_with_pydantic_ai(test_document)
    assert result.contract_type == ContractType.MSA

Entity and relation extraction

Single-entity extraction — one contract, one record — is the straightforward case. The harder case is documents with multiple entity types and relationships between them: a supplier database with companies, contacts, products, and pricing relationships; a medical record with diagnoses, medications, dosages, and the treating physicians; an e-commerce catalog with brands, categories, products, and variants.

The temptation is to build one big schema that captures everything in a single call:

# Don't do this for complex documents
class MedicalRecord(BaseModel):
    patient: Patient
    diagnoses: list[Diagnosis]
    medications: list[Medication]
    treating_physicians: list[Physician]
    diagnosis_medication_links: list[DiagnosisMedicationLink]
    physician_diagnosis_links: list[PhysicianDiagnosisLink]

This fails in practice. The schema gets long, the model loses track of which entities have already been identified, and the relational links between them frequently hallucinate entity references that don't match what was extracted.

The two-pass approach works better:

flowchart TD
    DOC["Input document"] --> PASS1["Pass 1: Entity extraction\nextract Patient, Diagnoses,\nMedications, Physicians"]
    PASS1 --> ENTITIES["Typed entity list\n(validated, deduplicated)"]
    ENTITIES --> PASS2["Pass 2: Relation extraction\nextract links between\nknown entity IDs"]
    PASS2 --> RELATIONS["Validated relation graph"]
    ENTITIES --> STORE["Store entities"]
    RELATIONS --> STORE

    style PASS1 fill:#a855f7,color:#fff
    style PASS2 fill:#a855f7,color:#fff
    style ENTITIES fill:#0e7490,color:#fff
    style RELATIONS fill:#0e7490,color:#fff

In the second pass, you feed back the extracted entity IDs as part of the prompt and define a relation schema that references them:

from pydantic import BaseModel
from typing import Literal

class DiagnosisMedicationLink(BaseModel):
    diagnosis_id: str = Field(
        description="ID from the extracted diagnoses list (e.g., 'diag_0', 'diag_1')"
    )
    medication_id: str = Field(
        description="ID from the extracted medications list (e.g., 'med_0', 'med_1')"
    )
    relationship: Literal["prescribed_for", "contraindicated_with", "monitored_alongside"]

class RelationExtractionResult(BaseModel):
    diagnosis_medication_links: list[DiagnosisMedicationLink]

# In the relation extraction call:
entity_context = f"""
Extracted entities (use these IDs exactly):
Diagnoses: {[f"diag_{i}: {d.name}" for i, d in enumerate(entities.diagnoses)]}
Medications: {[f"med_{i}: {m.name}" for i, m in enumerate(entities.medications)]}

Now extract relationships between them from the original document.
"""

The relational links are now grounded in real entity IDs, and the model has a smaller, focused task. Hallucinated cross-references drop significantly because the model is choosing from an explicit list rather than inventing identifiers.

What breaks

Schema conformance is not semantic correctness. This distinction is the central failure mode in production extraction pipelines, and it's worth being specific about what goes wrong.

OCR noise upstream of the model. The pipeline diagram starts with PDF/OCR pre-processing, and that stage quietly dominates field-level accuracy on scanned contracts and invoices. OCR that renders "$1,250,000" as "$l,25O,OOO" doesn't make the model fail — it makes the model guess, and a guess that passes schema validation is worse than an error. Plain-text extraction also flattens tables, so a rate card's rows and columns arrive as an undifferentiated word soup. Two mitigations: run a cheap garbage detector on the OCR output (character-confusion rate, dictionary-word ratio) and route low-quality scans to human review before the LLM ever sees them; or skip text extraction entirely and send page images to a multimodal model, which as of mid-2026 beats OCR-then-extract on degraded scans and complex layouts. The tradeoff — image tokens cost more and per-page calls add up — is covered in Multimodal RAG over images, PDFs, and text.

Hallucination on sparse fields. When a document doesn't contain a field, the model may invent a plausible value rather than return null — especially for fields that the model has strong prior beliefs about (dates in standard formats, common legal jurisdictions, US phone formats). The Optional[T] = None pattern plus an explicit null instruction in the system prompt reduces this, but doesn't eliminate it. A post-extraction check that queries the source document for the extracted value (via keyword search or another LLM call) catches the remaining cases.

Format drift on normalized fields. Even with type enforcement, models drift on normalization. Dates come back as 2024-03-01, 03/01/2024, and March 1, 2024 depending on which format appeared in the document. Phone numbers drop country codes. Money values appear with and without currency symbols. Pydantic's date type catches the format variation for dates, but for strings like phone numbers you need custom validators:

from pydantic import field_validator
import re

class Party(BaseModel):
    name: str
    phone: Optional[str] = None

    @field_validator("phone")
    @classmethod
    def normalize_phone(cls, v: str | None) -> str | None:
        if v is None:
            return None
        # strip everything except digits and leading +
        digits = re.sub(r"[^\d+]", "", v)
        return digits if len(digits) >= 10 else None

Context window exhaustion on long documents. A 50-page contract is 30,000–50,000 tokens. That fits in a modern context window, but the LLM's extraction quality degrades on information buried deep in a long document — the lost-in-the-middle effect is real. The practical fix is to chunk the document by logical section (preamble, definitions, payment terms, termination clauses) and extract each section independently, then merge the results with a deduplication pass. This also cuts cost significantly on large documents.

Conflicting information across document sections. Legal documents often contain the same information stated differently in two places — a cover page and a signature block may have different effective dates. The model may pick either one. Add an instruction to prefer the most specific statement ("use the date in the 'Effective Date' clause over the date in the preamble") and flag documents where extracted values appear in multiple inconsistent forms.

Retry loops that don't converge. If the schema is logically impossible to satisfy — an enum that doesn't cover a value appearing in the document, a field marked required but genuinely absent — the retry loop will exhaust max_retries and raise. Catch instructor.exceptions.InstructorRetryException and route these documents to a human review queue rather than letting them silently drop:

import instructor.exceptions

try:
    result = extract_contract(document_text)
except instructor.exceptions.InstructorRetryException as e:
    # Log the final attempt and validation error for review
    route_to_human_review(document_id, str(e))

This is why the retry cap exists at all. Watch what an uncapped loop looks like — the model keeps failing validation the same way, the token meter climbs, and nothing converges:

{ "type": "agent-loop", "scenario": "stuck-loop", "title": "Why you cap retries: a loop that never converges" }

Cap retries at 3–5. If three attempts with the validation error in context haven't fixed it, the problem is the schema or the document, not the sampling — more retries just burn tokens on the same failure.

The semantic validation layer

After schema validation passes, there's a second validation pass that no LLM-side mechanism can replace: checking that the extracted values are semantically plausible.

from datetime import date

def validate_contract_semantics(record: ContractExtraction, document_id: str) -> list[str]:
    errors = []

    # Date logic
    if record.effective_date and record.expiry_date:
        if record.expiry_date <= record.effective_date:
            errors.append(
                f"expiry_date ({record.expiry_date}) is not after effective_date ({record.effective_date})"
            )

    # Business rule: contracts over $10M require manual review
    if record.total_value_usd and record.total_value_usd > 10_000_000:
        errors.append(f"High-value contract (${record.total_value_usd:,.0f}) — flag for review")

    # Future-date sanity check
    if record.effective_date and record.effective_date > date(2030, 1, 1):
        errors.append(f"Effective date {record.effective_date} is suspiciously far in the future")

    return errors

This layer is deterministic and testable with standard unit tests. It catches the semantic errors that would otherwise silently enter your database: a renewal date before the start date, a $50M value on what should be a small vendor contract, a phone number that passes string validation but has the wrong digit count for the claimed country.

The pattern is exactly what's described in Building Reliable Tool-Using Systems: schema conformance is the structural guard, semantic validation is the domain guard, and anything that fails either goes to a human review queue with enough context to fix it.

Prompt caching on extraction pipelines

High-volume extraction pipelines have a structural property that makes them ideal for prompt caching: the system prompt and schema instructions are static across every document, while only the document content changes.

# OpenAI prompt cache: the prefix up to the first variable content is cached automatically
# Anthropic: use cache_control blocks explicitly

import anthropic

client = anthropic.Anthropic()

STATIC_SYSTEM = """You are a contract analysis assistant..."""  # 700 tokens, always the same

def extract_with_caching(document_text: str) -> ContractExtraction:
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": STATIC_SYSTEM,
                "cache_control": {"type": "ephemeral"}  # cache this prefix
            }
        ],
        tools=[...],  # ContractExtraction schema as tool
        messages=[{"role": "user", "content": document_text}],
    )

At 700 cached system tokens and 3,000 variable document tokens, Anthropic's cache pricing (roughly 10% of normal input price for cache reads, as of mid-2026 — check current pricing) saves about 17% of input cost per call once the cache is warm — 630 token-equivalents out of 3,700 total input. At 1M documents/month, that's a meaningful number.

OpenAI caches automatically for prompts longer than 1,024 tokens; Anthropic requires explicit cache_control markers. Either way, keep your system prompt and schema instructions at the front of the prompt, and put the variable document content at the end.

Caching is the smaller lever, though. If the pipeline has no latency requirement — and a 100k-docs/month backfill almost never does — run it on the batch endpoints. OpenAI's Batch API and Anthropic's Message Batches both price asynchronous work at roughly 50% of the synchronous rate (as of mid-2026), with results delivered within a 24-hour window. On the $77/month GPT-4o-mini profile from the summary block, that's ~$38/month back — an order of magnitude more than the caching math yields. The caveat is the 24-hour window: batch is for backfills and nightly runs, not for extraction that feeds a user-facing response. On Anthropic the two discounts stack — cache reads inside a batch keep their discounted rate — so a warm-cache batch pipeline is the cheapest configuration available.

Comparison: instructor vs Pydantic AI vs direct API

Dimension	Direct API call	instructor	Pydantic AI
Schema wiring	Manual (JSON Schema dict)	Automatic from Pydantic model	Automatic from Pydantic model
Validation	Manual after parsing	Automatic via Pydantic	Automatic via Pydantic
Retry on failure	Manual boilerplate	Built-in (max_retries)	Built-in (result validation)
Multi-step workflows	Build your own	Not the focus	First-class (agent framework)
Dependency injection (testing)	No	No	Yes
Provider support	Single provider	OpenAI, Anthropic, Gemini, 10+	OpenAI, Anthropic, Gemini, Groq, others
Learning curve	Lowest	Low	Medium
Best for	One-off scripts	Existing codebases, simple pipelines	Agentic workflows, team codebases

For most teams starting a new extraction pipeline, instructor is the right default: minimal overhead, broad provider support, the retry loop does the heavy lifting. Graduate to Pydantic AI when you need the full agent infrastructure — or when the extraction is part of a larger workflow you're already building with an agent framework.

The extraction-as-tool pattern

One more pattern worth naming: using the extraction call as a tool inside a larger agent, rather than as a standalone pipeline step.

The parent agent might receive a user query like "find all contracts expiring in Q4 with auto-renewal enabled and total value over $500k." It calls a search_contracts tool to retrieve relevant documents, then calls an extract_contract_fields tool for each one. The extraction schema is designed specifically for the fields the search query cares about — a narrow schema that's faster and cheaper than the full extraction.

This is the pattern Tool and Function Calling and Designing Tool Schemas go into depth on. The key point for extraction specifically: a focused schema on 5–7 fields consistently outperforms a 20-field schema on accuracy, because the model's attention isn't diluted across fields that don't matter for the current query.

The decision in practice

For teams deciding how to approach structured extraction:

Start with instructor and GPT-4o-mini for clean, well-structured documents where the schema is stable. Add Pydantic field descriptions that read like instructions, mark uncertain fields Optional with a null directive, and implement the semantic validation layer from day one — not after you've shipped hallucinated data to three customers.

Move to a frontier model (GPT-4o, Claude Sonnet) when you see consistent failures on specific field types in the mini model. Route selectively — most pipelines find that 15–20% of documents need the stronger model, and the cost math works out: $77/month on the cheap tier plus $230/month for the 20% complex documents is still far cheaper than processing everything through the frontier model at $1,155/month.

Adopt Pydantic AI when the extraction is one step in a multi-step agentic workflow, when you need proper test mocking, or when your team is building multiple agent systems and wants a unified framework.

The two-pass entity-then-relations approach is non-negotiable for any document type with more than two entity types and relationships between them. The complexity reduction in the per-call schema more than compensates for the extra API call.

And instrument everything from the start: log which documents hit the retry loop, which exhaust retries and route to humans, and which fields have the highest validation failure rates. That data tells you where to invest in schema improvement versus routing to a stronger model versus adding targeted post-processing. Without it you're guessing.

The pattern that holds across all of this: LLMs handle the format variation and the semantic understanding; deterministic code handles the validation and the business rules. Neither can do the other's job.