ai_ml / agents orchestration / 12_pydantic_ai.md

What is Pydantic AI?

4 interview angles 12 min read source

What is Pydantic AI?

Definition

Pydantic AI is a framework that combines Pydantic (data validation) with AI/LLM capabilities to create type-safe, structured AI applications. It provides a way to build AI agents and workflows with strong typing, validation, and structured outputs using Pydantic models.

Key Concepts

Core Principles

  1. Type Safety: Uses Pydantic models for type validation
  2. Structured Outputs: Guarantees structured, validated responses
  3. Agent Framework: Build AI agents with type safety
  4. Tool Integration: Type-safe tool definitions
  5. Validation: Automatic validation of AI outputs
  6. Pythonic: Leverages Python’s type system

Why Pydantic AI?

Problems it Solves:

  • Unstructured AI outputs (JSON strings, free text)
  • Type errors at runtime
  • Manual validation of AI responses
  • Lack of guarantees about output structure

Benefits:

  • Type-safe AI interactions
  • Automatic validation
  • Better IDE support
  • Runtime type checking
  • Structured, predictable outputs

Basic Usage

Simple Example

from pydantic_ai import Agent
from pydantic import BaseModel

# Define response model
class Answer(BaseModel):
    answer: str
    confidence: float
    sources: list[str]

# Create agent
agent = Agent(
    'openai:gpt-4',
    system_prompt='You are a helpful assistant.'
)

# Run with structured output
result = agent.run_sync(
    'What is machine learning?',
    response_model=Answer
)

print(result.data.answer)
print(result.data.confidence)
print(result.data.sources)

Agent with Tools

from pydantic_ai import Agent, Tool
from pydantic import BaseModel

# Define tool
def calculate(expression: str) -> float:
    """Calculate a mathematical expression."""
    return eval(expression)

# Create tool
calculator_tool = Tool(
    calculate,
    name='calculator',
    description='Performs mathematical calculations'
)

# Create agent with tools
agent = Agent(
    'openai:gpt-4',
    system_prompt='You are a math assistant.',
    tools=[calculator_tool]
)

# Use agent
result = agent.run_sync('What is 25 * 4 + 10?')
print(result.data)

Core Features

1. Structured Outputs

Without Pydantic AI:

# Unstructured output
response = llm.generate("What is Python?")
# Returns: string, need to parse manually

With Pydantic AI:

from pydantic import BaseModel

class PythonInfo(BaseModel):
    definition: str
    features: list[str]
    use_cases: list[str]

result = agent.run_sync(
    "What is Python?",
    response_model=PythonInfo
)

# Type-safe access
print(result.data.definition)  # str
print(result.data.features)    # list[str]
print(result.data.use_cases)   # list[str]

2. Type-Safe Tools

from pydantic_ai import Agent, Tool
from pydantic import BaseModel

# Define tool with types
def get_weather(city: str, unit: str = "celsius") -> dict:
    """Get weather for a city.
    
    Args:
        city: Name of the city
        unit: Temperature unit (celsius or fahrenheit)
    
    Returns:
        Dictionary with weather information
    """
    # Implementation
    return {"temperature": 22, "condition": "sunny"}

# Create typed tool
weather_tool = Tool(
    get_weather,
    name="get_weather",
    description="Get current weather for a city"
)

# Agent automatically understands types
agent = Agent(
    'openai:gpt-4',
    tools=[weather_tool]
)

result = agent.run_sync("What's the weather in Paris?")

3. Response Models

from pydantic import BaseModel, Field
from typing import Literal

class AnalysisResult(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"]
    score: float = Field(ge=0.0, le=1.0)
    keywords: list[str] = Field(min_length=1)
    summary: str = Field(min_length=10)

agent = Agent('openai:gpt-4')

result = agent.run_sync(
    "Analyze: 'I love this product!'",
    response_model=AnalysisResult
)

# Validated and typed
print(result.data.sentiment)  # "positive" | "negative" | "neutral"
print(result.data.score)      # 0.0 to 1.0
print(result.data.keywords)  # list[str]

4. Agent State Management

from pydantic_ai import Agent
from pydantic import BaseModel

class ConversationState(BaseModel):
    messages: list[str] = []
    topic: str = ""

agent = Agent('openai:gpt-4')

# Maintain state across interactions
state = ConversationState()

result1 = agent.run_sync(
    "Tell me about Python",
    state=state
)
state.messages.append(result1.data)
state.topic = "Python"

result2 = agent.run_sync(
    "What are its main features?",
    state=state
)

5. Async Support

import asyncio
from pydantic_ai import Agent

agent = Agent('openai:gpt-4')

async def main():
    # Async execution
    result = await agent.run(
        "What is async programming?",
        response_model=Answer
    )
    print(result.data)

# Run
asyncio.run(main())

Advanced Features

1. Custom Validators

from pydantic import BaseModel, field_validator
from pydantic_ai import Agent

class EmailResponse(BaseModel):
    email: str
    subject: str
    body: str
    
    @field_validator('email')
    @classmethod
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('Invalid email format')
        return v
    
    @field_validator('body')
    @classmethod
    def validate_body(cls, v):
        if len(v) < 10:
            raise ValueError('Body too short')
        return v

agent = Agent('openai:gpt-4')

# Validation happens automatically
result = agent.run_sync(
    "Generate an email to john@example.com",
    response_model=EmailResponse
)

2. Nested Models

from pydantic import BaseModel
from typing import List

class Author(BaseModel):
    name: str
    email: str

class Article(BaseModel):
    title: str
    content: str
    authors: List[Author]
    tags: List[str]

agent = Agent('openai:gpt-4')

result = agent.run_sync(
    "Write an article about AI",
    response_model=Article
)

# Type-safe nested access
print(result.data.authors[0].name)
print(result.data.authors[0].email)

3. Tool Result Validation

from pydantic_ai import Agent, Tool
from pydantic import BaseModel

class CalculationResult(BaseModel):
    expression: str
    result: float
    steps: list[str]

def calculate_with_steps(expression: str) -> CalculationResult:
    """Calculate and show steps."""
    # Implementation
    return CalculationResult(
        expression=expression,
        result=42.0,
        steps=["Step 1", "Step 2"]
    )

tool = Tool(
    calculate_with_steps,
    result_model=CalculationResult
)

agent = Agent('openai:gpt-4', tools=[tool])

4. Error Handling

from pydantic_ai import Agent
from pydantic import ValidationError

agent = Agent('openai:gpt-4')

try:
    result = agent.run_sync(
        "Generate data",
        response_model=StrictModel
    )
except ValidationError as e:
    print("Validation failed:", e)
    # Handle invalid response
except Exception as e:
    print("Error:", e)

5. Streaming Responses

from pydantic_ai import Agent

agent = Agent('openai:gpt-4')

# Stream tokens
for token in agent.run_stream("Tell a story"):
    print(token, end='', flush=True)

Comparison with Other Frameworks

Pydantic AI vs LangChain

Feature Pydantic AI LangChain
Type Safety Built-in with Pydantic Manual validation
Structured Outputs Automatic Manual parsing
Validation Automatic Manual
Python Types Full support Limited
Complexity Simpler for typed apps More features
Focus Type-safe AI General AI framework

Pydantic AI vs Direct LLM Calls

Direct LLM:

response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "What is Python?"}]
)
# Returns JSON string, need to parse
data = json.loads(response.choices[0].message.content)

Pydantic AI:

class PythonInfo(BaseModel):
    definition: str
    features: list[str]

result = agent.run_sync(
    "What is Python?",
    response_model=PythonInfo
)
# Returns validated Pydantic model
print(result.data.definition)  # Type-safe

Real-World Examples

Example 1: Data Extraction Agent

from pydantic import BaseModel, Field
from pydantic_ai import Agent
from typing import List

class Person(BaseModel):
    name: str
    age: int = Field(ge=0, le=150)
    email: str
    skills: List[str]

class TeamData(BaseModel):
    team_name: str
    members: List[Person]
    total_members: int

agent = Agent(
    'openai:gpt-4',
    system_prompt='Extract structured data from text.'
)

text = """
Our team consists of:
- John Doe, 30, john@example.com, knows Python and JavaScript
- Jane Smith, 28, jane@example.com, knows React and Node.js
"""

result = agent.run_sync(
    f"Extract team information: {text}",
    response_model=TeamData
)

# Type-safe access
print(f"Team: {result.data.team_name}")
for member in result.data.members:
    print(f"{member.name} ({member.age}): {member.email}")

Example 2: Code Analysis Agent

from pydantic import BaseModel
from pydantic_ai import Agent
from typing import List, Literal

class CodeIssue(BaseModel):
    line: int
    severity: Literal["error", "warning", "info"]
    message: str
    suggestion: str

class CodeAnalysis(BaseModel):
    language: str
    issues: List[CodeIssue]
    score: float
    recommendations: List[str]

agent = Agent(
    'openai:gpt-4',
    system_prompt='Analyze code and provide structured feedback.'
)

code = """
def calculate(x, y):
    return x + y
"""

result = agent.run_sync(
    f"Analyze this code: {code}",
    response_model=CodeAnalysis
)

# Type-safe analysis results
for issue in result.data.issues:
    print(f"Line {issue.line}: {issue.severity} - {issue.message}")

Example 3: Multi-Step Agent

from pydantic import BaseModel
from pydantic_ai import Agent, Tool
from typing import List

class ResearchResult(BaseModel):
    topic: str
    summary: str
    sources: List[str]
    key_points: List[str]

def search_web(query: str) -> str:
    """Search the web for information."""
    # Implementation
    return f"Search results for: {query}"

def summarize_text(text: str) -> str:
    """Summarize long text."""
    # Implementation
    return f"Summary: {text[:100]}..."

# Create tools
search_tool = Tool(search_web, name="search")
summarize_tool = Tool(summarize_text, name="summarize")

# Create agent
agent = Agent(
    'openai:gpt-4',
    tools=[search_tool, summarize_tool],
    system_prompt='Research topics and provide structured summaries.'
)

result = agent.run_sync(
    "Research machine learning and provide a summary",
    response_model=ResearchResult
)

print(result.data.summary)
print(f"Sources: {result.data.sources}")

Example 4: API Response Agent

from pydantic import BaseModel, HttpUrl
from pydantic_ai import Agent
from typing import List

class APIEndpoint(BaseModel):
    url: HttpUrl
    method: str
    description: str
    parameters: List[str]

class APIDocumentation(BaseModel):
    api_name: str
    base_url: HttpUrl
    endpoints: List[APIEndpoint]
    authentication: str

agent = Agent('openai:gpt-4')

result = agent.run_sync(
    "Document this API: GET /users, POST /users, GET /posts",
    response_model=APIDocumentation
)

# Validated URLs and structured data
print(f"API: {result.data.api_name}")
for endpoint in result.data.endpoints:
    print(f"{endpoint.method} {endpoint.url}")

Common Interview Questions and Answers

Q1: What is Pydantic AI and why use it?

Pydantic AI is a framework that combines Pydantic’s data validation with AI capabilities to create type-safe, structured AI applications.

Why use it:

  1. Type Safety: Guarantees structured, validated outputs
  2. Automatic Validation: No manual parsing or validation
  3. Better IDE Support: Full type hints and autocomplete
  4. Runtime Safety: Catches errors at runtime
  5. Structured Outputs: Guaranteed structure, not free-form text

Example:

# Without Pydantic AI: Manual parsing
response = llm.generate("...")
data = json.loads(response)  # May fail, no type safety

# With Pydantic AI: Type-safe
result = agent.run_sync("...", response_model=MyModel)
print(result.data.field)  # Type-safe, validated

Q2: How does Pydantic AI ensure type safety?

Through Pydantic models:

  1. Model Definition: Define expected structure
  2. Automatic Validation: Pydantic validates AI output
  3. Type Conversion: Converts to correct types
  4. Error Handling: Raises ValidationError if invalid
class Response(BaseModel):
    answer: str
    confidence: float  # Must be float

result = agent.run_sync("...", response_model=Response)
# result.data.confidence is guaranteed to be float
# If AI returns string, Pydantic converts or raises error

Q3: What’s the difference between Pydantic AI and LangChain?

Aspect Pydantic AI LangChain
Focus Type-safe AI General AI framework
Type Safety Built-in (Pydantic) Manual
Validation Automatic Manual
Complexity Simpler More features
Use Case Structured outputs Complex workflows
Learning Curve Lower Higher

Pydantic AI: Best for type-safe, structured AI applications LangChain: Best for complex workflows, agents, RAG

Q4: How do you handle errors in Pydantic AI?

Error handling:

from pydantic import ValidationError
from pydantic_ai import Agent

agent = Agent('openai:gpt-4')

try:
    result = agent.run_sync(
        "Generate data",
        response_model=StrictModel
    )
except ValidationError as e:
    # AI output didn't match model
    print("Validation errors:", e.errors())
    # Retry with different prompt or model
except Exception as e:
    # Other errors (API, network, etc.)
    print("Error:", e)

Strategies:

  1. Retry with different prompt: More explicit instructions
  2. Use more flexible model: Less strict validation
  3. Fallback model: Simpler response model
  4. Manual parsing: Parse and fix manually

Q5: How do you create tools in Pydantic AI?

Tool creation:

from pydantic_ai import Agent, Tool

# Simple function tool
def calculate(expression: str) -> float:
    """Calculate mathematical expression."""
    return eval(expression)

tool = Tool(
    calculate,
    name="calculator",
    description="Performs calculations"
)

# With type validation
class WeatherResult(BaseModel):
    temperature: float
    condition: str

def get_weather(city: str) -> WeatherResult:
    """Get weather for city."""
    # Implementation
    return WeatherResult(temperature=22.0, condition="sunny")

weather_tool = Tool(
    get_weather,
    result_model=WeatherResult  # Validate tool output
)

# Use in agent
agent = Agent('openai:gpt-4', tools=[tool, weather_tool])

Q6: Can you use Pydantic AI with async?

Yes, full async support:

import asyncio
from pydantic_ai import Agent

agent = Agent('openai:gpt-4')

async def main():
    # Async execution
    result = await agent.run(
        "What is async programming?",
        response_model=Answer
    )
    print(result.data)

# Run multiple in parallel
async def run_multiple():
    tasks = [
        agent.run("Question 1", response_model=Model1),
        agent.run("Question 2", response_model=Model2),
        agent.run("Question 3", response_model=Model3)
    ]
    results = await asyncio.gather(*tasks)
    return results

asyncio.run(main())

Q7: How do you validate nested structures?

Using nested Pydantic models:

from pydantic import BaseModel
from typing import List

class Author(BaseModel):
    name: str
    email: str

class Article(BaseModel):
    title: str
    authors: List[Author]  # Nested validation
    content: str

agent = Agent('openai:gpt-4')

result = agent.run_sync(
    "Write an article",
    response_model=Article
)

# All nested structures validated
for author in result.data.authors:
    print(author.name)  # Type-safe
    print(author.email)  # Validated email format

Q8: How do you customize validation in Pydantic AI?

Using Pydantic validators:

from pydantic import BaseModel, field_validator, model_validator

class CustomModel(BaseModel):
    email: str
    age: int
    score: float
    
    @field_validator('email')
    @classmethod
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('Invalid email')
        return v.lower()
    
    @field_validator('age')
    @classmethod
    def validate_age(cls, v):
        if v < 0 or v > 150:
            raise ValueError('Invalid age')
        return v
    
    @model_validator(mode='after')
    def validate_model(self):
        if self.age < 18 and self.score > 100:
            raise ValueError('Invalid combination')
        return self

agent = Agent('openai:gpt-4')
result = agent.run_sync("...", response_model=CustomModel)
# All validators run automatically

Q9: What are the limitations of Pydantic AI?

Limitations:

  1. LLM Dependency: Relies on LLM following structure
  2. Validation Failures: May need retries if validation fails
  3. Complex Workflows: Less suited for very complex workflows (use LangChain)
  4. Cost: LLM API calls can be expensive
  5. Latency: Network calls add latency

Mitigations:

  • Use clear prompts
  • Provide examples in prompts
  • Implement retry logic
  • Use appropriate models
  • Cache results when possible

Q10: How do you integrate Pydantic AI with existing code?

Integration strategies:

  1. Wrap Existing Functions:
def existing_function(data: dict) -> dict:
    # Existing code
    return result

# Wrap with Pydantic
class InputModel(BaseModel):
    field1: str
    field2: int

class OutputModel(BaseModel):
    result: str
    status: str

def wrapped_function(input_data: InputModel) -> OutputModel:
    result = existing_function(input_data.dict())
    return OutputModel(**result)
  1. Use as API Layer:
from fastapi import FastAPI
from pydantic_ai import Agent

app = FastAPI()
agent = Agent('openai:gpt-4')

@app.post("/analyze")
async def analyze(data: InputModel):
    result = await agent.run(
        f"Analyze: {data.text}",
        response_model=OutputModel
    )
    return result.data
  1. Replace Manual Parsing:
# Before: Manual parsing
response = api_call()
data = json.loads(response)
if 'field' in data:
    value = data['field']

# After: Type-safe
result = agent.run_sync("...", response_model=Model)
value = result.data.field  # Type-safe, validated

Best Practices

  1. Clear Models: Define clear, well-documented Pydantic models
  2. Validation: Use Pydantic validators for complex validation
  3. Error Handling: Implement robust error handling
  4. Prompts: Write clear prompts that guide structured output
  5. Testing: Test with various inputs and edge cases
  6. Type Hints: Use proper type hints throughout
  7. Documentation: Document models and their purpose
  8. Retry Logic: Implement retry for validation failures
  9. Caching: Cache results when appropriate
  10. Monitoring: Monitor validation failures and adjust

Summary

Pydantic AI:

  • Combines Pydantic validation with AI capabilities
  • Provides type-safe, structured AI interactions
  • Validates AI outputs automatically
  • Supports tools, async, and complex models
  • Enables building reliable AI applications

Key benefits:

  • Type safety
  • Automatic validation
  • Better IDE support
  • Runtime error detection
  • Structured outputs

Use cases:

  • Data extraction
  • Structured generation
  • Type-safe AI agents
  • API integration
  • Code analysis
  • Content generation

Pydantic AI is ideal when you need:

  • Guaranteed structure in AI outputs
  • Type safety in AI applications
  • Automatic validation
  • Better developer experience
  • Integration with existing type-safe code

Interview angle

  • “What is Pydantic AI and why choose it?” - a type-safe agent framework from the Pydantic team: agents declare a typed result and typed dependencies, so structured output and DI are first-class rather than bolted on. It’s a much smaller surface than LangGraph.
  • “Pydantic AI or LangGraph?” - Pydantic AI when you want type safety, a small dependency footprint and mostly single-agent behaviour. LangGraph when you need durable checkpointed state, human-in-the-loop interrupts, or non-trivial multi-agent topology.
  • “What does the typed result buy you?” - the agent’s output is a validated Pydantic model, so downstream code gets a real type instead of a string to parse, and mypy checks the boundary.
  • “How does dependency injection work there?” - a typed dependencies object is passed into the run and made available to tools, so a database session or HTTP client reaches tools without globals - and tests inject fakes at the same seam.