本文介绍基于PaddlePaddle框架复现ResNet18的项目，含详细注释助新手入门。代码定义了BasicBlock和ResNet类构建模型，提供Cifar10数据集下载及本地类ImageNet结构数据集的训练方法，设置20轮训练，验证准确率达0.7792，还包含模型保存步骤。

由于不同版本的paddlepaddle框架有部分特性不同，且随着版本更新会弃用一些方法，但虽然弃用，为了适配旧的版本的项目，仍会保存在框架源码中，如果你使用的版本较新，请运行下框的代码；

import warningswarnings.filterwarnings('ignore')登录后复制 ? ?

ResNet18 by paddlepaddle

本项目为使用paddlepaddle框架复现Resntet18，并添加大量注释作为入门文档，效果和各个框架封装的ResNet一致

运行配置为V100 32G，使用Cifar10验证时设置batch_size的大小为500时会占用约13.8G的显存，使用其他数据集时请调整batch_size大小防止爆显存

在Cifar数据集上，20轮大概可以达到AC率0.82，但实测调高轮次可以有更高的AC率

项目背景

很多成熟的，被认定有效的模型已经有完整的封装了，然而绝大部分模型都认为使用者是由足够知识基础的，而刚刚开始学习人工智能或学习使用人工智能的开发者往往没有足够的基础，甚至部分使用者是完全0基础的，本项目为基于paddlepaddle框架手写的Resnet18模型，并直接明确了各个函数的使用方法来帮助0基础的使用者入门，达到可以根据模型原论文/技术文档进行初步模型修改和调整的水平。

import paddleimport paddle.nn as nnimport paddle.nn.functional as Fimport paddle.optimizer as optimimport paddle.vision.transforms as Timport paddle.vision.datasets as Dfrom paddle.io import DataLoader登录后复制 ? ?In [3]

# ResNet-18 模型定义# 这里使用python的class来进行基模块的定义，这个基模块会在后面搭建模型时使用class BasicBlock(nn.Layer): expansion = 1 def __init__(self, in_channels, out_channels, stride=1, downsample=None): # in_channels为输入大小, out_channels输出大小, stride为卷积步长, downsample为下采样 super(BasicBlock, self).__init__()# 继承一下自己 self.conv1 = nn.Conv2D(in_channels, out_channels, kernel_size=3, stride=stride, padding=1, bias_attr=False) # Conv2D是paddle框架中封装好的对二位数据进行卷积的方法，和torch的封装结构一致，都封装在nn类中 # 这是第一个卷积层，卷积核大小为3*3,步长为1，padding为1，不使用bias（偏置） self.bn1 = nn.BatchNorm2D(out_channels) # 这是第一个BatchNorm层，对刚刚卷积后的数据进行批量归一化，out_channels为输出大小 self.relu = nn.ReLU() # 这是Relu激活函数，归一化后进入Relu激活函数，进行线性向非线性的变换 self.conv2 = nn.Conv2D(out_channels, out_channels, kernel_size=3, stride=1, padding=1, bias_attr=False) self.bn2 = nn.BatchNorm2D(out_channels) self.downsample = downsample self.stride = stride def forward(self, x):# 在这里定义前向传播 identity = x out = self.conv1(x)# 先定义一个值做个卷积 out = self.bn1(out)# 然后做个归一化 out = self.relu(out)# 再经过一次激活函数 out = self.conv2(out)# 接着卷积 out = self.bn2(out)# 继续归一化 if self.downsample is not None:# 在这里和进行对齐，避免无法进行残差链接 identity = self.downsample(x) out += identity# 残差链接，这里直接将上次的计算结果作为残差链接上来 out = self.relu(out) return out登录后复制 ? ?In [4]

class ResNet(nn.Layer): def __init__(self, block, layer_configs, num_classes=1000):# block是一个可更换的基模块，这里我们用的是上面声明的BasicBlock # layer_configs是一个列表，指定每个残差块组的块数量配置 super(ResNet, self).__init__() self.in_channels = 64 self.conv1 = nn.Conv2D(3, 64, kernel_size=7, stride=2, padding=3, bias_attr=False) self.bn1 = nn.BatchNorm2D(64) self.relu = nn.ReLU() self.maxpool = nn.MaxPool2D(kernel_size=3, stride=2, padding=1) # 构建残差块组 self.layer1 = self._make_layer(block, 64, layer_configs[0]) # 第一个残差块组 self.layer2 = self._make_layer(block, 128, layer_configs[1], stride=2) # 第二个残差块组，步长为2 self.layer3 = self._make_layer(block, 256, layer_configs[2], stride=2) # 第三个残差块组，步长为2 self.layer4 = self._make_layer(block, 512, layer_configs[3], stride=2) # 第四个残差块组，步长为2 self.avgpool = nn.AdaptiveAvgPool2D((1, 1)) self.fc = nn.Linear(512 * block.expansion, num_classes) def _make_layer(self, block, out_channels, num_blocks, stride=1): # 构建一个残差块组，包含多个残差块 # num_blocks: 残差块的数量 downsample = None # 判断是否需要下采样 if stride != 1 or self.in_channels != out_channels * block.expansion: downsample = nn.Sequential( nn.Conv2D(self.in_channels, out_channels * block.expansion, kernel_size=1, stride=stride, bias_attr=False), nn.BatchNorm2D(out_channels * block.expansion), ) layers = [] layers.append(block(self.in_channels, out_channels, stride, downsample)) self.in_channels = out_channels * block.expansion # 添加剩余的残差块，这些块不需要进行下采样 for _ in range(1, num_blocks): layers.append(block(self.in_channels, out_channels)) return nn.Sequential(*layers) def forward(self, x): x = self.conv1(x) x = self.bn1(x) x = self.relu(x) x = self.maxpool(x) x = self.layer1(x) x = self.layer2(x) x = self.layer3(x) x = self.layer4(x) x = self.avgpool(x) x = paddle.flatten(x, 1) x = self.fc(x) return x登录后复制 ? ?In [5]

def ResNet18(num_classes = 10): return ResNet(BasicBlock, [2, 2, 2, 2],num_classes=num_classes)# 在这里声明ResNet18，列表是每层的块数量，num_classes是分类的类别数# 可以试试更改模块数量，这样也可以说是改了模型登录后复制 ? ?

训练

后面两个代码框为两种不同的数据集读取方法，第一个是调用封装好的接口从服务器上下载Cifar10的代码；

第二个为使用本地的类似ImageNet数据存储结构的数据集进行训练的代码；

类似ImageNet的数据集存储结构：

数据集名称/├── train/│ ├── 类别1/│ │ ├── 类别1训练图像1（后缀可能是jpg，jepg，png，bmp等）│ │ ├── 类别1训练图像2│ │ ├── ...│ ├── 类别2/│ │ ├── 类别2训练图像1│ │ ├── 类别2训练图像2│ │ ├── ...│ ├── ...├── val/│ ├── 类别1/│ │ ├── 类别1验证图像1│ │ ├── 类别1验证图像2│ │ ├── ...│ ├── 类别2/│ │ ├── 类别2验证图像1│ │ ├── 类别2验证图像2│ │ ├── ...│ ├── ...

# 如果想用本地数据集进行训练，就注释掉这段代码，使用下一个方框内的代码# 由于Cifar10数据集是使用API接口下载的，犹如数据集官方可能更新API，该段代码可能报错# 如果该段代码报错，请注释掉该段代码，阅读下个方框使用本地数据集进行测试# 数据预处理和加载transform = T.Compose([ T.Resize((224, 224)),# 这个Resize其实是不合理的，因为Cifar10的图像是3*32*32，这里Resize为224*224，这样会导致信息有损 # 但是ResNet本来就是设计为224x224大小的，修改这里的Resize大小需要同时修改ResNet的结构，本文为使Paddle的原生Resnet18,故不做更改 T.ToTensor(), T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),])train_dataset = D.Cifar10(mode='train', transform=transform)val_dataset = D.Cifar10(mode='test', transform=transform)train_loader = DataLoader(train_dataset, batch_size=500, shuffle=True, drop_last=True)val_loader = DataLoader(val_dataset, batch_size=500, shuffle=False)# 模型、损失函数和优化器device = paddle.set_device('gpu' if paddle.is_compiled_with_cuda() else 'cpu')model = ResNet18(num_classes=10)model = model.to(device)criterion = nn.CrossEntropyLoss()optimizer = optim.Adam(parameters=model.parameters(), learning_rate=0.001)'''print('该代码框已屏蔽')# 这行代码用于标记这个框的代码是否有运行'''登录后复制 ? ? ? ?

W1118 14:08:18.540472 8281 gpu_resources.cc:119] Please NOTE: device: 0, GPU Compute Capability: 7.0, Driver API Version: 12.0, Runtime API Version: 11.8W1118 14:08:18.541822 8281 gpu_resources.cc:164] device: 0, cuDNN Version: 8.9.登录后复制 ? ? ? ?In [7]

'''# 如果要训练自己的数据集，就注释掉上面训练Cifar10的代码；# 注意，这段代码要求数据集的结构是 一个文件夹下有多个子文件夹，每个子文件夹的名称是标签，文件夹中的图片的训练图像# 这个文件的格式也是ImageNet的格式# 同时，这里还直接按读取到的子文件夹数量重新定义了ResNet18模型，可以不更改或不允许上面定义模型的代码# 在运行该段代码前，请解压data文件夹中的压缩包并修改路径到你的路径import osfrom PIL import Imageimport paddleimport paddle.nn as nnimport paddle.optimizer as optimimport paddle.vision.transforms as Tfrom paddle.io import DataLoader, Datasetfrom paddle.vision.models import resnet18# 自定义数据集类class CustomDataset(Dataset): def __init__(self, root_dir, transform=None): self.root_dir = root_dir self.transform = transform self.classes = sorted(os.listdir(root_dir)) self.class_to_idx = {cls_name: i for i, cls_name in enumerate(self.classes)} self.images = [] self.labels = [] for cls_name in self.classes: cls_dir = os.path.join(root_dir, cls_name) for img_name in os.listdir(cls_dir): img_path = os.path.join(cls_dir, img_name) self.images.append(img_path) self.labels.append(self.class_to_idx[cls_name]) def __len__(self): return len(self.images) def __getitem__(self, idx): img_path = self.images[idx] label = self.labels[idx] image = Image.open(img_path).convert('RGB') if self.transform: image = self.transform(image) return image, label# 数据预处理和加载transform = T.Compose([ T.Resize((224, 224)), T.ToTensor(), T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),])# 这里是训练集和验证集的路径，按你的路径进行修改，这里的 “./”为从当前文件所处文件夹路径开始向下寻找，具体参考python的路径操作train_dataset = CustomDataset(root_dir='./data/data303405/Training', transform=transform)val_dataset = CustomDataset(root_dir='./data/data303405/Testing', transform=transform)train_loader = DataLoader(train_dataset, batch_size=100, shuffle=True, drop_last=True)val_loader = DataLoader(val_dataset, batch_size=100, shuffle=False)# 模型、损失函数和优化器device = paddle.set_device('gpu' if paddle.is_compiled_with_cuda() else 'cpu')num_classes = len(train_dataset.classes) # 根据数据集的类别数调整模型输出类别数model = ResNet18(num_classes=num_classes)model = model.to(device)criterion = nn.CrossEntropyLoss()optimizer = optim.Adam(parameters=model.parameters(), learning_rate=0.001)#使用Adam优化器，初始学习率设置为0.001，也有很多其他优化器可以使用的，详细参考我都另一个项目或自行上网搜索# 训练和验证代码可以继续使用之前的代码'''print(”该代码框已屏蔽“)# 这行代码用于标记这个框的代码是否有运行登录后复制 ? ? ? ?

该代码框已屏蔽登录后复制 ? ? ? ?In [8]

# 训练循环num_epochs = 20 # 初始设置为20轮，可根据需求设置for epoch in range(num_epochs): model.train() for batch_id, (data, label) in enumerate(train_loader): data, label = data.to(device), label.to(device) optimizer.clear_grad() output = model(data) loss = criterion(output, label) loss.backward() optimizer.step() if batch_id % 100 == 0: print(f'Epoch [{epoch+1}/{num_epochs}], Step [{batch_id}/{len(train_loader)}], Loss: {loss.item():.4f}') # 验证 model.eval() correct = 0 total = 0 with paddle.no_grad(): for data, label in val_loader: data, label = data.to(device), label.to(device) output = model(data) _, predicted = paddle.topk(output, k=1) total += label.shape[0] correct += (predicted.squeeze() == label).astype('int32').sum().item() accuracy = correct / total print(f'Validation Accuracy: {accuracy:.4f}')# 保存模型登录后复制 ? ? ? ?

Epoch [1/20], Step [0/100], Loss: 2.5338Validation Accuracy: 0.5208Epoch [2/20], Step [0/100], Loss: 1.0902Validation Accuracy: 0.5667Epoch [3/20], Step [0/100], Loss: 0.8355Validation Accuracy: 0.7028Epoch [4/20], Step [0/100], Loss: 0.6159Validation Accuracy: 0.7269Epoch [5/20], Step [0/100], Loss: 0.5030Validation Accuracy: 0.7329Epoch [6/20], Step [0/100], Loss: 0.3750Validation Accuracy: 0.7699Epoch [7/20], Step [0/100], Loss: 0.2447Validation Accuracy: 0.7211Epoch [8/20], Step [0/100], Loss: 0.1741Validation Accuracy: 0.6971Epoch [9/20], Step [0/100], Loss: 0.1621Validation Accuracy: 0.7412Epoch [10/20], Step [0/100], Loss: 0.0974Validation Accuracy: 0.8108Epoch [11/20], Step [0/100], Loss: 0.0910Validation Accuracy: 0.8014Epoch [12/20], Step [0/100], Loss: 0.0521Validation Accuracy: 0.7784Epoch [13/20], Step [0/100], Loss: 0.0537Validation Accuracy: 0.7974Epoch [14/20], Step [0/100], Loss: 0.0147Validation Accuracy: 0.8082Epoch [15/20], Step [0/100], Loss: 0.0145Validation Accuracy: 0.7944Epoch [16/20], Step [0/100], Loss: 0.0341Validation Accuracy: 0.7840Epoch [17/20], Step [0/100], Loss: 0.0514Validation Accuracy: 0.7840Epoch [18/20], Step [0/100], Loss: 0.0350Validation Accuracy: 0.8077Epoch [19/20], Step [0/100], Loss: 0.0221Validation Accuracy: 0.7882Epoch [20/20], Step [0/100], Loss: 0.0220Validation Accuracy: 0.7792登录后复制 ? ? ? ?In [9]

paddle.save(model.state_dict(), 'resnet18_cifar10.pdparams')# 保存一下权重登录后复制 ? ?