Pydantic-AI Governance Agent

“Disclaimer: This article documents my personal exploration of AI governance frameworks and regulatory concepts. Nothing in this post constitutes legal advice, compliance certification, or a guarantee of regulatory conformance. AI governance requirements vary by jurisdiction, sector, and use case. Consult qualified legal and compliance professionals before making governance decisions. NosisTech LLC accepts no liability for outcomes arising from use of this content.”

Pydantic-AI caught my attention because it treats AI output like software, not like a loose conversation. The pattern is powerful: give the model controlled context, ask for a structured answer, validate the answer, and route the result through policy before anything operational happens.

What This Agent Does

This agent reviews fictional enterprise AI use cases and returns a governance decision: APPROVE, REVIEW, or REJECT.

The build uses a small policy context, three fictional business scenarios, a LiteLLM-routed model call, and Pydantic validation. Pydantic is a Python validation library that checks whether data matches the structure the code expects. In this build, it checks whether the model returned the right decision fields before the result is accepted.

The point is not to recreate the full Pydantic-AI framework. The point is to isolate the architectural lesson that matters: AI governance becomes more useful when the model’s answer is treated as structured data that can be checked, corrected, and routed.

Why I Built This

A governance rule that lives only in a document is different from a governance rule that changes what the system does.

Pydantic-AI’s architecture points in the right direction. It uses typed context, structured output, tools, and validation to turn model behavior into something software can inspect. I rebuilt that pattern in plain Python so the control flow is visible.

I wanted to test: can a small agent classify AI use cases, validate its own output shape, correct malformed responses, and enforce a local policy floor across more than one model provider?

What Actually Happened

What actually happened:
The first version worked, but dependency pinning created friction. LiteLLM 1.83.7 pinned python-dotenv==1.0.1, while other installed frameworks wanted newer dotenv versions.

What broke:
The first requirements file caused a resolver conflict during installation. I removed the direct dotenv dependency from this build and replaced it with a tiny local .env loader.

What surprised me:
DeepSeek V4 Pro initially approved the medium-risk customer support scenario even though the use case involved customer PII and external user impact. The structured output was valid, but the governance judgment needed a deterministic policy floor after validation.

What I would change next:
I would extract the policy floor into a small reusable governance wrapper. That wrapper could sit behind any LiteLLM-native agent that produces structured decisions.

Why this matters:
The build showed the difference between output validation and governance enforcement. Pydantic validation checks the shape of the answer. A policy floor checks whether the accepted answer violates local governance rules.

The Architecture: How It Works

The architecture has five parts.

First, the agent loads configuration from .env. The model name, LiteLLM base URL, and API key stay outside the code.

Second, each demo use case is represented as structured input. The model receives a controlled case file instead of an unbounded prompt.

Third, the model returns a structured governance decision. The expected fields are decision, risk score, reason, human review flag, and policy flags.

Fourth, Pydantic validates the returned data. If the model returns malformed JSON or the wrong field types, the agent sends a correction prompt and retries.

Fifth, a deterministic policy floor runs after validation. This matters because a model can return valid JSON that still makes the wrong governance call.

The two-provider test produced the same decision pattern:

Gemini Flash completed the three-case run in 15.54 seconds.
DeepSeek V4 Pro completed the same run in 64.26 seconds.

Both providers returned APPROVE for the low-risk internal summarizer and REVIEW for the customer support and eligibility scenarios. Gemini was much faster in this run. DeepSeek produced valid structured decisions, but took longer and initially exposed why local policy enforcement belongs outside the model.

The Build: Step by Step

class UseCase(BaseModel):
    """Structured input model for an enterprise AI use case."""

    title: str
    business_unit: str
    data_sensitivity: Literal["LOW", "MEDIUM", "HIGH"]
    automated_decisioning: bool
    external_user_impact: bool
    description: str

class GovernanceDecision(BaseModel):
    """Structured output model for a governance review decision."""

    decision: Literal["APPROVE", "REVIEW", "REJECT"]
    risk_score: int = Field(ge=0, le=10)
    reason: str
    required_human_review: bool
    policy_flags: list[str]

What This Code Is Actually Doing

This is the contract for the agent. A contract is a fixed agreement about what data is allowed to look like. UseCase defines the information the agent reviews. GovernanceDecision defines the answer the model has to return. The risk_score field is constrained from 0 to 10, so a model response with a value outside that range is rejected before it becomes part of the output.

def load_env_file() -> None:
    """Load local .env values without an extra dependency."""
    env_path = os.path.join(os.getcwd(), ".env")
    if not os.path.exists(env_path):
        return
    with open(env_path, encoding="utf-8") as env_file:
        for line in env_file:
            clean = line.strip()
            if clean and not clean.startswith("#") and "=" in clean:
                name, value = clean.split("=", 1)
                os.environ.setdefault(name.strip(), value.strip())

What This Code Is Actually Doing

This replaces a separate dotenv dependency with a small local loader. It reads the .env file, ignores blank lines and comments, and loads each setting into the environment. The reason I made this change was practical: LiteLLM pinned one dotenv version while other frameworks on the machine wanted another. Removing the direct dependency made this build easier to install without changing the architecture.

def build_prompt(use_case: UseCase, policy_context: str, correction: str = "") -> list[dict[str, str]]:
    """Build model messages for one governance classification."""
    schema = '{"decision":"APPROVE|REVIEW|REJECT","risk_score":0-10,"reason":"string","required_human_review":true|false,"policy_flags":["string"]}'
    prompt = "\n".join(["Return only valid JSON matching this schema:", schema, policy_context, "Use case:", use_case.model_dump_json()])
    if correction:
        prompt += "\nCorrect this validation problem: " + correction
    return [
        {"role": "system", "content": "You are a careful AI governance reviewer."},
        {"role": "user", "content": prompt},
    ]

What This Code Is Actually Doing

This function builds the instruction package for the model. It injects the policy context and the use case into one compact request. Dependency injection means passing the needed context into a function instead of letting that function search for it globally. In this rebuild, the policy context is passed directly into the prompt builder, which makes the governance rule source visible.

def extract_json_object(raw_text: str) -> dict:
    """Extract the first JSON object from model text."""
    cleaned = re.sub(r"```(?:json)?", "", raw_text).replace("```", "").strip()
    match = re.search(r"\{.*\}", cleaned, re.DOTALL)
    if not match:
        raise ValueError("No JSON object found in the model response.")
    return json.loads(match.group())

What This Code Is Actually Doing

This is the provider-tolerance layer. Some models return clean JSON. Others wrap JSON inside markdown fences or explanatory text. This function removes common wrapping and extracts the first JSON object. That gave the agent a better chance of working across Gemini, DeepSeek, and future LiteLLM providers.

def classify_use_case(client: OpenAI, model: str, use_case: UseCase, policy_context: str) -> GovernanceDecision:
    """Classify one use case with validation retries."""
    correction = ""
    for _ in range(3):
        raw_text = call_model_with_retries(client, model, build_prompt(use_case, policy_context, correction))
        try:
            decision = GovernanceDecision(**extract_json_object(raw_text))
            return enforce_policy_floor(use_case, decision)
        except (json.JSONDecodeError, ValidationError, ValueError) as error:
            correction = str(error)
    print("Could not validate the model's governance decision after three attempts.")
    sys.exit(1)

What This Code Is Actually Doing

This is the validation loop. The model gets three chances to return usable structured data. If parsing or Pydantic validation fails, the agent sends back a correction note and asks for a corrected JSON object. Once the answer validates, the agent still passes it through the local policy floor before printing the result.

def enforce_policy_floor(use_case: UseCase, decision: GovernanceDecision) -> GovernanceDecision:
    """Raise decisions that violate the local governance policy."""
    sensitive_external = use_case.data_sensitivity in ("MEDIUM", "HIGH") and use_case.external_user_impact
    automated_external = use_case.automated_decisioning and use_case.external_user_impact
    restricted_terms = ("compliance", "eligibility", "denial", "credit", "hiring", "housing", "insurance", "medical")
    restricted_decision = any(term in use_case.description.lower() for term in restricted_terms)
    if decision.decision == "APPROVE" and (sensitive_external or automated_external or restricted_decision):
        flags = list(dict.fromkeys([*decision.policy_flags, "LOCAL_POLICY_REVIEW_REQUIRED"]))
        score = max(decision.risk_score, 5 if use_case.data_sensitivity == "MEDIUM" else 7)
        return decision.model_copy(update={"decision": "REVIEW", "risk_score": score, "required_human_review": True, "policy_flags": flags})
    if decision.decision in ("REVIEW", "REJECT"):
        return decision.model_copy(update={"required_human_review": True})
    return decision

What This Code Is Actually Doing

This is the most important governance lesson in the build. The model can return valid data that still conflicts with local policy. This function checks the validated answer against local rules. If the model tries to approve a sensitive external-impact case, the local code raises the decision to REVIEW and turns on the human review flag.

Errors I Hit During This Build

The first error was a dependency conflict.

The terminal reported that litellm==1.83.7 required python-dotenv==1.0.1, while the requirements file pinned python-dotenv==1.1.1. That happened because I initially matched the visible environment instead of LiteLLM’s own dependency pin. I fixed it by removing direct dotenv from this build and adding a tiny .env loader in agent.py.

The second error came from PowerShell string replacement.

During an edit, literal newline markers were inserted into agent.py. Python caught the problem during py_compile with a syntax error. I repaired the affected section directly and reran the syntax check before continuing.

The third issue was not a crash. It was a governance behavior problem.

DeepSeek V4 Pro initially returned APPROVE for the customer support summarizer. The JSON shape was valid, but the decision was too permissive for a medium-sensitivity use case involving customer PII and external user impact. I fixed that by adding the deterministic local policy floor after Pydantic validation.

Why This Matters

This build separates three ideas that often get collapsed into one.

Prompting is asking the model for the right behavior.

Validation is checking whether the model returned the right shape.

Governance enforcement is checking whether the validated answer is acceptable under local rules.

That separation matters because structured output alone is not governance. It gives the system something reliable to inspect, but the final control still belongs in code. In this build, Pydantic handles the structure, LiteLLM handles model portability, and the local policy floor handles the final governance boundary.

The provider timing also mattered. Gemini Flash completed the run in 15.54 seconds. DeepSeek V4 Pro completed it in 64.26 seconds. For this specific governance workload, both models reached the same decision pattern, but speed and wording differed enough to make provider benchmarking part of the architecture decision.

That is the real takeaway: governance systems need model-agnostic execution, structured validation, and local enforcement. The model contributes judgment. The architecture decides how far that judgment is allowed to travel.