Структурированный JSON из LLM: валидация через Pydantic
Практический Python-конвейер: сохранить исходный ответ LLM, проверить JSON через Pydantic, выполнить одну repair-попытку и вернуть явную ошибку.
Содержание
Фраза «верни только JSON» описывает желаемый формат, но не создаёт контракт. Модель может добавить Markdown, пропустить поле, вернуть строку вместо числа или придумать новое значение enum. Поэтому ответ LLM нужно считать недоверенной строкой, пока локальный код не проверит его по схеме.
Надёжная граница выглядит так:
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: boolextra="forbid" важен для машинного контура. По умолчанию Pydantic может игнорировать лишние поля; здесь неожиданное поле должно остановить обработку. Ограничения длины не дают принять пустой заголовок или неограниченный список.
JSON Schema для prompt можно получить из той же модели:
import json
schema = SupportTicket.model_json_schema()
schema_text = json.dumps(schema, ensure_ascii=False)Так схема в prompt и локальная проверка происходят из одного источника. Однако наличие схемы в инструкции не гарантирует, что модель ей последует. Решение принимает только валидатор.
Отправьте обычный API-запрос и сохраните raw response
Пример ниже использует документированный OpenAI-compatible Chat Completions интерфейс BetterToken. Ключ и полный Model ID берутся из окружения, а не записываются в код.
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 ""Сохранить raw response означает не потерять исходную строку до валидации. Для короткой операции достаточно держать её в памяти. Если она нужна для диагностики, храните её в защищённом хранилище с ограниченным сроком и доступом. Не пишите в общий лог полный prompt, персональные данные, API Key или чувствительный ответ.
HTTP 200 доказывает, что API вернул ответ, но не то, что текст соответствует вашей бизнес-схеме.
Проверяйте именно JSON-строку
Pydantic v2 умеет валидировать JSON напрямую:
from pydantic import ValidationError
def validate_ticket(raw: str) -> SupportTicket:
return SupportTicket.model_validate_json(raw, strict=True)strict=True не позволит незаметно превратить, например, строку "true" в boolean. Один ValidationError может описывать разные проблемы:
- JSON синтаксически сломан;
- обязательное поле отсутствует;
- тип значения неверен;
- значение не входит в допустимый
Literal; - модель добавила неизвестное поле.
Не исправляйте такие значения с помощью тихих значений по умолчанию. Иначе ошибка модели превращается в данные, которые выглядят подтверждёнными приложением.
Для repair полезен компактный список ошибок без traceback:
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)
]include_input=False уменьшает риск снова записать чувствительное значение в диагностический объект.
Ограничьте repair одной попыткой
Repair — это новый запрос, а не доказательство того, что старый ответ был почти валиден. Передайте исходный ответ, список ошибок и ту же схему. Затем снова проверьте результат тем же валидатором.
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")Цикл имеет ровно два прохода: первый ответ и одна repair-попытка. Если второй ответ невалиден, value остаётся None. Вызывающий код должен обработать этот исход отдельно: запросить ручную проверку, положить задачу в очередь ошибок или показать пользователю безопасное сообщение.
Бесконечный retry скрывает дефект контракта, тратит лимит и может отправлять чувствительный текст снова и снова. Сетевые ошибки тоже обрабатывайте отдельно от ValidationError: таймаут не означает, что JSON был неправильным.
Минимальный локальный тест без API
Сначала проверьте контракт на фиксированных строках. Такой тест быстрый и не зависит от модели или сети.
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")В подготовке этой статьи локальная часть проверена на Python с Pydantic 2.12.5: валидный fixture принят, невалидный отклонён, а симуляция двух невалидных ответов завершилась явным failure. Сетевой запрос к модели не выполнялся, поэтому статья не утверждает, что конкретный Model ID всегда возвращает JSON без repair.
Что проверить перед production
- Версионируйте Pydantic-модель вместе с потребителем результата.
- Проверяйте обязательные поля, enum, длину списков и политику лишних полей.
- Сохраняйте raw response только в рамках вашей политики данных; редактируйте чувствительные значения в логах.
- Разделяйте ошибки транспорта, API и схемы в метриках.
- Ограничьте число repair-попыток и максимальный размер исходного ответа.
- Не выполняйте tool call, платёж или другую внешнюю операцию до успешной валидации.
- Добавьте fixtures для каждого изменения схемы и отдельный тест конечного failure.
Актуальную форму обычного OpenAI-compatible запроса, Base URL и способ выбора Model ID сверяйте в BetterToken Docs. Pydantic в этом конвейере остаётся локальной границей доверия: API генерирует текст, а ваше приложение решает, можно ли считать его данными.
