LLM से structured JSON: Pydantic से validation

“केवल JSON लौटाओ” format का निर्देश है, data contract नहीं। LLM Markdown जोड़ सकता है, आवश्यक field छोड़ सकता है, गलत type दे सकता है या नया enum value बना सकता है। इसलिए local validation पूरी होने तक हर response को अविश्वसनीय text मानें।

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

एक ही Pydantic contract रखें

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

Prompt का schema SupportTicket.model_json_schema() से बनाएँ। इससे prompt और local validator एक ही model से निकलते हैं। extra="forbid" अनपेक्षित field को रोकता है, पर schema को prompt में लिखना guarantee नहीं है; अंतिम निर्णय validator का है।

Parsing से पहले raw response बचाएँ

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

Validation खत्म होने तक मूल string रखें। Diagnostic storage की access और retention सीमित होनी चाहिए। Shared log में पूरा prompt, personal data, API key या sensitive response न लिखें। HTTP 200 केवल API response मिलने की पुष्टि करता है, business schema की नहीं।

JSON string को सीधे validate करें

from pydantic import ValidationError


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

strict=True होने पर string "true" चुपचाप boolean नहीं बनेगी। ValidationError invalid JSON syntax, missing field, wrong type, unsupported enum और extra field को अलग दिखा सकता है। Repair के लिए exc.errors(include_url=False, include_input=False) से केवल path, type और message लें, rejected input नहीं।

केवल एक repair attempt करें

Repair एक नया request है। मूल response, compact error list और वही schema भेजें तथा model को नया fact न जोड़ने को कहें। फिर उसी validator से परिणाम जाँचें। Loop में केवल दो passes हों: initial response और एक repair। दोनों invalid हों तो value=None, दोनों raw responses और अंतिम errors लौटाएँ; caller इसे manual review या error queue में भेजे।

Network timeout को schema error से अलग रखें। Validation सफल होने से पहले tool call, payment या कोई external operation न चलाएँ। Unlimited retry quota खर्च करता है और sensitive text बार-बार भेज सकता है।

API के बिना contract test करें

एक valid fixture और एक invalid fixture रखें जिसमें गलत enum, खाली list, boolean की जगह string और extra field हो। पहले की acceptance, दूसरे के expected error types और दो invalid responses के बाद explicit failure जाँचें।

इस लेख के local tests Python और Pydantic 2.12.5 पर pass हुए। Live model call नहीं किया गया, इसलिए किसी खास Model ID के हमेशा valid JSON लौटाने का दावा नहीं है।

Production से पहले model और consumer को साथ version करें, response size तथा attempts सीमित करें, transport/API/schema metrics अलग रखें, logs redact करें और schema बदलने पर fixtures जोड़ें।

वर्तमान request format BetterToken Chat Completions में देखें। API text बनाती है; Pydantic स्थानीय trust boundary है जो तय करती है कि उसे data माना जा सकता है या नहीं।

पूरा executable reference

ये blocks schema generation, compact errors, एक repair attempt और fixed fixtures के exact executable reference को सुरक्षित रखते हैं।

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

स्रोत

• Pydantic Models
• Pydantic JSON
• BetterToken Chat Completions