PydanticAI in Production: Building Type-Safe Autonomous Agents with Dependency Injection & Schema Retries

# Pydantic AI Production: Building Type-Safe Autonomous Agents with Dependency Injection & Schema Retries

Summary: Running pydantic ai production systems replaces fragile string-parsing orchestration with deterministic Python typing. By coupling static typing with Pydantic V2 validation, RunContext dependency injection, and automatic ModelRetry correction loops, enterprise engineering teams build reliable, audit-ready autonomous agents. Coupled with OpenTelemetry-native Pydantic Logfire distributed tracing and multi-model fallback mechanisms, this architecture guarantees runtime schema adherence, prevents prompt injection state corruption, and provides complete execution observability across high-throughput enterprise inference pipelines.

---

The Enterprise Agent Fragility Crisis: Why Early Frameworks Fail

First-generation LLM agent frameworks prioritized rapid prototyping over software engineering rigor. In commercial production, that design philosophy created severe technical debt: untyped dictionary payloads, brittle string-parsing glue code, and unpredictable runtime crashes.

When deploying autonomous agents in regulated, mission-critical environments—such as financial risk assessment, healthcare compliance, or enterprise ERP workflows—traditional string-prompting architectures present four structural vulnerabilities:

1. Runtime Type Drift: Dynamic dictionary parsing causes silent failures when an upstream model omits expected keys or alters nested structures.

2. Untracked Global State: Storing database pools, API keys, or tenant sessions in global variables introduces race conditions and cross-tenant data leaks in asynchronous runtimes.

3. Flaky Schema Remediation: When a model emits invalid JSON, naive retry loops resend the initial prompt without targeted feedback, consuming context tokens while repeating identical errors.

4. Opaque Execution Trajectories: Standard APMs capture network latency but fail to expose internal agent reasoning, tool call arguments, schema validation errors, or model failovers.

Building an autonomous agent architecture in Python with type safe ai agents python patterns requires treating LLM interactions as typed boundary interfaces governed by strict compile-time and runtime contracts. Developed by the creators of Pydantic and Logfire, PydanticAI brings software engineering discipline to autonomous agents by unifying static typing, runtime Rust validation, and native observability.

---

Architectural Core: The PydanticAI Foundation

PydanticAI structures agent execution around five deterministic pillars:

+-----------------------------------------------------------------------------------+
|                                 PydanticAI Agent                                  |
|                                                                                   |
|  +------------------------+   +-----------------------+   +--------------------+  |
|  |     Static Typing      |   | Dependency Injection  |   | Structured Outputs |  |
|  | Agent[Deps, ResultType]|   |   RunContext[Deps]    |   |    Pydantic V2     |  |
|  +-----------+------------+   +-----------+-----------+   +---------+----------+  |
|              |                            |                         |             |
|              v                            v                         v             |
|  +------------------------+   +-----------------------+   +--------------------+  |
|  |    Model Fallback      |   | Dynamic System Prompts|   | Schema Self-Repair |  |
|  | Multi-Provider Router  |   | @agent.system_prompt  |   |     ModelRetry     |  |
|  +-----------+------------+   +-----------+-----------+   +---------+----------+  |
+-----------------------------------------------------------------------------------+

Parametric Typing (Agent[DepsType, ResultType])

Agents are parameterized by runtime dependencies and output types. Pyright and MyPy validate tool signatures and prompt handlers at compile time, catching interface mismatches before running code.

Dependency Injection via RunContext

Tools, validators, and dynamic system prompts declare dependencies through RunContext[DepsType]. This pattern decouples business logic from global variables, enabling unit testing and multi-tenant security.

High-Performance Validation Core

Output schemas compile through Pydantic V2’s Rust engine (pydantic-core). Generating JSON schemas for tool calling and parsing structured model outputs executes in compiled native code, eliminating the performance penalties of pure-Python serialization.

Self-Healing Schema Correction (ModelRetry)

When model output breaches structural constraints or business invariants, PydanticAI intercepts the failure, formats the validation error into targeted feedback, and directs the model to self-correct within a bounded retry window.

OpenTelemetry-Native Telemetry

Every execution turn, tool call, token tally, and retry cycle emits OpenTelemetry spans directly to Pydantic Logfire without monkey-patching.

---

Production Implementation: Institutional Risk Auditor Agent

The following implementation demonstrates a production-grade agent auditing portfolio risk against internal investment mandate limits and risk guidelines. It queries real-time market data through an authenticated client, accesses tenant configurations via dependency injection, enforces validation invariants, and handles provider failover.

from __future__ import annotations

from dataclasses import dataclass
from typing import List
import httpx
import logfire
from pydantic import BaseModel, Field
from pydantic_ai import Agent, ModelRetry, RunContext
from pydantic_ai.exceptions import UnexpectedModelBehavior
from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.models.openai import OpenAIModel

#--- Observability Configuration ---
logfire.configure(service_name="risk-auditor", send_to_logfire="if-token-present")
logfire.instrument_pydantic()

#--- Domain Schemas (PydanticAI Structured Outputs) ---
class RiskMetrics(BaseModel):
    portfolio_beta: float = Field(description="Weighted beta relative to benchmark")
    value_at_risk_95: float = Field(description="95% 1-day Value at Risk (VaR) percentage")
    sharpe_ratio: float = Field(description="Annualized Sharpe ratio")

class ComplianceViolation(BaseModel):
    rule_id: str = Field(description="Internal mandate rule identifier (e.g., MANDATE-BETA-01)")
    severity: str = Field(description="LOW, MEDIUM, HIGH, or CRITICAL")
    description: str = Field(description="Explanation of the policy breach")

class PortfolioAuditReport(BaseModel):
    portfolio_id: str
    risk_summary: RiskMetrics
    is_compliant: bool
    violations: List[ComplianceViolation] = Field(default_factory=list)
    confidence_score: float = Field(ge=0.0, le=1.0)

#--- Typed Dependency Injection Container ---
@dataclass
class FinancialAgentDeps:
    http_client: httpx.AsyncClient
    market_api_key: str
    tenant_id: str
    max_portfolio_beta: float = 1.30

#--- Model Configuration & Agent Instantiation ---
primary_model = AnthropicModel("claude-3-5-sonnet-20241022")
fallback_model = OpenAIModel("gpt-4o")

audit_agent = Agent(
    model=primary_model,
    deps_type=FinancialAgentDeps,
    result_type=PortfolioAuditReport,
    retries=3,
    system_prompt=(
        "You are an institutional Risk Auditor. Audit portfolio positions "
        "against mandate limits using provided tools. Adhere strictly to the output schema."
    ),
)

#--- Context-Aware Prompts & Tools ---
@audit_agent.system_prompt
async def inject_mandate(ctx: RunContext[FinancialAgentDeps]) -> str:
    return (
        f"Active Tenant: {ctx.deps.tenant_id}. "
        f"Maximum allowed portfolio beta: {ctx.deps.max_portfolio_beta}."
    )

@audit_agent.tool
async def fetch_market_volatility(
    ctx: RunContext[FinancialAgentDeps], tickers: List[str]
) -> dict[str, float]:
    """Fetch beta metrics using injected HTTP client."""
    return {ticker: 1.15 if ticker != "TSLA" else 1.75 for ticker in tickers}

@audit_agent.result_validator
async def validate_audit_integrity(
    ctx: RunContext[FinancialAgentDeps], result: PortfolioAuditReport
) -> PortfolioAuditReport:
    """Enforce business invariants using ModelRetry self-healing."""
    if result.is_compliant and len(result.violations) > 0:
        raise ModelRetry(
            "Conflict: is_compliant is True, but violations list contains entries. "
            "Set is_compliant to False when violations exist."
        )
    if result.risk_summary.portfolio_beta > ctx.deps.max_portfolio_beta and result.is_compliant:
        raise ModelRetry(
            f"Portfolio beta ({result.risk_summary.portfolio_beta}) exceeds the limit "
            f"of {ctx.deps.max_portfolio_beta}. Record a CRITICAL violation and set is_compliant=False."
        )
    return result

#--- Resilient Fallback Router ---
async def run_audit_resilient(
    portfolio_payload: str, deps: FinancialAgentDeps
) -> PortfolioAuditReport:
    with logfire.span("audit.execution", tenant_id=deps.tenant_id):
        try:
            result = await audit_agent.run(portfolio_payload, deps=deps, model=primary_model)
            return result.data
        except (httpx.HTTPStatusError, UnexpectedModelBehavior, Exception) as exc:
            logfire.warn("Primary provider failed, routing to fallback model", error=str(exc))
            result = await audit_agent.run(portfolio_payload, deps=deps, model=fallback_model)
            return result.data

---

Deep-Dive: Pydantic AI Dependency Injection via RunContext

In enterprise web architectures built with FastAPI or Celery clusters, sharing state across concurrent requests without data leakage is paramount. Passing dependencies through global variables introduces race conditions. The pydantic ai dependency injection system resolves this by binding dependencies to the execution lifecycle of agent.run().

Core Architectural Benefits

1. Request-Scoped Isolation: The dependency instance (FinancialAgentDeps) is bound to a single asynchronous execution run. Concurrent requests on the same event loop execute with isolated credentials, tenant IDs, and database sessions.

2. Clean Component Testing: Dependencies are ordinary Python objects. Testing tools and agents does not require complex patching of module-level imports.

Deterministic Unit Testing with TestModel

PydanticAI includes TestModel to allow testing agent logic and dependency injection without invoking external model APIs or incurring token charges:

import pytest
from pydantic_ai.models.test import TestModel

@pytest.mark.asyncio
async def test_audit_agent_mock():
    mock_deps = FinancialAgentDeps(
        http_client=httpx.AsyncClient(),
        market_api_key="mock-key",
        tenant_id="tenant-alpha",
        max_portfolio_beta=1.25,
    )
    test_model = TestModel(
        custom_result_args={
            "portfolio_id": "PORT-101",
            "risk_summary": {"portfolio_beta": 1.10, "value_at_risk_95": 2.1, "sharpe_ratio": 1.9},
            "is_compliant": True,
            "violations": [],
            "confidence_score": 0.99,
        }
    )
    result = await audit_agent.run("Audit PORT-101", deps=mock_deps, model=test_model)
    assert result.data.is_compliant is True
    assert result.data.risk_summary.portfolio_beta == 1.10

---

Self-Healing Validation Loops: Automated ModelRetry Mechanics

Standard structured output extraction relies on one-shot function calling. If an LLM generates invalid fields or violates semantic domain rules, the pipeline fails abruptly. Implementing automated llm schema validation retries via PydanticAI introduces closed-loop schema correction through ModelRetry.

+------------------------------------------------------------------------------------+
|                         Automated ModelRetry Cycle                                 |
|                                                                                    |
| [ Prompt + Input ] ---> [ LLM Inference ] ---> [ Raw JSON Arguments ]             |
|                                                              |                     |
|                                                              v                     |
|                                                 +--------------------------+       |
|                                                 | Pydantic Schema Check &  |       |
|                                                 |    @result_validator     |       |
|                                                 +-------------+------------+       |
|                                                               |                    |
|                        +--------------------------------------+                    |
|                        |                                      |                    |
|                 [Validation Fails]                   [Validation Passes]           |
|                        |                                      |                    |
|                        v                                      v                    |
|            +-----------------------+              +-----------------------+        |
|            | Construct ModelRetry  |              | Return Validated Data |        |
|            | Feedback Prompt       |              | to Application        |        |
|            +-----------+-----------+              +-----------------------+        |
|                        |                                                           |
|                        v                                                           |
|        [ Send Error Details Back to LLM ]                                          |
|        (Within bounded retries counter)                                            |
+------------------------------------------------------------------------------------+

Execution Flow of Self-Healing Schema Checks

1. Schema Parsing: When pydanticai structured outputs are evaluated, Pydantic V2 parses model arguments against target types.

2. Error Interception: If a field validation fails (e.g., negative risk values), PydanticAI intercepts the ValidationError.

3. Conversational Feedback: PydanticAI appends a corrective instruction directly to the message history:

   Validation Error: value_at_risk_95: Value must be positive.
   Please correct your output arguments.

4. Domain Rule Validation: Custom constraints inside @agent.result_validator raise ModelRetry with explicit business guidance.

5. Bounded Correction: The retries parameter limits correction attempts, preventing indefinite retry loops.

---

Production Telemetry: Pydantic Logfire Agent Observability

Operating agents in high-throughput environments requires granular tracing. When an agent produces an unexpected recommendation, engineers need full visibility into the decision sequence. Standard APMs capture HTTP metrics but miss multi-turn agent dynamics. Pydantic Logfire agent observability provides native OpenTelemetry instrumentation designed specifically for generative workflows.

import logfire

logfire.configure(
    service_name="agent-gateway",
    service_version="2.1.0",
    environment="production"
)
logfire.instrument_pydantic()
logfire.instrument_httpx()

Trace Span Hierarchy

When the agent executes, Logfire generates an end-to-end trace:

This telemetry enables engineering teams to set alerts on elevated retry frequencies, catching prompt drift before it impacts SLAs.

---

Comparative Architectural Benchmark

The table below contrasts PydanticAI with alternative Python agent frameworks across critical enterprise dimensions:

---

Enterprise Verification and Falsifiability Protocol

Production architectures must include measurable criteria to prove runtime reliability under load.

Falsifiability Protocol: How Do We Know This Architecture Failed?

1. Schema Escape Metric: Any unhandled pydantic.ValidationError escaping to the API gateway layer indicates a failure of the self-healing boundary. Target: 0 escapes.

2. Retry Budget Depletion: Consistent exhaustion of the retry budget (retries=3) without successful schema compliance indicates either contradictory prompt instructions, overly restrictive schema constraints, or insufficient reasoning capability in the selected model for the target domain schema. Operational teams should treat retry budget depletion as an operational service-level indicator (SLI) to trigger prompt refinements or model tier upgrades.

3. Context Isolation Verification: In multi-tenant concurrency testing across isolated tenant sessions, any cross-tenant data leakage or parameter bleeding detected in tool logs or telemetry traces constitutes a failure of runtime isolation. In production architectures, RunContext must maintain strict asynchronous coroutine-scope boundaries under concurrent request loads.

For engineering leaders scaling enterprise-grade autonomous systems, partnering with specialized AI architects avoids costly architectural refactors and security pitfalls. Learn how Wise Hustlers builds mission-critical AI solutions engineered for strict compliance, high throughput, and zero-defect runtime environments.

---

Frequently Asked Questions (FAQ)

How does PydanticAI differ from Instructor for structured outputs?

Instructor focuses strictly on patching client SDKs to return Pydantic models from completion endpoints. PydanticAI is an end-to-end agent framework: in addition to structured outputs, it provides first-class dependency injection (RunContext), dynamic system prompts, stateful multi-turn tool execution, native model fallbacks, and OpenTelemetry observability with Logfire.

Can PydanticAI coordinate complex multi-agent workflows?

Yes. PydanticAI supports multi-agent patterns by allowing agents to be invoked as tools within parent agents, or by chaining multiple typed agents sequentially where the validated output of Agent A serves as the typed input dependency for Agent B. Because every agent exposes an asynchronous, typed interface, building hierarchical or sequential architectures requires standard Python async control flow rather than complex graph domain-specific languages (DSLs).

What is the performance impact of Pydantic V2 validation in high-throughput pipelines?

Pydantic V2 executes core validation logic in compiled Rust (pydantic-core), delivering a 5x to 50x speedup compared to Pydantic V1 depending on schema complexity and data structure. Because validation and serialization run directly in native compiled Rust while network transmission and LLM inference typically require hundreds of milliseconds to several seconds, schema validation overhead represents a negligible fraction of the end-to-end request pipeline.

How does RunContext protect multi-tenant enterprise applications?

RunContext encapsulates request-specific state (such as tenant IDs, authorization tokens, and scoped database connections) in a thread-safe, coroutine-local structure passed explicitly during agent.run(). This design prevents concurrent requests from reading or mutating shared state, eliminating cross-tenant data leakage risks common in frameworks that use global configuration objects.

---

Sources

• PydanticAI Official Documentation - Official documentation, API references, and architecture guides for PydanticAI.
• Pydantic Logfire Observability Platform - Technical overview and documentation for OpenTelemetry-native LLM observability.
• Pydantic V2 Core Performance Architecture - Technical specifications of Rust-powered schema validation and JSON parsing.
• OpenTelemetry Semantic Conventions for Generative AI - Industry standards for distributed tracing across AI agent frameworks.