Публикация AiManual

Pydantic AI в продакшене: типизированные ответы, валидация и управляемые ретраи LLM

Pydantic AI превращает свободный текст модели в валидированные объекты Python: разбираем модели результата через BaseModel и Field, обработку ValidationError, у

14 мин чтения 20.09.2026

Коротко

Что будет в материале

01
Почему типизация ответов LLM - это не роскошь, а необходимость
02
Модели результата: как описать ожидаемый ответ LLM
03
Валидация ответов LLM: как Pydantic AI обрабатывает ошибки
04
Управляемые ретраи: настройка и стратегии

Свободный текст модели удобно показывать человеку и тяжело принимать в коде. Pydantic AI закрывает этот разрыв: агент описывается Pydantic-моделью результата, LLM получает JSON Schema, а библиотека проверяет ответ и при несоответствии возвращает ошибку в диалог, чтобы модель исправилась на следующей попытке.

На практике схема даёт три эффекта. Код перестаёт заниматься ручным разбором строк, тип ответа известен статически, а количество обращений к модели становится управляемым параметром, а не лотереей. Ниже разобрано, из чего складывается такая конструкция, как настраивать ретраи и где проходят границы применимости.

Одно уточнение по версиям, без которого примеры не заведутся. API Pydantic AI менялся, и в актуальной документации параметр типа результата агента называется output_type (ai.pydantic.dev). Имена стоит сверять с установленной у вас версией: в материалах, на которые опирается статья, подтверждено только текущее имя output_type, а сведения о переименованиях result_type → output_type, system_prompt → instructions и data → output в версиях до 1.0 не подтверждены, поэтому перед обновлением проверяйте их по официальному миграционному руководству (MigrationGuide | Pydantic Docs).

Почему типизация ответов LLM - это не роскошь, а необходимость

Один и тот же промпт на одной и той же модели в понедельник возвращает чистый JSON, а в пятницу добавляет перед ним фразу «Конечно, вот результат». Продакшен ломается не на генерации, а на разборе того, что пришло.

Проблемы неструктурированного вывода LLM

Классы сбоев повторяются от проекта к проекту:

JSON обёрнут в markdown-блок или сопровождается пояснением, из-за чего json.loads падает на первом символе;
поле приходит под другим именем: user_name вместо name, age_years вместо age;
тип не совпадает: число отдано строкой «34», список превратился в строку «billing, access»;
обязательное поле отсутствует, зато появилось лишнее, которого нет в схеме;
значение проходит по типу, но нарушает смысл: priority = "срочно" вместо ожидаемого high, дата «вчера» вместо ISO-строки.

Последний пункт - самый неприятный. Регулярки и ручной парсинг ловят синтаксис и молчат про смысл. Строгий режим схем на стороне провайдера тоже не спасает: он проверяет структуру JSON, но не кросс-полевую логику вроде «если тип обращения billing, то сумма обязательна», и не гарантирует, что Literal заполнен осмысленным значением, а не первым попавшимся из перечисления.

Итог без типизации предсказуем: обработчики ошибок разрастаются, тесты покрывают не бизнес-логику, а устойчивость парсера, а падения выясняются в момент, когда пользователь уже получил неверный ответ.

Как Pydantic AI решает эти проблемы

Pydantic AI строится поверх Pydantic v2 и его ядра валидации. Схема ответа не пишется отдельно на JSON Schema: её выводят из обычной Pydantic-модели. Дальше работает конвейер из четырёх шагов:

Модель результата превращается в JSON Schema с описаниями полей и ограничениями.
Схема уходит провайдеру либо как нативный structured output, если модель его поддерживает, либо как описание инструмента, который модель обязана вызвать.
Ответ разбирается и валидируется средствами Pydantic: типы, диапазоны, Literal, собственные валидаторы.
При ValidationError текст ошибки не теряется, а отправляется обратно модели как сообщение с просьбой исправить ответ. Цикл повторяется до заданного лимита попыток.

Практическая выгода простая. На выходе вы получаете объект Python с гарантированными типами, а не строку, которую надо перепроверять. Ошибки формата превращаются в ещё один запрос к модели, а не в исключение посреди бизнес-транзакции. Ограничения тоже есть, и о них речь в конце статьи.

Модели результата: как описать ожидаемый ответ LLM

Модель результата - обычный класс Pydantic. Она описывает и структуру, и смысл: описания полей читает модель, ограничения проверяет валидатор.

Создание Pydantic-модели для ответа

from pydantic import BaseModel, Field, EmailStr

class UserInfo(BaseModel):
    name: str = Field(description="Имя пользователя, как оно указано в тексте")
    age: int = Field(ge=0, le=120, description="Возраст в полных годах")
    email: EmailStr = Field(description="Контактный email")
    tags: list[str] = Field(default_factory=list, description="Метки интересов, до 5 штук")

Три вещи в этом примере важнее остальных. description попадёт в схему и станет подсказкой для LLM, поэтому формулировка «Возраст в полных годах» снижает число ошибок заметнее, чем любые уговоры в промпте. Ограничения ge и le отсекают заведомо невозможные значения на стороне Python. EmailStr требует установленного пакета email-validator, иначе импорт упадёт.

Вложенные модели, списки объектов, Optional и union-типы поддерживаются, потому что Pydantic умеет всё это из коробки. Два ограничения на будущее: чем глубже вложенность, тем длиннее схема и тем чаще модель путает уровни, а поля с default провоцируют LLM вообще их не заполнять. Если значение критично, значение по умолчанию лучше не ставить.

Собственная логика добавляется валидаторами Pydantic. Они выполняются при разборе ответа, и их ValueError тоже попадает в текст, который увидит модель:

from pydantic import BaseModel, field_validator

class Ticket(BaseModel):
    title: str
    priority: str
    sla_hours: int = Field(gt=0)

    @field_validator("priority")
    @classmethod
    def normalize_priority(cls, v: str) -> str:
        allowed = {"low", "normal", "high"}
        value = v.strip().lower()
        if value not in allowed:
            raise ValueError(f"priority должен быть одним из {sorted(allowed)}")
        return value

Сообщение об ошибке формулируйте так, чтобы модели было понятно, что делать. «priority должен быть одним из ['high', 'low', 'normal']» даёт ей готовый выбор, а «неверное значение» не даёт ничего.

Подключение модели к агенту

from pydantic_ai import Agent

agent = Agent(
    "openai:gpt-4o",
    output_type=UserInfo,
    instructions="Извлекай данные пользователя из текста. Чего нет в тексте, не придумывай.",
)

result = agent.run_sync("Меня зовут Анна, 34 года, пишите на anna@example.com")
print(type(result.output))  # UserInfo
print(result.output.age)    # 34

Конструктор принимает строку вида provider:model, так что смена провайдера - это правка одной строки. Агент возвращает объект результата, у которого поле output содержит уже валидированный экземпляр UserInfo.

Про выбор модели: способ доставки схемы зависит от провайдера. По умолчанию Pydantic AI использует механизм tool calling модели: JSON-схема каждого типа вывода передаётся модели как схема параметров специального output-инструмента (Output | Pydantic Docs). Альтернативный режим Native Output опирается на нативную функцию модели «Structured Outputs» (JSON Schema response format), при которой модель обязана выводить только текст, соответствующий предоставленной JSON-схеме. Этот режим поддерживается не всеми моделями и иногда имеет ограничения. Практический вывод: под задачу с жёсткой схемой стоит проверять, умеет ли выбранная модель нативный structured output, а слабые локальные модели тестировать на своих реальных примерах перед выводом в продакшен.

Валидация ответов LLM: как Pydantic AI обрабатывает ошибки

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

Что такое ValidationError и как он возникает

ValidationError - исключение Pydantic, которое возникает, когда данные не соответствуют схеме модели. Внутри него лежит список ошибок с путём до проблемного поля, типом ошибки и человекочитаемым сообщением. Причины делятся на группы:

неверный тип, например модель вернула age: "тридцать четыре" вместо целого числа;
отсутствующее обязательное поле;
значение вне диапазона, заданного через Field(ge=..., le=...);
значение не входит в Literal или в перечисление;
ValueError из собственного валидатора или проверка на уровне всей модели через model_validator.

Смысл в том, что ошибка не «съедается». Pydantic AI перехватывает ValidationError, собирает читаемое описание проблемных полей и отправляет его модели как обратную связь вместе с предыдущим ответом. Модель видит, где именно она промахнулась, и правит конкретное место, а не переписывает ответ целиком наугад.

Автоматические повторные попытки при ошибках валидации

Механизм работает без вашего участия. Схема не сошлась - библиотека добавляет к диалогу сообщение с описанием ошибки и запрашивает новый ответ. Количество таких кругов ограничено параметрами агента:

agent = Agent(
    "openai:gpt-4o",
    output_type=UserInfo,
    retries=2,
)

Аргумент retries задаёт бюджеты повторных попыток: целое число переопределяет сразу оба бюджета - и для инструментов, и для вывода, - а словарь AgentRetries позволяет переопределить только один, например retries={'tools': 3} (pydantic_ai.agent | Pydantic Docs). Значение по умолчанию для retries на уровне агента равно единице: число повторных попыток для инструмента по умолчанию берётся из значения агента, которое равно 1. Для дорогих моделей разумно начинать с этого числа и повышать его только там, где статистика показывает частые промахи.

Что происходит, когда лимит исчерпан: попытки прекращаются, и исключение уходит наверх как ошибка выполнения агента. Это правильно. Бесконечный цикл «модель ошибается - модель исправляется» в продакшене означает неконтролируемый счёт за токены.

Управляемые ретраи: настройка и стратегии

Ретрай - это повторный полный запрос к модели с историей диалога, а не бесплатная операция. Арифметика простая: промпт на 4000 токенов и три попытки превращаются в 12 000 входных токенов на одну задачу. Лимиты повторов ставят прямо из этой логики, а не из соображения «пусть попробует побольше».

Настройка лимита повторных попыток

Разумные варианты по контексту:

дешёвая быстрая модель и простая плоская схема: retries=1, иногда 2;
дорогая модель и большая вложенная схема: retries=2 или 3, но с обязательным логированием каждого повтора;
пакетная ночная обработка, где задержка не критична: лимит можно поднять, если повторные вызовы стоят дешевле, чем ручной разбор сбойных записей.

Отдельный предохранитель - ограничение общего числа запросов на запуск. Лимит повторов защищает от зацикливания валидации, но не от агента, который десять раз вызывает инструменты. Для этого служит UsageLimits - структура для ограничения использования модели (токенов и/или запросов) на прогонах агента (pydantic_ai.usage | Pydantic Docs):

from pydantic_ai import Agent, UsageLimits

result = await agent.run(
    prompt,
    usage_limits=UsageLimits(request_limit=4),
)

Параметр request_limit задаёт максимальное число запросов к модели. Счётчик запросов отслеживается pydantic_ai, и лимит проверяется перед каждым запросом к модели. При превышении лимита выполнение прерывается исключением, и это заметно дешевле, чем разбираться с сюрпризом в счёте за месяц. Точные имена полей UsageLimits в вашей версии проверьте в документации: набор доступных ограничений может отличаться от релиза к релизу.

Использование ModelRetry для ручного управления

Автоматика реагирует на несоответствие схеме. Часть проблем схемой не выражается: значение корректно по типу, но противоречит данным во внешней системе. Здесь в дело вступает исключение ModelRetry. Его можно поднять из валидатора результата, и Pydantic AI вернёт модели текстовое пояснение вместо того, чтобы валить весь запуск.

from pydantic_ai import Agent, ModelRetry, RunContext

agent = Agent("openai:gpt-4o", output_type=Invoice, deps_type=Deps, retries=2)

@agent.output_validator
async def check_counterparty(ctx: RunContext[Deps], data: Invoice) -> Invoice:
    exists = await ctx.deps.registry.exists(data.counterparty_id)
    if not exists:
        raise ModelRetry(
            "Контрагент с таким id не найден. Выбери id из списка: "
            + ", ".join(await ctx.deps.registry.list_ids())
        )
    return data

Валидатор результата запускается после проверки схемой, получает доступ к зависимостям и может вернуть исправленный объект. Текст внутри ModelRetry попадает в диалог, поэтому он должен быть инструкцией, а не диагнозом: не «контрагент неверный», а «выбери id из этого списка». Валидаторы вывода выполняются и при пустом ответе (None) и могут сами инициировать повторную попытку, подняв ModelRetry (Retries | Pydantic Docs). Исключения, отличные от ModelRetry, не приводят к повтору: они поднимаются наверх, что удобно для проверки прав доступа и других ситуаций, где повтор бессмысленен.

Комбинированная стратегия выглядит так: схема ловит формальные ошибки, ModelRetry в валидаторе результата ловит смысловые, инструменты возвращают ModelRetry при временных сбоях внешних API, а UsageLimits ограничивает общий бюджет запуска. Каждый из этих слоёв нужно логировать отдельно, иначе непонятно, на каком уровне агент тратит попытки.

Зависимости и контекст выполнения

Инструментам и валидаторам обычно нужны внешние ресурсы: пул соединений, HTTP-клиент, конфиг, объект пользователя. Протаскивать их через глобальные переменные плохо, потому что тогда агент нельзя тестировать изолированно.

Определение и использование зависимостей

Pydantic AI передаёт зависимости через тип и объект RunContext. Тип объявляется в конструкторе агента, значение передаётся при запуске:

from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext

@dataclass
class Deps:
    http: httpx.AsyncClient
    tenant_id: str

agent = Agent("openai:gpt-4o", deps_type=Deps, output_type=Report)

@agent.tool
async def fetch_metrics(ctx: RunContext[Deps], service: str) -> dict:
    response = await ctx.deps.http.get(
        f"/metrics/{service}",
        headers={"X-Tenant": ctx.deps.tenant_id},
    )
    response.raise_for_status()
    return response.json()

result = await agent.run(
    "Собери отчёт по сервису payments",
    deps=Deps(http=client, tenant_id="acme"),
)

Доступ к зависимостям получают и инструменты, и валидаторы результата. Это даёт две вещи. Во-первых, изоляцию арендаторов и клиентов: один агент обслуживает разные контексты без глобального состояния. Во-вторых, тестируемость: в тестах подставляется словарь с подготовленными ответами вместо реального HTTP-клиента, и логика агента проверяется без обращения к сети.

Держите зависимости явными и узкими. Если в Deps оказывается полторы дюжины полей, это признак того, что агенту пытаются поручить слишком много.

Наблюдаемость: логирование и трейсинг

Агент с валидацией и ретраями ведёт себя нелинейно: один запрос пользователя может превратиться в три обращения к модели и два вызова инструментов. Без трассировки понять, где ушло время и деньги, невозможно.

Логирование ключевых событий

Минимальный набор событий, который стоит писать в лог:

идентификатор запуска и входной промпт в усечённом виде;
факт ошибки валидации с перечнем проблемных полей;
номер повторной попытки и причина, по которой она понадобилась;
итоговый статус и объект usage с числом токенов и вызовов;
общее время выполнения и время каждого обращения к модели.

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
log = logging.getLogger("agent")

result = await agent.run(prompt, deps=deps)
log.info(
    "run ok",
    extra={"usage": result.usage(), "retries_needed": retries_observed},
)

Структурированные логи с полями удобнее текстовых строк: по ним строится график доли запусков, которым потребовался повтор, и видно, какой тип ошибки доминирует. Если 20 процентов задач стабильно требуют второй попытки, дешевле поправить описания полей в модели или сузить схему, чем увеличивать лимит повторов.

Интеграция с OpenTelemetry

Pydantic AI является OpenTelemetry-native: capability Instrumentation создаёт стандартные OTel-спаны для каждого вызова модели и каждого вызова инструмента, и работает любой OTLP-бэкенд (Pydantic AI | Pydantic Docs). Спаны создаются для самого прогона агента, каждого запроса к модели и каждого выполнения инструмента, следуя OpenTelemetry Semantic Conventions for Generative AI (Instrumentation | Pydantic Docs).

Схема настройки стандартная: подключается экспортёр OTLP с адресом коллектора, а платформа для просмотра трейсов выбирается любая, поддерживающая OTel. Отдельно существует Logfire от той же команды - платформа наблюдаемости, которая показывает структуру запусков агента.

Что даёт трейсинг на практике: видно всю цепочку «запрос - вызов инструмента - ошибка валидации - повтор - успех», а не только финальные логи. Узкое место находится быстро, потому что длительность и атрибуты привязаны к конкретному шагу. Для локальной отладки достаточно консольного экспортёра, для продакшена ставят коллектор и хранилище с ретеншеном.

Границы применимости и ограничения Pydantic AI

Pydantic AI выигрывает там, где ответ модели должен превратиться в объект с известной структурой: извлечение данных из писем и документов, классификация обращений, формирование задач, разбор ответов внешних API в единый формат. Плохо он подходит для случаев, где структура ответа заранее неизвестна, а смысл важнее формы: свободный диалог, генерация длинных текстов, творческие задачи.

Ограничения, о которых стоит знать заранее:

качество следования схеме зависит от модели. Слабые локальные модели с длинной вложенной схемой нарушают её чаще, и число повторов растёт. Проверять нужно на своих данных, а не на демо из документации;
схема занимает место в контексте. Большая модель результата с десятками полей и описаний добавляет заметный объём входных токенов к каждому запросу;
режим Native Output, при котором модель обязана выводить текст по предоставленной JSON-схеме, поддерживается не всеми моделями и иногда имеет ограничения. Если нативный structured output недоступен, Pydantic AI по умолчанию работает через вызов инструмента: JSON-схема типа вывода передаётся модели как схема параметров специального output-инструмента;
валидация не проверяет истинность. Если сумма в схеме верна по типу, но противоречит реальности, Pydantic об этом не узнает. Смысловые проверки остаются вашей задачей, и делать их стоит через валидаторы результата;
сложная многошаговая логика с ветвлениями, очередями и фоновыми задачами требует отдельных инструментов. Pydantic AI закрывает слой типизации и вызова модели, а оркестрацию процессов и состояние вам придётся проектировать самим.

Отсюда практическое правило: заводите Pydantic AI под задачи с чётким контрактом ответа и не пытайтесь вытянуть им весь пайплайн сразу. Если агенту нужны собственные метрики надёжности, бюджеты и сложная оркестрация, стоит посмотреть на разбор архитектуры самописного AI-агента с метриками latency, cost и reliability и сравнить два подхода на своих сценариях.

Пример архитектуры: агент с типизацией, валидацией и ретраями

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

import logging
from dataclasses import dataclass
from typing import Literal

import httpx
from pydantic import BaseModel, Field, field_validator
from pydantic_ai import Agent, ModelRetry, RunContext, UsageLimits

log = logging.getLogger("triage")


class TicketData(BaseModel):
    summary: str = Field(description="Суть обращения, одна строка")
    category: Literal["billing", "access", "bug", "other"] = Field(
        description="Категория обращения"
    )
    severity: int = Field(ge=1, le=5, description="1 - косметика, 5 - сервис недоступен")
    affected_users: int = Field(default=0, ge=0, description="Сколько пользователей затронуто")

    @field_validator("summary")
    @classmethod
    def check_summary(cls, v: str) -> str:
        value = v.strip()
        if len(value) < 5:
            raise ValueError("summary слишком короткий, переформулируй суть обращения")
        return value


@dataclass
class Deps:
    http: httpx.AsyncClient


agent = Agent(
    "openai:gpt-4o",
    output_type=TicketData,
    deps_type=Deps,
    instructions="Разбирай обращения в поддержку. Числа бери только из текста.",
    retries=2,
)


@agent.output_validator
async def check_severity(ctx: RunContext[Deps], data: TicketData) -> TicketData:
    if data.severity >= 5:
        response = await ctx.deps.http.get("/oncall/current")
        if response.status_code != 200:
            raise ModelRetry("Дежурный сервис недоступен, поставь severity не выше 4")
    return data


async def triage(text: str) -> TicketData:
    async with httpx.AsyncClient(base_url="https://internal.local") as client:
        result = await agent.run(
            text,
            deps=Deps(http=client),
            usage_limits=UsageLimits(request_limit=4),
        )
    log.info("triage done", extra={"usage": result.usage()})
    return result.output

Разбор по слоям. Модель TicketData задаёт контракт и ограничения: категория ограничена перечислением, критичность диапазоном от 1 до 5, а валидатор отсекает пустые описания. Валидатор результата добавляет проверку, которую схема не выражает: пятая критичность уходит дежурному только при работающем сервисе. Зависимость Deps отдаёт HTTP-клиент и легко подменяется в тестах. UsageLimits страхует от разорения при зацикливании, а лог с объектом usage даёт цифры для последующей настройки лимита повторов.

Что стоит добавить при выводе в продакшен: идентификатор запуска в каждый лог, экспорт спанов через OTel, метрику доли запусков с повторами и алерт на рост этой доли. Начать можно с простого шага: опишите одну модель результата для самой частой задачи, включите логирование повторных попыток и посмотрите на реальные цифры через неделю работы. Они покажут, где именно схема требует уточнения, а где хватает текущих лимитов.

Подписаться на канал