3. Output Validation con Pydantic

Descripción

Tu LLM genera texto libre. A veces ese texto es exactamente lo que esperabas — un JSON limpio con los campos correctos, tipos adecuados, y valores dentro de rango. Otras veces, el modelo produce JSON incompleto, tipos incorrectos, campos que inventó, o directamente texto plano cuando pediste estructura. El modelo no tiene contrato — no garantiza cumplir tu schema. Si tu frontend, tu base de datos, o tu API downstream esperan una estructura específica y reciben algo diferente, tienes un bug en producción.

La validación de outputs con Pydantic convierte el "espero que el modelo responda bien" en "verifico que la respuesta cumple mi schema antes de usarla." Es el equivalente de usar tipos estrictos en un lenguaje con tipado fuerte — excepto que aquí el "compiler" eres tú, ejecutando la validación en runtime contra la salida probabilística de un LLM.

En esta cápsula construyes el segundo componente del Sanitization Pipeline: un Output Validator que usa Pydantic schemas, structured outputs de OpenAI, retry strategies, y fallback chains para garantizar que cada output del modelo sea válido antes de llegar al usuario o a un downstream system.

Por qué los outputs del LLM necesitan validación

El LLM no tiene contrato

En una API REST tradicional, el servidor tiene un contrato explícito (OpenAPI spec) y produce outputs determinísticos. Si defines que /users retorna {name: string, age: number}, eso es lo que retorna, siempre.

Un LLM no tiene esa garantía:

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{
        "role": "user",
        "content": "Dame el precio del iPhone 15 en formato JSON con campos: product, price, currency"
    }],
    temperature=0.7,
)

print(response.choices[0].message.content)
# Posibles outputs (todos válidos para el modelo):
#
# Intento 1: {"product": "iPhone 15", "price": 799, "currency": "USD"}  ← perfecto
# Intento 2: {"product": "iPhone 15", "price": "$799"}  ← falta currency, price es string
# Intento 3: Claro, aquí tienes: {"product": "iPhone 15"...}  ← texto + JSON parcial
# Intento 4: {"producto": "iPhone 15", "precio": 799}  ← campos en español
# Intento 5: No tengo acceso a precios actualizados.  ← rechazo sin JSON

Sin validación, tu código se rompe en los intentos 2-5. Con validación, detectas el problema y tomas acción.

Pydantic Basics para outputs AI

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}")

# Output esperado:
# 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 con 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]:
    """Extrae JSON de texto que puede incluir markdown o texto adicional."""
    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:
    """Valida un output del LLM contra un 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": "El iPhone 15 cuesta $799", "confidence": 0.95, "sources": ["apple.com"]}',
    'Aquí tienes: ```json\n{"answer": "Precio: $799", "confidence": 0.8}\n```',
    "No tengo esa información disponible.",
    '{"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()

# Output esperado:
# Input:  {"answer": "El iPhone 15 cuesta $799", "confidence": 0.95, "so...
# Valid:  True
#
# Input:  Aquí tienes: ```json...
# Valid:  True
#
# Input:  No tengo esa información disponible....
# Valid:  False
# Error:  No JSON found in output
#
# Input:  {"answer": "", "confidence": 1.5}...
# Valid:  False
# Error:  2 validation errors...

OpenAI Structured Outputs

OpenAI ofrece structured outputs como feature nativa, donde el modelo está forzado a producir JSON que cumple un schema específico. Esto reduce significativamente los errores de formato.

from openai import OpenAI
from pydantic import BaseModel, Field

client = OpenAI()


class ProductAnalysis(BaseModel):
    product_name: str = Field(description="Nombre del producto")
    category: str = Field(description="Categoría: electronics, clothing, food, other")
    sentiment: str = Field(description="Sentimiento: positive, negative, neutral")
    key_points: list[str] = Field(description="3-5 puntos clave del análisis")
    score: float = Field(ge=1.0, le=10.0, description="Puntuación de 1 a 10")


response = client.beta.chat.completions.parse(
    model="gpt-4o-mini",
    messages=[
        {
            "role": "system",
            "content": "Analiza el producto mencionado por el usuario."
        },
        {
            "role": "user",
            "content": "El iPhone 15 tiene una cámara increíble pero la batería podría ser mejor."
        },
    ],
    response_format=ProductAnalysis,
)

analysis = response.choices[0].message.parsed
print(f"Producto: {analysis.product_name}")
print(f"Categoría: {analysis.category}")
print(f"Sentimiento: {analysis.sentiment}")
print(f"Puntuación: {analysis.score}")
print(f"Puntos clave: {analysis.key_points}")

# Output esperado:
# Producto: iPhone 15
# Categoría: electronics
# Sentimiento: positive
# Puntuación: 7.5
# Puntos clave: ['Cámara de alta calidad', 'Batería mejorable', ...]

Structured outputs vs validación manual

Recomendación: Usa structured outputs cuando estés con OpenAI y necesites garantía de formato. Usa validación Pydantic manual como fallback y para proveedores que no soportan structured outputs.

Retry Strategies para outputs inválidos

Cuando el output no pasa la validación, necesitas una estrategia:

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"Tu respuesta anterior no fue válida. "
                        f"Error: {last_error[:200]}. "
                        f"Por favor, responde SOLO con JSON válido "
                        f"que cumpla el schema. Sin texto adicional."
                    ),
                })

            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"]],
        }


# Ejemplo de uso
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": "No pude procesar tu pregunta.", "confidence": 0.0},
    )
)

result = validator.validate_with_retry(
    messages=[
        {"role": "system", "content": "Responde en JSON con campos: answer (string), confidence (0-1)."},
        {"role": "user", "content": "¿Cuál es la capital de Francia?"},
    ],
    schema=SimpleAnswer,
)

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

# Output esperado:
# Success: True
# Attempts: 1
# Data: {'answer': 'La capital de Francia es París.', 'confidence': 0.99}

Estrategia de temperatura decreciente

Observa cómo el retry reduce la temperatura: 0.7 → 0.5 → 0.3. Esto hace que cada intento sea más "determinístico" y más probable que siga el schema exacto. Es una técnica efectiva para outputs estructurados.

Nested Validation para outputs complejos

Los outputs de AI suelen ser estructuras anidadas. Pydantic maneja esto naturalmente:

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": "El análisis muestra que el producto tiene buena recepción en el mercado.",
    "key_findings": ["Alta satisfacción del cliente", "Precio competitivo", "Batería mejorable"],
    "sources": [
        {"url": "https://example.com/review", "title": "Review oficial", "reliability": 0.9},
    ],
    "fact_checks": [
        {"claim": "Mejor cámara del mercado", "verified": False, "evidence": "Samsung S24 tiene mejor 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}")

# Output esperado:
# Valid: El análisis muestra que el producto tiene buena r...
# Findings: 3
# Sources: 1
# Confidence: 0.85

Custom Validators para AI Outputs

Los outputs de AI tienen problemas que los datos normales no tienen. Necesitas validators específicos:

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="La capital de Francia es París.",
    reasoning="Dato factual bien establecido.",
    confidence=0.99,
)
print(f"Valid: {valid.answer}, confidence: {valid.confidence}")

hedging = AIOutput(
    answer="La capital podría ser París.",
    reasoning="No estoy seguro de la respuesta exacta.",
    confidence=0.95,
)
print(f"Hedging: confidence adjusted to {hedging.confidence}")

try:
    AIOutput(
        answer='Mira esto: <script>alert("xss")</script>',
        reasoning="Generé un ejemplo.",
        confidence=0.5,
    )
except Exception as e:
    print(f"Blocked: {e}")

# Output esperado:
# Valid: La capital de Francia es París., confidence: 0.99
# Hedging: confidence adjusted to 0.6
# Blocked: 1 validation error... Output contains potentially dangerous code

Type Coercion para outputs imprecisos

Los LLMs a menudo retornan tipos "casi correctos" — un número como string, un booleano como "yes/no", una lista como string separado por comas:

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": "sí",
    "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__})")

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

Handling partial/malformed outputs

Los LLMs pueden producir outputs incompletos (por max_tokens) o malformados (JSON cortado). Necesitas manejar estos casos:

import json
from typing import Optional


def repair_partial_json(raw: str) -> Optional[dict]:
    """Intenta reparar JSON parcial producido por truncamiento."""
    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": "París", "confidence": 0.9',
    '{"answer": "Madrid", "items": ["a", "b", "c"',
    '```json\n{"answer": "Tokio"}\n```',
    '{"answer": "Berlin", "score": ',
]

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

# Output esperado:
# Input:    '{"answer": "París", "confidence": 0.9'
# Repaired: {'answer': 'París', 'confidence': 0.9}
#
# Input:    '{"answer": "Madrid", "items": ["a", "b", "c"'
# Repaired: {'answer': 'Madrid', 'items': ['a', 'b', 'c']}
#
# Input:    '```json\n{"answer": "Tokio"}\n```'
# Repaired: {'answer': 'Tokio'}
#
# Input:    '{"answer": "Berlin", "score": '
# Repaired: {'answer': 'Berlin'}

OutputValidator: la clase completa

Integrando todo en una clase reutilizable para el 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]],
            )


# --- Demostración ---

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": "No pude procesar tu pregunta.", "confidence": 0.0},
    allow_repair=True,
)

test_cases = [
    '{"answer": "París es la capital de Francia.", "confidence": 0.95}',
    '{"answer": "Madrid", "confidence": 0.8',
    "La capital es París.",
    '{"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()

# Output esperado:
# Raw:    {"answer": "París es la capital de Francia.", "confidence": 0.
# Action: valid
# Data:   {'answer': 'París es la capital de Francia.', 'confidence': 0.95}
#
# Raw:    {"answer": "Madrid", "confidence": 0.8
# Action: repaired
# Data:   {'answer': 'Madrid', 'confidence': 0.8}
# Issues: ['JSON repaired from partial output']
#
# Raw:    La capital es París.
# Action: fallback
# Data:   {'answer': 'No pude procesar tu pregunta.', 'confidence': 0.0}
# Issues: ['No JSON found, using fallback']
#
# Raw:    {"answer": "", "confidence": 2.0}
# Action: fallback
# Data:   {'answer': 'No pude procesar tu pregunta.', 'confidence': 0.0}
# Issues: ['Validation errors: 2', ...]

Troubleshooting

Problema 1: "Structured outputs de OpenAI agregan latencia"

Los structured outputs requieren que OpenAI procese el schema y constrain la generación. Esto puede agregar 100-500ms.

Solución: Usa structured outputs para calls donde la estructura es crítica (datos para DB, API responses). Para chat conversacional, usa validación Pydantic post-generación que no afecta la latencia del LLM.

Problema 2: "El modelo ignora el schema pedido en el prompt"

A veces el modelo genera texto libre aunque le pidas JSON. Esto pasa más con modelos pequeños y temperatures altas.

Solución: Combina: (1) instrucciones explícitas en el system prompt, (2) temperature baja (0.0-0.3), (3) ejemplo en el prompt, (4) retry con prompt más estricto. Si usas OpenAI, structured outputs resuelven esto directamente.

Problema 3: "Los retries cuestan tokens y dinero"

Cada retry es una llamada adicional al LLM que cuesta tokens.

Solución: Limita retries a 2-3. Usa temperature decreciente. Mantén un fallback response que no requiera LLM. Monitorea tu retry rate — si es > 10%, tu prompt necesita mejoras, no más retries.

Problema 4: "Los custom validators de Pydantic son lentos"

Los field validators con regex se ejecutan por cada campo. En outputs con muchos campos, esto se acumula.

Solución: Precompila regex patterns como constantes de clase. Usa model_validator(mode="after") para validaciones que necesitan múltiples campos — se ejecuta una vez, no por campo.

Ejercicios

Ejercicio 1: Schema para multi-step reasoning

Crea un Pydantic schema que valide outputs de un LLM que hace razonamiento paso a paso (chain-of-thought), donde cada paso debe referenciar al anterior.

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": "¿Cuánto es 2+2*3?",
    "steps": [
        {"step_number": 1, "description": "Identificar la operación de multiplicación: 2*3=6"},
        {"step_number": 2, "description": "Sumar al resultado: 2+6=8", "references_step": 1},
    ],
    "conclusion": "El resultado es 8",
    "confidence": 0.99,
}

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

# Output esperado:
# Steps: 2, Conclusion: El resultado es 8

Ejercicio 2: Validator que detecta hallucinations simples

Crea un validator que detecte cuando el modelo afirma hechos con alta confianza pero usa lenguaje hedging.

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="París es la capital de Francia.", 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}")

# Output esperado:
# Claim 1: confidence=0.99, factual=True
# Claim 2: confidence=0.35, factual=False

Ejercicio 3: Output validator con fallback chain

Implementa un sistema de fallback en cadena: intenta 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:
    """Intenta múltiples estrategias en orden."""
    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": "No pude procesar la respuesta.", "score": 0.0},
        "strategy": "fallback",
        "tried": 4,
    }


tests = [
    '{"text": "París", "score": 0.9}',
    '{"text": "Madrid", "score": 0.8',
    "La respuesta es Tokio.",
    "",
]

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

# Output esperado:
# Input: '{"text": "París", "score": 0.9}' → Strategy: direct_parse (tried: 1)
# Input: '{"text": "Madrid", "score": 0.8' → Strategy: json_repair (tried: 2)
# Input: 'La respuesta es Tokio.' → Strategy: text_extract (tried: 3)
# Input: '' → Strategy: fallback (tried: 4)

Ejercicio 4: Schema versionado para evolución de outputs

Implementa un sistema que soporte múltiples versiones de un schema y migre datos entre versiones.

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()}")

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

Resumen

• 🔑 Los outputs del LLM no tienen contrato — el modelo puede producir cualquier formato, tipo o estructura independientemente de lo que le pidas
• 🔑 Pydantic schemas son la herramienta principal para validar outputs: definen la estructura esperada y rechazan outputs que no cumplen
• 🔑 Los structured outputs de OpenAI fuerzan el schema a nivel del modelo, eliminando la mayoría de errores de formato (pero solo funcionan con OpenAI)
• 🔑 Las retry strategies con temperatura decreciente son efectivas para obtener outputs válidos sin fallback: primer intento a 0.7, segundo a 0.5, tercero a 0.3
• 🔑 La reparación de JSON parcial recupera outputs truncados por max_tokens, salvando llamadas al API
• 🔑 Los custom validators para AI outputs detectan problemas específicos: system prompt leaks, code injection, hedging inconsistente con confidence
• 🔑 La coerción de tipos maneja la imprecisión natural del LLM: "yes" → True, "1,234" → 1234, "85%" → 0.85
• 🔑 La fallback chain garantiza que siempre produces una respuesta válida: direct parse → repair → text extract → fallback
• 🔑 El OutputValidator es la segunda pieza del Sanitization Pipeline y se conecta con el Content Filter de la cápsula 04

Recursos adicionales

1. Pydantic V2 Documentation — Referencia completa de Pydantic para validación de datos, validators, y schemas
2. OpenAI Structured Outputs — Guía oficial para obtener outputs estructurados directamente del modelo
3. Pydantic — Field Validators — Custom validators para lógica de validación específica de tu dominio
4. OpenAI Function Calling — Alternativa a structured outputs para extraer datos estructurados
5. JSON Schema Specification — Estándar en el que Pydantic basa sus schemas, útil para interoperabilidad
6. OWASP LLM05: Improper Output Handling — La vulnerabilidad que la validación de outputs mitiga directamente
7. Instructor Library — Librería que simplifica structured outputs con Pydantic para múltiples proveedores LLM
8. Pydantic Settings Management — Gestión de configuración con Pydantic, útil para configurar validators

Creado: Marzo 2026 Versión: 1.0