Pytorch深度学习实践

PyTorch documentation — PyTorch 2.4 documentation

PyTorch is an optimized tensor library for deep learning using GPUs and CPUs.

https://pytorch.org/docs/stable/index.html

Basic Knowledge

线性代数

Hadamard product

范数

微积分

方向导数

梯度

1. 当前位置的梯度方向，为函数在该位置处方向导数最大的方向，也是函数值上升最快的方向，反方向为下降最快的方向；

2. 当前位置的梯度长度（模），为最大方向导数的值。

• 对于所有，都有

• 对于所有，都有

• 对于所有，都有

混淆矩阵

Overall Accuracy

Average accuracy

Recall

Precision

F1

• 当时，更偏好召回(Recall)

• 当时，更偏好精准(Precision)

• 当时，平衡精准和召回，即为 F1

• macro 方法：先计算每个 PR，取平均后，再计算 F1

• micro 方法：先计算混淆矩阵元素的平均，再计算 PR 和 F1

PR曲线

置信度

绘图

epoch、batch(mini-batch)、step、iteration

Tensor

x=torch.arange(12)
x.shape #输出形状
X=x.reshape(3,4) #改变形状
X=x.reshape(-1,4) #自动判断行数

torch.zeros((2,3,4))
torch.ones((2,3,4)) #2x3x4

torch.randn(3,4) #符合均值=0，标准差=1的正态分布

x=torch.tensor([1.,2.,4.,8.])
y=torch.tensor([2,4,8,16])
x+y
x-y
x**y
torch.exp(x)
x==y

import torch
a=torch.arange(3).reshape((3,1))
b=torch.arange(2).reshape((1,2))
a+b
# tensor([[0, 1],
#         [1, 2],
#         [2, 3]])

before=id(a)
a=a+b
id(a)==before #False

before=id(a)
a+=b
id(a)==before #True

Overview

线性模型

import numpy as np
import matplotlib.pyplot as plt

# 数据
x_data = [1.0, 2.0, 3.0]
y_data = [2.0, 4.0, 6.0]

# 前向计算函数（预测）
def forward(x, w):
    return x * w

# 损失函数（均方误差）
def loss(x, y, w):
    y_pred = forward(x, w)
    return (y_pred - y) ** 2

# 初始化列表存储不同权重和对应的均方误差
w_list = []
mse_list = []

# 遍历不同的权重值，计算MSE
for w in np.arange(0.0, 4.1, 0.1):
    print('w =', w)
    l_sum = 0
    for x_val, y_val in zip(x_data, y_data):
        y_pred_val = forward(x_val, w)
        loss_val = loss(x_val, y_val, w)
        l_sum += loss_val
        print('\t', x_val, y_val, y_pred_val, loss_val)
    mse = l_sum / 3  # 计算平均损失（MSE）
    print('MSE =', mse)
    w_list.append(w)
    mse_list.append(mse)

# 画出权重和MSE的关系图
plt.plot(w_list, mse_list)
plt.xlabel('w')
plt.ylabel('MSE')
plt.title('w vs MSE')
plt.show()

import numpy as np
import matplotlib.pyplot as plt

# 数据
x_data = np.array([1.0, 2.0, 3.0])
y_data = np.array([2.0, 4.0, 6.0])

# 前向计算函数（预测）
def forward(x, w, b):
    return x * w + b

# 损失函数（均方误差）
def loss(x_data, y_data, w, b):
    # 使用x_data的广播机制同时计算多个x的预测值
    y_pred = forward(x_data, w, b)
    return np.mean((y_pred - y_data) ** 2)  # 返回MSE

# 创建w和b的值范围
w_values = np.arange(0.0, 4.1, 0.1)
b_values = np.arange(-2.0, 2.1, 0.1)

# 创建网格
W, B = np.meshgrid(w_values, b_values)

# 初始化MSE矩阵
MSE = np.zeros(W.shape)  #shape表示矩阵大小(41,41)，shape[0] 表示数组的行数，而 shape[1] 表示数组的列数。

# 计算每一对(w, b)的MSE
for i in range(W.shape[0]):
    for j in range(W.shape[1]):
        w = W[i, j]
        b = B[i, j]
        MSE[i, j] = loss(x_data, y_data, w, b)

# 绘制3D图
fig = plt.figure()
ax = fig.add_subplot(111, projection='3d')

# 绘制网格点的三维图
ax.plot_surface(W, B, MSE, cmap='viridis')

ax.set_xlabel('w')
ax.set_ylabel('b')
ax.set_zlabel('MSE')
plt.title('w, b vs MSE')
plt.show()

梯度下降

批梯度下降 Batch Gradient Descent 

x_data = [1.0, 2.0, 3.0]
y_data = [2.0, 4.0, 6.0]
w = 1.0
def forward(x):
    return x * w
def cost(xs, ys):
    cost = 0
    for x, y in zip(xs, ys):
        y_pred = forward(x)
        cost += (y_pred - y) ** 2
    return cost / len(xs)
def gradient(xs, ys):
    grad = 0
    for x, y in zip(xs, ys):
        grad += 2 * x * (x * w - y)
    return grad / len(xs)
print('Predict (before training)', 4, forward(4))
for epoch in range(100):
    cost_val = cost(x_data, y_data)
    grad_val = gradient(x_data, y_data)
    w -= 0.01 * grad_val
    print('Epoch:', epoch, 'w=', w, 'loss=', cost_val)
print('Predict (after training)', 4, forward(4))

随机梯度下降 Stochastic Gradient Descent

详解随机梯度下降法（Stochastic Gradient Descent，SGD）_随机梯度下降公式-CSDN博客

文章浏览阅读3.9w次，点赞100次，收藏472次。深度学习最常用的优化方法就是随机梯度下降法，但是随机梯度下降法在某些情况下会失效，这是为什么呢？带着这个问题我们接着往下看。一个经典的例子就是假设你现在在山上，为了以最快的速度下山，且视线良好，你可以看清自己的位置以及所处位置的坡度，那么沿着坡向下走，最终你会走到山底。∑i=1n∇θf(θ;xi,yi)+∇θϕ(θ)\sum_{i=1}^{n} \nabla_{\theta} f\left(\theta ; x_{i}, y_{i}\right)+\nabla_{\theta} \phi(\theta_随机梯度下降公式

https://blog.csdn.net/qq_44614524/article/details/114241259

x_data = [1.0, 2.0, 3.0]
y_data = [2.0, 4.0, 6.0]
w = 1.0
def forward(x):
    return x * w
def loss(x, y):
    y_pred = forward(x)
    return (y_pred - y) ** 2
def gradient(x, y):
    return 2 * x * (x * w - y)
print('Predict (before training)', 4, forward(4))
for epoch in range(100):
    for x, y in zip(x_data, y_data):
        grad = gradient(x, y)
        w = w - 0.01 * grad
        print("\tgrad: ", x, y, grad)
        l = loss(x, y)
    print("progress:", epoch, "w=", w, "loss=", l)
print('Predict (after training)', 4, forward(4))

反向传播

1. 前馈（从前向后计算loss）

1. 反馈（从后向前计算）

import torch
x_data = [1.0, 2.0, 3.0]
y_data = [2.0, 4.0, 6.0]
w = torch.Tensor([1.0])
w.requires_grad = True
def forward(x):
    return x * w
def loss(x, y):
    y_pred = forward(x)
    return (y_pred - y) ** 2
print("predict (before training)", 4, forward(4).item())
for epoch in range(100):
    for x, y in zip(x_data, y_data):
        l = loss(x, y)
        l.backward()
        print('\tgrad:', x, y, w.grad.item())
        w.data = w.data - 0.01 * w.grad.data
        w.grad.data.zero_()
    print("progress:", epoch, l.item())
print("predict (after training)", 4, forward(4).item())

import torch
x_data = [1.0, 2.0, 3.0]
y_data = [2.0, 4.0, 6.0]

# 使用torch.tensor()来创建张量
w1 = torch.tensor([1.0], requires_grad=True)
w2 = torch.tensor([1.0], requires_grad=True)
b = torch.tensor([1.0], requires_grad=True)

def forward(x):
    return w1 * x * x + w2 * x + b

def loss(x, y):
    y_pred = forward(x)
    return (y_pred - y) ** 2

# 训练前的预测
print("predict (before training)", 4, forward(4).item())

for epoch in range(100):
    for x, y in zip(x_data, y_data):
        l = loss(x, y)
        l.backward()
        print(f'\tgrad: x={x}, y={y}, w1_grad={w1.grad.item()}, w2_grad={w2.grad.item()}, b_grad={b.grad.item()}')

        # 在torch.no_grad()上下文中更新参数
        with torch.no_grad():
            w1 -= 0.01 * w1.grad
            w2 -= 0.01 * w2.grad
            b -= 0.01 * b.grad

        # 将梯度清零
        w1.grad.zero_()
        w2.grad.zero_()
        b.grad.zero_()

    # 输出每个epoch的进展
    print("progress:", epoch, "loss:", l.item())

# 训练后的预测
print("predict (after training)", 4, forward(4).item())

y.backward()为什么 y 是标量时不需要传入同型向量，而 y 是向量时需要？

1. 标量的情况

x = torch.randn(3, requires_grad=True)
y = x.sum()  # y 是标量
y.backward()  # 直接计算梯度
print(x.grad)  # 输出 x 的梯度

2. 向量的情况

y.backward(v)

线性回归

1. Prepare dataset

x_data = torch.Tensor([[1.0], [2.0], [3.0]])
y_data = torch.Tensor([[2.0], [4.0], [6.0]])

2. Design Model

class LinearModel(torch.nn.Module):
    def __init__(self):
        super(LinearModel,self).__init__()
        self.linear=torch.nn.Linear(1,1)
    
    def forward(self,x):
        y_pred=self.linear(x)
        return y_pred
model=LinearModel()

# 构造函数
def __init__(self, in_features: int, out_features: int, bias: bool = True,device=None, dtype=None) -> None

# 自带的forward
def forward(self, input: Tensor) -> Tensor:
    return F.linear(input, self.weight, self.bias)#相当于执行pass

3. Construct Loss and Optimizer

criterion = torch.nn.MSELoss(reduction='sum')
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

4. Training Cycle

for epoch in range(1000):
	y_pred = model(x_data)
	loss = criterion(y_pred, y_data)
	print(epoch, loss.item())
	optimizer.zero_grad() #清零
	loss.backward()
	optimizer.step() #更新权重

完整代码

import torch.nn

x_data = torch.Tensor([[1.0], [2.0], [3.0]])
y_data = torch.Tensor([[2.0], [4.0], [6.0]])

class LinearModel(torch.nn.Module):
    def __init__(self):
        super(LinearModel,self).__init__()
        self.linear=torch.nn.Linear(1,1)
    
    def forward(self,x):
        y_pred=self.linear(x)
        return y_pred
model=LinearModel()

criterion = torch.nn.MSELoss(reduction='sum')
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)#用Adam的话在100000次训练之后可以算出f(4)=8

for epoch in range(1000):
	y_pred = model(x_data)
	loss = criterion(y_pred, y_data)
	print(epoch, loss.item())
	optimizer.zero_grad() #清零
	loss.backward()
	optimizer.step() #更新权重

print('w = ', model.linear.weight.item())
print('b = ', model.linear.bias.item())
x_test = torch.Tensor([[4.0]])
y_test = model(x_test)
print('y_pred = ', y_test.data)

Logistic Regression

import torch.nn.functional as F
x_data = torch.Tensor([[1.0], [2.0], [3.0]])
y_data = torch.Tensor([[0], [0], [1]])

class LogisticRegressionModel(torch.nn.Module):
    def __init__(self):
        super(LogisticRegressionModel, self).__init__()
        self.linear = torch.nn.Linear(1, 1)
    def forward(self, x):
        y_pred = F.sigmoid(self.linear(x))
        return y_pred
model = LogisticRegressionModel()

criterion = torch.nn.BCELoss(reduction='sum')
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

for epoch in range(1000):
    y_pred = model(x_data)
    loss = criterion(y_pred, y_data)
    print(epoch, loss.item())
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

Multiple Dimension Input

import torch
import numpy as np
xy = np.loadtxt('diabetes.csv.gz', delimiter=',', dtype=np.float32)
x_data = torch.from_numpy(xy[:,:-1])
y_data = torch.from_numpy(xy[:, [-1]])

class Model(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.linear1 = torch.nn.Linear(8, 6)
        self.linear2 = torch.nn.Linear(6, 4)
        self.linear3 = torch.nn.Linear(4, 1)
        self.activate = torch.nn.Sigmoid()
    def forward(self, x):
        x = self.activate(self.linear1(x))
        x = self.activate(self.linear2(x))
        x = self.activate(self.linear3(x))
        return x
model = Model()

criterion = torch.nn.BCELoss(reduction='mean')
optimizer = torch.optim.SGD(model.parameters(), lr=0.1)

for epoch in range(100):
    # Forward
    y_pred = model(x_data)
    loss = criterion(y_pred, y_data)
    print(epoch, loss.item())
    # Backward
    optimizer.zero_grad()
    loss.backward()
    # Update
    optimizer.step()

Dataset and DataLoader

class DiabetesDataset(Dataset):
    def __init__(self, filepath):
        xy = np.loadtxt(filepath, delimiter=',', dtype=np.float32)
        self.len = xy.shape[0]
        self.x_data = torch.from_numpy(xy[:, :-1])
        self.y_data = torch.from_numpy(xy[:, [-1]])
    def __getitem__(self, index): #The expression, dataset[index],will call this magic function.
        return self.x_data[index], self.y_data[index]
    def __len__(self):
        return self.len
dataset = DiabetesDataset('diabetes.csv.gz')
train_loader = DataLoader(dataset=dataset, batch_size=32, shuffle=True, num_workers=2) #num_worker规定读取数据的进程数

import torch
from torch.utils.data import Dataset, DataLoader
import numpy as np

class DiabetesDataset(Dataset):
    def __init__(self, filepath):
        xy = np.loadtxt(filepath, delimiter=',', dtype=np.float32)
        self.len = xy.shape[0]
        self.x_data = torch.from_numpy(xy[:, :-1])
        self.y_data = torch.from_numpy(xy[:, [-1]])
    def __getitem__(self, index):
        return self.x_data[index], self.y_data[index]
    def __len__(self):
        return self.len

class Model(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.linear1 = torch.nn.Linear(8, 6)
        self.linear2 = torch.nn.Linear(6, 4)
        self.linear3 = torch.nn.Linear(4, 1)
        self.sigmoid = torch.nn.Sigmoid()
    def forward(self, x):
        x = self.sigmoid(self.linear1(x))
        x = self.sigmoid(self.linear2(x))
        x = self.sigmoid(self.linear3(x))
        return x

if __name__ == "__main__":
    dataset = DiabetesDataset('diabetes.csv.gz')
    train_loader = DataLoader(dataset=dataset, batch_size=32, shuffle=True, num_workers=2)

    model = Model()
    criterion = torch.nn.BCELoss()
    optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

    for epoch in range(100):
        for i, data in enumerate(train_loader, 0):
            # 1. Prepare data
            inputs, labels = data
            # 2. Forward
            y_pred = model(inputs)
            loss = criterion(y_pred, labels)
            print(epoch, i, loss.item())
            # 3. Backward
            optimizer.zero_grad()
            loss.backward()
            # 4. Update
            optimizer.step()

import torch
from torch.utils.data import Dataset, DataLoader
import numpy as np


class DiabetesDataset(Dataset):
    def __init__(self, filepath):
        xy = np.loadtxt(filepath, delimiter=',', dtype=np.float32)
        self.len = xy.shape[0]
        self.x_data = torch.from_numpy(xy[:, :-1])
        self.y_data = torch.from_numpy(xy[:, [-1]])

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

    def __len__(self):
        return self.len


class Model(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.linear1 = torch.nn.Linear(8, 6)
        self.linear2 = torch.nn.Linear(6, 4)
        self.linear3 = torch.nn.Linear(4, 1)
        self.sigmoid = torch.nn.Sigmoid()

    def forward(self, x):
        x = self.sigmoid(self.linear1(x))
        x = self.sigmoid(self.linear2(x))
        x = self.sigmoid(self.linear3(x))
        return x

if torch.backends.mps.is_available():
    device = torch.device("mps")
else:
    device = torch.device("cpu")

if __name__ == "__main__":
    dataset = DiabetesDataset('diabetes.csv.gz')
    train_loader = DataLoader(dataset=dataset, batch_size=32, shuffle=True, num_workers=2)

    model = Model().to(device)
    criterion = torch.nn.BCELoss()
    optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

    for epoch in range(100):
        for i, data in enumerate(train_loader, 0):
            # 1. Prepare data
            inputs, labels = data
            inputs, labels = inputs.to(device), labels.to(device)
            # 2. Forward
            y_pred = model(inputs)
            loss = criterion(y_pred, labels)
            print(epoch, i, loss.item())
            # 3. Backward
            optimizer.zero_grad()
            loss.backward()
            # 4. Update
            optimizer.step()

Softmax Classifier

import torch

input=torch.rand(3,3)
target=torch.tensor([0,2,1])
softmax=torch.nn.Softmax(dim=1)
loss=torch.nn.NLLLoss()
CEL=torch.nn.CrossEntropyLoss()
print(loss(torch.log(softmax(input)),target))
print(CEL(input,target))

import torch
from torchvision import transforms
from torchvision import datasets
from torch.utils.data import DataLoader
import torch.nn.functional as F
import torch.optim as optim

# 检查 MPS 是否可用
device = torch.device("mps" if torch.backends.mps.is_available() else "cpu")

batch_size = 64
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.1307, ), (0.3081, ))
])

train_dataset = datasets.MNIST(root='../dataset/mnist/',
                               train=True,
                               download=True,
                               transform=transform)
train_loader = DataLoader(train_dataset,
                          shuffle=True,
                          batch_size=batch_size)

test_dataset = datasets.MNIST(root='../dataset/mnist/',
                              train=False,
                              download=True,
                              transform=transform)
test_loader = DataLoader(test_dataset,
                         shuffle=False,
                         batch_size=batch_size)

class Net(torch.nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.l1 = torch.nn.Linear(784, 512)
        self.l2 = torch.nn.Linear(512, 256)
        self.l3 = torch.nn.Linear(256, 128)
        self.l4 = torch.nn.Linear(128, 64)
        self.l5 = torch.nn.Linear(64, 10)

    def forward(self, x):
        x = x.view(-1, 784)
        x = F.relu(self.l1(x))
        x = F.relu(self.l2(x))
        x = F.relu(self.l3(x))
        x = F.relu(self.l4(x))
        return self.l5(x)

model = Net().to(device)

criterion = torch.nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.5)

def train(epoch):
    running_loss = 0.0
    for batch_idx, data in enumerate(train_loader, 0):
        inputs, target = data
        # 将输入和标签移到 MPS 设备上
        inputs, target = inputs.to(device), target.to(device)

        optimizer.zero_grad()
        # forward + backward + update
        outputs = model(inputs)
        loss = criterion(outputs, target)
        loss.backward()
        optimizer.step()

        running_loss += loss.item()
        if batch_idx % 300 == 299:
            print('[%d, %5d] loss: %.3f' % (epoch + 1, batch_idx + 1, running_loss / 300))
            running_loss = 0.0

def softtest():
    correct = 0
    total = 0
    with torch.no_grad():
        for data in test_loader:
            images, labels = data
            # 将输入和标签移到 MPS 设备上
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            _, predicted = torch.max(outputs.data, dim=1)
            total += labels.size(0)
            correct += (predicted == labels).sum().item()

    print('Accuracy on test set: %d %%' % (100 * correct / total))

if __name__ == '__main__':

    for epoch in range(10):
        train(epoch)
        softtest()

CNN

Convolution

import torch
in_channels, out_channels= 5, 10
width, height = 100, 100
kernel_size = 3
batch_size = 1
input = torch.randn(batch_size,in_channels,width,height)
conv_layer = torch.nn.Conv2d(in_channels,out_channels,kernel_size=kernel_size)
output = conv_layer(input)
print(input.shape)
print(output.shape)
print(conv_layer.weight.shape)

torch.Size([1, 5, 100, 100])
torch.Size([1, 10, 98, 98])
torch.Size([10, 5, 3, 3])

Padding

Stride

MaxPooling

• Input:  or

• Output:  or 

Example

import torch
from torchvision import transforms
from torchvision import datasets
from torch.utils.data import DataLoader
import torch.nn.functional as F
import torch.optim as optim
import matplotlib.pyplot as plt

# prepare dataset
batch_size = 64
transform = transforms.Compose([transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))])

train_dataset = datasets.MNIST(root='../dataset/mnist/', train=True, download=True, transform=transform)
train_loader = DataLoader(train_dataset, shuffle=True, batch_size=batch_size)
test_dataset = datasets.MNIST(root='../dataset/mnist/', train=False, download=True, transform=transform)
test_loader = DataLoader(test_dataset, shuffle=False, batch_size=batch_size)

# design model using class
class Net(torch.nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = torch.nn.Conv2d(1, 10, kernel_size=5)
        self.conv2 = torch.nn.Conv2d(10, 20, kernel_size=5)
        self.pooling = torch.nn.MaxPool2d(2)
        self.fc = torch.nn.Linear(320, 10)

    def forward(self, x):
        # flatten data from (n,1,28,28) to (n, 784)
        batch_size = x.size(0)
        x = F.relu(self.pooling(self.conv1(x)))
        x = F.relu(self.pooling(self.conv2(x)))
        x = x.view(batch_size, -1)  # -1 此处自动算出的是320
        x = self.fc(x)
        return x

model = Net()
device = torch.device("mps" if torch.mps.is_available() else "cpu")
model.to(device)

# construct loss and optimizer
criterion = torch.nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.5)

# training cycle forward, backward, update
def train(epoch):
    running_loss = 0.0
    for batch_idx, data in enumerate(train_loader, 0):
        inputs, target = data
        inputs, target = inputs.to(device), target.to(device)
        optimizer.zero_grad()

        outputs = model(inputs)
        loss = criterion(outputs, target)
        loss.backward()
        optimizer.step()

        running_loss += loss.item()
        if batch_idx % 300 == 299:
            print('[%d, %5d] loss: %.3f' % (epoch + 1, batch_idx + 1, running_loss / 300))
            running_loss = 0.0


def test():
    correct = 0
    total = 0
    with torch.no_grad():
        for data in test_loader:
            images, labels = data
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            _, predicted = torch.max(outputs.data, dim=1)
            total += labels.size(0)
            correct += (predicted == labels).sum().item()
    print('accuracy on test set: %d %% ' % (100 * correct / total))
    return correct / total

if __name__ == '__main__':
    epoch_list = []
    acc_list = []

    for epoch in range(10):
        train(epoch)
        acc = test()
        epoch_list.append(epoch)
        acc_list.append(acc)

    plt.plot(epoch_list, acc_list)
    plt.ylabel('accuracy')
    plt.xlabel('epoch')
    plt.show()

1x1 Convolution

Inception Module

class InceptionA(nn.Module):
	def __init__(self, in_channels):
		super(InceptionA, self).__init__()
		self.branch1x1 = nn.Conv2d(in_channels, 16, kernel_size=1)
		self.branch5x5_1 = nn.Conv2d(in_channels,16, kernel_size=1)
		self.branch5x5_2 = nn.Conv2d(16, 24, kernel_size=5, padding=2)
		self.branch3x3_1 = nn.Conv2d(in_channels, 16, kernel_size=1)
		self.branch3x3_2 = nn.Conv2d(16, 24, kernel_size=3, padding=1)
		self.branch3x3_3 = nn.Conv2d(24, 24, kernel_size=3, padding=1)
		self.branch_pool = nn.Conv2d(in_channels, 24, kernel_size=1)
	def forward(self, x):
		branch1x1 = self.branch1x1(x)
		branch5x5 = self.branch5x5_1(x)
		branch5x5 = self.branch5x5_2(branch5x5)
		branch3x3 = self.branch3x3_1(x)
		branch3x3 = self.branch3x3_2(branch3x3)
		branch3x3 = self.branch3x3_3(branch3x3)
		branch_pool = F.avg_pool2d(x, kernel_size=3, stride=1, padding=1)
		branch_pool = self.branch_pool(branch_pool)
		outputs = [branch1x1, branch5x5, branch3x3, branch_pool]
		return torch.cat(outputs, dim=1)

class Net(nn.Module):
	def __init__(self):
		super(Net, self).__init__()
		self.conv1 = nn.Conv2d(1, 10, kernel_size=5)
		self.conv2 = nn.Conv2d(88, 20, kernel_size=5)
		self.incep1 = InceptionA(in_channels=10)
		self.incep2 = InceptionA(in_channels=20)
		self.mp = nn.MaxPool2d(2)
		self.fc = nn.Linear(1408, 10)
	def forward(self, x):
		in_size = x.size(0)
		x = F.relu(self.mp(self.conv1(x)))
		x = self.incep1(x)
		x = F.relu(self.mp(self.conv2(x)))
		x = self.incep2(x)
		x = x.view(in_size, -1)
		x = self.fc(x)
		return x

Residual Network

class ResidualBlock(nn.Module):
	def __init__(self, channels):
		super(ResidualBlock, self).__init__()
		self.channels = channels
		self.conv1 = nn.Conv2d(channels, channels,
		kernel_size=3, padding=1)
		self.conv2 = nn.Conv2d(channels, channels,
		kernel_size=3, padding=1)
	def forward(self, x):
		y = F.relu(self.conv1(x))
		y = self.conv2(y)
		return F.relu(x + y) #先加再ReLU

class Net(nn.Module):
	def __init__(self):
		super(Net, self).__init__()
		self.conv1 = nn.Conv2d(1, 16, kernel_size=5)
		self.conv2 = nn.Conv2d(16, 32, kernel_size=5)
		self.mp = nn.MaxPool2d(2)
		self.rblock1 = ResidualBlock(16)
		self.rblock2 = ResidualBlock(32)
		self.fc = nn.Linear(512, 10)
	def forward(self, x):
		in_size = x.size(0)
		x = self.mp(F.relu(self.conv1(x)))
		x = self.rblock1(x)
		x = self.mp(F.relu(self.conv2(x)))
		x = self.rblock2(x)
		x = x.view(in_size, -1)
		x = self.fc(x)
		return x

权重共享

RNN

RNN Cell

• batchSize = 1

• seqLen = 3

• inputSize = 4

• hiddenSize = 2

• So the shape of inputs and outputs of RNNCell:
• input.shape = batchSize, inputSize
• output.shape = batchSize, hiddenSize

• The sequence can be wrapped in one Tensor with shape:
• dataset.shape = seqLen, batchSize, inputSize

import torch
batch_size = 1
seq_len = 3
input_size = 4
hidden_size = 2
cell = torch.nn.RNNCell(input_size=input_size, hidden_size=hidden_size)
# (seq, batch, features)
dataset = torch.randn(seq_len, batch_size, input_size)
hidden = torch.zeros(batch_size, hidden_size)
for idx, input in enumerate(dataset):
    print('=' * 20, idx, '=' * 20)
    print('Input size: ', input.shape)
    hidden = cell(input, hidden)
    print('outputs size: ', hidden.shape)
    print(hidden)

RNN

• batchSize

• seqLen

• inputSize, hiddenSize

• numLayers

• The shape of input and of RNN:
• input.shape = seqLen, batchSize, inputSize
• .shape = numLayers, batchSize, hiddenSize

• The shape of output and of RNN:
• output.shape = seqLen, batchSize, hiddenSize
• .shape = numLayers, batchSize, hiddenSize

import torch
batch_size = 1
seq_len = 3
input_size = 4
hidden_size = 2
num_layers = 1
cell = torch.nn.RNN(input_size=input_size, hidden_size=hidden_size,
num_layers=num_layers)
# (seqLen, batchSize, inputSize)
inputs = torch.randn(seq_len, batch_size, input_size)
hidden = torch.zeros(num_layers, batch_size, hidden_size)
out, hidden = cell(inputs, hidden)
print('Output size:', out.shape)
print('Output:', out)
print('Hidden size: ', hidden.shape)
print('Hidden: ', hidden)

One-Hot Vectors

Example

RNN Cell

import torch

input_size = 4 # 输入维度，每个字母对应张量维度
hidden_size = 4
batch_size = 1

# 准备数据集
idx2char = ['e', 'h', 'l', 'o']
x_data = [1, 0, 2, 3, 3] # hello对应编码
y_data = [3, 1, 2, 3, 2] # ohlol对应编码

one_hot_lookup = [[1, 0, 0, 0],
                  [0, 1, 0, 0],
                  [0, 0, 1, 0],
                  [0, 0, 0, 1]]

x_one_hot = [one_hot_lookup[x] for x in x_data]

inputs = torch.Tensor(x_one_hot).view(-1, batch_size, input_size)
labels = torch.LongTensor(y_data).view(-1, 1)
print(inputs.shape, labels.shape)

# 构建模型
class Model(torch.nn.Module):
    def __init__(self, input_size, hidden_size, batch_size):
        super(Model, self).__init__()
        self.batch_size = batch_size
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.rnncell = torch.nn.RNNCell(input_size=self.input_size, hidden_size=self.hidden_size)

    def forward(self, inputs, hidden):
        hidden = self.rnncell(inputs, hidden)
        return hidden

    def init_hidden(self):
        return torch.zeros(self.batch_size, self.hidden_size)

net = Model(input_size, hidden_size, batch_size)

# 损失函数和优化器
criterion = torch.nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(net.parameters(), lr=0.1)

# 训练
for epoch in range(15):
    loss = 0
    optimizer.zero_grad()
    hidden = net.init_hidden()
    print( 'Predicted string: ', end= '')
    for input, label in zip(inputs, labels):
        hidden = net(input, hidden)
        loss += criterion(hidden, label) #对每个seq的所有batch求loss，这里batch就1个
        _, idx = hidden.max(dim=1)
        print(idx2char[idx.item()], end= '')
    loss.backward()
    optimizer.step()
    print( ', Epoch [%d/15] loss=%.4f' % (epoch+1, loss.item()))

for epoch in range(15):
    loss = 0
    o = []
    optimizer.zero_grad()
    hidden = net.init_hidden()
    print('Predicted string: ', end='')

    for input, label in zip(inputs, labels):
        hidden = net(input, hidden)
        o.append(hidden)
        _, idx = hidden.max(dim=1)
        print(idx2char[idx.item()], end='')

    # Stack the list of hidden states to form a tensor
    o = torch.stack(o, dim=0)  # Now o is a tensor of shape [seqLen, batchSize, hiddenSize]

    # Reshape `o` to match the expected input shape for the loss function
    loss = criterion(o.view(-1, hidden_size), labels.view(-1))

    loss.backward()
    optimizer.step()

    print(', Epoch [%d/15] loss=%.4f' % (epoch + 1, loss.item()))

RNN Module

import torch

input_size = 4
hidden_size = 4
batch_size = 1
seq_len = 5
num_layers = 1 #1层和一个cell是一样的

# 准备数据集
idx2char = ['e', 'h', 'l', 'o']
x_data = [1, 0, 2, 3, 3] # hello对应编码
y_data = [3, 1, 2, 3, 2] # ohlol对应编码

one_hot_lookup = [[1, 0, 0, 0],
                  [0, 1, 0, 0],
                  [0, 0, 1, 0],
                  [0, 0, 0, 1]]
x_one_hot = [one_hot_lookup[x] for x in x_data]

inputs = torch.Tensor(x_one_hot).view(seq_len, batch_size, input_size)
labels = torch.LongTensor(y_data)
print(inputs.shape, labels.shape)

# 构建模型
class Model(torch.nn.Module):
    def __init__(self, input_size, hidden_size, batch_size, num_layers=1):
        super(Model, self).__init__()
        self.num_layers = num_layers
        self.batch_size = batch_size
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.rnn = torch.nn.RNN(input_size=self.input_size, hidden_size=self.hidden_size, num_layers=num_layers)

    def forward(self, inputs):
        hidden = torch.zeros(self.num_layers, self.batch_size, self.hidden_size)
        out, _ = self.rnn(inputs, hidden)    # 注意维度是(seqLen, batch_size, hidden_size)
        return out.view(-1, self.hidden_size) # 为了容易计算交叉熵这里调整维度为(seqLen * batch_size, hidden_size)

net = Model(input_size, hidden_size, batch_size)

# 损失函数和优化器
criterion = torch.nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(net.parameters(), lr=0.1)

# 训练
for epoch in range(15):
    optimizer.zero_grad()
    outputs = net(inputs)
    # print(outputs.shape, labels.shape)
    # 这里的outputs维度是([seqLen * batch_size, hidden]), labels维度是([seqLen*batch_size])，只不过这里batch_size刚好是1
    loss = criterion(outputs, labels) #这里是对所有seq对所有batch求了交叉熵，而默认reduction=mean，因此loss是之前Cell的loss的1/seq_len
    loss.backward()
    optimizer.step()

    _, idx = outputs.max(dim=1)
    idx = idx.data.numpy()
    print('Predicted: ', ''.join([idx2char[x] for x in idx]), end='')
    print(', Epoch [%d/15] loss = %.3f' % (epoch + 1, loss.item()))

Embedding

import torch

# parameters
num_class = 4
input_size = 4
hidden_size = 8
embedding_size = 10
num_layers = 2
batch_size = 1
seq_len = 5

# 准备数据集
idx2char = ['e', 'h', 'l', 'o']
x_data = [[1, 0, 2, 2, 3]]  # (batch, seq_len)
y_data = [[3, 1, 2, 3, 2]]  # (batch, seq_len)

inputs = torch.LongTensor(x_data)   # Input should be LongTensor: (batchSize, seqLen)
labels = torch.LongTensor(y_data).view(-1)   # Target should be LongTensor: (batchSize * seqLen)

# 构建模型
class Model(torch.nn.Module):
    def __init__(self):
        super(Model, self).__init__()
        self.emb = torch.nn.Embedding(input_size, embedding_size) #每个input(字符)映射到一个embedding_size维的向量
        self.rnn = torch.nn.RNN(input_size=embedding_size, hidden_size=hidden_size, num_layers=num_layers, batch_first=True)
        self.fc = torch.nn.Linear(hidden_size, num_class)

    def forward(self, x):
        hidden = torch.zeros(num_layers, x.size(0), hidden_size)
        x = self.emb(x)  # (batch, seqLen, embeddingSize)
        x, _ = self.rnn(x, hidden)  # 输出(𝒃𝒂𝒕𝒄𝒉𝑺𝒊𝒛𝒆, 𝒔𝒆𝒒𝑳𝒆𝒏, hidden_size)
        x = self.fc(x)  # 输出(𝒃𝒂𝒕𝒄𝒉𝑺𝒊𝒛𝒆, 𝒔𝒆𝒒𝑳𝒆𝒏, 𝒏𝒖𝒎𝑪𝒍𝒂𝒔𝒔)
        return x.view(-1, num_class)  # reshape to use Cross Entropy: (𝒃𝒂𝒕𝒄𝒉𝑺𝒊𝒛𝒆×𝒔𝒆𝒒𝑳𝒆𝒏, 𝒏𝒖𝒎𝑪𝒍𝒂𝒔𝒔)

net = Model()

# 损失函数和优化器
criterion = torch.nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(net.parameters(), lr=0.05)

# 训练模型
for epoch in range(15):
    optimizer.zero_grad()
    outputs = net(inputs)
    loss = criterion(outputs, labels)
    loss.backward()
    optimizer.step()

    _, idx = outputs.max(dim=1)
    idx = idx.data.numpy()
    print('Predicted: ', ''.join([idx2char[x] for x in idx]), end='')
    print(', Epoch [%d/15] loss = %.3f' % (epoch + 1, loss.item()))

LSTM

GRU

Bidirectional

Example 根据人名判断国家

数据处理

pack_padded_sequence

完整代码

import csv
import gzip
import time
import math
import torch
from torch.nn.utils.rnn import pack_padded_sequence
from torch.utils.data import Dataset, DataLoader
import matplotlib.pyplot as plt

# 超参数设置
HIDDEN_SIZE = 100
BATCH_SIZE = 256
N_LAYER = 2
N_EPOCHS = 100
N_CHARS = 128
USE_MPS = torch.backends.mps.is_available() # 自动检测MPS,不知道为什么CPU跑的比GPU快

# 定义数据集
class NameDataset(Dataset):
    def __init__(self, is_train_set=True):
        filename = 'names_train.csv.gz' if is_train_set else 'names_test.csv.gz'
        with gzip.open(filename, 'rt') as f:
            reader = csv.reader(f)
            rows = list(reader)
        self.names = [row[0] for row in rows] #提取名字
        self.len = len(self.names)
        self.countries = [row[1] for row in rows] #提取国家
        self.country_list = list(sorted(set(self.countries))) #先去重，再排序
        self.country_dict = self.get_country_dict()
        self.country_num = len(self.country_list)

    def __getitem__(self, index):
        return self.names[index], self.country_dict[self.countries[index]] #返回名字和国家的索引

    def __len__(self):
        return self.len

    def get_country_dict(self):
        country_dict = dict()
        for idx, country_name in enumerate(self.country_list, 0): #可以省略0，0表示从0开始遍历
            country_dict[country_name] = idx
        return country_dict

    def idx2country(self, index):
        return self.country_list[index]

    def get_countries_num(self):
        return self.country_num

# 数据加载
trainset = NameDataset(is_train_set=True)
trainloader = DataLoader(trainset, batch_size=BATCH_SIZE, shuffle=True)
testset = NameDataset(is_train_set=False)
testloader = DataLoader(testset, batch_size=BATCH_SIZE, shuffle=False)
N_COUNTRY = trainset.get_countries_num()

# 定义模型
class RNNClassifier(torch.nn.Module):
    def __init__(self, input_size, hidden_size, output_size, n_layers=1, bidirectional=True):
        super(RNNClassifier, self).__init__()
        self.hidden_size = hidden_size
        self.n_layers = n_layers
        self.n_directions = 2 if bidirectional else 1
        self.embedding = torch.nn.Embedding(input_size, hidden_size) #input_size=N_CHARS，input_size其实是字典的大小
        self.gru = torch.nn.GRU(hidden_size, hidden_size, n_layers, bidirectional=bidirectional)
        self.fc = torch.nn.Linear(hidden_size * self.n_directions, output_size)

    def _init_hidden(self, batch_size):
        hidden = torch.zeros(self.n_layers * self.n_directions, batch_size, self.hidden_size)
        return create_tensor(hidden)

    def forward(self, input, seq_lengths):
        input = input.t()  # 转置 input 形状：B x S -> S x B
        batch_size = input.size(1)
        hidden = self._init_hidden(batch_size)
        embedding = self.embedding(input)
        gru_input = pack_padded_sequence(embedding, seq_lengths) #embedding是SxBxH，且按照seq_length降序排列，seq_lengths是每个序列的长度
        output, hidden = self.gru(gru_input, hidden)
        if self.n_directions == 2:
            hidden_cat = torch.cat([hidden[-1], hidden[-2]], dim=1) #双向RNN的输出是两个方向的输出拼接
        else:
            hidden_cat = hidden[-1]
        fc_output = self.fc(hidden_cat)
        return fc_output

# 将张量放置到 MPS 或 CPU
def create_tensor(tensor):
    if USE_MPS:
        return tensor.to('mps')
    else:
        return tensor

# 计算经过的时间
def time_since(since):
    now = time.time()
    s = now - since
    m = math.floor(s / 60)
    s -= m * 60
    return f'{m}m {s:.0f}s'

# 构建输入张量
def make_tensors(names, countries):
    sequences_and_lengths = [name2list(name) for name in names] #将名字转换为ASCII序列
    name_sequences = [sl[0] for sl in sequences_and_lengths] #提取ASCII序列
    seq_lengths = torch.LongTensor([sl[1] for sl in sequences_and_lengths]) #提取序列长度
    countries = countries.long()    #将国家转换为整数
    seq_tensor = torch.zeros(len(name_sequences), seq_lengths.max()).long() #创建一个全0的张量，大小为len(name_sequences) x seq_lengths.max()
    for idx, (seq, seq_len) in enumerate(zip(name_sequences, seq_lengths), 0):
        seq_tensor[idx, :seq_len] = torch.LongTensor(seq) #填充张量
    seq_lengths, perm_idx = seq_lengths.sort(dim=0, descending=True) #按照序列长度降序排列
    seq_tensor = seq_tensor[perm_idx] #按照perm_idx的顺序排列，tensor或者np.array都可以这样操作，普通的list不行
    countries = countries[perm_idx]
    return create_tensor(seq_tensor), seq_lengths, create_tensor(countries) #保留seq_lengths在CPU上

# 将名字转换为 ASCII 序列
def name2list(name):
    arr = [ord(c) for c in name]
    return arr, len(arr)

# 训练模型
def train_model():
    total_loss = 0
    for i, (names, countries) in enumerate(trainloader, 1):
        inputs, seq_lengths, target = make_tensors(names, countries)
        output = classifier(inputs, seq_lengths)
        loss = criterion(output, target)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
        if i % 10 == 0:
            print(f'[{time_since(start)}] Epoch {epoch} [{i * len(inputs)}/{len(trainset)}] loss={total_loss / (i * len(inputs)):.4f}')
    return total_loss

# 测试模型
def test_model():
    correct = 0
    total = 0
    with torch.no_grad():
        for i, (names, countries) in enumerate(testloader, 1):
            inputs, seq_lengths, target = make_tensors(names, countries)
            output = classifier(inputs, seq_lengths)
            pred = output.max(dim=1, keepdim=True)[1]
            correct += pred.eq(target.view_as(pred)).sum().item()
            total += target.size(0)
    accuracy = correct / total * 100
    return accuracy

# 初始化模型、损失函数、优化器
classifier = RNNClassifier(N_CHARS, HIDDEN_SIZE, N_COUNTRY, N_LAYER)
if USE_MPS:
    classifier.to('mps')

criterion = torch.nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(classifier.parameters(), lr=0.001)

# 开始训练并在每个 epoch 后测试
start = time.time()
accuracy_list=[]
for epoch in range(1, N_EPOCHS + 1):
    print(f"\nEpoch {epoch}/{N_EPOCHS}")
    train_loss = train_model()
    print(f'Epoch {epoch} training loss: {train_loss:.4f}')
    accuracy=test_model()
    accuracy_list.append(accuracy)
    print(f'Epoch {epoch} accuracy: {accuracy:.2f}%')

# Save the Model
torch.save(classifier.state_dict(), 'rnn.pth')

# 作图
plt.plot(range(1,N_EPOCHS+1),accuracy_list)
plt.xlabel('Epoch')
plt.ylabel('Accuracy')
plt.show()

结束

無限進步