JSON estruturado de um LLM: validação com Pydantic

“Retorne apenas JSON” é uma instrução de formato, não um contrato de dados. Um LLM ainda pode adicionar Markdown, omitir um campo, usar o tipo errado ou inventar um valor de enum. Trate toda resposta como texto não confiável até a validação local.

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

Defina um ú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

Gere o schema do prompt com SupportTicket.model_json_schema(). Dessa forma, prompt e validador partem do mesmo modelo. extra="forbid" rejeita campos inesperados, mas somente a validação local decide se a saída pode ser aceita.

Preserve a resposta bruta

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

Mantenha a string original até o fim da validação. Para diagnóstico, use armazenamento protegido, acesso restrito e retenção curta. Não registre prompts completos, dados pessoais, chaves ou respostas sensíveis. HTTP 200 comprova que houve resposta, não que o texto respeita o contrato.

Valide a string JSON diretamente

from pydantic import ValidationError


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

Com strict=True, a string "true" não vira booleano silenciosamente. ValidationError separa JSON inválido, campo ausente, tipo errado, enum não permitido e campo extra. Para o reparo, extraia somente caminho, tipo e mensagem com exc.errors(include_url=False, include_input=False).

Permita apenas um reparo

Reparo é uma nova solicitação. Envie a resposta original, o resumo dos erros e o mesmo schema, com a instrução de não inventar fatos. Valide novamente com a mesma função. O fluxo tem duas passagens: resposta inicial e um reparo. Se ambas falharem, retorne value=None, preserve as duas respostas e encaminhe os erros finais para revisão manual ou uma fila de falhas.

Erros de transporte ficam em outra categoria: timeout não significa JSON incorreto. Não execute tool calls, pagamentos ou qualquer ação externa antes da validação bem-sucedida.

Teste sem chamar a API

Crie fixtures fixos: um válido e outro com enum inválido, lista vazia, string no lugar de booleano e campo extra. Confirme a aceitação do primeiro, os erros do segundo e a falha explícita após duas respostas inválidas.

Neste artigo, os testes locais passaram com Python e Pydantic 2.12.5. Nenhuma chamada real ao modelo foi feita; portanto, não há promessa de que um Model ID específico sempre produza JSON válido.

Antes de produção, versione o modelo junto do consumidor, limite tamanho e tentativas, separe métricas de transporte/API/schema, proteja logs e adicione fixtures a cada mudança do contrato.

Veja o formato atual em BetterToken Chat Completions. A API gera texto; o Pydantic é a fronteira local que decide se ele pode ser tratado como dado.

Referência executável completa

Estes blocos preservam a referência executável exata para gerar o schema, resumir erros, limitar o reparo e testar fixtures fixos.

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

Fontes

• Pydantic Models
• Pydantic JSON
• BetterToken Chat Completions