Шпаргалка по коду Pydantic AI

Все ключевые конструкции курса на одной странице. Удобно держать под рукой при чтении кода и постановке задач.

Установка

pip install pydantic-ai

Минимальный агент

from pydantic_ai import Agent

agent = Agent('anthropic:claude-sonnet-4-6', instructions='Отвечай кратко.')
result = agent.run_sync('Вопрос')
print(result.output)

Выбор модели

Agent('openai:gpt-5.2')               # строка «провайдер:модель»
Agent('google:gemini-3-flash-preview')
Agent('test')                          # TestModel: без API и затрат

Свой OpenAI-совместимый провайдер (MyRouter, OpenRouter, прокси)

from pydantic_ai.models.openai import OpenAIChatModel

model = OpenAIChatModel('myrouter/auto',
    base_url='https://api.gildiya-razvitiya.com/v1', api_key='myr_sk_...')
agent = Agent(model)

Резервные модели

from pydantic_ai.models.fallback import FallbackModel

agent = Agent(FallbackModel('openai:gpt-5.2', 'anthropic:claude-sonnet-4-5'))

Структурированный вывод

from pydantic import BaseModel, Field

class Lead(BaseModel):
    name: str
    phone: str
    score: int = Field(ge=1, le=10)   # с ограничением

class Failed(BaseModel):
    """Не удалось обработать."""

agent = Agent('openai:gpt-5.2', output_type=[Lead, Failed])  # «или-или»

Своя проверка результата

from pydantic_ai import ModelRetry

@agent.output_validator
async def check(ctx, output):
    if not is_ok(output):
        raise ModelRetry('Исправь: ...')   # модель попробует снова
    return output

Инструменты

@agent.tool_plain                          # без контекста
def roll_dice() -> str:
    """Бросить кубик."""
    ...

@agent.tool                                # с контекстом и deps
async def find_order(ctx: RunContext[MyDeps], order_id: int) -> str:
    """Найти заказ в базе."""
    return await ctx.deps.db.get(order_id)

Зависимости

from dataclasses import dataclass

@dataclass
class MyDeps:
    api_key: str
    db: Database

agent = Agent('openai:gpt-5.2', deps_type=MyDeps)
result = await agent.run('...', deps=MyDeps(key, db))

# подмена в тестах:
with agent.override(deps=test_deps):
    ...

Диалог с историей

r1 = agent.run_sync('Первый вопрос')
r2 = agent.run_sync('Уточнение', message_history=r1.new_messages())

Стриминг

async with agent.run_stream('Вопрос') as response:
    async for text in response.stream_text():
        print(text)

Лимиты расходов

from pydantic_ai import UsageLimits

result = agent.run_sync('...',
    usage_limits=UsageLimits(request_limit=5, total_tokens_limit=500))

Делегирование агенту

@main_agent.tool
async def sub_task(ctx: RunContext, n: int) -> list[str]:
    r = await helper_agent.run(f'Сделай {n} вариантов', usage=ctx.usage)
    return r.output

MCP и способности

from pydantic_ai.capabilities import MCP, WebSearch, Thinking

agent = Agent('openai:gpt-5.2',
    capabilities=[Thinking(), WebSearch(), MCP(url='https://...')])

Наблюдаемость (Logfire)

import logfire
logfire.configure()
logfire.instrument_pydantic_ai()