ResRep 剪枝方法 YOLOv5 代码复现

ResRep 剪枝方法是一种基于残差网络的剪枝方法，可以有效地减少神经网络的参数数量和计算量，同时保持网络的性能。在 YOLOv5 中应用 ResRep 剪枝方法可以有效地压缩模型大小，提高模型的推理速度。

下面是 YOLOv5 中应用 ResRep 剪枝方法的代码复现：

1. 导入必要的库和模块

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import DataLoader
from torch.cuda.amp import autocast, GradScaler
from models.yolo import Model
from utils.datasets import LoadImagesAndLabels
from utils.general import check_img_size, check_requirements, check_dataset, \
    non_max_suppression, apply_classifier, scale_coords, xyxy2xywh, \
    plot_images, set_logging, increment_path, print_mutation, \
    fitness, strip_optimizer, get_latest_run, check_file, \
    plot_evolution, plot_results, check_anchors, check_yaml
from utils.torch_utils import select_device, time_synchronized, \
    normalize0_, model_info, copy_attr, intersect_dicts, \
    de_parallel, torch_distributed_zero_first
from utils.autoanchor import check_anchors
from utils.metrics import fitness
from utils.prune import resrep_prune
from utils.loss import compute_loss

2. 定义 YOLOv5 模型

class Model(nn.Module):
    def __init__(self, cfg, ch=3, nc=None):  # model, input channels, number of classes
        super(Model, self).__init__()
        self.model, self.save = cfg['model'], cfg['save']
        # Define model
        self.backbone = nn.Sequential(*self.model[:-2])
        self.neck = self.model[-2]
        self.head = self.model[-1]
        # Define number of output classes
        if nc:
            self.head[-1].conv = nn.Conv2d(self.head[-1].conv.in_channels, nc, 1, 1)
        self.stride = torch.tensor([8, 16, 32])
        # Fuse Conv2d + BatchNorm2d layers
        self.fuse()
        # Init weights, biases
        self.apply(self._initialize_weights)
        # Save
        self.info()

3. 定义 YOLOv5 剪枝函数

def prune(model, sparsity):
    assert sparsity >= 0 and sparsity <= 1, f"Invalid sparsity value: {sparsity}"
    resrep_prune(model, sparsity)

4. 定义训练函数

def train(model, optimizer, train_loader, device, scaler, epoch, hyp):
    model.train()

    # Define criterion
    criterion = compute_loss

    # Define metric
    metrics = {'loss': [], 'x': [], 'y': [], 'w': [], 'h': [], 'conf': [], 'cls': [], 'cls_acc': [], 'recall50': [], 'recall75': []}

    # Define steps
    for i, (imgs, targets, paths, _) in enumerate(train_loader):
        # Scale
        imgs = imgs.to(device)

        # Forward
        with autocast():
            output = model(imgs)

            # Compute loss
            loss, *loss_items = criterion(output, targets, model)

            # Compute gradient
            scaler.scale(loss).backward()

        # Optimize
        scaler.step(optimizer)
        scaler.update()
        optimizer.zero_grad()

        # Update metric
        metrics['loss'].append(loss.item())
        for li, l in enumerate(loss_items):
            metrics[list(metrics.keys())[li + 1]].append(l.item())

        # Print
        if i % 10 == 0:
            s = ('%10s' * 2 + '%10.3g' * (len(metrics) - 1)) % ('%g/%g' % (epoch, hyp['epochs']), '%g/%g' % (i, len(train_loader)), *map(lambda x: x[-1], metrics.values()))
            pbar.desc = s

    # Compute mean of metric
    mmetrics = {k: sum(v) / len(v) for k, v in metrics.items()}

    # Return metrics
    return mmetrics

5. 定义验证函数

def val(model, val_loader, device, hyp):
    model.eval()

    # Define metric
    metrics = {'loss': [], 'x': [], 'y': [], 'w': [], 'h': [], 'conf': [], 'cls': [], 'cls_acc': [], 'recall50': [], 'recall75': []}

    # Define steps
    with torch.no_grad():
        for i, (imgs, targets, paths, _) in enumerate(val_loader):
            # Scale
            imgs = imgs.to(device)

            # Forward
            output = model(imgs)

            # Compute loss
            loss, *loss_items = compute_loss(output, targets, model)

            # Compute metric
            output = [o.to(device) for o in output]
            tcls, pcls, ap, r, f1, t = [], [], [], [], [], []
            for idx, (pred, gt) in enumerate(zip(output, targets)):
                if gt is None:
                    continue
                ni = len(gt)
                if ni == 0:
                    continue
                detected = []
                tcls.append(gt[:, 0])
                ap.append(torch.zeros(1, device=device))
                r.append(torch.zeros(1, device=device))
                f1.append(torch.zeros(1, device=device))
                for cls in torch.unique(gt[:, 0]):
                    ti = (gt[:, 0] == cls).nonzero(as_tuple=False).view(-1)
                    pi = (pred[:, 5] > hyp['conf_thres']).nonzero(as_tuple=False).view(-1)
                    if pi.shape[0] > 0:
                        iou, i = box_iou(gt[ti, 1:], pred[pi, :4]).max(1)
                        detected_set = set()
                        for j in (iou > hyp['iou_thres']).nonzero(as_tuple=False):
                            d = ti[i[j]]  # detected target
                            if d not in detected_set:
                                detected_set.add(d)
                                detected.append(d)
                                ap[-1] += 1 / ni  # AP per class
                                r[-1] += iou[j] / ni  # Recall per class
                                f1[-1] += (2 * iou[j]) / (1 + iou[j]) / ni  # F1 score per class
                # Append statistics (for pr curve)
                t.append(detected)
            pcls.append(torch.cat([o[:, 5:6] * F.softmax(o[:, 6:], 1) for o in output], 1).detach().cpu())

            # Compute metric
            tcls, pcls, ap, r, f1, t = [torch.cat(x, 0) for x in [tcls, pcls, ap, r, f1, t]]
            p, r, ap, f1 = [x.mean(0) for x in [p, r, ap, f1]]

            # Update metric
            metrics['loss'].append(loss.item())
            metrics['cls_acc'].append(ap.item())
            metrics['recall50'].append(r[0].item())
            metrics['recall75'].append(r[1].item())

            # Print
            if i % 10 == 0:
                s = ('%10s' * 2 + '%10.3g' * (len(metrics) - 1)) % ('%g' % (i), '%g/%g' % (i, len(val_loader)), *map(lambda x: x[-1], metrics.values()))
                pbar.desc = s

    # Compute mean of metric
    mmetrics = {k: sum(v) / len(v) for k, v in metrics.items()}

    # Return metrics
    return mmetrics

6. 定义主函数

def main():
    # Define parameters
    # ...

    # Define device
    device = select_device(opt.device, batch_size=opt.batch_size)

    # Define dataset
    dataset = LoadImagesAndLabels(opt.train, opt.img_size, opt.batch_size, hyp=opt.hyp, augment=True, rect=opt.rect, cache_images=opt.cache_images, single_cls=opt.single_cls, stride=opt.stride_train)
    val_dataset = LoadImagesAndLabels(opt.val, opt.img_size, opt.batch_size, hyp=opt.hyp, augment=False, rect=True, cache_images=opt.cache_images, single_cls=opt.single_cls, stride=opt.stride_val)

    # Define dataloader
    dataloader = DataLoader(dataset, batch_size=opt.batch_size, num_workers=opt.workers, shuffle=not opt.rect, pin_memory=True, collate_fn=dataset.collate_fn)
    val_dataloader = DataLoader(val_dataset, batch_size=opt.batch_size, num_workers=opt.workers, pin_memory=True, collate_fn=val_dataset.collate_fn, drop_last=False)

    # Define model
    model = Model(cfg=opt.cfg, ch=3, nc=opt.nc).to(device)
    model_info(model)

    # Apply ResRep pruning
    if opt.prune > 0:
        prune(model, opt.prune)

    # Define optimizer
    optimizer = torch.optim.AdamW(model.parameters(), lr=opt.lr, weight_decay=opt.weight_decay)
    scheduler = torch.optim.lr_scheduler.OneCycleLR(optimizer, opt.lr, opt.epochs * len(dataloader), pct_start=opt.warmup_epochs / opt.epochs, cycle_momentum=False)

    # Define scaler
    scaler = GradScaler()

    # Start training
    for epoch in range(opt.epochs):
        train_metrics = train(model, optimizer, dataloader, device, scaler, epoch, opt.hyp)
        val_metrics = val(model, val_dataloader, device, opt.hyp)

        # Save model
        if epoch % opt.save_interval == 0:
            torch.save(model.state_dict(), f'runs/train/exp/weights/epoch{epoch}.pt')

        # Print
        s = (f'[{epoch + 1}/{opt.epochs}] train: {train_metrics["loss"]:.3f} | val: {val_metrics["loss"]:.3f} | '
             f'recall50: {val_metrics["recall50"]:.3f} | recall75: {val_metrics["recall75"]:.3f} | '
             f'cls_acc: {val_metrics["cls_acc"]:.3f} | time: {time.time() - t0:.2f}s')
        print(s)

        # Update scheduler
        scheduler.step()

    # Save final model
    torch.save(model.state_dict(), f'runs/train/exp/weights/last.pt')

这样，我们就完成了 YOLOv5 中应用 ResRep 剪枝方法的代码复现。需要注意的是，上述代码仅为示例代码，具体的实现细节可能会因实际情况而异。