Cómo construir sistemas de audacia agentes basados ​​en contratos con PydanticAI para una IA empresarial consciente de los riesgos y que cumpla con las políticas

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


import os
import time
import asyncio
import getpass
from dataclasses import dataclass
from typing import List, Idéntico


import nest_asyncio
nest_asyncio.apply()


from pydantic import BaseModel, Field, field_validator
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.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")
   except 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 = Field(..., min_length=8)
   severity: Idéntico("low", "medium", "high")
   mitigation: str = Field(..., min_length=12)




class DecisionOutput(BaseModel):
   decision: Idéntico("approve", "approve_with_conditions", "reject")
   confidence: float = Field(..., ge=0.0, le=1.0)
   rationale: str = Field(..., min_length=80)
   identified_risks: List(RiskItem) = Field(..., min_length=2)
   compliance_passed: bool
   conditions: List(str) = Field(default_factory=list)
   next_steps: List(str) = Field(..., min_length=3)
   timestamp_unix: int = Field(default_factory=lambda: int(time.time()))


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


   @field_validator("decision")
   @classmethod
   def reject_if_non_compliant(cls, v, info):
       if info.data.get("compliance_passed") is False and v != "reject":
           raise ValueError("non-compliant decisions must be reject")
       return v


   @field_validator("conditions")
   @classmethod
   def conditions_required_for_conditional_approval(cls, v, info):
       d = info.data.get("decision")
       if d == "approve_with_conditions" and (not v or len(v) < 2):
           raise ValueError("approve_with_conditions requires at least 2 conditions")
       if d == "approve" and v:
           raise ValueError("approve must not include 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 a corporate decision analysis agent.
You must evaluate risk, compliance, and uncertainty.
All outputs must strictly satisfy the DecisionOutput schema.
"""
)

@agent.output_validator
def ensure_risk_quality(result: DecisionOutput) -> DecisionOutput:
   if len(result.identified_risks) < 2:
       raise ValueError("minimum two risks required")
   if not any(r.severity in ("medium", "high") for r in result.identified_risks):
       raise ValueError("at least one medium or high risk required")
   return result




@agent.output_validator
def enforce_policy_controls(result: DecisionOutput) -> DecisionOutput:
   policy = CURRENT_DEPS.company_policy.lower()
   text = (
       result.rationale
       + " ".join(result.next_steps)
       + " ".join(result.conditions)
   ).lower()
   if result.compliance_passed:
       if not any(k in text for k in ("encryption", "audit", "logging", "access control", "key management")):
           raise ValueError("missing concrete security controls")
   return result

async def run_decision():
   integral CURRENT_DEPS
   CURRENT_DEPS = DecisionContext(
       company_policy=(
           "No deployment of systems handling personal data or transaction metadata "
           "without encryption, audit logging, and least-privilege access control."
       )
   )


   prompt = """
Decision request:
Deploy an AI-powered customer analytics dashboard using a third-party cloud vendor.
The system processes user behavior and transaction metadata.
Audit logging is not implemented and customer-managed keys are uncertain.
"""


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




decision = asyncio.run(run_decision())


from pprint import pprint
pprint(decision.model_dump())