Skip to main content

On This Page

Building Reliable Agentic Workflows with PydanticAI and Strict Schemas

3 min read
Share

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