Deterministic Control Over Non-Deterministic AI: Why Production Agents Demand PydanticAI and LangGraph

In production services relying on raw string interpolation, unhandled execution faults stem directly from four output failure modes: dropped mandatory payload keys, string-to-float coercion failures, malformed JSON due to unescaped control characters, and hallucinated object keys. Parsing unstructured model outputs with standard string loaders guarantees runtime crashes downstream.

Large language models (LLMs) operate as probabilistic prediction engines and fail to adhere to API contracts by default. When an LLM alters a payload schema, standard application stacks fail without explicit deterministic boundaries. To isolate non-deterministic model outputs inside core services, we enforce four operational constraints:

• Strict input and output runtime contracts guaranteed by memory-level type validation.
• Bounded cyclic state graphs that enforce hard step ceilings to eliminate infinite recursion.
• Full OpenTelemetry trace visibility across model calls, schema validation checks, and graph state transitions.
• Deterministic dependency injection to isolate network, database, and infrastructure calls during testing.

1. Structural Schema Enforcement via PydanticAI

String-in and string-out execution paradigms shift data validation downstream, delegating error handling to runtime application code. We eliminate this vulnerability by embedding Python typing contracts directly into the LLM execution pipeline via PydanticAI.

By defining outputs as standard pydantic.BaseModel objects, we force model responses into typed schemas governed by runtime coercion, numerical range constraints, and field-level validators.

When an LLM response violates schema parameters, PydanticAI intercepts the validation failure at the framework layer, extracts the precise exception traceback, and appends that structural failure log into a reflection prompt context. The model then executes a self-correction pass before execution returns to the host application logic.

from typing import List
from pydantic import BaseModel, Field, field_validator
from pydantic_ai import Agent, RunContext

class RiskAssessment(BaseModel):
    account_id: str
    risk_score: float = Field(ge=0.0, le=1.0, description="Normalized risk score between 0.0 and 1.0")
    flagged_categories: List[str]
    action_required: bool

    @field_validator('flagged_categories')
    def validate_categories(cls, value: List[str]) -> List[str]:
        allowed = {'SUSPICIOUS_IP', 'VELOCITY_EXCEEDED', 'LARGE_TRANSFER', 'NEW_DEVICE'}
        invalid = set(value) - allowed
        if invalid:
            raise ValueError(f"Invalid categories detected: {invalid}. Must be subset of {allowed}")
        return value

risk_agent = Agent(
    'openai:gpt-4o',
    result_type=RiskAssessment,
    system_prompt="Analyze provided telemetry objects. Classify risk parameters and return structural JSON."
)

We encountered a significant framework defect during automated load testing: when an LLM repeatedly fails schema validation, appending multi-line tracebacks into message history causes context length to explode exponentially. At retry counts above 3, context inflation degrades inference latency from 850ms to over 4,200ms per call, while inducing attention drift where the model continuously repeats the identical validation error. We currently hack around this behavior by hard-capping model reflection at max_retries=3, forcing an execution abort and offloading the unparsed payload directly to an async dead-letter queue for offline analysis.

2. Deterministic State Machines and Control Flow via LangGraph

Managing multi-agent execution using imperative while loops or unstructured chaining creates unmaintainable software architectures. Imperative control loops obfuscate branching logic, conceal state mutations, complicate error recovery, and risk infinite recursive loops that consume compute budget.

We enforce execution bounds by shifting orchestration into explicit state graphs managed by LangGraph. By modeling agent interactions as state machines with defined nodes and edges, we maintain explicit control over cyclic execution, step limits, and state transitions.

We configure LangGraph execution pipelines by defining explicit state schemas and conditional routing logic across graph nodes:

from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    account_id: str
    risk_data: dict
    approval_required: bool

def router_node(state: AgentState) -> dict:
    return {"account_id": state["account_id"]}

def evaluate_risk_node(state: AgentState) -> dict:
    run_result = risk_agent.run_sync(f"Evaluate account {state['account_id']}")
    return {
        "risk_data": run_result.data.model_dump(),
        "approval_required": run_result.data.risk_score > 0.8
    }

def route_next(state: AgentState) -> str:
    if state["approval_required"]:
        return "human_approval"
    return "complete"

builder = StateGraph(AgentState)
builder.add_node("router", router_node)
builder.add_node("evaluate_risk", evaluate_risk_node)
builder.set_entry_point("router")
builder.add_edge("router", "evaluate_risk")
builder.add_conditional_edges(
    "evaluate_risk",
    route_next,
    {
        "human_approval": END,
        "complete": END
    }
)
graph = builder.compile()

During high-concurrency stress testing at 500 requests per second, we identified a second unresolved issue: LangGraph’s PostgreSQL checkpointer leaks connection pool slots when saving state histories with payloads exceeding 2MB. Because the framework serializes the complete conversation graph array on every node edge transition, high throughput causes database connection pool exhaustion within 120 seconds. We currently hack around this state bloat by patching the persistence layer with a custom serializer that truncates message context objects older than 5 execution steps prior to writing checkpoint records to disk.

3. OpenTelemetry Tracing and Observability via Logfire

Standard Application Performance Monitoring (APM) tools fail when applied to non-deterministic agent workflows. A basic HTTP trace captures overall function duration and status codes, but provides no visibility into system prompt construction, model token consumption breakdown, tool parameter schema validation, or retry cycles.

We address this visibility gap using Pydantic Logfire, an OpenTelemetry-native observability provider built to track structural execution paths without manual log injection across functions.

Logfire captures and displays four telemetry metrics natively:

• Inference latency per model execution broken down by prompt and completion generation time.
• Exact prompt and completion token counts mapped directly to execution costs.
• Field-level Pydantic validation failures and structural schema auto-correction loops.
• Node transition latencies and edge evaluations across LangGraph execution instances.

import logfire
from pydantic_ai import Agent

logfire.configure(
    service_name='enterprise-risk-engine',
    pydantic_plugin=logfire.PydanticPlugin(record='all')
)

logfire.instrument_pydantic_ai()

When an agent encounters a schema failure or triggers a state graph route, Logfire captures the execution tree directly as structured spans:

4. Test Isolation via Dependency Injection

Production CI/CD pipelines cannot rely on execution tests that issue live calls to external LLM providers or require direct database infrastructure connections. Relying on live external calls during integration testing induces four major pipeline failures:

• Non-deterministic build results caused by API variance.
• Unbounded network latency stalling test suites.
• Variable compute billing on every pull request commit.
• State mutations in shared database environments.

We enforce test isolation by decoupling application context dependencies—such as database connection pools, external APIs, and session managers—using PydanticAI’s Deps injection system.

Tool calls receive context instances using RunContext[Deps]. During production execution, runtime services receive real connection pools. During test runs, tests inject mock dependencies without altering agent implementation logic, tool code, or state routing rules.

from dataclasses import dataclass
import pytest
from pydantic_ai import Agent, RunContext

@dataclass
class DatabaseDependencies:
    async def fetch_transaction_count(self, account_id: str) -> int:
        raise NotImplementedError("Database connection required")

typed_agent = Agent[DatabaseDependencies, RiskAssessment](
    'openai:gpt-4o',
    deps_type=DatabaseDependencies,
    result_type=RiskAssessment
)

@typed_agent.tool
async def fetch_account_history(ctx: RunContext[DatabaseDependencies], account_id: str) -> int:
    return await ctx.deps.fetch_transaction_count(account_id)

@pytest.mark.asyncio
async def test_risk_agent_isolation():
    class MockDatabaseDependencies(DatabaseDependencies):
        async def fetch_transaction_count(self, account_id: str) -> int:
            return 42

    result = await typed_agent.run(
        "Evaluate risk profile for account ACC-9901",
        deps=MockDatabaseDependencies()
    )
    assert result.data.risk_score >= 0.0
    assert result.data.account_id == "ACC-9901"

5. End-to-End System Execution Sequence

The interaction between client interfaces, state management, schema enforcement, model generation, and telemetry collection follows an explicit runtime execution sequence.