Structured Output Generation: JSON Mode, Function Schemas, and Pydantic Validation

Free-text LLM output is hard to integrate into production systems. Parsing prose, extracting fields, and handling unexpected formats adds fragility to every downstream component. Structured output — asking the model to produce machine-readable output that conforms to a schema — is the solution.

This article covers the main approaches to structured output in production: JSON mode, function/tool schemas, grammar-constrained decoding, and Pydantic-based validation pipelines. The goal is to help you choose the right method and build systems that fail predictably when the model deviates from the schema.

Why Structured Output Matters

Consider a pipeline that extracts information from a customer support ticket. If the LLM returns:

The customer is frustrated. Their order number is 12345 and they want a refund.

your downstream code must parse prose, handle variations in phrasing, and deal with missing fields. Now consider:

{
  "sentiment": "negative",
  "order_id": "12345",
  "intent": "refund_request",
  "urgency": "high"
}

The second form can be directly validated, stored, routed, and processed without natural language parsing logic.

Method 1: JSON Mode

Most major LLM providers offer a JSON mode that constrains the model to always return valid JSON.

OpenAI JSON mode

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    response_format={"type": "json_object"},
    messages=[
        {
            "role": "system",
            "content": "You are a data extraction assistant. Always respond with valid JSON."
        },
        {
            "role": "user",
            "content": f"Extract the key fields from this ticket:\n{ticket_text}"
        }
    ]
)

import json
result = json.loads(response.choices[0].message.content)

Limitation: JSON mode guarantees syntactically valid JSON but does not enforce a specific schema. The model may return different keys across calls, include unexpected fields, or omit required fields.

Method 2: Function Calling / Tool Schemas

Function calling (OpenAI) and tool use (Anthropic) ask the model to fill in a schema that you define as a JSON Schema object. This is more reliable than JSON mode because the schema is explicit.

OpenAI function calling

from openai import OpenAI
import json

client = OpenAI()

ticket_schema = {
    "name": "extract_ticket_fields",
    "description": "Extract structured fields from a support ticket",
    "parameters": {
        "type": "object",
        "properties": {
            "sentiment": {
                "type": "string",
                "enum": ["positive", "neutral", "negative"],
                "description": "Overall customer sentiment"
            },
            "order_id": {
                "type": "string",
                "description": "Order ID if mentioned, null otherwise"
            },
            "intent": {
                "type": "string",
                "enum": ["refund_request", "shipping_inquiry", "product_question", "complaint", "other"],
                "description": "Primary customer intent"
            },
            "urgency": {
                "type": "string",
                "enum": ["low", "medium", "high"],
                "description": "Estimated urgency level"
            }
        },
        "required": ["sentiment", "intent", "urgency"]
    }
}

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": f"Extract fields from: {ticket_text}"}
    ],
    tools=[{"type": "function", "function": ticket_schema}],
    tool_choice={"type": "function", "function": {"name": "extract_ticket_fields"}}
)

tool_call = response.choices[0].message.tool_calls[0]
result = json.loads(tool_call.function.arguments)

The tool_choice parameter forces the model to call the specified function rather than generating text.

Method 3: Structured Output with Pydantic (OpenAI v1.40+)

OpenAI’s .parse() method accepts a Pydantic model and returns a validated object directly:

from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Literal, Optional
from enum import Enum

client = OpenAI()

class Intent(str, Enum):
    refund = "refund_request"
    shipping = "shipping_inquiry"
    product = "product_question"
    complaint = "complaint"
    other = "other"

class TicketExtraction(BaseModel):
    sentiment: Literal["positive", "neutral", "negative"]
    order_id: Optional[str] = Field(default=None, description="Order ID if mentioned")
    intent: Intent
    urgency: Literal["low", "medium", "high"]
    summary: str = Field(description="One-sentence summary of the ticket", max_length=200)

response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "Extract fields from the support ticket."},
        {"role": "user", "content": ticket_text}
    ],
    response_format=TicketExtraction,
)

ticket = response.choices[0].message.parsed  # typed TicketExtraction instance
print(ticket.intent)     # Intent.refund
print(ticket.order_id)   # "12345" or None

This approach uses constrained decoding under the hood — the model’s token probabilities are masked to only allow tokens that conform to the Pydantic schema’s JSON Schema representation. It is the most reliable method currently available.

Method 4: Instructor Library

The instructor library wraps multiple LLM providers and adds Pydantic validation with automatic retry on validation failure:

import instructor
from openai import OpenAI
from pydantic import BaseModel, field_validator
from typing import List

client = instructor.from_openai(OpenAI())

class LineItem(BaseModel):
    description: str
    quantity: int
    unit_price: float
    total: float

    @field_validator("total")
    def check_total(cls, v, info):
        expected = info.data.get("quantity", 0) * info.data.get("unit_price", 0)
        if abs(v - expected) > 0.01:
            raise ValueError(f"Total {v} does not match quantity × unit_price = {expected}")
        return v

class Invoice(BaseModel):
    vendor: str
    invoice_number: str
    line_items: List[LineItem]
    subtotal: float
    tax_rate: float
    total: float

invoice = client.chat.completions.create(
    model="gpt-4o",
    response_model=Invoice,
    messages=[
        {"role": "user", "content": f"Extract invoice data from:\n{invoice_text}"}
    ],
    max_retries=3,  # retries automatically on validation failure
)

print(f"Invoice {invoice.invoice_number}: ${invoice.total}")

instructor automatically feeds the validation error back to the model and asks it to correct its output — up to max_retries times.

Method 5: Grammar-Constrained Decoding (Local Models)

For locally hosted models (llama.cpp, vLLM, Outlines), grammar-constrained decoding enforces the schema at the token level — invalid tokens are masked to probability zero.

Outlines

import outlines
from pydantic import BaseModel
from typing import Literal

class Classification(BaseModel):
    category: Literal["technical", "billing", "general", "complaint"]
    confidence: float
    escalate: bool

model = outlines.models.transformers("mistralai/Mistral-7B-Instruct-v0.2")
generator = outlines.generate.json(model, Classification)

result = generator(
    f"Classify this support ticket:\n{ticket_text}"
)
print(result)  # Classification(category='billing', confidence=0.91, escalate=False)

Grammar-constrained decoding guarantees schema conformance at generation time. No retries needed.

vLLM guided decoding

from vllm import LLM, SamplingParams
from pydantic import BaseModel

class AnalysisResult(BaseModel):
    summary: str
    topics: list[str]
    sentiment_score: float  # -1.0 to 1.0

llm = LLM(model="mistralai/Mistral-7B-Instruct-v0.2")

sampling_params = SamplingParams(
    temperature=0.1,
    guided_decoding_backend="outlines",
    guided_json=AnalysisResult.model_json_schema(),
)

output = llm.generate(prompt, sampling_params)
import json
result = AnalysisResult(**json.loads(output[0].outputs[0].text))

Method 6: Anthropic Structured Output

Anthropic’s Claude uses tool use for structured extraction:

import anthropic
import json

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=[{
        "name": "extract_fields",
        "description": "Extract structured fields from text",
        "input_schema": {
            "type": "object",
            "properties": {
                "sentiment": {"type": "string", "enum": ["positive", "neutral", "negative"]},
                "main_topic": {"type": "string"},
                "action_required": {"type": "boolean"}
            },
            "required": ["sentiment", "main_topic", "action_required"]
        }
    }],
    tool_choice={"type": "tool", "name": "extract_fields"},
    messages=[{"role": "user", "content": ticket_text}]
)

tool_use_block = next(b for b in response.content if b.type == "tool_use")
result = tool_use_block.input

Validation Patterns

Schema-first design

Define your Pydantic model before writing the prompt. The schema is the contract between your system and the LLM.

from pydantic import BaseModel, Field, field_validator
from typing import Optional, List
import re

class ExtractedPerson(BaseModel):
    full_name: str = Field(min_length=2, max_length=100)
    email: Optional[str] = None
    phone: Optional[str] = None
    role: Optional[str] = None

    @field_validator("email")
    def validate_email(cls, v):
        if v is not None:
            pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
            if not re.match(pattern, v):
                raise ValueError(f"Invalid email format: {v}")
        return v

    @field_validator("phone")
    def normalize_phone(cls, v):
        if v is not None:
            digits_only = re.sub(r'\D', '', v)
            if len(digits_only) < 7:
                raise ValueError("Phone number too short")
            return digits_only
        return v

Retry with corrective feedback

When validation fails, feed the error back to the model:

import json
from pydantic import ValidationError

def extract_with_retry(text: str, schema: type, llm, max_retries: int = 3):
    messages = [{"role": "user", "content": f"Extract from:\n{text}"}]
    
    for attempt in range(max_retries):
        raw = llm.generate(messages)
        try:
            data = json.loads(raw)
            return schema(**data)
        except (json.JSONDecodeError, ValidationError) as e:
            if attempt == max_retries - 1:
                raise
            messages.append({"role": "assistant", "content": raw})
            messages.append({
                "role": "user",
                "content": f"Your output was invalid:\n{e}\n\nPlease correct it and return valid JSON."
            })

Choosing the Right Method

Method	Schema enforcement	Provider	Retry needed	Complexity
JSON mode	Syntax only	Most providers	Sometimes	Low
Function calling	Schema-defined	OpenAI, Anthropic	Rarely	Medium
Pydantic parse (OpenAI)	Full schema	OpenAI only	No	Medium
Instructor	Full + validators	Multi-provider	Automatic	Medium
Grammar decoding (Outlines)	Full + grammar	Local models	No	High (infra)

Decision guide:

• Use Pydantic parse for OpenAI models in new projects — most reliable with least code.
• Use instructor for multi-provider setups or complex validation logic.
• Use grammar-constrained decoding for local models or when you need hard guarantees.
• Avoid raw JSON mode for anything beyond prototyping.

Production Checklist

• Define schema as Pydantic model before writing prompts
• Validate all model outputs before using them downstream
• Log validation failures separately from LLM errors
• Set max_retries and a fallback behavior when all retries fail
• Include schema as part of your prompt versioning
• Test schema against edge cases: null fields, long strings, boundary values
• Monitor schema conformance rate as a production metric

---

Advanced: Complex Nested Schemas

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import List, Optional, Literal
from datetime import date
import re

class Address(BaseModel):
    street:    str
    city:      str
    state:     str = Field(min_length=2, max_length=2)
    zip_code:  str

    @field_validator("zip_code")
    def validate_zip(cls, v):
        if not re.match(r"^\d{5}(-\d{4})?$", v):
            raise ValueError(f"Invalid ZIP code: {v}")
        return v

class ContactInfo(BaseModel):
    email:  Optional[str] = None
    phone:  Optional[str] = None

    @model_validator(mode="after")
    def at_least_one_contact(self):
        if not self.email and not self.phone:
            raise ValueError("At least one of email or phone is required")
        return self

class JobApplication(BaseModel):
    applicant_name:  str = Field(min_length=2)
    applied_date:    date
    position:        str
    experience_years: int = Field(ge=0, le=50)
    skills:          List[str] = Field(min_length=1)
    salary_expected: Optional[float] = Field(default=None, ge=30_000, le=500_000)
    contact:         ContactInfo
    address:         Address
    status:          Literal["pending", "reviewing", "shortlisted", "rejected"] = "pending"
    notes:           List[str] = Field(default_factory=list)

# Extraction with OpenAI structured output
from openai import OpenAI
client = OpenAI()

def extract_job_application(text: str) -> JobApplication:
    response = client.beta.chat.completions.parse(
        model="gpt-4o-2024-08-06",
        messages=[
            {"role": "system", "content": "Extract job application data from the provided text. Be precise."},
            {"role": "user",   "content": text},
        ],
        response_format=JobApplication,
        temperature=0,
    )
    return response.choices[0].message.parsed

---

Real-Time Schema Validation in Production

from pydantic import ValidationError
from typing import Type, TypeVar, Generic
import time
import json
import logging

T = TypeVar("T")

class SchemaValidatedLLMCall(Generic[T]):
    """Production wrapper that validates, retries, logs, and monitors."""

    def __init__(
        self,
        schema:       Type[T],
        llm_client,
        model:        str  = "gpt-4o-mini",
        max_retries:  int  = 3,
        logger=None,
    ):
        self.schema      = schema
        self.llm         = llm_client
        self.model       = model
        self.max_retries = max_retries
        self.logger      = logger or logging.getLogger(__name__)
        self._stats = {"calls": 0, "successes": 0, "failures": 0, "retries": 0}

    def invoke(self, messages: list[dict], **kwargs) -> tuple[Optional[T], dict]:
        start = time.perf_counter()
        meta  = {"retries": 0, "model": self.model, "success": False}
        self._stats["calls"] += 1

        for attempt in range(self.max_retries):
            try:
                response = self.llm.beta.chat.completions.parse(
                    model=self.model,
                    messages=messages,
                    response_format=self.schema,
                    temperature=0,
                    **kwargs,
                )
                result = response.choices[0].message.parsed
                if result is None:
                    raise ValidationError("Parsed result is None", [])

                meta.update({
                    "success":        True,
                    "retries":        attempt,
                    "latency_ms":     round((time.perf_counter() - start) * 1000, 1),
                    "prompt_tokens":  response.usage.prompt_tokens,
                    "output_tokens":  response.usage.completion_tokens,
                })
                self._stats["successes"] += 1
                return result, meta

            except (ValidationError, Exception) as e:
                meta["retries"] += 1
                self._stats["retries"] += 1

                if attempt < self.max_retries - 1:
                    # Add correction context and retry
                    messages = messages + [{
                        "role":    "user",
                        "content": f"Your response had validation errors: {e}. Please fix it and try again.",
                    }]
                else:
                    meta.update({
                        "success":    False,
                        "error":      str(e),
                        "latency_ms": round((time.perf_counter() - start) * 1000, 1),
                    })
                    self._stats["failures"] += 1
                    self.logger.error(json.dumps({"event": "schema_validation_failure", **meta}))

        return None, meta

    def get_stats(self) -> dict:
        total = self._stats["calls"]
        return {
            **self._stats,
            "success_rate":    round(self._stats["successes"] / max(total, 1), 3),
            "avg_retries":     round(self._stats["retries"]   / max(total, 1), 2),
        }

---

Monitoring Schema Conformance

import time
from collections import defaultdict
from threading import Lock

class SchemaConformanceMonitor:
    """Track schema validation metrics for operational dashboards."""

    def __init__(self, window_seconds: int = 3600):
        self.window  = window_seconds
        self._events: list[dict] = []
        self._lock   = Lock()

    def record(self, schema_name: str, success: bool, retries: int = 0):
        with self._lock:
            self._events.append({
                "ts":          time.time(),
                "schema":      schema_name,
                "success":     success,
                "retries":     retries,
            })
            # Trim old events outside window
            cutoff = time.time() - self.window
            self._events = [e for e in self._events if e["ts"] > cutoff]

    def report(self) -> dict:
        with self._lock:
            events = list(self._events)

        if not events:
            return {}

        by_schema = defaultdict(lambda: {"total": 0, "failures": 0, "retries": 0})
        for e in events:
            s = by_schema[e["schema"]]
            s["total"]    += 1
            s["failures"] += 0 if e["success"] else 1
            s["retries"]  += e["retries"]

        return {
            schema: {
                "conformance_rate": round(1 - data["failures"] / data["total"], 3),
                "total_calls":      data["total"],
                "failure_count":    data["failures"],
                "avg_retries":      round(data["retries"] / data["total"], 2),
            }
            for schema, data in by_schema.items()
        }

# Usage
monitor = SchemaConformanceMonitor()

def tracked_extraction(text: str) -> Optional[TicketExtraction]:
    result, meta = extractor.invoke([{"role": "user", "content": text}])
    monitor.record("TicketExtraction", meta["success"], meta.get("retries", 0))
    return result

# Periodically log the report:
# print(monitor.report())
# → {'TicketExtraction': {'conformance_rate': 0.983, 'total_calls': 1200, 'failure_count': 20, 'avg_retries': 0.04}}

---

Streaming Structured Output

from openai import OpenAI
from pydantic import BaseModel

client = OpenAI()

class StreamedReport(BaseModel):
    title:       str
    sections:    list[str]
    word_count:  int

# Stream with parse — returns tokens incrementally, validates at the end
with client.beta.chat.completions.stream(
    model="gpt-4o-2024-08-06",
    messages=[{
        "role": "user",
        "content": "Generate a 3-section report outline on quantum computing.",
    }],
    response_format=StreamedReport,
) as stream:
    # Stream events as they arrive
    for event in stream:
        if event.type == "content.delta":
            print(event.delta, end="", flush=True)

    # Get final parsed result after stream completes
    final: StreamedReport = stream.get_final_completion().choices[0].message.parsed
    print(f"\n\nFinal: {final.title} | {len(final.sections)} sections | {final.word_count} words")

        print(event.delta, end="", flush=True)

# Get final parsed result after stream completes
final: StreamedReport = stream.get_final_completion().choices[0].message.parsed
print(f"\n\nFinal: {final.title} | {len(final.sections)} sections | {final.word_count} words") ```

---

Multi-Step Structured Pipelines

from pydantic import BaseModel
from typing import List, Optional
from openai import OpenAI
import instructor

client = instructor.from_openai(OpenAI())

# Step 1: Extract entities
class EntityList(BaseModel):
    people:       List[str]
    organizations: List[str]
    locations:    List[str]

# Step 2: Analyze relationships  
class Relationship(BaseModel):
    entity_a:  str
    entity_b:  str
    relation:  str
    confidence: float

class RelationshipGraph(BaseModel):
    relationships: List[Relationship]
    summary:       str

# Step 3: Generate report
class IntelligenceReport(BaseModel):
    title:          str
    key_findings:   List[str]
    risk_level:     str   # low, medium, high, critical
    recommendations: List[str]
    confidence:     float

def analyze_document_pipeline(text: str) -> IntelligenceReport:
    # Step 1
    entities = client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=EntityList,
        messages=[{"role": "user", "content": f"Extract all people, orgs, and locations from:\n{text}"}],
    )

    # Step 2
    rels = client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=RelationshipGraph,
        messages=[{"role": "user", "content": f"""
Analyze relationships between these entities:
People: {entities.people}
Organizations: {entities.organizations}
Locations: {entities.locations}

Original text: {text}
"""}],
    )

    # Step 3
    report = client.chat.completions.create(
        model="gpt-4o",
        response_model=IntelligenceReport,
        messages=[{"role": "user", "content": f"""
Generate an intelligence report based on:
Entities: {entities.model_dump_json()}
Relationships: {rels.model_dump_json()}
"""}],
    )
    return report

---

Schema Versioning

from pydantic import BaseModel, Field
from typing import Literal
from datetime import datetime

class TicketExtractionV1(BaseModel):
    """Schema version 1 — basic extraction."""
    schema_version: Literal["v1"] = "v1"
    sentiment:      str
    order_id:       str | None
    intent:         str

class TicketExtractionV2(BaseModel):
    """Schema version 2 — adds urgency and category."""
    schema_version: Literal["v2"] = "v2"
    sentiment:      str
    order_id:       str | None
    intent:         str
    urgency:        str           # NEW in v2
    category:       str           # NEW in v2
    extracted_at:   datetime = Field(default_factory=datetime.utcnow)

def migrate_v1_to_v2(v1: TicketExtractionV1) -> TicketExtractionV2:
    return TicketExtractionV2(
        sentiment=v1.sentiment,
        order_id=v1.order_id,
        intent=v1.intent,
        urgency="medium",   # default for migrated records
        category="general", # default for migrated records
    )

# Schema registry
SCHEMA_REGISTRY = {
    "v1": TicketExtractionV1,
    "v2": TicketExtractionV2,
}
CURRENT_VERSION = "v2"

def extract(text: str) -> dict:
    schema   = SCHEMA_REGISTRY[CURRENT_VERSION]
    result   = client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=schema,
        messages=[{"role": "user", "content": f"Extract ticket data from: {text}"}],
    )
    return result.model_dump()

---

LLM Output Schema Anti-Patterns

Anti-Pattern	Problem	Fix
Dict[str, Any] return type	No validation, no type safety	Define explicit Pydantic model
Optional fields everywhere	Model never refuses unclear extractions	Use Optional only when field is truly optional
Free-text enum fields	“high”, “High”, “HIGH” all valid	Use Literal["low", "medium", "high"]
Nested unbounded lists	Model generates arbitrary structure	Set explicit list item types
Parsing LLM output with regex	Fragile, breaks on format changes	Use response_format={"type": "json_object"}
Schema embedded in prompt only	No enforcement at generation time	Use constrained decoding (Outlines) or .parse()
Same schema for all tasks	Over-fitting to one use case	Use task-specific schemas

---

Testing Structured Outputs

import pytest
from pydantic import ValidationError

class TestTicketExtraction:
    def test_valid_extraction(self, extractor):
        result = extractor.extract("My order #12345 arrived broken. Very angry!")
        assert result.order_id == "12345"
        assert result.sentiment == "negative"
        assert result.urgency in ("high", "critical")

    def test_handles_no_order_id(self, extractor):
        result = extractor.extract("General question about your return policy.")
        assert result.order_id is None
        assert result.intent == "policy_question"

    def test_schema_enforced(self, raw_llm):
        """Direct model without schema enforcement fails."""
        response = raw_llm.invoke("Extract from: Order 99 is late!")
        # Without schema, output is unpredictable
        assert isinstance(response.content, str)

    def test_retry_on_validation_failure(self, mock_llm_that_fails_once, extractor):
        """Extractor retries automatically on validation error."""
        result = extractor.extract("Order 777 needs refund")
        assert result is not None   # succeeded on retry

---

Structured output is not a feature — it is a reliability requirement for any LLM system that integrates with other software components. The schema is the contract between your application and the model, and enforcing it rigorously — through constrained decoding, Pydantic validation, automatic retry, conformance monitoring, and schema versioning — is what separates prototype-quality LLM code from production-grade engineering.

---

Structured Output Conclusion

Structured output generation sits at the intersection of language modeling and software engineering. It is the mechanism that makes LLM outputs programmable — reliable enough to drive database writes, API calls, and business workflows. The methods range from simple JSON mode for prototyping to grammar-constrained decoding for hard guarantees. The tooling (Pydantic, instructor, Outlines) has matured significantly. What has not changed is the fundamental principle: define the schema first, enforce it at every boundary, and monitor conformance in production as seriously as you monitor error rates.

Structured Output Resources

• instructor library — Multi-provider structured output with automatic retry
• Outlines — Grammar-constrained decoding for local models
• OpenAI Structured Outputs — Official .parse() API documentation
• Pydantic documentation — Data validation and schema definition
• vLLM guided decoding — Schema enforcement in vLLM

Structured Output: Key Takeaways

The engineering discipline for structured LLM output:

• Define schemas first; write prompts second
• Use constrained decoding (Outlines, .parse()) over parse-and-retry where possible
• Monitor conformance rate as a first-class SLO metric
• Version schemas; coordinate breaking changes across the system
• Validate at system boundaries, not just in tests
• Log validation failures separately from model errors—they indicate different problems
• Test edge cases explicitly: null fields, long strings, Unicode, boundary values

Structured output is the contract that makes LLM integration with the rest of your software stack reliable. Enforce it rigorously.

---

Structured Output Quick Reference

Method	Schema Enforcement	Provider	Best For
JSON mode	Syntax only	Most	Prototyping
Function calling	Schema-defined	OpenAI, Anthropic	General production
OpenAI .parse()	Full (constrained)	OpenAI only	New OpenAI projects
instructor library	Full + validators	Multi-provider	Cross-provider apps
Outlines grammar	Full + grammar	Local models	Hard guarantees

Common Pydantic Validators for LLM Output

from pydantic import BaseModel, Field, field_validator
import re

class ExtractedData(BaseModel):
    # Enum validation
    sentiment: Literal["positive", "neutral", "negative"]

    # Length constraint
    summary: str = Field(min_length=10, max_length=500)

    # Pattern validation
    email: Optional[str] = None
    @field_validator("email")
    def validate_email(cls, v):
        if v and not re.match(r'^[^@]+@[^@]+\.[^@]+$', v):
            raise ValueError(f"Invalid email: {v}")
        return v

    # Range validation
    confidence: float = Field(ge=0.0, le=1.0)

    # Cross-field validation
    start_date: Optional[str] = None
    end_date:   Optional[str] = None
    @model_validator(mode="after")
    def dates_consistent(self):
        if self.start_date and self.end_date:
            if self.start_date > self.end_date:
                raise ValueError("start_date must be before end_date")
        return self

---

Schema-First Development Workflow

1. Define Pydantic schema FIRST
   └─ Think: what fields, types, constraints, and validators are needed?

2. Write the extraction prompt SECOND
   └─ Include schema example in prompt for few-shot guidance

3. Choose enforcement method
   └─ Cloud: OpenAI .parse() or instructor
   └─ Local: Outlines grammar-constrained generation

4. Add validation tests
   └─ Test happy path + edge cases + adversarial inputs

5. Monitor conformance in production
   └─ Track: schema_valid_rate, retry_rate, fallback_rate

6. Iterate on schema when new fields are needed
   └─ Version the schema; migrate persisted data if needed

# Minimal conformance monitoring
class ConformanceCounter:
    def __init__(self):
        self.total   = 0
        self.success = 0
        self.retries = 0

    def record(self, success: bool, retries: int = 0):
        self.total   += 1
        self.success += success
        self.retries += retries

    @property
    def rate(self) -> float:
        return self.success / max(self.total, 1)

conformance = ConformanceCounter()

def tracked_parse(text: str, schema, llm) -> Optional[object]:
    result, meta = extractor.invoke([{"role": "user", "content": text}])
    conformance.record(meta["success"], meta.get("retries", 0))
    return result

# Periodic check:
# if conformance.rate < 0.95:
#     alert(f"Schema conformance degraded: {conformance.rate:.1%}")

---

Structured Output Engineering Principles

1. Schema first: Define your Pydantic model before writing the prompt
2. Constrained decoding over retries: For local models, use Outlines; for APIs, use .parse(); both are more reliable than parse-and-retry
3. Validate at system boundary: Treat model output as untrusted input — validate before using it in downstream logic
4. Track conformance rate: If schema_valid_rate drops below 95%, something changed in the model, prompt, or data
5. Design for partial failures: Some retries will exhaust; define a fallback behavior (default values, human escalation)
6. Version your schemas: Breaking schema changes must be coordinated with migration of persisted data
7. Log validation failures separately: Validation errors indicate a different problem than model errors or network errors
8. Test edge cases explicitly: null fields, maximum-length strings, boundary values, and Unicode are common sources of schema failure

Structured output is the contract between your LLM application and the rest of your software system. Enforcing it rigorously — at generation time with constrained decoding, at runtime with Pydantic, and in production with conformance monitoring — is what makes LLM outputs reliable enough to integrate into critical workflows. ```

Summary

The schema is the contract. Define it first, enforce it at every boundary, monitor conformance in production, and version it when it changes. These four practices combined with constrained decoding for local models and Pydantic validation with automatic retry for API models are the foundation of reliable structured LLM output. Any LLM system that generates output consumed by downstream software must treat schema enforcement as a first-class engineering concern, not an afterthought.