3. Output Validation with Pydantic

Overview

Your LLM generates free text. Sometimes that text is exactly what you expected — clean JSON with the right fields, appropriate types, and values within range. Other times, the model produces incomplete JSON, incorrect types, fields it made up, or plain text when you asked for structure. The model has no contract — it doesn't guarantee it will meet your schema. If your frontend, your database, or your downstream API expects a specific structure and receives something different, you have a production bug.

Output validation with Pydantic turns "I hope the model responds well" into "I verify that the response meets my schema before using it." It's the equivalent of using strict types in a strongly-typed language — except here the "compiler" is you, running validation at runtime against the probabilistic output of an LLM.

In this capsule you build the second component of the Sanitization Pipeline: an Output Validator that uses Pydantic schemas, OpenAI structured outputs, retry strategies, and fallback chains to guarantee that every model output is valid before it reaches the user or a downstream system.

Why LLM outputs need validation

The LLM has no contract

In a traditional REST API, the server has an explicit contract (OpenAPI spec) and produces deterministic outputs. If you define that /users returns {name: string, age: number}, that's what it returns, always.

An LLM has no such guarantee:

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{
        "role": "user",
        "content": "Give me the price of the iPhone 15 in JSON format with fields: product, price, currency"
    }],
    temperature=0.7,
)

print(response.choices[0].message.content)
# Possible outputs (all valid for the model):
#
# Attempt 1: {"product": "iPhone 15", "price": 799, "currency": "USD"}  ← perfect
# Attempt 2: {"product": "iPhone 15", "price": "$799"}  ← missing currency, price is a string
# Attempt 3: Sure, here you go: {"product": "iPhone 15"...}  ← text + partial JSON
# Attempt 4: {"producto": "iPhone 15", "precio": 799}  ← fields in Spanish
# Attempt 5: I don't have access to up-to-date prices.  ← refusal without JSON

Without validation, your code breaks on attempts 2-5. With validation, you detect the problem and take action.

Pydantic Basics for AI outputs

Schema definition

from pydantic import BaseModel, Field, field_validator
from typing import Optional
from enum import Enum


class Currency(str, Enum):
    USD = "USD"
    EUR = "EUR"
    MXN = "MXN"
    GBP = "GBP"


class ProductPrice(BaseModel):
    product: str = Field(min_length=1, max_length=200)
    price: float = Field(gt=0, le=100000)
    currency: Currency
    in_stock: bool = True
    description: Optional[str] = Field(None, max_length=500)

    @field_validator("product")
    @classmethod
    def clean_product_name(cls, v: str) -> str:
        return v.strip()

    @field_validator("price")
    @classmethod
    def round_price(cls, v: float) -> float:
        return round(v, 2)


valid_data = {"product": "iPhone 15", "price": 799.99, "currency": "USD"}
product = ProductPrice(**valid_data)
print(f"Valid: {product}")

try:
    bad_data = {"product": "", "price": -50, "currency": "BITCOIN"}
    ProductPrice(**bad_data)
except Exception as e:
    print(f"Validation error: {e}")

# Expected output:
# Valid: product='iPhone 15' price=799.99 currency=<Currency.USD: 'USD'> in_stock=True description=None
# Validation error: 3 validation errors for ProductPrice
#   product: String should have at least 1 character
#   price: Input should be greater than 0
#   currency: Input should be 'USD', 'EUR', 'MXN' or 'GBP'

Parsing LLM output with Pydantic

import json
import re
from pydantic import BaseModel, Field, ValidationError
from typing import Optional


class LLMResponse(BaseModel):
    answer: str = Field(min_length=1, max_length=2000)
    confidence: float = Field(ge=0.0, le=1.0)
    sources: list[str] = Field(default_factory=list, max_length=10)


def extract_json_from_text(text: str) -> Optional[dict]:
    """Extracts JSON from text that may include markdown or additional text."""
    json_pattern = re.compile(r"```(?:json)?\s*([\s\S]*?)```")
    match = json_pattern.search(text)
    if match:
        try:
            return json.loads(match.group(1).strip())
        except json.JSONDecodeError:
            pass

    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass

    brace_pattern = re.compile(r"\{[\s\S]*\}")
    match = brace_pattern.search(text)
    if match:
        try:
            return json.loads(match.group())
        except json.JSONDecodeError:
            pass

    return None


def validate_llm_output(raw_output: str, schema: type[BaseModel]) -> dict:
    """Validates an LLM output against a Pydantic schema."""
    extracted = extract_json_from_text(raw_output)

    if extracted is None:
        return {
            "valid": False,
            "error": "No JSON found in output",
            "raw": raw_output[:200],
            "data": None,
        }

    try:
        validated = schema(**extracted)
        return {
            "valid": True,
            "error": None,
            "data": validated.model_dump(),
        }
    except ValidationError as e:
        return {
            "valid": False,
            "error": str(e),
            "data": extracted,
        }


test_outputs = [
    '{"answer": "The iPhone 15 costs $799", "confidence": 0.95, "sources": ["apple.com"]}',
    'Here you go: ```json\n{"answer": "Price: $799", "confidence": 0.8}\n```',
    "I don't have that information available.",
    '{"answer": "", "confidence": 1.5}',
]

for output in test_outputs:
    result = validate_llm_output(output, LLMResponse)
    print(f"Input:  {output[:60]}...")
    print(f"Valid:  {result['valid']}")
    if not result["valid"]:
        print(f"Error:  {str(result['error'])[:80]}")
    print()

# Expected output:
# Input:  {"answer": "The iPhone 15 costs $799", "confidence": 0.95, "...
# Valid:  True
#
# Input:  Here you go: ```json...
# Valid:  True
#
# Input:  I don't have that information available....
# Valid:  False
# Error:  No JSON found in output
#
# Input:  {"answer": "", "confidence": 1.5}...
# Valid:  False
# Error:  2 validation errors...

OpenAI Structured Outputs

OpenAI offers structured outputs as a native feature, where the model is forced to produce JSON that meets a specific schema. This significantly reduces format errors.

from openai import OpenAI
from pydantic import BaseModel, Field

client = OpenAI()


class ProductAnalysis(BaseModel):
    product_name: str = Field(description="Product name")
    category: str = Field(description="Category: electronics, clothing, food, other")
    sentiment: str = Field(description="Sentiment: positive, negative, neutral")
    key_points: list[str] = Field(description="3-5 key points of the analysis")
    score: float = Field(ge=1.0, le=10.0, description="Score from 1 to 10")


response = client.beta.chat.completions.parse(
    model="gpt-4o-mini",
    messages=[
        {
            "role": "system",
            "content": "Analyze the product mentioned by the user."
        },
        {
            "role": "user",
            "content": "The iPhone 15 has an amazing camera but the battery could be better."
        },
    ],
    response_format=ProductAnalysis,
)

analysis = response.choices[0].message.parsed
print(f"Product: {analysis.product_name}")
print(f"Category: {analysis.category}")
print(f"Sentiment: {analysis.sentiment}")
print(f"Score: {analysis.score}")
print(f"Key points: {analysis.key_points}")

# Expected output:
# Product: iPhone 15
# Category: electronics
# Sentiment: positive
# Score: 7.5
# Key points: ['High-quality camera', 'Battery could be better', ...]

Structured outputs vs manual validation

Recommendation: Use structured outputs when you're on OpenAI and need a format guarantee. Use manual Pydantic validation as a fallback and for providers that don't support structured outputs.

Retry Strategies for invalid outputs

When the output doesn't pass validation, you need a strategy:

import time
from openai import OpenAI
from pydantic import BaseModel, Field, ValidationError
from typing import Optional

client = OpenAI()


class RetryConfig(BaseModel):
    max_retries: int = 3
    retry_delay_seconds: float = 0.5
    escalate_prompt: bool = True
    fallback_response: Optional[dict] = None


class OutputValidator:
    def __init__(self, config: RetryConfig = RetryConfig()):
        self.config = config
        self.attempt_log: list[dict] = []

    def validate_with_retry(
        self,
        messages: list[dict],
        schema: type[BaseModel],
        model: str = "gpt-4o-mini",
    ) -> dict:
        for attempt in range(1, self.config.max_retries + 1):
            current_messages = messages.copy()

            if attempt > 1 and self.config.escalate_prompt:
                last_error = self.attempt_log[-1].get("error", "")
                current_messages.append({
                    "role": "user",
                    "content": (
                        f"Your previous response was not valid. "
                        f"Error: {last_error[:200]}. "
                        f"Please respond ONLY with valid JSON "
                        f"that meets the schema. No additional text."
                    ),
                })

            try:
                response = client.chat.completions.create(
                    model=model,
                    messages=current_messages,
                    temperature=max(0, 0.7 - (attempt * 0.2)),
                )
                raw_output = response.choices[0].message.content

                extracted = extract_json_from_text(raw_output)
                if extracted is None:
                    raise ValueError("No JSON found in output")

                validated = schema(**extracted)
                self.attempt_log.append({
                    "attempt": attempt,
                    "success": True,
                })
                return {
                    "success": True,
                    "data": validated.model_dump(),
                    "attempts": attempt,
                }

            except (ValidationError, ValueError, Exception) as e:
                self.attempt_log.append({
                    "attempt": attempt,
                    "success": False,
                    "error": str(e)[:200],
                })
                if attempt < self.config.max_retries:
                    time.sleep(self.config.retry_delay_seconds)

        if self.config.fallback_response:
            return {
                "success": False,
                "data": self.config.fallback_response,
                "attempts": self.config.max_retries,
                "used_fallback": True,
            }

        return {
            "success": False,
            "data": None,
            "attempts": self.config.max_retries,
            "errors": [log["error"] for log in self.attempt_log if not log["success"]],
        }


# Example usage
class SimpleAnswer(BaseModel):
    answer: str = Field(min_length=1)
    confidence: float = Field(ge=0.0, le=1.0)


validator = OutputValidator(
    config=RetryConfig(
        max_retries=3,
        fallback_response={"answer": "I couldn't process your question.", "confidence": 0.0},
    )
)

result = validator.validate_with_retry(
    messages=[
        {"role": "system", "content": "Respond in JSON with fields: answer (string), confidence (0-1)."},
        {"role": "user", "content": "What is the capital of France?"},
    ],
    schema=SimpleAnswer,
)

print(f"Success: {result['success']}")
print(f"Attempts: {result['attempts']}")
print(f"Data: {result['data']}")

# Expected output:
# Success: True
# Attempts: 1
# Data: {'answer': 'The capital of France is Paris.', 'confidence': 0.99}

Decreasing-temperature strategy

Notice how the retry lowers the temperature: 0.7 → 0.5 → 0.3. This makes each attempt more "deterministic" and more likely to follow the exact schema. It's an effective technique for structured outputs.

Nested Validation for complex outputs

AI outputs are often nested structures. Pydantic handles this naturally:

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


class Source(BaseModel):
    url: str = Field(min_length=1)
    title: str = Field(min_length=1)
    reliability: float = Field(ge=0.0, le=1.0)


class FactCheck(BaseModel):
    claim: str
    verified: bool
    evidence: Optional[str] = None


class AnalysisResult(BaseModel):
    summary: str = Field(min_length=10, max_length=2000)
    key_findings: list[str] = Field(min_length=1, max_length=10)
    sources: list[Source] = Field(min_length=1, max_length=5)
    fact_checks: list[FactCheck] = Field(default_factory=list)
    overall_confidence: float = Field(ge=0.0, le=1.0)
    generated_at: str = Field(default_factory=lambda: datetime.now().isoformat())

    @field_validator("key_findings")
    @classmethod
    def validate_findings(cls, v: list[str]) -> list[str]:
        return [f.strip() for f in v if len(f.strip()) > 0]


test_data = {
    "summary": "The analysis shows that the product has good market reception.",
    "key_findings": ["High customer satisfaction", "Competitive price", "Battery could be better"],
    "sources": [
        {"url": "https://example.com/review", "title": "Official review", "reliability": 0.9},
    ],
    "fact_checks": [
        {"claim": "Best camera on the market", "verified": False, "evidence": "Samsung S24 has better zoom"},
    ],
    "overall_confidence": 0.85,
}

result = AnalysisResult(**test_data)
print(f"Valid: {result.summary[:50]}...")
print(f"Findings: {len(result.key_findings)}")
print(f"Sources: {len(result.sources)}")
print(f"Confidence: {result.overall_confidence}")

# Expected output:
# Valid: The analysis shows that the product has good marke...
# Findings: 3
# Sources: 1
# Confidence: 0.85

Custom Validators for AI Outputs

AI outputs have problems that normal data doesn't have. You need specific validators:

from pydantic import BaseModel, Field, field_validator, model_validator
import re


class AIOutput(BaseModel):
    answer: str = Field(min_length=1, max_length=2000)
    reasoning: str = Field(min_length=1, max_length=1000)
    confidence: float = Field(ge=0.0, le=1.0)

    @field_validator("answer")
    @classmethod
    def no_system_prompt_leak(cls, v: str) -> str:
        leak_patterns = [
            r"you are an? (?:ai|assistant|helpful)",
            r"your (?:instructions|system prompt|rules)",
            r"as an ai (?:language )?model",
            r"i (?:was|am) (?:programmed|instructed|told) to",
        ]
        for pattern in leak_patterns:
            if re.search(pattern, v.lower()):
                raise ValueError(
                    f"Output may contain system prompt leak: matched '{pattern}'"
                )
        return v

    @field_validator("answer")
    @classmethod
    def no_code_injection(cls, v: str) -> str:
        dangerous_patterns = [
            r"<script[\s>]",
            r"javascript:",
            r"on\w+\s*=",
            r"eval\s*\(",
            r"exec\s*\(",
            r"__import__\s*\(",
        ]
        for pattern in dangerous_patterns:
            if re.search(pattern, v.lower()):
                raise ValueError(
                    f"Output contains potentially dangerous code: '{pattern}'"
                )
        return v

    @model_validator(mode="after")
    def confidence_matches_reasoning(self) -> "AIOutput":
        reasoning_lower = self.reasoning.lower()
        hedging_words = ["perhaps", "maybe", "not sure", "quizás", "tal vez", "no estoy seguro"]
        has_hedging = any(word in reasoning_lower for word in hedging_words)

        if has_hedging and self.confidence > 0.8:
            self.confidence = min(self.confidence, 0.6)
        return self


# Test
valid = AIOutput(
    answer="The capital of France is Paris.",
    reasoning="Well-established factual fact.",
    confidence=0.99,
)
print(f"Valid: {valid.answer}, confidence: {valid.confidence}")

hedging = AIOutput(
    answer="The capital might be Paris.",
    reasoning="I'm not sure of the exact answer.",
    confidence=0.95,
)
print(f"Hedging: confidence adjusted to {hedging.confidence}")

try:
    AIOutput(
        answer='Look at this: <script>alert("xss")</script>',
        reasoning="I generated an example.",
        confidence=0.5,
    )
except Exception as e:
    print(f"Blocked: {e}")

# Expected output:
# Valid: The capital of France is Paris., confidence: 0.99
# Hedging: confidence adjusted to 0.6
# Blocked: 1 validation error... Output contains potentially dangerous code

Type Coercion for imprecise outputs

LLMs often return "almost correct" types — a number as a string, a boolean as "yes/no", a list as a comma-separated string:

from pydantic import BaseModel, Field, field_validator
from typing import Union


class FlexibleOutput(BaseModel):
    count: int
    available: bool
    tags: list[str]
    score: float

    @field_validator("count", mode="before")
    @classmethod
    def coerce_count(cls, v):
        if isinstance(v, str):
            cleaned = v.replace(",", "").strip()
            try:
                return int(float(cleaned))
            except ValueError:
                raise ValueError(f"Cannot coerce '{v}' to int")
        return v

    @field_validator("available", mode="before")
    @classmethod
    def coerce_bool(cls, v):
        if isinstance(v, str):
            truthy = {"yes", "true", "sí", "si", "1", "available", "disponible"}
            falsy = {"no", "false", "0", "unavailable", "no disponible"}
            lower = v.lower().strip()
            if lower in truthy:
                return True
            if lower in falsy:
                return False
            raise ValueError(f"Cannot coerce '{v}' to bool")
        return v

    @field_validator("tags", mode="before")
    @classmethod
    def coerce_tags(cls, v):
        if isinstance(v, str):
            return [tag.strip() for tag in v.split(",") if tag.strip()]
        return v

    @field_validator("score", mode="before")
    @classmethod
    def coerce_score(cls, v):
        if isinstance(v, str):
            cleaned = v.replace("%", "").strip()
            try:
                val = float(cleaned)
                if val > 1 and "%" not in v:
                    return val
                if val > 1:
                    return val / 100
                return val
            except ValueError:
                raise ValueError(f"Cannot coerce '{v}' to float")
        return v


messy_llm_output = {
    "count": "1,234",
    "available": "yes",
    "tags": "electronics, apple, smartphone",
    "score": "85%",
}

result = FlexibleOutput(**messy_llm_output)
print(f"Count: {result.count} (type: {type(result.count).__name__})")
print(f"Available: {result.available} (type: {type(result.available).__name__})")
print(f"Tags: {result.tags} (type: {type(result.tags).__name__})")
print(f"Score: {result.score} (type: {type(result.score).__name__})")

# Expected output:
# Count: 1234 (type: int)
# Available: True (type: bool)
# Tags: ['electronics', 'apple', 'smartphone'] (type: list)
# Score: 0.85 (type: float)

Handling partial/malformed outputs

LLMs can produce incomplete outputs (due to max_tokens) or malformed ones (truncated JSON). You need to handle these cases:

import json
from typing import Optional


def repair_partial_json(raw: str) -> Optional[dict]:
    """Tries to repair partial JSON produced by truncation."""
    cleaned = raw.strip()
    if cleaned.startswith("```"):
        lines = cleaned.split("\n")
        cleaned = "\n".join(lines[1:])
        if cleaned.endswith("```"):
            cleaned = cleaned[:-3]
        cleaned = cleaned.strip()

    try:
        return json.loads(cleaned)
    except json.JSONDecodeError:
        pass

    open_braces = cleaned.count("{")
    close_braces = cleaned.count("}")
    missing = open_braces - close_braces
    if missing > 0:
        last_comma = cleaned.rfind(",")
        last_colon = cleaned.rfind(":")
        if last_comma > last_colon:
            cleaned = cleaned[:last_comma]
        cleaned += "}" * missing
        try:
            return json.loads(cleaned)
        except json.JSONDecodeError:
            pass

    open_brackets = cleaned.count("[")
    close_brackets = cleaned.count("]")
    if open_brackets > close_brackets:
        cleaned += "]" * (open_brackets - close_brackets)
        if open_braces > cleaned.count("}"):
            cleaned += "}" * (open_braces - cleaned.count("}"))
        try:
            return json.loads(cleaned)
        except json.JSONDecodeError:
            pass

    return None


partial_outputs = [
    '{"answer": "Paris", "confidence": 0.9',
    '{"answer": "Madrid", "items": ["a", "b", "c"',
    '```json\n{"answer": "Tokyo"}\n```',
    '{"answer": "Berlin", "score": ',
]

for output in partial_outputs:
    repaired = repair_partial_json(output)
    print(f"Input:    {output!r}")
    print(f"Repaired: {repaired}")
    print()

# Expected output:
# Input:    '{"answer": "Paris", "confidence": 0.9'
# Repaired: {'answer': 'Paris', 'confidence': 0.9}
#
# Input:    '{"answer": "Madrid", "items": ["a", "b", "c"'
# Repaired: None
#
# Input:    '```json\n{"answer": "Tokyo"}\n```'
# Repaired: {'answer': 'Tokyo'}
#
# Input:    '{"answer": "Berlin", "score": '
# Repaired: None

OutputValidator: the complete class

Integrating everything into a reusable class for the Sanitization Pipeline:

import json
import re
import time
from pydantic import BaseModel, Field, ValidationError
from typing import Optional, Any
from dataclasses import dataclass, field
from enum import Enum


class ValidationAction(Enum):
    VALID = "valid"
    COERCED = "coerced"
    REPAIRED = "repaired"
    FALLBACK = "fallback"
    FAILED = "failed"


@dataclass
class ValidationResult:
    action: ValidationAction
    data: Optional[dict]
    raw_output: str
    issues: list[str] = field(default_factory=list)
    attempts: int = 1

    @property
    def success(self) -> bool:
        return self.action in (
            ValidationAction.VALID,
            ValidationAction.COERCED,
            ValidationAction.REPAIRED,
        )


class OutputValidator:
    def __init__(
        self,
        schema: type[BaseModel],
        fallback: Optional[dict] = None,
        allow_repair: bool = True,
        allow_coercion: bool = True,
    ):
        self.schema = schema
        self.fallback = fallback
        self.allow_repair = allow_repair
        self.allow_coercion = allow_coercion

    def validate(self, raw_output: str) -> ValidationResult:
        issues: list[str] = []

        # Step 1: Try direct parse
        extracted = extract_json_from_text(raw_output)

        if extracted is None and self.allow_repair:
            extracted = repair_partial_json(raw_output)
            if extracted:
                issues.append("JSON repaired from partial output")

        if extracted is None:
            if self.fallback:
                return ValidationResult(
                    action=ValidationAction.FALLBACK,
                    data=self.fallback,
                    raw_output=raw_output,
                    issues=["No JSON found, using fallback"],
                )
            return ValidationResult(
                action=ValidationAction.FAILED,
                data=None,
                raw_output=raw_output,
                issues=["No JSON found in output"],
            )

        # Step 2: Validate against schema
        try:
            validated = self.schema(**extracted)
            action = (
                ValidationAction.REPAIRED if issues
                else ValidationAction.VALID
            )
            return ValidationResult(
                action=action,
                data=validated.model_dump(),
                raw_output=raw_output,
                issues=issues,
            )
        except ValidationError as e:
            issues.append(f"Validation errors: {e.error_count()}")
            if self.fallback:
                return ValidationResult(
                    action=ValidationAction.FALLBACK,
                    data=self.fallback,
                    raw_output=raw_output,
                    issues=issues + [str(e)[:200]],
                )
            return ValidationResult(
                action=ValidationAction.FAILED,
                data=None,
                raw_output=raw_output,
                issues=issues + [str(e)[:200]],
            )


# --- Demonstration ---

class ChatResponse(BaseModel):
    answer: str = Field(min_length=1, max_length=2000)
    confidence: float = Field(ge=0.0, le=1.0, default=0.5)


validator = OutputValidator(
    schema=ChatResponse,
    fallback={"answer": "I couldn't process your question.", "confidence": 0.0},
    allow_repair=True,
)

test_cases = [
    '{"answer": "Paris is the capital of France.", "confidence": 0.95}',
    '{"answer": "Madrid", "confidence": 0.8',
    "The capital is Paris.",
    '{"answer": "", "confidence": 2.0}',
]

for raw in test_cases:
    result = validator.validate(raw)
    print(f"Raw:    {raw[:60]}")
    print(f"Action: {result.action.value}")
    print(f"Data:   {result.data}")
    if result.issues:
        print(f"Issues: {result.issues}")
    print()

# Expected output:
# Raw:    {"answer": "Paris is the capital of France.", "confidence": 0.9
# Action: valid
# Data:   {'answer': 'Paris is the capital of France.', 'confidence': 0.95}
#
# Raw:    {"answer": "Madrid", "confidence": 0.8
# Action: repaired
# Data:   {'answer': 'Madrid', 'confidence': 0.8}
# Issues: ['JSON repaired from partial output']
#
# Raw:    The capital is Paris.
# Action: fallback
# Data:   {'answer': "I couldn't process your question.", 'confidence': 0.0}
# Issues: ['No JSON found, using fallback']
#
# Raw:    {"answer": "", "confidence": 2.0}
# Action: fallback
# Data:   {'answer': "I couldn't process your question.", 'confidence': 0.0}
# Issues: ['Validation errors: 2', ...]

Troubleshooting

Problem 1: "OpenAI structured outputs add latency"

Structured outputs require OpenAI to process the schema and constrain the generation. This can add 100-500ms.

Solution: Use structured outputs for calls where structure is critical (data for the DB, API responses). For conversational chat, use post-generation Pydantic validation that doesn't affect the LLM's latency.

Problem 2: "The model ignores the schema requested in the prompt"

Sometimes the model generates free text even when you ask for JSON. This happens more with small models and high temperatures.

Solution: Combine: (1) explicit instructions in the system prompt, (2) low temperature (0.0-0.3), (3) an example in the prompt, (4) retry with a stricter prompt. If you use OpenAI, structured outputs solve this directly.

Problem 3: "Retries cost tokens and money"

Each retry is an additional call to the LLM that costs tokens.

Solution: Limit retries to 2-3. Use decreasing temperature. Keep a fallback response that doesn't require an LLM. Monitor your retry rate — if it's > 10%, your prompt needs improvements, not more retries.

Problem 4: "Pydantic custom validators are slow"

Field validators with regex run for each field. In outputs with many fields, this accumulates.

Solution: Precompile regex patterns as class constants. Use model_validator(mode="after") for validations that need multiple fields — it runs once, not per field.

Exercises

Exercise 1: Schema for multi-step reasoning

Create a Pydantic schema that validates outputs from an LLM that does step-by-step reasoning (chain-of-thought), where each step must reference the previous one.

from pydantic import BaseModel, Field, model_validator


class ReasoningStep(BaseModel):
    step_number: int = Field(ge=1)
    description: str = Field(min_length=5)
    references_step: int = Field(ge=0, default=0)

class ChainOfThought(BaseModel):
    question: str = Field(min_length=5)
    steps: list[ReasoningStep] = Field(min_length=1, max_length=10)
    conclusion: str = Field(min_length=5)
    confidence: float = Field(ge=0.0, le=1.0)

    @model_validator(mode="after")
    def validate_step_references(self) -> "ChainOfThought":
        step_numbers = {s.step_number for s in self.steps}
        for step in self.steps:
            if step.references_step > 0 and step.references_step not in step_numbers:
                raise ValueError(
                    f"Step {step.step_number} references non-existent step {step.references_step}"
                )
            if step.references_step >= step.step_number:
                raise ValueError(
                    f"Step {step.step_number} cannot reference future step {step.references_step}"
                )
        return self


data = {
    "question": "What is 2+2*3?",
    "steps": [
        {"step_number": 1, "description": "Identify the multiplication operation: 2*3=6"},
        {"step_number": 2, "description": "Add to the result: 2+6=8", "references_step": 1},
    ],
    "conclusion": "The result is 8",
    "confidence": 0.99,
}

result = ChainOfThought(**data)
print(f"Steps: {len(result.steps)}, Conclusion: {result.conclusion}")

# Expected output:
# Steps: 2, Conclusion: The result is 8

Exercise 2: Validator that detects simple hallucinations

Create a validator that detects when the model claims facts with high confidence but uses hedging language.

from pydantic import BaseModel, Field, model_validator
import re

HEDGING_PATTERNS = [
    r"\b(perhaps|maybe|might|could be|possibly)\b",
    r"\b(quizás|tal vez|podría ser|posiblemente)\b",
    r"\b(i think|i believe|creo que|pienso que)\b",
    r"\b(not sure|no estoy seguro|uncertain)\b",
    r"\b(approximately|roughly|around|cerca de|aproximadamente)\b",
]


class FactualClaim(BaseModel):
    claim: str = Field(min_length=5)
    confidence: float = Field(ge=0.0, le=1.0)
    is_factual: bool = True

    @model_validator(mode="after")
    def check_hedging_vs_confidence(self) -> "FactualClaim":
        hedging_count = 0
        for pattern in HEDGING_PATTERNS:
            if re.search(pattern, self.claim.lower()):
                hedging_count += 1

        if hedging_count > 0 and self.confidence > 0.8:
            self.confidence = max(0.3, self.confidence - (hedging_count * 0.2))
            self.is_factual = False
        return self


test1 = FactualClaim(claim="Paris is the capital of France.", confidence=0.99)
test2 = FactualClaim(claim="I think perhaps the answer might be 42.", confidence=0.95)

print(f"Claim 1: confidence={test1.confidence}, factual={test1.is_factual}")
print(f"Claim 2: confidence={test2.confidence}, factual={test2.is_factual}")

# Expected output:
# Claim 1: confidence=0.99, factual=True
# Claim 2: confidence=0.5499999999999999, factual=False

Exercise 3: Output validator with a fallback chain

Implement a chained fallback system: try structured output → Pydantic parse → JSON repair → fallback response.

from pydantic import BaseModel, Field, ValidationError
from typing import Optional


class Answer(BaseModel):
    text: str = Field(min_length=1)
    score: float = Field(ge=0.0, le=1.0, default=0.5)


def fallback_chain(raw_output: str) -> dict:
    """Tries multiple strategies in order."""
    strategies_tried = []

    # Strategy 1: Direct Pydantic parse
    extracted = extract_json_from_text(raw_output)
    if extracted:
        try:
            result = Answer(**extracted)
            return {"data": result.model_dump(), "strategy": "direct_parse", "tried": 1}
        except ValidationError:
            strategies_tried.append("direct_parse_failed")

    # Strategy 2: JSON repair
    repaired = repair_partial_json(raw_output)
    if repaired:
        try:
            result = Answer(**repaired)
            return {"data": result.model_dump(), "strategy": "json_repair", "tried": 2}
        except ValidationError:
            strategies_tried.append("repair_failed")

    # Strategy 3: Extract text as answer
    clean_text = raw_output.strip()
    if len(clean_text) > 0:
        try:
            result = Answer(text=clean_text, score=0.3)
            return {"data": result.model_dump(), "strategy": "text_extract", "tried": 3}
        except ValidationError:
            strategies_tried.append("text_extract_failed")

    # Strategy 4: Fallback
    return {
        "data": {"text": "I couldn't process the response.", "score": 0.0},
        "strategy": "fallback",
        "tried": 4,
    }


tests = [
    '{"text": "Paris", "score": 0.9}',
    '{"text": "Madrid", "score": 0.8',
    "The answer is Tokyo.",
    "",
]

for t in tests:
    result = fallback_chain(t)
    print(f"Input: {t[:40]!r} → Strategy: {result['strategy']} (tried: {result['tried']})")

# Expected output:
# Input: '{"text": "Paris", "score": 0.9}' → Strategy: direct_parse (tried: 1)
# Input: '{"text": "Madrid", "score": 0.8' → Strategy: json_repair (tried: 2)
# Input: 'The answer is Tokyo.' → Strategy: text_extract (tried: 3)
# Input: '' → Strategy: fallback (tried: 4)

Exercise 4: Versioned schema for output evolution

Implement a system that supports multiple versions of a schema and migrates data between versions.

from pydantic import BaseModel, Field
from typing import Optional


class AnswerV1(BaseModel):
    text: str
    schema_version: int = 1

class AnswerV2(BaseModel):
    text: str
    confidence: float = Field(ge=0.0, le=1.0, default=0.5)
    schema_version: int = 2

class AnswerV3(BaseModel):
    text: str
    confidence: float = Field(ge=0.0, le=1.0, default=0.5)
    sources: list[str] = Field(default_factory=list)
    schema_version: int = 3


SCHEMAS = {1: AnswerV1, 2: AnswerV2, 3: AnswerV3}
CURRENT_VERSION = 3

def migrate_to_current(data: dict) -> dict:
    version = data.get("schema_version", 1)
    while version < CURRENT_VERSION:
        if version == 1:
            data["confidence"] = 0.5
            data["schema_version"] = 2
        elif version == 2:
            data["sources"] = []
            data["schema_version"] = 3
        version = data["schema_version"]
    return data


old_data = {"text": "Hello", "schema_version": 1}
migrated = migrate_to_current(old_data)
result = AnswerV3(**migrated)
print(f"Migrated: {result.model_dump()}")

# Expected output:
# Migrated: {'text': 'Hello', 'confidence': 0.5, 'sources': [], 'schema_version': 3}

Summary

• 🔑 LLM outputs have no contract — the model can produce any format, type, or structure regardless of what you ask for
• 🔑 Pydantic schemas are the main tool for validating outputs: they define the expected structure and reject outputs that don't comply
• 🔑 OpenAI structured outputs enforce the schema at the model level, eliminating most format errors (but they only work with OpenAI)
• 🔑 Retry strategies with decreasing temperature are effective for getting valid outputs without a fallback: first attempt at 0.7, second at 0.5, third at 0.3
• 🔑 Partial JSON repair recovers outputs truncated by max_tokens, saving API calls
• 🔑 Custom validators for AI outputs detect specific problems: system prompt leaks, code injection, hedging inconsistent with confidence
• 🔑 Type coercion handles the LLM's natural imprecision: "yes" → True, "1,234" → 1234, "85%" → 0.85
• 🔑 The fallback chain guarantees that you always produce a valid response: direct parse → repair → text extract → fallback
• 🔑 The OutputValidator is the second piece of the Sanitization Pipeline and connects with the Content Filter from capsule 04

Additional resources

1. Pydantic V2 Documentation — Complete Pydantic reference for data validation, validators, and schemas
2. OpenAI Structured Outputs — Official guide to getting structured outputs directly from the model
3. Pydantic — Field Validators — Custom validators for validation logic specific to your domain
4. OpenAI Function Calling — Alternative to structured outputs for extracting structured data
5. JSON Schema Specification — The standard Pydantic bases its schemas on, useful for interoperability
6. OWASP LLM05: Improper Output Handling — The vulnerability that output validation directly mitigates
7. Instructor Library — Library that simplifies structured outputs with Pydantic for multiple LLM providers
8. Pydantic Settings Management — Configuration management with Pydantic, useful for configuring validators

Created: March 2026 Version: 1.0