The Concept
The Boundary Problem in AI Systems
Every AI system has a boundary: on one side is the messy, unstructured world (user input, LLM output, scraped text, JSON from external APIs); on the other side is your structured system (databases, typed code, business logic). At this boundary, assumptions break. A field that was always a string is suddenly null. A number comes back as a string. An array that always had 5 elements comes back empty. An LLM returns valid JSON 99% of the time — and garbage the other 1%.
Traditional Python code handles this with if checks scattered everywhere: if data is None, if isinstance(x, str), if len(arr) > 0. This is fragile, repetitive, and easy to forget. Pydantic replaces all of that with a data contract: you define what valid data looks like once, as a typed class, and every instance is automatically validated at creation time. If the data doesn't match the contract, you get a structured error message explaining exactly which fields failed and why — not a cryptic AttributeError 200 lines later.
Pydantic Models: Data Contracts for AI
A Pydantic model is a class that inherits from BaseModel and declares its fields with types and constraints. When you instantiate it, Pydantic validates every field against its declaration. If validation passes, you get an object with typed attributes (your editor will autocomplete them). If validation fails, you get a ValidationError listing every problem.
from pydantic import BaseModel, Field, field_validator
from typing import Literal, Optional
from datetime import datetime
class ChatMessage(BaseModel):
role: Literal["system", "user", "assistant"]
content: str = Field(..., min_length=1, max_length=32000)
class LLMResponse(BaseModel):
content: str
model: str
prompt_tokens: int = Field(ge=0)
completion_tokens: int = Field(ge=0)
finish_reason: Literal["stop", "length", "tool_calls", "content_filter"]
latency_ms: float = Field(ge=0)
class EmbeddingResult(BaseModel):
vector: list[float] = Field(..., min_length=1)
model: str
dimensions: int = Field(ge=1)
What happens when you pass {"text": "", "model": "gpt-5"} to a Pydantic model with text: str = Field(min_length=1) and model: Literal["gpt-4", "claude"]?
Pydantic validates all fields and raises a single ValidationError listing every failure. Each error includes the field path, the failed constraint, and the invalid input — making it easy to return structured error messages to API clients.
@field_validator("dimensions") @classmethod def check_dimensions(cls, v, info): vector = info.data.get("vector") if vector and v != len(vector): raise ValueError(f"dimensions ({v}) != len(vector) ({len(vector)})") return v
### Custom Validators for Domain Logic
from pydantic import BaseModel, field_validator, model_validator
class RAGQuery(BaseModel):
question: str = Field(..., min_length=3, max_length=2000)
top_k: int = Field(default=5, ge=1, le=50)
min_similarity: float = Field(default=0.7, ge=0.0, le=1.0)
filters: dict[str, str] = Field(default_factory=dict)
@field_validator("question")
@classmethod
def no_injection(cls, v: str) -> str:
suspicious = ["DROP TABLE", "DELETE FROM", "--", "/*"]
lower = v.upper()
for pattern in suspicious:
if pattern in lower:
raise ValueError(f"Potential injection pattern detected: {pattern}")
return v.strip()
@model_validator(mode="after")
def check_filter_keys(self):
allowed_filters = {"source", "date", "author", "category"}
invalid = set(self.filters.keys()) - allowed_filters
if invalid:
raise ValueError(f"Unknown filter keys: {invalid}. Allowed: {allowed_filters}")
return self
# Usage — validation happens automatically
try:
query = RAGQuery(question="What is RAG?", top_k=10, filters={"source": "docs"})
print(f"Valid query: {query.model_dump()}")
except ValueError as e:
print(f"Validation failed: {e}")
pytest: Testing AI Code
import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from myapp.rag import RAGEngine, RAGQuery
# Fixtures: reusable test setup
@pytest.fixture
def sample_query():
return RAGQuery(question="What is gradient descent?", top_k=3)
@pytest.fixture
def mock_embeddings():
return [[0.1, 0.2, 0.3] for _ in range(10)]
@pytest.fixture
def rag_engine(mock_embeddings):
engine = MagicMock(spec=RAGEngine)
engine.search.return_value = [
{"text": "Gradient descent is...", "score": 0.95},
{"text": "Optimization in ML...", "score": 0.88},
{"text": "Learning rates...", "score": 0.82},
]
return engine
# Parametrized tests: test many cases at once
@pytest.mark.parametrize("question,should_pass", [
("What is RAG?", True),
("How do transformers work?", True),
("ab", False), # too short
("DROP TABLE users;", False), # injection
("", False), # empty
])
def test_query_validation(question, should_pass):
if should_pass:
query = RAGQuery(question=question, top_k=5)
assert query.question == question.strip()
else:
with pytest.raises(ValueError):
RAGQuery(question=question, top_k=5)
# Mocking external APIs
@pytest.mark.asyncio
async def test_rag_search_calls_llm(rag_engine, sample_query):
results = rag_engine.search(sample_query)
assert len(results) == 3
assert all("text" in r and "score" in r for r in results)
rag_engine.search.assert_called_once_with(sample_query)
# Testing stochastic code with seeds
def test_embedding_determinism():
import numpy as np
np.random.seed(42)
emb1 = np.random.randn(768)
np.random.seed(42)
emb2 = np.random.randn(768)
np.testing.assert_array_equal(emb1, emb2)
# Markers for test organization
@pytest.mark.slow
@pytest.mark.integration
def test_full_rag_pipeline():
"""Only runs with: pytest -m 'not slow'"""
engine = RAGEngine(index_path="data/test_index")
result = engine.query("What is attention?")
assert len(result.sources) > 0
JSON Schema Generation
Pydantic models automatically generate JSON schemas — useful for API docs and LLM structured output:
class SearchRequest(BaseModel):
query: str
top_k: int = 5
filters: dict[str, str] = {}
# Generate JSON schema for API documentation
schema = SearchRequest.model_json_schema()
# This schema can also be passed to OpenAI's structured output:
# response = client.chat.completions.create(
# model="gpt-4o",
# response_format={"type": "json_schema", "json_schema": {...schema}},
# )