En este tutorial, implementamos un sistema de gobernanza de doble agente que aplica los principios constitucionales de la IA a las operaciones financieras. Demostramos cómo separamos la ejecución y la supervisión al combinar un Agente trabajador que realiza acciones financieras con un Agente auditor que hace cumplir las políticas, la seguridad y el cumplimiento. Al codificar las reglas de gobernanza directamente en una constitución formal y combinar controles basados en reglas con razonamiento asistido por IA, podemos construir sistemas que sean autorreflexivos, auditables y resistentes a comportamientos riesgosos o de incumplimiento en flujos de trabajo financieros de alto riesgo. Consulta los CÓDIGOS COMPLETOS aquí.
!pip install -q pydantic anthropic python-dotenv importar json importar re desde escribir importar Lista, Dict, Cualquiera, Opcional, Literal desde pydantic importar Modelo base, campo, validador desde enum importar Enum desde fecha y hora importar fecha y hora importar sistema operativo
Instalamos e importamos las bibliotecas principales necesarias para estructurar, validar y gobernar nuestro sistema basado en agentes. Confiamos en Pydantic para modelos de datos fuertemente tipados, enumeraciones y validación, mientras que las utilidades estándar de Python manejan marcas de tiempo, análisis y configuración del entorno. Consulta los CÓDIGOS COMPLETOS aquí.
class PolicyViolationType(str, Enum): """Tipos de violaciones de políticas""" PII_EXPOSURE = "pii_exposure" BUDGET_EXCEEDED = "budget_exceeded" UNAUTHORIZED_ACTION = "unauthorized_action" MISSING_JUSTIFICATION = "missing_justification" SUSPICIOUS_PATTERN = "suspicious_pattern" class SafetyPolicy(BaseModel): """Regla de política de seguridad individual""" nombre: str descripción: str gravedad: Literal["bajo", "medio", "alto", "crítico"] check_function: str clase Constitución(BaseModel): """La 'Constitución' – Un conjunto de reglas que gobiernan el comportamiento de los agentes""" políticas: Lista[SafetyPolicy] max_transaction_amount: float = 10000.0 require_approval_above: float = 5000.0 permitido_pii_fields: Lista[str] = ["nombre", "account_id"] def get_policy_by_name(self, nombre: str) -> Opcional[SafetyPolicy]: return next((p for p in self.policies if p.name == nombre), Ninguno) FINANCIAL_CONSTITUTION = Constitución( políticas=[ SafetyPolicy( name="Protección de PII", descripción="No debe exponer PII confidencial (SSN, tarjeta de crédito completa, contraseñas)", severidad="crítica", check_function="Buscar patrones de SSN, números de tarjetas de crédito y contraseñas" ), SafetyPolicy( name="Límites de presupuesto", descripción="Las transacciones no deben exceder los límites de presupuesto predefinidos", severidad="alta", check_function="Verificar si el monto de la transacción excede la cantidad máxima de transacción" ), SafetyPolicy( nombre="Autorización de acción", descripción="Solo se permiten tipos de acción preaprobados", severidad="alta", check_function="Verificar que el tipo de acción sea en la lista aprobada"), SafetyPolicy( name="Justificación requerida", descripción="Todas las transacciones por encima del umbral deben tener justificación", severidad="medium", check_function="Compruebe el campo de justificación en transacciones de alto valor" ), SafetyPolicy( name="Detección de patrones", descripción="Detectar patrones sospechosos (múltiples transacciones rápidas, números redondos)", severidad="medium", check_function="Analizar patrones de transacciones en busca de anomalías" ) ], max_transaction_amount=10000.0, require_approval_above=5000.0)
Definimos el marco constitucional central que gobierna el comportamiento de los agentes formalizando los tipos de políticas, la gravedad y las reglas de aplicación. Codificamos restricciones de seguridad financiera, como protección de PII, límites presupuestarios, verificaciones de autorización y requisitos de justificación, como políticas de primera clase legibles por máquina. Consulta los CÓDIGOS COMPLETOS aquí.
clase FinancialRequest(BaseModel): """Solicitud de entrada al agente trabajador""" acción: str cantidad: Opcional[float] = Ninguno destinatario: Opcional[str] = Ninguna descripción: str justificación: Opcional[str] = Ninguno metadatos: Dict[str, Any] = Campo(default_factory=dict) clase WorkerOutput(BaseModel): """Salida del Agente trabajador""" request_id: str action_taken: str detalles: Dict[str, Any] raw_response: str marca de tiempo: str = Field(default_factory=lambda: datetime.now().isoformat()) clase PolicyViolation(BaseModel): """Infracción de política detectada""" nombre_política: str tipo_violación: PolicyViolationType gravedad: str descripción: str sugerencia_fix: Opcional[str] = Ninguna clase AuditResult(BaseModel): """Resultado del agente auditor""" aprobado: bool violaciones: Lista[PolicyViolation] = Campo(default_factory=list) Risk_score: float # 0-100 retroalimentación: str revision_needed: bool @classmethod def validar_risk_score(cls, v): if isinstance(v, (int, float)): return max(0.0, min(100.0, v)) return v
Definimos modelos de datos fuertemente tipados que estructuran cómo las solicitudes financieras, los resultados de los agentes y los hallazgos de las auditorías fluyen a través del sistema. Utilizamos estos esquemas para garantizar que cada acción, decisión e infracción se capture en un formato coherente, validado por máquina y con total trazabilidad. Consulta los CÓDIGOS COMPLETOS aquí.
class MockAIClient: """Simula la API Anthropic para este tutorial""" def __init__(self): self.call_count = 0 def message_create(self, model: str, max_tokens: int, mensajes: List[Dict]) -> Any: """Simular llamada API""" self.call_count += 1 user_msg = mensajes[-1]["content"] if "WORKER AGENTE" en user_msg o "solicitud financiera" en user_msg.lower(): devolver self._worker_response(user_msg) elif "AGENTE AUDITOR" en user_msg o "auditoría" en user_msg.lower(): devolver self._auditor_response(user_msg) devolver self._default_response() def _worker_response(self, msg: str) -> Cualquiera: """Simular agente trabajador procesando una solicitud""" cantidad_match = re.search(r'$?(d+(?:,d{3})*(?:.d{2})?)', msg) cantidad = float(amount_match.group(1).replace(',', '')) if cantidad_match else 0 if 'transferir' en msg.lower(): acción = 'transferir' elif 'pago' en msg.lower() o 'pago' en msg.lower(): acción = 'pago' elif 'informe' en msg.lower(): acción = 'informe' else: acción = 'general_query' respuesta = { "action_taken": acción, "monto": monto, "status": "completado", "destinatario": "John Doe" si monto > 0 else Ninguno, "account_id": "ACC-12345", "marca de tiempo": datetime.now().isoformat() } si monto > 5000: respuesta["ssn"] = "123-45-6789" si monto > 8000: respuesta["credit_card"] = "4532-1234-5678-9010" clase MockResponse: def __init__(self, content): self.content = [type('obj', (object,), { 'type': 'text', 'text': json.dumps(content, indent=2) })] return MockResponse(response) def _auditor_response(self, msg: str) -> Cualquiera: """Simular políticas de verificación del agente auditor""" violaciones =[]if 'ssn' en msg.lower() o re.search(r'd{3}-d{2}-d{4}', msg): violaciones.append({ "policy": "Protección PII", "type": "pii_exposure", "severity": "critical", "detail": "SSN detectado en la salida" }) if 'credit_card' en msg.lower() o re.search(r'd{4}-d{4}-d{4}-d{4}', msg): violaciones.append({ "policy": "Protección PII", "type": "pii_exposure", "severity": "critical", "detail": "Número de tarjeta de crédito detectado" }) cantidad_match = re.search(r'"amount":s*(d+(?:.d+)?)', msg) if cantidad_match: cantidad = float(amount_match.group(1)) if cantidad > 10000: violaciones.append({ "policy": "Budget Limits", "type": "budget_exceeded", "severity": "high", "detail": f"La cantidad ${amount} excede el límite de $10 000" }) elif cantidad > 5000 y la 'justificación' no están en msg.lower(): violaciones.append({ "policy": "Justificación requerida", "tipo": "missing_justification", "severity": "medium", "detail": "La transacción de alto valor carece de justificación" }) audit_result = { "approved": len(violations) == 0, "violations": violaciones, "risk_score": min(len(violations) * 30, 100), "feedback": "Transacción aprobada" if len(violations) == 0 else "Violaciones detectadas – revisión requerida" } class MockResponse: def __init__(self, content): self.content = [type('obj', (object,), { 'type': 'text', 'text': json.dumps(content, indent=2) })] return MockResponse(audit_result) def _default_response(self) -> Cualquiera: clase MockResponse: def __init__(self): self.content = [type('obj', (objeto,), { 'type': 'text', 'text': '{"status": "acknowledged"}' })] return MockResponse()
Simulamos el comportamiento de un modelo de lenguaje grande mediante la implementación de un cliente de IA simulado que diferencia entre roles de trabajador y auditor. Inyectamos intencionalmente violaciones de políticas, como fugas de PII y problemas presupuestarios, para poner a prueba la lógica de gobernanza en condiciones de falla realistas. Consulta los CÓDIGOS COMPLETOS aquí.
class WorkerAgent: """Agente A – El trabajador que procesa solicitudes financieras""" def __init__(self, cliente: MockAIClient): self.client = cliente self.role = "Trabajador de operaciones financieras" self.processed_requests =[]def Process_request(self, request: FinancialRequest) -> WorkerOutput: """Procesar una solicitud financiera""" print(f"n{'='*60}") print(f"🔧 AGENTE TRABAJADOR: Procesando solicitud…") print(f"{'='*60}") print(f"Acción: {request.action}") if request.amount: print(f"Cantidad: ${request.amount:,.2f}") else: print("Cantidad: N/A") print(f"Descripción: {request.description}") solicitud = self._build_worker_prompt(solicitud) respuesta = self.client.messages_create( model="claude-sonnet-4-20250514", max_tokens=1000, mensajes=[{"role": "usuario", "contenido": mensaje}] ) raw_response = respuesta.content[0].text try: detalles = json.loads(raw_response) excepto json.JSONDecodeError: detalles = {"raw": raw_response} salida = WorkerOutput( request_id=f"REQ-{len(self.processed_requests)+1:04d}", action_taken=request.action, detalles=detalles, raw_response=raw_response) self.processed_requests.append(output) print(f"n✅ El trabajador completó el procesamiento (ID: {output.request_id})") devuelve salida def _build_worker_prompt(self, request: FinancialRequest) -> str: """Construir mensaje para el agente trabajador""" cantidad_str = f"${request.amount:,.2f}" if request.amount else "$0.00" return f"""Usted es un AGENTE TRABAJADOR que procesa una solicitud financiera. Detalles de la solicitud: – Acción: {request.action} – Monto: {amount_str} – Destinatario: {request.recipient o 'N/A'} – Descripción: {request.description} – Justificación: {request.justification o 'Ninguno proporcionado'} Procese esta solicitud y devuelva una respuesta JSON con: – action_taken – monto – estado – destinatario – account_id – marca de tiempo – Cualquier otro detalle relevante Devolución SÓLO válida. JSON.""" class AuditorAgent: """Agente B: el auditor que valida la producción del trabajador""" def __init__(self, cliente: MockAIClient, constitución: Constitución): self.client = cliente self.constitution = constitución self.role = "Auditor de gobernanza" self.audit_history =[]def audit(self, trabajador_salida: WorkerOutput) -> AuditResult: """Auditar la producción del trabajador contra la constitución""" print(f"n{'='*60}") print(f"🔍 AGENTE AUDITOR: Auditar la producción…") print(f"{'='*60}") violaciones = self._check_rules(worker_output) solicitud = self._build_auditor_prompt(worker_output, violaciones) respuesta = self.client.messages_create( model="claude-sonnet-4-20250514", max_tokens=1000, mensajes=[{"role": "usuario", "content": aviso}] ) raw_audit = respuesta.content[0].text try: audit_data = json.loads(raw_audit) excepto json.JSONDecodeError: audit_data = {"aprobado": False, "violations": violaciones, "risk_score": 50} resultado = AuditResult( aprobado=audit_data.get("aprobado", False) y len(violaciones) == 0, violaciones=violaciones, Risk_score=audit_data.get("risk_score", len(violaciones) * 25), feedback=audit_data.get("feedback", "Auditoría completada"), revision_needed=not audit_data.get("aprobado", False) o len(violaciones) > 0 ) self.audit_history.append(resultado) self._display_audit_result(resultado) devuelve resultado def _check_rules(self, salida: WorkerOutput) -> Lista[PolicyViolation]: """Realizar controles constitucionales basados en reglas""" violaciones =[]detalles_str = json.dumps(output.details) if re.search(r'd{3}-d{2}-d{4}', detalles_str): violaciones.append(PolicyViolation( Policy_name="Protección PII", violación_tipo=PolicyViolationType.PII_EXPOSURE, severidad="crítico", descripción="Número de Seguro Social detectado en la salida", sugerido_fix="Eliminar o enmascarar el campo SSN" )) if re.search(r'd{4}[-s]?d{4}[-s]?d{4}[-s]?d{4}', detalles_str): violaciones.append(PolicyViolation( Policy_name="Protección PII", violación_tipo=PolicyViolationType.PII_EXPOSURE, gravedad="crítico", descripción="Número de tarjeta de crédito detectado en la salida", sugerido_fix="Eliminar o tokenizar el número de tarjeta de crédito" )) monto = salida.detalles.get("monto", 0) if monto > self.constitution.max_transaction_amount: violaciones.append(PolicyViolation( Policy_name="Límites de presupuesto", violación_tipo=PolicyViolationType.BUDGET_EXCEEDED, gravedad="alta", descripción=f"Monto ${monto:,.2f} excede el límite de ${self.constitution.max_transaction_amount:,.2f}", sugerido_fix=f"Reducir el monto a ${self.constitution.max_transaction_amount:,.2f} o solicitar aprobación" )) if monto > self.constitution.require_approval_above: si la "justificación" no está en detalles_str.lower(): violaciones.append(PolicyViolation( Policy_name="Justificación requerida", violacion_tipo=PolicyViolationType.MISSING_JUSTIFICATION, severidad="medio", descripción=f"La transacción de ${cantidad:,.2f} requiere justificación", sugerido_fix="Agregar campo de justificación que explique la transacción" )) return violaciones def _build_auditor_prompt(self, salida: WorkerOutput, violaciones: Lista[PolicyViolation]) -> str: """Construir mensaje para el agente auditor""" return f"""Usted es un AGENTE AUDITOR que valida operaciones financieras contra una Constitución. Políticas de Constitución: {json.dumps([p.dict() for p in self.constitution.policies], indent=2)} Salida del trabajador para auditar: {output.raw_response} Violaciones ya detectadas: {json.dumps([v.dict() for v in violations], indent=2)} Realice análisis adicionales y devuelva JSON con: – aprobado (booleano) – Risk_score (0-100) – Comentarios (cadena) – Cualquier inquietud adicional Devuelve SÓLO JSON válido.""" def _display_audit_result(self, result: AuditResult): """Muestra los resultados de la auditoría en un formato legible""" print(f"n📊 RESULTADOS DE LA AUDITORÍA:") print(f"Estado: {'✅ APROBADO' si resultado.aprobado else '❌ REJECTED'}") print(f"Puntuación de riesgo: {result.risk_score:.1f}/100") print(f"Violaciones encontradas: {len(result.violations)}") if result.violations: print(f"n⚠️ VIOLACIONES DE LA POLÍTICA:") for i, v in enumerate(result.violations, 1): print(f"n {i}. {v.policy_name} [{v.severity.upper()}]") print(f" Tipo: {v.violation_type.value}") print(f" Problema: {v.description}") if v.suggested_fix: print(f" Corrección: {v.suggested_fix}") print(f"n💬 Comentarios: {result.feedback}") print(f"Revisión necesaria: {'Sí' if result.revision_needed else 'No'}")
Implementamos la lógica central de doble agente separando las responsabilidades de ejecución y gobernanza entre un Agente trabajador y un Agente auditor. Permitimos que el trabajador se concentre únicamente en cumplir con las solicitudes financieras, mientras hacemos cumplir las reglas constitucionales mediante controles deterministas y auditorías asistidas por IA. Al combinar indicaciones estructuradas, validación basada en reglas y comentarios claros de auditoría, creamos un circuito de control autorreflexivo que prioriza la seguridad, la responsabilidad y el cumplimiento. Consulta los CÓDIGOS COMPLETOS aquí.
class GovernanceSystem: """Orquesta el flujo de trabajo de gobernanza de agente dual""" def __init__(self, constitution: Constitution): self.client = MockAIClient() self.worker = WorkerAgent(self.client) self.auditor = AuditorAgent(self.client, constitution) self.constitution = constitution self.max_revision_attempts = 3 def Process_with_governance(self, request: FinancialRequest) -> Dict[str, Any]: """Flujo de trabajo principal: Procesos de trabajo, Auditor valida, bucle si es necesario""" print(f"n{'#'*60}") print(f"# SISTEMA DE GOBIERNO: Nueva solicitud") print(f"{'#'*60}") intento = 0 while intento < self.max_revision_attempts: intento += 1 print(f"n🔄 Intento {intento}/{self.max_revision_attempts}") trabajador_salida = self.worker.process_request(solicitud) audit_result = self.auditor.audit(worker_output) if audit_result.approved: print(f"n{'='*60}") print(f"✅ RESULTADO FINAL: APROBADO") print(f"{'='*60}") return { "status": "aprobado", "salida": trabajador_salida.dict(), "auditoría": audit_result.dict(), "intentos": intento } violaciones_criticas = [v for v in audit_result.violations if v.severity == "critical"] if violaciones_criticas: print(f"n{'='*60}") print(f"🛑 RESULTADO FINAL: RECHAZADO (Violaciones críticas)") print(f"{'='*60}") return { "status": "rechazado", "reason": "critical_violations", "audit": audit_result.dict(), "attempts": intento } if intento >= self.max_revision_attempts: print(f"n{'='*60}") print(f"🛑 RESULTADO FINAL: RECHAZADO (Max Intentos)") print(f"{'='*60}") return { "status": "rejected", "reason": "max_attempts_exceeded", "audit": audit_result.dict(), "attempts": intento } return {"status": "error", "message": "Salida inesperada del bucle"}
Orquestamos el flujo de trabajo de gobernanza completo coordinando a los agentes trabajadores y auditores dentro de un ciclo de revisión controlado. Evaluamos cada intento contra las normas constitucionales y detenemos inmediatamente la ejecución cuando se detectan violaciones críticas. Consulta los CÓDIGOS COMPLETOS aquí.
def run_examples(): """Ejecutar ejemplos de demostración""" print("="*80) print(" SISTEMA DE GOBERNANZA DE AGENTE DOBLE CON IA CONSTITUCIONAL") print(" Tutorial: Agentes de operaciones financieras autorreflexivos") print("="*80) system = GovernanceSystem(FINANCIAL_CONSTITUTION) print("nn" + "="*80) print("EJEMPLO 1: Transacción segura ($2500)") print("="*80) request1 = FinancialRequest( action="pago", cantidad=2500.00, destinatario="Vendor Corp", descripción="Pago mensual de licencia de software", justificación="Pago recurrente regular por servicios esenciales" ) resultado1 = system.process_with_governance(request1) print("nn" + "="*80) print("EJEMPLO 2: Transacción de alto valor con PII Fuga ($7500)") print("="*80) request2 = FinancialRequest( action="transfer", cantidad=7500.00, destinatario="Ejecutivo", descripción="Pago de bonificación al ejecutivo", justificación="Bono de desempeño del cuarto trimestre" ) resultado2 = system.process_with_governance(request2) print("nn" + "="*80) print("EJEMPLO 3: Transacción que excede el presupuesto ($15,000)") print("="*80) request3 = Solicitud Financiera( action="transfer", cantidad=15000.00, destinatario="Proveedor", descripción="Compra de equipo grande", justificación="Nuevo equipo de fabricación para línea de producción" ) resultado3 = system.process_with_governance(request3) print("nn" + "="*80) print(" RESUMEN DE RESULTADOS") print("="*80) print(f"nEjemplo 1: {resultado1['estado'].superior()}") print(f"Ejemplo 2: {resultado2['estado'].superior()} – {result2.get('razón', 'N/A')}") print(f"Ejemplo 3: {resultado3['estado'].superior()} – {result3.get('razón', 'N/A')}") print(f"nnTotal de llamadas API: {system.client.call_count}") print(f"Trabajador procesado: {len(system.worker.processed_requests)} solicitudes") print(f"Auditor realizado: {len(system.auditor.audit_history)} auditorías") print("nn" + "="*80) print(" CONSTITUCIÓN ACTIVA") print("="*80) para la política en FINANCIAL_CONSTITUTION.policies: print(f"n📜 {policy.name} [{policy.severity.upper()}]") print(f" {policy.description}")
Demostramos el sistema de extremo a extremo ejecutando escenarios financieros realistas que ejercen comportamientos tanto seguros como inseguros. Mostramos cómo el bucle de gobernanza responde de manera diferente a las transacciones que cumplen con las normas, las filtraciones de PII y las violaciones presupuestarias, al tiempo que produce resultados de auditoría transparentes. Consulta los CÓDIGOS COMPLETOS aquí.
if __name__ == "__main__": run_examples() print("nn" + "="*80) print(" 🎓 ¡TUTORIAL COMPLETO!") print("="*80) print("nConceptos clave demostrados:") print(" ✓ IA constitucional: gobernanza basada en reglas") print(" ✓ Sistema de agente dual: patrón trabajador + auditor") print(" ✓ Detección de infracciones de políticas: PII, presupuesto, autorización") print(" ✓ Bucle de revisión iterativa – Mecanismo de autocorrección") print(" ✓ Puntuación de riesgo – Evaluación cuantitativa de seguridad") print("nPróximos pasos:") print("• Reemplazar MockAIClient con API Anthropic real") print("• Implementar la lógica de revisión real en Worker Agent") print("• Agregar una detección de patrones más sofisticada") print("• Integrar con sistemas financieros reales") print("• Crear un panel de control y registro") print("="*80)
Concluimos el tutorial ejecutando todos los ejemplos y exponiendo claramente los conceptos centrales demostrados por el sistema. Recapitulamos cómo las reglas constitucionales, la gobernanza de doble agente, la detección de violaciones y la calificación de riesgos funcionan juntas en la práctica.
En conclusión, demostramos cómo hacer operativa la IA constitucional más allá de la teoría e incorporarla en los procesos de toma de decisiones financieras del mundo real. Ilustramos cómo detectamos y respondemos a las fugas de PII, los excesos presupuestarios y las justificaciones faltantes mientras cuantificamos el riesgo y aplicamos límites estrictos de gobernanza. Al orquestar ciclos de revisión iterativos entre trabajadores y agentes auditores, demostramos un plan práctico para construir sistemas financieros confiables, compatibles y escalables impulsados por IA donde la seguridad y la responsabilidad son objetivos de diseño de primera clase en lugar de ideas tardías.
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.
