Pydantic AI — это Python-фреймворк нового поколения от команды Samuel Colvin, создателей легендарной библиотеки Pydantic. Если Pydantic навсегда изменил то, как мы валидируем данные в Python-экосистеме, то Pydantic AI делает то же самое для мира LLM-агентов. Главная идея проста и гениальна: каждый ответ языковой модели должен быть не просто строкой, а строго типизированным объектом, прошедшим валидацию на уровне схемы.

Представьте: вы пишете агента, который анализирует резюме кандидатов. Без Pydantic AI вы получаете от модели текстовый ответ и пишете регулярные выражения, чтобы вытащить оттуда имя, опыт и навыки — хрупко, ненадёжно, больно. С Pydantic AI вы описываете Pydantic-модель Candidate, передаёте её как result_type — и на выходе получаете гарантированно валидный объект с полями правильных типов. Никакого парсинга, никаких исключений.

Фреймворк построен вокруг трёх ключевых абстракций: Agent (агент — основная единица логики), Model (языковая модель — OpenAI, Anthropic, Groq, Google и десятки других через провайдеры) и Tools (инструменты — обычные Python-функции, которые агент может вызывать для получения внешних данных). Всё это собирается в единый пайплайн, где каждый этап типизирован, а ошибки обнаруживаются на этапе написания кода, а не в рантайме.

В этом гайде мы пройдём полный путь: от первой установки библиотеки до построения мультиагентной системы с потоковым выводом и структурированными результатами. Вы научитесь создавать агентов, которые не просто «говорят», а возвращают данные — в том виде, в котором их можно сразу передать в базу данных, REST-ответ или следующий этап пайплайна.

Начнём с установки. Pydantic AI доступен через pip и поддерживает Python 3.10+. Библиотека спроектирована с минимальным количеством обязательных зависимостей — сам Pydantic, httpx для HTTP-вызовов и провайдеры для конкретных LLM, которые вы подключаете по необходимости. Такой подход позволяет не тащить в проект десятки ненужных пакетов, если вы работаете только с одной моделью.

Для работы потребуется API-ключ от провайдера языковой модели. Фреймворк поддерживает два способа передачи ключей: через переменные окружения (рекомендуемый способ для продакшена) и напрямую в коде (удобно для локального прототипирования). В примерах ниже будем использовать OpenAI-совместимый эндпоинт, но синтаксис идентичен для Anthropic, Google Gemini, Groq, Mistral и других.

# Установка ядра Pydantic AI с поддержкой OpenAI
pip install pydantic-ai pydantic-ai-slim

# Если нужен только OpenAI-провайдер
pip install pydantic-ai[openai]

# Или все доступные провайдеры сразу
pip install pydantic-ai[all]

# Добавьте в ~/.bashrc, ~/.zshrc или .env
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export GOOGLE_API_KEY="AIza..."

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from pydantic import BaseModel

# Простейший агент: проверяем, что всё работает
model = OpenAIModel("gpt-4o")
agent = Agent(model=model, system_prompt="Ты — лаконичный ассистент.")

result = agent.run_sync("Скажи 'Pydantic AI работает!' одним предложением.")
print(result.data)  # → Pydantic AI работает! Установка прошла успешно.

Обратите внимание: метод run_sync() — синхронный и блокирующий. Он идеален для скриптов и Jupyter-ноутбуков, но в веб-приложениях лучше использовать асинхронный agent.run() — к нему вернёмся в секции про стриминг. Главное, что вы увидели: агент уже работает, и result.data возвращает строку — самый простой тип результата.

Центральная абстракция Pydantic AI — класс Agent. Это не просто обёртка над LLM: агент хранит системный промпт, список инструментов, тип ожидаемого результата и всю конфигурацию выполнения. Конструктор принимает обязательный параметр model и опциональные system_prompt, result_type, tools.

Системный промпт определяет «личность» агента — его роль, тон, ограничения. В отличие от обычных вызовов ChatGPT API, где вы каждый раз передаёте system message, здесь промпт задаётся один раз при создании агента и автоматически подставляется во все вызовы. Это архитектурно чище: агент становится самодостаточной сущностью со своим поведением.

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from pydantic import BaseModel, Field
from datetime import date


class WeatherReport(BaseModel):
    """Структурированный отчёт о погоде."""
    city: str = Field(description="Название города")
    temperature_c: float = Field(description="Температура в градусах Цельсия")
    condition: str = Field(description="Погодные условия: sunny, cloudy, rainy и т.д.")
    humidity_pct: int = Field(ge=0, le=100, description="Влажность в процентах")
    forecast_date: date = Field(description="Дата прогноза")


model = OpenAIModel("gpt-4o")
weather_agent = Agent(
    model=model,
    system_prompt=(
        "Ты — метеоролог-ассистент. Отвечай информативно, но кратко. "
        "Всегда указывай реалистичные данные для указанного города и даты."
    ),
    result_type=WeatherReport,
)

# Синхронный вызов
result = weather_agent.run_sync("Какой прогноз погоды в Токио на завтра?")

# result.data — это уже экземпляр WeatherReport
report: WeatherReport = result.data
print(f"Город: {report.city}")
print(f"Температура: {report.temperature_c}°C")
print(f"Условия: {report.condition}")
print(f"Влажность: {report.humidity_pct}%")
print(f"Дата: {report.forecast_date}")

Ключевой момент: result_type=WeatherReport. Передавая Pydantic-модель в этот параметр, вы говорите фреймворку: «Я ожидаю на выходе объект именно этой структуры». Pydantic AI автоматически добавляет в системный промпт инструкцию для модели о том, какой JSON нужно вернуть, парсит ответ и валидирует его. Если модель вернёт температуру строкой "двадцать пять" вместо числа 25.0 — валидация упадёт с понятным сообщением об ошибке, и вы сможете либо запросить у модели исправление (retry), либо обработать исключение.

Обратите внимание на Field(description=...) — эти описания попадают в промпт модели, помогая ей понять, что именно ожидается в каждом поле. Чем точнее описания, тем выше качество структурированного вывода.

Инструменты — это то, что превращает языковую модель из «говорящей головы» в полезного агента, способного взаимодействовать с внешним миром. В Pydantic AI инструменты — это обычные Python-функции, декорированные через @agent.tool (метод агента) или переданные в конструктор через параметр tools=[].

Каждый инструмент имеет чёткую сигнатуру с типами аргументов и возвращаемого значения. Эти типы автоматически конвертируются в JSON Schema и передаются языковой модели как описание function calling. Модель решает, когда и с какими аргументами вызывать инструмент, а Pydantic AI берёт на себя выполнение и передачу результата обратно модели для формирования финального ответа.

Важное преимущество перед нативным function calling OpenAI: здесь аргументы функции проходят Pydantic-валидацию ещё до вызова. Если модель передаст некорректные данные, вы получите исключение на уровне фреймворка, а не сломанную бизнес-логику.

import httpx
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from pydantic import BaseModel


class ProductInfo(BaseModel):
    name: str
    price_rub: float
    in_stock: bool
    rating: float


model = OpenAIModel("gpt-4o")
shop_agent = Agent(
    model=model,
    system_prompt="Ты — консультант интернет-магазина. Используй инструменты для получения данных.",
    result_type=ProductInfo,
)


# Инструмент 1: поиск в «базе данных» (заглушка)
@shop_agent.tool
def search_product(query: str, max_price: float | None = None) -> dict:
    """Ищет товар в каталоге по названию. Можно ограничить максимальную цену."""
    # В реальном проекте — запрос к БД или поисковому индексу
    catalog = {
        "наушники": {"name": "Sony WH-1000XM5", "price_rub": 34990, "in_stock": True, "rating": 4.8},
        "клавиатура": {"name": "Keychron K8 Pro", "price_rub": 12990, "in_stock": True, "rating": 4.6},
        "монитор": {"name": "Dell U2723QE", "price_rub": 54990, "in_stock": False, "rating": 4.7},
    }
    for key, product in catalog.items():
        if query.lower() in key.lower():
            if max_price is not None and product["price_rub"] > max_price:
                return {"error": "Товар не укладывается в бюджет"}
            return product
    return {"error": "Товар не найден"}


# Инструмент 2: запрос к внешнему API курса валют
@shop_agent.tool
async def get_usd_rate() -> float:
    """Возвращает актуальный курс доллара к рублю."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.exchangerate-api.com/v4/latest/USD",
            timeout=10.0
        )
        data = response.json()
        return data["rates"]["RUB"]


# Запрос к агенту — он сам решит, какие инструменты вызвать
result = shop_agent.run_sync(
    "Найди беспроводные наушники до 40000 рублей и скажи их цену в долларах."
)
print(result.data)
# → ProductInfo(name='Sony WH-1000XM5', price_rub=34990, in_stock=True, rating=4.8)
print(f"Цена в USD: ${result.data.price_rub / result.data.usd_rate:.0f}")

Обратите внимание: инструменты могут быть как синхронными, так и асинхронными. Pydantic AI корректно обрабатывает оба случая. Фреймворк сам строит JSON Schema для каждого инструмента на основе сигнатуры функции и docstring — вам не нужно вручную описывать function calling схемы. Модель получает полную информацию о том, что делает инструмент, какие у него параметры и что он возвращает, и на основе этого принимает решение о вызове.

Ещё одна мощная возможность — валидация аргументов. Если модель передаст строку вместо float для параметра max_price, фреймворк перехватит это и либо попытается повторить вызов, либо вернёт ошибку агенту для самоисправления.

Структурированный вывод — это главная «фишка» Pydantic AI, которая отличает его от всех остальных фреймворков для LLM. Вместо того чтобы просить модель «верни JSON» и потом парсить его вручную, вы описываете Pydantic-модель — и фреймворк гарантирует, что на выходе будет валидный экземпляр этой модели.

Механизм работает так: на этапе подготовки запроса Pydantic AI генерирует JSON Schema из вашей модели, добавляет её в системный промпт как ожидаемый формат ответа, а после получения ответа от LLM парсит JSON и пропускает через валидатор Pydantic. Если данные не соответствуют схеме, выбрасывается исключение с детальным описанием ошибки — какие поля отсутствуют, какие имеют неверный тип, какие значения нарушают ограничения.

Для сложных случаев можно использовать вложенные модели, списки, Literal-типы, Optional поля и любые другие конструкции из арсенала Pydantic — включая кастомные валидаторы.

from typing import Literal
from pydantic import BaseModel, Field, field_validator
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel


class Skill(BaseModel):
    """Отдельный навык кандидата."""
    name: str = Field(description="Название навыка")
    level: Literal["junior", "middle", "senior", "expert"] = Field(
        description="Уровень владения"
    )
    years: float = Field(ge=0, le=50, description="Лет опыта с этим навыком")


class Contact(BaseModel):
    email: str | None = Field(default=None, description="Email (если указан в резюме)")
    phone: str | None = Field(default=None, description="Телефон (если указан в резюме)")


class Candidate(BaseModel):
    """Полная модель данных кандидата."""
    full_name: str = Field(description="ФИО кандидата")
    position: str = Field(description="Желаемая должность")
    total_experience_years: float = Field(ge=0, le=50, description="Общий опыт работы в годах")
    skills: list[Skill] = Field(description="Список ключевых навыков (минимум 1)", min_length=1)
    contact: Contact = Field(description="Контактные данные")
    salary_expectation_rub: int | None = Field(
        default=None, ge=0, description="Ожидаемая зарплата в рублях"
    )
    relocation_ready: bool = Field(description="Готовность к переезду")

    @field_validator("full_name")
    @classmethod
    def name_must_have_spaces(cls, v: str) -> str:
        if " " not in v:
            raise ValueError("Имя должно содержать хотя бы фамилию и имя")
        return v


model = OpenAIModel("gpt-4o")
hr_agent = Agent(
    model=model,
    system_prompt=(
        "Ты — HR-ассистент, анализирующий резюме. Извлекай все указанные данные. "
        "Если какого-то поля нет в резюме, используй null/None."
    ),
    result_type=Candidate,
)

resume_text = """
Иван Сергеевич Петров, 32 года. Python-разработчик с опытом 7 лет.
Ключевые навыки: Python (senior, 6 лет), PostgreSQL (middle, 4 года),
Docker (middle, 3 года), FastAPI (middle, 2 года).
Email: ivan@example.com, телефон: +7-999-123-45-67.
Ожидаемая зарплата: 350000 рублей. Готов к переезду.
"""

result = hr_agent.run_sync(resume_text)
candidate: Candidate = result.data

print(f"Имя: {candidate.full_name}")
print(f"Должность: {candidate.position}")
print(f"Опыт: {candidate.total_experience_years} лет")
for skill in candidate.skills:
    print(f"  • {skill.name} — {skill.level} ({skill.years} лет)")
print(f"Email: {candidate.contact.email}")
print(f"Зарплата: {candidate.salary_expectation_rub} ₽")
print(f"Переезд: {'да' if candidate.relocation_ready else 'нет'}")

Здесь мы используем сразу несколько продвинутых приёмов: вложенные Pydantic-модели (Skill и Contact внутри Candidate), Literal-типы для ограничения значений (level: Literal["junior", "middle", ...]), числовые ограничения через ge и le, кастомные валидаторы через @field_validator. Все эти проверки выполняются автоматически — вам не нужно писать ни строчки кода для валидации ответа модели.

В продакшен-сценариях редко можно позволить себе ждать полного ответа модели — особенно если она генерирует большой объём данных или вызывает несколько инструментов. Pydantic AI предлагает два режима асинхронной работы: потоковый вывод (streaming) для текстовых ответов и асинхронные вызовы для параллельного выполнения нескольких агентов.

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

import asyncio
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel


model = OpenAIModel("gpt-4o")
storyteller = Agent(
    model=model,
    system_prompt="Ты — писатель-фантаст. Пиши увлекательно, но не затягивай.",
)


async def stream_example():
    """Потоковый вывод: печатаем каждое слово по мере генерации."""
    print("🤖 Генерация истории...\n")

    async with storyteller.run_stream(
        "Напиши короткий рассказ про робота, который научился мечтать."
    ) as stream:
        async for chunk in stream:
            # chunk — это дельта (новый кусочек текста)
            print(chunk, end="", flush=True)

    print("\n\n✅ История завершена!")


# Запуск
asyncio.run(stream_example())

import asyncio
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from pydantic import BaseModel


class AnalysisResult(BaseModel):
    topic: str
    summary: str
    sentiment: str
    key_points: list[str]


model = OpenAIModel("gpt-4o")

# Три агента для параллельного анализа текста
summarizer = Agent(model=model, system_prompt="Сделай краткую выжимку текста.", result_type=AnalysisResult)
sentiment_agent = Agent(model=model, system_prompt="Определи тональность текста.", result_type=AnalysisResult)
keywords_agent = Agent(model=model, system_prompt="Выдели ключевые тезисы.", result_type=AnalysisResult)


async def parallel_analysis(text: str):
    """Запускаем трёх агентов одновременно и ждём все результаты."""
    tasks = [
        summarizer.run(text),
        sentiment_agent.run(text),
        keywords_agent.run(text),
    ]

    # asyncio.gather — параллельный запуск
    results = await asyncio.gather(*tasks)

    for i, result in enumerate(results, 1):
        data: AnalysisResult = result.data
        print(f"\n--- Агент {i}: {data.topic} ---")
        print(f"Тональность: {data.sentiment}")
        print(f"Резюме: {data.summary[:100]}...")
        print(f"Тезисы: {', '.join(data.key_points[:3])}")

    return results


# Запуск
sample_text = """
Pydantic AI — это прорывной фреймворк для работы с LLM, который решает
проблему неструктурированного вывода. Разработчики получают типизированные
ответы, встроенную валидацию через Pydantic и удобный API для создания
сложных AI-агентов. Фреймворк поддерживает все популярные LLM-провайдеры
и идеально подходит для продакшен-систем.
"""

asyncio.run(parallel_analysis(sample_text))

Ключевое преимущество асинхронного подхода: три вызова к LLM выполняются одновременно, а не последовательно. Если каждый вызов занимает ~2 секунды, параллельное выполнение завершится за те же 2 секунды, тогда как последовательное заняло бы 6. На практике это даёт колоссальный выигрыш в скорости для продакшен-систем, обрабатывающих множество запросов.

Обратите внимание на метод run_stream() — он возвращает асинхронный контекстный менеджер, внутри которого можно итерироваться по чанкам. Это нативный Python-паттерн, никаких коллбэков или сложных абстракций.

Настоящая сила Pydantic AI раскрывается при построении мультиагентных систем — архитектур, где несколько специализированных агентов работают сообща, передавая данные друг другу. Это паттерн «конвейер» (pipeline) или «оркестратор-исполнители» (orchestrator-workers), знакомый по LangChain и AutoGen, но реализованный с жёсткой типизацией на каждом стыке.

В Pydantic AI агенты могут зависеть друг от друга через механизм dependency injection. Один агент может использовать результат работы другого как контекст, а промежуточные Pydantic-модели гарантируют, что данные не «испортятся» при передаче. Это позволяет строить сложные конвейеры обработки: агент-классификатор определяет тип запроса → агент-маршрутизатор выбирает исполнителя → профильный агент решает задачу → агент-валидатор проверяет результат.

import asyncio
from typing import Literal
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel


# ── Модели данных для меж-агентного взаимодействия ──

class TicketClassification(BaseModel):
    """Классификация входящего тикета поддержки."""
    category: Literal["billing", "technical", "account", "general"]
    priority: Literal["low", "medium", "high", "critical"]
    language: Literal["ru", "en"]
    summary: str = Field(description="Краткое описание проблемы (1 предложение)")


class TicketSolution(BaseModel):
    """Предложенное решение тикета."""
    action_required: str = Field(description="Что нужно сделать")
    steps: list[str] = Field(description="Пошаговая инструкция")
    estimated_time_minutes: int = Field(ge=1, le=480)
    needs_human: bool = Field(description="Требуется ли участие человека")


class FinalResponse(BaseModel):
    """Финальный ответ пользователю."""
    greeting: str
    solution_text: str
    follow_up_question: str | None = None


# ── Агенты пайплайна ──

model = OpenAIModel("gpt-4o")

# Агент 1: Классификатор
classifier = Agent(
    model=model,
    system_prompt=(
        "Ты — классификатор обращений в техподдержку. "
        "Определяй категорию, приоритет и язык обращения."
    ),
    result_type=TicketClassification,
)

# Агент 2: Решатель (специализируется на категории)
solver = Agent(
    model=model,
    system_prompt=(
        "Ты — опытный специалист техподдержки. На основе классификации тикета "
        "предложи конкретное решение. Будь точен и практичен."
    ),
    result_type=TicketSolution,
)

# Агент 3: Формирователь ответа клиенту
responder = Agent(
    model=model,
    system_prompt=(
        "Ты — вежливый сотрудник поддержки. Сформулируй ответ клиенту "
        "на основе готового решения. Пиши на языке клиента."
    ),
    result_type=FinalResponse,
)


async def process_ticket(customer_message: str) -> FinalResponse:
    """Полный пайплайн обработки тикета: классификация → решение → ответ."""

    # Шаг 1: Классифицируем обращение
    print("🔍 Классификация тикета...")
    classification_result = await classifier.run(customer_message)
    classification: TicketClassification = classification_result.data
    print(f"   Категория: {classification.category}, приоритет: {classification.priority}")

    # Шаг 2: Находим решение, передавая контекст классификации
    print("🛠️ Поиск решения...")
    solver_prompt = (
        f"Категория: {classification.category}\n"
        f"Приоритет: {classification.priority}\n"
        f"Проблема: {classification.summary}\n"
        f"Исходное обращение: {customer_message}"
    )
    solution_result = await solver.run(solver_prompt)
    solution: TicketSolution = solution_result.data
    print(f"   Действие: {solution.action_required}")
    print(f"   Нужен человек: {solution.needs_human}")

    # Шаг 3: Формируем финальный ответ клиенту
    print("✉️ Формирование ответа...")
    response_prompt = (
        f"Обращение клиента: {customer_message}\n"
        f"Найденное решение: {solution.action_required}\n"
        f"Шаги: {', '.join(solution.steps)}\n"
        f"Язык ответа: {classification.language}"
    )
    final_result = await responder.run(response_prompt)
    final: FinalResponse = final_result.data

    return final


# ── Запуск пайплайна ──

async def main():
    ticket = "Не могу войти в личный кабинет уже третий день! Пароль сбрасывал, не помогает."
    response = await process_ticket(ticket)

    print("\n📬 Финальный ответ клиенту:")
    print(f"   {response.greeting}")
    print(f"   {response.solution_text}")
    if response.follow_up_question:
        print(f"   {response.follow_up_question}")

asyncio.run(main())

В этом примере мы построили полноценный трёхступенчатый пайплайн обработки обращений в техподдержку. Каждый агент специализирован и получает на вход строго типизированные данные от предыдущего этапа. Обратите внимание: мы не передаём сырой текст между агентами — мы передаём Pydantic-объекты. Если классификатор вернёт некорректную категорию, валидация упадёт ещё до того, как данные попадут в следующий агент.

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

📋 Итог: что мы изучили