Phase 11 - Lesson 03

Outputs Estructurados: JSON, Validación de Schema, Decodificación Restringida

This lesson includes a graded coding exercise that runs in your browser, unlocked with lifetime access.

Tu LLM devuelve un string. Tu aplicación necesita JSON. Esa brecha ha colapsado más sistemas en producción que cualquier alucinación de modelo. El output estructurado es el puente entre el lenguaje natural y los datos tipados. Hazlo bien y tu LLM se convierte en una API confiable. Hazlo mal y estarás parseando texto libre con regex a las 3 am.

Tipo: Build Lenguajes: Python Requisitos previos: Fase 10, Lecciones 01-05 (LLMs desde cero) Tiempo: ~90 minutos Relacionado: Fase 5 · 20 (Outputs Estructurados & Decodificación Restringida) cubre la teoría a nivel de decodificador (procesadores de logits FSM/CFG, Outlines, XGrammar). Esta lección se enfoca en la superficie del SDK de producción (OpenAI response_format, uso de herramientas de Anthropic, Instructor) — lee la Fase 5 · 20 primero si deseas entender qué está sucediendo debajo de la API.

Objetivos de Aprendizaje

• Implementar modo JSON y outputs restringidos por schema utilizando parámetros de API de OpenAI y Anthropic
• Construir una capa de validación Pydantic que rechace outputs malformados de LLM y reintente con feedback de error
• Explicar cómo la decodificación restringida fuerza un JSON válido a nivel de token sin posprocesamiento
• Diseñar prompts de extracción robustos que conviertan de manera confiable texto no estructurado en estructuras de datos tipadas

El Problema

Le preguntas a un LLM: "Extrae el nombre del producto, precio y disponibilidad de este texto." Responde:

The product is the Sony WH-1000XM5 headphones, which cost $348.00 and are currently in stock.

Esa es una respuesta perfectamente correcta. También es completamente inútil para tu aplicación. Tu sistema de inventario necesita {"product": "Sony WH-1000XM5", "price": 348.00, "in_stock": true}. Necesitas un objeto JSON con claves específicas, tipos específicos y restricciones de valor específicas. No necesitas una oración.

La solución ingenua: agregar "Responde en JSON" a tu prompt. Esto funciona el 90% de las veces. El otro 10% el modelo envuelve el JSON en bloques de código markdown, o agrega un preámbulo como "Aquí está el JSON:", o produce un JSON sintácticamente inválido porque cerró una llave antes de tiempo. Tu parser de JSON falla. Tu pipeline se rompe. Agregas try/except y un ciclo de reintento. El reintento a veces produce datos diferentes. Ahora tienes un problema de consistencia además de un problema de parsing.

Esto no es un problema de ingeniería de prompts. Es un problema de decodificación. El modelo genera tokens de izquierda a derecha. En cada posición, elige el siguiente token más probable de un vocabulario de más de 100K opciones. La mayoría de esas opciones producirían un JSON inválido en cualquier posición dada. Si el modelo simplemente emitiera {"price":, el siguiente token debe ser un dígito, una comilla (para string), null, true, false o un signo negativo. Cualquier otra cosa produce un JSON inválido. Sin restricciones, el modelo podría elegir una palabra en inglés perfectamente razonable pero catastróficamente incorrecta desde el punto de vista sintático.

El Concepto

El Espectro de Outputs Estructurados

Existen cuatro niveles de control de output estructurado, cada uno más confiable que el anterior.

graph LR
    subgraph Spectrum["Structured Output Spectrum"]
        direction LR
        A["Prompt-based\n'Return JSON'\n~90% valid"] --> B["JSON Mode\nGuaranteed valid JSON\nNo schema guarantee"]
        B --> C["Schema Mode\nJSON + matches schema\nGuaranteed compliance"]
        C --> D["Constrained Decoding\nToken-level enforcement\n100% compliance"]
    end

    style A fill:#1a1a2e,stroke:#ff6b6b,color:#fff
    style B fill:#1a1a2e,stroke:#ffa500,color:#fff
    style C fill:#1a1a2e,stroke:#51cf66,color:#fff
    style D fill:#1a1a2e,stroke:#0f3460,color:#fff

Basado en prompt ("Responda en JSON válido"): sin cumplimiento forzado. El modelo suele cumplir pero a veces no. Confiabilidad: ~90%. Modos de falla: bloques de markdown, texto de preámbulo, output truncado, estructura incorrecta.

Modo JSON: la API garantiza que el output sea un JSON válido. El parámetro response_format: { type: "json_object" } de OpenAI habilita esto. El output se parseará sin errores. Pero es posible que no coincida con el schema esperado -- claves adicionales, tipos incorrectos, campos faltantes.

Modo Schema: la API recibe un JSON Schema y garantiza que el output coincida con él. En 2026, todos los proveedores principales lo soportan de forma nativa: response_format: { type: "json_schema", json_schema: {...} } de OpenAI (también como tool_choice="required"), el uso de herramientas de Anthropic con input_schema, y response_schema + response_mime_type: "application/json" de Gemini. El output tiene exactamente las claves, tipos y restricciones que especificaste.

Decodificación restringida: en cada posición de token durante la generación, el decodificador enmascara todos los tokens que producirían un output inválido. Si el schema requiere un número y el modelo está a punto de emitir una letra, ese token se establece con probabilidad cero. El modelo solo puede producir tokens que lleven a un output válido. Esto es lo que el modo de output estructurado de OpenAI y librerías como Outlines y Guidance implementan internamente.

JSON Schema: El Lenguaje de Contrato

JSON Schema es la forma en que le dices al modelo (o a la capa de validación) qué forma debe tener el output. Cada sistema principal de output estructurado lo utiliza.

{
  "type": "object",
  "properties": {
    "product": { "type": "string" },
    "price": { "type": "number", "minimum": 0 },
    "in_stock": { "type": "boolean" },
    "categories": {
      "type": "array",
      "items": { "type": "string" }
    }
  },
  "required": ["product", "price", "in_stock"]
}

Este schema dice: el output debe ser un objeto con un string product, un número no negativo price, un booleano in_stock y un array opcional de strings categories. Cualquier output que no coincida es rechazado.

Los schemas manejan los casos difíciles: objetos aninados, arrays con ítems tipados, enums (restringir un string a valores específicos), coincidencia de patrones (regex en strings) y combinadores (oneOf, anyOf, allOf para outputs polimórficos).

El Patrón Pydantic

En Python, no escribes JSON Schema a mano. Defines un modelo Pydantic y este genera el schema por ti.

from pydantic import BaseModel

class Product(BaseModel):
    product: str
    price: float
    in_stock: bool
    categories: list[str] = []

Esto produce el mismo JSON Schema que el anterior. La librería Instructor (y el SDK de OpenAI) aceptan modelos Pydantic directamente: pasas la clase del modelo y obtienes una instancia validada. Si el output del LLM no coincide, Instructor lo reintenta automáticamente.

Llamada de Funciones / Uso de Herramientas

Una interfaz alternativa para el mismo problema. En lugar de pedirle al modelo que produzca JSON directamente, defines "herramientas" (funciones) con parámetros tipados. El modelo devuelve una llamada a la función con argumentos estructurados. OpenAI llama a esto "llamada de funciones" (function calling). Anthropic lo llama "uso de herramientas" (tool use). El resultado es el mismo: datos estructurados.

graph TD
    subgraph ToolUse["Tool Use Flow"]
        U["User: Extract product info\nfrom this review text"] --> M["Model processes input"]
        M --> TC["Tool Call:\nextract_product(\n  product='Sony WH-1000XM5',\n  price=348.00,\n  in_stock=true\n)"]
        TC --> V["Validate against\nfunction schema"]
        V --> R["Structured Result:\n{product, price, in_stock}"]
    end

    style U fill:#1a1a2e,stroke:#0f3460,color:#fff
    style TC fill:#1a1a2e,stroke:#e94560,color:#fff
    style V fill:#1a1a2e,stroke:#ffa500,color:#fff
    style R fill:#1a1a2e,stroke:#51cf66,color:#fff

El uso de herramientas es preferible cuando el modelo debe elegir qué función llamar, no solo completar parámetros. Si tienes 10 schemas de extracción diferentes y el modelo debe elegir el correcto basándose en la entrada, el uso de herramientas te ofrece tanto la selección del schema como el output estructurado.

Modos de Falla Comunes

Incluso con la aplicación forzada de schemas, los outputs estructurados pueden fallar de maneras sutiles.

Valores alucinados: el output coincide con el schema pero contiene datos inventados. El modelo produce {"price": 299.99} cuando el texto dice $348. La validación del schema no puede detectar esto -- el tipo es correcto, el valor es incorrecto.

Confusión de Enum: restringes un campo a ["in_stock", "out_of_stock", "preorder"]. El modelo devuelve "available" -- semánticamente correcto, pero no está en el conjunto permitido. Una buena decodificación restringida previene esto. Los enfoques basados en prompts no.

Profundidad de objetos aninados: los schemas profundamente aninados (4+ niveles) producen más errores. Cada nivel de anidamiento es otro lugar donde el modelo puede perder el control de la estructura.

Longitud de arrays: el modelo puede producir demasiados o muy pocos elementos en un array. Los schemas soportan minItems and maxItems pero no todos los proveedores los imponen a nivel de decodificación.

Omisión de campos opcionales: el modelo omite campos que son técnicamente opcionales pero semánticamente importantes para tu caso de uso. Establécelos como requeridos en el schema incluso si los datos a veces faltan -- fuerza al modelo a producir null explícitamente.

Constrúyelo

Paso 1: Validador de JSON Schema

Construye un validador desde cero que verifique si un objeto de Python coincide con un JSON Schema. Esto es lo que se ejecuta en el lado del output para verificar el cumplimiento.

import json

def validate_schema(data, schema):
    errors = []
    _validate(data, schema, "", errors)
    return errors

def _validate(data, schema, path, errors):
    schema_type = schema.get("type")

    if schema_type == "object":
        if not isinstance(data, dict):
            errors.append(f"{path}: expected object, got {type(data).__name__}")
            return
        for key in schema.get("required", []):
            if key not in data:
                errors.append(f"{path}.{key}: required field missing")
        properties = schema.get("properties", {})
        for key, value in data.items():
            if key in properties:
                _validate(value, properties[key], f"{path}.{key}", errors)

    elif schema_type == "array":
        if not isinstance(data, list):
            errors.append(f"{path}: expected array, got {type(data).__name__}")
            return
        min_items = schema.get("minItems", 0)
        max_items = schema.get("maxItems", float("inf"))
        if len(data) < min_items:
            errors.append(f"{path}: array has {len(data)} items, minimum is {min_items}")
        if len(data) > max_items:
            errors.append(f"{path}: array has {len(data)} items, maximum is {max_items}")
        items_schema = schema.get("items", {})
        for i, item in enumerate(data):
            _validate(item, items_schema, f"{path}[{i}]", errors)

    elif schema_type == "string":
        if not isinstance(data, str):
            errors.append(f"{path}: expected string, got {type(data).__name__}")
            return
        enum_values = schema.get("enum")
        if enum_values and data not in enum_values:
            errors.append(f"{path}: '{data}' not in allowed values {enum_values}")

    elif schema_type == "number":
        if not isinstance(data, (int, float)):
            errors.append(f"{path}: expected number, got {type(data).__name__}")
            return
        minimum = schema.get("minimum")
        maximum = schema.get("maximum")
        if minimum is not None and data < minimum:
            errors.append(f"{path}: {data} is less than minimum {minimum}")
        if maximum is not None and data > maximum:
            errors.append(f"{path}: {data} is greater than maximum {maximum}")

    elif schema_type == "boolean":
        if not isinstance(data, bool):
            errors.append(f"{path}: expected boolean, got {type(data).__name__}")

    elif schema_type == "integer":
        if not isinstance(data, int) or isinstance(data, bool):
            errors.append(f"{path}: expected integer, got {type(data).__name__}")

Paso 2: Modelo Estilo Pydantic a Schema

Construye un convertidor minimalista de clase a schema. Define una clase de Python y genera su JSON Schema automáticamente.

class SchemaField:
    def __init__(self, field_type, required=True, default=None, enum=None, minimum=None, maximum=None):
        self.field_type = field_type
        self.required = required
        self.default = default
        self.enum = enum
        self.minimum = minimum
        self.maximum = maximum

def python_type_to_schema(field):
    type_map = {
        str: "string",
        int: "integer",
        float: "number",
        bool: "boolean",
    }

    schema = {}

    if field.field_type in type_map:
        schema["type"] = type_map[field.field_type]
    elif field.field_type == list:
        schema["type"] = "array"
        schema["items"] = {"type": "string"}
    elif isinstance(field.field_type, dict):
        schema = field.field_type

    if field.enum:
        schema["enum"] = field.enum
    if field.minimum is not None:
        schema["minimum"] = field.minimum
    if field.maximum is not None:
        schema["maximum"] = field.maximum

    return schema

def model_to_schema(name, fields):
    properties = {}
    required = []

    for field_name, field in fields.items():
        properties[field_name] = python_type_to_schema(field)
        if field.required:
            required.append(field_name)

    return {
        "type": "object",
        "properties": properties,
        "required": required,
    }

Paso 3: Filtro de Token Restringido

Simula una decodificación restringida. Dada una cadena JSON parcial y un schema, determina qué categorías de tokens son válidas en la posición actual.

def next_valid_tokens(partial_json, schema):
    stripped = partial_json.strip()

    if not stripped:
        return ["{"]

    try:
        json.loads(stripped)
        return ["<EOS>"]
    except json.JSONDecodeError:
        pass

    last_char = stripped[-1] if stripped else ""

    if last_char == "{":
        return ['"', "}"]
    elif last_char == '"':
        if stripped.endswith('":'):
            return ['"', "0-9", "true", "false", "null", "[", "{"]
        return ["a-z", '"']
    elif last_char == ":":
        return [" ", '"', "0-9", "true", "false", "null", "[", "{"]
    elif last_char == ",":
        return [" ", '"', "{", "["]
    elif last_char in "0123456789":
        return ["0-9", ".", ",", "}", "]"]
    elif last_char == "}":
        return [",", "}", "]", "<EOS>"]
    elif last_char == "]":
        return [",", "}", "<EOS>"]
    elif last_char == "[":
        return ['"', "0-9", "true", "false", "null", "{", "[", "]"]
    else:
        return ["any"]

def demonstrate_constrained_decoding():
    partial_states = [
        '',
        '{',
        '{"product"',
        '{"product":',
        '{"product": "Sony"',
        '{"product": "Sony",',
        '{"product": "Sony", "price":',
        '{"product": "Sony", "price": 348',
        '{"product": "Sony", "price": 348}',
    ]

    print(f"{'Partial JSON':<45} {'Valid Next Tokens'}")
    print("-" * 80)
    for state in partial_states:
        valid = next_valid_tokens(state, {})
        display = state if state else "(empty)"
        print(f"{display:<45} {valid}")

Paso 4: Pipeline de Extracción

Combina todo en una pipeline de extracción: define un schema, simula un LLM produciendo output estructurado, valida el output y maneja los reintentos.

def simulate_llm_extraction(text, schema, attempt=0):
    if "headphones" in text.lower() or "sony" in text.lower():
        if attempt == 0:
            return '{"product": "Sony WH-1000XM5", "price": 348.00, "in_stock": true, "categories": ["audio", "headphones"]}'
        return '{"product": "Sony WH-1000XM5", "price": 348.00, "in_stock": true}'

    if "laptop" in text.lower():
        return '{"product": "MacBook Pro 16", "price": 2499.00, "in_stock": false, "categories": ["computers"]}'

    return '{"product": "Unknown", "price": 0, "in_stock": false}'

def extract_with_retry(text, schema, max_retries=3):
    for attempt in range(max_retries):
        raw = simulate_llm_extraction(text, schema, attempt)

        try:
            data = json.loads(raw)
        except json.JSONDecodeError as e:
            print(f"  Attempt {attempt + 1}: JSON parse error -- {e}")
            continue

        errors = validate_schema(data, schema)
        if not errors:
            return data

        print(f"  Attempt {attempt + 1}: Schema validation errors -- {errors}")

    return None

product_schema = {
    "type": "object",
    "properties": {
        "product": {"type": "string"},
        "price": {"type": "number", "minimum": 0},
        "in_stock": {"type": "boolean"},
        "categories": {"type": "array", "items": {"type": "string"}},
    },
    "required": ["product", "price", "in_stock"],
}

Paso 5: Ejecutar la Pipeline Completa

def run_demo():
    print("=" * 60)
    print("  Structured Output Pipeline Demo")
    print("=" * 60)

    print("\n--- Schema Definition ---")
    product_fields = {
        "product": SchemaField(str),
        "price": SchemaField(float, minimum=0),
        "in_stock": SchemaField(bool),
        "categories": SchemaField(list, required=False),
    }
    generated_schema = model_to_schema("Product", product_fields)
    print(json.dumps(generated_schema, indent=2))

    print("\n--- Schema Validation ---")
    test_cases = [
        ({"product": "Test", "price": 10.0, "in_stock": True}, "Valid object"),
        ({"product": "Test", "price": -5.0, "in_stock": True}, "Negative price"),
        ({"product": "Test", "in_stock": True}, "Missing price"),
        ({"product": "Test", "price": "ten", "in_stock": True}, "String as price"),
        ("not an object", "String instead of object"),
    ]

    for data, label in test_cases:
        errors = validate_schema(data, product_schema)
        status = "PASS" if not errors else f"FAIL: {errors}"
        print(f"  {label}: {status}")

    print("\n--- Constrained Decoding Simulation ---")
    demonstrate_constrained_decoding()

    print("\n--- Extraction Pipeline ---")
    texts = [
        "The Sony WH-1000XM5 headphones are priced at $348 and currently available.",
        "The new MacBook Pro 16-inch laptop costs