LLMの構造化JSONをPydanticで検証する

「JSONだけを返して」は形式の指示であり、データ契約ではありません。LLMはMarkdownを追加したり、必須フィールドを省略したり、型やenum値を誤ったりします。ローカル検証が終わるまで、応答は信頼できないテキストとして扱います。

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

Pydanticの契約を一つにする

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とvalidatorが同じmodelを参照します。extra="forbid"は未知のfieldを拒否しますが、Promptにschemaを書いても保証にはならず、受理を決めるのはローカルvalidatorです。

Parse前に生レスポンスを保持する

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

検証が終わるまで元のstringを残します。診断用に保存する場合は、保護されたstorageでaccessとretentionを制限してください。共有logに完全なPrompt、個人情報、API key、機密応答を書いてはいけません。HTTP 200はAPIが応答したことを示すだけで、business schemaへの適合は証明しません。

JSON文字列を直接検証する

from pydantic import ValidationError


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

strict=Trueなら、stringの"true"はbooleanへ暗黙変換されません。ValidationErrorは壊れたJSON、必須field不足、型違い、許可されないenum、追加fieldを区別します。Repairにはexc.errors(include_url=False, include_input=False)からpath、type、messageだけを渡し、拒否された入力を診断objectに重複させません。

Repairは1回に制限する

Repairは新しいrequestです。元の応答、短いerror list、同じschemaを送り、事実を追加しないよう指示します。結果は同じvalidatorで再検証します。Loopはinitial responseと1回のrepairの2 passesだけです。両方がinvalidならvalue=None、2つのraw responses、最後のerrorsを返し、manual reviewかerror queueへ送ります。

Network timeoutはschema errorと分けます。検証成功前にtool call、payment、その他のexternal operationを実行してはいけません。無制限retryはquotaを消費し、機密textを何度も送る原因になります。

APIなしで契約をテストする

有効なfixtureと、enum違反、空list、booleanの代わりのstring、追加fieldを含む無効fixtureを用意します。前者のaccept、後者のexpected error types、2回ともinvalidな場合の明示的failureを確認します。

この記事ではPythonとPydantic 2.12.5でlocal testが成功しました。Live model callは実行していないため、特定のModel IDが常にvalid JSONを返すという主張ではありません。

Production前にはmodelとconsumerを一緒にversion管理し、response sizeとattemptsを制限し、transport/API/schema metricsを分離し、logを保護し、schema変更ごとにfixturesを追加します。

現在のrequest形式はBetterToken Chat Completionsで確認できます。APIはtextを生成し、Pydanticがdataとして扱えるかを決めるローカルtrust boundaryです。

完全な実行用リファレンス

以下はschema生成、短いerror summary、1回のrepair、固定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")

参考資料

• Pydantic Models
• Pydantic JSON
• BetterToken Chat Completions