Pydantic AI & Instructor Basics

python -m venv .venv && source .venv/bin/activate
pip install "pydantic-ai" instructor
export OPENAI_API_KEY="sk-..."   # or provider-specific vars; verify at build

from pydantic_ai import Agent
 
agent = Agent(
    "openai:gpt-4o-mini",  # verify model slug at build
    instructions="Be concise. Reply in one sentence.",
)
 
result = agent.run_sync('Where does "hello world" come from?')
print(result.output)  # str
print(result.usage)  # token / request usage; verify attribute at build

from pydantic import BaseModel, Field
from pydantic_ai import Agent
 
 
class CityLocation(BaseModel):
    city: str
    country: str
    year: int = Field(description="Year of the event asked about")
 
 
agent = Agent(
    "openai:gpt-4o-mini",  # verify at build
    output_type=CityLocation,
    instructions="Extract the city, country, and year from the question or known facts.",
)
 
result = agent.run_sync("Where were the Olympics held in 2012?")
print(result.output)
# CityLocation(city='London', country='United Kingdom', year=2012)
print(result.output.city)  # typed attribute access

import instructor
from pydantic import BaseModel
 
 
class Person(BaseModel):
    name: str
    age: int
    occupation: str
 
 
client = instructor.from_provider("openai/gpt-4o-mini")  # verify slug at build
 
person = client.create(
    response_model=Person,
    messages=[
        {
            "role": "user",
            "content": "Extract: Ada is a 36-year-old data engineer.",
        }
    ],
)
print(person)
# Person(name='Ada', age=36, occupation='data engineer')

from enum import Enum
from pydantic import BaseModel, Field
from pydantic_ai import Agent
 
 
class Priority(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"
 
 
class TicketDraft(BaseModel):
    title: str = Field(description="Short issue title, max ~80 chars")
    priority: Priority
    summary: str = Field(description="One paragraph for the human agent")
 
 
agent = Agent(
    "openai:gpt-4o-mini",
    output_type=TicketDraft,
    instructions="Turn user complaints into support tickets.",
)
 
result = agent.run_sync("My card was charged twice and I need this fixed today.")
print(result.output.model_dump())

from pydantic import BaseModel
from pydantic_ai import Agent
 
 
class WeatherReport(BaseModel):
    city: str
    temp_c: float
    advice: str
 
 
agent = Agent(
    "openai:gpt-4o-mini",
    output_type=WeatherReport,
    instructions="Use the weather tool, then return a WeatherReport.",
)
 
 
@agent.tool_plain
def get_temp_c(city: str) -> float:
    """Return a demo temperature in Celsius for a city."""
    # Replace with a real weather API in production.
    return {"paris": 18.0, "tokyo": 24.5}.get(city.lower(), 20.0)
 
 
result = agent.run_sync("What is the weather like in Paris for a walk?")
print(result.output)

from dataclasses import dataclass
 
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
 
 
@dataclass
class SupportDeps:
    customer_id: int
 
 
class SupportOutput(BaseModel):
    advice: str
    risk: int = Field(ge=0, le=10)
 
 
agent = Agent(
    "openai:gpt-4o-mini",
    deps_type=SupportDeps,
    output_type=SupportOutput,
    instructions="Give brief support advice and a risk score 0-10.",
)
 
 
@agent.tool
async def get_customer_id(ctx: RunContext[SupportDeps]) -> int:
    """Return the current customer id from dependencies."""
    return ctx.deps.customer_id
 
 
result = agent.run_sync(
    "I lost my card and saw a strange charge.",
    deps=SupportDeps(customer_id=42),
)
print(result.output)

import instructor
from pydantic import BaseModel, field_validator
 
 
class UserAge(BaseModel):
    name: str
    age: int
 
    @field_validator("age")
    @classmethod
    def age_must_be_positive(cls, v: int) -> int:
        if v < 0:
            raise ValueError("age must be >= 0")
        return v
 
 
client = instructor.from_provider("openai/gpt-4o-mini")
 
user = client.create(
    response_model=UserAge,
    max_retries=3,
    messages=[
        {
            "role": "user",
            "content": "Extract name and age. Text: Riley is -3 years old.",
        }
    ],
)
print(user)  # age corrected after retry feedback when the model cooperates

# extraction.py - pure structured extract, no tools
import instructor
from pydantic import BaseModel
 
 
class InvoiceLine(BaseModel):
    description: str
    amount: float
 
 
class Invoice(BaseModel):
    vendor: str
    lines: list[InvoiceLine]
 
 
extract_client = instructor.from_provider("openai/gpt-4o-mini")
 
 
def extract_invoice(text: str) -> Invoice:
    return extract_client.create(
        response_model=Invoice,
        messages=[{"role": "user", "content": text}],
        max_retries=2,
    )
 
 
# agent.py - multi-step tool use + structured finish
from pydantic_ai import Agent
 
triage_agent = Agent(
    "openai:gpt-4o-mini",
    output_type=Invoice,
    instructions="If needed, call tools; finish with a validated Invoice.",
)