Валидация ответа LLM в Python: схемы и повторный запрос

Вы просите модель вернуть JSON с полями intent, confidence и entities. В девяти случаях из десяти она это делает. В десятом — оборачивает JSON в тройные бэктики с подписью json, ставит confidence: "high" вместо числа, добавляет поле reasoning, которого вы не просили, или обрывает ответ на середине из-за лимита токенов. Ваш json.loads() падает с JSONDecodeError, а дальше по цепочке — KeyError и 500 в API.

Разбираем один узкий приём: как валидировать структуру ответа Pydantic-схемой и автоматически перезапрашивать модель, подставляя в новый промпт текст ошибки валидации. Это формат для тех, кто уже дергает LLM из Python и упёрся в нестабильность структуры ответа. Промпт-инжиниринг (как правильно сформулировать инструкцию) и стриминг — отдельные темы, здесь их не трогаем. Версии: Python 3.11+, Pydantic 2.x, клиент openai 1.x.

Почему прямой парсинг ломается

Модель — генеративная, а не детерминированная. Даже с temperature=0 и системной инструкцией «отвечай только JSON» вы не получаете гарантии структуры. Типичные сбои, которые я вижу на живых логах:

• Обёртка Markdown. Ответ приходит как ```json\n{...}\n```. json.loads на такой строке падает сразу.
• Неверный тип. Поле объявлено числом, модель вернула строку "0.9" или слово "высокая". JSON валиден, а данные — нет.
• Лишние или недостающие поля. Модель домыслила explanation или забыла обязательное id.
• Обрыв по лимиту. Ответ упёрся в max_tokens, JSON не закрыт. Это не лечится валидацией — это лечится увеличением лимита, но поймать факт обрыва схема помогает.

Три первых случая — задача для схемы. Вопрос в том, что делать после того, как схема поймала ошибку: падать или дать модели второй шанс.

Схема как контракт

Pydantic описывает ожидаемую структуру и сам приводит совместимые типы: строку "0.9" он превратит в float, если поле объявлено числом. Это уже снимает часть проблем без единого перезапроса.

from pydantic import BaseModel, Field, ValidationError
from typing import Literal

# Pydantic 2.x
class Entity(BaseModel):
    name: str
    type: Literal["person", "org", "location"]

class Analysis(BaseModel):
    intent: str
    confidence: float = Field(ge=0.0, le=1.0)
    entities: list[Entity]

Field(ge=0.0, le=1.0) отсекает confidence вне диапазона — модель любит возвращать 1.5 или -0.2. Literal запрещает произвольные значения типа: если модель напишет "company" вместо "org", вы это увидите на этапе валидации, а не в отчёте через неделю.

Снятие Markdown-обёртки

Прежде чем отдавать строку в Pydantic, вырежьте возможные бэктики. Не полагайтесь на то, что модель их не поставит — она поставит.

import json
import re

def extract_json(raw: str) -> str:
    """Убирает обёртку ```json ... ``` и лишние пробелы."""
    text = raw.strip()
    # снимаем ограждение markdown, если оно есть
    match = re.search(r"```(?:json)?\s*(.*?)```", text, re.DOTALL)
    if match:
        text = match.group(1).strip()
    return text

Перезапрос с текстом ошибки

Ключевая идея: если валидация упала, не выбрасывайте ответ, а покажите модели, что именно не так. Pydantic отдаёт машиночитаемое описание ошибок через ValidationError.errors() — его можно человекочитаемо свернуть и подставить в следующий запрос. Модель почти всегда исправляется со второй попытки, потому что видит конкретную претензию, а не абстрактное «сделай правильно».

from openai import OpenAI

client = OpenAI()  # ключ берётся из переменной окружения OPENAI_API_KEY

SYSTEM = (
    "Ты извлекаешь структуру из текста. "
    "Отвечай только валидным JSON без markdown-обёртки."
)

def ask_with_retry(user_text: str, max_attempts: int = 3) -> Analysis:
    messages = [
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": user_text},
    ]

    for attempt in range(1, max_attempts + 1):
        completion = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            temperature=0,
            response_format={"type": "json_object"},
        )
        raw = completion.choices[0].message.content
        cleaned = extract_json(raw)

        try:
            return Analysis.model_validate_json(cleaned)
        except (ValidationError, json.JSONDecodeError) as err:
            if attempt == max_attempts:
                raise
            # добавляем в диалог, что вернула модель и в чём ошибка
            messages.append({"role": "assistant", "content": raw})
            messages.append({
                "role": "user",
                "content": (
                    f"Ответ не прошёл валидацию: {err}. "
                    "Верни исправленный JSON по той же схеме."
                ),
            })
    raise RuntimeError("недостижимо")

Здесь работают две вещи одновременно. Первая — response_format={"type": "json_object"} заставляет модель отдавать синтаксически валидный JSON на уровне API (поддерживается моделями OpenAI, где заявлен JSON mode). Вторая — Pydantic проверяет уже семантику: типы, диапазоны, набор полей. JSON mode не гарантирует вашу схему, только то, что это будет разбираемый JSON.

Перезапрос с текстом ошибки дешевле, чем усложнение промпта под каждый новый тип сбоя. Схема — единственный источник правды о структуре, а модель подстраивается под неё сама.

Structured Outputs вместо ручного цикла

У новых моделей OpenAI есть режим строгой схемы (Structured Outputs): вы передаёте JSON Schema, и модель гарантированно возвращает структуру, ей соответствующую. В SDK 1.x это метод client.beta.chat.completions.parse с параметром response_format=Analysis (Pydantic-класс передаётся напрямую). Точное имя метода и список поддерживаемых моделей сверяйте в актуальной документации OpenAI — API этой области менялся между версиями SDK. Где режим доступен, цикл перезапроса нужен реже: он остаётся страховкой на случай отказа сервиса или обрыва по токенам.

Сколько попыток и сколько это стоит

Каждая попытка — это ещё один запрос к модели: деньги за токены и добавленная задержка. Три попытки в худшем случае утраивают латентность одного вызова. Поэтому:

• Два-три перезапроса — потолок. Если модель не справилась за три раза с подставленной ошибкой, четвёртый обычно тоже не поможет — проблема в промпте или в самой задаче.
• Обрыв по токенам не перезапрашивайте вслепую. Сначала проверьте completion.choices[0].finish_reason: если там "length", увеличивайте max_tokens, а не шлите тот же запрос заново.
• Логируйте номер попытки. Если в проде доля вторых и третьих попыток растёт, это сигнал, что промпт или схема разошлись с тем, что реально просят пользователи.

Когда какой подход

Что ломается

Три места, где связка даёт сбой на практике:

• Слишком строгая схема на старте. Если навесить Literal на поле, где модель законно возвращает варианты, вы уйдёте в потолок перезапросов на каждом ответе. Сначала соберите реальные значения из логов, потом ужесточайте.
• Ошибка Pydantic раздувает промпт. При большой схеме str(err) может быть многословным. Отдавайте модели только релевантные ошибки через err.errors(), а не полный дамп.
• Молчаливое приведение типов. Pydantic по умолчанию приведёт "5" к 5. Иногда это скрывает баг в промпте. Если нужна строгость, включайте strict-режим для конкретного поля и сверяйте его поведение с документацией Pydantic 2.x.

Что настроить дальше

Оберните ask_with_retry в замер: логируйте finish_reason, номер удачной попытки и текст ошибки при неудаче. Через неделю в проде вы увидите, какие поля модель путает чаще всего — это подскажет, что уточнить в промпте или ослабить в схеме. Следующий шаг после стабилизации структуры — таймауты и ретраи на уровне сети (отдельно от перезапросов по валидации), чтобы падение сервиса не выглядело как ошибка схемы.