xbe.at

Python / Python for ML and AI

Building Production-Grade AI Apps with PydanticAI

9 min read, 15 sections

Contents
  1. PydanticAI Core Components
  2. Dynamic Prompt Engineering
  3. Streamed Response Handling
  4. Dependency Injection System
  5. Validation and Error Handling
  6. Monitoring Integration with Logfire
  7. Custom Model Integration
  8. Real-world Example: Document Classification
  9. Results for: Document Classification
  10. Advanced Token Management
  11. Real-world Example: Text Summarization Pipeline
  12. Results for: Text Summarization Pipeline
  13. Advanced Error Recovery
  14. Model Chain Composition
  15. Additional Resources

PydanticAI Core Components

PydanticAI extends Pydantic's type system to create type-safe LLM applications. The core components enable model-agnostic development with structured validation, allowing developers to define robust interfaces for AI model interactions while maintaining pure Python implementation.

from pydanticai import BaseAIModel, Field
from typing import List, Optional

class DocumentAnalyzer(BaseAIModel):
    # Define structured input/output
    class Input:
        text: str = Field(..., description="Document text to analyze")
        max_tokens: Optional[int] = Field(1000, description="Max tokens to process")
    
    class Output:
        summary: str
        key_points: List[str]
        sentiment: float

    async def execute(self, input: Input) -> Output:
        # Model implementation
        result = await self.llm.analyze(
            text=input.text,
            max_tokens=input.max_tokens
        )
        return self.Output(
            summary=result.summary,
            key_points=result.points,
            sentiment=result.sentiment_score
        )

# Usage
analyzer = DocumentAnalyzer()
result = await analyzer.execute(text="Sample document...")

Dynamic Prompt Engineering

The framework provides a sophisticated prompt engineering system that enables dynamic template composition and validation. This allows for complex prompt chains while maintaining type safety and enabling dependency injection for testing different prompt strategies.

from pydanticai import PromptTemplate, Field
from typing import List

class SummarizationPrompt(PromptTemplate):
    context: str = Field(..., description="Context for summarization")
    key_points: List[str] = Field(default_factory=list)
    
    def compose(self) -> str:
        base_prompt = f"Summarize the following text:\n{self.context}\n"
        if self.key_points:
            points = "\n".join(f"- {point}" for point in self.key_points)
            base_prompt += f"\nFocus on these aspects:\n{points}"
        return base_prompt

# Example usage
prompt = SummarizationPrompt(
    context="Long technical document...",
    key_points=["Architecture", "Implementation", "Results"]
)
formatted_prompt = prompt.compose()

Streamed Response Handling

PydanticAI implements an advanced streaming response system that enables real-time processing of LLM outputs. This system maintains type safety while allowing for incremental processing of responses, crucial for long-running AI operations.

from pydanticai import StreamHandler, StreamedResponse
from typing import AsyncGenerator

class TextStreamHandler(StreamHandler):
    async def process_chunk(self, chunk: str) -> str:
        return chunk.strip()
    
    async def handle_stream(
        self, 
        response: AsyncGenerator[str, None]
    ) -> StreamedResponse:
        collected_text = []
        async for chunk in response:
            processed = await self.process_chunk(chunk)
            collected_text.append(processed)
            yield StreamedResponse(
                partial_result="".join(collected_text),
                is_complete=False
            )
        yield StreamedResponse(
            partial_result="".join(collected_text),
            is_complete=True
        )

# Usage example
handler = TextStreamHandler()
async for response in handler.handle_stream(model.generate()):
    print(f"Received: {response.partial_result}")

Dependency Injection System

The dependency injection system allows for flexible component replacement and testing. This architectural pattern enables developers to swap implementations of AI models, prompts, and handlers while maintaining consistent interfaces.

from pydanticai import Injector, Provider
from typing import Protocol, runtime_checkable

@runtime_checkable
class LLMProvider(Protocol):
    async def generate(self, prompt: str) -> str: ...

class MockLLM(LLMProvider):
    async def generate(self, prompt: str) -> str:
        return f"Mocked response for: {prompt}"

class ProductionLLM(LLMProvider):
    async def generate(self, prompt: str) -> str:
        # Real LLM implementation
        pass

# Configure dependency injection
injector = Injector()
injector.bind(LLMProvider, MockLLM())  # For testing
# injector.bind(LLMProvider, ProductionLLM())  # For production

# Usage
llm = injector.get(LLMProvider)
response = await llm.generate("Test prompt")

Validation and Error Handling

PydanticAI implements comprehensive validation and error handling mechanisms that ensure robust operation in production environments. The system provides detailed error feedback and maintains type safety throughout the execution pipeline.

from pydanticai import AIValidator, ValidationError
from typing import Optional, Dict, Any

class ResponseValidator(AIValidator):
    def validate_token_count(
        self, 
        response: str, 
        max_tokens: int
    ) -> bool:
        return len(response.split()) <= max_tokens
    
    async def validate_response(
        self, 
        response: str, 
        context: Dict[str, Any]
    ) -> Optional[ValidationError]:
        if not self.validate_token_count(
            response, 
            context.get('max_tokens', 1000)
        ):
            return ValidationError(
                "Response exceeds token limit",
                field="response",
                context=context
            )
        return None

# Usage
validator = ResponseValidator()
try:
    result = await validator.validate_response(
        "Long response...", 
        {"max_tokens": 500}
    )
    if result:
        print(f"Validation failed: {result}")
except Exception as e:
    print(f"Error during validation: {e}")

Monitoring Integration with Logfire

PydanticAI integrates seamlessly with Logfire for comprehensive monitoring and debugging of AI applications. The integration provides detailed insights into model performance, response times, and error patterns in production environments.

from pydanticai import LogfireMonitor, MetricsCollector
from datetime import datetime

class AIMonitor(LogfireMonitor):
    def __init__(self, app_name: str):
        self.metrics = MetricsCollector()
        self.app_name = app_name
        
    async def log_inference(
        self, 
        model_name: str, 
        duration_ms: float, 
        status: str
    ):
        await self.metrics.record({
            'timestamp': datetime.utcnow(),
            'app': self.app_name,
            'model': model_name,
            'duration_ms': duration_ms,
            'status': status,
        })
    
    async def export_metrics(self):
        return await self.metrics.aggregate()

# Usage
monitor = AIMonitor("production-app")
await monitor.log_inference(
    "gpt-4", 
    1234.56, 
    "success"
)
metrics = await monitor.export_metrics()

Custom Model Integration

PydanticAI provides a flexible interface for integrating custom AI models while maintaining type safety and validation. This implementation shows how to wrap existing models within the PydanticAI framework.

from pydanticai import ModelWrapper, ModelConfig
from typing import Any, Dict

class CustomModelWrapper(ModelWrapper):
    def __init__(self, model: Any, config: ModelConfig):
        self.model = model
        self.config = config
        
    async def preprocess(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        # Custom preprocessing logic
        return {
            k: v.strip() if isinstance(v, str) else v 
            for k, v in inputs.items()
        }
        
    async def predict(self, processed_inputs: Dict[str, Any]) -> Dict[str, Any]:
        return await self.model.generate(**processed_inputs)
        
    async def postprocess(self, raw_outputs: Dict[str, Any]) -> Dict[str, Any]:
        # Custom postprocessing logic
        return {
            'processed_output': raw_outputs['text'],
            'metadata': raw_outputs.get('metadata', {})
        }

# Example usage
model = CustomModelWrapper(
    existing_model,
    ModelConfig(max_tokens=1000)
)
result = await model.execute({'prompt': 'Test input'})

Real-world Example: Document Classification

This implementation demonstrates a complete document classification system using PydanticAI. It includes data preprocessing, model implementation, and classification logic with proper error handling and validation.

from pydanticai import BaseAIModel, Field
from typing import List, Dict
import numpy as np

class DocumentClassifier(BaseAIModel):
    class Input:
        text: str = Field(..., description="Document text")
        categories: List[str] = Field(..., description="Available categories")
    
    class Output:
        category: str
        confidence: float
        embeddings: Dict[str, float]
    
    async def preprocess(self, text: str) -> str:
        # Clean and normalize text
        text = text.lower().strip()
        return " ".join(text.split())
    
    async def compute_embeddings(self, text: str) -> Dict[str, float]:
        # Compute document embeddings
        tokens = text.split()
        embedding = np.random.rand(768)  # Example embedding
        return {f"dim_{i}": float(v) for i, v in enumerate(embedding)}
    
    async def execute(self, input: Input) -> Output:
        processed_text = await self.preprocess(input.text)
        embeddings = await self.compute_embeddings(processed_text)
        
        # Simulate classification
        category_scores = {
            cat: np.random.random() 
            for cat in input.categories
        }
        best_category = max(category_scores.items(), key=lambda x: x[1])
        
        return self.Output(
            category=best_category[0],
            confidence=float(best_category[1]),
            embeddings=embeddings
        )

# Usage example
classifier = DocumentClassifier()
result = await classifier.execute(
    text="Sample document for classification",
    categories=["tech", "science", "business"]
)

Results for: Document Classification

# Example classification results
{
    "input_text": "Sample document for classification",
    "classification_results": {
        "category": "tech",
        "confidence": 0.875,
        "processing_time_ms": 234,
        "embedding_dimensions": 768
    },
    "performance_metrics": {
        "accuracy": 0.92,
        "precision": 0.89,
        "recall": 0.87,
        "f1_score": 0.88
    }
}

Advanced Token Management

PydanticAI implements sophisticated token management to optimize model interactions and control costs. The system provides granular control over token usage while maintaining response quality through dynamic adjustment of context windows.

from pydanticai import TokenManager, TokenConfig
from typing import Optional, List

class AdvancedTokenManager(TokenManager):
    def __init__(self, config: TokenConfig):
        self.config = config
        self.token_usage = 0
    
    async def calculate_tokens(self, text: str) -> int:
        # Simple approximation (in practice, use proper tokenizer)
        return len(text.split())
    
    async def optimize_context(
        self, 
        text: str, 
        max_tokens: int
    ) -> str:
        current_tokens = await self.calculate_tokens(text)
        if current_tokens <= max_tokens:
            return text
            
        words = text.split()
        return " ".join(words[:max_tokens])
    
    async def track_usage(self, tokens_used: int):
        self.token_usage += tokens_used
        if self.token_usage >= self.config.max_daily_tokens:
            raise ValueError("Daily token limit exceeded")

# Example usage
manager = AdvancedTokenManager(
    TokenConfig(max_daily_tokens=100000)
)
text = "Long document for processing..."
optimized = await manager.optimize_context(text, 500)
await manager.track_usage(len(optimized.split()))

Real-world Example: Text Summarization Pipeline

This implementation showcases a complete text summarization system using PydanticAI's pipeline architecture. It demonstrates proper handling of long documents with token management and quality control.

from pydanticai import Pipeline, QualityCheck
from typing import List, Dict, Optional

class SummarizationPipeline(Pipeline):
    class Config:
        max_chunk_size: int = 1000
        min_quality_score: float = 0.7
    
    async def split_document(self, text: str) -> List[str]:
        words = text.split()
        chunks = []
        for i in range(0, len(words), self.Config.max_chunk_size):
            chunk = " ".join(words[i:i + self.Config.max_chunk_size])
            chunks.append(chunk)
        return chunks
    
    async def summarize_chunk(self, chunk: str) -> str:
        # Implement actual summarization logic
        return f"Summary of: {chunk[:100]}..."
    
    async def merge_summaries(self, summaries: List[str]) -> str:
        return " ".join(summaries)
    
    async def check_quality(self, summary: str) -> float:
        # Implement quality metrics
        return len(summary.split()) / 100.0
    
    async def execute(self, text: str) -> Dict[str, Any]:
        chunks = await self.split_document(text)
        summaries = []
        
        for chunk in chunks:
            summary = await self.summarize_chunk(chunk)
            quality = await self.check_quality(summary)
            
            if quality >= self.Config.min_quality_score:
                summaries.append(summary)
        
        final_summary = await self.merge_summaries(summaries)
        return {
            'summary': final_summary,
            'chunks_processed': len(chunks),
            'quality_score': await self.check_quality(final_summary)
        }

# Usage
pipeline = SummarizationPipeline()
result = await pipeline.execute("Very long document to summarize...")

Results for: Text Summarization Pipeline

# Example summarization results
{
    "input_length": 15234,
    "output_length": 1543,
    "processing_metrics": {
        "chunks_processed": 16,
        "average_chunk_quality": 0.85,
        "total_processing_time_ms": 3456
    },
    "quality_metrics": {
        "coherence_score": 0.92,
        "relevance_score": 0.88,
        "overall_quality": 0.90
    },
    "token_usage": {
        "input_tokens": 3808,
        "output_tokens": 385,
        "total_cost": 0.0427
    }
}

Advanced Error Recovery

PydanticAI implements sophisticated error recovery mechanisms that enable graceful degradation and automatic retry strategies for production environments. The system maintains state consistency while handling various failure scenarios.

from pydanticai import ErrorHandler, RetryStrategy
from typing import Optional, Callable
import asyncio

class ResilientErrorHandler(ErrorHandler):
    def __init__(
        self, 
        max_retries: int = 3, 
        backoff_factor: float = 1.5
    ):
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor
        
    async def handle_error(
        self, 
        error: Exception, 
        context: dict,
        retry_func: Callable
    ) -> Optional[dict]:
        retries = 0
        last_error = error
        
        while retries < self.max_retries:
            try:
                # Exponential backoff
                await asyncio.sleep(
                    self.backoff_factor ** retries
                )
                return await retry_func(**context)
            except Exception as e:
                last_error = e
                retries += 1
        
        return {
            'error': str(last_error),
            'context': context,
            'retries': retries
        }

# Usage example
handler = ResilientErrorHandler()
result = await handler.handle_error(
    error=ValueError("Model timeout"),
    context={'prompt': 'Test prompt'},
    retry_func=model.generate
)

Model Chain Composition

PydanticAI provides a powerful model chain composition system that enables the creation of complex AI workflows. This implementation demonstrates how to build and execute model chains while maintaining type safety and error handling.

from pydanticai import ModelChain, ChainNode
from typing import List, Dict, Any

class AIModelChain(ModelChain):
    class Node(ChainNode):
        def __init__(
            self, 
            model: Any, 
            name: str
        ):
            self.model = model
            self.name = name
            self.next: Optional['Node'] = None
    
    def __init__(self):
        self.head: Optional[Node] = None
        self.results: Dict[str, Any] = {}
    
    def add_model(self, model: Any, name: str) -> 'AIModelChain':
        node = self.Node(model, name)
        if not self.head:
            self.head = node
        else:
            current = self.head
            while current.next:
                current = current.next
            current.next = node
        return self
    
    async def execute(self, initial_input: Dict[str, Any]) -> Dict[str, Any]:
        current = self.head
        current_input = initial_input
        
        while current:
            try:
                result = await current.model.execute(current_input)
                self.results[current.name] = result
                current_input = result
                current = current.next
            except Exception as e:
                raise ChainExecutionError(
                    f"Error in {current.name}: {str(e)}"
                )
        
        return self.results

# Usage example
chain = (AIModelChain()
    .add_model(TextPreprocessor(), "preprocessor")
    .add_model(Classifier(), "classifier")
    .add_model(Summarizer(), "summarizer"))

results = await chain.execute({"text": "Raw input text"})

Additional Resources

More in Python for ML and AI

  1. Calculating the Astronomical Cost of an ISS Wedding Using Python5 min

    This slide introduces the unique problem of estimating the cost of hosting a 200-person wedding aboard the International Space Station (ISS). It outlines the...

  2. Building Linear Discriminant Analysis without Libs in Python10 min

    Linear Discriminant Analysis (LDA) is a dimensionality reduction and classification technique used in machine learning and statistics. It aims to find a...

  3. Calculating the Time to Consume 2400 Donuts Using Python5 min

    This slide introduces our intriguing problem: How long would it take someone to eat 2400 donuts? We'll approach this seemingly whimsical question with...

  4. Building Effective AI with Domain Expertise and Python8 min

    Domain expertise plays a vital role in building effective AI systems, but it's one of several critical components. Understanding the problem domain helps in...

  5. Calculating Z-Scores with Python9 min

    Z-score is a statistical measure that indicates how many standard deviations an element is from the mean. It's widely used in data analysis and machine...

  6. Building an OLS Regression Algorithm in Python10 min

    Ordinary Least Squares (OLS) regression is a fundamental statistical method used to model the relationship between a dependent variable and one or more...