Edge Computing para IA: Ejecutando Modelos Donde Importan

import onnxruntime as ort
from onnxruntime.quantization import quantize_dynamic, QuantType
 
# Cuantización dinámica — el punto de inicio más fácil
# Convierte pesos a INT8, mantiene activaciones en FP32
# Típicamente 2-4x de aceleración con <2% de pérdida de precisión
quantize_dynamic(
    model_input="model_fp32.onnx",
    model_output="model_int8.onnx",
    weight_type=QuantType.QInt8
)
 
# Cuantización estática — mejor rendimiento, más trabajo
# Requiere un dataset de calibración para determinar factores de escala óptimos
from onnxruntime.quantization import quantize_static, CalibrationDataReader
 
class MyCalibrationReader(CalibrationDataReader):
    """Alimenta datos representativos para determinar parámetros de cuantización."""
 
    def __init__(self, calibration_data: list):
        self.data = iter(calibration_data)
 
    def get_next(self):
        try:
            sample = next(self.data)
            return {"input": sample}
        except StopIteration:
            return None
 
# Usa 100-500 muestras representativas
calibration_reader = MyCalibrationReader(calibration_samples)
 
quantize_static(
    model_input="model_fp32.onnx",
    model_output="model_int8_static.onnx",
    calibration_data_reader=calibration_reader,
)

import torch
import torch.nn.utils.prune as prune
 
# Poda no estructurada — elimina pesos individuales
# Bueno para: reducción general de tamaño
model = load_your_model()
 
for name, module in model.named_modules():
    if isinstance(module, torch.nn.Linear):
        # Elimina 30% de los pesos con menor magnitud
        prune.l1_unstructured(module, name='weight', amount=0.3)
 
# Poda estructurada — elimina neuronas/canales completos
# Bueno para: aceleración real (no solo archivos más chicos)
for name, module in model.named_modules():
    if isinstance(module, torch.nn.Conv2d):
        # Elimina 20% de los canales de salida
        prune.ln_structured(
            module, name='weight', amount=0.2, n=2, dim=0
        )
 
# Hacer la poda permanente (eliminar las máscaras)
for name, module in model.named_modules():
    if isinstance(module, (torch.nn.Linear, torch.nn.Conv2d)):
        prune.remove(module, 'weight')
 
# Fine-tune por unas cuantas epochs para recuperar precisión
# Este paso importa — no te lo saltes
fine_tune(model, train_loader, epochs=5, lr=1e-5)

import torch
import torch.nn.functional as F
 
def distillation_loss(
    student_logits: torch.Tensor,
    teacher_logits: torch.Tensor,
    labels: torch.Tensor,
    temperature: float = 3.0,
    alpha: float = 0.7
) -> torch.Tensor:
    """
    Loss combinado: soft targets del profesor + hard targets de las etiquetas.
 
    temperature: Mayor = distribución de probabilidad más suave = más
                 transferencia de conocimiento. 3-5 funciona bien en general.
    alpha: Balance entre conocimiento del profesor y verdad fundamental.
           0.7 significa 70% profesor, 30% etiquetas duras.
    """
    # Soft loss — aprender de la distribución de probabilidad del profesor
    soft_loss = F.kl_div(
        F.log_softmax(student_logits / temperature, dim=-1),
        F.softmax(teacher_logits / temperature, dim=-1),
        reduction='batchmean'
    ) * (temperature ** 2)
 
    # Hard loss — cross-entropy estándar con la verdad fundamental
    hard_loss = F.cross_entropy(student_logits, labels)
 
    return alpha * soft_loss + (1 - alpha) * hard_loss
 
# Loop de entrenamiento
teacher_model.eval()  # El profesor está congelado
student_model.train()
 
for batch in train_loader:
    inputs, labels = batch
 
    with torch.no_grad():
        teacher_logits = teacher_model(inputs)
 
    student_logits = student_model(inputs)
 
    loss = distillation_loss(student_logits, teacher_logits, labels)
    loss.backward()
    optimizer.step()
    optimizer.zero_grad()

import torch
import onnx
 
# Exportar modelo de PyTorch a ONNX
model = load_your_trained_model()
model.eval()
 
# Crear input dummy que coincida con la forma esperada del modelo
dummy_input = torch.randn(1, 3, 224, 224)  # batch, channels, H, W
 
torch.onnx.export(
    model,
    dummy_input,
    "model.onnx",
    export_params=True,
    opset_version=17,             # Usar el último opset estable
    do_constant_folding=True,     # Optimizar expresiones constantes
    input_names=["input"],
    output_names=["output"],
    dynamic_axes={                 # Permitir tamaño de batch variable
        "input": {0: "batch_size"},
        "output": {0: "batch_size"}
    }
)
 
# Verificar el modelo exportado
onnx_model = onnx.load("model.onnx")
onnx.checker.check_model(onnx_model)
print(f"Modelo exportado: {onnx_model.graph.input[0].type}")

// TypeScript — servidor edge en Node.js o app Electron
import * as ort from 'onnxruntime-node';
 
interface PredictionResult {
  label: string;
  confidence: number;
  inferenceTimeMs: number;
}
 
async function runEdgeInference(
  modelPath: string,
  inputData: Float32Array
): Promise<PredictionResult> {
  // Crear sesión con flags de optimización
  const session = await ort.InferenceSession.create(modelPath, {
    executionProviders: ['CUDAExecutionProvider', 'CPUExecutionProvider'],
    graphOptimizationLevel: 'all',
    enableCpuMemArena: true,
    enableMemPattern: true,
  });
 
  // Crear tensor de entrada
  const inputTensor = new ort.Tensor('float32', inputData, [1, 3, 224, 224]);
 
  // Ejecutar inferencia y medir tiempo
  const start = performance.now();
  const results = await session.run({ input: inputTensor });
  const inferenceTime = performance.now() - start;
 
  // Procesar salida
  const output = results.output.data as Float32Array;
  const maxIdx = output.indexOf(Math.max(...output));
 
  return {
    label: LABELS[maxIdx],
    confidence: output[maxIdx],
    inferenceTimeMs: Math.round(inferenceTime * 100) / 100,
  };
}
 
// Caché de sesión — ¡no recargues el modelo en cada request!
const sessionCache = new Map<string, ort.InferenceSession>();
 
async function getOrCreateSession(
  modelPath: string
): Promise<ort.InferenceSession> {
  if (!sessionCache.has(modelPath)) {
    const session = await ort.InferenceSession.create(modelPath, {
      executionProviders: ['CPUExecutionProvider'],
      graphOptimizationLevel: 'all',
      intraOpNumThreads: 4,       // Ajustar según los cores del dispositivo
      interOpNumThreads: 2,
    });
    sessionCache.set(modelPath, session);
  }
  return sessionCache.get(modelPath)!;
}

import tensorrt as trt
import numpy as np
 
# Convertir ONNX a engine de TensorRT
def build_tensorrt_engine(
    onnx_path: str,
    engine_path: str,
    precision: str = "fp16",      # "fp32", "fp16", o "int8"
    max_batch_size: int = 1,
    workspace_gb: float = 1.0
) -> None:
    """Construir un engine de TensorRT optimizado desde un modelo ONNX."""
    logger = trt.Logger(trt.Logger.WARNING)
    builder = trt.Builder(logger)
    network = builder.create_network(
        1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
    )
    parser = trt.OnnxParser(network, logger)
 
    # Parsear modelo ONNX
    with open(onnx_path, 'rb') as f:
        if not parser.parse(f.read()):
            for i in range(parser.num_errors):
                print(f"Error de parseo: {parser.get_error(i)}")
            raise RuntimeError("Falló el parseo del modelo ONNX")
 
    # Configurar builder
    config = builder.create_builder_config()
    config.set_memory_pool_limit(
        trt.MemoryPoolType.WORKSPACE,
        int(workspace_gb * (1 << 30))
    )
 
    # Establecer precisión
    if precision == "fp16":
        if builder.platform_has_fast_fp16:
            config.set_flag(trt.BuilderFlag.FP16)
            print("Usando precisión FP16")
        else:
            print("FP16 no soportado, cayendo a FP32")
    elif precision == "int8":
        if builder.platform_has_fast_int8:
            config.set_flag(trt.BuilderFlag.INT8)
            # INT8 requiere calibración — ver sección de cuantización
            config.int8_calibrator = create_calibrator(calibration_data)
            print("Usando precisión INT8")
 
    # Establecer formas dinámicas
    profile = builder.create_optimization_profile()
    input_shape = network.get_input(0).shape
    profile.set_shape(
        network.get_input(0).name,
        min=(1, *input_shape[1:]),
        opt=(max_batch_size // 2, *input_shape[1:]),
        max=(max_batch_size, *input_shape[1:])
    )
    config.add_optimization_profile(profile)
 
    # Construir engine (esto toma minutos — hazlo una vez, guarda el resultado)
    print("Construyendo engine de TensorRT (esto toma un rato)...")
    serialized_engine = builder.build_serialized_network(network, config)
 
    with open(engine_path, 'wb') as f:
        f.write(serialized_engine)
 
    print(f"Engine guardado en {engine_path}")

#!/usr/bin/env python3
"""Servicio de inferencia edge para dispositivos NVIDIA Jetson."""
 
import asyncio
import time
import logging
from pathlib import Path
from dataclasses import dataclass
from contextlib import asynccontextmanager
 
import numpy as np
import onnxruntime as ort
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
 
logger = logging.getLogger(__name__)
 
@dataclass
class ModelConfig:
    model_path: str
    input_name: str = "input"
    input_shape: tuple = (1, 3, 224, 224)
    device: str = "cuda"     # "cuda" para GPU Jetson, "cpu" para fallback
    num_threads: int = 4
 
class InferenceRequest(BaseModel):
    data: list[float]
 
class InferenceResponse(BaseModel):
    predictions: list[float]
    inference_ms: float
    device: str
    model_version: str
 
class EdgeInferenceEngine:
    """Gestiona el ciclo de vida del modelo e inferencia en dispositivos edge."""
 
    def __init__(self, config: ModelConfig):
        self.config = config
        self.session = None
        self.model_version = "unknown"
        self._inference_count = 0
        self._total_inference_ms = 0.0
 
    def load(self) -> None:
        """Cargar modelo con el execution provider apropiado."""
        providers = []
 
        if self.config.device == "cuda":
            providers.append(('CUDAExecutionProvider', {
                'device_id': 0,
                'arena_extend_strategy': 'kSameAsRequested',
                'gpu_mem_limit': 512 * 1024 * 1024,  # 512MB — dejar espacio
                'cudnn_conv_algo_search': 'HEURISTIC',
            }))
 
        providers.append(('CPUExecutionProvider', {
            'arena_extend_strategy': 'kSameAsRequested',
        }))
 
        session_options = ort.SessionOptions()
        session_options.graph_optimization_level = (
            ort.GraphOptimizationLevel.ORT_ENABLE_ALL
        )
        session_options.intra_op_num_threads = self.config.num_threads
        session_options.enable_mem_pattern = True
 
        self.session = ort.InferenceSession(
            self.config.model_path,
            sess_options=session_options,
            providers=providers
        )
 
        actual_provider = self.session.get_providers()[0]
        logger.info(f"Modelo cargado en {actual_provider}")
 
        # Corrida de calentamiento — la primera inferencia siempre es lenta
        dummy = np.random.randn(*self.config.input_shape).astype(np.float32)
        self.session.run(None, {self.config.input_name: dummy})
        logger.info("Inferencia de calentamiento completa")
 
    def predict(self, input_data: np.ndarray) -> tuple[np.ndarray, float]:
        """Ejecutar inferencia y devolver (output, time_ms)."""
        if self.session is None:
            raise RuntimeError("Modelo no cargado")
 
        start = time.perf_counter()
        outputs = self.session.run(
            None,
            {self.config.input_name: input_data}
        )
        elapsed_ms = (time.perf_counter() - start) * 1000
 
        self._inference_count += 1
        self._total_inference_ms += elapsed_ms
 
        return outputs[0], elapsed_ms
 
    @property
    def avg_inference_ms(self) -> float:
        if self._inference_count == 0:
            return 0.0
        return self._total_inference_ms / self._inference_count
 
# --- Servicio FastAPI ---
engine = EdgeInferenceEngine(ModelConfig(
    model_path="/models/anomaly_detector_int8.onnx",
    device="cuda"
))
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    engine.load()
    yield
    logger.info(f"Apagando. Inferencia prom: {engine.avg_inference_ms:.2f}ms")
 
app = FastAPI(title="Servicio de Inferencia Edge", lifespan=lifespan)
 
@app.post("/predict", response_model=InferenceResponse)
async def predict(request: InferenceRequest):
    try:
        input_array = np.array(
            request.data, dtype=np.float32
        ).reshape(engine.config.input_shape)
 
        predictions, inference_ms = engine.predict(input_array)
 
        return InferenceResponse(
            predictions=predictions.flatten().tolist(),
            inference_ms=round(inference_ms, 2),
            device=engine.session.get_providers()[0],
            model_version=engine.model_version,
        )
    except Exception as e:
        logger.error(f"Inferencia falló: {e}")
        raise HTTPException(status_code=500, detail=str(e))
 
@app.get("/health")
async def health():
    return {
        "status": "healthy",
        "inference_count": engine._inference_count,
        "avg_inference_ms": round(engine.avg_inference_ms, 2),
        "provider": engine.session.get_providers()[0] if engine.session else None,
    }

# Despliegue en Raspberry Pi — solo CPU o con Coral USB Accelerator
# Restricciones clave:
#   - 4-8GB RAM (el modelo debe caber con espacio para el SO)
#   - CPU ARM Cortex (sin GPU)
#   - Thermal throttling sobre ~80°C
 
import onnxruntime as ort
import numpy as np
 
def create_pi_session(model_path: str) -> ort.InferenceSession:
    """Crear una sesión optimizada para Raspberry Pi."""
    options = ort.SessionOptions()
 
    # Crítico: limitar threads a cores físicos
    # Pi 5 tiene 4 cores — usar más causa thrashing
    options.intra_op_num_threads = 4
    options.inter_op_num_threads = 1
 
    # Habilitar todas las optimizaciones de grafo
    options.graph_optimization_level = (
        ort.GraphOptimizationLevel.ORT_ENABLE_ALL
    )
 
    # Optimización de memoria — importante en Pi de 4GB
    options.enable_cpu_mem_arena = True
    options.enable_mem_pattern = True
 
    session = ort.InferenceSession(
        model_path,
        sess_options=options,
        providers=['CPUExecutionProvider']
    )
 
    return session
 
# Pro tip: monitorear temperatura y reducir velocidad si hay throttling
def get_cpu_temp() -> float:
    """Leer temperatura del CPU del Pi."""
    with open('/sys/class/thermal/thermal_zone0/temp', 'r') as f:
        return float(f.read().strip()) / 1000.0
 
def should_throttle() -> bool:
    """Verificar si debemos reducir la frecuencia de inferencia."""
    temp = get_cpu_temp()
    if temp > 75.0:
        logger.warning(f"Temp CPU {temp}°C — reduciendo inferencia")
        return True
    return False

// Inferencia en el navegador con ONNX Runtime Web
// Funciona en todos los navegadores modernos via WebAssembly
 
import * as ort from 'onnxruntime-web';
 
// Configurar backend WASM
ort.env.wasm.numThreads = navigator.hardwareConcurrency || 4;
ort.env.wasm.simd = true;
 
interface BrowserInferenceConfig {
  modelUrl: string;            // URL o ruta al archivo .onnx
  executionProvider: 'wasm' | 'webgl' | 'webgpu';
  inputShape: number[];
}
 
class BrowserInferenceEngine {
  private session: ort.InferenceSession | null = null;
  private config: BrowserInferenceConfig;
 
  constructor(config: BrowserInferenceConfig) {
    this.config = config;
  }
 
  async initialize(): Promise<void> {
    // Cargar modelo — esto descarga y compila el módulo WASM
    // Muestra un indicador de carga porque esto toma 1-5 segundos
    this.session = await ort.InferenceSession.create(
      this.config.modelUrl,
      {
        executionProviders: [this.config.executionProvider],
        graphOptimizationLevel: 'all',
      }
    );
 
    // Corrida de calentamiento
    const dummy = new Float32Array(
      this.config.inputShape.reduce((a, b) => a * b, 1)
    );
    const tensor = new ort.Tensor('float32', dummy, this.config.inputShape);
    await this.session.run({ input: tensor });
 
    console.log('Modelo cargado y calentado');
  }
 
  async predict(inputData: Float32Array): Promise<{
    result: Float32Array;
    timeMs: number;
  }> {
    if (!this.session) throw new Error('Modelo no inicializado');
 
    const tensor = new ort.Tensor(
      'float32', inputData, this.config.inputShape
    );
 
    const start = performance.now();
    const output = await this.session.run({ input: tensor });
    const timeMs = performance.now() - start;
 
    const resultKey = this.session.outputNames[0];
    return {
      result: output[resultKey].data as Float32Array,
      timeMs: Math.round(timeMs * 100) / 100,
    };
  }
 
  dispose(): void {
    this.session?.release();
    this.session = null;
  }
}
 
// Uso en un componente React
// const engine = new BrowserInferenceEngine({
//   modelUrl: '/models/classifier_int8.onnx',
//   executionProvider: 'wasm',
//   inputShape: [1, 3, 224, 224],
// });
// await engine.initialize();
// const { result, timeMs } = await engine.predict(preprocessedImage);

import time
import json
import logging
from dataclasses import dataclass, field, asdict
from collections import deque
from typing import Optional
 
@dataclass
class InferenceMetric:
    timestamp: float
    inference_ms: float
    input_hash: str              # Para detectar data drift
    prediction: list[float]
    confidence: float
    model_version: str
    device_id: str
    cpu_temp: Optional[float] = None
    memory_usage_mb: Optional[float] = None
 
class EdgeTelemetryCollector:
    """Recolecta y agrupa métricas de inferencia edge para subir."""
 
    def __init__(
        self,
        device_id: str,
        buffer_size: int = 1000,
        upload_interval_s: int = 300,  # Subir cada 5 minutos
    ):
        self.device_id = device_id
        self.buffer: deque[InferenceMetric] = deque(maxlen=buffer_size)
        self.upload_interval = upload_interval_s
        self._last_upload = time.time()
 
        # Estadísticas rolling para alertas locales
        self._recent_latencies: deque[float] = deque(maxlen=100)
        self._recent_confidences: deque[float] = deque(maxlen=100)
 
    def record(self, metric: InferenceMetric) -> None:
        """Registrar una métrica individual de inferencia."""
        self.buffer.append(metric)
        self._recent_latencies.append(metric.inference_ms)
        self._recent_confidences.append(metric.confidence)
 
        # Detección de anomalías local — no esperar al cloud
        self._check_local_alerts(metric)
 
    def _check_local_alerts(self, metric: InferenceMetric) -> None:
        """Detectar problemas localmente sin necesitar conectividad al cloud."""
        # Detección de picos de latencia
        if len(self._recent_latencies) >= 10:
            avg = sum(self._recent_latencies) / len(self._recent_latencies)
            if metric.inference_ms > avg * 3:
                logging.warning(
                    f"Pico de latencia: {metric.inference_ms:.1f}ms "
                    f"(prom: {avg:.1f}ms) — posible thermal throttling"
                )
 
        # Detección de drift de confianza
        if len(self._recent_confidences) >= 50:
            avg_conf = (
                sum(self._recent_confidences)
                / len(self._recent_confidences)
            )
            if avg_conf < 0.5:
                logging.warning(
                    f"Confianza promedio baja: {avg_conf:.2f} — "
                    "posible data drift o degradación del modelo"
                )
 
        # Alerta de temperatura
        if metric.cpu_temp and metric.cpu_temp > 80:
            logging.warning(
                f"Temp CPU alta: {metric.cpu_temp}°C — "
                "degradación de rendimiento probable"
            )
 
    def should_upload(self) -> bool:
        """Verificar si es hora de subir métricas almacenadas."""
        return (
            time.time() - self._last_upload > self.upload_interval
            and len(self.buffer) > 0
        )
 
    def get_upload_batch(self) -> list[dict]:
        """Obtener métricas para subir y limpiar buffer."""
        batch = [asdict(m) for m in self.buffer]
        self.buffer.clear()
        self._last_upload = time.time()
        return batch

# A veces la mejor estrategia edge es... no hacer edge en absoluto.
# Aquí va un patrón inteligente de fallback:
 
import aiohttp
import asyncio
 
async def hybrid_inference(
    input_data: np.ndarray,
    edge_engine: EdgeInferenceEngine,
    cloud_url: str,
    cloud_timeout_ms: float = 200,
) -> dict:
    """
    Intenta edge primero. Cae a cloud si edge no está saludable.
    Esto te da lo mejor de ambos mundos.
    """
    # Verificar si el modelo edge está saludable
    if (
        edge_engine.session is not None
        and edge_engine.avg_inference_ms < 100  # No tiene thermal throttling
    ):
        try:
            predictions, inference_ms = edge_engine.predict(input_data)
            return {
                "source": "edge",
                "predictions": predictions.tolist(),
                "latency_ms": inference_ms,
            }
        except Exception as e:
            logger.warning(f"Inferencia edge falló: {e}, cayendo a cloud")
 
    # Fallback a cloud
    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(
                cloud_url,
                json={"data": input_data.tolist()},
                timeout=aiohttp.ClientTimeout(
                    total=cloud_timeout_ms / 1000
                ),
            ) as resp:
                result = await resp.json()
                return {
                    "source": "cloud",
                    "predictions": result["predictions"],
                    "latency_ms": result.get("latency_ms", -1),
                }
    except asyncio.TimeoutError:
        logger.error("Edge y cloud fallaron — retornando resultado cacheado")
        return {
            "source": "cache",
            "predictions": get_cached_prediction(input_data),
            "latency_ms": 0,
        }