LLM의 구조화 JSON을 Pydantic으로 검증하기

“JSON만 반환하라”는 형식 지시일 뿐 데이터 계약이 아닙니다. LLM은 Markdown을 추가하거나 필수 field를 빼고, 잘못된 type이나 새로운 enum 값을 반환할 수 있습니다. 로컬 검증이 끝나기 전까지 응답을 신뢰할 수 없는 text로 취급해야 합니다.

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를 거부하지만, schema를 Prompt에 넣는 것 자체는 보장이 아니며 최종 판단은 로컬 validator가 합니다.

Parsing 전에 원본 응답 보존

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, 잘못된 type, 허용되지 않은 enum, 추가 field를 구분합니다. Repair에는 exc.errors(include_url=False, include_input=False)에서 path, type, message만 추려 전달하여 거부된 input이 진단 object에 다시 들어가지 않게 합니다.

Repair는 한 번만 허용

Repair는 새로운 request입니다. 원본 응답, 간단한 error list, 같은 schema를 보내고 사실을 추가하지 말라고 지시합니다. 결과도 같은 validator로 다시 검사합니다. Loop는 initial response와 한 번의 repair, 총 2 passes뿐입니다. 둘 다 invalid라면 value=None, 두 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를 만듭니다. 첫 번째의 acceptance, 두 번째의 expected error types, 두 응답이 모두 invalid일 때의 명시적 failure를 확인합니다.

이 글의 local test는 Python과 Pydantic 2.12.5에서 통과했습니다. 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, 한 번의 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