3. 基于Taylor展开的滤波器重要性评估
任务目标
本案例实现基于Taylor展开的滤波器重要性评估方法,对VGG-16模型进行通道剪枝,并在CIFAR-100数据集上验证效果,同时与L1范数重要性评估方法进行对比。Taylor展开方法利用损失函数对滤波器输出的梯度信息,评估每个滤波器对最终损失的贡献程度,相比仅考虑权重幅值的L1范数方法,能更准确地识别对模型预测至关重要的滤波器。
核心原理与实现
Taylor展开重要性评估的核心思想是:将滤波器移除后引起的损失变化近似为损失函数对该滤波器输出的一阶Taylor展开项。对于第l层第i个滤波器,其重要性分数计算为:
Importance(l,i) = |∂L/∂zl,i · zl,i|
其中zl,i为该滤波器的输出特征图,∂L/∂zl,i为损失对该输出的梯度。实现时通过一次前向传播和反向传播即可获得所有滤波器的重要性分数,计算效率高。L1范数方法则直接计算滤波器权重值之和作为重要性分数。
完整实现代码
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torchvision import datasets, transforms, models
import numpy as np
import copy
# -------------------- 数据加载 --------------------
def get_dataloaders(batch_size=64):
transform_train = transforms.Compose([
transforms.RandomCrop(32, padding=4),
transforms.RandomHorizontalFlip(),
transforms.ToTensor(),
transforms.Normalize((0.5071, 0.4867, 0.4408), (0.2675, 0.2565, 0.2761))
])
transform_test = transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.5071, 0.4867, 0.4408), (0.2675, 0.2565, 0.2761))
])
train_set = datasets.CIFAR100(root='./data', train=True, download=True, transform=transform_train)
test_set = datasets.CIFAR100(root='./data', train=False, download=True, transform=transform_test)
train_loader = torch.utils.data.DataLoader(train_set, batch_size=batch_size, shuffle=True, num_workers=2)
test_loader = torch.utils.data.DataLoader(test_set, batch_size=batch_size, shuffle=False, num_workers=2)
return train_loader, test_loader
# -------------------- 模型定义(VGG-16简化版) --------------------
class VGG16(nn.Module):
def __init__(self, num_classes=100):
super().__init__()
self.features = nn.Sequential(
# Block 1
nn.Conv2d(3, 64, 3, padding=1), nn.ReLU(inplace=True),
nn.Conv2d(64, 64, 3, padding=1), nn.ReLU(inplace=True),
nn.MaxPool2d(2),
# Block 2
nn.Conv2d(64, 128, 3, padding=1), nn.ReLU(inplace=True),
nn.Conv2d(128, 128, 3, padding=1), nn.ReLU(inplace=True),
nn.MaxPool2d(2),
# Block 3
nn.Conv2d(128, 256, 3, padding=1), nn.ReLU(inplace=True),
nn.Conv2d(256, 256, 3, padding=1), nn.ReLU(inplace=True),
nn.Conv2d(256, 256, 3, padding=1), nn.ReLU(inplace=True),
nn.MaxPool2d(2),
# Block 4
nn.Conv2d(256, 512, 3, padding=1), nn.ReLU(inplace=True),
nn.Conv2d(512, 512, 3, padding=1), nn.ReLU(inplace=True),
nn.Conv2d(512, 512, 3, padding=1), nn.ReLU(inplace=True),
nn.MaxPool2d(2),
# Block 5
nn.Conv2d(512, 512, 3, padding=1), nn.ReLU(inplace=True),
nn.Conv2d(512, 512, 3, padding=1), nn.ReLU(inplace=True),
nn.Conv2d(512, 512, 3, padding=1), nn.ReLU(inplace=True),
nn.MaxPool2d(2),
)
self.classifier = nn.Sequential(
nn.Linear(512, 4096), nn.ReLU(inplace=True), nn.Dropout(),
nn.Linear(4096, 4096), nn.ReLU(inplace=True), nn.Dropout(),
nn.Linear(4096, num_classes)
)
def forward(self, x):
x = self.features(x)
x = x.view(x.size(0), -1)
x = self.classifier(x)
return x
# -------------------- 重要性评估 --------------------
def compute_taylor_importance(model, dataloader, device):
"""基于Taylor展开计算每个卷积层滤波器的重要性分数"""
model.eval()
importance = {}
gradients = {}
activations = {}
# 注册钩子收集梯度和激活值
hooks = []
for name, module in model.features.named_modules():
if isinstance(module, nn.Conv2d):
def hook_fn(module, input, output, name=name):
activations[name] = output.detach()
def grad_hook_fn(module, grad_input, grad_output, name=name):
gradients[name] = grad_output[0].detach()
hooks.append(module.register_forward_hook(hook_fn))
hooks.append(module.register_full_backward_hook(grad_hook_fn))
total_loss = 0.0
num_batches = 0
criterion = nn.CrossEntropyLoss()
for data, target in dataloader:
data, target = data.to(device), target.to(device)
model.zero_grad()
output = model(data)
loss = criterion(output, target)
loss.backward()
total_loss += loss.item()
num_batches += 1
# 计算重要性分数
for name, module in model.features.named_modules():
if isinstance(module, nn.Conv2d):
if name in activations and name in gradients:
# Taylor展开: |grad * activation|,对空间维度求和
score = torch.abs(gradients[name] * activations[name])
score = score.sum(dim=(0, 2, 3)) # 每个滤波器一个分数
importance[name] = score.cpu().numpy()
for hook in hooks:
hook.remove()
return importance
def compute_l1_importance(model):
"""基于L1范数计算滤波器重要性"""
importance = {}
for name, module in model.features.named_modules():
if isinstance(module, nn.Conv2d):
# 对权重求L1范数,每个滤波器一个分数
score = torch.abs(module.weight).sum(dim=(1, 2, 3))
importance[name] = score.detach().cpu().numpy()
return importance
# -------------------- 剪枝实现 --------------------
def prune_model(model, importance_dict, prune_ratio):
"""根据重要性分数剪枝,返回剪枝后的模型"""
pruned_model = copy.deepcopy(model)
conv_layers = []
bn_layers = []
# 收集所有卷积层和BN层
for name, module in pruned_model.features.named_modules():
if isinstance(module, nn.Conv2d):
conv_layers.append((name, module))
elif isinstance(module, nn.BatchNorm2d):
bn_layers.append((name, module))
# 确定每层保留的通道数
keep_indices = {}
for name, module in conv_layers:
scores = importance_dict[name]
num_filters = len(scores)
num_keep = int(num_filters * (1 - prune_ratio))
# 保留重要性分数最高的滤波器
keep_idx = np.argsort(scores)[-num_keep:]
keep_idx.sort()
keep_indices[name] = keep_idx
# 逐层剪枝
prev_keep = None
for i, (name, conv) in enumerate(conv_layers):
keep_idx = keep_indices[name]
# 剪枝当前卷积层
new_conv = nn.Conv2d(
in_channels=conv.in_channels if prev_keep is None else len(prev_keep),
out_channels=len(keep_idx),
kernel_size=conv.kernel_size,
stride=conv.stride,
padding=conv.padding
)
# 复制权重
with torch.no_grad():
if prev_keep is not None:
new_conv.weight.copy_(conv.weight[keep_idx][:, prev_keep])
else:
new_conv.weight.copy_(conv.weight[keep_idx])
if conv.bias is not None:
new_conv.bias.copy_(conv.bias[keep_idx])
# 替换层
parent = pruned_model.features
layer_name = name.split('.')[-1]
if len(name.split('.')) > 1:
parent = pruned_model.features[int(name.split('.')[0])]
layer_name = name.split('.')[1]
parent[int(layer_name)] = new_conv
# 剪枝对应的BN层
if i < len(bn_layers):
bn_name, bn = bn_layers[i]
new_bn = nn.BatchNorm2d(len(keep_idx))
with torch.no_grad():
new_bn.weight.copy_(bn.weight[keep_idx])
new_bn.bias.copy_(bn.bias[keep_idx])
new_bn.running_mean.copy_(bn.running_mean[keep_idx])
new_bn.running_var.copy_(bn.running_var[keep_idx])
bn_parent = pruned_model.features
bn_layer_name = bn_name.split('.')[-1]
if len(bn_name.split('.')) > 1:
bn_parent = pruned_model.features[int(bn_name.split('.')[0])]
bn_layer_name = bn_name.split('.')[1]
bn_parent[int(bn_layer_name)] = new_bn
prev_keep = keep_idx
# 调整分类器输入维度
last_conv_out = len(prev_keep)
old_fc = pruned_model.classifier[0]
new_fc = nn.Linear(last_conv_out, 4096)
with torch.no_grad():
# 由于最后一层卷积输出是512,这里需要根据实际结构调整
# 简化处理:直接复制权重的前last_conv_out*7*7部分
new_fc.weight.copy_(old_fc.weight[:, :last_conv_out*7*7])
new_fc.bias.copy_(old_fc.bias)
pruned_model.classifier[0] = new_fc
return pruned_model
# -------------------- 训练与评估 --------------------
def evaluate(model, test_loader, device):
model.eval()
correct = 0
total = 0
with torch.no_grad():
for data, target in test_loader:
data, target = data.to(device), target.to(device)
output = model(data)
_, predicted = output.max(1)
total += target.size(0)
correct += predicted.eq(target).sum().item()
return 100.0 * correct / total
def train(model, train_loader, device, epochs=5):
model.train()
criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.9, weight_decay=5e-4)
scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=2, gamma=0.1)
for epoch in range(epochs):
running_loss = 0.0
for data, target in train_loader:
data, target = data.to(device), target.to(device)
optimizer.zero_grad()
output = model(data)
loss = criterion(output, target)
loss.backward()
optimizer.step()
running_loss += loss.item()
scheduler.step()
print(f'Epoch {epoch+1}/{epochs}, Loss: {running_loss/len(train_loader):.4f}')
# -------------------- 主流程 --------------------
def main():
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f'Using device: {device}')
# 加载数据
train_loader, test_loader = get_dataloaders(batch_size=64)
# 创建原始模型
original_model = VGG16().to(device)
print('Original model accuracy:', evaluate(original_model, test_loader, device))
# 训练原始模型(实际使用时需充分训练,这里仅演示流程)
print('Training original model...')
train(original_model, train_loader, device, epochs=1)
# 计算两种重要性分数
print('Computing Taylor importance...')
taylor_importance = compute_taylor_importance(original_model, train_loader, device)
print('Computing L1 importance...')
l1_importance = compute_l1_importance(original_model)
# 执行剪枝
prune_ratio = 0.3 # 剪枝30%的滤波器
print(f'Pruning with ratio {prune_ratio}...')
taylor_pruned = prune_model(original_model, taylor_importance, prune_ratio)
l1_pruned = prune_model(original_model, l1_importance, prune_ratio)
# 评估剪枝后模型(未微调)
print('Taylor pruned model accuracy (before fine-tune):', evaluate(taylor_pruned, test_loader, device))
print('L1 pruned model accuracy (before fine-tune):', evaluate(l1_pruned, test_loader, device))
# 微调
print('Fine-tuning Taylor pruned model...')
train(taylor_pruned, train_loader, device, epochs=3)
print('Fine-tuning L1 pruned model...')
train(l1_pruned, train_loader, device, epochs=3)
# 最终评估
taylor_acc = evaluate(taylor_pruned, test_loader, device)
l1_acc = evaluate(l1_pruned, test_loader, device)
print(f'Final accuracy - Taylor pruned: {taylor_acc:.2f}%')
print(f'Final accuracy - L1 pruned: {l1_acc:.2f}%')
if __name__ == '__main__':
main()
运行方式
- 安装依赖:
pip install torch torchvision numpy - 运行脚本:
python taylor_pruning.py - 首次运行会自动下载CIFAR-100数据集(约160MB)
验证方法与判定标准
运行完成后观察输出:
- 原始模型在CIFAR-100上应有一定准确率(随机初始化约1%,训练后应显著提升)
- 剪枝后模型准确率会下降,但Taylor方法剪枝后的准确率应高于或等于L1方法
- 微调后两种方法的准确率都应回升,Taylor方法通常恢复得更好
判定通过条件:Taylor剪枝模型的最终准确率不低于L1剪枝模型,且两者均高于随机剪枝基线(约1%)。若Taylor方法准确率明显低于L1方法,说明实现存在错误。
关键实现说明:Taylor重要性评估需要同时获取激活值和梯度,代码通过PyTorch的hook机制实现。注意
register_full_backward_hook在PyTorch 1.8+中可用,旧版本使用register_backward_hook。剪枝时需同步调整卷积层、BN层和全连接层的维度,确保模型结构一致。