Structured Output & Schema Validation

The Structured Output Problem

Extracting structured data from LLM responses by parsing text is fragile:

# Dangerous pattern
response = llm.invoke("Extract the price and product name from: " + text)
lines = response.content.split("\n")
# split("\n") assumes the model always uses newlines: it won't, consistently
price = float(lines[0].split(":")[1].strip().replace("$", ""))  
# 💥 crashes if format is slightly different

This works in demos and breaks in production. The solution: make the LLM return guaranteed-parseable JSON.

OpenAI Strict Structured Output

from openai import AsyncOpenAI
from pydantic import BaseModel, Field
from typing import Optional

client = AsyncOpenAI()

# Define your schema as a Pydantic model: field descriptions become instructions to the LLM
class ProductExtraction(BaseModel):
    product_name: str = Field(description="Full product name")
    price_usd: float = Field(description="Price in USD (convert if in other currency)")
    category: str = Field(description="Product category")
    in_stock: bool = Field(description="Whether product is currently available")
    discount_percent: Optional[float] = Field(
        default=None,
        # Optional[float] tells the model to return null when no discount exists, not 0.0
        description="Discount percentage if on sale, else null"
    )
    specifications: dict[str, str] = Field(
        default_factory=dict,
        description="Key technical specifications as key-value pairs"
    )

async def extract_product(text: str) -> ProductExtraction:
    response = await client.beta.chat.completions.parse(
        model="gpt-5.6-sol",  # all current GPT-5.6 tiers support strict structured output natively
        messages=[
            {"role": "system", "content": "Extract structured product data from the provided text."},
            {"role": "user", "content": text}
        ],
        # response_format=ProductExtraction: OpenAI converts the Pydantic model to a JSON schema
        # and uses constrained decoding to guarantee the output matches it exactly
        response_format=ProductExtraction,
    )
    
    # client.beta.chat.completions.parse() validates the response against the Pydantic model before returning
    # response.choices[0].message.parsed is already a ProductExtraction instance: no json.loads() needed
    return response.choices[0].message.parsed

LangChain with_structured_output

Works across providers with a unified interface:

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from pydantic import BaseModel, Field, field_validator
from typing import Literal

class SentimentAnalysis(BaseModel):
    # Literal constrains the model to one of these exact string values: no free-form text
    sentiment: Literal["positive", "negative", "neutral", "mixed"]
    # ge/le are Pydantic validators: ge=0.0 means "greater than or equal to 0.0"
    confidence: float = Field(ge=0.0, le=1.0)
    # min_length/max_length apply to the list length, not string length
    key_phrases: list[str] = Field(min_length=1, max_length=5)
    reasoning: str

# with_structured_output() adds the JSON schema to the API call and auto-parses the response
# Works with both OpenAI and Anthropic behind the same interface
openai_extractor = ChatOpenAI(model="gpt-5.6-sol").with_structured_output(
    SentimentAnalysis,
    method="json_schema",  # uses strict mode, guarantees schema compliance, not just best-effort
)

# Anthropic's tool_calling method is used by default; no method arg needed
anthropic_extractor = ChatAnthropic(model="claude-sonnet-5").with_structured_output(
    SentimentAnalysis,
)

# ainvoke() returns a SentimentAnalysis instance directly: no parsing step in caller code
result: SentimentAnalysis = await openai_extractor.ainvoke(
    "This product is incredible! Best purchase I've made all year."
)
print(f"Sentiment: {result.sentiment}, Confidence: {result.confidence:.0%}")

Complex Nested Extraction

# Author is a nested Pydantic model: ResearchPaper.authors will be a list of Author instances
class Author(BaseModel):
    name: str
    affiliation: Optional[str] = None  # None default means "omit if not found in the document"
    is_corresponding: bool = False

class ResearchPaper(BaseModel):
    title: str
    authors: list[Author] = Field(min_length=1)  # at least one author is required
    abstract: str = Field(max_length=2000)        # cap prevents the model from dumping the full paper
    publication_year: int = Field(ge=1900, le=2030)
    journal: Optional[str] = None
    doi: Optional[str] = None
    keywords: list[str] = Field(default_factory=list)
    methodology: Literal["experimental", "theoretical", "review", "case_study", "survey"] 
    key_findings: list[str] = Field(min_length=1, max_length=10)
    citations_count: Optional[int] = Field(default=None, ge=0)
    
    # @field_validator runs after the model parses the raw JSON: adds business-logic validation
    @field_validator('doi')
    @classmethod
    def validate_doi_format(cls, v: str | None) -> str | None:
        # All valid DOIs start with "10.": reject anything that looks malformed
        if v and not v.startswith("10."):
            raise ValueError("DOI must start with '10.'")
        return v

paper_extractor = ChatOpenAI(model="gpt-5.6-sol").with_structured_output(ResearchPaper)

async def extract_paper(pdf_text: str) -> ResearchPaper:
    return await paper_extractor.ainvoke([
        # "Only extract information explicitly stated" reduces hallucination into optional fields
        {"role": "system", "content": "Extract structured metadata from this research paper. Only extract information explicitly stated in the text."},
        {"role": "user", "content": pdf_text[:8000]}  # truncate to avoid context limit overruns
    ])

Classification with Discriminated Unions

For extraction where the schema type depends on the data:

from pydantic import BaseModel, Field
from typing import Annotated, Union, Literal

# Each intent class uses Literal["<tag>"] on its `type` field: this is the discriminator key
class EmailIntent(BaseModel):
    type: Literal["email"]
    subject: str
    to: list[str]
    body: str

class MeetingIntent(BaseModel):
    type: Literal["meeting"]
    title: str
    attendees: list[str]
    duration_minutes: int
    proposed_times: list[str]

class TaskIntent(BaseModel):
    type: Literal["task"]
    title: str
    description: str
    due_date: Optional[str] = None
    assignee: Optional[str] = None

class UserIntentExtraction(BaseModel):
    # discriminator="type" tells Pydantic which sub-model to instantiate based on the "type" value
    # the model sees all three schemas and picks the right one from context
    intent: Union[EmailIntent, MeetingIntent, TaskIntent] = Field(
        discriminator="type",
        description="The user's intent"
    )
    urgency: Literal["low", "medium", "high"]
    requires_confirmation: bool

intent_extractor = ChatOpenAI(model="gpt-5.6-sol").with_structured_output(UserIntentExtraction)

async def extract_user_intent(user_message: str) -> UserIntentExtraction:
    return await intent_extractor.ainvoke(
        f"Extract the intent from this message: {user_message}"
    )

# Usage
result = await extract_user_intent(
    "Can you set up a 30-minute meeting with Sarah and John about the Q4 report?"
)
# Pydantic validates the discriminated union: result.intent is guaranteed to be a MeetingIntent here
assert result.intent.type == "meeting"
assert isinstance(result.intent, MeetingIntent)
print(f"Duration: {result.intent.duration_minutes} mins")

Post-processing and Transformation

Raw extraction often needs business logic post-processing:

from datetime import datetime

# RawExtraction uses plain strings: lets the LLM capture the text as-is without guessing format
class RawExtraction(BaseModel):
    raw_date: str  # "January 15th, 2024" or "15/01/24" or "2024-01-15"
    raw_amount: str  # "$1,500" or "1.5k" or "fifteen hundred"
    raw_status: str  # "shipped" or "in transit" or "on the way"

# NormalizedRecord holds the canonical, typed representation after post-processing
class NormalizedRecord(BaseModel):
    date: datetime
    amount_cents: int  # store money as integer cents to avoid float precision issues
    status: Literal["pending", "shipped", "delivered", "cancelled"]

def normalize_extraction(raw: RawExtraction) -> NormalizedRecord:
    # Parse date (simplified: use dateparser in production for arbitrary natural language dates)
    date = datetime.fromisoformat(raw.raw_date)
    
    # Parse amount to cents: handles "$1,500", "1.5k", "1500" formats
    amount_str = raw.raw_amount.replace("$", "").replace(",", "").strip()
    if amount_str.endswith("k"):
        # "1.5k" → 1500.0 → 150000 cents
        amount_cents = int(float(amount_str[:-1]) * 1000 * 100)
    else:
        amount_cents = int(float(amount_str) * 100)
    
    # Map free-text statuses to canonical enum values: many phrases map to "shipped"
    status_map = {
        "shipped": "shipped", "in transit": "shipped", "on the way": "shipped",
        "delivered": "delivered", "received": "delivered",
        "cancelled": "cancelled", "canceled": "cancelled",
        "pending": "pending", "processing": "pending",
    }
    # Default to "pending" if the LLM returned a status phrase not in the map
    status = status_map.get(raw.raw_status.lower(), "pending")
    
    return NormalizedRecord(date=date, amount_cents=amount_cents, status=status)

Structured output combined with post-processing validation gives you a reliable pipeline from unstructured text to typed, validated, business-logic-aware data objects that the rest of your application can trust.