Pydantic AI Framework Cheat Sheet

Sinopsis

Pydantic AI es un poderoso marco de agente de Python diseñado para hacer aplicaciones de producción de edificios con IA Generativa más directa y confiable. Construido sobre la popular biblioteca de validación de datos Pydantic, Pydantic AI supera la brecha entre las aplicaciones tradicionales de Python y los flujos de trabajo impulsados por AI proporcionando seguridad de tipo robusto, validación de esquemas y salidas estructuradas para interacciones AI.

Lo que distingue a Pydantic AI es su enfoque en la preparación de la producción y la seguridad del tipo. Si bien muchos marcos de IA priorizan la flexibilidad a expensas de la fiabilidad, Pydantic AI asegura que los productos generados por IA se ajusten a los esquemas esperados, lo que facilita la construcción de aplicaciones robustas que puedan manejar la imprevisibilidad de los modelos de idiomas grandes. Al aprovechar las anotaciones tipo Python y las capacidades de validación de Pydantic, el marco proporciona un enfoque estructurado para el desarrollo de la IA que reduce el código de caldera y los posibles errores de tiempo de ejecución.

Pydantic La IA es particularmente valiosa para los desarrolladores que necesitan integrar las capacidades de IA en las aplicaciones existentes de Python o crear nuevas características impulsadas por IA con confianza. Su énfasis en seguridad de tipo, validación e integración limpia con las prácticas modernas de Python hace que sea una excelente opción para entornos de producción donde la fiabilidad y la sostenibilidad son primordiales.

Instalación y configuración

Instalación básica

# 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

Configuración del medio ambiente

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
)

Estructura del proyecto

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

Conceptos básicos

Agente

La clase Agente es la interfaz principal para interactuar con los modelos de lenguaje. Maneja la comunicación con el LLM subyacente y proporciona métodos para generar respuestas.

Prompt

Prompts define las instrucciones enviadas al modelo de idioma. Pydantic AI proporciona formas estructuradas de crear y gestionar los impulsos.

Producto

Las clases de productos definen la estructura prevista de las respuestas del modelo de idioma. Aprovechan las capacidades de validación de Pydantic para asegurar que las respuestas se ajusten al esquema esperado.

Validador

Los validadores son funciones que verifican si la salida generada cumple criterios específicos más allá de la validación básica del esquema.

Uso básico

Crear un agente simple

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)

Producto estructurado con modelos pidánticos

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()

Plantillas Prompt

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)

Características avanzadas

Promptas estructuradas con Pydantic

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)

Validadores personalizados

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

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())

Función de llamada

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)

Integración de herramientas

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)

Patrones avanzados

Cadena de la razón del pensamiento

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\\\\}")

Procesamiento multi-procesamiento

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)\\\\}")

Manejo de errores y registros

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

Operaciones Async

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())

Integración con los marcos web

Integración FastAPI

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

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

Buenas prácticas

Diseño de esquemas

• Iniciar Simple: Empezar con esquemas simples y agregar gradualmente complejidad
• Use Field Descriptions: Siempre incluye descripciones claras para cada campo
• ** Tipos de apropiación**: Elija los tipos apropiados para cada campo (estring, int, flotador, lista, etc.)
• Optional Fields: Use Optional for fields that might not always be present
• Constraints de validación: Añadir limitaciones como min/max longitud, patrones regex, etc.
• Nested Models: Utilice modelos anidados para estructuras de datos complejas
• Enums: Use Clases de Enum para campos con un conjunto fijo de posibles valores

Prompt Engineering

• ** Instrucciones de vuelo**: Proporcionar instrucciones claras y específicas en los avisos
• Contexto: Incluir el contexto pertinente para la tarea
• Ejemplos: Incluir ejemplos de salida esperada cuando sea posible
• ** Formato Estructurado**: Use formatos estructurados como listas numeradas o secciones
• Evitar la ambigüedad: Sé explícito sobre lo que quieres que haga el modelo
• Refinamiento alternativo: Refinar los impulsos basados en productos modelo

Manejo de errores

• ** Errores de validación**: Errores de validación manual
• Retry Logic: Implementar la lógica de retry para errores transitorios
• Fallbacks: Tener estrategias de retroceso cuando el enfoque primario falla
• Logging: Errores de registro y respuestas modelo para depurar
• Retroalimentación del usuario: Proporcionar mensajes de error útiles a los usuarios

Optimización del rendimiento

• Procesamiento de lotes: Procesar varios elementos en lotes cuando sea posible
• Caching: Cache responses for identical or similar requests
• ** Operaciones anónimas**: Utilizar funciones asinc para operaciones simultáneas
• ** Selección modelo**: Elija modelos apropiados basados en la complejidad de la tarea
• Token Management: Tenga en cuenta el uso de token para controlar los costos

Solución de problemas

Cuestiones comunes

Fallos de validación de Schema

• Causa: La salida del modelo no coincide con el esquema esperado
• Solución: Refinar los impulsos para ser más explícito sobre el formato requerido, o ajustar el esquema para ser más flexible

Productos inconsistentes

• Causa: Indicaciones de vacío o configuración de alta temperatura
• Solución: Hacer los impulsos más específicos y reducir la temperatura para productos más consistentes

Rendimiento lento

• Causa: Grandes solicitudes, esquemas complejos o procesamiento ineficiente
• Solución: Optimize prompts, use async operations, or implement caching

Uso de alta señal

• Causa: Incitaciones de verbos o contexto innecesario
• Solution: Streamline prompts, use modelos más eficientes o implemente monitoreo de usos de token

-...

*Esta completa hoja de trampa Pydantic AI proporciona todo lo necesario para construir aplicaciones de IA de grado de producción con seguridad de tipo y validación. Desde la configuración básica hasta patrones avanzados, utilice estos ejemplos y mejores prácticas para crear características sólidas y sostenibles con IA Pydantic enfoque estructurado. *