4. Output Validation with Pydantic

Description

The LLM can produce malformed JSON, missing fields, incorrect types, or out-of-range values. Pydantic validates the output against a schema defined with types and constraints. But validation isn't just "Pydantic validates or fails" — there are three fallback strategies when it fails: retry, partial parsing, and default. This capsule covers schemas, validators, the three fallback strategies, and how to use OpenAI structured outputs to reduce errors at the source.

The problem it solves

Without output validation, these cases break your app:

# Case 1: Truncated JSON (insufficient max_tokens)
raw = '{"sentiment": "positive", "score": 0.92, "keywords": ["incred'
# JSONDecodeError → your app crashes

# Case 2: Field with incorrect type
raw = '{"sentiment": "positive", "score": "high", "keywords": []}'
# Parse ok, but result["score"] is a string, not a float
# result["score"] * 2 → TypeError in downstream code

# Case 3: Out-of-range field (the LLM hallucinated the value)
raw = '{"sentiment": "positive", "score": 1.5, "keywords": []}'
# Passes JSON parse but score=1.5 violates the 0-1 contract

# Case 4: Unexpected extra field (may not matter, or it may)
raw = '{"sentiment": "positive", "score": 0.9, "keywords": [], "password": "secret"}'
# Does your app expose the "password" in the response to the frontend?

# Case 5: Completely unexpected structure
raw = '{"error": "I didn\'t understand the instruction", "message": "Please repeat"}'
# The LLM didn't follow the format — you need to detect it and handle it

Basic schema with Pydantic

# src/guardrails/output_validator.py
from pydantic import BaseModel, field_validator, model_validator
from typing import Literal, Optional
from enum import Enum

class SentimentOutput(BaseModel):
    """Schema for the sentiment analyzer output."""
    
    # Field 1: Categorical (only 3 allowed values)
    sentiment: Literal["positive", "negative", "neutral"]
    
    # Field 2: Float with a constraint
    score: float
    
    # Field 3: String with a length constraint
    explanation: str
    
    # Field 4: List of strings (can be empty)
    keywords: list[str]
    
    # ─── Validators ──────────────────────────────────────────────
    
    @field_validator("score")
    @classmethod
    def score_must_be_in_range(cls, v: float) -> float:
        if not (0.0 <= v <= 1.0):
            raise ValueError(f"score must be between 0.0 and 1.0, received: {v}")
        return round(v, 4)  # Normalize to 4 decimals
    
    @field_validator("explanation")
    @classmethod
    def explanation_must_have_content(cls, v: str) -> str:
        v = v.strip()
        if len(v) > 500:
            v = v[:500]  # Silently truncate if it's too long
        return v
    
    @field_validator("keywords")
    @classmethod
    def keywords_must_be_strings(cls, v: list) -> list[str]:
        cleaned = []
        for kw in v:
            if isinstance(kw, str) and kw.strip():
                cleaned.append(kw.strip()[:50])  # Max 50 chars per keyword
        return cleaned[:10]  # Max 10 keywords
    
    # Model configuration
    model_config = {
        "extra": "ignore"  # Ignore extra fields — don't fail if there are more fields
    }

Advanced validators

# Validators with more complex logic:

class SummaryOutput(BaseModel):
    summary: str
    confidence: float
    word_count: int
    sources: Optional[list[str]] = None
    
    @field_validator("summary")
    @classmethod
    def summary_length_check(cls, v: str) -> str:
        v = v.strip()
        if len(v) < 10:
            raise ValueError(f"summary too short: {len(v)} chars")
        if len(v) > 2000:
            raise ValueError(f"summary too long: {len(v)} chars")
        return v
    
    @field_validator("confidence")
    @classmethod
    def confidence_range(cls, v: float) -> float:
        if not (0.0 <= v <= 1.0):
            raise ValueError(f"confidence out of range: {v}")
        return v
    
    @model_validator(mode="after")
    def word_count_matches_summary(self) -> "SummaryOutput":
        """Cross-field validator: checks consistency between fields."""
        actual_word_count = len(self.summary.split())
        # The LLM reported word_count — verify it's approximately correct
        if abs(actual_word_count - self.word_count) > 20:
            # Silently correct
            self.word_count = actual_word_count
        return self

The three fallback strategies

When Pydantic fails, there are three strategies. Choose based on the use case:

Strategy 1: Retry (for transient errors)

import time
from pydantic import ValidationError

def validate_with_retry(
    raw: str,
    schema: type[BaseModel],
    llm_callable,
    max_retries: int = 2,
    retry_delay: float = 1.0
) -> BaseModel | None:
    """
    Validates the output and retries if it fails.
    
    When to use: When the error is probably transient
    (truncated JSON, slightly different format).
    
    When NOT to use: When the error is systematic (the LLM never
    produces this format) — retrying won't help.
    """
    # Try to parse the current raw
    try:
        return schema.model_validate_json(raw)
    except (ValidationError, Exception) as first_error:
        pass
    
    # Retry with the LLM (reformulating the prompt)
    for attempt in range(max_retries):
        try:
            retry_prompt = f"Please reformat your response as valid JSON with this exact structure: {schema.model_json_schema()}"
            new_raw = llm_callable(retry_prompt)
            return schema.model_validate_json(new_raw)
        except Exception:
            if attempt < max_retries - 1:
                time.sleep(retry_delay)
    
    return None  # All attempts failed

Strategy 2: Default (for persistent errors)

def validate_with_default(
    raw: str,
    schema: type[BaseModel],
    default: BaseModel
) -> BaseModel:
    """
    Validates the output and returns a default if it fails.
    
    When to use: When the system must keep working
    even if the LLM fails. The default is a "safe fallback".
    
    When NOT to use: When the failure is critical and returning
    a default could mislead the user.
    """
    try:
        return schema.model_validate_json(raw)
    except Exception as e:
        import logging
        logging.getLogger("guardrails").warning(
            "validation_failed_using_default",
            extra={"error": str(e), "raw_truncated": raw[:100]}
        )
        return default

# Usage:
DEFAULT_SENTIMENT = SentimentOutput(
    sentiment="neutral",
    score=0.5,
    explanation="Could not analyze the sentiment of the text.",
    keywords=[]
)

result = validate_with_default(
    raw=llm_response,
    schema=SentimentOutput,
    default=DEFAULT_SENTIMENT
)

Strategy 3: Partial parsing (for complex outputs)

import json
import re

def extract_and_validate(
    raw: str,
    schema: type[BaseModel],
    required_fields: list[str] = None
) -> BaseModel | None:
    """
    Tries multiple strategies to extract valid JSON.
    
    Pipeline:
    1. Direct JSON
    2. JSON in a markdown code block
    3. JSON searched for in free text
    4. Partial parsing with only the required fields
    
    When to use: When the LLM includes JSON inside a longer text,
    or when you only need some fields from the output.
    """
    # Strategy 1: Parse directly
    try:
        return schema.model_validate_json(raw.strip())
    except Exception:
        pass
    
    # Strategy 2: Extract from a markdown code block
    markdown_match = re.search(r'```(?:json)?\s*\n?(.*?)\n?```', raw, re.DOTALL)
    if markdown_match:
        try:
            return schema.model_validate_json(markdown_match.group(1).strip())
        except Exception:
            pass
    
    # Strategy 3: Find the first JSON object in the text
    json_match = re.search(r'\{.*\}', raw, re.DOTALL)
    if json_match:
        try:
            return schema.model_validate_json(json_match.group())
        except Exception:
            pass
    
    # Strategy 4: Partial parsing — build with only the available fields
    try:
        data = {}
        # Try to extract key fields with individual regexes
        for field_name, field_info in schema.model_fields.items():
            # Look for the field in the text as "field_name": value
            pattern = f'"{field_name}"\\s*:\\s*([^,}}]+)'
            match = re.search(pattern, raw)
            if match:
                try:
                    value = json.loads(match.group(1).strip().rstrip(',}'))
                    data[field_name] = value
                except Exception:
                    pass
        
        if required_fields and not all(f in data for f in required_fields):
            return None
        
        return schema.model_validate(data)
    except Exception:
        return None

OpenAI Structured Outputs: reduce errors at the source

# Option 1: JSON mode (ensures valid JSON, not a specific structure)
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{
        "role": "user",
        "content": "Analyze the sentiment. Respond with JSON: {sentiment, score, explanation, keywords}"
    }],
    response_format={"type": "json_object"},  # ← Guarantees valid JSON
    temperature=0.0
)
# With this: we eliminate JSONDecodeError. Schema validation can still fail.

# Option 2: Structured outputs with a schema (gpt-4o, more robust)
# With this approach the model guarantees it follows the schema:
from pydantic import BaseModel
import openai

response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",  # Requires gpt-4o with a specific date
    messages=[{
        "role": "user",
        "content": "Analyze: 'I love this product'"
    }],
    response_format=SentimentOutput,  # ← The Pydantic schema directly
)
result = response.choices[0].message.parsed  # ← Already a validated SentimentOutput

Complete output validation function

# src/guardrails/output_validator.py

def validate_llm_output(
    raw: str,
    schema: type[BaseModel],
    strategy: str = "extract_and_default",
    default: BaseModel | None = None,
    llm_callable = None,
    max_retries: int = 1
) -> BaseModel | None:
    """
    Validates the LLM output with the specified strategy.
    
    Strategies:
    - "strict": Only perfect JSON. Fails otherwise.
    - "extract": Tries to extract JSON from free text.
    - "retry": Retries with the LLM if it fails.
    - "extract_and_default": Extracts, and uses a default if it fails. (Recommended)
    """
    import logging
    logger = logging.getLogger("guardrails.validator")
    
    if strategy == "strict":
        return schema.model_validate_json(raw)
    
    if strategy in ("extract", "extract_and_default"):
        result = extract_and_validate(raw, schema)
        if result:
            return result
        
        if strategy == "extract_and_default" and default:
            logger.warning(
                "output_validation_failed_using_default",
                extra={"raw_preview": raw[:100]}
            )
            return default
        return None
    
    if strategy == "retry" and llm_callable:
        return validate_with_retry(raw, schema, llm_callable, max_retries)
    
    return None

Complete validator tests

# tests/unit/guardrails/test_output_validator.py
import pytest
from pydantic import ValidationError
from src.guardrails.output_validator import SentimentOutput, validate_llm_output

class TestSentimentOutputSchema:
    """Tests for the Pydantic schema."""
    
    def test_valid_output_passes(self):
        data = {
            "sentiment": "positive",
            "score": 0.85,
            "explanation": "The text uses positive language",
            "keywords": ["excellent", "recommended"]
        }
        result = SentimentOutput(**data)
        assert result.sentiment == "positive"
        assert result.score == 0.85
    
    def test_invalid_sentiment_fails(self):
        with pytest.raises(ValidationError):
            SentimentOutput(sentiment="very positive", score=0.9, explanation="OK", keywords=[])
    
    def test_score_above_1_fails(self):
        with pytest.raises(ValidationError):
            SentimentOutput(sentiment="positive", score=1.5, explanation="OK", keywords=[])
    
    def test_score_below_0_fails(self):
        with pytest.raises(ValidationError):
            SentimentOutput(sentiment="positive", score=-0.1, explanation="OK", keywords=[])
    
    def test_extra_fields_ignored(self):
        """model_config = extra: ignore — extra fields must not cause an error."""
        result = SentimentOutput(
            sentiment="positive",
            score=0.8,
            explanation="OK",
            keywords=[],
            unexpected_field="value"  # Extra field
        )
        assert result.sentiment == "positive"
        assert not hasattr(result, "unexpected_field")
    
    def test_keywords_cleaned(self):
        result = SentimentOutput(
            sentiment="positive",
            score=0.8,
            explanation="OK",
            keywords=["  keyword1  ", "", "  ", "valid"]
        )
        assert "keyword1" in result.keywords  # Trimmed
        assert "" not in result.keywords     # Empty removed

class TestValidateLlmOutput:
    """Tests for the complete validation function."""
    
    def test_valid_json_passes(self):
        raw = '{"sentiment":"positive","score":0.9,"explanation":"OK","keywords":[]}'
        result = validate_llm_output(raw, SentimentOutput)
        assert result is not None
        assert result.sentiment == "positive"
    
    def test_json_in_markdown_extracted(self):
        raw = '```json\n{"sentiment":"positive","score":0.9,"explanation":"OK","keywords":[]}\n```'
        result = validate_llm_output(raw, SentimentOutput, strategy="extract")
        assert result is not None
    
    def test_invalid_json_with_default(self):
        from src.guardrails.output_validator import DEFAULT_SENTIMENT
        raw = "This is a free-text response, not JSON."
        result = validate_llm_output(
            raw, SentimentOutput,
            strategy="extract_and_default",
            default=DEFAULT_SENTIMENT
        )
        # Should return the default, not None
        assert result is not None
        assert result.sentiment == "neutral"  # The default

    def test_truncated_json_handled(self):
        """Truncated JSON must not crash — it must handle it gracefully."""
        raw = '{"sentiment": "positive", "score": 0.9, "explanation": "OK", "keywo'
        result = validate_llm_output(raw, SentimentOutput, strategy="extract_and_default")
        # Must not raise an exception — may return None or the default
        # assert result is None  # or assert result == DEFAULT_SENTIMENT

Exercises

Exercise 1: Schema for classification

Define a Pydantic schema for document classification:

• category: one of ["technology", "science", "sports", "politics", "entertainment"]
• confidence: float 0-1
• tags: list of strings (maximum 5)
• language: optional string (may be absent)

from pydantic import BaseModel, field_validator
from typing import Literal, Optional

class ClassificationOutput(BaseModel):
    category: Literal["technology", "science", "sports", "politics", "entertainment"]
    confidence: float
    tags: list[str]
    language: Optional[str] = None
    
    @field_validator("confidence")
    @classmethod
    def confidence_range(cls, v):
        if not (0 <= v <= 1):
            raise ValueError(f"confidence out of range: {v}")
        return v
    
    @field_validator("tags")
    @classmethod
    def max_tags(cls, v):
        return v[:5]  # Maximum 5 tags
    
    model_config = {"extra": "ignore"}

Exercise 2: Fallback strategy

For a user-facing /analyze endpoint, which fallback strategy would you use and why?

DEFAULT_SENTIMENT = SentimentOutput(
    sentiment="neutral",
    score=0.5,
    explanation="It was not possible to analyze the sentiment at this time.",
    keywords=[]
)

Exercise 3: JSON in markdown

The LLM returns this response:

Here is my analysis:

```json
{"sentiment": "positive", "score": 0.85, "explanation": "Positive text", "keywords": ["good"]}

Anything else?

Write the code to extract and validate the JSON from this response:

<details>
<summary>See solution</summary>

```python
import re
import json
from pydantic import ValidationError

def extract_from_markdown(raw: str) -> SentimentOutput | None:
    # Look for JSON in a code block
    match = re.search(r'```(?:json)?\s*\n?(.*?)\n?```', raw, re.DOTALL)
    if match:
        try:
            return SentimentOutput.model_validate_json(match.group(1).strip())
        except (ValidationError, Exception):
            return None
    return None

# Test:
raw = '''Here is my analysis:
```json
{"sentiment": "positive", "score": 0.85, "explanation": "Positive text", "keywords": ["good"]}

Anything else?'''

result = extract_from_markdown(raw) assert result is not None assert result.sentiment == "positive"

</details>

---

### Exercise 4: Cross-field model validator

Write a `model_validator` that checks that if `sentiment == "positive"`, the `score` must be >= 0.5 (and vice versa for "negative"):

<details>
<summary>See solution</summary>

```python
from pydantic import model_validator

class SentimentOutputStrict(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"]
    score: float
    explanation: str
    keywords: list[str]
    
    @model_validator(mode="after")
    def sentiment_score_consistency(self):
        if self.sentiment == "positive" and self.score < 0.5:
            raise ValueError(
                f"Inconsistent: sentiment='positive' but score={self.score} < 0.5"
            )
        if self.sentiment == "negative" and self.score > 0.5:
            raise ValueError(
                f"Inconsistent: sentiment='negative' but score={self.score} > 0.5"
            )
        return self

Summary

• Pydantic validates structure, types, and constraints — the first line of defense for the LLM output
• Three fallback strategies: retry (transient errors), default (persistent errors), partial parsing (embedded JSON)
• model_config = {"extra": "ignore"}: avoid failures from unexpected extra fields
• extract_and_validate: handle JSON in markdown code blocks (very common in LLMs)
• OpenAI Structured Outputs / JSON mode: reduce errors at the source
• model_validator cross-field: validate consistency between fields when a single one isn't enough

Additional resources

1. Pydantic V2 Documentation — Complete reference
2. Pydantic field_validator — Per-field validators
3. Pydantic model_validator — Cross-field validators
4. OpenAI Structured Outputs — For guaranteed outputs
5. OpenAI JSON Mode — Basic JSON mode
6. Pydantic model_json_schema — Export the schema to use it in prompts