Structured outputs, JSON mode, and response validation

Why this matters

Unstructured text responses are fine for a chat interface. Every other LLM feature — classifiers, extractors, form-filling assistants, API-connected workflows — needs the model to return data your code can act on. The gap between "the model usually returns JSON" and "my application reliably parses and validates the response" is where most production LLM bugs live.

This lesson covers the three approaches to structured output, when to use each, how to validate every layer, and how to build a model fallback chain so a primary model failure does not bring down your feature.

Core concepts

JSON mode. Both OpenAI and Anthropic support a JSON mode that constrains the model to return syntactically valid JSON. The API-level flag differs by provider:

# OpenAI JSON mode
response = await openai_client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    response_format={"type": "json_object"},  # OpenAI
    max_tokens=512,
)

# Anthropic does not have a direct JSON mode flag —
# instead, prefill the assistant turn with an opening brace
messages_with_prefill = [
    *messages,
    # Anthropic allows starting the assistant response with a prefix
]
# Alternatively, use tool use (preferred)

JSON mode gives you valid JSON syntax but zero schema enforcement. You still need to validate the response shape with Pydantic.

Tool use / function calling (preferred approach). Define your expected output as a JSON Schema and force the model to use it:

from pydantic import BaseModel
from typing import Literal

class TicketClassification(BaseModel):
    category: Literal["billing", "technical", "account", "other"]
    priority: int  # 1-5
    sentiment: Literal["positive", "neutral", "negative"]
    requires_escalation: bool
    one_line_summary: str

# Convert Pydantic model to JSON Schema for the tool definition
CLASSIFICATION_SCHEMA = {
    "name": "classify_ticket",
    "description": "Classify a customer support ticket into structured categories.",
    "input_schema": {
        "type": "object",
        "properties": {
            "category": {
                "type": "string",
                "enum": ["billing", "technical", "account", "other"],
                "description": "The primary issue category"
            },
            "priority": {
                "type": "integer",
                "minimum": 1,
                "maximum": 5,
                "description": "Priority 1 (lowest) to 5 (highest)"
            },
            "sentiment": {
                "type": "string",
                "enum": ["positive", "neutral", "negative"]
            },
            "requires_escalation": {"type": "boolean"},
            "one_line_summary": {
                "type": "string",
                "maxLength": 120,
                "description": "One sentence capturing the core issue"
            },
        },
        "required": ["category", "priority", "sentiment", "requires_escalation", "one_line_summary"],
    },
}

async def classify_ticket(ticket_text: str) -> TicketClassification | None:
    response = await anthropic_client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=256,
        tools=[CLASSIFICATION_SCHEMA],
        tool_choice={"type": "tool", "name": "classify_ticket"},
        messages=[{"role": "user", "content": f"Classify this support ticket:\n\n{ticket_text}"}],
    )

    for block in response.content:
        if block.type == "tool_use" and block.name == "classify_ticket":
            try:
                return TicketClassification(**block.input)
            except Exception:
                return None

    return None

The tool_choice={"type": "tool", "name": "classify_ticket"} forces the model to use your tool rather than responding in free text. Combined with Pydantic validation, this is the most reliable approach for production.

OpenAI structured outputs (newer). OpenAI's newer API supports response_format={"type": "json_schema", "json_schema": {...}} with strict schema enforcement. This is equivalent to forced function calling with a cleaner API surface:

from openai.lib._pydantic import to_strict_json_schema

response = await openai_client.beta.chat.completions.parse(
    model="gpt-4o-mini",
    messages=messages,
    response_format=TicketClassification,  # pass the Pydantic class directly
)
result = response.choices[0].message.parsed  # already a TicketClassification instance

The .beta.chat.completions.parse method handles the schema generation and Pydantic deserialization automatically. This is the cleanest path for OpenAI-only integrations.

Defense-in-depth parsing. No single layer is sufficient for production. Use all three:

import json
import re
import logging
from pydantic import ValidationError

logger = logging.getLogger(__name__)

def extract_json_block(text: str) -> dict | None:
    """Three-layer JSON extraction from model text output."""
    # Layer 1: try direct parse
    text = text.strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass

    # Layer 2: strip markdown code fences
    fence_match = re.search(r"```(?:json)?\s*(.+?)\s*```", text, re.DOTALL)
    if fence_match:
        try:
            return json.loads(fence_match.group(1))
        except json.JSONDecodeError:
            pass

    # Layer 3: find the outermost JSON object
    obj_match = re.search(r"(\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\})", text, re.DOTALL)
    if obj_match:
        try:
            return json.loads(obj_match.group(1))
        except json.JSONDecodeError:
            pass

    return None


def parse_structured_output(
    raw_text: str,
    model_class: type[BaseModel],
    fallback: BaseModel | None = None,
) -> BaseModel | None:
    """Parse model output with validation. Returns fallback on failure."""
    raw_dict = extract_json_block(raw_text)
    if raw_dict is None:
        logger.warning("json_extraction_failed", extra={"text_preview": raw_text[:100]})
        return fallback

    try:
        return model_class.model_validate(raw_dict)
    except ValidationError as exc:
        logger.warning("pydantic_validation_failed", extra={"errors": exc.errors(), "raw": raw_dict})
        return fallback

Model fallback chains. When a primary model fails (rate limit, timeout, or provider outage), fall back to an alternative:

from dataclasses import dataclass

@dataclass
class ModelConfig:
    provider: str
    model: str
    max_tokens: int = 1024
    is_fallback: bool = False


FALLBACK_CHAIN = [
    ModelConfig(provider="anthropic", model="claude-sonnet-4-5"),
    ModelConfig(provider="anthropic", model="claude-haiku-4-5", is_fallback=True),
    ModelConfig(provider="openai", model="gpt-4o-mini", is_fallback=True),
]


async def generate_with_fallback(
    system: str,
    messages: list[dict],
    clients: dict[str, ProviderClient],
    chain: list[ModelConfig] = FALLBACK_CHAIN,
) -> tuple[LLMResult, ModelConfig]:
    last_error = None

    for config in chain:
        try:
            client = clients[config.provider]
            result = await client.generate(
                system=system,
                messages=messages,
                model=config.model,
                max_tokens=config.max_tokens,
            )
            if config.is_fallback:
                logger.warning("fallback_model_used", extra={"model": config.model})
            return result, config
        except Exception as exc:
            last_error = exc
            logger.warning(
                "model_failed_trying_next",
                extra={"model": config.model, "error": str(exc)},
            )
            continue

    raise RuntimeError(f"All models in fallback chain failed. Last error: {last_error}")

The fallback chain starts with the best model and degrades to cheaper alternatives. Each failure is logged so you can see when fallbacks are being triggered.

Working example

A production-grade classifier that combines tool use, Pydantic validation, and fallback:

async def classify_with_fallback(
    ticket_text: str,
    clients: dict[str, ProviderClient],
) -> TicketClassification:
    system = "You are a support ticket classifier. Always use the classify_ticket tool."
    messages = [{"role": "user", "content": f"Classify:\n\n{ticket_text}"}]

    result, config = await generate_with_fallback(system, messages, clients)

    # For tool use, the result.text is the raw tool_use block —
    # in a real implementation, extract from the full response object
    parsed = parse_structured_output(result.text, TicketClassification)

    if parsed is None:
        logger.error("classification_parse_failed", extra={"model": config.model})
        # Return a safe default rather than raising
        return TicketClassification(
            category="other",
            priority=3,
            sentiment="neutral",
            requires_escalation=False,
            one_line_summary="Classification failed — needs manual review",
        )

    return parsed

Common mistakes

Trusting tool use output without Pydantic validation. Tool use is more reliable than free-text parsing, but the model can still produce unexpected field values — a priority of "high" instead of 4, or a category value not in your enum. Always run through Pydantic.

No fallback for parse failures. A ValidationError that propagates to the user is a broken feature. Every structured output path needs either a fallback or a clean error response.

Overly complex JSON schemas. Deeply nested schemas with many optional fields confuse models and produce lower-quality outputs. Keep schemas flat: five to ten fields with clear types and descriptions is more reliable than twenty fields with nesting.

Forgetting to set tool_choice. Without tool_choice={"type": "tool", "name": "..."}, the model may choose to respond in plain text. Force the tool call to get deterministic structured output.

Using structured output for everything. Classification and extraction benefit from structured output. Open-ended reasoning, creative writing, and explanation tasks do not. Forcing a schema onto a free-text task produces worse results.

Try it yourself

1. Build a TicketClassifier class that uses tool use on Anthropic and OpenAI structured outputs. Verify both paths produce valid TicketClassification objects on a sample of 10 real support tickets.
2. Implement generate_with_fallback and write a test that injects a mock primary client that always raises RateLimitError and a mock fallback that succeeds. Verify the fallback is used and the warning is logged.
3. Write a fuzz test that passes 20 malformed model responses (wrong types, missing fields, markdown fences, JSON with prose) to parse_structured_output. Verify all cases return the fallback rather than raising.