圖像分類項目-食物分類(監督學習和半監督學習)

項目介紹

數據處理

設定隨機種子

由于神經網絡的訓練具有隨機性，為了保證之前得到的好的訓練效果可以得到復現，設定隨機種子，讓訓練過程中的隨機行為每次訓練都是相同。

def seed_everything(seed):torch.manual_seed(seed)torch.cuda.manual_seed(seed)torch.cuda.manual_seed_all(seed)torch.backends.cudnn.benchmark = Falsetorch.backends.cudnn.deterministic = Truerandom.seed(seed)np.random.seed(seed)os.environ['PYTHONHASHSEED'] = str(seed)
#################################################################
seed_everything(0)
###############################################

讀取文件內容

進行數據處理前，需要了解數據的形式訓練集中，有標簽的數據按照11類分別存放在11個文件夾中，因此要循環依次讀取這11個文件夾的內容：

首先從文件夾中讀出每張圖片和對應標簽（讀取的是帶標簽的數據）：

HW = 224def read_file(path):for i in tqdm(range(11)):file_dir = path + "/%02d" % ifile_list = os.listdir(file_dir)  # 列出文件夾下所有文件名字xi = np.zeros((len(file_list), HW, HW, 3), dtype=np.uint8)  # 每個元素存一個圖片,圖片為整形類型yi = np.zeros(len(file_list))for j, img_name in enumerate(file_list):img_path = os.path.join(file_dir, img_name)  # 拼接地址img = Image.open(img_path)  # 打開圖片img = img.resize((HW, HW))  # 修改成模型接受的大小xi[j, ...] = imgyi[j] = iif i == 0:#第一個數據賦值X = xiY = yielse:#后續數據尾插X = np.concatenate((X, xi), axis=0)Y = np.concatenate((Y, yi), axis=0)print("讀到了%d個數據" % len(Y))return X, Y

圖像增廣

模型對訓練使用的圖片數據有好的效果，但是如果對圖片數據進行一定的變化，模型的效果就變差，因此在訓練時，不僅使用原圖片訓練，還要對圖片進行旋轉，放大裁切等圖像操作，將原圖片和操作后的圖片都作為訓練數據，也就是圖像增廣，讓模型的效果更好。

train_transform = transforms.Compose(#定義訓練集增廣方式[transforms.ToPILImage(), #224,224,3模型：3，224，244transforms.RandomResizedCrop(224),#隨機放大裁切transforms.RandomRotation(50),#50度以內隨機旋轉transforms.ToTensor()#模型運行的數據類型為張量]
)val_transform = transforms.Compose(#驗證集不需要增廣[transforms.ToPILImage(),  # 224,224,3模型：3，224，244transforms.ToTensor()  # 模型運行的數據類型為張量]
)

定義Dataset類

class food_Dataset(Dataset):#繼承Dateset類def __init__(self, path, mode="train"):self.X, self.Y = read_file(path)self.Y = torch.LongTensor(self.Y)#圖片數據類型為整形if mode == "train":#根據模式選擇增廣類型self.transform = train_transformelse:self.transform = val_transformdef __getitem__(self, item):return self.transform(self.X[item]), self.Y[item] #使用圖片增廣def __len__(self):return len(self.Y)

模型定義

在模型中設定一些卷積、歸一化、池化、激活函數對數據進行特征提取。

class myModel(nn.Module):def __init__(self, num_class):super(myModel, self).__init__()#3*224*224->512*7*7->拉直->全連接分類self.conv1 = nn.Conv2d(3, 64, 3, 1, 1) #3厚度，64個卷積核，卷積核大小3，padding為1，步長為1 64*224*224self.bn1 = nn.BatchNorm2d(64)#歸一化self.relu = nn.ReLU()self.pool1 = nn.MaxPool2d(2)   #64*112*112self.layer1 = nn.Sequential(nn.Conv2d(64, 128, 3, 1, 1),    # 128*112*112nn.BatchNorm2d(128),nn.ReLU(),nn.MaxPool2d(2)   #128*56*56)self.layer2 = nn.Sequential(nn.Conv2d(128, 256, 3, 1, 1),nn.BatchNorm2d(256),nn.ReLU(),nn.MaxPool2d(2)   #256*28*28)self.layer3 = nn.Sequential(nn.Conv2d(256, 512, 3, 1, 1),nn.BatchNorm2d(512),nn.ReLU(),nn.MaxPool2d(2)   #512*14*14)self.pool2 = nn.MaxPool2d(2)    #512*7*7self.fc1 = nn.Linear(25088, 1000)   #25088->1000self.relu2 = nn.ReLU()self.fc2 = nn.Linear(1000, num_class)  #1000-11def forward(self, x):#使用定義的模型進行前向過程x = self.conv1(x)x = self.bn1(x)x = self.relu(x)x = self.pool1(x)x = self.layer1(x)x = self.layer2(x)x = self.layer3(x)x = self.pool2(x)x = x.view(x.size()[0], -1) #拉直，x.size(0) 就是批量大小（batch_size），表示你有多少個樣本輸入到模型中。-1 是自動計算剩下的維度以便將數據展平。x = self.fc1(x)x = self.relu2(x)x = self.fc2(x)return x

遷移學習

良好的模型是需要大量的數據訓練得到的，由于我們設備加上數據量的限制訓練出來的模型效果不會特別好，甚至預測效果接近隨機預測，因此我們要進行遷移學習。簡單來說，**遷移學習就是使用大佬們用大量數據訓練出來的現成模型，**由于大佬的模型經過訓練后有很好的特征提取效果，因此我們只需要使用大佬的模型然后加上分類頭作為訓練的模型即可。

預訓練是指在無關當前任務的模型訓練，遷移學習使用的模型就是進行過預訓練的模型，遷移學習時可以進行線性探測和微調，線性探測就是在訓練中不進行參數的調整，完全信任遷移學習使用的模型，微調就是在訓練過程中會進行參數調整。

遷移學習時可選擇只使用架構和使用架構和參數，雖然遷移學習使用的架構很優秀但是參數是更加重要的部分，因此使用架構和參數的效果要更好，要想使用遷移學習的預訓練參數要保持架構一致。

from torchvision.models import resnet18#導入模型
model = resnet18(pretrained=True)#使用架構和參數
in_fetures = model.fc.in_features#獲取模型的特征提取后的輸出維度
model.fc = nn.Linear(in_fetures, 11)#全連接分類頭

定義超參

定義學習率、損失函數、優化器、訓練輪次等超參數。

Adam和AdamW

Adam優化器不僅考慮當前點的梯度還考慮之前的梯度，并且會自動更改學習率，由于參數更改時要減去學習率×梯度，當這個值過大時，Adam會自動更改學習率，AdamW是在Adam的基礎上增加了權重衰減使得模型曲線更加平滑。

訓練過程

def train_val(model, train_loader, val_loader, device, epochs, optimizer, loss, save_path):model = model.to(device)plt_train_loss = [] #記錄所有輪次的LOSSplt_val_loss = []plt_train_acc = [] #記錄準確率plt_val_acc = []max_acc = 0.0for epoch in range(epochs): #開始訓練train_loss = 0.0val_loss = 0.0train_acc = 0.0#用準確率表示模型效果val_acc = 0.0start_time = time.time()model.train() #模型調為訓練模式，有時訓練模式和測試模式的模型不同for batch_x, batch_y in train_loader:x, target = batch_x.to(device), batch_y.to(device)pred = model(x)train_bat_loss = loss(pred, target) #獲取一批數據的LOSStrain_bat_loss.backward() #梯度回傳optimizer.step()#更新模型optimizer.zero_grad()train_loss += train_bat_loss.cpu().item() #將gpu上的張量數據放到cpu上取出數據計算,累加記錄本輪LOSStrain_acc += np.sum(np.argmax(pred.detach().cpu().numpy(), axis=1) == target.cpu().numpy())#記錄預測對的數量plt_train_loss.append(train_loss / train_loader.__len__()) #除以輪次數，得到每個輪次的LOSS平均值plt_train_acc.append(train_acc/train_loader.dataset.__len__()) #記錄準確率，model.eval()#調為驗證模式with torch.no_grad():#所有模型中的張量計算都積攢梯度，而驗證時不需要梯度for batch_x, batch_y in val_loader:x, target = batch_x.to(device), batch_y.to(device)pred = model(x)val_bat_loss = loss(pred, target)val_loss += val_bat_loss.cpu().item()val_acc += np.sum(np.argmax(pred.detach().cpu().numpy(), axis=1) == target.cpu().numpy())  # 記錄預測對的數量plt_val_loss.append(val_loss / val_loader.dataset.__len__())plt_val_acc.append(val_acc/val_loader.dataset.__len__()) #記錄準確率，if val_acc > max_acc: #如果當前模型效果更好，進行記錄torch.save(model, save_path)max_acc = val_loss#訓練效果打印print('[%03d/%03d] %2.2f sec(s) TrainLoss : %.6f | valLoss: %.6f Trainacc : %.6f | valacc: %.6f' % \(epoch, epochs, time.time() - start_time, plt_train_loss[-1], plt_val_loss[-1], plt_train_acc[-1],plt_val_acc[-1]))  # 打印訓練結果。 注意python語法， %2.2f 表示小數位為2的浮點數， 后面可以對應。plt.plot(plt_train_loss)plt.plot(plt_val_loss)plt.title("loss")plt.legend(["train", "val"])plt.show()plt.plot(plt_train_acc)plt.plot(plt_val_acc)plt.title("loss")plt.legend(["train", "val"])plt.show()

半監督學習

監督學習是指每個訓練樣本都有對應的標簽，模型通過學習這些標注數據訓練，目標是讓模型能夠根據新的輸入數據預測正確的標簽。

半監督學習是介于監督學習和無監督學習之間的一種方法。在半監督學習中，訓練數據包含大量的未標注數據和少量的標注數據。模型利用少量的標注數據來進行學習，同時也借助未標注數據來進一步提高模型的性能。

• 模型首先使用標注數據進行訓練。
• 模型的效果達到一定程度后，用訓練得到的模型對未標注數據進行預測。
• 若預測結果結果的置信值（成功率）達到一定值后，將預測結果（偽標簽）添加到訓練數據集中。

為了加入半監督學習，對監督學習的代碼進行修改。

定義Dataset類

class food_Dataset(Dataset):def __init__(self, path, mode="train"):self.mode = modeif mode == "semi":#若為半監督模式，數據只有X，沒有標簽Yself.X = self.read_file(path)else:self.X, self.Y = self.read_file(path)self.Y = torch.LongTensor(self.Y)  #標簽轉為長整形if mode == "train":#訓練模式需要圖片增廣等操作self.transform = train_transformelse:#非訓練模式，包括半監督模式下，只需要讓數據轉換成符合模型輸入的格式即可self.transform = val_transformdef read_file(self, path):#讀取數據函數if self.mode == "semi":file_list = os.listdir(path)xi = np.zeros((len(file_list), HW, HW, 3), dtype=np.uint8)# 列出文件夾下所有文件名字for j, img_name in enumerate(file_list):img_path = os.path.join(path, img_name)img = Image.open(img_path)img = img.resize((HW, HW))xi[j, ...] = imgprint("讀到了%d個數據" % len(xi))return xielse:for i in tqdm(range(11)):file_dir = path + "/%02d" % ifile_list = os.listdir(file_dir)xi = np.zeros((len(file_list), HW, HW, 3), dtype=np.uint8)yi = np.zeros(len(file_list), dtype=np.uint8)# 列出文件夾下所有文件名字for j, img_name in enumerate(file_list):img_path = os.path.join(file_dir, img_name)img = Image.open(img_path)img = img.resize((HW, HW))xi[j, ...] = imgyi[j] = iif i == 0:X = xiY = yielse:X = np.concatenate((X, xi), axis=0)Y = np.concatenate((Y, yi), axis=0)print("讀到了%d個數據" % len(Y))return X, Ydef __getitem__(self, item):if self.mode == "semi":return self.transform(self.X[item]), self.X[item]#前者為為了輸入模型進行轉換的X用于訓練得到偽標簽，后者為原始數據X用于加入半監督數據集else:return self.transform(self.X[item]), self.Y[item]def __len__(self):return len(self.X)class semiDataset(Dataset):#半監督數據集Dataset類def __init__(self, no_label_loder, model, device, thres=0.99):#傳入無標簽數據，預測模型，置信度x, y = self.get_label(no_label_loder, model, device, thres)if x == []:#如果預測得到的偽標簽都不符合要求，如置信度低，導致半監督數據集為空進行標記self.flag = Falseelse:self.flag = Trueself.X = np.array(x)self.Y = torch.LongTensor(y)self.transform = train_transform#得到的半監督數據集同樣用于模型訓練def get_label(self, no_label_loder, model, device, thres):#給半監督數據打標簽model = model.to(device)pred_prob = []#記錄預測類型中最高概率labels = []#記錄最高概率對應的標簽x = []y = []soft = nn.Softmax()with torch.no_grad():#只要通過模型就會積攢梯度，只要不進行模型訓練調整，積攢的梯度就沒用for bat_x, _ in no_label_loder:bat_x = bat_x.to(device)pred = model(bat_x)pred_soft = soft(pred)pred_max, pred_value = pred_soft.max(1)#維度1為橫向，取出最高概率和其對應的標簽，由于loader中一個元素是一批數據，因此pred_max和pred_value的一個元素中包含對應批數個值pred_prob.extend(pred_max.cpu().numpy().tolist())labels.extend(pred_value.cpu().numpy().tolist())for index, prob in enumerate(pred_prob):if prob > thres:#大于置信度加入半監督數據集x.append(no_label_loder.dataset[index][1])   #調用到原始的getitem，因為要加入半監督數據集y.append(labels[index])return x, ydef __getitem__(self, item):return self.transform(self.X[item]), self.Y[item]def __len__(self):return len(self.X)def get_semi_loader(no_label_loder, model, device, thres):#獲取半監督數據集semiset = semiDataset(no_label_loder, model, device, thres)if semiset.flag == False:return Noneelse:semi_loader = DataLoader(semiset, batch_size=16, shuffle=False)return semi_loader

模型定義

加入半監督學習只需要復用監督學習的訓練模型進行預測即可。

定義超參

加入半監督學習要額外定義包括置信度的超參。

訓練過程

def train_val(model, train_loader, val_loader, no_label_loader, device, epochs, optimizer, loss, thres, save_path):model = model.to(device)semi_loader = Noneplt_train_loss = []plt_val_loss = []plt_train_acc = []plt_val_acc = []max_acc = 0.0for epoch in range(epochs):train_loss = 0.0val_loss = 0.0train_acc = 0.0val_acc = 0.0semi_loss = 0.0#半監督數據集LOSSsemi_acc = 0.0#對半監督數據集的預測準確率start_time = time.time()model.train()#訓練模式for batch_x, batch_y in train_loader:#使用有標簽訓練集訓練x, target = batch_x.to(device), batch_y.to(device)pred = model(x)train_bat_loss = loss(pred, target)train_bat_loss.backward()optimizer.step()  # 更新參數 之后要梯度清零否則會累積梯度optimizer.zero_grad()train_loss += train_bat_loss.cpu().item()train_acc += np.sum(np.argmax(pred.detach().cpu().numpy(), axis=1) == target.cpu().numpy())plt_train_loss.append(train_loss / train_loader.__len__())plt_train_acc.append(train_acc/train_loader.dataset.__len__()) #記錄準確率，if semi_loader!= None:#若半監督數據集非空，使用半監督數據集進行訓練for batch_x, batch_y in semi_loader:x, target = batch_x.to(device), batch_y.to(device)pred = model(x)semi_bat_loss = loss(pred, target)semi_bat_loss.backward()optimizer.step()  # 更新參數 之后要梯度清零否則會累積梯度，因為下一輪數據要重新計算梯度optimizer.zero_grad()semi_loss += train_bat_loss.cpu().item()semi_acc += np.sum(np.argmax(pred.detach().cpu().numpy(), axis=1) == target.cpu().numpy())print("半監督數據集的訓練準確率為", semi_acc/train_loader.dataset.__len__())model.eval()with torch.no_grad():for batch_x, batch_y in val_loader:x, target = batch_x.to(device), batch_y.to(device)pred = model(x)val_bat_loss = loss(pred, target)val_loss += val_bat_loss.cpu().item()val_acc += np.sum(np.argmax(pred.detach().cpu().numpy(), axis=1) == target.cpu().numpy())plt_val_loss.append(val_loss / val_loader.dataset.__len__())plt_val_acc.append(val_acc / val_loader.dataset.__len__())if epoch%3 == 0 and plt_val_acc[-1] > 0.6:#將模型訓練至一定能力后，再進行半監督學習semi_loader = get_semi_loader(no_label_loader, model, device, thres)if val_acc > max_acc:torch.save(model, save_path)max_acc = val_lossprint('[%03d/%03d] %2.2f sec(s) TrainLoss : %.6f | valLoss: %.6f Trainacc : %.6f | valacc: %.6f' % \(epoch, epochs, time.time() - start_time, plt_train_loss[-1], plt_val_loss[-1], plt_train_acc[-1], plt_val_acc[-1]))  # 打印訓練結果。 注意python語法， %2.2f 表示小數位為2的浮點數， 后面可以對應。plt.plot(plt_train_loss)plt.plot(plt_val_loss)plt.title("loss")plt.legend(["train", "val"])plt.show()plt.plot(plt_train_acc)plt.plot(plt_val_acc)plt.title("acc")plt.legend(["train", "val"])plt.show()