JSON structuré produit par un LLM : validation avec Pydantic

« Renvoyer uniquement du JSON » est une consigne de format, pas un contrat de données. Un LLM peut encore ajouter du Markdown, oublier un champ, utiliser le mauvais type ou inventer une valeur d’énumération. La réponse reste donc un texte non fiable jusqu’à sa validation locale.

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

Définir un contrat unique

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

Générez le schéma envoyé dans le prompt avec SupportTicket.model_json_schema(). Le prompt et le validateur reposent ainsi sur le même modèle. extra="forbid" bloque les champs inattendus, mais seule la validation locale autorise la suite du traitement.

Conserver la réponse brute

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

Gardez la chaîne originale jusqu’à la fin de la validation. Pour le diagnostic, utilisez un stockage protégé, un accès limité et une rétention courte. N’inscrivez pas dans un journal partagé les prompts complets, données personnelles, clés ou réponses sensibles. HTTP 200 indique seulement que l’API a répondu, pas que le texte respecte le schéma.

Valider directement la chaîne JSON

from pydantic import ValidationError


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

Avec strict=True, la chaîne "true" n’est pas convertie silencieusement en booléen. ValidationError distingue une syntaxe JSON invalide, un champ absent, un type erroné, une valeur enum interdite ou un champ supplémentaire. Pour la réparation, n’extrayez que le chemin, le type et le message avec exc.errors(include_url=False, include_input=False).

Limiter la réparation à une tentative

Une réparation est une nouvelle requête. Envoyez la réponse originale, la liste compacte des erreurs et le même schéma, en interdisant l’ajout de faits. Validez ensuite avec exactement la même fonction. Le flux comporte deux passages : réponse initiale puis une réparation. Si les deux échouent, renvoyez value=None, conservez les deux réponses et transmettez les erreurs finales à une revue humaine ou à une file d’échecs.

Traitez les erreurs réseau séparément : un timeout ne dit rien sur la validité du JSON. N’exécutez aucun tool call, paiement ou effet externe avant une validation réussie.

Tester sans appel API

Préparez un fixture valide et un autre avec enum invalide, liste vide, chaîne à la place d’un booléen et champ supplémentaire. Vérifiez l’acceptation du premier, les erreurs du second et l’échec explicite après deux réponses invalides.

Pour cet article, ces tests locaux ont réussi avec Python et Pydantic 2.12.5. Aucun appel réel au modèle n’a été effectué ; cela ne garantit donc pas qu’un Model ID précis renvoie toujours un JSON valide.

Avant la production, versionnez le modèle avec son consommateur, limitez la taille des réponses et les tentatives, séparez les métriques transport/API/schéma, protégez les journaux et ajoutez des fixtures à chaque modification du contrat.

Consultez le format actuel dans BetterToken Chat Completions. L’API génère du texte ; Pydantic constitue la frontière locale qui décide s’il peut devenir une donnée.

Référence exécutable complète

Ces blocs conservent la référence exécutable exacte pour générer le schéma, résumer les erreurs, limiter la réparation et tester des fixtures fixes.

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

Sources

• Pydantic Models
• Pydantic JSON
• BetterToken Chat Completions