Skip to content

AI Governance Platform: Arquitectura Técnica Completa

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:

  1. IA (Inteligencia Artificial) → Velocidad y automatización
  2. MPC (Multi-Party Computation) → Privacidad sin exposición
  3. Blockchain → Confianza y trazabilidad
# Core Framework
Flask 2.3.3
SQLAlchemy 2.0.21
MySQL 8.0
# AI & ML
openai>=2.6.1
langchain==0.1.0
tiktoken>=0.5.0
# MPC & Crypto
mpc-lib==2.1.0
shamir-secret-sharing==1.0.0
cryptography==41.0.4
pycryptodome==3.19.0
# Blockchain
web3==6.11.0
eth-account==0.10.0
py-solc-x==2.0.0
hyperledger-fabric-sdk==1.0.0
# Security
Flask-JWT-Extended==4.5.2
bcrypt
python-dotenv==1.0.1
{
"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"
}
}
  • 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
┌────────────────────────────────────────────────────────────────────────┐
│ 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 │
└──────────────────────┘ └──────────────┘ └────────────────────┘

Responsabilidad: Gestiona el ciclo de vida completo de las tareas IA.

backend/portfolio_app/services/ai_agent_service.py
from langchain import OpenAI, LLMChain
from langchain.prompts import PromptTemplate
from typing import Dict, Any, Tuple
import hashlib
import 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 confidence

Flujo 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│
└──────────────────┘

Responsabilidad: Procesar datos sensibles sin exponerlos completamente.

backend/portfolio_app/services/mpc_service.py
from shamir_secret_sharing import split_secret, reconstruct_secret
import requests
import asyncio
from typing import List, Dict, Any
import 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 │
└──────────────────────┘

Responsabilidad: Registro inmutable de todas las decisiones críticas.

backend/portfolio_app/services/blockchain_service.py
from web3 import Web3
from eth_account import Account
import json
from 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):

blockchain/contracts/AuditTrail.sol
// SPDX-License-Identifier: MIT
pragma 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;
}
}

Responsabilidad: Gestionar cola de aprobaciones y supervisión humana.

backend/portfolio_app/services/approval_service.py
from typing import Dict, Any, List
from datetime import datetime, timedelta
from ..models.tbl_human_approvals import HumanApproval
from ..models.tbl_ai_tasks import AITask
from .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.
pass
ArtículoRequirementImplementation
Art. 22Right not to be subject to automated decision-making✅ Human approval for low confidence
Art. 30Records of processing activities✅ Blockchain audit trail
Art. 32Security of processing✅ MPC + encryption + access control
Art. 33/34Breach notification✅ Automated alerts + audit logs
  • A.9: Access Control → RBAC integration
  • A.12: Operations Security → Audit logging
  • A.14: System Acquisition → Secure SDLC
  • A.18: Compliance → Automated compliance checks
# Redis caching for frequent queries
CACHE_CONFIG = {
'policy_cache_ttl': 300, # 5 minutes
'agent_config_cache_ttl': 600, # 10 minutes
'blockchain_query_cache_ttl': 60 # 1 minute
}
# Celery tasks for heavy computations
@celery.task
def process_large_dataset_with_mpc(data_id: str):
# Async MPC computation
pass
@celery.task
def generate_compliance_report(start_date: str, end_date: str):
# Async report generation
pass
# Prometheus metrics
ai_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')
# Structured logging
import 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
)
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=3

Esta arquitectura proporciona:

  1. Automatización confiable con supervisión humana
  2. Privacidad de datos mediante MPC
  3. Auditoría inmutable con blockchain
  4. Compliance con GDPR, ISO 27001, SOC 2
  5. Escalabilidad horizontal (MPC nodes, blockchain nodes)
  6. 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