Pytorch基于卷积神经网络的猫狗识别

实验环境

Pytorch 1.4.0
conda 4.7.12
Jupyter Notebook 6.0.1
Python 3.7

数据集介绍

实验采用的猫和狗的图片来自 Kaggle 竞赛的一个赛题 Cat vs Dog 的数据集，其中训练数据集包括 25000 张图片，其中类别为猫的图片有 12500 张图片，类别为狗的图片有 12500 张，两种类别比例为 1:1。训练集有 25000 张，猫狗各占一半。测试集 12500 张，猫狗各占一半。实际上该数据集是 Asirra 数据集的子集。

Asirra 数据集的来源:
Web 服务有时通过行为验证信息来保护自身不被网络攻击，因为类似识别一个物品这样的问题对人们来说很容易解决，但对计算机却很难。这种挑战通常称为 CAPTCHA 完全自动化的公共 Turing 测试，以区分计算机和人类)或 HIP(人类互动证明)。HIP 有多种用途，例如减少电子邮件和博客垃圾邮件，以及防止对网站密码的暴力攻击。Asirra(用于限制访问的动物物种图像识别)是一种 HIP，其工作原理是要求用户识别猫和狗的照片。对于计算机而言，此任务很困难，但研究表明，人们可以快速而准确地完成此任务。Asirra 之所以与众不同，是因为它与 Petfinder.com 合作，Petfinder.com 是全球最大的致力于寻找无家可归宠物的家的网站。他们为 Microsoft Research 提供了超过三百万张猫和狗的图像，这些图像由美国数千家动物收容所中的人手动分类。Kaggle 很幸运能够提供这些数据的子集，以供娱乐和研究之用。

训练过程

数据准备

数据预处理:首先，导入实验所需的库，定义一些宏参数，BATCH_SIZE 表示每个 batch 加载多少个样本、EPOCHS 表示总共训练批次。如果支持 cuda 就用 gpu 来 run，不支持就用 cpu 来 run。

import numpy as np # linear algebra
import pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)
import os
import torch
import torch.nn as nn
import cv2
import matplotlib.pyplot as plt
import torchvision
from torch.utils.data import Dataset, DataLoader, ConcatDataset
from torchvision import transforms,models
from torch.optim.lr_scheduler import *
import copy
import random
import tqdm
from PIL import Image
import torch.nn.functional as F%matplotlib inlineBATCH_SIZE = 20
EPOCHS = 10
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")

从 Kaggle 官网下载好数据集 train.zip 和 test1.zip，解压到项目目录 data 文件夹下，重命名训练集和测试集文件夹名字。由于 listdir 参数不允许有”…” 和”.”，所以我先获取项目路径，再拼接上项目目录下训练集和测试集的位置，构成训练集和测试集的路径地址，最后通过 listdir 获取相应目录下文件名的集合。

cPath = os.getcwd()
train_dir = cPath + '/data/train'
test_dir = cPath + '/data/test'
train_files = os.listdir(train_dir)
test_files = os.listdir(test_dir)

训练集的图片命名规则是:类型. 序号.jpg，我定义一个数据集处理类 CatDogDataset 来对数据集进行预处理，狗的 label 为 1，猫的 label 为 0。以及在 getitem 时调用 transform 处理输入数据，根据 mode 返回不同的信息，mode=train 则返回训练图片和标签，其他则返回图片和图片文件名。

class CatDogDataset(Dataset):def __init__(self, file_list, dir, mode='train', transform = None):self.file_list = file_listself.dir = dirself.mode= modeself.transform = transformif self.mode == 'train':if 'dog' in self.file_list[0]:self.label = 1else:self.label = 0def __len__(self):return len(self.file_list)def __getitem__(self, idx):img = Image.open(os.path.join(self.dir, self.file_list[idx]))if self.transform:img = self.transform(img)if self.mode == 'train':img = img.numpy()return img.astype('float32'), self.labelelse:img = img.numpy()return img.astype('float32'), self.file_list[idx]

使用自定义的 transform 进行数据增强，它是对训练集进行变换，使训练集更丰富，从而让模型更具泛化能力，以及数据处理统一输入图片格式大小和归一化。train_transforms 先调整图片大小至 256x256 重置图像分辨率，再按照 224x224 随机剪裁，然后随机的图像水平翻转，转化成 tensor，最后采用 ImageNet 给出的数值归一化。接着构造 train dataloader，目的是为了方便读取和使用，设置 batch 大小，采用多线程，shuffle=True 设置在每个 epoch 重新打乱数据，保证数据的随机性。
test_transform 重置图片分辨率 224x224，转化成 tensor，同样采用 ImageNet 给出的数值归一化。接着构造 test dataloader，设置 batch size，采用多线程，shuffle=False。

train_transform = transforms.Compose([transforms.Resize((256, 256)),  # 先调整图片大小至256x256transforms.RandomCrop((224, 224)),  # 再随机裁剪到224x224transforms.RandomHorizontalFlip(),  # 随机的图像水平翻转，通俗讲就是图像的左右对调transforms.ToTensor(),transforms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225))  # 归一化，数值是用ImageNet给出的数值
])cat_files = [tf for tf in train_files if 'cat' in tf]
dog_files = [tf for tf in train_files if 'dog' in tf]cats = CatDogDataset(cat_files, train_dir, transform = train_transform)
dogs = CatDogDataset(dog_files, train_dir, transform = train_transform)train_set = ConcatDataset([cats, dogs])
train_loader = DataLoader(train_set, batch_size = BATCH_SIZE, shuffle=True, num_workers=0)test_transform = transforms.Compose([transforms.Resize((224, 224)),transforms.ToTensor(),transforms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225))
])test_set = CatDogDataset(test_files, test_dir, mode='test', transform = test_transform)
test_loader = DataLoader(test_set, batch_size = BATCH_SIZE, shuffle=False, num_workers=0)

查看训练集经过 transform 处理的一个 batch 的图片集。在张量里, image 是 (batch, width, height), 所以我们需要转置成 (width, height, batch) 来展示。

samples, labels = iter(train_loader).next()
plt.figure(figsize=(16,24))
grid_imgs = torchvision.utils.make_grid(samples[:BATCH_SIZE])
np_grid_imgs = grid_imgs.numpy()
# in tensor, image is (batch, width, height), so you have to transpose it to (width, height, batch) in numpy to show it.
plt.imshow(np.transpose(np_grid_imgs, (1,2,0)))

网络配置

自己搭建一个卷积神经网络，网络分为特征提取、平均池化、分类器、softmax。定义如下:


class MineNet(nn.Module):def __init__(self,num_classes=2):super().__init__()self.features=nn.Sequential(nn.Conv2d(3,64,kernel_size=11,stride=4,padding=2),   #(224+2*2-11)/4+1=55nn.ReLU(inplace=True),nn.MaxPool2d(kernel_size=3,stride=2),   #(55-3)/2+1=27nn.Conv2d(64,128,kernel_size=5,stride=1,padding=2), #(27+2*2-5)/1+1=27nn.ReLU(inplace=True),nn.MaxPool2d(kernel_size=3,stride=2),   #(27-3)/2+1=13nn.Conv2d(128,256,kernel_size=3,stride=1,padding=1),    #(13+1*2-3)/1+1=13nn.ReLU(inplace=True),nn.Conv2d(256,128,kernel_size=3,stride=1,padding=1),    #(13+1*2-3)/1+1=13nn.ReLU(inplace=True),nn.Conv2d(128,128,kernel_size=3,stride=1,padding=1),    #13+1*2-3)/1+1=13nn.ReLU(inplace=True),nn.MaxPool2d(kernel_size=3,stride=2),   #(13-3)/2+1=6)   #6*6*128=9126self.avgpool=nn.AdaptiveAvgPool2d((6,6))self.classifier=nn.Sequential(nn.Dropout(),nn.Linear(128*6*6,2048),nn.ReLU(inplace=True),nn.Dropout(),nn.Linear(2048,512),nn.ReLU(inplace=True),nn.Linear(512,num_classes),)# softmaxself.logsoftmax = nn.LogSoftmax(dim=1)def forward(self,x):x=self.features(x)x=self.avgpool(x)x=x.view(x.size(0),-1)x=self.classifier(x)x=self.logsoftmax(x)return x

初始化模型，用交叉熵计算 loss;优化器采用随机梯度下降法 SGD，学习率设置为 0.01，动量因
子设置为 0.9，权重衰减设置为 5;使用 StepLR 在每 5 个 epoch 结束时调整学习率。

model = MineNet()
# model = MyConvNet().to(DEVICE)
optimizer = torch.optim.SGD(model.parameters(), lr=0.01, momentum=0.9, weight_decay=5e-4)  # 设置训练细节
scheduler = StepLR(optimizer, step_size=5)
criterion = nn.CrossEntropyLoss()

模型训练
由于测试集没有 label，所以我把训练集按 9:1 划分成训练集和验证集，其中验证集是每轮 epoch 开始之前调用 refreshdataloader 在猫狗子集合中各抽 10%，意思就是验证集中猫狗图片各占一半，训练集 22500 张图片，验证集 2500 张图片，验证集数据使用 test_transform 处理。

def refreshdataloader():    cat_files = [tf for tf in train_files if 'cat' in tf]dog_files = [tf for tf in train_files if 'dog' in tf]val_cat_files = []val_dog_files = []for i in range(0,1250):r = random.randint(0,len(cat_files)-1)val_cat_files.append(cat_files[r])val_dog_files.append(dog_files[r])cat_files.remove(cat_files[r])dog_files.remove(dog_files[r])cats = CatDogDataset(cat_files, train_dir, transform = train_transform)dogs = CatDogDataset(dog_files, train_dir, transform = train_transform)train_set = ConcatDataset([cats, dogs])train_loader = DataLoader(train_set, batch_size = BATCH_SIZE, shuffle=True, num_workers=1)val_cats = CatDogDataset(val_cat_files, train_dir, transform = test_transform)val_dogs = CatDogDataset(val_dog_files, train_dir, transform = test_transform)val_set = ConcatDataset([val_cats, val_dogs])val_loader = DataLoader(val_set, batch_size = BATCH_SIZE, shuffle=True, num_workers=1)return train_loader,val_loader

定义训练函数，print 进度和过程中的 loss，损失函数使用交叉熵 loss，每一 epoch 结束后调整学习率，print 每一 epoch 的平均 loss 和 accuracy。

def train(model, device, train_loader, optimizer, epoch):model.train()train_loss = 0.0train_acc = 0.0percent = 10for batch_idx, (sample, target) in enumerate(train_loader):sample, target = sample.to(device), target.to(device)optimizer.zero_grad()output = model(sample)loss = criterion(output, target)loss.backward()optimizer.step()loss = loss.item()train_loss += losspred = output.max(1, keepdim = True)[1] train_acc += pred.eq(target.view_as(pred)).sum().item()if (batch_idx+1)%percent == 0:print('train epoch: {} [{}/{} ({:.0f}%)]\tloss: {:.6f}\t'.format(epoch, (batch_idx+1) * len(sample), len(train_loader.dataset),100. * batch_idx / len(train_loader), loss))train_loss *= BATCH_SIZEtrain_loss /= len(train_loader.dataset)                     train_acc = train_acc/len(train_loader.dataset)    print('\ntrain epoch: {}\tloss: {:.6f}\taccuracy:{:.4f}% '.format(epoch,train_loss,100.*train_acc))scheduler.step()return train_loss,train_acc

定义验证函数，计算并打印每一 epoch 的验证集的平均 loss 和总体 accuracy, 计算平均 loss 时记得乘上 batch size，因为 CrossEntropyLoss 计算出的 loss 对 mini-batch 的大小取了平均。

def val(model, device, val_loader,epoch):model.eval()val_loss =0.0correct = 0for sample, target in val_loader:with torch.no_grad():sample,target = sample.to(device),target.to(device)output = model(sample)val_loss += criterion(output, target).item()pred = output.max(1, keepdim = True)[1]correct += pred.eq(target.view_as(pred)).sum().item()val_loss *= BATCH_SIZEval_loss /= len(val_loader.dataset)val_acc= correct / len(val_loader.dataset)print("\nval set: epoch{} average loss: {:.4f}, accuracy: {}/{} ({:.4f}%) \n".format(epoch, val_loss, correct, len(val_loader.dataset),100.* val_acc))return val_loss,100.*val_acc

定义测试函数，把测试集的预测结果按(文件名，结果)的格式存成 csv 文件。


def test(model, device, test_loader,epoch):model.eval()filename_list = []pred_list = []for sample, filename in test_loader:with torch.no_grad():sample = sample.to(device)output = model(sample)pred = torch.argmax(output, dim=1)filename_list += [n[:-4] for n in filename]pred_list += [p.item() for p in pred]print("\ntest epoch: {}\n".format(epoch))submission = pd.DataFrame({"id":filename_list, "label":pred_list})submission.to_csv('preds_' + str(epoch) + '.csv', index=False)

预定义 4 个数组，分别存储训练集的 loss 和 accuracy、验证集的 loss 和 accuracy，然后开始训练、验证和测试，保存每一轮 epoch 结束的模型，以便之后继续训练和异常情况。

train_losses = []
train_acces = []
val_losses = []
val_acces = []for epoch in range(1, EPOCHS + 1):train_loader,val_loader = refreshdataloader()tr_loss,tr_acc = train(model,  DEVICE, train_loader, optimizer, epoch)train_losses.append(tr_loss)train_acces.append(tr_acc)vl,va = val(model, DEVICE, val_loader,epoch)val_losses.append(vl)val_acces.append(va)filename_pth = 'catdog_mineresnet_' + str(epoch) + '.pth'torch.save(model.state_dict(), filename_pth)test(model,DEVICE,test_loader)

ResNet18

Pytorch 的 ResNet18 接口的最后一层全连接层的输出维度是 1000。这不符合猫狗大战数据集，因为猫狗大战数据集是二分类的，所以最后一层全连接层输出的维度应该是 2 才对。因此我们需要对 ResNet18 进行最后一层的修改。取掉 ResNet18 model 的后 1 层，加上一层参数修改好的全连接层，输出为 2。训练、验证方法不变。

class Net(nn.Module):def __init__(self, model):super(Net, self).__init__()self.resnet_layer = nn.Sequential(*list(model.children())[:-1])self.Linear_layer = nn.Linear(512, 2)def forward(self, x):x = self.resnet_layer(x) x = x.view(x.size(0), -1) x = self.Linear_layer(x)return xfrom torchvision.models.resnet import resnet18
resnet = resnet18(pretrained=True)
model = Net(resnet)
model = model.to(DEVICE)
optimizer = torch.optim.SGD(model.parameters(), lr=0.001, momentum=0.9, weight_decay=5e-4)  # 设置训练细节
scheduler = StepLR(optimizer, step_size=3)
criterion = nn.CrossEntropyLoss()

VGG16 网络

因为猫狗大战数据集是二分类的，所以最后一层全连接层输出的维度应该是 2 才对。因此我们需要对 VGG16 进行最后一层的修改。把 Pytorch 的 VGG16 接口 model 的 classifier 替换成输出为 2 分类的。训练、验证方法不变。

from torchvision.models.vgg import vgg16
model = vgg16(pretrained=True)
for parma in model.parameters():parma.requires_grad = Falsemodel.classifier = nn.Sequential(nn.Linear(25088, 4096),nn.ReLU(),nn.Dropout(p=0.5),nn.Linear(4096, 4096),nn.ReLU(),nn.Dropout(p=0.5),nn.Linear(4096, 2))for index, parma in enumerate(model.classifier.parameters()):if index == 6:parma.requires_grad = Truemodel = model.to(DEVICE)
optimizer = torch.optim.SGD(model.parameters(), lr=0.001, momentum=0.9, weight_decay=5e-4)  # 设置训练细节
scheduler = StepLR(optimizer, step_size=3)
criterion = nn.CrossEntropyLoss()

结果

我的网络训练 8 epoch 后验证集 accuracy 为 91.16%，ResNet18 网络训练 1 epoch 后验证集 accuracy 为 98.72%,VGG16 验证集训练 1 epoch 后验证集准确度达到了 98.76%。