4. Output Validation con Pydantic

Descripción

El LLM puede producir JSON malformado, campos faltantes, tipos incorrectos, o valores fuera de rango. Pydantic valida el output contra un schema definido con tipos y constraints. Pero la validación no es solo "Pydantic valida o falla" — hay tres estrategias de fallback cuando falla: retry, parseo parcial, y default. Esta cápsula cubre schemas, validators, las tres estrategias de fallback, y cómo usar structured outputs de OpenAI para reducir errores desde el origen.

El problema que resuelve

Sin validación de output, estos casos rompen tu app:

# Caso 1: JSON truncado (max_tokens insuficiente)
raw = '{"sentiment": "positivo", "score": 0.92, "keywords": ["increí'
# JSONDecodeError → tu app crashea

# Caso 2: Campo con tipo incorrecto
raw = '{"sentiment": "positivo", "score": "alta", "keywords": []}'
# Parseo ok, pero result["score"] es un string, no float
# result["score"] * 2 → TypeError en código downstream

# Caso 3: Campo fuera de rango (LLM alucinó el valor)
raw = '{"sentiment": "positivo", "score": 1.5, "keywords": []}'
# Pasa JSON parse pero score=1.5 viola el contrato 0-1

# Caso 4: Campo extra inesperado (puede no importar, o puede)
raw = '{"sentiment": "positivo", "score": 0.9, "keywords": [], "password": "secret"}'
# ¿Tu app expone el "password" en el response al frontend?

# Caso 5: Estructura completamente inesperada
raw = '{"error": "No entendí la instrucción", "message": "Repite por favor"}'
# El LLM no siguió el formato — necesitas detectarlo y manejar

Schema básico con 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 para el output del analizador de sentimiento."""
    
    # Campo 1: Categorical (solo 3 valores permitidos)
    sentiment: Literal["positivo", "negativo", "neutral"]
    
    # Campo 2: Float con constraint
    score: float
    
    # Campo 3: String con constraint de longitud
    explanation: str
    
    # Campo 4: Lista de strings (puede estar vacía)
    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 debe estar entre 0.0 y 1.0, recibido: {v}")
        return round(v, 4)  # Normalizar a 4 decimales
    
    @field_validator("explanation")
    @classmethod
    def explanation_must_have_content(cls, v: str) -> str:
        v = v.strip()
        if len(v) > 500:
            v = v[:500]  # Truncar silenciosamente si es demasiado larga
        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 por keyword
        return cleaned[:10]  # Max 10 keywords
    
    # Configuración del modelo
    model_config = {
        "extra": "ignore"  # Ignorar campos extra — no fallar si hay más campos
    }

Validators avanzados

# Validators con lógica más compleja:

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 demasiado corto: {len(v)} chars")
        if len(v) > 2000:
            raise ValueError(f"summary demasiado largo: {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 fuera de rango: {v}")
        return v
    
    @model_validator(mode="after")
    def word_count_matches_summary(self) -> "SummaryOutput":
        """Validator cross-field: verifica consistencia entre campos."""
        actual_word_count = len(self.summary.split())
        # El LLM reportó word_count — verificar que es aproximadamente correcto
        if abs(actual_word_count - self.word_count) > 20:
            # Corregir silenciosamente
            self.word_count = actual_word_count
        return self

Las tres estrategias de fallback

Cuando Pydantic falla, hay tres estrategias. Elegir según el caso de uso:

Estrategia 1: Retry (para errores transitorios)

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:
    """
    Valida el output y retries si falla.
    
    Cuándo usar: Cuando el error es probablemente transitorio
    (JSON truncado, formato ligeramente diferente).
    
    Cuándo NO usar: Cuando el error es sistemático (el LLM nunca
    produce este formato) — el retry no ayudará.
    """
    # Intentar parsear el raw actual
    try:
        return schema.model_validate_json(raw)
    except (ValidationError, Exception) as first_error:
        pass
    
    # Retry con el LLM (reformulando el prompt)
    for attempt in range(max_retries):
        try:
            retry_prompt = f"Por favor reformatea tu respuesta como JSON válido con esta estructura exacta: {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  # Todos los intentos fallaron

Estrategia 2: Default (para errores persistentes)

def validate_with_default(
    raw: str,
    schema: type[BaseModel],
    default: BaseModel
) -> BaseModel:
    """
    Valida el output y retorna un default si falla.
    
    Cuándo usar: Cuando el sistema debe seguir funcionando
    aunque el LLM falle. El default es un "safe fallback".
    
    Cuándo NO usar: Cuando el fallo es crítico y devolver
    un default podría inducir a error al usuario.
    """
    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

# Uso:
DEFAULT_SENTIMENT = SentimentOutput(
    sentiment="neutral",
    score=0.5,
    explanation="No se pudo analizar el sentimiento del texto.",
    keywords=[]
)

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

Estrategia 3: Parseo parcial (para outputs complejos)

import json
import re

def extract_and_validate(
    raw: str,
    schema: type[BaseModel],
    required_fields: list[str] = None
) -> BaseModel | None:
    """
    Intenta múltiples estrategias para extraer JSON válido.
    
    Pipeline:
    1. JSON directo
    2. JSON en markdown code block
    3. JSON buscado en texto libre
    4. Parseo parcial con solo campos requeridos
    
    Cuándo usar: Cuando el LLM incluye JSON en un texto más largo,
    o cuando solo necesitas algunos campos del output.
    """
    # Estrategia 1: Parsear directamente
    try:
        return schema.model_validate_json(raw.strip())
    except Exception:
        pass
    
    # Estrategia 2: Extraer de 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
    
    # Estrategia 3: Encontrar primer objeto JSON en el texto
    json_match = re.search(r'\{.*\}', raw, re.DOTALL)
    if json_match:
        try:
            return schema.model_validate_json(json_match.group())
        except Exception:
            pass
    
    # Estrategia 4: Parseo parcial — construir con solo los campos disponibles
    try:
        data = {}
        # Intentar extraer campos clave con regex individuales
        for field_name, field_info in schema.model_fields.items():
            # Buscar el campo en el texto como "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: reducir errores desde el origen

# Opción 1: JSON mode (asegura JSON válido, no estructura específica)
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{
        "role": "user",
        "content": "Analiza el sentimiento. Responde con JSON: {sentiment, score, explanation, keywords}"
    }],
    response_format={"type": "json_object"},  # ← Garantiza JSON válido
    temperature=0.0
)
# Con esto: eliminamos JSONDecodeError. Todavía puede fallar validación de schema.

# Opción 2: Structured outputs con schema (gpt-4o, más robusto)
# Con este enfoque el modelo garantiza seguir el schema:
from pydantic import BaseModel
import openai

response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",  # Requiere gpt-4o con fecha específica
    messages=[{
        "role": "user",
        "content": "Analiza: 'Me encanta este producto'"
    }],
    response_format=SentimentOutput,  # ← El schema de Pydantic directamente
)
result = response.choices[0].message.parsed  # ← Ya es un SentimentOutput validado

Función completa de validación de output

# 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:
    """
    Valida el output del LLM con la estrategia especificada.
    
    Strategies:
    - "strict": Solo JSON perfecto. Falla si no.
    - "extract": Intenta extraer JSON de texto libre.
    - "retry": Reintenta con el LLM si falla.
    - "extract_and_default": Extrae, y usa default si falla. (Recomendado)
    """
    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

Tests completos del validador

# 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 del schema de Pydantic."""
    
    def test_valid_output_passes(self):
        data = {
            "sentiment": "positivo",
            "score": 0.85,
            "explanation": "El texto usa lenguaje positivo",
            "keywords": ["excelente", "recomendado"]
        }
        result = SentimentOutput(**data)
        assert result.sentiment == "positivo"
        assert result.score == 0.85
    
    def test_invalid_sentiment_fails(self):
        with pytest.raises(ValidationError):
            SentimentOutput(sentiment="muy positivo", score=0.9, explanation="OK", keywords=[])
    
    def test_score_above_1_fails(self):
        with pytest.raises(ValidationError):
            SentimentOutput(sentiment="positivo", score=1.5, explanation="OK", keywords=[])
    
    def test_score_below_0_fails(self):
        with pytest.raises(ValidationError):
            SentimentOutput(sentiment="positivo", score=-0.1, explanation="OK", keywords=[])
    
    def test_extra_fields_ignored(self):
        """model_config = extra: ignore — campos extra no deben causar error."""
        result = SentimentOutput(
            sentiment="positivo",
            score=0.8,
            explanation="OK",
            keywords=[],
            unexpected_field="valor"  # Campo extra
        )
        assert result.sentiment == "positivo"
        assert not hasattr(result, "unexpected_field")
    
    def test_keywords_cleaned(self):
        result = SentimentOutput(
            sentiment="positivo",
            score=0.8,
            explanation="OK",
            keywords=["  keyword1  ", "", "  ", "valid"]
        )
        assert "keyword1" in result.keywords  # Trimmed
        assert "" not in result.keywords     # Empty removed

class TestValidateLlmOutput:
    """Tests de la función de validación completa."""
    
    def test_valid_json_passes(self):
        raw = '{"sentiment":"positivo","score":0.9,"explanation":"OK","keywords":[]}'
        result = validate_llm_output(raw, SentimentOutput)
        assert result is not None
        assert result.sentiment == "positivo"
    
    def test_json_in_markdown_extracted(self):
        raw = '```json\n{"sentiment":"positivo","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 = "Esta es una respuesta en texto libre, no JSON."
        result = validate_llm_output(
            raw, SentimentOutput,
            strategy="extract_and_default",
            default=DEFAULT_SENTIMENT
        )
        # Debe retornar el default, no None
        assert result is not None
        assert result.sentiment == "neutral"  # El default

    def test_truncated_json_handled(self):
        """JSON truncado no debe crashear — debe manejar gracefully."""
        raw = '{"sentiment": "positivo", "score": 0.9, "explanation": "OK", "keywo'
        result = validate_llm_output(raw, SentimentOutput, strategy="extract_and_default")
        # No debe lanzar excepción — puede retornar None o default
        # assert result is None  # o assert result == DEFAULT_SENTIMENT

Ejercicios

Ejercicio 1: Schema para clasificación

Define un schema Pydantic para clasificación de documentos:

• category: uno de ["tecnología", "ciencia", "deportes", "política", "entretenimiento"]
• confidence: float 0-1
• tags: lista de strings (máximo 5)
• language: string opcional (puede no estar)

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

class ClassificationOutput(BaseModel):
    category: Literal["tecnología", "ciencia", "deportes", "política", "entretenimiento"]
    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 fuera de rango: {v}")
        return v
    
    @field_validator("tags")
    @classmethod
    def max_tags(cls, v):
        return v[:5]  # Máximo 5 tags
    
    model_config = {"extra": "ignore"}

Ejercicio 2: Estrategia de fallback

Para un endpoint /analyze de cara al usuario, ¿qué estrategia de fallback usarías y por qué?

DEFAULT_SENTIMENT = SentimentOutput(
    sentiment="neutral",
    score=0.5,
    explanation="No fue posible analizar el sentimiento en este momento.",
    keywords=[]
)

Ejercicio 3: JSON en markdown

El LLM retorna esta respuesta:

Aquí está mi análisis:

```json
{"sentiment": "positivo", "score": 0.85, "explanation": "Texto positivo", "keywords": ["bueno"]}

¿Necesitas algo más?

Escribe el código para extraer y validar el JSON de esta respuesta:

<details>
<summary>Ver solución</summary>

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

def extract_from_markdown(raw: str) -> SentimentOutput | None:
    # Buscar JSON en 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 = '''Aquí está mi análisis:
```json
{"sentiment": "positivo", "score": 0.85, "explanation": "Texto positivo", "keywords": ["bueno"]}

¿Necesitas algo más?'''

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

</details>

---

### Ejercicio 4: Model validator cross-field

Escribe un `model_validator` que verifique que si `sentiment == "positivo"`, el `score` debe ser >= 0.5 (y viceversa para "negativo"):

<details>
<summary>Ver solución</summary>

```python
from pydantic import model_validator

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

Resumen

• Pydantic valida estructura, tipos y constraints — la primera línea de defensa para el output del LLM
• Tres estrategias de fallback: retry (errores transitorios), default (errores persistentes), parseo parcial (JSON embebido)
• model_config = {"extra": "ignore"}: evitar fallos por campos extra inesperados
• extract_and_validate: manejar JSON en markdown code blocks (muy común en LLMs)
• OpenAI Structured Outputs / JSON mode: reducir errores desde el origen
• model_validator cross-field: validar consistencia entre campos cuando una sola no es suficiente

Recursos adicionales

1. Pydantic V2 Documentation — Referencia completa
2. Pydantic field_validator — Validators por campo
3. Pydantic model_validator — Validators cross-field
4. OpenAI Structured Outputs — Para outputs garantizados
5. OpenAI JSON Mode — JSON mode básico
6. Pydantic model_json_schema — Exportar schema para usarlo en prompts