AI Governance Platform: Arquitectura Técnica Completa
Resumen Ejecutivo
Section titled “Resumen Ejecutivo”La AI Governance Platform es una solución empresarial que permite automatizar operaciones críticas manteniendo:
- ✅ Control humano estratégico
- ✅ Privacidad de datos sensibles (GDPR/ISO)
- ✅ Auditoría inmutable de todas las decisiones
- ✅ Cumplimiento normativo automatizado
Esta arquitectura combina lo mejor de tres mundos:
- IA (Inteligencia Artificial) → Velocidad y automatización
- MPC (Multi-Party Computation) → Privacidad sin exposición
- Blockchain → Confianza y trazabilidad
Stack Tecnológico
Section titled “Stack Tecnológico”Backend
Section titled “Backend”# Core FrameworkFlask 2.3.3SQLAlchemy 2.0.21MySQL 8.0
# AI & MLopenai>=2.6.1langchain==0.1.0tiktoken>=0.5.0
# MPC & Cryptompc-lib==2.1.0shamir-secret-sharing==1.0.0cryptography==41.0.4pycryptodome==3.19.0
# Blockchainweb3==6.11.0eth-account==0.10.0py-solc-x==2.0.0hyperledger-fabric-sdk==1.0.0
# SecurityFlask-JWT-Extended==4.5.2bcryptpython-dotenv==1.0.1Frontend
Section titled “Frontend”{ "core": { "react": "^18.2.0", "@reduxjs/toolkit": "^1.9.7", "react-router-dom": "^6.15.0" }, "blockchain": { "web3": "^1.10.0", "ethers": "^6.9.0", "@web3-react/core": "^8.2.0" }, "ui": { "tailwindcss": "^3.3.3", "echarts-for-react": "^3.0.2", "@heroicons/react": "^2.2.0" }}Infrastructure
Section titled “Infrastructure”- Database: MySQL 8.0 (Docker)
- Blockchain: Hyperledger Fabric 2.5 (private network)
- MPC Nodes: Python 3.11 (distributed)
- API Gateway: Nginx 1.25
- Containerization: Docker Compose
Arquitectura de Alto Nivel
Section titled “Arquitectura de Alto Nivel”┌────────────────────────────────────────────────────────────────────────┐│ PRESENTATION LAYER ││ ││ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌───────────┐ ││ │ AI Agent │ │ Approval │ │ Compliance │ │ Policy │ ││ │ Dashboard │ │ Queue UI │ │ Dashboard │ │ Manager │ ││ └──────────────┘ └──────────────┘ └──────────────┘ └───────────┘ ││ ││ React + RTK Query + Tailwind CSS │└────────────────────────────────┬───────────────────────────────────────┘ │ HTTPS / WebSocket┌────────────────────────────────┴───────────────────────────────────────┐│ APPLICATION LAYER ││ ││ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌───────────┐ ││ │ AI Agent │ │ Approval │ │ Policy │ │ Task │ ││ │ Service │ │ Service │ │ Engine │ │ Queue │ ││ └──────────────┘ └──────────────┘ └──────────────┘ └───────────┘ ││ ││ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌───────────┐ ││ │ MPC │ │ Blockchain │ │ Audit │ │ Auth │ ││ │ Service │ │ Service │ │ Service │ │ Service │ ││ └──────────────┘ └──────────────┘ └──────────────┘ └───────────┘ ││ ││ Flask + Python 3.11 │└────────────────────────────────┬───────────────────────────────────────┘ │ ┌──────────────────┼──────────────────┐ │ │ │┌─────────────▼────────┐ ┌─────▼────────┐ ┌─────▼──────────────┐│ DATA LAYER │ │ BLOCKCHAIN │ │ MPC NETWORK ││ │ │ LAYER │ │ ││ ┌────────────────┐ │ │ │ │ ┌──────────────┐ ││ │ MySQL 8.0 │ │ │ Hyperledger │ │ │ MPC Node 1 │ ││ │ │ │ │ Fabric │ │ │ MPC Node 2 │ ││ │ - Users │ │ │ │ │ │ MPC Node 3 │ ││ │ - AI Tasks │ │ │ - Audit Log │ │ │ MPC Node 4 │ ││ │ - Approvals │ │ │ - Smart │ │ │ MPC Node 5 │ ││ │ - Policies │ │ │ Contracts │ │ └──────────────┘ ││ │ - Agents │ │ │ │ │ ││ └────────────────┘ │ │ │ │ 3-of-5 Threshold │└──────────────────────┘ └──────────────┘ └────────────────────┘Componentes Core
Section titled “Componentes Core”1. AI Agent Orchestrator
Section titled “1. AI Agent Orchestrator”Responsabilidad: Gestiona el ciclo de vida completo de las tareas IA.
from langchain import OpenAI, LLMChainfrom langchain.prompts import PromptTemplatefrom typing import Dict, Any, Tupleimport hashlibimport json
class AIAgentService: def __init__(self, agent_config: Dict[str, Any]): self.agent_id = agent_config['agent_id'] self.model = OpenAI( model_name=agent_config.get('model_name', 'gpt-4'), temperature=0.3 ) self.confidence_threshold = agent_config['confidence_threshold']
async def execute_task(self, task_data: Dict[str, Any]) -> Tuple[Any, float, bool]: """ Executes an AI task and returns: - result: The task output - confidence: Confidence score (0-1) - requires_approval: Whether human approval is needed """ # 1. Classify data sensitivity is_sensitive = self._classify_sensitivity(task_data)
# 2. Process with MPC if sensitive if is_sensitive: processed_data = await self._process_with_mpc(task_data) else: processed_data = task_data
# 3. Execute AI model result = await self._run_model(processed_data)
# 4. Calculate confidence score confidence = self._calculate_confidence(result)
# 5. Check policy compliance policy_check = await self._check_policies(task_data, result)
# 6. Determine if approval needed requires_approval = ( confidence < self.confidence_threshold or is_sensitive or not policy_check['compliant'] )
# 7. Log to blockchain tx_hash = await self._log_to_blockchain({ 'agent_id': self.agent_id, 'task_id': task_data['task_id'], 'confidence': confidence, 'requires_approval': requires_approval, 'timestamp': datetime.utcnow().isoformat() })
return result, confidence, requires_approval
def _classify_sensitivity(self, data: Dict) -> bool: """Classifies if data contains sensitive information""" sensitive_keywords = ['ssn', 'credit_card', 'password', 'email', 'phone'] data_str = json.dumps(data).lower() return any(keyword in data_str for keyword in sensitive_keywords)
async def _process_with_mpc(self, data: Dict) -> Dict: """Processes sensitive data using MPC""" from .mpc_service import MPCService mpc = MPCService() return await mpc.secure_compute(data)
def _calculate_confidence(self, result: Any) -> float: """Calculates confidence score based on model output""" # Implementation depends on model type # For classification: use probability # For generation: use perplexity/likelihood if hasattr(result, 'confidence'): return float(result.confidence) return 0.85 # Default confidenceFlujo de Ejecución:
┌─────────────────┐│ Task Received │└────────┬────────┘ │ ▼┌─────────────────────┐│ Classify Sensitivity│ ← PII detection└────────┬────────────┘ │ ├─────────► Sensitive? │ │ │ ▼ │ ┌──────────┐ │ │ MPC │ │ │ Process │ │ └────┬─────┘ │ │ ▼ ▼┌──────────────────────────┐│ Execute AI Model │└────────┬─────────────────┘ │ ▼┌──────────────────────────┐│ Calculate Confidence │└────────┬─────────────────┘ │ ▼┌──────────────────────────┐│ Check Policies │└────────┬─────────────────┘ │ ▼ Confidence < Threshold? Sensitive Data? Policy Violation? │ ┌────┴────┐ │ YES │ NO ▼ ▼┌─────────┐ ┌──────────┐│ Approval│ │ Execute ││ Queue │ │ Directly │└─────────┘ └──────────┘ │ │ └──────┬──────┘ ▼ ┌──────────────────┐ │ Log to Blockchain│ └──────────────────┘2. MPC (Multi-Party Computation) Service
Section titled “2. MPC (Multi-Party Computation) Service”Responsabilidad: Procesar datos sensibles sin exponerlos completamente.
from shamir_secret_sharing import split_secret, reconstruct_secretimport requestsimport asynciofrom typing import List, Dict, Anyimport json
class MPCService: def __init__(self): self.nodes = [ {'id': 1, 'url': 'http://mpc-node-1:8000'}, {'id': 2, 'url': 'http://mpc-node-2:8000'}, {'id': 3, 'url': 'http://mpc-node-3:8000'}, {'id': 4, 'url': 'http://mpc-node-4:8000'}, {'id': 5, 'url': 'http://mpc-node-5:8000'}, ] self.threshold = 3 self.total_shares = 5
async def secure_compute(self, sensitive_data: Dict[str, Any]) -> Dict[str, Any]: """ Performs computation on sensitive data using MPC Returns: Computed result without exposing original data """ # 1. Convert data to secret secret = json.dumps(sensitive_data).encode('utf-8')
# 2. Split secret into shares using Shamir Secret Sharing shares = split_secret(secret, self.threshold, self.total_shares)
# 3. Distribute shares to MPC nodes distribution_tasks = [ self._send_share_to_node(node, share) for node, share in zip(self.nodes, shares) ] await asyncio.gather(*distribution_tasks)
# 4. Request computation from nodes computation_tasks = [ self._request_computation(node, computation_type='aggregate') for node in self.nodes ] results = await asyncio.gather(*computation_tasks)
# 5. Reconstruct result from threshold shares valid_results = [r for r in results if r['success']][:self.threshold] final_result = self._reconstruct_result(valid_results)
# 6. Generate Zero-Knowledge Proof zkp = self._generate_zkp(final_result)
return { 'result': final_result, 'proof': zkp, 'nodes_used': len(valid_results), 'threshold': self.threshold }
async def _send_share_to_node(self, node: Dict, share: bytes): """Sends a secret share to an MPC node""" async with aiohttp.ClientSession() as session: async with session.post( f"{node['url']}/receive_share", json={'share': share.hex()} ) as response: return await response.json()
async def _request_computation(self, node: Dict, computation_type: str): """Requests computation from an MPC node""" async with aiohttp.ClientSession() as session: async with session.post( f"{node['url']}/compute", json={'type': computation_type} ) as response: return await response.json()
def _reconstruct_result(self, partial_results: List[Dict]) -> Any: """Reconstructs final result from partial computations""" shares = [bytes.fromhex(r['share']) for r in partial_results] reconstructed = reconstruct_secret(shares, self.threshold) return json.loads(reconstructed.decode('utf-8'))
def _generate_zkp(self, result: Any) -> Dict: """Generates Zero-Knowledge Proof of correct computation""" # Simplified ZKP - in production use zk-SNARKs import hashlib result_hash = hashlib.sha256(json.dumps(result).encode()).hexdigest() return { 'type': 'sha256', 'commitment': result_hash, 'timestamp': datetime.utcnow().isoformat() }MPC en Acción:
┌───────────────────┐│ Sensitive Data ││ (e.g., SSN, PII) │└────────┬──────────┘ │ ▼┌────────────────────┐│ Shamir Secret │ ← Split into 5 shares│ Sharing (5,3) │ Need 3 to reconstruct└────────┬───────────┘ │ ┌────┴────┬────┬────┬────┐ │ │ │ │ │ ▼ ▼ ▼ ▼ ▼┌─────┐ ┌─────┐ ... ┌─────┐│Node1│ │Node2│ │Node5││Share│ │Share│ │Share││ 1/5 │ │ 2/5 │ │ 5/5 │└──┬──┘ └──┬──┘ └──┬──┘ │ │ │ └────┬────┴────┬───────┘ │ │ ▼ ▼ ┌──────────────────────┐ │ Compute on shares │ │ (never see full data)│ └──────────┬───────────┘ │ ▼ ┌──────────────────────┐ │ Reconstruct Result │ │ (3 of 5 shares) │ └──────────┬───────────┘ │ ▼ ┌──────────────────────┐ │ Final Result + ZKP │ └──────────────────────┘3. Blockchain Service
Section titled “3. Blockchain Service”Responsabilidad: Registro inmutable de todas las decisiones críticas.
from web3 import Web3from eth_account import Accountimport jsonfrom typing import Dict, Any
class BlockchainService: def __init__(self): # Connect to Hyperledger Fabric or Ethereum node self.w3 = Web3(Web3.HTTPProvider('http://blockchain:7051')) self.contract_address = os.getenv('SMART_CONTRACT_ADDRESS') self.contract_abi = self._load_contract_abi() self.contract = self.w3.eth.contract( address=self.contract_address, abi=self.contract_abi ) self.account = Account.from_key(os.getenv('BLOCKCHAIN_PRIVATE_KEY'))
async def log_decision(self, event_data: Dict[str, Any]) -> str: """ Logs a decision to blockchain Returns: Transaction hash """ # 1. Hash sensitive data (don't store on-chain) data_hash = self._hash_data(event_data)
# 2. Prepare transaction tx = self.contract.functions.logDecision( eventType=event_data['event_type'], actorId=event_data['actor_id'], dataHash=data_hash, confidence=int(event_data.get('confidence', 0) * 10000), requiresApproval=event_data.get('requires_approval', False), timestamp=int(datetime.utcnow().timestamp()) ).build_transaction({ 'from': self.account.address, 'nonce': self.w3.eth.get_transaction_count(self.account.address), 'gas': 200000, 'gasPrice': self.w3.eth.gas_price })
# 3. Sign transaction signed_tx = self.w3.eth.account.sign_transaction(tx, self.account.key)
# 4. Send transaction tx_hash = self.w3.eth.send_raw_transaction(signed_tx.rawTransaction)
# 5. Wait for receipt (optional - can be async) receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)
# 6. Store local index in MySQL await self._index_transaction(tx_hash.hex(), event_data, receipt)
return tx_hash.hex()
async def verify_decision(self, tx_hash: str) -> Dict[str, Any]: """ Verifies a decision by querying blockchain Returns: Verification result with proof """ receipt = self.w3.eth.get_transaction_receipt(tx_hash) block = self.w3.eth.get_block(receipt['blockNumber'])
# Decode event logs event = self.contract.events.DecisionLogged().process_receipt(receipt)[0]
return { 'valid': True, 'block_number': receipt['blockNumber'], 'block_hash': block['hash'].hex(), 'timestamp': block['timestamp'], 'event_data': dict(event['args']), 'confirmations': self.w3.eth.block_number - receipt['blockNumber'] }
def _hash_data(self, data: Dict) -> str: """Creates SHA-256 hash of data""" import hashlib data_str = json.dumps(data, sort_keys=True) return hashlib.sha256(data_str.encode()).hexdigest()
async def _index_transaction(self, tx_hash: str, event_data: Dict, receipt: Dict): """Stores transaction in local MySQL for fast queries""" from ..models.tbl_blockchain_audit import BlockchainAudit from ..extensions import db
audit_record = BlockchainAudit( event_type=event_data['event_type'], entity_id=event_data.get('entity_id'), actor_id=event_data.get('actor_id'), action=event_data.get('action'), data_hash=self._hash_data(event_data), blockchain_tx_hash=tx_hash, block_number=receipt['blockNumber'] ) db.session.add(audit_record) await db.session.commit()Smart Contract (Solidity):
// SPDX-License-Identifier: MITpragma solidity ^0.8.0;
contract AuditTrail { struct Decision { string eventType; uint256 actorId; bytes32 dataHash; uint16 confidence; // 0-10000 (0.00% - 100.00%) bool requiresApproval; uint256 timestamp; address logger; }
mapping(bytes32 => Decision) public decisions; bytes32[] public decisionHashes;
event DecisionLogged( bytes32 indexed decisionHash, string eventType, uint256 indexed actorId, uint256 timestamp );
function logDecision( string memory eventType, uint256 actorId, bytes32 dataHash, uint16 confidence, bool requiresApproval, uint256 timestamp ) public returns (bytes32) { bytes32 decisionHash = keccak256( abi.encodePacked( eventType, actorId, dataHash, timestamp, msg.sender ) );
require(decisions[decisionHash].timestamp == 0, "Decision already logged");
decisions[decisionHash] = Decision({ eventType: eventType, actorId: actorId, dataHash: dataHash, confidence: confidence, requiresApproval: requiresApproval, timestamp: timestamp, logger: msg.sender });
decisionHashes.push(decisionHash);
emit DecisionLogged(decisionHash, eventType, actorId, timestamp);
return decisionHash; }
function getDecision(bytes32 decisionHash) public view returns (Decision memory) { return decisions[decisionHash]; }
function getTotalDecisions() public view returns (uint256) { return decisionHashes.length; }}4. Human Approval Service
Section titled “4. Human Approval Service”Responsabilidad: Gestionar cola de aprobaciones y supervisión humana.
from typing import Dict, Any, Listfrom datetime import datetime, timedeltafrom ..models.tbl_human_approvals import HumanApprovalfrom ..models.tbl_ai_tasks import AITaskfrom .blockchain_service import BlockchainService
class ApprovalService: def __init__(self): self.blockchain = BlockchainService()
async def create_approval_request( self, task_id: str, assigned_to: int, original_output: Any ) -> str: """ Creates a new approval request Returns: approval_id """ approval = HumanApproval( approval_id=self._generate_uuid(), task_id=task_id, assigned_to=assigned_to, status='pending', original_output=json.dumps(original_output), created_at=datetime.utcnow() )
db.session.add(approval) await db.session.commit()
# Send notification (email/webhook) await self._notify_supervisor(approval)
return approval.approval_id
async def approve( self, approval_id: str, supervisor_id: int, modifications: Dict = None, justification: str = "" ) -> Dict[str, Any]: """ Approves a decision (with optional modifications) """ approval = await HumanApproval.query.get(approval_id)
if not approval: raise ValueError("Approval not found")
if approval.assigned_to != supervisor_id: raise PermissionError("Not authorized to approve this task")
# Update approval record approval.status = 'approved' if not modifications else 'modified' approval.modified_output = json.dumps(modifications) if modifications else None approval.justification = justification approval.approved_at = datetime.utcnow()
# Log to blockchain tx_hash = await self.blockchain.log_decision({ 'event_type': 'human_approval', 'actor_id': supervisor_id, 'entity_id': approval_id, 'action': 'approve', 'modifications': bool(modifications), 'timestamp': datetime.utcnow().isoformat() })
approval.blockchain_tx_hash = tx_hash
# Update task status task = await AITask.query.get(approval.task_id) task.status = 'approved' task.blockchain_tx_hash = tx_hash
await db.session.commit()
return { 'approval_id': approval_id, 'status': 'approved', 'tx_hash': tx_hash, 'modifications_made': bool(modifications) }
async def reject( self, approval_id: str, supervisor_id: int, justification: str ) -> Dict[str, Any]: """ Rejects a decision """ approval = await HumanApproval.query.get(approval_id)
approval.status = 'rejected' approval.justification = justification approval.approved_at = datetime.utcnow()
# Log to blockchain tx_hash = await self.blockchain.log_decision({ 'event_type': 'human_rejection', 'actor_id': supervisor_id, 'entity_id': approval_id, 'action': 'reject', 'justification': justification })
approval.blockchain_tx_hash = tx_hash
# Update task task = await AITask.query.get(approval.task_id) task.status = 'rejected'
await db.session.commit()
return { 'approval_id': approval_id, 'status': 'rejected', 'tx_hash': tx_hash }
async def get_pending_approvals(self, supervisor_id: int) -> List[Dict]: """ Gets all pending approvals for a supervisor """ approvals = await HumanApproval.query.filter_by( assigned_to=supervisor_id, status='pending' ).order_by(HumanApproval.created_at.asc()).all()
return [self._serialize_approval(a) for a in approvals]
async def _notify_supervisor(self, approval: HumanApproval): """Sends notification to supervisor""" # Implementation: Email, Slack, webhook, etc. passSeguridad y Compliance
Section titled “Seguridad y Compliance”GDPR Compliance
Section titled “GDPR Compliance”| Artículo | Requirement | Implementation |
|---|---|---|
| Art. 22 | Right not to be subject to automated decision-making | ✅ Human approval for low confidence |
| Art. 30 | Records of processing activities | ✅ Blockchain audit trail |
| Art. 32 | Security of processing | ✅ MPC + encryption + access control |
| Art. 33/34 | Breach notification | ✅ Automated alerts + audit logs |
ISO 27001 Controls
Section titled “ISO 27001 Controls”- A.9: Access Control → RBAC integration
- A.12: Operations Security → Audit logging
- A.14: System Acquisition → Secure SDLC
- A.18: Compliance → Automated compliance checks
Performance Optimization
Section titled “Performance Optimization”Caching Strategy
Section titled “Caching Strategy”# Redis caching for frequent queriesCACHE_CONFIG = { 'policy_cache_ttl': 300, # 5 minutes 'agent_config_cache_ttl': 600, # 10 minutes 'blockchain_query_cache_ttl': 60 # 1 minute}Async Processing
Section titled “Async Processing”# Celery tasks for heavy computations@celery.taskdef process_large_dataset_with_mpc(data_id: str): # Async MPC computation pass
@celery.taskdef generate_compliance_report(start_date: str, end_date: str): # Async report generation passMonitoring & Observability
Section titled “Monitoring & Observability”Key Metrics
Section titled “Key Metrics”# Prometheus metricsai_tasks_total = Counter('ai_tasks_total', 'Total AI tasks processed')ai_tasks_approved = Counter('ai_tasks_approved', 'Tasks requiring approval')mpc_operations_duration = Histogram('mpc_operations_duration_seconds', 'MPC operation duration')blockchain_tx_duration = Histogram('blockchain_tx_duration_seconds', 'Blockchain transaction duration')Logging
Section titled “Logging”# Structured loggingimport structlog
log = structlog.get_logger()
log.info( "ai_task_executed", task_id=task_id, agent_id=agent_id, confidence=confidence, requires_approval=requires_approval, duration_ms=duration)Deployment
Section titled “Deployment”Docker Compose
Section titled “Docker Compose”version: '3.8'
services: backend: # ... existing backend config environment: - BLOCKCHAIN_URL=http://blockchain:7051 - MPC_NODES=http://mpc-node-1:8000,http://mpc-node-2:8000,http://mpc-node-3:8000
blockchain: image: hyperledger/fabric-peer:2.5 ports: - "7051:7051" volumes: - ./blockchain/config:/etc/hyperledger/fabric - ./blockchain/data:/var/hyperledger/production
mpc-node-1: build: ./mpc-nodes environment: - NODE_ID=1 - MPC_THRESHOLD=3
mpc-node-2: build: ./mpc-nodes environment: - NODE_ID=2 - MPC_THRESHOLD=3
mpc-node-3: build: ./mpc-nodes environment: - NODE_ID=3 - MPC_THRESHOLD=3Conclusión
Section titled “Conclusión”Esta arquitectura proporciona:
- ✅ Automatización confiable con supervisión humana
- ✅ Privacidad de datos mediante MPC
- ✅ Auditoría inmutable con blockchain
- ✅ Compliance con GDPR, ISO 27001, SOC 2
- ✅ Escalabilidad horizontal (MPC nodes, blockchain nodes)
- ✅ Trazabilidad completa de todas las decisiones
Next Steps:
- Implementar módulos core
- Configurar blockchain privada
- Desplegar MPC nodes
- Integrar con frontend React
- Testing exhaustivo de seguridad
Autor: Joseph Ruiz (ruizdev7)
Fecha: 22 de Noviembre, 2024
Versión: 1.0