Sample Page Title

!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):
   threat: str = Discipline(..., min_length=8)
   severity: Literal["low", "medium", "high"]
   mitigation: str = Discipline(..., min_length=12)




class DecisionOutput(BaseModel):
   resolution: 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
   situations: Checklist[str] = Discipline(default_factory=checklist)
   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:
           elevate ValueError("confidence too excessive given high-severity dangers")
       return v


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


   @field_validator("situations")
   @classmethod
   def conditions_required_for_conditional_approval(cls, v, data):
       d = data.information.get("resolution")
       if d == "approve_with_conditions" and (not v or len(v) < 2):
           elevate ValueError("approve_with_conditions requires not less than 2 situations")
       if d == "approve" and v:
           elevate ValueError("approve should not embody situations")
       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 resolution evaluation agent.
You should consider threat, compliance, and uncertainty.
All outputs should strictly fulfill the DecisionOutput schema.
"""
)

@agent.output_validator
def ensure_risk_quality(outcome: DecisionOutput) -> DecisionOutput:
   if len(outcome.identified_risks) < 2:
       elevate ValueError("minimal two dangers required")
   if not any(r.severity in ("medium", "excessive") for r in outcome.identified_risks):
       elevate ValueError("not less than one medium or excessive threat required")
   return outcome




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

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


   immediate = """
Determination request:
Deploy an AI-powered buyer analytics dashboard utilizing a third-party cloud vendor.
The system processes person habits and transaction metadata.
Audit logging will not be carried out and customer-managed keys are unsure.
"""


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




resolution = asyncio.run(run_decision())


from pprint import pprint
pprint(resolution.model_dump())