Pydantic AI Agent Framework: Building Intelligent AI Agents with Python

The Pydantic AI agent framework has emerged as a powerful solution for developers building intelligent AI agents with Python. By combining the robust validation capabilities of Pydantic with advanced agent orchestration features, this framework provides a type-safe, production-ready platform for creating AI applications that are reliable, maintainable, and scalable.

Whether you're building chatbots, autonomous agents, or complex multi-agent systems, the Pydantic AI agent framework offers the tools and patterns you need to develop sophisticated AI applications with confidence. This comprehensive guide explores the framework's capabilities, demonstrates practical implementations, and shares best practices for building production-grade AI agents.

What is the Pydantic AI Agent Framework?

The Pydantic AI agent framework is a Python library that streamlines the development of AI agents by providing structured data validation, type safety, and agent orchestration capabilities. Built on top of Pydantic, the popular data validation library, it extends these capabilities specifically for AI agent development.

At its core, the Pydantic AI agent framework solves several critical challenges in AI development. It ensures that data flowing between your application and language models is properly validated and typed, provides clear patterns for structuring agent logic, handles errors gracefully, and makes testing AI applications more straightforward and reliable.

The framework integrates seamlessly with major language model providers including OpenAI, Anthropic, Google, and others, while maintaining a consistent API that makes switching between providers straightforward. This flexibility is essential as the AI landscape evolves and new models become available.

Key Features of Pydantic AI Agent Framework

Understanding the core features helps developers leverage the full power of the Pydantic AI agent framework in their applications.

Type-Safe Agent Definitions

The Pydantic AI agent framework uses Python's type system to ensure correctness at development time. Agents are defined with clear input and output types, making it impossible to accidentally pass wrong data types or receive unexpected responses.

from pydantic_ai import Agent
from pydantic import BaseModel

class UserQuery(BaseModel):
    question: str
    context: dict[str, str] = {}

class AgentResponse(BaseModel):
    answer: str
    confidence: float
    sources: list[str] = []

# Define a type-safe agent
agent = Agent(
    'openai:gpt-4',
    result_type=AgentResponse,
    system_prompt='You are a helpful research assistant.'
)

Structured Output Validation

One of the most powerful features is automatic validation of language model outputs. The framework ensures that responses conform to your defined schemas, automatically retrying if the model produces invalid output.

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class ProductRecommendation(BaseModel):
    product_name: str = Field(description="Name of the recommended product")
    price: float = Field(gt=0, description="Product price in USD")
    reasoning: str = Field(min_length=20, description="Why this product is recommended")
    category: str

recommendation_agent = Agent(
    'anthropic:claude-sonnet-4',
    result_type=ProductRecommendation
)

# The framework ensures the output matches the schema
result = recommendation_agent.run_sync("Recommend a laptop for software development")
print(f"Recommended: {result.data.product_name} at ${result.data.price}")

Dependency Injection

The Pydantic AI agent framework supports dependency injection, making it easy to provide agents with access to databases, APIs, or other services without tight coupling.

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class DatabaseConnection:
    connection_string: str
    
    def query(self, sql: str) -> list[dict]:
        # Database query logic here
        pass

agent = Agent('openai:gpt-4', deps_type=DatabaseConnection)

@agent.tool
def search_database(ctx: RunContext[DatabaseConnection], query: str) -> list[dict]:
    """Search the database for relevant information."""
    return ctx.deps.query(f"SELECT * FROM data WHERE content LIKE '%{query}%'")

# Run with dependencies
db = DatabaseConnection("postgresql://localhost/mydb")
result = agent.run_sync("Find all customers from California", deps=db)

Tool and Function Calling

Agents can be equipped with tools that extend their capabilities beyond text generation. The Pydantic AI agent framework handles function calling automatically, including parameter validation and result processing.

from pydantic_ai import Agent, RunContext
import httpx

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

@agent.tool
def get_weather(ctx: RunContext[None], city: str) -> dict:
    """Get current weather for a city."""
    response = httpx.get(f"https://api.weather.com/v1/weather?city={city}")
    return response.json()

@agent.tool
def calculate_distance(ctx: RunContext[None], city1: str, city2: str) -> float:
    """Calculate distance between two cities in miles."""
    # Distance calculation logic
    return 523.7

# Agent automatically decides which tools to use
result = agent.run_sync("What's the weather like in San Francisco and how far is it from Los Angeles?")

Building Your First Agent with Pydantic AI

Let's walk through creating a practical AI agent using the Pydantic AI agent framework. This example demonstrates a customer support agent with structured responses and tool integration.

Installation and Setup

First, install the Pydantic AI agent framework along with your preferred model provider:

# Install Pydantic AI with OpenAI support
pip install pydantic-ai[openai]

# Or with Anthropic support
pip install pydantic-ai[anthropic]

# Or install all providers
pip install pydantic-ai[all]

Complete Customer Support Agent Example

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from dataclasses import dataclass
from datetime import datetime

# Define the response structure
class SupportResponse(BaseModel):
    message: str = Field(description="Response message to the customer")
    category: str = Field(description="Support category: technical, billing, general")
    priority: int = Field(ge=1, le=5, description="Priority level 1-5")
    requires_escalation: bool = Field(description="Whether to escalate to human agent")
    suggested_articles: list[str] = Field(default=[], description="Knowledge base articles")

# Define dependencies
@dataclass
class SupportDeps:
    customer_id: str
    ticket_history: list[dict]
    
    def get_customer_info(self) -> dict:
        """Fetch customer information from database."""
        return {
            "name": "John Doe",
            "tier": "premium",
            "account_age_days": 730
        }
    
    def search_knowledge_base(self, query: str) -> list[str]:
        """Search internal knowledge base."""
        # Simulate KB search
        return [
            "How to reset your password",
            "Billing cycle explained",
            "API rate limits"
        ]

# Create the agent
support_agent = Agent(
    'anthropic:claude-sonnet-4',
    result_type=SupportResponse,
    deps_type=SupportDeps,
    system_prompt="""You are an expert customer support agent. 
    Analyze customer inquiries and provide helpful, professional responses.
    Categorize issues appropriately and escalate when necessary.
    Always check the knowledge base for relevant articles."""
)

# Add tools
@support_agent.tool
def get_customer_details(ctx: RunContext[SupportDeps]) -> dict:
    """Retrieve customer account information."""
    return ctx.deps.get_customer_info()

@support_agent.tool
def search_help_articles(ctx: RunContext[SupportDeps], search_query: str) -> list[str]:
    """Search knowledge base for relevant help articles."""
    return ctx.deps.search_knowledge_base(search_query)

@support_agent.tool
def check_ticket_history(ctx: RunContext[SupportDeps]) -> list[dict]:
    """Review customer's previous support tickets."""
    return ctx.deps.ticket_history

# Use the agent
def handle_support_ticket(customer_id: str, message: str):
    deps = SupportDeps(
        customer_id=customer_id,
        ticket_history=[
            {"date": "2024-01-15", "issue": "Password reset", "resolved": True},
            {"date": "2024-02-03", "issue": "Billing question", "resolved": True}
        ]
    )
    
    result = support_agent.run_sync(message, deps=deps)
    
    print(f"Category: {result.data.category}")
    print(f"Priority: {result.data.priority}")
    print(f"Escalate: {result.data.requires_escalation}")
    print(f"Response: {result.data.message}")
    print(f"Suggested articles: {', '.join(result.data.suggested_articles)}")
    
    return result.data

# Example usage
response = handle_support_ticket(
    "CUST-12345",
    "I've been charged twice this month and my API calls are failing"
)

Advanced Patterns with Pydantic AI Agent Framework

As your applications grow in complexity, the Pydantic AI agent framework provides advanced patterns for sophisticated use cases.

Multi-Agent Collaboration

Create systems where multiple specialized agents work together to solve complex problems.

from pydantic_ai import Agent
from pydantic import BaseModel

class ResearchFindings(BaseModel):
    summary: str
    key_facts: list[str]
    sources: list[str]

class WrittenArticle(BaseModel):
    title: str
    content: str
    word_count: int

# Research agent
research_agent = Agent(
    'openai:gpt-4',
    result_type=ResearchFindings,
    system_prompt='You are a thorough researcher who finds accurate information.'
)

# Writing agent
writing_agent = Agent(
    'anthropic:claude-sonnet-4',
    result_type=WrittenArticle,
    system_prompt='You are a skilled writer who creates engaging articles.'
)

async def create_article(topic: str) -> WrittenArticle:
    # First agent researches the topic
    research = await research_agent.run(f"Research comprehensive information about: {topic}")
    
    # Second agent writes based on research
    article_prompt = f"""Write an article about {topic}.
    
    Use these research findings:
    Summary: {research.data.summary}
    Key facts: {', '.join(research.data.key_facts)}
    Sources: {', '.join(research.data.sources)}
    """
    
    article = await writing_agent.run(article_prompt)
    return article.data

Streaming Responses

The Pydantic AI agent framework supports streaming for real-time user experiences.

from pydantic_ai import Agent

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

async def stream_response(query: str):
    async with agent.run_stream(query) as response:
        async for text in response.stream_text():
            print(text, end='', flush=True)
        
        # Get final validated result
        final_result = await response.get_data()
        return final_result

# Usage
import asyncio
asyncio.run(stream_response("Explain quantum computing"))

Error Handling and Retries

Robust error handling is essential for production applications.

from pydantic_ai import Agent, ModelRetry, UnexpectedModelBehavior
from pydantic import BaseModel, ValidationError

class AnalysisResult(BaseModel):
    sentiment: str
    confidence: float
    
agent = Agent('openai:gpt-4', result_type=AnalysisResult)

async def analyze_with_retry(text: str, max_attempts: int = 3):
    for attempt in range(max_attempts):
        try:
            result = await agent.run(f"Analyze sentiment: {text}")
            return result.data
            
        except ModelRetry as e:
            print(f"Retry attempt {attempt + 1}: Model needs to retry")
            if attempt == max_attempts - 1:
                raise
                
        except ValidationError as e:
            print(f"Validation failed: {e}")
            # Could implement custom retry logic
            if attempt == max_attempts - 1:
                raise
                
        except UnexpectedModelBehavior as e:
            print(f"Unexpected behavior: {e}")
            raise

# Usage
result = asyncio.run(analyze_with_retry("This product exceeded my expectations!"))

Context Management and Memory

Maintain conversation context across multiple interactions.

from pydantic_ai import Agent
from pydantic import BaseModel

class ConversationMessage(BaseModel):
    role: str
    content: str

class ConversationAgent:
    def __init__(self, model: str = 'openai:gpt-4'):
        self.agent = Agent(model)
        self.history: list[ConversationMessage] = []
    
    async def chat(self, user_message: str) -> str:
        # Add user message to history
        self.history.append(ConversationMessage(role="user", content=user_message))
        
        # Build context from history
        context = "\n".join([
            f"{msg.role}: {msg.content}" 
            for msg in self.history[-10:]  # Keep last 10 messages
        ])
        
        # Get response
        result = await self.agent.run(
            f"Previous conversation:\n{context}\n\nRespond to the latest message."
        )
        
        # Add assistant response to history
        response_text = str(result.data)
        self.history.append(ConversationMessage(role="assistant", content=response_text))
        
        return response_text
    
    def clear_history(self):
        """Reset conversation history."""
        self.history = []

# Usage
conv_agent = ConversationAgent()
asyncio.run(conv_agent.chat("Hello, what can you help me with?"))
asyncio.run(conv_agent.chat("Tell me more about the first option"))

Integration with Popular AI Models

The Pydantic AI agent framework works seamlessly with multiple model providers, making it easy to switch or compare models.

OpenAI Integration

from pydantic_ai import Agent

# GPT-4 agent
gpt4_agent = Agent('openai:gpt-4')

# GPT-3.5 for faster, cheaper responses
gpt35_agent = Agent('openai:gpt-3.5-turbo')

# Specify custom parameters
custom_agent = Agent(
    'openai:gpt-4',
    model_settings={
        'temperature': 0.7,
        'max_tokens': 2000,
        'top_p': 0.9
    }
)

Anthropic Claude Integration

from pydantic_ai import Agent

# Claude Opus for complex reasoning
opus_agent = Agent('anthropic:claude-opus-4')

# Claude Sonnet for balanced performance
sonnet_agent = Agent('anthropic:claude-sonnet-4')

# Claude Haiku for speed
haiku_agent = Agent('anthropic:claude-haiku-4')

Google Gemini Integration

from pydantic_ai import Agent

# Gemini Pro
gemini_agent = Agent('google:gemini-pro')

# With custom settings
gemini_custom = Agent(
    'google:gemini-pro',
    model_settings={
        'temperature': 0.8,
        'top_k': 40,
        'top_p': 0.95
    }
)

Testing AI Agents

The Pydantic AI agent framework makes testing AI applications more reliable and straightforward.

Unit Testing with Mock Responses

import pytest
from pydantic_ai import Agent
from pydantic_ai.testing import TestModel

@pytest.fixture
def test_agent():
    return Agent('test', result_type=str)

def test_agent_response(test_agent):
    # Use TestModel for deterministic testing
    with TestModel() as test_model:
        test_model.add_response("This is a test response")
        
        result = test_agent.run_sync("Test query")
        assert result.data == "This is a test response"

def test_structured_output():
    from pydantic import BaseModel
    
    class Output(BaseModel):
        answer: str
        score: int
    
    agent = Agent('test', result_type=Output)
    
    with TestModel() as test_model:
        test_model.add_response(Output(answer="test", score=95))
        
        result = agent.run_sync("Query")
        assert result.data.answer == "test"
        assert result.data.score == 95

Integration Testing

import pytest
from pydantic_ai import Agent
from pydantic import BaseModel

class AnalysisResult(BaseModel):
    category: str
    confidence: float

@pytest.mark.integration
async def test_real_model_integration():
    agent = Agent('openai:gpt-3.5-turbo', result_type=AnalysisResult)
    
    result = await agent.run("Classify this as positive or negative: Great product!")
    
    assert result.data.category in ["positive", "negative", "neutral"]
    assert 0 <= result.data.confidence <= 1.0

@pytest.mark.integration  
async def test_tool_execution():
    agent = Agent('openai:gpt-4')
    
    @agent.tool
    def get_data(ctx, key: str) -> dict:
        return {"key": key, "value": "test_value"}
    
    result = await agent.run("Get data for key 'test123'")
    # Verify the tool was called and used correctly
    assert "test123" in str(result.data) or "test_value" in str(result.data)

Production Best Practices

Deploying Pydantic AI agents to production requires attention to reliability, performance, and observability.

Logging and Monitoring

import logging
from pydantic_ai import Agent
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MonitoredAgent:
    def __init__(self, model: str):
        self.agent = Agent(model)
        self.call_count = 0
        self.error_count = 0
    
    async def run_with_monitoring(self, prompt: str):
        start_time = datetime.now()
        self.call_count += 1
        
        try:
            logger.info(f"Agent call #{self.call_count} started")
            result = await self.agent.run(prompt)
            
            duration = (datetime.now() - start_time).total_seconds()
            logger.info(f"Agent call completed in {duration}s")
            
            return result.data
            
        except Exception as e:
            self.error_count += 1
            logger.error(f"Agent call failed: {e}")
            raise
    
    def get_metrics(self) -> dict:
        return {
            "total_calls": self.call_count,
            "errors": self.error_count,
            "error_rate": self.error_count / max(self.call_count, 1)
        }

Rate Limiting and Cost Control

import asyncio
from datetime import datetime, timedelta
from collections import deque

class RateLimitedAgent:
    def __init__(self, agent: Agent, max_calls_per_minute: int = 60):
        self.agent = agent
        self.max_calls = max_calls_per_minute
        self.call_times = deque()
    
    async def run(self, prompt: str):
        # Clean old entries
        cutoff = datetime.now() - timedelta(minutes=1)
        while self.call_times and self.call_times[0] < cutoff:
            self.call_times.popleft()
        
        # Check rate limit
        if len(self.call_times) >= self.max_calls:
            wait_time = 60 - (datetime.now() - self.call_times[0]).total_seconds()
            if wait_time > 0:
                await asyncio.sleep(wait_time)
        
        # Make call
        self.call_times.append(datetime.now())
        return await self.agent.run(prompt)

Caching Responses

from functools import lru_cache
import hashlib
import json

class CachedAgent:
    def __init__(self, agent: Agent):
        self.agent = agent
        self.cache = {}
    
    def _get_cache_key(self, prompt: str, **kwargs) -> str:
        cache_data = {"prompt": prompt, **kwargs}
        cache_str = json.dumps(cache_data, sort_keys=True)
        return hashlib.md5(cache_str.encode()).hexdigest()
    
    async def run(self, prompt: str, use_cache: bool = True, **kwargs):
        if use_cache:
            cache_key = self._get_cache_key(prompt, **kwargs)
            
            if cache_key in self.cache:
                return self.cache[cache_key]
        
        result = await self.agent.run(prompt, **kwargs)
        
        if use_cache:
            self.cache[cache_key] = result
        
        return result
    
    def clear_cache(self):
        self.cache = {}

Real-World Use Cases

The Pydantic AI agent framework excels in various production scenarios.

Content Moderation System

from pydantic import BaseModel, Field
from pydantic_ai import Agent
from enum import Enum

class ModerationCategory(str, Enum):
    SAFE = "safe"
    SPAM = "spam"
    HARASSMENT = "harassment"
    HATE_SPEECH = "hate_speech"
    EXPLICIT = "explicit"

class ModerationResult(BaseModel):
    category: ModerationCategory
    confidence: float = Field(ge=0, le=1)
    flagged: bool
    reasoning: str

moderation_agent = Agent(
    'openai:gpt-4',
    result_type=ModerationResult,
    system_prompt="""You are a content moderation system. 
    Analyze content for policy violations including spam, harassment, 
    hate speech, and explicit material. Be accurate and fair."""
)

async def moderate_content(content: str) -> ModerationResult:
    result = await moderation_agent.run(f"Moderate this content: {content}")
    return result.data

Data Extraction Pipeline

from pydantic import BaseModel
from pydantic_ai import Agent
from typing import Optional

class ExtractedData(BaseModel):
    company_name: Optional[str] = None
    contact_email: Optional[str] = None
    phone_number: Optional[str] = None
    address: Optional[str] = None
    industry: Optional[str] = None

extraction_agent = Agent(
    'anthropic:claude-sonnet-4',
    result_type=ExtractedData,
    system_prompt='Extract structured data from unstructured text.'
)

async def extract_business_info(text: str) -> ExtractedData:
    result = await extraction_agent.run(
        f"Extract business information from this text:\n\n{text}"
    )
    return result.data

# Example usage
text = """
Acme Corporation is a leading tech company based at 123 Main St, San Francisco, CA.
Contact us at info@acme.com or call (555) 123-4567. We specialize in AI solutions.
"""

data = asyncio.run(extract_business_info(text))
print(f"Company: {data.company_name}")
print(f"Email: {data.contact_email}")

Performance Optimization

Optimize your Pydantic AI agents for production-grade performance.

Async Batch Processing

import asyncio
from pydantic_ai import Agent

async def process_batch(items: list[str], agent: Agent) -> list:
    tasks = [agent.run(item) for item in items]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    processed = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Item {i} failed: {result}")
        else:
            processed.append(result.data)
    
    return processed

# Usage
agent = Agent('openai:gpt-3.5-turbo')
items = ["Query 1", "Query 2", "Query 3", "Query 4", "Query 5"]
results = asyncio.run(process_batch(items, agent))

Connection Pooling

from pydantic_ai import Agent
import httpx

# Create persistent HTTP client for better performance
http_client = httpx.AsyncClient(
    limits=httpx.Limits(max_keepalive_connections=20, max_connections=100),
    timeout=httpx.Timeout(30.0)
)

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

# Remember to close the client when done
# await http_client.aclose()

Common Patterns and Solutions

Dynamic System Prompts

from pydantic_ai import Agent

class DynamicAgent:
    def __init__(self, base_model: str = 'openai:gpt-4'):
        self.base_model = base_model
    
    def create_agent(self, role: str, context: dict) -> Agent:
        system_prompt = f"""You are a {role}.
        
        Additional context:
        {chr(10).join(f"- {k}: {v}" for k, v in context.items())}
        """
        
        return Agent(self.base_model, system_prompt=system_prompt)

# Usage
dynamic = DynamicAgent()
agent = dynamic.create_agent(
    role="financial advisor",
    context={"risk_tolerance": "moderate", "time_horizon": "10 years"}
)

Conclusion

The Pydantic AI agent framework represents a significant advancement in building production-ready AI agents with Python. By combining Pydantic's robust validation with powerful agent orchestration capabilities, it provides developers with the tools needed to create reliable, type-safe, and maintainable AI applications.

From simple chatbots to complex multi-agent systems, the framework handles the heavy lifting of data validation, error management, and model integration. This allows developers to focus on building great AI experiences rather than wrestling with infrastructure concerns.

As AI continues to evolve, frameworks like Pydantic AI that prioritize developer experience, type safety, and production readiness will become increasingly essential. Whether you're building your first AI agent or scaling to production systems handling millions of requests, the Pydantic AI agent framework provides a solid foundation for success.

Start exploring the Pydantic AI agent framework today and discover how it can accelerate your AI development while maintaining the reliability and maintainability that production applications demand.