小董的BLOG
0%

发表于 更新于

对比学习入门，记录一下simclr的学习过程

1.模型建立

基于lightly框架

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import torch
import torchvision
from torch import nn

from lightly.loss import NTXentLoss
from lightly.models.modules import SimCLRProjectionHead
from lightly.transforms.simclr_transform import SimCLRTransform


class SimCLR(nn.Module):
    def __init__(self, backbone):
        super().__init__()
        self.backbone = backbone	#resnet最后平均池化后为512*1*1
        self.projection_head = SimCLRProjectionHead(512, 512, 128)#非线性化

    def forward(self, x):
        x = self.backbone(x).flatten(start_dim=1) #nn.flatten为展平为1维
        z = self.projection_head(x)	#
        return z


resnet = torchvision.models.resnet18()	#使用resnet作为特征识别网络
backbone = nn.Sequential(*list(resnet.children())[:-1])#返回残差结构迭代器列表并拼接为resnet，此处resnet.children()不会返回resnet本来的全连接层！仅到全连接层前的平均池化
model = SimCLR(backbone)

device = "cuda" if torch.cuda.is_available() else "cpu"
model.to(device)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
#resnet.children()测试
import torchvision
from torch import nn

resnet = torchvision.models.resnet18()
backbone = nn.Sequential(*list(resnet.children())[:-1])
print(backbone)

'''
Sequential(
  (0): Conv2d(3, 64, kernel_size=(7, 7), stride=(2, 2), padding=(3, 3), bias=False)
  (1): BatchNorm2d(64, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
  (2): ReLU(inplace=True)
  (3): MaxPool2d(kernel_size=3, stride=2, padding=1, dilation=1, ceil_mode=False)
  (4): Sequential(
    (0): BasicBlock(
      (conv1): Conv2d(64, 64, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn1): BatchNorm2d(64, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      (relu): ReLU(inplace=True)
      (conv2): Conv2d(64, 64, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn2): BatchNorm2d(64, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
    )
    (1): BasicBlock(
      (conv1): Conv2d(64, 64, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn1): BatchNorm2d(64, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      (relu): ReLU(inplace=True)
      (conv2): Conv2d(64, 64, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn2): BatchNorm2d(64, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
    )
  )
  (5): Sequential(
    (0): BasicBlock(
      (conv1): Conv2d(64, 128, kernel_size=(3, 3), stride=(2, 2), padding=(1, 1), bias=False)
      (bn1): BatchNorm2d(128, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      (relu): ReLU(inplace=True)
      (conv2): Conv2d(128, 128, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn2): BatchNorm2d(128, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      (downsample): Sequential(
        (0): Conv2d(64, 128, kernel_size=(1, 1), stride=(2, 2), bias=False)
        (1): BatchNorm2d(128, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      )
    )
    (1): BasicBlock(
      (conv1): Conv2d(128, 128, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn1): BatchNorm2d(128, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      (relu): ReLU(inplace=True)
      (conv2): Conv2d(128, 128, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn2): BatchNorm2d(128, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
    )
  )
  (6): Sequential(
    (0): BasicBlock(
      (conv1): Conv2d(128, 256, kernel_size=(3, 3), stride=(2, 2), padding=(1, 1), bias=False)
      (bn1): BatchNorm2d(256, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      (relu): ReLU(inplace=True)
      (conv2): Conv2d(256, 256, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn2): BatchNorm2d(256, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      (downsample): Sequential(
        (0): Conv2d(128, 256, kernel_size=(1, 1), stride=(2, 2), bias=False)
        (1): BatchNorm2d(256, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      )
    )
    (1): BasicBlock(
      (conv1): Conv2d(256, 256, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn1): BatchNorm2d(256, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      (relu): ReLU(inplace=True)
      (conv2): Conv2d(256, 256, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn2): BatchNorm2d(256, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
    )
  )
  (7): Sequential(
    (0): BasicBlock(
      (conv1): Conv2d(256, 512, kernel_size=(3, 3), stride=(2, 2), padding=(1, 1), bias=False)
      (bn1): BatchNorm2d(512, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      (relu): ReLU(inplace=True)
      (conv2): Conv2d(512, 512, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn2): BatchNorm2d(512, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      (downsample): Sequential(
        (0): Conv2d(256, 512, kernel_size=(1, 1), stride=(2, 2), bias=False)
        (1): BatchNorm2d(512, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      )
    )
    (1): BasicBlock(
      (conv1): Conv2d(512, 512, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn1): BatchNorm2d(512, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
      (relu): ReLU(inplace=True)
      (conv2): Conv2d(512, 512, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False)
      (bn2): BatchNorm2d(512, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
    )
  )
  (8): AdaptiveAvgPool2d(output_size=(1, 1))
)
'''

2.transform、dataset

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
transform = SimCLRTransform(input_size=32, gaussian_blur=0.0)
dataset = torchvision.datasets.CIFAR10(
    "datasets/cifar10", download=True, transform=transform
)



class SimCLRTransform(MultiViewTransform):	#继承MultiViewTransform父类
    def __init__(
        self,
        input_size: int = 224,
        cj_prob: float = 0.8,
        cj_strength: float = 1.0,
        cj_bright: float = 0.8,
        cj_contrast: float = 0.8,
        cj_sat: float = 0.8,
        cj_hue: float = 0.2,
        min_scale: float = 0.08,
        random_gray_scale: float = 0.2,
        gaussian_blur: float = 0.5,
        kernel_size: Optional[float] = None,
        sigmas: Tuple[float, float] = (0.1, 2),
        vf_prob: float = 0.0,
        hf_prob: float = 0.5,
        rr_prob: float = 0.0,
        rr_degrees: Optional[Union[float, Tuple[float, float]]] = None,
        normalize: Union[None, Dict[str, List[float]]] = IMAGENET_NORMALIZE,
    ):
        view_transform = SimCLRViewTransform(
            input_size=input_size,
            cj_prob=cj_prob,
            cj_strength=cj_strength,
            cj_bright=cj_bright,
            cj_contrast=cj_contrast,
            cj_sat=cj_sat,
            cj_hue=cj_hue,
            min_scale=min_scale,
            random_gray_scale=random_gray_scale,
            gaussian_blur=gaussian_blur,
            kernel_size=kernel_size,
            sigmas=sigmas,
            vf_prob=vf_prob,
            hf_prob=hf_prob,
            rr_prob=rr_prob,
            rr_degrees=rr_degrees,
            normalize=normalize,
        )
        super().__init__(transforms=[view_transform, view_transform])
			#通过父类MultiViewTransform生成一张图片的不同“视角”
         
class SimCLRViewTransform:
    def __init__(
        self,
        input_size: int = 224,
        cj_prob: float = 0.8,
        cj_strength: float = 1.0,
        cj_bright: float = 0.8,
        cj_contrast: float = 0.8,
        cj_sat: float = 0.8,
        cj_hue: float = 0.2,
        min_scale: float = 0.08,
        random_gray_scale: float = 0.2,
        gaussian_blur: float = 0.5,
        kernel_size: Optional[float] = None,
        sigmas: Tuple[float, float] = (0.1, 2),
        vf_prob: float = 0.0,
        hf_prob: float = 0.5,
        rr_prob: float = 0.0,
        rr_degrees: Optional[Union[float, Tuple[float, float]]] = None,
        normalize: Union[None, Dict[str, List[float]]] = IMAGENET_NORMALIZE,
    ):
        color_jitter = T.ColorJitter(
            brightness=cj_strength * cj_bright,
            contrast=cj_strength * cj_contrast,
            saturation=cj_strength * cj_sat,
            hue=cj_strength * cj_hue,
        )
			#随机数据增强
        transform = [
            T.RandomResizedCrop(size=input_size, scale=(min_scale, 1.0)),
            random_rotation_transform(rr_prob=rr_prob, rr_degrees=rr_degrees),
            T.RandomHorizontalFlip(p=hf_prob),
            T.RandomVerticalFlip(p=vf_prob),
            T.RandomApply([color_jitter], p=cj_prob),	#颜色抖动，对于对比学习很重要
            T.RandomGrayscale(p=random_gray_scale),	#随机灰度，同上
            GaussianBlur(kernel_size=kernel_size, sigmas=sigmas, prob=gaussian_blur),
            T.ToTensor(),
        ]
        if normalize:
            #归一化，此处列表可直接通过+=添加新元素
            transform += [T.Normalize(mean=normalize["mean"], std=normalize["std"])]
        self.transform = T.Compose(transform)

    def __call__(self, image: Union[Tensor, Image]) -> Tensor:  #该方法会在访问dataloader实例时自动调用（理解为图片在使用前才进行设置好的预处理），每次调用时transform的结果都是随机的
        transformed: Tensor = self.transform(image)
        return transformed

3.训练

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
dataloader = torch.utils.data.DataLoader(
    dataset,
    batch_size=256,
    shuffle=True,
    drop_last=True,
    num_workers=8,
)
criterion = NTXentLoss()	#loss，用于计算2个输入张量的相似度
optimizer = torch.optim.SGD(model.parameters(), lr=0.06)

print("Starting Training")
for epoch in range(10):
    total_loss = 0
    for batch in dataloader:
        x0, x1 = batch[0]	#batch[0]为image，[1]为label，对比学习不需要label
        x0 = x0.to(device)
        x1 = x1.to(device)
        z0 = model(x0)
        z1 = model(x1)
        loss = criterion(z0, z1)	#基于相似度的loss
        total_loss += loss.detach()
        loss.backward()
        optimizer.step()
        optimizer.zero_grad()
    avg_loss = total_loss / len(dataloader)
    print(f"epoch: {epoch:>02}, loss: {avg_loss:.5f}")

在for batch in dataloader:时，经过transform预处理后一张图片会返回2个随机view，也就是说x0,x1各自长度都为256

因此NTXentLoss()函数会比较一个批次中所有图像视图的嵌入表示（z0 和 z1），不仅仅是同一图像的不同视图（即正样本对，positive pair），还包括不同图像的不同视图（即负样本对，negative pair）。

3.基于simclr的下游分类任务

要使用经过 SimCLR 训练的模型进行图片分类任务，需要将其转换为一个有监督的分类模型。SimCLR 通过对比学习学习到的是图像的特征表示，但并没有学到类别信息。因此，接下来需要在 SimCLR 训练好的特征提取器基础上，加上一个新的分类头（如全连接层），并对这个分类器进行训练。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class ClassificationHead(nn.Module):
    def __init__(self, backbone, num_classes=10):# cirf10有10个类别
        super(ClassificationHead, self).__init__()
        self.backbone = backbone
        self.fc = nn.Linear(512, num_classes) #添加1个全连接层，输出为10个类别，此处512需根据使用的backbone种类确定（我也不知道是不是都是512）

    def forward(self, x):
        features = self.backbone(x).flatten(start_dim=1)	#记得全连接之前要展成1维
        return self.fc(features)

model.load_state_dict(torch.load('simclr.pth'))
backbone = model.backbone  # 从 训练好的SimCLR 模型中获取 backbone
    
# 冻结 backbone 的参数
for param in backbone.parameters():
    param.requires_grad = False

# 实例化分类模型
model = ClassificationHead(backbone, num_classes=10)
model = model.to(device)

# 加载 CIFAR-10 训练集和测试集（带标签的有监督训练）
class_transform = torchvision.transforms.Compose(
        [torchvision.transforms.Resize((32, 32)),
         torchvision.transforms.ToTensor(),
         torchvision.transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])

train_dataset = torchvision.datasets.CIFAR10(
        root="dataset", train=True, download=True, transform=class_transform
    )
test_dataset = torchvision.datasets.CIFAR10(
        root="dataset", train=False, download=True, transform=class_transform
    )

train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True, num_workers=4)
test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False, num_workers=4)

    # 定义损失函数和优化器
    fc_loss = nn.CrossEntropyLoss()  # 分类任务的损失函数，使用交叉熵损失函数
    optimizer = optim.Adam(model.fc.parameters(), lr=1e-3)  # 只更新全连接层的参数

    # 训练分类器
    print('classification fc training start')
    for epoch in range(10):
        model.train()
        total_loss = 0
        correct = 0
        total = 0

        for inputs, labels in train_loader:
            inputs, labels = inputs.to(device), labels.to(device)

            # 前向传播
            outputs = model(inputs)
            loss = fc_loss(outputs, labels)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            total_loss += loss.item()

            _, predicted = outputs.max(1)
            total += labels.size(0)
            correct += predicted.eq(labels).sum().item()

        avg_loss = total_loss / len(train_loader)
        accuracy = 100. * correct / total
        print(f"stage2 Epoch [{epoch + 1}/5], Loss: {avg_loss:.4f}, Accuracy: {accuracy:.2f}%")

    # 测试集
    model.eval()
    correct = 0
    total = 0
    with torch.no_grad():
        for inputs, labels in test_loader:
            inputs, labels = inputs.to(device), labels.to(device)
            outputs = model(inputs)
            total += labels.size(0)
            correct += predicted.eq(labels).sum().item()

    test_accuracy = 100. * correct / total
    print(f"stage2 Test Accuracy: {test_accuracy:.2f}%")

关于model.train和.eval

  • 训练模式（Training Mode）：如表格所示，在此模式下，模型会进行前向传播、反向传播以及参数更新。某些层，如Dropout层和BatchNorm层，在此模式下的行为会与评估模式下不同。例如，Dropout层会在训练过程中随机将一部分输入设置为0，以防止过拟合。

  • 评估模式（Evaluation Mode）：如表格所示，在此模式下，模型只会进行前向传播，不会进行反向传播和参数更新。Dropout层会停止dropout，BatchNorm层会使用在训练阶段计算得到的全局统计数据，而不是测试集中的批统计数据。

评估模式一般用于验证、测试集

关于torch.max

_, predicted = outputs.max(1)与torch.max(outputs, dim=1)相同，对于其返回值：

  • [0]：在全连接层所有输出中的最大值
  • [1]：最大值对应的标签值，通常我们只关心标签值（索引），不关心[0]

4.predict

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
import torch
import torchvision.transforms as transforms
import torchvision
import torch.nn as nn
from PIL import Image
from examples.pytorch.simclr import SimCLR
from examples.pytorch.simclr_classification import ClassificationHead


def main():
    transform = transforms.Compose(
        [transforms.Resize((32, 32)),
         transforms.ToTensor(),
         transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])

    classes = ('plane', 'car', 'bird', 'cat',
               'deer', 'dog', 'frog', 'horse', 'ship', 'truck')

   #实例化simclr模型并加载训练好的权重
    resnet = torchvision.models.resnet18()
    backbone = nn.Sequential(*list(resnet.children())[:-1])
    simclr_model = SimCLR(backbone)
    simclr_model.load_state_dict(torch.load('simclr.pth'))
   
   #将训练好的simclr的backbone取出与全连接层拼成新模型
    backbone = simclr_model.backbone
    class_model = ClassificationHead(backbone, num_classes=10)
   
   #加载已训练好的全连接层权重
    class_model.load_state_dict(torch.load('simclr_classification.pth'))

    class_model.eval()
    im = Image.open('1.jpg')
    im = transform(im)  # [C, H, W]
    im = torch.unsqueeze(im, dim=0)  # [N, C, H, W]

    with torch.no_grad():
        outputs = class_model(im)
        predict = outputs.max(1)[1].numpy()
    print(classes[int(predict)])


if __name__ == '__main__':
    main()

总结

  • 先建立基于resnet18的模型，使用backbone这种形式大概是为了方便在下游任务使用？
  • 对数据集进行随机裁剪、颜色抖动等操作，生成同一张图的不同view(通过dataloader访问数据集时会通过transform生成2张图)
  • 使用NTXentLoss进行对比学习，该损失函数会会比较一个批次中所有图像视图，经过对比学习后的backbone会有更好的特征提取能力
  • 若要用于分类任务，则将对比学习训练后的backbone接上全连接层并对全连接层单独进行监督训练
  • 进行predict时，需先加载训练好的resnet backbone参数，再加载训练好的全连接层参数

发表于 更新于

最近看了几个算法的数据集建立方法，在此记录一下，方便以后借用

对于数量较少且每种图片数量相同的情况

若要使用该方法，图片在文件夹中需按顺序排列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
import glob
from torch.utils.data import Dataset, DataLoader
from PIL import Image
from torchvision import transforms
from sklearn.model_selection import train_test_split


class Mydataset(Dataset):
    def __init__(self, images, labels, transform):
        self.images = images
        self.labels = labels
        self.transform = transform
        dataset = []
        for i in range(len(labels)):	#每个图片都有对应的标签元素，因此标签数量=图片数量
            temp_img = Image.open(images[i])
            temp_img = self.transform(temp_img)  # transforms.Compose对象可直接输入图像
            dataset.append((temp_img, labels[i]))#经过train_test_split的图片标签一一对应
        self.dataset = dataset

    def __getitem__(self, index):
        return self.dataset[index]

    def __len__(self):
        return len(self.labels)

#先看这个
def load_dataset(self):
    data = []
    all_imgs_path = glob.glob(r'dataset\*.png')	#查找文件夹中的所有png文件
    for ip in all_imgs_path:
        data.append(ip)		#此处是保存了每个图像路径的列表
    labels = []
    for i in range(20): #只适用于知道每个类别图片数量的情况，且数量最好相等且不多
        labels.extend([i] * 72) #图片有20种，每种有72张
      #分别生成72个0、1...20，此处i即为标签
    tr_imgs, te_imgs, tr_labs, te_labs = train_test_split(data, labels, train_size=0.9)	#因为data按照文件顺序排列，因此图片与标签可以一一对应，训练集:验证集=9:1
   
    transform = transforms.Compose([
        transforms.Resize((128, 128)),
        transforms.ToTensor()
    ])
      
    tr_set = Mydataset(tr_imgs, tr_labs, transform)
    te_set = Mydataset(te_imgs, te_labs, transform)
    tr_loader = DataLoader(tr_set, batch_size=self.args.train_batch_size, shuffle=True, num_workers=self.args.workers,
                           pin_memory=True)
    te_loader = DataLoader(te_set, batch_size=self.args.train_batch_size, shuffle=True, num_workers=self.args.workers,
                           pin_memory=True)
    return tr_loader, te_loader

对于图片已按照文件夹分好的情况

数据集划分

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
import os
from shutil import copy
import random


def mkfile(file):
    if not os.path.exists(file):
        os.makedirs(file)


# 获取 flower_photos 文件夹下除 .txt 文件以外所有文件夹名（即5种花的类名）
file_path = 'flower_data/flower_photos'
flower_class = [cla for cla in os.listdir(file_path) if ".txt" not in cla]
#os.listdir会返回文件夹下的文件夹名的列表

# 创建 训练集train 文件夹，并由5种类名在其目录下创建5个子目录
mkfile('flower_data/train')
for cla in flower_class:
    mkfile('flower_data/train/' + cla)

# 创建 验证集val 文件夹，并由5种类名在其目录下创建5个子目录
mkfile('flower_data/val')
for cla in flower_class:
    mkfile('flower_data/val/' + cla)

# 划分比例，训练集 : 验证集 = 9 : 1
split_rate = 0.1

# 遍历5种花的全部图像并按比例分成训练集和验证集
for cla in flower_class:
    cla_path = file_path + '/' + cla + '/'  # 某一类别花的子目录
    images = os.listdir(cla_path)  # iamges 列表存储了该目录下所有图像的名称
    num = len(images)
    eval_index = random.sample(images, k=int(num * split_rate))  # 从images列表中随机抽取 k 个图像名称
    for index, image in enumerate(images):  #创建验证集
        # eval_index 中保存验证集val的图像名称
        if image in eval_index:
            image_path = cla_path + image
            new_path = 'flower_data/val/' + cla
            copy(image_path, new_path)  # 将选中的图像复制到新路径

        # 其余的图像保存在训练集train中
        else:
            image_path = cla_path + image
            new_path = 'flower_data/train/' + cla
            copy(image_path, new_path)
        print("\r[{}] processing [{}/{}]".format(cla, index + 1, num), end="")  # processing bar
    print()

print("processing done!")

pytorch加载数据集

1
2
3
4
5
6
7
8
# 获取图像数据集的路径
data_root = os.path.abspath(os.path.join(os.getcwd(), "../.."))  # get data root path 返回上上层目录
image_path = data_root + "/data_set/flower_data/"  			# flower data_set path

# 导入训练集并进行预处理
train_dataset = datasets.ImageFolder(root=image_path + "/train",		
                                     transform=data_transform["train"])

ImageFolder()返回的对象是一个包含数据集所有图像及对应标签构成的二维元组容器，支持索引和迭代，可作为torch.utils.data.DataLoader的输入，其有三种方法：

  • self.classes：用一个列表保存类别名称，例如[‘dog’,’cat’]

  • self.class_to_idx：得到字典：类别 + 对应的索引值，例如{‘dog’:1,’cat’:2}

    1
    2
    3
    flower_list = train_dataset.class_to_idx
    # 将 flower_list 中的 key 和 val 调换位置
    cla_dict = dict((val, key) for key, val in flower_list.items())

  • self.imgs：保存(图像路径, 它所属于的类别index) tuple的 list

同时，imageFolder()过程只会加载index，而不会执行transform操作（transform操作是懒加载的，只有使用dataLoader的时候才会执行）

对同一数据集进行分割

1
2
3
4
5
6
7
# 从训练集的50000个样本中，取49000个作为训练集，剩余1000个作为验证集
NUM_TRAIN = 49000

loader_train = DataLoader(dataset, batch_size=64, sampler=sampler.SubsetRandomSampler(range(NUM_TRAIN)))

loader_val = DataLoader(dataset, batch_size=64, sampler=sampler.SubsetRandomSampler(range(NUM_TRAIN, 50000)))

发表于 更新于

Lenet

model

卷积后的矩阵深度由卷积核组数决定，大小由卷积核尺寸通过公式计算得到

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 使用torch.nn包来构建神经网络.
import torch.nn as nn
import torch.nn.functional as F

class LeNet(nn.Module): 					# 继承于nn.Module这个父类
    def __init__(self):						# 初始化网络结构
        super(LeNet, self).__init__()    	# 多继承需用到super函数
        self.conv1 = nn.Conv2d(3, 16, 5)	
        #2维卷积：输入channel（矩阵深度），输出channel（卷积核组数），卷积核尺寸
        self.pool1 = nn.MaxPool2d(2, 2)
        #最大池化：池化核大小，布距
        self.conv2 = nn.Conv2d(16, 32, 5)
        self.pool2 = nn.MaxPool2d(2, 2)
        self.fc1 = nn.Linear(32*5*5, 120)	#全连接层，输入需将矩阵转为1维
        self.fc2 = nn.Linear(120, 84)
        self.fc3 = nn.Linear(84, 10)	#最后一层根据需求更改输出个数

    def forward(self, x):			 # 正向传播过程，pytorch 中tensor的定义：[batch，channel，height，weight]
        x = F.relu(self.conv1(x)) # input(3, 32, 32) output(16, 28, 28)
        x = self.pool1(x)            # output(16, 14, 14) 池化层只改变H，W
        x = F.relu(self.conv2(x))    # output(32, 10, 10)
        x = self.pool2(x)            # output(32, 5, 5)
        x = x.view(-1, 32*5*5)       # output(32*5*5)，tensor展平
        x = F.relu(self.fc1(x))      # output(120)
        x = F.relu(self.fc2(x))      # output(84)
        x = self.fc3(x)              # output(10)
        return x

train

数据预处理

1
2
3
4
5
6
7
8
9
10
11
transform = transforms.Compose(
   [transforms.ToTensor(), #将图片转化为torch的tensor格式
    transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))#归一化
   ])

'''
还可以添加
transforms.RandomResizedCrop(size)      # 随机裁剪，再缩放成 size*size
transforms.RandomHorizontalFlip(p=0.5) 	#概率为0.5的随机水平翻转
'''

下载数据集并通过dataloader加载

对于打包的训练集需要通过dataloader加载并设置batchsize等信息

训练集

1
2
3
train_set = torchvision.datasets.CIFAR10(root='./data', train=True,download=False, transform=transform)

train_loader = torch.utils.data.DataLoader(train_set, batch_size=36,shuffle=True, num_workers=0)#shuffle:打乱

测试集

1
2
3
val_set = torchvision.datasets.CIFAR10(root='./data', train=False,download=False, transform=transform)
#测试集train为false
val_loader = torch.utils.data.DataLoader(val_set, batch_size=5000,shuffle=True, num_workers=0)

关于dataloader

如果对dataloader对象取迭代器，则其返回为值与标签

1
2
3
#创建迭代器方便访问（dataloader返回值为数据和标签）
val_data_iter = iter(val_loader)
val_image, val_label = next(val_data_iter)

若对其：

1
for step, data in enumerate(train_loader, start=0):

则enumerate函数会返回当前的索引步数和data，其中data组成为[value，label]

加载lenet模型、损失函数、优化器

1
2
3
net = LeNet().to(device)#若使用gpu，则所有张量（模型、经过transform处理的对象）都要转到gpu上
loss_function = nn.CrossEntropyLoss()#交叉熵损失函数
optimizer = optim.Adam(net.parameters(), lr=0.001)#adam优化器

开始训练

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
for epoch in range(7): 

  running_loss = 0.0
  for step, data in enumerate(train_loader, start=0):
      #enumerate将dataloader转化为索引值+数据的迭代器结构，其中数据data包括图像值和标签
      inputs, labels = data

      # zero the parameter gradients
      optimizer.zero_grad()
      
      # forward + backward + optimize
      outputs = net(inputs.to(device))	#正向传播
      loss = loss_function(outputs, labels.to(device))	
      loss.backward()	#根据损失函数反向传播
      optimizer.step()	#优化器更新

      # print statistics
      running_loss += loss.item()
      if step % 500 == 499:    # print every 500 mini-batches
         with torch.no_grad():	#在with块中临时关闭梯度计算，因为此处为验证而非训练
             outputs = net(val_image.to(device))  #将验证集放入训练好的模型中
             predict_y = torch.max(outputs, dim=1)[1]	#选取输出中概率最大的类别作为预测结果
             accuracy = torch.eq(predict_y, val_label.to(device)).sum().item() / 					 val_label.size(0)
             '''
             torch.eq(predict_y, val_label.to(device)) 比较预测结果 predict_y 和实际标签 				val_label 是否相等，生成一个布尔张量，值为 True 表示预测正确，False 表示预测错误。
				sum() 计算预测正确的样本数，.item() 将结果从张量转换为Python数值类型。最后，用正确预					测的数量除以验证集的总样本数 val_label.size(0)，得到准确率。
             '''
             print('[%d, %5d] train_loss: %.3f  test_accuracy: %.3f' %(epoch + 1, 					 step + 1, running_loss / 500, accuracy))
             running_loss = 0.0

predict

需注意单张图片的处理格式，在train中，由于训练集（验证集）都经过了dataloader的batchsize的设置，最终输入到模型中的格式为[N, C, H, W]，预测中由于是单张图片，因此需要手动通过torch.unsqueeze设置为[1, C, H, W]

1
2
3
4
5
6
7
8
9
transform = transforms.Compose(
        	[transforms.Resize((32, 32)),
         transforms.ToTensor(),
         transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))]
im = Image.open('1.jpg')
im = transform(im)  # [C, H, W]
im = torch.unsqueeze(im, dim=0)  # [N, C, H, W]
with torch.no_grad():
   	outputs = net(im)

model的一些改进

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class LeNet(nn.Module):
    def __init__(self):
        super(LeNet, self).__init__()
			#在更为复杂的模型中常使用nn.Sequential集成CNN的各部分
        self.feature = nn.Sequential(
            nn.Conv2d(3, 16, 5),
            nn.ReLU(inplace=True),	#该方法可节约内存
            nn.MaxPool2d(2, 2),
            nn.Conv2d(16, 32, 5),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(2, 2)
        )
      	#全连接层
        self.classifiter = nn.Sequential(
            nn.Linear(32 * 5 * 5, 120),
            nn.ReLU(inplace=True),
            nn.Linear(120, 84),
            nn.ReLU(inplace=True),
            nn.Linear(84, 10)
        )

    def forward(self, x):
        x = self.feature(x)
        x = x.view(-1, 32*5*5)	#输入全连接层时记得展平为1维
        x = self.classifiter(x)
        return x

Alexnet

由图可得，Alexnet共有三次池化，且2-3之间堆叠了3次3x3的卷积

Resnet

传统CNN缺点

实验证明，当网络堆叠到一定深度时，会出现两个问题：

  • 梯度消失或梯度爆炸
  • 退化问题(degradation problem)：在解决了梯度消失、爆炸问题后，仍然存在深层网络的效果可能比浅层网络差的现象

而Resnet通过以下方法解决了这些问题：

  • 对于梯度消失或梯度爆炸问题，ResNet论文提出通过数据的预处理以及在网络中使用 BN（Batch Normalization）层来解决。

  • 对于退化问题，ResNet论文提出了 residual结构（残差结构）来减轻退化问题，并且随着网络的不断加深，效果并没有变差，而是变的更好了。

残差结构

2个适用于不同深度神经网络的残差结构。

人为地让神经网络某些层跳过下一层神经元的连接，隔层相连，弱化每层之间的强联系。这种神经网络被称为 残差网络 (ResNets)。

这里要注意最下方是求和后再经过激活函数，其中1x1的卷积核通过调整步数从而达到降维效果

resnet所有网络的具体信息

需注意的是18、32的输出深度是512，而50以上的输出深度是2048

resnet18结构

由于特征矩阵相加时深度必须相同，因此虚线部分需通过1x1矩阵降维

pytorch搭建残差网络

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
import torch.nn as nn
import torch


# ResNet18/34的残差结构，用的是2个3x3的卷积
class BasicBlock(nn.Module):
    expansion = 1  # 残差结构中，主分支的卷积核个数是否发生变化，不变则为1

    def __init__(self, in_channel, out_channel, stride=1, downsample=None):  				#downsample对应虚线残差结构
        super(BasicBlock, self).__init__()
    #主线路    
        self.conv1 = nn.Conv2d(in_channels=in_channel, out_channels=out_channel,
                             kernel_size=3, stride=stride, padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(out_channel)
        self.relu = nn.ReLU()
        self.conv2 = nn.Conv2d(in_channels=out_channel, out_channels=out_channel,
                               kernel_size=3, stride=1, padding=1, bias=False)
        self.bn2 = nn.BatchNorm2d(out_channel)
      #下采样，表示虚线部分的降维，不一定每个残差结构都有   
        self.downsample = downsample

    def forward(self, x):
        identity = x
        if self.downsample is not None:  # 虚线残差结构，需要下采样
            identity = self.downsample(x)  # 捷径分支 short cut

        out = self.conv1(x)
        out = self.bn1(out)
        out = self.relu(out)

        out = self.conv2(out)
        out = self.bn2(out)

        out += identity 	#相加后再relu
        out = self.relu(out)

        return out

# ResNet50/101/152的残差结构，用的是1x1+3x3+1x1的卷积
class Bottleneck(nn.Module):
    expansion = 4  # 残差结构中第三层卷积核个数是第一/二层卷积核个数的4倍

    def __init__(self, in_channel, out_channel, stride=1, downsample=None):
        super(Bottleneck, self).__init__()
        self.conv1 = nn.Conv2d(in_channels=in_channel, out_channels=out_channel,
                               kernel_size=1, stride=1, bias=False)  # squeeze channels
        self.bn1 = nn.BatchNorm2d(out_channel)
        # -----------------------------------------
        self.conv2 = nn.Conv2d(in_channels=out_channel, out_channels=out_channel,
                               kernel_size=3, stride=stride, bias=False, padding=1)
        self.bn2 = nn.BatchNorm2d(out_channel)
        # -----------------------------------------
        self.conv3 = nn.Conv2d(in_channels=out_channel, out_channels=out_channel * self.expansion,
                               kernel_size=1, stride=1, bias=False)  # unsqueeze channels
        self.bn3 = nn.BatchNorm2d(out_channel * self.expansion)
        self.relu = nn.ReLU(inplace=True)
        self.downsample = downsample

    def forward(self, x):
        identity = x
        if self.downsample is not None:
            identity = self.downsample(x)  # 捷径分支 short cut

        out = self.conv1(x)
        out = self.bn1(out)
        out = self.relu(out)

        out = self.conv2(out)
        out = self.bn2(out)
        out = self.relu(out)

        out = self.conv3(out)
        out = self.bn3(out)

        out += identity
        out = self.relu(out)

        return out


class ResNet(nn.Module):
    # block = BasicBlock or Bottleneck
    # block_num为残差结构中conv2_x~conv5_x中残差块个数，是一个列表，共4层，每层有若干卷积层
    def __init__(self, block, blocks_num, num_classes=1000, include_top=True):
        super(ResNet, self).__init__()
        self.include_top = include_top
        self.in_channel = 64

        self.conv1 = nn.Conv2d(3, self.in_channel, kernel_size=7, stride=2,
                               padding=3, bias=False)
        self.bn1 = nn.BatchNorm2d(self.in_channel)
        self.relu = nn.ReLU(inplace=True)
        self.maxpool = nn.MaxPool2d(kernel_size=3, stride=2, padding=1)
        self.layer1 = self._make_layer(block, 64, blocks_num[0])             
        self.layer2 = self._make_layer(block, 128, blocks_num[1], stride=2) 
        self.layer3 = self._make_layer(block, 256, blocks_num[2], stride=2)  
        self.layer4 = self._make_layer(block, 512, blocks_num[3], stride=2) 
        if self.include_top:
            self.avgpool = nn.AdaptiveAvgPool2d((1, 1))  # output size = (1, 1)
            self.fc = nn.Linear(512 * block.expansion, num_classes)

        for m in self.modules():
            if isinstance(m, nn.Conv2d):
                nn.init.kaiming_normal_(m.weight, mode='fan_out', nonlinearity='relu')

    #重点函数
    def _make_layer(self, block, channel, block_num, stride=1):
        downsample = None

        # ResNet50/101/152的残差结构，block.expansion=4，满足条件则生成下采样降维的旁支
        if stride != 1 or self.in_channel != channel * block.expansion:
            downsample = nn.Sequential(
                nn.Conv2d(self.in_channel, channel * block.expansion, kernel_size=1, 					  stride=stride, bias=False),
                nn.BatchNorm2d(channel * block.expansion))

        layers = []
        layers.append(block(self.in_channel, channel, downsample=downsample, stride=stride))
        self.in_channel = channel * block.expansion

        for _ in range(1, block_num):
            layers.append(block(self.in_channel, channel))

        return nn.Sequential(*layers)
   #对列表取*相对于将其元素依次顺序传入，此时nn.Sequential则会顺序拼接各个残差块的内容

    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)

        if self.include_top:
            x = self.avgpool(x)
            x = torch.flatten(x, 1)
            x = self.fc(x)

        return x


def resnet34(num_classes=1000, include_top=True):
    return ResNet(BasicBlock, [3, 4, 6, 3], num_classes=num_classes, include_top=include_top)


def resnet101(num_classes=1000, include_top=True):
    return ResNet(Bottleneck, [3, 4, 23, 3], num_classes=num_classes, include_top=include_top)


值得注意的是，对于nn.Sequential，其可以通过nn.Sequential(class1,class2)的方式顺序拼接两个神经网络块，也可以通过lenet中的方式依次集成神经网络

由于残差结构（块）的具体参数（卷积层数，是否下采样）需要根据使用的resnet的深度决定，因此使用这种拼接的方式会比较方便