Strukturiertes JSON aus einem LLM mit Pydantic validieren

„Nur JSON zurückgeben“ ist eine Formatanweisung, aber kein Datenvertrag. Ein LLM kann weiterhin Markdown ergänzen, ein Pflichtfeld auslassen, einen falschen Typ liefern oder einen neuen Enum-Wert erfinden. Behandeln Sie die Antwort deshalb als nicht vertrauenswürdigen Text, bis der lokale Code sie validiert hat.

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

Einen einzigen Vertrag definieren

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

Erzeugen Sie das Schema für den Prompt mit SupportTicket.model_json_schema(). Damit stammen Prompt und Validator aus demselben Modell. extra="forbid" weist unerwartete Felder zurück; die endgültige Entscheidung trifft trotzdem nur die lokale Validierung.

Die Rohantwort aufbewahren

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

Bewahren Sie die ursprüngliche Zeichenfolge bis zum Abschluss der Prüfung auf. Für Diagnosen gehören solche Daten in geschützten Speicher mit begrenztem Zugriff und kurzer Aufbewahrung. Vollständige Prompts, personenbezogene Daten, Schlüssel und sensible Antworten haben in gemeinsamen Logs nichts zu suchen. HTTP 200 bestätigt nur eine API-Antwort, nicht die Einhaltung des Schemas.

Den JSON-String direkt validieren

from pydantic import ValidationError


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

Mit strict=True wird die Zeichenfolge "true" nicht still in einen booleschen Wert umgewandelt. ValidationError unterscheidet ungültige JSON-Syntax, fehlende Felder, falsche Typen, unzulässige Enum-Werte und zusätzliche Felder. Für den Reparaturauftrag genügen Pfad, Typ und Meldung aus exc.errors(include_url=False, include_input=False).

Nur einen Reparaturversuch erlauben

Eine Reparatur ist eine neue Anfrage. Senden Sie die Rohantwort, die kompakte Fehlerliste und dasselbe Schema mit der Anweisung, keine Fakten hinzuzufügen. Prüfen Sie das Ergebnis mit derselben Funktion. Die Schleife hat genau zwei Durchläufe: ursprüngliche Antwort und eine Reparatur. Scheitern beide, geben Sie value=None, beide Rohantworten und die letzte Fehlerliste an eine manuelle Prüfung oder Fehlerwarteschlange weiter.

Netzwerkfehler bleiben getrennt: Ein timeout beweist kein ungültiges JSON. Führen Sie keinen tool call, keine Zahlung und keine andere externe Aktion aus, bevor die Validierung erfolgreich war.

Den Vertrag ohne API testen

Verwenden Sie einen gültigen Fixture und einen zweiten mit ungültigem Enum, leerer Liste, String statt Boolean und Zusatzfeld. Prüfen Sie die Annahme des ersten, die erwarteten Fehler des zweiten und den expliziten Fehlschlag nach zwei ungültigen Antworten.

Für diesen Artikel liefen die lokalen Tests mit Python und Pydantic 2.12.5 erfolgreich. Es gab keinen echten Modellaufruf; daraus folgt keine Garantie, dass eine bestimmte Model ID immer gültiges JSON erzeugt.

Vor der Produktion sollten Sie Modell und Consumer gemeinsam versionieren, Antwortgröße und Versuche begrenzen, Transport/API/Schema-Metriken trennen, Logs schützen und bei jeder Schemaänderung Fixtures ergänzen.

Das aktuelle Anfrageformat steht in BetterToken Chat Completions. Die API erzeugt Text; Pydantic ist die lokale Vertrauensgrenze, die entscheidet, ob daraus Daten werden.

Vollständige ausführbare Referenz

Diese Blöcke erhalten die exakte ausführbare Referenz für Schemaerzeugung, kompakte Fehler, einen Reparaturversuch und feste Fixtures.

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

Quellen

• Pydantic Models
• Pydantic JSON
• BetterToken Chat Completions