Análise de Performance e Identificação de Gargalos

Antes de aplicar qualquer técnica de otimização, é fundamental identificar os gargalos computacionais do modelo. Uma análise de profiling revela quais camadas consomem mais recursos e onde os esforços de otimização terão maior impacto.

import torch
import time

class ModelProfiler:
    def __init__(self, weights_path, device='cpu'):
        from yolov5.models.experimental import attempt_load
        self.device = torch.device(device)
        self.net = attempt_load(weights_path, map_location=self.device)
        self.net.eval()
    
    def measure_inference(self, img_size=(1, 3, 640, 640), runs=100):
        dummy = torch.randn(img_size, device=self.device)
        
        # Fase de aquecimento
        with torch.no_grad():
            for _ in range(10):
                self.net(dummy)
        
        # Medição
        if self.device.type == 'cuda':
            torch.cuda.synchronize()
        
        t0 = time.perf_counter()
        with torch.no_grad():
            for _ in range(runs):
                self.net(dummy)
        
        if self.device.type == 'cuda':
            torch.cuda.synchronize()
        
        t1 = time.perf_counter()
        latency = (t1 - t0) / runs * 1000
        return latency

# profiler = ModelProfiler('yolov5s.pt')
# print(f"Latência média: {profiler.measure_inference():.2f} ms")

Técnicas de Pruning para Redução de Parâmetros

Pruning Estruturado

O pruning estruturado remove canais inteiros ou filtros, resultando em redução efetiva do custo computacional, diferentemente do pruning não-estruturado que apenas esparsifica a matriz de pesos.

import torch.nn as nn
import torch.nn.utils.prune as prune

def apply_channel_pruning(network, ratio=0.3):
    """Aplica pruning L1 não-estruturado em camadas convolucionais."""
    for module_name, layer in network.named_modules():
        if isinstance(layer, nn.Conv2d):
            prune.l1_unstructured(layer, name='weight', amount=ratio)
            prune.remove(layer, 'weight')
    return network

def apply_global_pruning(network, threshold_value=0.01):
    """Aplica pruning global baseado em magnitude."""
    target_layers = []
    for _, layer in network.named_modules():
        if isinstance(layer, nn.Conv2d):
            target_layers.append((layer, 'weight'))
    
    prune.global_unstructured(
        target_layers,
        pruning_method=prune.L1Unstructured,
        threshold=threshold_value
    )
    return network

# pruned_net = apply_channel_pruning(model, ratio=0.25)

Quantização de Modelos

Quantização Dinâmica

A quantização dinâmica converte pesos para INT8 durante a inferência, com ativações quantizadas dinamicamente. Esta abordagem oferece bom equilíbrio entre facilidade de implementação e ganho de performance.

def quantize_dynamic(net):
    return torch.quantization.quantize_dynamic(
        net,
        {nn.Linear, nn.Conv2d},
        dtype=torch.qint8
    )

# quantized_net = quantize_dynamic(model)

Quantização Estática Pós-Treinamento

A quantização estática requer um conjunto de dados de calibração para determinar os parâmetros de quantização das ativações. Oferece maior aceleração em comparação à quantização dinâmica.

def quantize_static(net, calibration_loader):
    net.eval()
    net.qconfig = torch.quantization.get_default_qconfig('fbgemm')
    
    prepared_net = torch.quantization.prepare(net)
    
    with torch.no_grad():
        for idx, batch in enumerate(calibration_loader):
            if idx >= 100:
                break
            prepared_net(batch)
    
    quantized_net = torch.quantization.convert(prepared_net)
    return quantized_net

Exportação e Otimização com ONNX

O formato ONNX (Open Neural Network Exchange) facilita a interoperabilidade entre frameworks e serve como intermediário para deploy em diversos runtimes de inferência otimizados.

import onnx
from onnxsim import simplify

def export_onnx(net, filepath="model.onnx", imgsz=640):
    net.eval()
    dummy = torch.randn(1, 3, imgsz, imgsz)
    
    torch.onnx.export(
        net,
        dummy,
        filepath,
        export_params=True,
        opset_version=12,
        do_constant_folding=True,
        input_names=['input'],
        output_names=['output'],
        dynamic_axes={
            'input': {0: 'batch'},
            'output': {0: 'batch'}
        }
    )
    print(f"Modelo exportado para {filepath}")

def simplify_onnx(input_path, output_path):
    onnx_model = onnx.load(input_path)
    simplified, success = simplify(onnx_model)
    
    if success:
        onnx.save(simplified, output_path)
        print(f"Modelo simplificado salvo em {output_path}")
    else:
        print("Falha na simplificação")

Inferência com ONNX Runtime

import onnxruntime as ort
import numpy as np

class OnnxEngine:
    def __init__(self, model_path):
        providers = ['CUDAExecutionProvider', 'CPUExecutionProvider']
        self.session = ort.InferenceSession(model_path, providers=providers)
        
        self.input_key = self.session.get_inputs()[0].name
        self.output_key = self.session.get_outputs()[0].name
    
    def predict(self, img_tensor):
        return self.session.run(
            [self.output_key],
            {self.input_key: img_tensor}
        )[0]
    
    def benchmark(self, iterations=500):
        dummy = np.random.randn(1, 3, 640, 640).astype(np.float32)
        
        for _ in range(10):
            self.predict(dummy)
        
        t0 = time.perf_counter()
        for _ in range(iterations):
            self.predict(dummy)
        t1 = time.perf_counter()
        
        latency = (t1 - t0) / iterations * 1000
        print(f"ONNX Runtime - Latência: {latency:.2f} ms")
        return latency

Aceleração com TensorRT

TensorRT é o mecanismo de inferência de alta performance da NVIDIA, otimizado para GPUs. Realiza otimizações em nível de grafo, fusão de kernels e seleção ótima de algoritmos.

Construção do Engine TensorRT

import tensorrt as trt
import pycuda.driver as cuda
import pycuda.autoinit

class TrtEngineBuilder:
    def __init__(self):
        self.logger = trt.Logger(trt.Logger.WARNING)
        self.builder = trt.Builder(self.logger)
        self.config = self.builder.create_builder_config()
    
    def build_from_onnx(self, onnx_path, engine_path, use_fp16=True):
        explicit_batch = 1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
        network = self.builder.create_network(explicit_batch)
        parser = trt.OnnxParser(network, self.logger)
        
        with open(onnx_path, 'rb') as f:
            if not parser.parse(f.read()):
                for i in range(parser.num_errors):
                    print(f"Erro: {parser.get_error(i)}")
                raise RuntimeError("Falha ao parsear ONNX")
        
        self.config.max_workspace_size = 1 << 30  # 1GB
        
        if use_fp16 and self.builder.platform_has_fast_fp16:
            self.config.set_flag(trt.BuilderFlag.FP16)
        
        engine = self.builder.build_engine(network, self.config)
        
        with open(engine_path, 'wb') as f:
            f.write(engine.serialize())
        
        return engine

# builder = TrtEngineBuilder()
# builder.build_from_onnx('model.onnx', 'model_fp16.trt')

Inferência com TensorRT

class TrtInferenceEngine:
    def __init__(self, engine_path):
        self.logger = trt.Logger(trt.Logger.WARNING)
        with open(engine_path, 'rb') as f:
            self.engine = trt.Runtime(self.logger).deserialize_cuda_engine(f.read())
        
        self.context = self.engine.create_execution_context()
        self.stream = cuda.Stream()
        
        self.device_buffers = []
        self.input_info = None
        self.output_info = None
        
        for i in range(self.engine.num_bindings):
            shape = self.engine.get_binding_shape(i)
            dtype = trt.nptype(self.engine.get_binding_dtype(i))
            size = trt.volume(shape)
            buffer = cuda.mem_alloc(size * np.dtype(dtype).itemsize)
            self.device_buffers.append(buffer)
            
            info = {'ptr': buffer, 'shape': shape, 'dtype': dtype, 'size': size}
            if self.engine.binding_is_input(i):
                self.input_info = info
            else:
                self.output_info = info
    
    def __call__(self, input_array):
        cuda.memcpy_htod_async(self.input_info['ptr'], input_array.ravel(), self.stream)
        
        self.context.execute_async_v2(
            bindings=[int(b) for b in self.device_buffers],
            stream_handle=self.stream.handle
        )
        
        output = np.empty(self.output_info['size'], dtype=self.output_info['dtype'])
        cuda.memcpy_dtoh_async(output, self.output_info['ptr'], self.stream)
        self.stream.synchronize()
        
        return output
    
    def profile(self, iterations=1000):
        dummy = np.random.randn(1, 3, 640, 640).astype(np.float32)
        
        for _ in range(15):
            self(dummy)
        
        t0 = time.perf_counter()
        for _ in range(iterations):
            self(dummy)
        t1 = time.perf_counter()
        
        latency = (t1 - t0) / iterations * 1000
        print(f"TensorRT - Latência: {latency:.2f} ms")
        return latency

Pipeline de Otimização Completa

class YOLOOptimizer:
    def __init__(self, weights_path):
        self.weights_path = weights_path
        self.results = {}
    
    def execute_pipeline(self):
        from yolov5.models.experimental import attempt_load
        
        # Carregar modelo base
        base_model = attempt_load(self.weights_path, map_location='cpu')
        self.results['original_latency'] = self._measure(base_model)
        
        # Aplicar pruning
        pruned_model = apply_channel_pruning(base_model, ratio=0.25)
        self.results['pruned_latency'] = self._measure(pruned_model)
        
        # Exportar para ONNX
        export_onnx(pruned_model, 'yolo_optimized.onnx')
        simplify_onnx('yolo_optimized.onnx', 'yolo_final.onnx')
        
        # Benchmark ONNX Runtime
        onnx_engine = OnnxEngine('yolo_final.onnx')
        self.results['onnx_latency'] = onnx_engine.benchmark()
        
        # Benchmark TensorRT
        trt_builder = TrtEngineBuilder()
        trt_builder.build_from_onnx('yolo_final.onnx', 'yolo_fp16.trt')
        
        trt_engine = TrtInferenceEngine('yolo_fp16.trt')
        self.results['trt_latency'] = trt_engine.profile()
        
        self._report()
    
    def _measure(self, net, runs=50):
        profiler = ModelProfiler.__new__(ModelProfiler)
        profiler.net = net
        profiler.device = torch.device('cpu')
        return profiler.measure_inference(runs=runs)
    
    def _report(self):
        base = self.results['original_latency']
        print("\n=== Relatório de Performance ===")
        print(f"{'Estágio':<20} {'Latência (ms)':<15} {'Speedup':<10}")
        print("-" * 45)
        
        stages = [
            ('Original', 'original_latency'),
            ('Pruning', 'pruned_latency'),
            ('ONNX Runtime', 'onnx_latency'),
            ('TensorRT FP16', 'trt_latency')
        ]
        
        for name, key in stages:
            lat = self.results[key]
            speedup = base / lat if lat > 0 else 0
            print(f"{name:<20} {lat:<15.2f} {speedup:<10.2f}x")

Comparativo de Performance

Boas Práticas e Monitoramento

Validação de Precisão

def validate_accuracy(original, optimized, val_dataset):
    def evaluate(net, data):
        net.eval()
        correct = 0
        total = 0
        with torch.no_grad():
            for images, targets in data:
                outputs = net(images)
                # Lógica de avaliação específica
                total += targets.size(0)
        return correct / total if total > 0 else 0
    
    base_acc = evaluate(original, val_dataset)
    opt_acc = evaluate(optimized, val_dataset)
    
    degradation = (base_acc - opt_acc) * 100
    
    if degradation > 2.0:
        print(f"ALERTA: Degradação de precisão de {degradation:.2f}%")
    
    return degradation

Otimização para Dispositivos Edge

def optimize_for_target(net, target_device):
    if target_device == 'jetson':
        net = net.half()
        torch.backends.cudnn.benchmark = True
    elif target_device == 'edge_tpu':
        net = quantize_for_tpu(net)
    elif target_device == 'mobile':
        net = torch.quantization.quantize_dynamic(net)
    return net

A aplicação sistemática das técnicas apresentadas permite acelerações de 5 a 10 vezes na inferência de modelos YOLO, mantendo degradação de precisão aceitável para a maioria das aplicações industriais. A escolha do pipeline de otimização deve considerar os requisitos específicos de latência, precisão e recursos computacionais disponíveis no ambiente de deploy.