PydanticAI Tutorial: Type-Safe Agent Outputs
Source: mortalapps.com- PydanticAI is a type-safe agent framework built by the creators of Pydantic to guarantee structured LLM outputs.
- It solves the problem of non-deterministic JSON schemas and validation failures in production LLM integrations.
- Crucial for regulated environments (finance, healthcare) where schema drift or malformed JSON violates compliance.
- Demonstrates how to build validated agent loops, implement self-correction, and integrate with LangGraph pipelines.
Why This Matters
In highly regulated industries such as finance, healthcare, and legal tech, non-deterministic outputs from Large Language Models (LLMs) present a severe operational risk. Standard LLM APIs return unstructured text or loosely validated JSON that can easily drift, break downstream parsers, or violate strict compliance schemas. This pydanticai tutorial demonstrates how to eliminate this class of runtime errors by enforcing strict, compile-time and runtime type safety on agent outputs.
Historically, developers relied on manual regex parsing, fragile post-processing scripts, or basic JSON-mode prompts to coerce LLMs into structured formats. When these methods fail, systems crash, corrupted data enters database transactions, and automated workflows halt. PydanticAI was designed specifically to address this vulnerability by unifying Pydantic's industry-standard data validation with LLM orchestration. By defining agent outputs as native Pydantic models, developers gain access to automatic schema generation, runtime type checking, and structured error recovery out of the box.
Ignoring type safety in production agent pipelines leads to cascading failures, silent data corruption, and high latency due to unmanaged retry loops. While alternatives like raw LangChain or custom OpenAI structured outputs exist, they often lack deep validation lifecycle hooks, local testing capabilities, or seamless integration with broader state-machine frameworks. PydanticAI provides a lightweight, robust, and highly testable abstraction layer that guarantees every LLM response strictly conforms to your enterprise domain models before any downstream action is executed.
Core Concepts
To build robust, type-safe agent systems, you must understand the core vocabulary and components that PydanticAI introduces:
- Agent: The primary orchestrator in PydanticAI, combining an LLM model, system prompts, and dependency injection.
- Result Type: The expected Pydantic model that the agent must return. The framework automatically translates this model into a JSON schema for the LLM.
- Run Context: A generic type-safe container passed to system prompts and tools, enabling dependency injection without global state.
- Model Validator: A Pydantic validator that runs on the structured output *after* the LLM returns it, allowing custom business logic validation.
- Self-Correction Loop: The automatic process where validation failures are fed back to the LLM with error messages, prompting a corrected output.
Output Validation Strategies
| Strategy | Type Safety | Latency Overhead | Error Recovery |
|---|---|---|---|
| Raw JSON Mode | Low (JSON only) | Low | Manual |
| OpenAI Structured Outputs | Medium (JSON Schema) | Low | Hard Fail |
| PydanticAI Validation | High (Python + Schema) | Medium (on retry) | Automatic Self-Correction |
How It Works
PydanticAI guarantees type safety through a structured execution lifecycle that intercepts, validates, and corrects LLM outputs before returning them to your application.
Phase 1: Schema Generation and Injection
When you initialize a PydanticAI agent with a defined result_type, the framework inspects the Pydantic model and generates a standard JSON schema. This schema is injected directly into the LLM payload as a tool definition or structured output instruction, depending on the underlying provider's capabilities.
Phase 2: Execution and Dependency Injection
When the agent runs, it resolves any dynamic dependencies defined in the RunContext. These dependencies (such as database pools or HTTP clients) are injected into system prompts and tool definitions, ensuring that the LLM has access to real-time, validated context.
Phase 3: Parsing and Validation
The LLM returns a raw string response. PydanticAI intercepts this response, parses it into JSON, and attempts to instantiate the target Pydantic model.
Phase 4: Self-Correction Loop (Failure Path)
If the LLM's response violates the schema or fails custom @model_validator checks, PydanticAI does not crash. Instead, it captures the ValidationError, formats the error trace into a readable prompt, appends it to the conversation history, and re-invokes the LLM. This self-correction loop continues until the output passes validation or the configured max_retries limit is reached, at which point a ModelRetryError is raised.
Architecture
The architecture consists of a Client Application that invokes a LangGraph State Machine. Within the state machine, specific nodes route execution to a PydanticAI Agent. The PydanticAI Agent acts as the orchestrator, utilizing a Dependency Injector to populate the RunContext with runtime connections (like databases). The Agent serializes the target Pydantic model into a JSON schema and sends it alongside the user query to the LLM Provider. The LLM Provider returns a JSON payload, which flows directly into the Validation Engine. If validation succeeds, the parsed Pydantic model is returned to the LangGraph node to update the global state. If validation fails, the Validation Engine formats the error, feeds it back to the LLM Provider for a correction attempt, and loops until success or the retry limit is breached.
Defining Strict Schemas for Regulated Environments
In regulated environments, schemas must do more than enforce basic types like str or int. They must enforce domain-specific constraints. Pydantic allows you to embed these constraints directly into the model definition using Field metadata. These constraints are compiled into the JSON schema sent to the LLM, forcing the model to respect boundaries during generation.
from pydantic import BaseModel, Field
class AuditReport(BaseModel):
compliance_score: float = Field(
description="The calculated compliance score of the transaction.",
ge=0.0,
le=1.0
)
flagged_violations: list[str] = Field(
min_length=1,
description="List of specific regulatory violations identified."
)
Modern LLMs natively understand JSON schema properties like minimum, maximum, and minItems. By declaring these constraints in Pydantic, you reduce the reliance on natural language prompting to enforce boundaries.
The Self-Correction Loop and Retry Mechanics
When an LLM fails validation, PydanticAI's self-correction mechanism executes. This is not a simple retry; it is an informed correction loop. The framework appends the validation error message to the chat history as a user message, prompting the model to correct its specific mistake.
For example, if the LLM returns a compliance_score of 1.5 (violating the le=1.0 constraint), the validation engine catches the error and sends the following feedback to the LLM:
1 validation error for AuditReport
compliance_score
Input should be less than or equal to 1.0 [type=less_than_equal, input_value=1.5, input_type=float]
The LLM uses this feedback to adjust its output in the next turn. This loop is bounded by the max_retries parameter, preventing infinite loops and runaway token costs.
Dependency Injection and Context Management
Production agents must interact with external systems securely. PydanticAI provides a type-safe dependency injection system via Deps and RunContext. This design pattern avoids global state, making your agents thread-safe and easy to unit test.
from pydantic_ai import Agent, RunContext
class DatabaseClient:
def fetch_user_role(self, user_id: str) -> str:
return "admin"
agent = Agent('openai:gpt-4o', deps_type=DatabaseClient)
@agent.system_prompt
def define_system_prompt(ctx: RunContext[DatabaseClient]) -> str:
# Securely access the injected database client
role = ctx.deps.fetch_user_role("user_123")
return f"You are auditing transactions for a user with role: {role}."
Embedding PydanticAI inside LangGraph Pipelines
While PydanticAI excels at single-node type-safe execution, complex enterprise workflows often require multi-agent routing, state persistence, and human-in-the-loop approvals. LangGraph is the industry standard for these orchestrations.
To combine them, you embed PydanticAI agents inside LangGraph nodes. The LangGraph state holds the raw transaction data and the final validated Pydantic model. When the node executes, it runs the PydanticAI agent, which guarantees that the output written back to the LangGraph state is fully validated and typed.
Code Example
import os
import logging
from typing import List, Optional
from pydantic import BaseModel, Field, model_validator
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIModel
from langgraph.graph import StateGraph, END
# Configure logging for production observability
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("pydantic_ai_integration")
# Define the structured output schema with strict validation rules
class TransactionAudit(BaseModel):
transaction_id: str = Field(description="The unique identifier of the transaction.")
risk_score: float = Field(
description="A risk score between 0.0 (low) and 1.0 (high).",
ge=0.0,
le=1.0
)
flagged: bool = Field(description="Whether the transaction is flagged for review.")
reasons: List[str] = Field(
default=[],
description="List of reasons for the risk score. Must not be empty if flagged is True."
)
@model_validator(mode="after")
def validate_flagged_reasons(self) -> "TransactionAudit":
# Custom business logic validation: if flagged, reasons must be provided
if self.flagged and not self.reasons:
raise ValueError("Flagged transactions must contain at least one reason.")
return self
# Define dependencies to inject into the agent context
class AuditDeps:
def __init__(self, db_connection_string: str):
self.db_connection_string = db_connection_string
def log_audit_attempt(self, tx_id: str, status: str):
# Mock database logging operation
logger.info(f"Logging audit attempt for {tx_id} to DB: {status}")
# Initialize the LLM model using environment variables
api_key = os.environ.get("OPENAI_API_KEY", "mock-key-for-testing")
model = OpenAIModel("gpt-4o-mini", api_key=api_key)
# Initialize the PydanticAI Agent
audit_agent = Agent(
model,
result_type=TransactionAudit,
deps_type=AuditDeps,
system_prompt=(
"You are an expert financial compliance auditor. Analyze the transaction data "
"and produce a structured audit report. Be highly critical of high-value transfers."
)
)
# Define a dynamic system prompt modifier using dependencies
@audit_agent.system_prompt
def add_context(ctx: RunContext[AuditDeps]) -> str:
return f"Database context active: {ctx.deps.db_connection_string}."
# Define the LangGraph State
class AuditState(BaseModel):
transaction_data: dict
audit_report: Optional[TransactionAudit] = None
error: Optional[str] = None
# Define the LangGraph node that executes the PydanticAI Agent
async def run_audit_node(state: AuditState) -> dict:
db_url = os.environ.get("DATABASE_URL", "sqlite:///:memory:")
deps = AuditDeps(db_connection_string=db_url)
tx_id = state.transaction_data.get("id", "unknown")
deps.log_audit_attempt(tx_id, "STARTED")
try:
# Execute the agent with automatic self-correction retries
result = await audit_agent.run(
user_prompt=f"Analyze this transaction: {state.transaction_data}",
deps=deps,
usage_limits=None # Can be configured to limit tokens in production
)
deps.log_audit_attempt(tx_id, "SUCCESS")
return {"audit_report": result.data, "error": None}
except Exception as e:
logger.error(f"Audit agent failed: {str(e)}")
deps.log_audit_attempt(tx_id, f"FAILED: {str(e)}")
return {"error": str(e)}
# Build the LangGraph workflow
workflow = StateGraph(AuditState)
workflow.add_node("audit_transaction", run_audit_node)
workflow.set_entry_point("audit_transaction")
workflow.add_edge("audit_transaction", END)
app = workflow.compile()
INFO:pydantic_ai_integration:Logging audit attempt for tx_101 to DB: STARTED INFO:pydantic_ai_integration:Logging audit attempt for tx_101 to DB: SUCCESS (Returns an AuditState containing a validated TransactionAudit model instance)