Appearance
Pydantic AI Framework Cheat Sheet
Overview
Pydantic AI is a powerful Python agent framework designed to make building production-grade applications with Generative AI more straightforward and reliable. Built on top of the popular Pydantic data validation library, Pydantic AI bridges the gap between traditional Python applications and AI-driven workflows by providing robust type safety, schema validation, and structured outputs for AI interactions.
What sets Pydantic AI apart is its focus on production readiness and type safety. While many AI frameworks prioritize flexibility at the expense of reliability, Pydantic AI ensures that AI-generated outputs conform to expected schemas, making it easier to build robust applications that can handle the unpredictability of large language models. By leveraging Python's type annotations and Pydantic's validation capabilities, the framework provides a structured approach to AI development that reduces boilerplate code and potential runtime errors.
Pydantic AI is particularly valuable for developers who need to integrate AI capabilities into existing Python applications or build new AI-powered features with confidence. Its emphasis on type safety, validation, and clean integration with modern Python practices makes it an excellent choice for production environments where reliability and maintainability are paramount.
Installation and Setup
Basic Installation
bash
# Install Pydantic AI
pip install pydantic-ai
# Install with optional dependencies
pip install "pydantic-ai[all]"
# Install development version
pip install git+https://github.com/pydantic/pydantic-ai.git
Environment Configuration
python
import os
from pydantic_ai import Agent, Output, Prompt
# Set up API keys
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
# Basic agent configuration
agent = Agent(
model="gpt-4",
temperature=0.7,
max_tokens=1000
)
Project Structure
pydantic_ai_project/
├── agents/
│ ├── __init__.py
│ ├── text_agent.py
│ └── structured_agent.py
├── models/
│ ├── __init__.py
│ ├── input_models.py
│ └── output_models.py
├── prompts/
│ ├── __init__.py
│ └── templates.py
├── services/
│ ├── __init__.py
│ └── ai_service.py
├── utils/
│ ├── __init__.py
│ └── validators.py
├── config/
│ ├── __init__.py
│ └── settings.py
└── main.py
Core Concepts
Agent
The Agent class is the primary interface for interacting with language models. It handles the communication with the underlying LLM and provides methods for generating responses.
Prompt
Prompts define the instructions sent to the language model. Pydantic AI provides structured ways to create and manage prompts.
Output
Output classes define the expected structure of responses from the language model. They leverage Pydantic's validation capabilities to ensure responses conform to the expected schema.
Validator
Validators are functions that check if the generated output meets specific criteria beyond basic schema validation.
Basic Usage
Creating a Simple Agent
python
from pydantic_ai import Agent
# Create a basic agent
agent = Agent(
model="gpt-4",
temperature=0.7,
max_tokens=1000
)
# Generate a simple text response
response = agent.run("Explain quantum computing in simple terms.")
print(response)
Structured Output with Pydantic Models
python
from pydantic_ai import Agent, Output
from pydantic import BaseModel, Field
from typing import List
# Define output schema
class MovieRecommendation(BaseModel):
title: str = Field(description="The title of the movie")
year: int = Field(description="The year the movie was released")
director: str = Field(description="The director of the movie")
genre: List[str] = Field(description="List of genres for the movie")
rating: float = Field(description="Rating from 0.0 to 10.0")
class MovieRecommendations(BaseModel):
recommendations: List[MovieRecommendation] = Field(description="List of movie recommendations")
reasoning: str = Field(description="Explanation for these recommendations")
# Create output definition
movie_output = Output(MovieRecommendations)
# Create agent and get structured recommendations
agent = Agent(model="gpt-4")
result = agent.run(
"Recommend 3 science fiction movies from the 1980s",
output=movie_output
)
# Access structured data
for movie in result.recommendations:
print(f"{movie.title} ({movie.year}) - Directed by {movie.director}")
print(f"Genres: {', '.join(movie.genre)}")
print(f"Rating: {movie.rating}/10")
print()
Prompt Templates
python
from pydantic_ai import Agent, Prompt
from typing import List
# Create a prompt template
movie_prompt = Prompt(
"""
You are a movie expert with extensive knowledge of cinema history.
Please recommend {num_movies} movies that match these criteria:
- Genre: {genre}
- Time period: {era}
- Mood: {mood}
For each movie, provide:
1. Title
2. Year of release
3. Director
4. Brief plot summary
5. Why it matches the criteria
"""
)
# Use the template with parameters
agent = Agent(model="gpt-4")
response = agent.run(
movie_prompt.format(
num_movies=3,
genre="Film Noir",
era="1940s",
mood="Suspenseful"
)
)
print(response)
Advanced Features
Structured Prompts with Pydantic
python
from pydantic_ai import Agent, StructuredPrompt
from pydantic import BaseModel, Field
from typing import List, Optional
# Define input schema
class MovieCriteria(BaseModel):
genres: List[str] = Field(description="List of movie genres")
release_years: Optional[List[int]] = Field(description="List of years or range of years")
directors: Optional[List[str]] = Field(description="List of directors to include")
exclude_directors: Optional[List[str]] = Field(description="List of directors to exclude")
min_rating: Optional[float] = Field(description="Minimum rating threshold (0-10)")
keywords: Optional[List[str]] = Field(description="Thematic keywords to match")
# Create structured prompt
movie_prompt = StructuredPrompt(
"""
As a film expert, recommend 5 movies that match the following criteria:
{{criteria}}
For each movie, provide the title, year, director, and a brief explanation of why it matches the criteria.
""",
criteria=MovieCriteria
)
# Use structured prompt
agent = Agent(model="gpt-4")
response = agent.run(
movie_prompt,
criteria=MovieCriteria(
genres=["Science Fiction", "Horror"],
release_years=[1970, 1971, 1972, 1973, 1974, 1975, 1976, 1977, 1978, 1979],
directors=["John Carpenter", "David Cronenberg", "Ridley Scott"],
min_rating=7.5,
keywords=["Body horror", "Dystopia", "Alien"]
)
)
print(response)
Custom Validators
python
from pydantic_ai import Agent, Output, Validator
from pydantic import BaseModel, Field
from typing import List
# Define output schema
class NewsArticleSummary(BaseModel):
title: str = Field(description="The title of the article")
source: str = Field(description="The source of the article")
summary: str = Field(description="A concise summary of the article")
key_points: List[str] = Field(description="Key points from the article")
bias_assessment: str = Field(description="Assessment of potential bias in the article")
# Create custom validators
def check_summary_length(output: NewsArticleSummary) -> bool:
"""Ensure summary is between 100 and 300 characters."""
return 100 <= len(output.summary) <= 300
def check_key_points(output: NewsArticleSummary) -> bool:
"""Ensure there are 3-5 key points and none are too long."""
if not (3 <= len(output.key_points) <= 5):
return False
return all(len(point) <= 100 for point in output.key_points)
# Create validator with custom checks
news_validator = Validator(
checks=[check_summary_length, check_key_points],
max_retries=3
)
# Create output with validator
news_output = Output(
NewsArticleSummary,
validator=news_validator
)
# Use with agent
agent = Agent(model="gpt-4")
result = agent.run(
"Summarize this news article about climate change: [article text here]",
output=news_output
)
print(f"Title: {result.title}")
print(f"Source: {result.source}")
print(f"Summary: {result.summary}")
print("Key Points:")
for point in result.key_points:
print(f"- {point}")
print(f"Bias Assessment: {result.bias_assessment}")
Streaming Responses
python
from pydantic_ai import Agent
import asyncio
# Create agent
agent = Agent(model="gpt-4")
# Synchronous streaming
for chunk in agent.stream("Write a short story about a robot learning to paint."):
print(chunk, end="", flush=True)
# Asynchronous streaming
async def stream_response():
async for chunk in agent.astream("Write a poem about artificial intelligence."):
print(chunk, end="", flush=True)
asyncio.run(stream_response())
Function Calling
python
from pydantic_ai import Agent, Function
from pydantic import BaseModel, Field
from typing import List, Optional
# Define function input schema
class WeatherRequest(BaseModel):
location: str = Field(description="City and country or postal code")
forecast_days: Optional[int] = Field(default=1, description="Number of days to forecast")
units: Optional[str] = Field(default="metric", description="Units system (metric or imperial)")
# Define function
def get_weather(request: WeatherRequest) -> str:
"""Get the current weather and forecast for a location."""
# In a real application, this would call a weather API
return f"Weather for {request.location}: Sunny, 25°C, with light winds."
# Register function with agent
agent = Agent(model="gpt-4")
weather_function = Function(get_weather)
agent.register_function(weather_function)
# Use function calling
response = agent.run(
"What's the weather like in Paris, France today?",
functions=[weather_function]
)
print(response)
Tool Integration
python
from pydantic_ai import Agent, Tool
from pydantic import BaseModel, Field
from typing import List, Optional
import requests
# Define tool input schema
class SearchQuery(BaseModel):
query: str = Field(description="Search query string")
num_results: Optional[int] = Field(default=5, description="Number of results to return")
# Define tool output schema
class SearchResult(BaseModel):
title: str = Field(description="Title of the search result")
url: str = Field(description="URL of the search result")
snippet: str = Field(description="Text snippet from the search result")
class SearchResults(BaseModel):
results: List[SearchResult] = Field(description="List of search results")
# Create search tool
def web_search(query: SearchQuery) -> SearchResults:
"""Search the web for information."""
# In a real application, this would call a search API
mock_results = [
SearchResult(
title=f"Result {i} for {query.query}",
url=f"https://example.com/result{i}",
snippet=f"This is a snippet for result {i} about {query.query}..."
)
for i in range(1, query.num_results + 1)
]
return SearchResults(results=mock_results)
# Register tool with agent
agent = Agent(model="gpt-4")
search_tool = Tool(
name="web_search",
function=web_search,
description="Search the web for current information"
)
agent.register_tool(search_tool)
# Use tool in conversation
response = agent.run(
"Find me information about the latest advancements in quantum computing",
tools=[search_tool]
)
print(response)
Advanced Patterns
Chain of Thought Reasoning
python
from pydantic_ai import Agent, Output
from pydantic import BaseModel, Field
from typing import List
# Define output schema with reasoning steps
class MathSolution(BaseModel):
problem: str = Field(description="The original math problem")
reasoning_steps: List[str] = Field(description="Step-by-step reasoning process")
final_answer: str = Field(description="The final numerical answer")
confidence: float = Field(description="Confidence in the answer (0-1)")
# Create output
math_output = Output(MathSolution)
# Use with agent
agent = Agent(model="gpt-4")
result = agent.run(
"Solve this math problem: If a train travels at 120 km/h and covers a distance of 360 km, how long does the journey take?",
output=math_output
)
print(f"Problem: {result.problem}")
print("Reasoning Steps:")
for i, step in enumerate(result.reasoning_steps, 1):
print(f"{i}. {step}")
print(f"Final Answer: {result.final_answer}")
print(f"Confidence: {result.confidence:.2f}")
Multi-Step Processing
python
from pydantic_ai import Agent, Output
from pydantic import BaseModel, Field
from typing import List
# Define intermediate and final output schemas
class TextAnalysis(BaseModel):
main_topics: List[str] = Field(description="Main topics identified in the text")
sentiment: str = Field(description="Overall sentiment of the text")
key_entities: List[str] = Field(description="Key entities mentioned in the text")
class ArticleSummary(BaseModel):
title: str = Field(description="Suggested title for the article")
summary: str = Field(description="Concise summary of the article")
keywords: List[str] = Field(description="Keywords for SEO")
# Create agent
agent = Agent(model="gpt-4")
# Multi-step process
def process_article(text: str) -> ArticleSummary:
# Step 1: Analyze the text
analysis_output = Output(TextAnalysis)
analysis = agent.run(
f"Analyze this text and identify main topics, sentiment, and key entities:\n\n{text}",
output=analysis_output
)
# Step 2: Generate summary based on analysis
summary_output = Output(ArticleSummary)
summary = agent.run(
f"""
Based on the following analysis, create a title, summary, and keywords:
Main Topics: {', '.join(analysis.main_topics)}
Sentiment: {analysis.sentiment}
Key Entities: {', '.join(analysis.key_entities)}
Original Text:
{text}
""",
output=summary_output
)
return summary
# Use the multi-step process
article_text = """
[Long article text here...]
"""
result = process_article(article_text)
print(f"Title: {result.title}")
print(f"Summary: {result.summary}")
print(f"Keywords: {', '.join(result.keywords)}")
Error Handling and Retries
python
from pydantic_ai import Agent, Output, Validator
from pydantic import BaseModel, Field, ValidationError
from typing import List, Optional
import time
# Define output schema
class CodeSolution(BaseModel):
problem_statement: str = Field(description="The original coding problem")
language: str = Field(description="Programming language used")
code: str = Field(description="The solution code")
explanation: str = Field(description="Explanation of how the code works")
test_cases: List[str] = Field(description="Example test cases")
# Custom validator with specific code checks
def validate_python_code(solution: CodeSolution) -> bool:
if solution.language.lower() != "python":
return True # Skip validation for non-Python code
# Check for common Python syntax issues
code = solution.code
if "import" not in code and "def " not in code:
return False
# Check for test cases
if len(solution.test_cases) < 2:
return False
return True
# Create validator with retry logic
code_validator = Validator(
checks=[validate_python_code],
max_retries=3,
retry_delay=1.0
)
# Create output with validator
code_output = Output(
CodeSolution,
validator=code_validator
)
# Use with agent and handle errors
agent = Agent(model="gpt-4")
try:
result = agent.run(
"Write a Python function to find the longest palindromic substring in a given string.",
output=code_output
)
print(f"Language: {result.language}")
print(f"Solution:\n{result.code}")
print(f"Explanation: {result.explanation}")
print("Test Cases:")
for test in result.test_cases:
print(f"- {test}")
except ValidationError as e:
print(f"Validation error: {e}")
# Handle validation failure
except Exception as e:
print(f"Error: {e}")
# Handle other errors
Async Operations
python
from pydantic_ai import Agent, Output
from pydantic import BaseModel, Field
import asyncio
from typing import List
# Define output schemas
class TranslationResult(BaseModel):
original_text: str = Field(description="The original text")
translated_text: str = Field(description="The translated text")
language: str = Field(description="The target language")
# Create agent
agent = Agent(model="gpt-4")
# Async translation function
async def translate_text(text: str, languages: List[str]) -> List[TranslationResult]:
# Create output
translation_output = Output(TranslationResult)
# Create translation tasks
tasks = []
for language in languages:
task = agent.arun(
f"Translate the following text to {language}:\n\n{text}",
output=translation_output
)
tasks.append(task)
# Run tasks concurrently
results = await asyncio.gather(*tasks)
return results
# Use async function
async def main():
text = "Artificial intelligence is transforming the world in unprecedented ways."
languages = ["Spanish", "French", "German", "Japanese", "Russian"]
translations = await translate_text(text, languages)
for translation in translations:
print(f"{translation.language}: {translation.translated_text}")
# Run async function
asyncio.run(main())
Integration with Web Frameworks
FastAPI Integration
python
from fastapi import FastAPI, HTTPException
from pydantic_ai import Agent, Output
from pydantic import BaseModel, Field
from typing import List
app = FastAPI()
# Initialize agent
agent = Agent(model="gpt-4")
# Define request and response models
class SummaryRequest(BaseModel):
text: str = Field(description="Text to summarize")
max_length: int = Field(default=200, description="Maximum summary length")
class SummaryResponse(BaseModel):
original_length: int = Field(description="Length of original text")
summary: str = Field(description="Generated summary")
keywords: List[str] = Field(description="Key topics in the text")
# Define output schema
class TextSummary(BaseModel):
summary: str = Field(description="Concise summary of the text")
keywords: List[str] = Field(description="Key topics extracted from the text")
# Create endpoint
@app.post("/summarize", response_model=SummaryResponse)
async def summarize_text(request: SummaryRequest):
try:
# Create output
summary_output = Output(TextSummary)
# Generate summary
result = await agent.arun(
f"""
Summarize the following text in no more than {request.max_length} characters:
{request.text}
""",
output=summary_output
)
# Create response
return SummaryResponse(
original_length=len(request.text),
summary=result.summary,
keywords=result.keywords
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# Run with: uvicorn app:app --reload
Flask Integration
python
from flask import Flask, request, jsonify
from pydantic_ai import Agent, Output
from pydantic import BaseModel, Field
from typing import List
app = Flask(__name__)
# Initialize agent
agent = Agent(model="gpt-4")
# Define output schema
class ContentGenerator(BaseModel):
title: str = Field(description="Attention-grabbing title")
introduction: str = Field(description="Engaging introduction paragraph")
sections: List[str] = Field(description="Main content sections")
conclusion: str = Field(description="Concluding paragraph")
# Create output
content_output = Output(ContentGenerator)
@app.route("/generate-content", methods=["POST"])
def generate_content():
data = request.json
topic = data.get("topic", "")
tone = data.get("tone", "informative")
length = data.get("length", "medium")
if not topic:
return jsonify({"error": "Topic is required"}), 400
try:
# Generate content
result = agent.run(
f"""
Generate content about "{topic}" with a {tone} tone.
The content should be {length} length.
""",
output=content_output
)
# Create response
return jsonify({
"title": result.title,
"introduction": result.introduction,
"sections": result.sections,
"conclusion": result.conclusion
})
except Exception as e:
return jsonify({"error": str(e)}), 500
# Run with: flask run
Best Practices
Schema Design
- Start Simple: Begin with simple schemas and gradually add complexity
- Use Field Descriptions: Always include clear descriptions for each field
- Appropriate Types: Choose appropriate types for each field (string, int, float, list, etc.)
- Optional Fields: Use Optional for fields that might not always be present
- Validation Constraints: Add constraints like min/max length, regex patterns, etc.
- Nested Models: Use nested models for complex data structures
- Enums: Use Enum classes for fields with a fixed set of possible values
Prompt Engineering
- Clear Instructions: Provide clear, specific instructions in prompts
- Context: Include relevant context for the task
- Examples: Include examples of expected output when possible
- Structured Format: Use structured formats like numbered lists or sections
- Avoid Ambiguity: Be explicit about what you want the model to do
- Iterative Refinement: Refine prompts based on model outputs
Error Handling
- Validation Errors: Handle validation errors gracefully
- Retry Logic: Implement retry logic for transient errors
- Fallbacks: Have fallback strategies when the primary approach fails
- Logging: Log errors and model responses for debugging
- User Feedback: Provide helpful error messages to users
Performance Optimization
- Batch Processing: Process multiple items in batch when possible
- Caching: Cache responses for identical or similar requests
- Async Operations: Use async functions for concurrent operations
- Model Selection: Choose appropriate models based on task complexity
- Token Management: Be mindful of token usage to control costs
Troubleshooting
Common Issues
Schema Validation Failures
- Cause: Model output doesn't match the expected schema
- Solution: Refine prompts to be more explicit about the required format, or adjust the schema to be more flexible
Inconsistent Outputs
- Cause: Vague prompts or high temperature settings
- Solution: Make prompts more specific and reduce temperature for more consistent outputs
Slow Performance
- Cause: Large requests, complex schemas, or inefficient processing
- Solution: Optimize prompts, use async operations, or implement caching
High Token Usage
- Cause: Verbose prompts or unnecessary context
- Solution: Streamline prompts, use more efficient models, or implement token usage monitoring
This comprehensive Pydantic AI cheat sheet provides everything needed to build production-grade AI applications with type safety and validation. From basic setup to advanced patterns, use these examples and best practices to create robust, maintainable AI-powered features with Pydantic AI's structured approach.