卷积神经网络

import torch
from torch import nn
import torch.nn.functional as F
# 这个二维卷积层使用4D输入和4D输出格式
# (batch_size, channel, height, width)，其中批量大小和通道数 
# 均为 1
X = torch.ones((6, 8))
X[:, 2:6] = 0
X.unsqueeze_(0).unsqueeze_(0)  # X.shape [1,1,6,8]
K = torch.tensor([[[[1.0, -1.0]]]])  # K.shape [1,1,1,2]
Y = F.conv2d(X, K)  # Y.shape [1,1,6,7]
# 使用Conv2d构建一个具有1个输出通道的一维卷积层
# 卷积核形状为 (1, 2)
conv2d = nn.Conv2d(1, 1, kernel_size=(1, 2), bias=False)
lr = 3e-2  # 学习率
for i in range(10):
    Y_hat = conv2d(X)
    l = (Y_hat - Y) ** 2
    conv2d.zero_grad()
    l.sum().backward()
    # 使用梯度更新权重
    conv2d.weight.data[:] -= lr * conv2d.weight.grad
    if (i + 1) % 2 == 0:
        print(f'epoch {i+1}, loss {l.sum():.3f}')
print("convolution weights =", conv2d.weight.data.reshape((1, 2)))

# CIFAR10 数据集图像分类
import time 
import numpy as np 
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import DataLoader
from torchvision import datasets
from torchvision import transforms
import matplotlib.pyplot as plt
# 配置中文字体支持
plt.rcParams['font.sans-serif'] = [
    'Noto Sans CJK SC', 'Noto Sans CJK JP', 'SimHei',
    'Microsoft YaHei', 'WenQuanYi Micro Hei', 'DejaVu Sans'
]
plt.rcParams['axes.unicode_minus'] = False


# 超参数
RANDOM_SEED = 1
LEARNING_RATE = 0.01
BATCH_SIZE = 128
NUM_EPOCHS = 10
# 网络参数
NUM_FEATURES = 32 * 32
NUM_CLASSES = 10
# 其他
DEVICE = torch.device("cuda:0")
torch.manual_seed(0)

# 加载数据集，transforms.ToTensor() 将数据归一化为 [0.0,1.0]
train_dataset = datasets.CIFAR10(root='../../../dataset/cifar10', train=True, transform=transforms.ToTensor(),download=True)
test_dataset = datasets.CIFAR10(root='../../../dataset/cifar10', train=False, transform=transforms.ToTensor())
train_loader = DataLoader(dataset=train_dataset, batch_size=BATCH_SIZE, 
                          shuffle=True)
test_loader = DataLoader(dataset=test_dataset, batch_size=BATCH_SIZE, 
                         shuffle=False)
for images, labels in train_loader:  # 验证数据
    print('Image batch dimensions:', images.shape)
    print('Image label dimensions:', labels.shape)
    break

class ConvNet(torch.nn.Module):
    def __init__(self, num_classes, n_channels):
        super().__init__()
        self.n_channels = n_channels
        self.features = nn.Sequential(
                            nn.Conv2d(in_channels=n_channels, # 32x32x3 => 28x28x18
                             out_channels=6*n_channels, kernel_size=5,
                             padding=0), # (1(28-1) - 32 + 5) / 2 = 0
                            nn.ReLU(),
                            nn.MaxPool2d(kernel_size=2, # 28x28x18 => 14x14x18
                                         stride=2,
                                         padding=0), # (2(14-1) - 28 + 2) = 0
                            nn.Conv2d(in_channels=6*n_channels, # 14x14x18 => 14x14x36
                            out_channels=12*n_channels, kernel_size=5, stride=(1, 1),
                                      padding=2), # (1(14-1) - 14 + 5) / 2 = 2  
                            nn.ReLU(),
                            nn.MaxPool2d(kernel_size=(2, 2), # 14x14x36 => 7x7x36
                                         stride=(2, 2),
                                         padding=0)) # (2(7-1) - 14 + 2) = 0

        self.classifier = torch.nn.Linear(7 * 7 * 12 * n_channels, num_classes)
        # 可以使用高斯分布初始化权重，但高斯初始化 
        # 权重很少使用
        for m in self.modules():
            if isinstance(m, torch.nn.Conv2d) or isinstance(m, torch.nn.Linear):
                m.weight.data.normal_(0.0, 0.01)
                if m.bias is not None:
                    m.bias.detach().zero_()

    def forward(self, x):
        out = self.features(x)
        logits = self.classifier(out.view(-1, 7*7*12*self.n_channels))
        probas = F.softmax(logits, dim=1)
        return logits, probas

def compute_accuracy(model, data_loader, device):
    """计算模型在给定数据加载器上的准确率。"""
    model.eval()
    correct_pred, num_examples = 0, 0
    with torch.no_grad():
        for features, targets in data_loader:
            features, targets = features.to(device), targets.to(device)
            logits, probas = model(features)
            _, predicted_labels = torch.max(probas, 1)
            num_examples += targets.size(0)
            correct_pred += (predicted_labels == targets).sum()
    return correct_pred.float() / num_examples * 100


def train_model(model, train_loader, test_loader, num_epochs, learning_rate, 
                device, log_interval=50, momentum=0.9):
    """
    训练神经网络模型。
    
    参数:
        model: 要训练的PyTorch模型
        train_loader: 训练数据的数据加载器
        test_loader: 测试数据的数据加载器
        num_epochs: 训练轮数
        learning_rate: 优化器的学习率
        device: 训练设备 (cuda/cpu)
        log_interval: 每 N 个批次打印日志 (默认: 50)
        momentum: SGD 优化器的动量 (默认: 0.9)
    
    返回:
        训练好的模型
    """
    model = model.to(device)
    optimizer = torch.optim.SGD(model.parameters(), lr=learning_rate, momentum=momentum)
    
    print(model)
    total_params = sum(p.numel() for p in model.parameters())
    print(f'Total parameters: {total_params:,}')
    
    start_time = time.time()
    for epoch in range(num_epochs):
        model.train()
        for batch_idx, (features, targets) in enumerate(train_loader):
            features, targets = features.to(device), targets.to(device)
            
            # 前向传播
            logits, probas = model(features)
            loss = F.cross_entropy(logits, targets)
            
            # 反向传播
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            
            if batch_idx % log_interval == 0:
                print(f'Epoch: {epoch+1:03d}/{num_epochs:03d} | '
                      f'Batch {batch_idx:04d}/{len(train_loader):04d} | '
                      f'Loss: {loss:.4f}')

        # 在训练集上评估
        train_acc = compute_accuracy(model, train_loader, device)
        print(f'Epoch: {epoch+1:03d}/{num_epochs:03d} | Train Accuracy: {train_acc:.3f}%')
        print(f'Time elapsed: {(time.time() - start_time)/60:.2f} min')
    
    print(f'Total Training Time: {(time.time() - start_time)/60:.2f} min')
    
    # 最终测试评估
    test_acc = compute_accuracy(model, test_loader, device)
    print(f'Test Accuracy: {test_acc:.2f}%')
    
    return model


def visualize_prediction(model, test_loader, device, class_names=None):
    """
    可视化测试集中的样本预测。
    
    参数:
        model: 训练好的模型
        test_loader: 测试数据的数据加载器
        device: 设备 (cuda/cpu)
        class_names: 类别名称列表 (可选)
    """
    if class_names is None:
        class_names = ['airplane', 'automobile', 'bird', 'cat', 'deer', 
                       'dog', 'frog', 'horse', 'ship', 'truck']
    
    # 从测试集中获取一个批次
    features, targets = next(iter(test_loader))
    
    # 显示图像
    nhwc_img = np.transpose(features[0], axes=(1, 2, 0))
    plt.imshow(nhwc_img)
    plt.title(f'True label: {class_names[targets[0]]}')
    plt.show()
    
    # 进行预测
    model.eval()
    with torch.no_grad():
        logits, probas = model(features.to(device)[0:1])
    
    pred_class = torch.argmax(probas, dim=1).item()
    confidence = probas[0][pred_class].item() * 100
    print(f'Predicted: {class_names[pred_class]} ({confidence:.2f}%)')
    print(f'True label: {class_names[targets[0]]}')


# 在 CIFAR-10 上训练 ConvNet 模型
NUM_CHANNELS = images.shape[1]
torch.manual_seed(RANDOM_SEED)

model = ConvNet(NUM_CLASSES, NUM_CHANNELS)
model = train_model(model, train_loader, test_loader, 
                    num_epochs=NUM_EPOCHS, learning_rate=LEARNING_RATE, device=DEVICE)
visualize_prediction(model, test_loader, DEVICE)

class LeNet5(nn.Module):
    def __init__(self, num_classes, in_channels=1):
        super().__init__()
        
        self.in_channels = in_channels
        self.num_classes = num_classes
        self.features = nn.Sequential(
            nn.Conv2d(in_channels, 6*in_channels, kernel_size=5), nn.ReLU(),
            nn.MaxPool2d(kernel_size=2),
            nn.Conv2d(6*in_channels, 16*in_channels, kernel_size=5), nn.ReLU(),
            nn.MaxPool2d(kernel_size=2))
        self.classifier = nn.Sequential(
            nn.Linear(16*5*5*in_channels, 120*in_channels), nn.ReLU(),
            nn.Linear(120*in_channels, 84*in_channels), nn.ReLU(),
            nn.Linear(84*in_channels, num_classes))
    
    def forward(self, x):
        x = self.features(x)
        x = torch.flatten(x, 1)
        logits = self.classifier(x)
        probas = F.softmax(logits, dim=1)
        return logits, probas

# 在 CIFAR-10 数据集上训练 LeNet5
torch.manual_seed(RANDOM_SEED)

# 创建 LeNet5 模型 (in_channels=3 对应 RGB 图像)
lenet_model = LeNet5(num_classes=NUM_CLASSES, in_channels=3)

# 使用可重用的 train_model 函数训练模型
lenet_model = train_model(
    model=lenet_model,
    train_loader=train_loader,
    test_loader=test_loader,
    num_epochs=NUM_EPOCHS,
    learning_rate=LEARNING_RATE,
    device=DEVICE
)

# 可视化预测
visualize_prediction(lenet_model, test_loader, DEVICE)

class AlexNet(nn.Module):
    """
    为 224x224 ImageNet 图像设计的原始 AlexNet 架构。
    
    注意：这将不能直接用于 CIFAR-10 (32x32) 图像。
    对于较小的图像，请使用 AlexNetCIFAR10。
    """
    def __init__(self, num_classes):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(3, 64, kernel_size=11, stride=4, padding=2),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=3, stride=2),
            
            nn.Conv2d(64, 192, kernel_size=5, padding=2),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=3, stride=2),
            
            nn.Conv2d(192, 384, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),
            
            nn.Conv2d(384, 256, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),

            nn.Conv2d(256, 256, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=3, stride=2),
        )
        self.avgpool = nn.AdaptiveAvgPool2d((6, 6))
        self.classifier = nn.Sequential(
            nn.Dropout(0.5),
            nn.Linear(256 * 6 * 6, 4096),
            nn.ReLU(inplace=True),
            nn.Dropout(0.5),
            nn.Linear(4096, 4096),
            nn.ReLU(inplace=True),
            nn.Linear(4096, num_classes))
    
    def forward(self, x):
        x = self.avgpool(self.features(x))
        x = x.view(x.size(0), 256 * 6 * 6)
        logits = self.classifier(x)
        probas = F.softmax(logits, dim=1)
        return logits, probas


class AlexNetCIFAR10(nn.Module):
    """
    适用于 CIFAR-10 (32x32 图像) 的 AlexNet。
    
    使用较小的核和步长来处理 32x32 输入，同时
    保持与原始 AlexNet 相同的架构精神。
    """
    def __init__(self, num_classes):
        super().__init__()
        self.features = nn.Sequential(
            # 第一层: 32x32x3 -> 32x32x64 -> 16x16x64
            nn.Conv2d(3, 64, kernel_size=3, stride=1, padding=1),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=2, stride=2),
            
            # 第二层: 16x16x64 -> 16x16x192 -> 8x8x192
            nn.Conv2d(64, 192, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=2, stride=2),
            
            # 第三层: 8x8x192 -> 8x8x384
            nn.Conv2d(192, 384, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),
            
            # 第四层: 8x8x384 -> 8x8x256
            nn.Conv2d(384, 256, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),

            # 第五层: 8x8x256 -> 8x8x256 -> 4x4x256
            nn.Conv2d(256, 256, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=2, stride=2),
        )
        self.classifier = nn.Sequential(
            nn.Dropout(0.5),
            nn.Linear(256 * 4 * 4, 4096),
            nn.ReLU(inplace=True),
            nn.Dropout(0.5),
            nn.Linear(4096, 4096),
            nn.ReLU(inplace=True),
            nn.Linear(4096, num_classes))
    
    def forward(self, x):
        x = self.features(x)
        x = x.view(x.size(0), 256 * 4 * 4)
        logits = self.classifier(x)
        probas = F.softmax(logits, dim=1)
        return logits, probas

# 在 CIFAR-10 数据集上训练 AlexNetCIFAR10
# 注意：对于 32x32 图像使用 AlexNetCIFAR10（而不是 AlexNet）
torch.manual_seed(RANDOM_SEED)

# 创建 AlexNetCIFAR10 模型（适用于 32x32 图像）
alexnet_model = AlexNetCIFAR10(num_classes=NUM_CLASSES)

# 使用可重用的 train_model 函数训练模型
alexnet_model = train_model(
    model=alexnet_model,
    train_loader=train_loader,
    test_loader=test_loader,
    num_epochs=NUM_EPOCHS,
    learning_rate=LEARNING_RATE,
    device=DEVICE
)

# 可视化预测
visualize_prediction(alexnet_model, test_loader, DEVICE)

class VGG11(torch.nn.Module):
    def vgg_block(self, num_convs, in_channels, out_channels):
        net = [nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1), nn.ReLU(True)]
        for i in range(num_convs - 1):  # 添加若干卷积层
            net.append(nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1))
            net.append(nn.ReLU(True))
        net.append(nn.MaxPool2d(2, 2))  # 添加池化层
        return nn.Sequential(*net)
    
    def __init__(self, num_classes, n_channels):
        super().__init__()
        num_convs = (1, 1, 2, 2, 2)
        channels = ((n_channels, 64), (64, 128), (128, 256), (256, 512), (512, 512))
        net = []
        for n, c in zip(num_convs, channels):
            in_c, out_c = c[0], c[1]
            net.append(self.vgg_block(n, in_c, out_c))
        self.features = nn.Sequential(*net)
        self.classifier = nn.Sequential(nn.Linear(512, 100), nn.ReLU(True),
                                        nn.Linear(100, num_classes))
        
        for m in self.modules():
            if isinstance(m, torch.nn.Conv2d) or isinstance(m, torch.nn.Linear):
                nn.init.kaiming_uniform_(m.weight, mode='fan_in', nonlinearity='relu')
                if m.bias is not None:
                    m.bias.detach().zero_()
    
    def forward(self, x):
        out = self.features(x)
        logits = self.classifier(out.view(x.shape[0], -1))
        probas = F.softmax(logits, dim=1)
        return logits, probas

class Inception(nn.Module): #定义Inception块
    def __init__(self,in_channels,c1,c2,c3,c4):
        super().__init__()
        self.conv1 = nn.Conv2d(in_channels,c1,kernel_size = 1)

        self.conv2 = nn.Sequential(nn.Conv2d(in_channels,c2[0],kernel_size = 1),nn.ReLU(),
                                   nn.Conv2d(c2[0],c2[1],kernel_size=3,padding=1),nn.ReLU())

        self.conv3 = nn.Sequential(nn.Conv2d(in_channels,c3[0],kernel_size=1),nn.ReLU(),
                                   nn.Conv2d(c3[0],c3[1],kernel_size=5, padding=2),nn.ReLU())

        self.conv4 = nn.Sequential(nn.MaxPool2d(kernel_size=3,stride=1,padding=1),
                                   nn.Conv2d(in_channels,c4,kernel_size = 1),nn.ReLU())
    def forward(self,X):
        return torch.cat((self.conv1(X),
                          self.conv2(X),
                          self.conv3(X),
                          self.conv4(X)), dim = 1)

#GoogleNet网络
class GoogleNet(nn.Module):
    def __init__(self,in_channels,classes):
        super().__init__()
        self.model = nn.Sequential(
                  nn.Conv2d(in_channels,out_channels=64,kernel_size=7,stride=2,padding=3),nn.ReLU(),
                  nn.MaxPool2d(kernel_size=3,stride=2),
                  nn.Conv2d(in_channels=64,out_channels=64,kernel_size=1),nn.ReLU(),
                  nn.Conv2d(in_channels=64,out_channels=192,kernel_size=3,padding=1),nn.ReLU(),
                  nn.MaxPool2d(kernel_size=3,stride=2),
                  Inception(192,c1=64,c2=[96,128],c3=[16,32],c4=32),
                  Inception(256,c1=128,c2=[128,192],c3=[32,96],c4=64),
                  nn.MaxPool2d(kernel_size=3,stride=2,padding=1),
                  Inception(480,c1=192,c2=[96,208],c3=[16,48],c4=64),
                  Inception(512,c1=160,c2=[112,224],c3=[24,64],c4=64),
                  Inception(512,c1=128,c2=[128,256],c3=[24,64],c4=64),
                  Inception(512,c1=112,c2=[144,288],c3=[32,64],c4=64),
                  Inception(528,c1=256,c2=[160,320],c3=[32,128],c4=128),
                  nn.MaxPool2d(kernel_size=3,stride=2,padding=1),
                  Inception(832,c1=256,c2=[160,320],c3=[32,128],c4=128),
                  Inception(832,c1=384,c2=[192,384],c3=[48,128],c4=128),
                  nn.AvgPool2d(kernel_size=7,stride=1),
                  nn.Dropout(p=0.4),
                  nn.Flatten(),
                  nn.Linear(1024,classes),
                  nn.Softmax(dim=1)
        )
    def forward(self,X:torch.tensor):
        return self.model(X)

#定义残差块
class ResBlock(nn.Module):
    def __init__(self, inchannel, outchannel, stride=1):
        super(ResBlock, self).__init__()

        #这定义了残差块中两个连续的卷积层
        self.left = nn.Sequential(
            nn.Conv2d(inchannel, outchannel, kernel_size=3, stride=stride, padding=1, bias=False),
            nn.BatchNorm2d(outchannel),
            nn.ReLU(inplace=True),
            nn.Conv2d(outchannel, outchannel, kernel_size=3, stride=1, padding=1, bias=False),
            nn.BatchNorm2d(outchannel))
        self.shortcut = nn.Sequential() #普通的快捷连接
        if stride != 1 or inchannel != outchannel:
            # 下采样快捷连接，如果形状不同，快捷连接
            # 需要使用1x1卷积，将步长设置为大于
            # 1用于下采样
            self.shortcut = nn.Sequential(
                nn.Conv2d(inchannel, outchannel, kernel_size=1, stride=stride, bias=False),
                nn.BatchNorm2d(outchannel))
    def forward(self, x):
        out = self.left(x)
        #将两个卷积层的输出与处理后的x相加，以实现
        #残差块的基本结构
        out = out + self.shortcut(x)
        out = F.relu(out)
        return out

#定义残差块
class ResNet18(nn.Module):
    """ResNet-18 架构实现"""
    def __init__(self, ResBlock, num_classes, in_channels):
        super().__init__()
        self.inchannel = 64
        # 第一个卷积层
        self.conv1 = nn.Sequential(
            nn.Conv2d(in_channels, 64, kernel_size=3, stride=1, padding=1, bias=False),
            nn.BatchNorm2d(64),
            nn.ReLU()
        )
        # 构建不同通道大小的残差层
        res_block_list = self.make_layer(ResBlock, 64, 2, stride=1)
        res_block_list += self.make_layer(ResBlock, 128, 2, stride=2)
        res_block_list += self.make_layer(ResBlock, 256, 2, stride=2)
        res_block_list += self.make_layer(ResBlock, 512, 2, stride=2)
        self.res_layers = nn.Sequential(*res_block_list)
        # 用于分类的全连接层
        self.fc = nn.Linear(512, num_classes)

    def make_layer(self, block, channels, num_blocks, stride):
        """残差层的工厂函数"""
        strides = [stride] + [1] * (num_blocks - 1)
        layers = []
        for stride in strides:
            layers.append(ResBlock(self.inchannel, channels, stride))
            self.inchannel = channels
        return layers
    
    def forward(self, x):
        # 通过第一个卷积层
        out = self.conv1(x)
        # 通过残差层
        out = self.res_layers(out)
        # 全局平均池化
        out = F.avg_pool2d(out, 4)
        # 为全连接层展平
        out = out.view(out.size(0), -1)
        # 分类
        logits = self.fc(out)
        probas = F.softmax(logits, dim=1)
        return probas