JSON estructurado desde un LLM: validación con Pydantic

«Devuelve solo JSON» es una instrucción de formato, no un contrato de datos. Un LLM todavía puede añadir Markdown, omitir un campo, usar un tipo incorrecto o inventar un valor de enum. La aplicación debe tratar la respuesta como texto no confiable hasta validarla localmente.

LLM -> raw response -> Pydantic validation -> typed object
                       | validation error
                       v
                 one repair attempt -> success or explicit failure

Define un único contrato

from typing import Literal

from pydantic import BaseModel, ConfigDict, Field


class SupportTicket(BaseModel):
    model_config = ConfigDict(extra="forbid")

    title: str = Field(min_length=1, max_length=120)
    priority: Literal["low", "medium", "high"]
    affected_services: list[str] = Field(min_length=1, max_length=5)
    needs_human_review: bool

Genera el esquema para el prompt con SupportTicket.model_json_schema(). Así el prompt y el validador nacen del mismo modelo. extra="forbid" hace que un campo inesperado detenga el proceso, y la validación local sigue siendo la autoridad aunque el esquema aparezca en el prompt.

Conserva la respuesta sin procesar

import os

from openai import OpenAI


client = OpenAI(
    api_key=os.environ["BETTERTOKEN_API_KEY"],
    base_url="https://www.bettertoken.ai/v1",
)


def ask_model(user_text: str, schema_text: str) -> str:
    response = client.chat.completions.create(
        model=os.environ["BETTERTOKEN_MODEL_ID"],
        messages=[
            {
                "role": "system",
                "content": (
                    "Return one JSON object without Markdown. "
                    "It must match this JSON Schema:\n" + schema_text
                ),
            },
            {"role": "user", "content": user_text},
        ],
    )
    return response.choices[0].message.content or ""

Guarda la cadena original hasta terminar la validación. Si necesitas conservarla para diagnóstico, utiliza almacenamiento protegido y retención limitada. No registres prompts completos, datos personales, claves ni respuestas sensibles. Un HTTP 200 confirma una respuesta del servicio, no el cumplimiento del esquema.

Valida la cadena JSON directamente

from pydantic import ValidationError


def validate_ticket(raw: str) -> SupportTicket:
    return SupportTicket.model_validate_json(raw, strict=True)

Con strict=True, una cadena como "true" no se convierte silenciosamente en booleano. ValidationError permite distinguir sintaxis rota, campos ausentes, tipos erróneos, valores enum no permitidos y campos adicionales. Para reparar, extrae solo ruta, tipo y mensaje con exc.errors(include_url=False, include_input=False).

Limita la reparación a un intento

La reparación es otra solicitud. Envía la respuesta original, el resumen de errores y el mismo esquema, indicando que no se añadan hechos. Valida otra vez con la misma función. El bucle debe tener dos pasos: respuesta inicial y una reparación. Si ambas fallan, devuelve value=None, conserva las dos respuestas y entrega la lista final de errores a revisión manual o a una cola de fallos.

Trata los errores de red por separado: un timeout no demuestra que el JSON sea incorrecto. Tampoco ejecutes herramientas, pagos u operaciones externas antes de obtener un objeto válido.

Comprueba el contrato sin llamar a la API

Usa fixtures fijos: uno válido y otro con enum inválido, lista vacía, string en lugar de booleano y campo extra. Verifica que el primero se acepta, que el segundo produce los tipos de error esperados y que dos respuestas inválidas terminan en un fallo explícito.

Para este artículo, estas pruebas locales pasaron con Python y Pydantic 2.12.5. No se hizo una llamada real al modelo, por lo que no se afirma que un Model ID concreto produzca siempre JSON válido.

Antes de producción, versiona el modelo con su consumidor, limita el tamaño de respuesta y los intentos, separa métricas de transporte/API/esquema, protege los logs y añade fixtures cada vez que cambie el contrato.

Consulta el formato actual en BetterToken Chat Completions. La API genera texto; Pydantic es la frontera local que decide si puede convertirse en datos.

Referencia ejecutable completa

Estos bloques conservan la referencia ejecutable exacta para generar el esquema, resumir errores, limitar la reparación y probar fixtures fijos.

import json

schema = SupportTicket.model_json_schema()
schema_text = json.dumps(schema, ensure_ascii=False)

def error_summary(exc: ValidationError) -> list[dict[str, object]]:
    return [
        {
            "path": ".".join(str(part) for part in item["loc"]),
            "type": item["type"],
            "message": item["msg"],
        }
        for item in exc.errors(include_url=False, include_input=False)
    ]

import json
from dataclasses import dataclass


@dataclass
class ParseResult:
    value: SupportTicket | None
    raw_responses: list[str]
    errors: list[dict[str, object]]


def repair_model(
    raw: str,
    errors: list[dict[str, object]],
    schema_text: str,
) -> str:
    response = client.chat.completions.create(
        model=os.environ["BETTERTOKEN_MODEL_ID"],
        messages=[
            {
                "role": "system",
                "content": (
                    "Repair the JSON. Return one JSON object without Markdown. "
                    "Do not add facts. The object must match this schema:\n"
                    + schema_text
                ),
            },
            {
                "role": "user",
                "content": json.dumps(
                    {"raw": raw, "validation_errors": errors},
                    ensure_ascii=False,
                ),
            },
        ],
    )
    return response.choices[0].message.content or ""


def parse_with_one_repair(user_text: str) -> ParseResult:
    raw_responses: list[str] = []
    raw = ask_model(user_text, schema_text)

    for attempt in range(2):
        raw_responses.append(raw)
        try:
            value = validate_ticket(raw)
            return ParseResult(value=value, raw_responses=raw_responses, errors=[])
        except ValidationError as exc:
            errors = error_summary(exc)
            if attempt == 1:
                return ParseResult(
                    value=None,
                    raw_responses=raw_responses,
                    errors=errors,
                )
            raw = repair_model(raw, errors, schema_text)

    raise AssertionError("unreachable")

valid_raw = """{
  "title": "Login fails after token refresh",
  "priority": "high",
  "affected_services": ["auth-api"],
  "needs_human_review": true
}"""

invalid_raw = """{
  "title": "Login fails",
  "priority": "urgent",
  "affected_services": [],
  "needs_human_review": "yes",
  "confidence": 0.98
}"""

ticket = validate_ticket(valid_raw)
assert ticket.priority == "high"

try:
    validate_ticket(invalid_raw)
except ValidationError as exc:
    assert {item["type"] for item in exc.errors()} >= {
        "literal_error",
        "too_short",
        "bool_type",
        "extra_forbidden",
    }
else:
    raise AssertionError("invalid fixture was accepted")

Fuentes

• Pydantic Models
• Pydantic JSON
• BetterToken Chat Completions