Pydantic AI: как собрать типобезопасного AI-агента на Python с MCP и Ollama

Stable v2 вышел 23 июня 2026 года. Текущий v2.46.0 уже относится к production/stable линии. Старые примеры v0.x/v1.x могут использовать другие имена аргументов, старый MCP API и прежнюю структуру integrations.

from pydantic import BaseModel
from pydantic_ai import Agent


class DeviceFault(BaseModel):
    device: str
    severity: str
    probable_cause: str
    needs_human: bool


agent = Agent(
    "openai:gpt-5.2",
    output_type=DeviceFault,
)

result = agent.run_sync(
    "Блок питания греется, напряжение проседает под нагрузкой."
)

fault = result.output

print(fault.device)
print(fault.needs_human)

Внутри framework строит JSON Schema и валидирует model output. Если модель не возвращает структуру нужного вида, результат не должен бесконтрольно пройти дальше как «почти JSON».

from dataclasses import dataclass

from pydantic_ai import Agent, RunContext


@dataclass
class Deps:
    user_id: int
    database: object


agent = Agent(
    "openai:gpt-5.2",
    deps_type=Deps,
)


@agent.tool
def get_user_name(ctx: RunContext[Deps]) -> str:
    """Read user data from application dependencies."""
    return f"user-{ctx.deps.user_id}"


deps = Deps(
    user_id=42,
    database=object(),
)

result = agent.run_sync(
    "Кто текущий пользователь?",
    deps=deps,
)

Такой подход полезен для production: agent code получает только те зависимости, которые приложение передало конкретному run, а type checker видит их тип.

Для локального Ollama используется OpenAI-compatible endpoint http://localhost:11434/v1.

# Linux/macOS
export OLLAMA_BASE_URL="http://localhost:11434/v1"

# Windows PowerShell
$env:OLLAMA_BASE_URL = "http://localhost:11434/v1"

from pydantic_ai import Agent

agent = Agent("ollama:qwen3")

result = agent.run_sync(
    "Объясни разницу между RAG и fine-tuning."
)

print(result.output)

Structured output с локальным Ollama

from pydantic import BaseModel

from pydantic_ai import Agent
from pydantic_ai.models.ollama import OllamaModel
from pydantic_ai.output import NativeOutput
from pydantic_ai.providers.ollama import OllamaProvider


class CheckResult(BaseModel):
    ok: bool
    reason: str


model = OllamaModel(
    "qwen3",
    provider=OllamaProvider(
        base_url="http://localhost:11434/v1"
    ),
)

agent = Agent(
    model,
    output_type=NativeOutput(CheckResult),
)

Self-hosted Ollama умеет schema-constrained structured output, поэтому Pydantic AI может использовать NativeOutput для локальной модели.

from pydantic_evals import Case, Dataset


dataset = Dataset(
    name="support-agent",
    cases=[
        Case(
            name="simple",
            inputs="Сеть пропала после перезагрузки роутера",
            expected_output="check_wan",
        ),
    ],
)


def my_task(inputs: str) -> str:
    # Здесь вызывается agent/application code.
    return "check_wan"


report = dataset.evaluate_sync(my_task)

report.print()

Pydantic AI поддерживает несколько durability engines: Temporal, DBOS, Prefect, Restate, AWS Lambda, а также внешние integrations с Kitaru и Apache Airflow. Есть builder для подключения других engines.

DBOS: простой пример

from dbos import DBOS, DBOSConfig

from pydantic_ai import Agent
from pydantic_ai.durable_exec.dbos import DBOSDurability


DBOS(
    config={
        "name": "agent-app",
        "system_database_url": "sqlite:///dbos.sqlite",
    }
)

agent = Agent(
    "openai:gpt-5.2",
    name="support-agent",
    capabilities=[DBOSDurability()],
)


@DBOS.workflow()
async def answer(question: str) -> str:
    result = await agent.run(question)
    return result.output

Model requests и MCP communication становятся durable steps. После restart workflow может продолжить работу с checkpoint, а не повторять всё с начала.