5. Structured Extraction

Description

Structured extraction turns documents (invoices, receipts, contracts, IDs) into schema-validated data. In this capsule you'll learn to use Pydantic to define typed schemas, design prompts that return reliable JSON, validate the LLM's output and handle errors robustly. It's the foundation for automating data entry from real documents.

Why it matters: Without schemas, the LLM's output is free text that requires manual parsing. With Pydantic you get typed, validated data ready for databases or APIs. The difference between a prototype and a production system is validation.

Connection with the module: In capsule 02 you learned to process PDFs. In capsule 03 you saw OCR vs Vision APIs. In capsule 04 you prepared document images for Vision. Here you close the loop: you take the image, define what data you need with a schema, and get validated data.

Key Concepts

Schema-first extraction

1. Define the schema (Pydantic model with types, validators, optional fields)
2. Include the schema in the prompt as a contract
3. Use response_format={"type": "json_object"} (OpenAI) or explicit instructions (Anthropic)
4. Parse the JSON response and validate with Pydantic
5. Handle errors with intelligent retries

Why Pydantic and not dictionaries

Schemas with Pydantic

Schema: Invoice

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional

class InvoiceItem(BaseModel):
    description: str = Field(..., description="Product or service name")
    quantity: float = Field(default=1.0, ge=0)
    unit_price: float = Field(..., ge=0)
    amount: float = Field(..., ge=0)

    @model_validator(mode="after")
    def verify_amount(self):
        expected = round(self.quantity * self.unit_price, 2)
        if abs(self.amount - expected) > 0.5:
            self.amount = expected
        return self

class Invoice(BaseModel):
    date: Optional[str] = Field(None, description="YYYY-MM-DD")
    number: Optional[str] = Field(None, description="Number or reference")
    vendor: Optional[str] = None
    customer: Optional[str] = None
    total: Optional[float] = Field(None, ge=0)
    subtotal: Optional[float] = Field(None, ge=0)
    tax: Optional[float] = Field(None, ge=0)
    currency: str = Field(default="MXN")
    items: list[InvoiceItem] = Field(default_factory=list)

    @field_validator("currency")
    @classmethod
    def valid_currency(cls, v: str) -> str:
        v = v.upper().strip()
        if v not in {"MXN", "USD", "EUR", "COP", "ARS", "CLP", "PEN"}:
            return "MXN"
        return v

    @model_validator(mode="after")
    def calculate_total(self):
        if self.total is None and self.items:
            self.total = round(sum(i.amount for i in self.items), 2)
        return self

Schema: Receipt

class Receipt(BaseModel):
    merchant: Optional[str] = Field(None, description="Establishment name")
    address: Optional[str] = None
    date: Optional[str] = None
    time: Optional[str] = None
    items: list[dict] = Field(default_factory=list)
    subtotal: Optional[float] = None
    tax: Optional[float] = None
    total: Optional[float] = None
    payment_method: Optional[str] = None

    @field_validator("total")
    @classmethod
    def total_positive(cls, v):
        if v is not None and v < 0:
            raise ValueError("total cannot be negative")
        return v

Schema: Contract

class ContractParty(BaseModel):
    name: str
    role: Optional[str] = Field(None, description="E.g.: lessor, vendor")
    id: Optional[str] = None

class Contract(BaseModel):
    type: Optional[str] = Field(None, description="lease, services, sale")
    parties: list[ContractParty] = Field(default_factory=list)
    signing_date: Optional[str] = None
    start_date: Optional[str] = None
    end_date: Optional[str] = None
    subject: Optional[str] = None
    amount: Optional[float] = None
    currency: str = "MXN"
    key_clauses: list[str] = Field(default_factory=list)

Schema: ID Document

class IdDocument(BaseModel):
    document_type: Optional[str] = Field(None, description="ID card, passport, license")
    full_name: Optional[str] = None
    birth_date: Optional[str] = None
    document_number: Optional[str] = None
    issue_date: Optional[str] = None
    expiry_date: Optional[str] = None
    nationality: Optional[str] = None
    sex: Optional[str] = None

    @field_validator("sex")
    @classmethod
    def normalize_sex(cls, v):
        if v is None:
            return v
        mapping = {"M": "M", "F": "F", "MASCULINO": "M", "FEMENINO": "F", "H": "M", "MUJER": "F"}
        return mapping.get(v.upper().strip(), v)

Prompt Engineering for Extraction

Generate a prompt from a Pydantic schema

def schema_to_prompt(model: type[BaseModel]) -> str:
    """Generates a field description for the prompt from the Pydantic schema."""
    schema = model.model_json_schema()
    props = schema.get("properties", {})
    required = schema.get("required", [])
    lines = []
    for name, info in props.items():
        t = info.get("type", "string")
        desc = info.get("description", "")
        req = "(required)" if name in required else "(optional)"
        line = f"- {name}: {t} {req}"
        if desc:
            line += f" — {desc}"
        lines.append(line)
    return "\n".join(lines)

Complete prompt with few-shot

def build_extraction_prompt(model: type[BaseModel], few_shot: bool = True) -> str:
    """Builds an extraction prompt with schema and example."""
    schema_desc = schema_to_prompt(model)
    prompt = f"""You are a document data extraction system.
Analyze the image and extract the data in JSON format.

## Fields to extract:
{schema_desc}

## Rules:
- Respond ONLY with valid JSON, no extra text.
- Use null for fields not found. Dates in YYYY-MM-DD. Amounts as numbers.
- Do not invent data. If it's not visible, use null."""

    if few_shot:
        prompt += """

## Example output:
{
    "date": "2024-03-15", "number": "INV-001234",
    "vendor": "Tech Services Inc.", "total": 15680.50, "currency": "MXN",
    "items": [{"description": "Consulting", "quantity": 10, "unit_price": 1500.00, "amount": 15000.00}]
}"""
    return prompt

Extraction with critical parameters

from openai import OpenAI
import base64
import json

client = OpenAI()

def extract_with_schema(image_path: str, model_class: type[BaseModel]) -> BaseModel:
    """Extracts structured data from an image using a Pydantic schema."""
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": build_extraction_prompt(model_class)},
                {"type": "image_url", "image_url": {
                    "url": f"data:image/jpeg;base64,{b64}", "detail": "high"
                }}
            ]
        }],
        response_format={"type": "json_object"},
        temperature=0,
        max_tokens=4000
    )

    data = json.loads(response.choices[0].message.content)
    return model_class(**data)

Validation with Pydantic

Response pre-processing

from pydantic import ValidationError

def clean_llm_data(data: dict) -> dict:
    """Cleans common problematic values in LLM responses."""
    cleaned = {}
    for key, value in data.items():
        if isinstance(value, str) and value.strip().lower() in ("n/a", "n/d", "-", "none", "null", ""):
            cleaned[key] = None
        elif isinstance(value, list):
            cleaned[key] = [clean_llm_data(item) if isinstance(item, dict) else item for item in value]
        elif isinstance(value, dict):
            cleaned[key] = clean_llm_data(value)
        else:
            cleaned[key] = value
    return cleaned

def parse_llm_response(raw_json: str, model_class: type[BaseModel]) -> BaseModel | None:
    """Parses the LLM's JSON response with cleaning and validation."""
    try:
        data = json.loads(raw_json)
        data = clean_llm_data(data)
        return model_class(**data)
    except json.JSONDecodeError as e:
        print(f"Invalid JSON: {e}")
        return None
    except ValidationError as e:
        print(f"Validation failed: {e.error_count()} errors")
        for error in e.errors():
            field_path = " → ".join(str(x) for x in error["loc"])
            print(f"  {field_path}: {error['msg']}")
        return None

Retries with error context

If validation fails, include the error in the second attempt so the LLM corrects itself:

def extract_with_retry(
    image_path: str,
    model_class: type[BaseModel],
    max_retries: int = 2
) -> BaseModel | None:
    """Extracts data with retries that include previous errors."""
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    base_prompt = build_extraction_prompt(model_class)
    messages = [{
        "role": "user",
        "content": [
            {"type": "text", "text": base_prompt},
            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
        ]
    }]

    for attempt in range(max_retries + 1):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            response_format={"type": "json_object"},
            temperature=0,
            max_tokens=4000
        )
        raw = response.choices[0].message.content

        try:
            data = clean_llm_data(json.loads(raw))
            return model_class(**data)
        except (json.JSONDecodeError, ValidationError) as e:
            if attempt < max_retries:
                messages.append({"role": "assistant", "content": raw})
                messages.append({
                    "role": "user",
                    "content": f"JSON with errors:\n{e}\n\nCorrect it and respond with valid JSON."
                })
            else:
                print(f"Failed after {max_retries + 1} attempts: {e}")
                return None

Multi-Schema Extraction

DOCUMENT_SCHEMAS = {
    "invoice": Invoice,
    "receipt": Receipt,
    "contract": Contract,
    "id_document": IdDocument,
}

def classify_document(image_path: str) -> str:
    """Classifies the document type using Vision."""
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": (
                    "Classify this document: invoice, receipt, contract, id_document.\n"
                    "Respond with JSON: {\"type\": \"...\"}"
                )},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
            ]
        }],
        response_format={"type": "json_object"},
        temperature=0,
        max_tokens=50
    )
    return json.loads(response.choices[0].message.content).get("type", "invoice")

Classify and extract in a single request

To reduce latency and cost, you can classify and extract in a single call. The key is to include the fields of all the schemas in the prompt and ask the model to select the type:

def classify_and_extract(image_path: str) -> tuple[str, dict]:
    """Classifies the document type and extracts data in a single request."""
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    all_fields = {name: list(m.model_json_schema()["properties"].keys())
                  for name, m in DOCUMENT_SCHEMAS.items()}

    prompt = (f"Analyze this document. Determine the type (invoice/receipt/contract/id_document) "
              f"and extract fields.\nFields per type: {json.dumps(all_fields, ensure_ascii=False)}\n"
              f"Respond with JSON: {{\"type\": \"...\", \"data\": {{...}}}}. Use null if you can't find a field.")

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": [
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
        ]}],
        response_format={"type": "json_object"}, temperature=0
    )

    result = json.loads(response.choices[0].message.content)
    doc_type = result.get("type", "invoice")
    data = clean_llm_data(result.get("data", {}))
    schema_class = DOCUMENT_SCHEMAS.get(doc_type)
    if schema_class:
        return doc_type, schema_class(**data).model_dump()
    return doc_type, data

OpenAI JSON Mode vs Prompt Engineering

Anthropic: no native JSON mode

import anthropic
import re

def extract_with_anthropic(image_b64: str, prompt: str) -> dict:
    """Extracts JSON from Anthropic with a regex fallback."""
    client_anthropic = anthropic.Anthropic()

    response = client_anthropic.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4000,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": image_b64}},
                {"type": "text", "text": prompt + "\n\nRespond ONLY with valid JSON. No ```json fences. Just the JSON."}
            ]
        }]
    )

    text = response.content[0].text.strip()
    text = re.sub(r"^```(?:json)?\s*", "", text)
    text = re.sub(r"\s*```$", "", text)
    match = re.search(r"\{.*\}", text, re.DOTALL)
    return json.loads(match.group()) if match else json.loads(text)

Recommendation: OpenAI → response_format + temperature=0. Anthropic → explicit prompt + regex. In both, Pydantic validates the schema.

Pipeline: Image → Schema → Validated Data

from dataclasses import dataclass, field as dc_field
from enum import Enum

class ExtractionStatus(str, Enum):
    SUCCESS = "success"
    PARTIAL = "partial"
    FAILED = "failed"

@dataclass
class ExtractionResult:
    status: ExtractionStatus
    doc_type: str
    data: dict = dc_field(default_factory=dict)
    errors: list[str] = dc_field(default_factory=list)
    attempts: int = 0

def full_extraction_pipeline(
    image_path: str, force_type: str | None = None, max_retries: int = 2
) -> ExtractionResult:
    """Complete pipeline: image → classification → extraction → validation."""
    result = ExtractionResult(status=ExtractionStatus.FAILED, doc_type="unknown")
    doc_type = force_type if force_type in DOCUMENT_SCHEMAS else classify_document(image_path)
    result.doc_type = doc_type

    schema_class = DOCUMENT_SCHEMAS.get(doc_type)
    if not schema_class:
        result.errors.append(f"No schema for type: {doc_type}")
        return result

    extracted = extract_with_retry(image_path, schema_class, max_retries)
    result.attempts = max_retries + 1
    if extracted is None:
        result.errors.append("Extraction failed after all retries")
        return result

    data_dict = extracted.model_dump()
    non_null = sum(1 for v in data_dict.values() if v is not None and v != [] and v != "")
    completeness = non_null / len(data_dict) if data_dict else 0
    result.data = data_dict
    result.status = ExtractionStatus.SUCCESS if completeness >= 0.5 else ExtractionStatus.PARTIAL
    if completeness < 0.5:
        result.errors.append(f"Only {completeness:.0%} of fields extracted")
    return result

# Usage
result = full_extraction_pipeline("scanned_invoice.jpg")
print(f"Status: {result.status.value} | Type: {result.doc_type}")
print(json.dumps(result.data, indent=2, ensure_ascii=False, default=str))

Difficult Fields

Dates in multiple formats

from datetime import datetime

MONTH_NAMES = {"january": "01", "february": "02", "march": "03", "april": "04", "may": "05",
               "june": "06", "july": "07", "august": "08", "september": "09",
               "october": "10", "november": "11", "december": "12"}

def normalize_date(raw: str | None) -> str | None:
    """Normalizes dates to YYYY-MM-DD from ISO, US and English-text formats."""
    if not raw:
        return None
    text = raw.lower().strip()
    for name, num in MONTH_NAMES.items():
        if name in text:
            text = re.sub(rf"\b{name}\b", num, text)
            text = re.sub(r"\bof\b", "", text).strip()
            text = re.sub(r"\s+", "/", text)
            try:
                return datetime.strptime(text, "%d/%m/%Y").strftime("%Y-%m-%d")
            except ValueError:
                pass
    for fmt in ["%Y-%m-%d", "%d/%m/%Y", "%d-%m-%Y", "%m/%d/%Y", "%d.%m.%Y"]:
        try:
            return datetime.strptime(raw.strip(), fmt).strftime("%Y-%m-%d")
        except ValueError:
            continue
    return raw

Currencies and amounts

def normalize_amount(raw: str | float | None) -> float | None:
    """Converts amount strings to float. Handles Latin format (1.234,56) and Anglo (1,234.56)."""
    if raw is None:
        return None
    if isinstance(raw, (int, float)):
        return float(raw)
    cleaned = re.sub(r"[^\d.,\-]", "", str(raw))
    if re.match(r"^\d{1,3}(\.\d{3})+(,\d{2})?$", cleaned):  # Latin
        cleaned = cleaned.replace(".", "").replace(",", ".")
    elif re.match(r"^\d{1,3}(,\d{3})+(\.\d{2})?$", cleaned):  # Anglo
        cleaned = cleaned.replace(",", "")
    elif "," in cleaned and "." not in cleaned:
        cleaned = cleaned.replace(",", ".")
    try:
        return float(cleaned)
    except ValueError:
        return None

Nested items (invoice line items)

The most problematic field — LLMs return items as strings, or with keys in Spanish/English:

def normalize_items(raw_items: list) -> list[dict]:
    """Normalizes items that may come as strings or dicts with variable keys."""
    normalized = []
    for item in raw_items:
        if isinstance(item, str):
            normalized.append({"description": item, "quantity": 1, "unit_price": 0, "amount": 0})
        elif isinstance(item, dict):
            normalized.append({
                "description": item.get("description", item.get("name", "")),
                "quantity": float(item.get("quantity", item.get("qty", 1))),
                "unit_price": float(item.get("unit_price", item.get("price", 0))),
                "amount": float(item.get("amount", item.get("total", 0)))
            })
    return normalized

Complete post-processor

def postprocess_extraction(data: dict) -> dict:
    """Applies all the normalizations to the extracted data."""
    for key in ["date", "signing_date", "start_date", "end_date",
                "birth_date", "issue_date", "expiry_date"]:
        if key in data:
            data[key] = normalize_date(data[key])
    for key in ["total", "subtotal", "tax", "amount"]:
        if key in data and isinstance(data[key], str):
            data[key] = normalize_amount(data[key])
    if "items" in data and isinstance(data["items"], list):
        data["items"] = normalize_items(data["items"])
    return data

Troubleshooting

Problem 1: Malformed JSON in the response

Cause: The model includes markdown (```json) or extra text, especially without response_format.

Solution: Extract the JSON with regex:

def extract_json_from_text(text: str) -> dict:
    text = re.sub(r"^```(?:json)?\s*", "", text.strip())
    text = re.sub(r"\s*```$", "", text)
    match = re.search(r"\{.*\}", text, re.DOTALL)
    if match:
        return json.loads(match.group())
    raise json.JSONDecodeError("No JSON found", text, 0)

Problem 2: Missing fields that do exist in the document

Cause: The prompt doesn't specify where to look for the fields, or the layout is complex.

Solution: Include the expected location: "The invoice number is usually in the top-right corner. The total is at the end of the document."

Problem 3: Incorrect types ("N/A" instead of null)

Cause: The model uses "N/A", "Not available", "-" as strings.

Solution: Use clean_llm_data() before Pydantic. Add to the prompt: "Use null (not 'N/A', not '-') for fields not found."

Problem 4: Hallucinated values

Cause: The LLM invents data for fields that don't exist in the document.

Solution: Cross-validation (total vs sum of items). In the prompt: "If a field is NOT visible, use null. NEVER invent values." Add a model_validator that detects inconsistencies.

Problem 5: Inconsistent format across documents

Cause: Invoices from different vendors use different field names.

Solution: model_validator(mode="before") with alias mapping:

class FlexibleInvoice(Invoice):
    @model_validator(mode="before")
    @classmethod
    def map_aliases(cls, data):
        alias_map = {
            "invoice_number": "number", "total_amount": "total",
            "seller": "vendor", "moneda": "currency",
            "fecha": "date", "line_items": "items",
        }
        return {alias_map.get(k, k): v for k, v in data.items()}

Exercises

Exercise 1: Schema with cross-validation

Create a SalesNote schema with: number, date, items (list), subtotal, discount, total. Add a model_validator that verifies that total == subtotal - discount (tolerance $1).

class SaleItem(BaseModel):
    product: str
    quantity: float = 1.0
    price: float
    amount: float

class SalesNote(BaseModel):
    number: Optional[str] = None
    date: Optional[str] = None
    items: list[SaleItem] = Field(default_factory=list)
    subtotal: Optional[float] = None
    discount: Optional[float] = Field(default=0.0)
    total: Optional[float] = None
    _warning: str = ""

    @model_validator(mode="after")
    def verify_totals(self):
        if self.subtotal is not None and self.total is not None:
            expected = self.subtotal - (self.discount or 0)
            if abs(self.total - expected) > 1.0:
                self._warning = f"Total ({self.total}) ≠ subtotal - discount ({expected})"
        if self.subtotal is None and self.items:
            self.subtotal = round(sum(i.amount for i in self.items), 2)
        return self

Exercise 2: Multi-provider extractor

Create extract_universal(image_path, schema_class, provider) that supports "openai" (with response_format) and "anthropic" (with regex parsing). Both must validate with Pydantic.

def extract_universal(
    image_path: str, schema_class: type[BaseModel], provider: str = "openai"
) -> BaseModel | None:
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    prompt = build_extraction_prompt(schema_class)

    if provider == "openai":
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": [
                {"type": "text", "text": prompt},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
            ]}],
            response_format={"type": "json_object"},
            temperature=0
        )
        data = json.loads(response.choices[0].message.content)
    elif provider == "anthropic":
        data = extract_with_anthropic(b64, prompt)
    else:
        raise ValueError(f"Unsupported provider: {provider}")

    return schema_class(**clean_llm_data(data))

Exercise 3: Batch of mixed documents

Create a function that takes a list of mixed image paths, classifies each one, extracts with the correct schema, and returns a report with totals by type and status.

def process_mixed_batch(image_paths: list[str]) -> dict:
    results = {"total": len(image_paths), "successful": 0, "failed": 0, "by_type": {}}
    for path in image_paths:
        try:
            ext = full_extraction_pipeline(path)
            ok = ext.status != ExtractionStatus.FAILED
            results["successful" if ok else "failed"] += 1
            doc_type = ext.doc_type
            results["by_type"].setdefault(doc_type, {"successful": 0, "failed": 0})
            results["by_type"][doc_type]["successful" if ok else "failed"] += 1
        except Exception:
            results["failed"] += 1
    return results

Exercise 4: Date normalization with tests

Implement safe_extract_date that takes a raw string and returns a Python date or None. It must handle at least 5 formats. Include asserts.

from datetime import date as date_type

def safe_extract_date(raw: str | None) -> date_type | None:
    if not raw or not isinstance(raw, str):
        return None
    text = raw.lower().strip()
    for name, num in MONTH_NAMES.items():
        if name in text:
            text = re.sub(rf"\b{name}\b", num, text)
            text = re.sub(r"\bof\b", "", text).strip()
            text = re.sub(r"\s+", "/", text)
            try:
                return datetime.strptime(text, "%d/%m/%Y").date()
            except ValueError:
                pass
    for fmt in ["%Y-%m-%d", "%d/%m/%Y", "%d-%m-%Y", "%m/%d/%Y", "%d.%m.%Y"]:
        try:
            return datetime.strptime(raw.strip(), fmt).date()
        except ValueError:
            continue
    return None

assert safe_extract_date("2024-03-15") == date_type(2024, 3, 15)
assert safe_extract_date("15/03/2024") == date_type(2024, 3, 15)
assert safe_extract_date("15-03-2024") == date_type(2024, 3, 15)
assert safe_extract_date("15 March 2024") == date_type(2024, 3, 15)
assert safe_extract_date("15.03.2024") == date_type(2024, 3, 15)
assert safe_extract_date(None) is None
print("All tests passed.")

Additional Resources

1. Pydantic v2 Documentation
2. OpenAI JSON Mode / Structured Outputs
3. Anthropic Vision — Extracting structured data
4. Pydantic model_validator