The Method To Assemble Contract-First Agentic Decision Packages With PydanticAI For Risk-Acutely aware, Protection-Compliant Enterprise AI

!pip -q arrange -U pydantic-ai pydantic openai nest_asyncio


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


import nest_asyncio
nest_asyncio.apply()


from pydantic import BaseModel, Self-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:
   try:
       from google.colab import userdata
       OPENAI_API_KEY = userdata.get("OPENAI_API_KEY")
   moreover Exception:
       OPENAI_API_KEY = None
if not OPENAI_API_KEY:
   OPENAI_API_KEY = getpass.getpass("Enter OPENAI_API_KEY: ").strip()

class RiskItem(BaseModel):
   risk: str = Self-discipline(..., min_length=8)
   severity: Literal["low", "medium", "high"]
   mitigation: str = Self-discipline(..., min_length=12)




class DecisionOutput(BaseModel):
   selection: Literal["approve", "approve_with_conditions", "reject"]
   confidence: float = Self-discipline(..., ge=0.0, le=1.0)
   rationale: str = Self-discipline(..., min_length=80)
   identified_risks: Itemizing[RiskItem] = Self-discipline(..., min_length=2)
   compliance_passed: bool
   conditions: Itemizing[str] = Self-discipline(default_factory=document)
   next_steps: Itemizing[str] = Self-discipline(..., min_length=3)
   timestamp_unix: int = Self-discipline(default_factory=lambda: int(time.time()))


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


   @field_validator("selection")
   @classmethod
   def reject_if_non_compliant(cls, v, knowledge):
       if knowledge.data.get("compliance_passed") is False and v != "reject":
           elevate ValueError("non-compliant picks need to be reject")
       return v


   @field_validator("conditions")
   @classmethod
   def conditions_required_for_conditional_approval(cls, v, knowledge):
       d = knowledge.data.get("selection")
       if d == "approve_with_conditions" and (not v or len(v) < 2):
           elevate ValueError("approve_with_conditions requires a minimal of two conditions")
       if d == "approve" and v:
           elevate ValueError("approve shouldn't embrace conditions")
       return v

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




model = OpenAIChatModel(
   "gpt-5",
   provider=OpenAIProvider(api_key=OPENAI_API_KEY),
)


agent = Agent(
   model=model,
   deps_type=DecisionContext,
   output_type=DecisionOutput,
   system_prompt="""
You are an organization selection analysis agent.
It is important to contemplate risk, compliance, and uncertainty.
All outputs ought to strictly fulfill the DecisionOutput schema.
"""
)

@agent.output_validator
def ensure_risk_quality(consequence: DecisionOutput) -> DecisionOutput:
   if len(consequence.identified_risks) < 2:
       elevate ValueError("minimal two risks required")
   if not any(r.severity in ("medium", "extreme") for r in consequence.identified_risks):
       elevate ValueError("a minimal of 1 medium or extreme risk required")
   return consequence




@agent.output_validator
def enforce_policy_controls(consequence: DecisionOutput) -> DecisionOutput:
   protection = CURRENT_DEPS.company_policy.lower()
   textual content material = (
       consequence.rationale
       + " ".be a part of(consequence.next_steps)
       + " ".be a part of(consequence.conditions)
   ).lower()
   if consequence.compliance_passed:
       if not any(okay in textual content material for okay in ["encryption", "audit", "logging", "access control", "key management"]):
           elevate ValueError("missing concrete security controls")
   return consequence

async def run_decision():
   world CURRENT_DEPS
   CURRENT_DEPS = DecisionContext(
       company_policy=(
           "No deployment of methods coping with personal data or transaction metadata "
           "with out encryption, audit logging, and least-privilege entry administration."
       )
   )


   quick = """
Decision request:
Deploy an AI-powered purchaser analytics dashboard using a third-party cloud vendor.
The system processes client habits and transaction metadata.
Audit logging won't be utilized and customer-managed keys are not sure.
"""


   consequence = await agent.run(quick, deps=CURRENT_DEPS)
   return consequence.output




selection = asyncio.run(run_decision())


from pprint import pprint
pprint(selection.model_dump())