Easy methods to Construct Contract-First Agentic Resolution Methods with PydanticAI for Danger-Conscious, Coverage-Compliant Enterprise AI

!pip -q set up -U pydantic-ai pydantic openai nest_asyncio


import os
import time
import asyncio
import getpass
from dataclasses import dataclass
from typing import Checklist, Literal


import nest_asyncio
nest_asyncio.apply()


from pydantic import BaseModel, Discipline, field_validator
from pydantic_ai import Agent
from pydantic_ai.fashions.openai import OpenAIChatModel
from pydantic_ai.suppliers.openai import OpenAIProvider


OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
   strive:
       from google.colab import userdata
       OPENAI_API_KEY = userdata.get("OPENAI_API_KEY")
   besides Exception:
       OPENAI_API_KEY = None
if not OPENAI_API_KEY:
   OPENAI_API_KEY = getpass.getpass("Enter OPENAI_API_KEY: ").strip()

class RiskItem(BaseModel):
   danger: str = Discipline(..., min_length=8)
   severity: Literal["low", "medium", "high"]
   mitigation: str = Discipline(..., min_length=12)




class DecisionOutput(BaseModel):
   determination: Literal["approve", "approve_with_conditions", "reject"]
   confidence: float = Discipline(..., ge=0.0, le=1.0)
   rationale: str = Discipline(..., min_length=80)
   identified_risks: Checklist[RiskItem] = Discipline(..., min_length=2)
   compliance_passed: bool
   circumstances: Checklist[str] = Discipline(default_factory=listing)
   next_steps: Checklist[str] = Discipline(..., min_length=3)
   timestamp_unix: int = Discipline(default_factory=lambda: int(time.time()))


   @field_validator("confidence")
   @classmethod
   def confidence_vs_risk(cls, v, data):
       dangers = data.information.get("identified_risks") or []
       if any(r.severity == "excessive" for r in dangers) and v > 0.70:
           increase ValueError("confidence too excessive given high-severity dangers")
       return v


   @field_validator("determination")
   @classmethod
   def reject_if_non_compliant(cls, v, data):
       if data.information.get("compliance_passed") is False and v != "reject":
           increase ValueError("non-compliant choices should be reject")
       return v


   @field_validator("circumstances")
   @classmethod
   def conditions_required_for_conditional_approval(cls, v, data):
       d = data.information.get("determination")
       if d == "approve_with_conditions" and (not v or len(v) < 2):
           increase ValueError("approve_with_conditions requires at the least 2 circumstances")
       if d == "approve" and v:
           increase ValueError("approve should not embody circumstances")
       return v

@dataclass
class DecisionContext:
   company_policy: str
   risk_threshold: float = 0.6




mannequin = OpenAIChatModel(
   "gpt-5",
   supplier=OpenAIProvider(api_key=OPENAI_API_KEY),
)


agent = Agent(
   mannequin=mannequin,
   deps_type=DecisionContext,
   output_type=DecisionOutput,
   system_prompt="""
You're a company determination evaluation agent.
You could consider danger, compliance, and uncertainty.
All outputs should strictly fulfill the DecisionOutput schema.
"""
)

@agent.output_validator
def ensure_risk_quality(end result: DecisionOutput) -> DecisionOutput:
   if len(end result.identified_risks) < 2:
       increase ValueError("minimal two dangers required")
   if not any(r.severity in ("medium", "excessive") for r in end result.identified_risks):
       increase ValueError("at the least one medium or excessive danger required")
   return end result




@agent.output_validator
def enforce_policy_controls(end result: DecisionOutput) -> DecisionOutput:
   coverage = CURRENT_DEPS.company_policy.decrease()
   textual content = (
       end result.rationale
       + " ".be part of(end result.next_steps)
       + " ".be part of(end result.circumstances)
   ).decrease()
   if end result.compliance_passed:
       if not any(ok in textual content for ok in ["encryption", "audit", "logging", "access control", "key management"]):
           increase ValueError("lacking concrete safety controls")
   return end result

async def run_decision():
   world CURRENT_DEPS
   CURRENT_DEPS = DecisionContext(
       company_policy=(
           "No deployment of techniques dealing with private information or transaction metadata "
           "with out encryption, audit logging, and least-privilege entry management."
       )
   )


   immediate = """
Resolution request:
Deploy an AI-powered buyer analytics dashboard utilizing a third-party cloud vendor.
The system processes consumer habits and transaction metadata.
Audit logging isn't carried out and customer-managed keys are unsure.
"""


   end result = await agent.run(immediate, deps=CURRENT_DEPS)
   return end result.output




determination = asyncio.run(run_decision())


from pprint import pprint
pprint(determination.model_dump())