CNN

代码

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torchvision import datasets,transforms
import matplotlib.pyplot as plt
import numpy as np
%matplotlib inline

读取数据

#定义超参数
input_size=28
num_class=10
num_epochs=3
batch_size=64

#训练集
train_dataset=datasets.MNIST(root='./data',train=True,transform=transforms.ToTensor(),download=True)test_dataset=datasets.MNIST(root='./data',train=False,transform=transforms.ToTensor())

#构建batch数据
train_loader=torch.utils.data.DataLoader(dataset=train_dataset,batch_size=batch_size,shuffle=True) #num_worker=4 使用4个子线程加载数据
test_loader=torch.utils.data.DataLoader(dataset=test_dataset,batch_size=batch_size,shuffle=False)

train_data_iter=iter(train_loader)
#获取训练集的第一个批次数据（第一个快递包）
batch_x,batch_y=next(train_data_iter)
print(batch_x.shape,batch_y.shape)test_data_iter=iter(test_loader)
batch_x_test,batch_y_test=next(test_data_iter)
print(batch_x_test.shape,batch_y_test.shape)

搭建CNN

class CNN(nn.Module):def __init__(self):super(CNN,self).__init__() #batch_size,1,28,28self.conv1=nn.Sequential(nn.Conv2d(in_channels=1,out_channels=16,kernel_size=5,stride=1,padding=2), #batch_size,16,28,28nn.ReLU(),nn.MaxPool2d(kernel_size=2), #batch_size,16,14,14)self.conv2=nn.Sequential(nn.Conv2d(16,32,5,1,2), #batch_size,32,14,14nn.ReLU(),nn.Conv2d(32,32,5,1,2), #batch_size,32,14,14  #输入输出通道不变，让其在隐藏层里面更进一步提取特征nn.ReLU(),nn.MaxPool2d(2), #batch_size,32,7,7)self.conv3=nn.Sequential(nn.Conv2d(32,64,5,1,2), #batch_size,64,7,7nn.ReLU(),)#batch_size,64*7*7self.out=nn.Linear(64*7*7,10)def forward(self,x):x=self.conv1(x)x=self.conv2(x)x=nn.Flatten(self.conv3(x))output=self.out(x)return output

def accuracy(prediction,labels):pred=torch.argmax(prediction.data,dim=1) #prediction.data中加data是为了防止数据里面单独数据可能会带来梯度信息rights=pred.eq(labels.data,view_as(pred)).sum()return rights,len(labels) #(batch_size,)/(batch_size,1)

训练网络模型

net=CNN()criterion=nn.CrossEntropyLoss() #不需要在CNN中将logistic转换为概率，因为pytorch的交叉熵损失函数会自动进行optimizer=optim.Adam(net.parameters(),lr=0.001)for epoch in range(num_epochs):train_rights=[]for batch_idx,(data,target) in enumerate(train_loader):net.train() #进入训练状态，也就是所有网络参数都处于可更新状态output=net(data) #output只是logits得分loss=criterion(output,target)optimizer.zero_grad()loss.backward()optimizer.step()right=accuracy(output,target)train_rights.append(right)if batch_idx %100 ==0:net.eval() #进入评估模式，自动关闭求导机制和模型中的BN层drop out层val_rights=[]for (data,target) in test_loader:output=net(data)right=accuracy(output,target)val_rights.append(right)train_r=(sum([tup[0] for tup in train_rights]),sum([tup[1] for tup in train_rights]))val_r=(sum([tup[0] for tup in val_rights]),sum([tup[1] for tup in val_rights]))print('当前epoch:{} [{}/{} ({:.0f}%)]\t损失:{:.6f}\t训练集准确率:{:.2f}%\t测试集准确率:{:.2f}%'.format(epoch,batch_idx*batch_size,len(train_loader.dataset),100.*batch_idx/len(train_loader),loss.data,100.*train_r[0].numpy()/train_r[1],100.*val_r[0].numpy()/val_r[1]))

数据增强

比如数据不够，可以对数据进行旋转，翻转等操作来添加数据

迁移学习

例如使用预训练模型

图像识别策略

输出为102

数据读取

data_dir = './汪学长的随堂资料/2/flower_data/'
train_dir = data_dir + '/train' # 训练数据的文件路径
valid_dir = data_dir + '/valid' # 验证数据的文件路径

定义数据预处理操作

data_transforms = {'train':transforms.Compose([transforms.Resize([96, 96]),transforms.RandomRotation(45), # 随机旋转， -45~45度之间transforms.CenterCrop(64), #对中心进行裁剪，变成64*64transforms.RandomHorizontalFlip(p=0.5),transforms.RandomVerticalFlip(p=0.5),transforms.ColorJitter(brightness=0.2, contrast=0.1, saturation=0.1, hue=0.1), # 亮度、对比度、饱和度、色调transforms.RandomGrayscale(p=0.025), #彩色图变成灰度图transforms.ToTensor(), # 0-255 ——> 0-1transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]) #这组均值和标准差是最适合图片进行使用的，因为是3通道所以有3组]),'valid':transforms.Compose([transforms.Resize([64, 64]),transforms.ToTensor(),transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]) ]),}

image_datasets

dataloaders

dataset_sizes

model_name = "resnet18" # resnet34, resnet50, feature_extract = True #使用训练好的参数

冻结resnet18的函数

def set_parameter_requires_gard(model ,feature_extracting):if feature_extracting:for param in model.parameters():param.requires_grad = False

model_ft = models.resnet18() #内置的resnet18
model_ft

改最后一层的，因为默认的是1000输出

把模型输出层改成自己的

def initialize_model(feature_extract, use_pretrained=True):model_ft = models.resnet18(pretrained = use_pretrained)set_parameter_requires_gard(model_ft, feature_extract)model_ft.fc = nn.Linear(512, 102)input_size = 64return model_ft, input_size

设置哪些层需要训练

model_ft, input_size = initialize_model(feature_extract, use_pretrained=True)device = torch.device("mps") # cuda/cpumodel_ft = model_ft.to(device)filename = 'best.pt' # .pt .pthparams_to_update = model_ft.parameters()if feature_extract:params_to_update = []for name, parm in model_ft.named_parameters():if parm.requires_grad == True:params_to_update.append(parm)print(name)

model_ft

设置优化器和损失函数

optimizer_ft = optim.Adam(params_to_update, lr=1e-3)# 定义学习率调度器
scheduler = optim.lr_scheduler.StepLR(optimizer_ft, step_size=10, gamma=0.1)criterion = nn.CrossEntropyLoss()

optimizer_ft.param_groups[0]

训练

def train_model(model, dataloaders, criterion, optimizer, num_epochs=50, filename="best.pt"):# 初始化一些变量since = time.time() # 记录初始时间best_acc = 0 # 记录验证集上的最佳精度model.to(device)train_acc_history = []val_acc_history = []train_losses = []valid_losses = []LRS = [optimizer.param_groups[0]['lr']]best_model_wts = copy.deepcopy(model.state_dict())for epoch in range(num_epochs):print('Epoch {}/{}'.format(epoch + 1, num_epochs))print('-' * 10)# 在每个epoch内，遍历训练和验证两个阶段for phase in ['train', 'valid']:if phase == 'train':model.train()else:model.eval()running_loss = 0.0 # 累积训练过程中的损失running_corrects = 0 # 累积训练过程中的正确预测的样本数量for inputs, labels in dataloaders[phase]:inputs = inputs.to(device)labels = labels.to(device)outputs = model(inputs)loss = criterion(outputs, labels)preds = torch.argmax(outputs, dim=1)optimizer.zero_grad()if phase == 'train':loss.backward()optimizer.step()running_loss += loss.item() * inputs.size(0)running_corrects += torch.sum(preds == labels.data)epoch_loss = running_loss / len(dataloaders[phase].dataset)# 整个epoch的平均损失epoch_acc = running_corrects.float() / len(dataloaders[phase].dataset) # 整个epoch的准确率time_elapsed = time.time() - sinceprint('Time elapsed {:.0f}m {:.0f}s'.format(time_elapsed // 60, time_elapsed % 60))print('{} Loss: {:.4f}; ACC: {:.4f}'.format(phase, epoch_loss, epoch_acc))if phase == "valid" and epoch_acc > best_acc:best_acc = epoch_accbest_model_wts = copy.deepcopy(model.state_dict())state = {'state_dict': model.state_dict(),'best_acc': best_acc,'optimizer': optimizer.state_dict()}torch.save(state, filename)if phase == 'valid':val_acc_history.append(epoch_acc)valid_losses.append(epoch_loss)if phase == 'train':train_acc_history.append(epoch_acc)train_losses.append(epoch_loss)print('Optimizer learning rate : {:.7f}'.format(optimizer.param_groups[0]['lr']))LRS.append(optimizer.param_groups[0]['lr'])print()scheduler.step() # 调用学习率调度器来进行学习率更新操作# 已经全部训练完了time_elapsed = time.time() - sinceprint('Training complete in {:.0f}m {:.0f}s'.format(time_elapsed // 60, time_elapsed % 60))print('Best val Acc: {:.4f}'.format(best_acc))model.load_state_dict(best_model_wts)return model, val_acc_history, train_acc_history, valid_losses, train_losses ,LRS

开始训练

# def train_model(model, dataloaders, criterion, optimizer, num_epochs=50, filename="best.pt"):
model_ft, val_acc_history, train_acc_history, valid_losses, train_losses ,LRS = train_model(model_ft, dataloaders, criterion, optimizer_ft, num_epochs=5)

再训练所有层

# 解冻
for param in model_ft.parameters():parm.requires_grad = Trueoptimizer = optim.Adam(model_ft.parameters(), lr=1e-3)
scheduler = optim.lr_scheduler.StepLR(optimizer_ft, step_size=7, gamma=0.1) # 每7个epoch, 学习率衰减1/10
criterion = nn.CrossEntropyLoss()

# 加载之间训练好的权重参数
checkpoint = torch.load(filename)
best_acc = checkpoint['best_acc']
model_ft.load_state_dict(checkpoint['state_dict'])

model_ft, val_acc_history, train_acc_history, valid_losses, train_losses ,LRS = train_model(model_ft, dataloaders, criterion, optimizer_ft, num_epochs=3)

关机了，再开机，加载训练好的模型

model_ft, input_size = initialize_model(feature_extract, use_pretrained=True)filename = 'best.pt'# 加载模型
checkpoint = torch.load(filename)
model_ft.load_state_dict(checkpoint['state_dict'])