Building Reliable Agentic Workflows with PydanticAI and Strict Schemas
These articles are AI-generated summaries. Please check the original sources for full details.
A Coding Implementation to Build Bulletproof Agentic Workflows with PydanticAI Using Strict Schemas, Tool Injection, and Model-Agnostic Execution
PydanticAI enables engineers to move beyond fragile chatbot patterns by enforcing strict typed outputs at every step of an agentic workflow. This implementation demonstrates a production-ready system that prioritizes reliability over best-effort generation through schema validation and tool injection.
Why This Matters
Traditional LLM interactions often suffer from non-deterministic outputs that break downstream systems, leading to high failure rates in automated pipelines. By using PydanticAI’s strict schema enforcement and dependency injection, developers can guarantee that agents interact safely with external databases and business logic, significantly reducing the maintenance cost of agentic AI. This move from best-effort generation to enforced contracts is critical for enterprise workflows where malformed data can halt execution.
Key Insights
- Strict Schema Enforcement: PydanticAI uses Pydantic’s BaseModel to define clear response contracts, preventing malformed outputs from propagating through workflows.
- Dependency Injection: The SupportDeps dataclass allows for injecting runtime dependencies like SQLite connections and tenant-specific policies directly into the agent context.
- Model-Agnostic Execution: The framework supports swapping underlying models, such as GPT-4o-mini to GPT-4o, without modifying core tool logic or response schemas.
- Self-Correction via Output Validators: The @agent.output_validator decorator enables agents to retry and correct decisions that violate business rules, such as missing ticket IDs for updates.
- Async-First Architecture: The implementation leverages Python’s async/await patterns for efficient, notebook-friendly execution of complex support triage tasks.
Working Examples
Environment setup and core library imports for the PydanticAI workflow.
!pip -q install "pydantic-ai-slim[openai]" pydantic
import os, json, sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Literal, Optional, List
from pydantic import BaseModel, Field, field_validator
from pydantic_ai import Agent, RunContext, ModelRetry
Defining the strict TicketDraft schema with validation rules for length and content.
Priority = Literal["low", "medium", "high", "critical"]
ActionType = Literal["create_ticket", "update_ticket", "query_ticket", "list_open_tickets", "no_action"]
class TicketDraft(BaseModel):
title: str = Field(..., min_length=8, max_length=120)
customer: str = Field(..., min_length=2, max_length=60)
priority: Priority
category: Literal["billing", "bug", "feature_request", "security", "account", "other"]
description: str = Field(..., min_length=20, max_length=1000)
expected_outcome: str = Field(..., min_length=10, max_length=250)
Agent construction using model-agnostic strings and dependency-injected tools.
def build_agent(model_name: str) -> Agent[SupportDeps, AgentDecision]:
agent = Agent(
f"openai:{model_name}",
output_type=AgentDecision,
output_retries=2,
instructions="You are a production support triage agent."
)
@agent.tool
def create_ticket(ctx: RunContext[SupportDeps], ticket: TicketDraft) -> int:
deps = ctx.deps
return seed_ticket(deps.db, deps.tenant, ticket, status="open")
return agent
Practical Applications
- Production Support Triage: An agent automatically categorizes tickets and queries database state using injected tools for companies like Acme Corp.
- Pitfall: Relying on unstructured text prompts for database updates, which leads to SQL errors or invalid state transitions if the LLM hallucinates IDs.
- Feature Request Automation: Scaling feedback loops by allowing agents to draft tickets directly into a system of record while validating priority levels.
- Pitfall: Ignoring output validation, resulting in ‘critical’ tickets being created for tenants that explicitly forbid that priority level via local policy.
References:
Continue reading
Next article
Build a Task Tracker MCP Server in Python with FastMCP
Related Content
Design Tool-Driven Agentic Workflows for Deterministic Route Optimization
Learn to build a production-style Route Optimizer Agent using LangChain and Pydantic that computes precise ETAs and distances deterministically instead of hallucinating results.
Building Production-Grade Support Pipelines with Griptape and Agentic Reasoning
Learn how to build an automated support pipeline using Griptape to sanitize PII, categorize issues, and assign SLAs with deterministic tools before using GPT-4 for synthesis.
Building Production-Ready Agentic Workflows with AgentScope and ReAct Agents
Learn to build production-ready AgentScope workflows using ReAct agents, custom toolkits, and Pydantic for structured outputs. This tutorial demonstrates how to orchestrate multi-agent debates and concurrent analysis pipelines using OpenAI models to achieve high-fidelity reasoning and automated tool execution for enterprise-grade AI applications.