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

En este tutorial, demostramos cómo diseñar un sistema de decisión agente basado en contratos utilizando PydanticAI, tratando los esquemas estructurados como contratos de gobernanza no negociables en lugar de formatos de salida opcionales. Mostramos cómo definimos un modelo de decisión estricto que codifica el cumplimiento de políticas, la evaluación de riesgos, la calibración de confianza y los próximos pasos procesables directamente en el esquema de salida del agente. Al combinar los validadores de Pydantic con los mecanismos de autocorrección y reintento de PydanticAI, garantizamos que el agente no pueda producir decisiones lógicamente inconsistentes o que no cumplan. A lo largo del flujo de trabajo, nos centramos en crear un agente de decisiones de nivel empresarial que razona bajo restricciones, lo que lo hace adecuado para escenarios de riesgo, cumplimiento y gobernanza del mundo real en lugar de demostraciones basadas en sugerencias de juguete. Consulta los CÓDIGOS COMPLETOS aquí.

!pip -q install -U pydantic-ai pydantic openai nest_asyncio importar os importar tiempo importar asyncio importar getpass desde clases de datos importar clase de datos desde escribir importar Lista, Importar literal nest_asyncio nest_asyncio.apply() desde pydantic importar BaseModel, Campo, field_validator desde pydantic_ai importar Agente desde pydantic_ai.models.openai importar OpenAIChatModel desde pydantic_ai.providers.openai import OpenAIProvider OPENAI_API_KEY = os.getenv(“OPENAI_API_KEY”) si no es OPENAI_API_KEY: intente: desde google.colab importe datos de usuario OPENAI_API_KEY = userdata.get(“OPENAI_API_KEY”) excepto Excepción: OPENAI_API_KEY = Ninguno si no es OPENAI_API_KEY: OPENAI_API_KEY = getpass.getpass(“Ingrese OPENAI_API_KEY: “).strip()

Configuramos el entorno de ejecución instalando las bibliotecas necesarias y configurando la ejecución asincrónica para Google Colab. Cargamos de forma segura la clave API de OpenAI y nos aseguramos de que el tiempo de ejecución esté listo para manejar llamadas de agentes asíncronos. Esto establece una base estable para ejecutar el agente de contrato primero sin problemas relacionados con el medio ambiente. Consulta los CÓDIGOS COMPLETOS aquí.

clase RiskItem(BaseModel): riesgo: str = Field(…, min_length=8) severidad: Literal[“low”, “medium”, “high”]
mitigación: str = Field(…, min_length=12) clase DecisionOutput(BaseModel): decisión: Literal[“approve”, “approve_with_conditions”, “reject”]
confianza: float = Field(…, ge=0.0, le=1.0) justificación: str = Field(…, min_length=80) riesgos_identificados: Lista[RiskItem] = Campo(…, min_length=2) cumplimiento_aprobado: condiciones bool: Lista[str] = Campo(default_factory=lista) next_steps: Lista[str] = Campo(…, min_length=3) timestamp_unix: int = Field(default_factory=lambda: int(time.time())) @field_validator(“confianza”) @classmethod def confianza_vs_risk(cls, v, info): riesgos = info.data.get(“identified_risks”) o []
si hay alguno(r.severity == “alto” para r en riesgos) y v > 0,70: aumentar ValueError(“confianza demasiado alta dados riesgos de alta gravedad”) devolver v @field_validator(“decisión”) @classmethod def rechazar_if_non_compliant(cls, v, info): si info.data.get(“compliance_passed”) es falso y v != “rechazar”: aumentar ValueError(“las decisiones no conformes deben rechazarse”) return v @field_validator(“condiciones”) @classmethod def condition_required_for_conditional_approval(cls, v, info): d = info.data.get(“decisión”) if d == “approve_with_conditions” y (no v o len(v) < 2): rise ValueError("approve_with_conditions requiere al menos 2 condiciones") si d == "aprobar" y v: aumentar ValueError ("aprobar no debe incluir condiciones") return v

Definimos el contrato de decisión central utilizando modelos Pydantic estrictos que describen con precisión una decisión válida. Codificamos restricciones lógicas, como la alineación entre confianza y riesgo, el rechazo impulsado por el cumplimiento y las aprobaciones condicionales, directamente en el esquema. Esto garantiza que la salida de cualquier agente debe satisfacer la lógica empresarial, no sólo la estructura sintáctica. Consulta los CÓDIGOS COMPLETOS aquí.

@dataclass clase DecisionContext: company_policy: str Risk_threshold: float = 0.6 model = OpenAIChatModel( “gpt-5″, proveedor=OpenAIProvider(api_key=OPENAI_API_KEY), ) agente = Agente( model=model, deps_type=DecisionContext, output_type=DecisionOutput, system_prompt=””” Usted es un agente de análisis de decisiones corporativas. Debe evaluar el riesgo, el cumplimiento y la incertidumbre. Todas las salidas deben cumplir estrictamente el esquema DecisionOutput “”” ).

Inyectamos contexto empresarial a través de un objeto de dependencia escrito e inicializamos el agente PydanticAI respaldado por OpenAI. Configuramos el agente para que produzca únicamente decisiones estructuradas que se ajusten al contrato predefinido. Este paso formaliza la separación entre el contexto empresarial y el razonamiento del modelo. Consulta los CÓDIGOS COMPLETOS aquí.

@agent.output_validator def asegurar_riesgo_calidad(resultado: DecisionOutput) -> DecisionOutput: if len(result.identified_risks) < 2: aumentar ValueError("se requieren dos riesgos mínimos") si no hay ninguno(r.severity in ("medium", "high") for r in result.identified_risks): elevar ValueError("se requiere al menos un riesgo medio o alto") devolver resultado @agent.output_validator def enforce_policy_controls(resultado: DecisionOutput) -> DecisionOutput: política = CURRENT_DEPS.company_policy.lower() texto = ( resultado.rationale + ” “.join(result.next_steps) + ” “.join(result.conditions) ).lower() si resultado.compliance_passed: si no hay ninguno(k en texto para k en [“encryption”, “audit”, “logging”, “access control”, “key management”]): genera ValueError (“faltan controles de seguridad concretos”) y devuelve el resultado

Agregamos validadores de salida que actúan como puntos de control de gobernanza después de que el modelo genera una respuesta. Obligamos al agente a identificar riesgos significativos y a hacer referencia explícita a controles de seguridad concretos al reclamar el cumplimiento. Si se violan estas restricciones, activamos reintentos automáticos para imponer la autocorrección. Consulta los CÓDIGOS COMPLETOS aquí.

async def run_decision(): global CURRENT_DEPS CURRENT_DEPS = DecisionContext( company_policy=( “No se implementan sistemas que manejen datos personales o metadatos de transacciones ” “sin cifrado, registro de auditoría y control de acceso con privilegios mínimos.” ) ) Prompt = “”” Solicitud de decisión: implementar un panel de análisis de clientes basado en inteligencia artificial utilizando un proveedor de nube externo. El sistema procesa el comportamiento del usuario y los metadatos de transacciones. El registro de auditoría no está implementado ni es administrado por el cliente. las claves son inciertas. “”” resultado = await agent.run(prompt, deps=CURRENT_DEPS) return result.output decision = asyncio.run(run_decision()) from pprint import pprint pprint(decision.model_dump())

Ejecutamos el agente según una solicitud de decisión realista y capturamos el resultado estructurado validado. Demostramos cómo el agente evalúa el riesgo, el cumplimiento de la política y la confianza antes de tomar una decisión final. Esto completa el flujo de trabajo de extremo a extremo para tomar la primera decisión sobre el contrato en una configuración de estilo de producción.

En conclusión, demostramos cómo pasar de resultados de LLM de formato libre a sistemas de decisión confiables y gobernados utilizando PydanticAI. Mostramos que al hacer cumplir contratos estrictos a nivel de esquema, podemos alinear automáticamente las decisiones con los requisitos de las políticas, la gravedad del riesgo y el realismo de la confianza sin un ajuste manual rápido. Este enfoque nos permite crear agentes que fallan de manera segura, se autocorrigen cuando se violan las restricciones y producen resultados estructurados y auditables en los que los sistemas posteriores pueden confiar. En última instancia, demostramos que el diseño de agente basado en el contrato nos permite implementar IA agente como una capa de decisión confiable dentro de los entornos empresariales y de producción.

Consulta los CÓDIGOS COMPLETOS aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.

Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como empresario e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.