Salidas estructuradas de LLMs que no se rompen en producción

La demo funciona cada vez: pídele al modelo que devuelva JSON según un schema, aplica json.loads() a la respuesta, y sigue adelante. Luego se hace el deploy, y tres semanas después alguien encuentra un stack trace donde el modelo envolvió su JSON en un code fence de Markdown, o agregó un amigable “Aquí están los datos extraídos:” antes del objeto real, o — peor — devolvió JSON sintácticamente válido al que le falta un campo que el código downstream asumió que siempre estaría ahí. Nada de esto es el modelo siendo poco confiable de alguna manera misteriosa. Es el resultado predecible de tratar “pedir amablemente JSON” como equivalente a “obtener un objeto JSON validado”, cuando son dos niveles diferentes de garantía que resultan verse iguales en una demo de happy-path.

El arreglo no es un prompt más inteligente. Es un pipeline que asume que el modelo ocasionalmente hará exactamente lo que no pediste, y tiene un camino definido y probado para lo que sucede cuando lo hace — un schema que se hace cumplir en lugar de solo describirse, un loop de retry que le devuelve al modelo su propio error de validación, y un fallback para el caso en que nada de eso funcione y de todas formas obtengas prosa. Este artículo construye ese pipeline en Python.

Aprenderás:

• Por qué hacer prompting para JSON y validar JSON son problemas diferentes con diferentes modos de fallo
• Cómo definir un contrato de salida de LLM con Pydantic y obtener una instancia validada directamente
• La diferencia real entre constrained decoding y prompting “por favor responde en JSON”
• Cómo construir un loop de retry que alimenta errores de validación de vuelta al modelo en lugar de solo volver a intentar a ciegas
• Qué hacer cuando el modelo devuelve prosa a pesar de todos los guardarraíles
• Cómo manejar JSON parcial y en streaming sin que tu parser se ahogue en un objeto incompleto
• Un checklist de producción para un pipeline de salida estructurada en el que realmente puedas confiar

Tabla de contenidos

1. Dos problemas diferentes con un mismo nombre
2. Schemas Pydantic como el contrato
3. Constrained Decoding vs JSON por Prompt
4. Retry en fallo de validación
5. Cuando el modelo devuelve prosa de todas formas
6. Manejar JSON parcial y en streaming
7. Schemas anidados y Union
8. Salidas estructuradas dentro de un loop de agente
9. Checklist de producción
10. Conclusión

Dos problemas diferentes con un mismo nombre

“Obtener salida estructurada de un LLM” es usualmente una abreviación de dos garantías separadas que no se juntan automáticamente. La primera es la forma: ¿Es la respuesta JSON sintácticamente válido, versus prosa, un code fence, o JSON con una oración adjuntada? La segunda es el schema: dado JSON válido, ¿realmente tiene los campos que necesitas, con los tipos correctos, satisfaciendo las restricciones que asume tu código downstream?

Un prompt que solo dice “responde en JSON” no aborda ninguna garantía de manera confiable — es una petición, no una restricción, y el modelo es libre de estar a algunos tokens de distancia de lo que pediste. Las técnicas en este artículo abordan cada garantía en una capa diferente: constrained decoding (o un modo de API consciente del schema) aborda la forma estructuralmente, de modo que el formato literalmente no puede salir mal; la validación Pydantic aborda el schema, detectando una salida que es JSON válido pero el JSON incorrecto — un string donde necesitabas un int, un campo requerido faltante, un valor de enum que no existe. Tratarlos como un solo problema es cómo los equipos terminan con un único try/except json.JSONDecodeError que silenciosamente deja pasar un objeto perfectamente parseable al que le falta la mitad de los campos de los que depende el resto del código.

Schemas Pydantic como el contrato

Define la forma de salida una vez como un modelo Pydantic, y deja que sirva tres propósitos a la vez — el schema que la documentación de otro modo describiría, el validador que se ejecuta en cada respuesta, y el tipo con el que tu código downstream realmente trabaja.

from pydantic import BaseModel, Field
from typing import Literal

class TicketClassification(BaseModel):
    category: Literal["billing", "technical", "account", "other"]
    urgency: Literal["low", "medium", "high"]
    summary: str = Field(description="One-sentence summary of the customer's issue")
    requires_escalation: bool

Con el SDK de Anthropic, client.messages.parse() toma este modelo directamente y devuelve una instancia validada en lugar de un bloque de contenido que tienes que parsear tú mismo:

import anthropic

client = anthropic.Anthropic()

response = client.messages.parse(
    model="claude-opus-5",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": f"Classify this support ticket:\n\n{ticket_text}",
    }],
    output_format=TicketClassification,
)

ticket = response.parsed_output
print(ticket.category, ticket.urgency)  # typed, validated attribute access — not dict lookups

response.parsed_output es una instancia real de TicketClassification, no un dict del que esperas que tenga las claves correctas — un typo como ticket.catgory falla en tiempo de desarrollo con un AttributeError normal de Python, en lugar de fallar silenciosamente en runtime con un KeyError tres capas más profundo en tu call stack. Es una cosa pequeña que suma: el schema deja de ser documentación que esperas que se haya mantenido actualizada y se convierte en algo que tu type checker y tu IDE realmente hacen cumplir.

Para casos donde necesitas el JSON schema crudo en lugar de un parse dirigido por Pydantic — hablar con un sistema que te entrega un schema dinámicamente, por ejemplo — la misma garantía está disponible a nivel de request mediante output_config:

response = client.messages.create(
    model="claude-opus-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": f"Classify this ticket:\n\n{ticket_text}"}],
    output_config={
        "format": {
            "type": "json_schema",
            "schema": {
                "type": "object",
                "properties": {
                    "category": {"type": "string", "enum": ["billing", "technical", "account", "other"]},
                    "urgency": {"type": "string", "enum": ["low", "medium", "high"]},
                    "summary": {"type": "string"},
                    "requires_escalation": {"type": "boolean"},
                },
                "required": ["category", "urgency", "summary", "requires_escalation"],
                "additionalProperties": False,
            },
        }
    },
)

output_config.format garantiza que el contenido de la respuesta es texto que contiene JSON válido que coincide con ese schema — la garantía de forma — pero nota que en ese punto es el dict crudo; igual tendrías que llamar a TicketClassification.model_validate_json(...) tú mismo para obtener la garantía de schema (coerción de tipos, verificación de enum) que .parse() da automáticamente. En la práctica, prefiere .parse() con un modelo Pydantic cada vez que controles el schema en código — es estrictamente menos qué mantener que definir las mismas restricciones dos veces, una vez como modelo Pydantic y una vez como dict de JSON schema crudo.

Constrained Decoding vs JSON por Prompt

Vale la pena ser preciso sobre lo que cambió aquí respecto a los enfoques anteriores, porque “pedir al modelo que devuelva JSON en el system prompt” sigue siendo un consejo común y es significativamente más débil que un modo de output consciente del schema. El JSON por prompt es una petición que el modelo puede satisfacer imperfectamente — puede agregar un preámbulo, olvidar una llave de cierre bajo condiciones de sampling inusuales, o usar una clave que suena plausible que no especificaste en lugar de la que especificaste. Usualmente funciona, justo hasta que la entrada es suficientemente inusual como para que no lo haga.

output_config.format (y el helper .parse() construido sobre él) restringe la respuesta en tiempo de generación de modo que la garantía de forma es estructural en lugar de solicitada — el modelo está impedido de emitir contenido fuera del schema, no solo amablemente pedido que se quede dentro. Esta es la razón real para usar un modo de API consciente del schema sobre un prompt bien elaborado: convierte “el modelo usualmente lo hace bien” en “el modelo no puede hacer la forma incorrecta”, que es una garantía de confiabilidad categóricamente diferente, aunque todavía deja la validación a nivel de schema (la segunda garantía de la sección anterior) como algo que tu propio código tiene que verificar.

Nota una interacción que vale la pena conocer: output_config.format no es compatible con citas de documentos (citations: {enabled: true}) — si tu pipeline necesita tanto salida estructurada como rastreo de citas contra un documento fuente, esa es una restricción de diseño real que considerar, no un edge case que descubrir en producción.

Retry en fallo de validación

Incluso con una respuesta schema-restringida, la validación Pydantic aún puede fallar — un valor adyacente a un enum que el schema permitió como string genérico pero tu tipo Literal no acepta, un campo numérico que el modelo populó con un valor fuera de un rango que haces cumplir en un validador personalizado. La única cosa más efectiva que puedes hacer aquí no es “intentar de nuevo con el mismo prompt” sino “intentar de nuevo con el propio mensaje de error del modelo incluido”, porque un error de validación es feedback específico y accionable sobre el que el modelo usualmente puede actuar directamente.

from pydantic import ValidationError

def classify_with_retry(ticket_text: str, max_attempts: int = 3) -> TicketClassification:
    messages = [{"role": "user", "content": f"Classify this support ticket:\n\n{ticket_text}"}]

    for attempt in range(max_attempts):
        response = client.messages.create(
            model="claude-opus-5",
            max_tokens=1024,
            messages=messages,
            output_config={"format": {"type": "json_schema", "schema": TicketClassification.model_json_schema()}},
        )
        text = next(b.text for b in response.content if b.type == "text")

        try:
            return TicketClassification.model_validate_json(text)
        except ValidationError as e:
            if attempt == max_attempts - 1:
                raise
            messages.append({"role": "assistant", "content": text})
            messages.append({
                "role": "user",
                "content": f"That response failed validation with this error:\n{e}\n"
                           f"Return corrected JSON matching the schema exactly.",
            })

    raise RuntimeError("unreachable")

Dos detalles hacen que este loop sea realmente efectivo en lugar de solo más lento. Primero, el intento fallido y el error ambos vuelven a la conversación — el modelo ve exactamente qué salió mal, no un genérico “inténtalo de nuevo.” Segundo, max_attempts es un techo real con una excepción real al final, no un retry infinito que convierte un request malo en un multiplicador de costos silencioso; tres intentos capturan la abrumadora mayoría de los fallos de validación en la práctica, y un request que aún falla después de tres te está diciendo algo sobre el input o el schema, no que solo necesitaba un intento más.

Cuando el modelo devuelve prosa de todas formas

Incluso con un modo schema-restringido, planifica para que la respuesta ocasionalmente no sea lo que esperabas — un rechazo, un stop_reason inusual, un edge case de la biblioteca cliente, o un modelo que decidió que la respuesta honesta a “extrae estos datos” es “no hay datos que extraer” y lo dijo en una oración. El parsing defensivo aquí es un seguro barato contra una clase de incidente de producción que de otro modo parece un crash aleatorio e irreproducible.

import json
import re

def extract_json_defensively(text: str) -> dict | None:
    # Try a direct parse first — the common case.
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass

    # Strip a markdown code fence if the model wrapped its output in one.
    fence_match = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
    if fence_match:
        try:
            return json.loads(fence_match.group(1))
        except json.JSONDecodeError:
            pass

    # Last resort: grab the first {...} block and hope it's balanced.
    brace_match = re.search(r"\{.*\}", text, re.DOTALL)
    if brace_match:
        try:
            return json.loads(brace_match.group(0))
        except json.JSONDecodeError:
            pass

    return None

Conéctalo como fallback detrás del path de modo restringido, no como reemplazo — recurre a la extracción regex solo después de que la llamada consciente del schema falle completamente (un stop_reason que no esperabas, un error de API) en lugar de enrutar cada respuesta por defecto a través de cirugía de strings de mejor esfuerzo. Y lo que sea que esto devuelva, igual pásalo por TicketClassification.model_validate(...) antes de confiar en él; recuperar JSON de aspecto válido de un string desordenado no es lo mismo que recuperar JSON que satisface tu schema; esas son las mismas dos garantías del inicio de este artículo, y saltarse la segunda aquí anula todo el punto del ejercicio.

Manejar JSON parcial y en streaming

Hacer streaming de una respuesta estructurada introduce una versión del problema de forma que no obtienes con una sola llamada bloqueante: en cualquier punto a mitad del stream, el texto acumulado es por definición JSON incompleto, y parsearlo antes de que el objeto esté cerrado lanzará en cada chunk hasta el último. No parsees cada chunk — buferiza y parsea defensivamente, o usa un parser de JSON en streaming construido exactamente para esto si necesitas resultados parciales renderizados incrementalmente (un formulario que se llena progresivamente en una UI, por ejemplo).

def stream_classification(ticket_text: str):
    buffer = ""
    with client.messages.stream(
        model="claude-opus-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": f"Classify this ticket:\n\n{ticket_text}"}],
        output_config={"format": {"type": "json_schema", "schema": TicketClassification.model_json_schema()}},
    ) as stream:
        for text_delta in stream.text_stream:
            buffer += text_delta
            # Only attempt a full parse once the stream is done, unless you have
            # a genuine incremental-rendering need that justifies a partial-JSON parser.

    return TicketClassification.model_validate_json(buffer)

Si genuinamente necesitas rendering incremental, trata los parses parciales estrictamente como estado de display de mejor esfuerzo — nunca valides un objeto parcial contra tu schema Pydantic completo, ya que los campos requeridos que simplemente no han llegado en streaming aún fallarán la validación de una manera que no tiene nada que ver con que el modelo haga algo mal.

Schemas anidados y Union

Las tareas de extracción reales rara vez son un objeto plano. Los modelos anidados y los unions de Pydantic funcionan a través del mismo path de output_format / output_config sin plumbing adicional, lo cual es uno de los argumentos más fuertes para definir schemas en Pydantic en lugar de escribir a mano dicts de JSON Schema a medida que crece la complejidad.

class LineItem(BaseModel):
    description: str
    quantity: int
    unit_price: float

class RefundRequest(BaseModel):
    order_id: str
    items: list[LineItem]
    reason: Literal["damaged", "wrong_item", "changed_mind", "other"]

response = client.messages.parse(
    model="claude-opus-5", max_tokens=1024,
    messages=[{"role": "user", "content": f"Extract the refund request:\n\n{email_text}"}],
    output_format=RefundRequest,
)
refund = response.parsed_output
total = sum(item.quantity * item.unit_price for item in refund.items)

El modo de fallo que vale la pena probar deliberadamente aquí es una lista vacía o de un solo elemento donde tu prompt implicaba que normalmente habría varios — items: [] es válido contra el schema de arriba pero es muy probablemente una señal de que el modelo no encontró ningún line item en lugar de que el email genuinamente tuviera cero, y esa ambigüedad es exactamente el tipo de cosa que un schema no puede distinguir por ti. Agrega una restricción explícita de min_length=1 si una lista vacía nunca es realmente válida para tu caso de uso, de modo que esa ambigüedad se convierta en un fallo de validación que tu loop de retry pueda capturar en lugar de que un 0 fluya silenciosamente hacia downstream.

Salidas estructuradas dentro de un loop de agente

Todo lo de arriba asume una sola llamada aislada, pero la validación de salidas estructuradas importa aún más una vez que está sentada dentro de un nodo de un agente más grande en lugar de estar sola — una mala clasificación allí no solo devuelve una respuesta incorrecta, enruta el resto de la ejecución del agente por el camino incorrecto. Si estás construyendo un agente lo suficientemente complejo como para necesitar branching explícito y ciclos en lugar de una sola llamada lineal, ese es el punto donde una chain simple deja de ser suficiente y algo como LangGraph se gana su complejidad — y un nodo estilo TicketClassification es exactamente el tipo de cosa que termina condicionando un edge condicional.

def classify_node(state: AgentState) -> dict:
    response = client.messages.parse(
        model="claude-opus-5", max_tokens=1024,
        messages=[{"role": "user", "content": f"Classify this ticket:\n\n{state['ticket_text']}"}],
        output_format=TicketClassification,
    )
    return {"classification": response.parsed_output}

def route_on_category(state: AgentState) -> str:
    return "escalate" if state["classification"].requires_escalation else "auto_respond"

El loop de retry-on-validation-failure de antes pertenece dentro de classify_node, no envuelto alrededor de toda la invocación del grafo — un fallo de validación en la salida de un nodo no debería forzar a repetir cada nodo que ya se ejecutó exitosamente. Mantener el retry local al nodo que produce la salida estructurada es lo que hace que esto se componga limpiamente con el resto del manejo de errores de un grafo en lugar de pelear con él.

Checklist de producción

• Usa un modo de output consciente del schema (.parse() o output_config.format), no un request solo por prompt para JSON, para cualquier cosa de la que dependa el código downstream para parsear exitosamente.
• Valida a través de Pydantic incluso cuando la API garantiza la forma — forma y schema son garantías diferentes, y solo una de ellas se hace cumplir en tiempo de generación.
• Retry con el error de validación incluido, limitado a un conteo de intentos pequeño y explícito con una excepción real al final.
• Mantén un fallback de extracción defensiva para el caso de que una llamada restringida aún no regrese como se espera — pero valida lo que sea que recupere antes de confiar en ello.
• Nunca valides un buffer parcial en streaming contra el schema completo — trata el estado parcial como solo para display.
• Agrega restricciones explícitas (min_length, rangos de valores) para cualquier cosa que un valor meramente-válido-pero-vacío o meramente-válido-pero-implausible rompería downstream.
• Registra los fallos de validación con la salida del modelo cruda adjunta — un bucket creciente de fallos de retry en el mismo campo es una señal de que tu schema o prompt necesita cambiar, no solo que los retries están absorbiendo el problema.

Un detalle relacionado que vale la pena señalar para quien venga de una codebase más antigua: algunos SDKs todavía exponen un parámetro output_format de nivel superior directamente en messages.create() de antes de que existieran el helper .parse() actual y la forma de output_config.format. Ese parámetro está deprecado — el código nuevo debería usar output_config: {format: {...}} en messages.create() para el path de schema crudo, o .parse() con un modelo Pydantic para el path tipado mostrado a lo largo de este artículo. Si mantienes una integración existente y ves el nombre del parámetro antiguo, es una señal de que el código es anterior a la superficie actual de salida estructurada y vale la pena migrar en lugar de dejarlo como está, ya que los parámetros deprecados son exactamente el tipo de cosa que silenciosamente deja de ser compatible en una versión futura del SDK sin advertencia en tiempo de llamada.

Conclusión

La brecha entre una demo de salida estructurada y un pipeline de salida estructurada en el que puedes confiar en producción está completamente en cómo cada capa maneja que el modelo no haga exactamente lo que pediste. Un modelo Pydantic te da un contrato real y ejecutable en lugar de un comentario que describe uno; un modo de output consciente del schema hace que la forma de la respuesta sea estructural en lugar de solicitada; un loop de retry que le entrega al modelo su propio error de validación se recupera de los fallos de schema que de todas formas logran pasar; y un fallback defensivo captura el resto sin pretender que el texto recuperado es automáticamente válido. Ninguno de estos individualmente es ingeniería exótica — la confiabilidad viene de usar los cuatro juntos, deliberadamente, en lugar de descubrir en producción cuál omitiste.