今天我们正式开启大模型技术的学习之旅。

根据我在冲击大厂：手把手带你从0-1精通企业真正需要的大模型技术中规划的学习路线，第一阶段的首要任务是夯实深度学习基础。

S1.1学习方法

对于计算机相关专业的同学，相信大家对李沐老师的深度学习课程都不陌生。他不仅提供了系统化的教学视频，还编写了配套教材。针对零基础学员，我强烈推荐结合视频和教材同步学习。虽然需要投入较多时间，但李沐老师的课程能帮助大家真正掌握深度学习的基础知识和核心概念。

当时看到课程内容繁多，纯文字学习让我难以吸收。尤其对于计算机专业来说，理解概念并不等于能看懂代码。因此，我推荐稍有机器学习或者深度学习基础的同学，可以参考B站"刘二大人"的《PyTorch深度学习实践》课程。这套视频总时长约十多个小时，分为每节1小时左右的课程，一周就能学完。老师采用理论讲解+图示演示+代码实践的教学方式，深入浅出地讲解了线性模型、梯度下降、反向传播、逻辑斯蒂回归、CNN、RNN等核心内容。下面我将基于这套课程，详细解析其中的关键知识点。大家可以边看老师的视频边敲代码，也可以跟着我来~

S1.1《Pytorch深度学习实践》

(1)线性模型

在深度学习中，我们主要分为三个步骤

·dataset-构建高质量的数据集

·模型选择与设计-根据数据特性和任务需求，选择或设计适合的模型架构，需权衡模型复杂度与性能

·模型训练：大部分的模型都需要training，而且越是复杂的模型，推理的时间就会越长。当然有些也可以不做训练，好比knn模型(K近邻)，把数据都存进去，将来若是有新的数据，就可以直接找跟他的特征值最接近的，从而决定他的类别。

·模型推理-当training之后，我们就把模型里面的权重都保留下来，然后我们就可以投入实际应用场景进行inferring。

我们现在有这样一个情景：假设学生要在期末考试里得到y的成绩，那么她要每天花费x个小时。每个学生花费的时间不同，最终的成绩也会不同

前三行的数据是我们进行采样测量的结果，现在我们要推理每周学习4个小时的学生最终会得到怎样的成绩y。那么前三行的数据将会作为Training的部分，？部分则是在训练之后的测试部分或推理部分才会看到。那么machine learing的过程则是把数据集交给算法，经过训练之后再获得新的输入，从而得到相应的prediction。其实在训练的时候，由于我们知道训练集中x对应y的值，所以可以根据x和y之间的差异来对模型进行调整，但测试集在训练过程中是不可以偷看的，我们要用它来判定模型的泛化能力。即便我们知道最终结果，我们也只能拿它来和模型预测的结果来计算相应的指标。

关于本节涉及的数学验证方法，刘二大人的视频讲解得非常详细，建议大家参考学习。下面我们通过代码演示如何用穷举法求解线性函数的最优权重。

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):
    return x*w
 
#定义损失函数
def loss(x, y):
    y_pred = forward(x)
    return (y_pred - y)**2
 
 
# 穷举法
#每一个权重和该权重对应的损失值一一保存到列表
w_list = []
mse_list = []

for w in np.arange(0.0, 4.1, 0.1):#间隔0.1，从0.0-4.0
    print("w=", w)
    l_sum = 0
    for x_val, y_val in zip(x_data, y_data):#将x_data, y_data的值拿出来，拼成x_val, y_val的值
        y_pred_val = forward(x_val)
        loss_val = loss(x_val, y_val)
        l_sum += loss_val
        print('\t', x_val, y_val, y_pred_val, loss_val)
    print('MSE=', l_sum/3)
    w_list.append(w)
    mse_list.append(l_sum/3)#MSE
    
plt.plot(w_list,mse_list)
plt.ylabel('Loss')
plt.xlabel('w')
plt.show()    

有人可能会疑惑，这里为何没有采用深度学习方法。实际上，在进行深度学习训练时，我们需要绘制图表——以epoch为横坐标，观察每轮训练的损失值变化。这样能帮助我们判断模型何时收敛或过拟合，从而及时调整模型参数或超参数。所以，我们通常会用线性模型在深度学习的训练过程中进行一些可视化，时刻监督模型的训练过程。

(2)梯度下降

基本思想

梯度下降是一种用于寻找函数局部最小值的优化算法。其核心思想是通过迭代的方式逐步调整参数，使得目标函数的值不断减小。具体来说，算法通过计算目标函数的梯度（即偏导数）来确定参数的更新方向，并沿着梯度的反方向调整参数。

梯度下降的注意事项

1.学习率的选择：学习率过大可能导致算法不收敛，过小则收敛速度慢。

2.局部最优问题：梯度下降可能陷入局部最小值而非全局最小值。

3.特征缩放：在应用梯度下降前，对特征进行标准 化或归一化可以加快收敛速度。

那么我们回到数学的角度，一条复杂的曲线可能包括多个导数为0的点，他们可能是局部最优点，但我们没办法找到全局最小或最大的那个点，即全局最优点。实际上，在深度神经网络中，他的损失函数里并没有过多的局部最优点，但是有一种点叫做“鞍点”，即梯度G为0,那么这个时候梯度无法进行迭代，因为(为学习率)。

在本节课里还有一些数学理论和公式推导大家一定要详细去听老师的讲解，否则代码部分的数学运算你可能不知是怎么得出的，下面我们直接示例代码。

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]
#初始权重猜测
w = 1.0

def forward(x):
 
    return x * w
 
#计算MSE
def cost(xs, ys):
    """
    :param xs: 特征
    :param 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):
    """
    求导（误差求导）
    :param xs: 特征
    :param ys: 对应的真实值
    """
    grad = 0
    for x, y in zip(xs, ys):
    #由数学公式推导而来的最终损失对w的偏导，并求和
        grad += 2 * x * (x * w - y)
 
    return grad / len(xs)
 
 
print('Predict (before traning)', 4, forward(4))
 
for epoch in range(100):
    """
    遍历100次
    调用cost函数计算损失值即误差值
    调用gradient函数求得求导值
    更新w值
    输出第几次迭代，对应的更新的权值，损失值
    """
    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))

 随机梯度下降(SGD)

这里解释一下梯度下降与随机梯度下降(SGD)的区别：在标准梯度下降中，我们需要计算所有样本损失的平均值来参与权重更新；而SGD则是随机选取单个样本的损失来进行计算。

采用SGD主要有两个优势：首先，训练数据可能存在噪声，随机采样可以避免噪声对整体结果的过度影响；其次，当遇到损失函数的鞍点时，计算全部样本的梯度可能导致优化停滞，而随机采样可能帮助我们突破这些局部停滞点，继续优化模型。

#SGD代码优化
import matplotlib.pyplot as plt
 
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)#不再是导数求和
 
epoch_list = []
loss_list = []
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)
    epoch_list.append(epoch)
    loss_list.append(l)
 
print('predict (after training)', 4, forward(4))
plt.plot(epoch_list,loss_list)
plt.ylabel('loss')
plt.xlabel('epoch')
plt.show() 

思考

在深度学习里面我们会遇到这样一个问题：在使用梯度下降时，对于每一个样本，模型在计算的时候是可以并行计算的，即和是没有依赖关系的。但在SGD中，因为我们每一个样本求梯度之后就进行更新，下一次计算所用的梯度是上一次计算的结果。所以使用SGD计算是不能并行的。

所以在深度学习中，我们采取折中的方式。如果是全部数据都用上，那么梯度下降的效率更高。但是如果把数据切割，使用SGD会拿到更好的最优值，但时间复杂度将会变高。那么这种折中的方法，就是mini-batch，即批量的随机梯度下降。

(3)反向传播

基本思想

反向传播（Backpropagation）是一种用于训练人工神经网络的算法，其核心思想是通过链式法则计算损失函数对网络参数的梯度，从而利用梯度下降法优化参数。反向传播的关键在于将输出层的误差逐层反向传递到网络的每一层。输出层的误差通过计算预测值与真实值的差异得到，随后根据链式法则将误差分摊到每一层的权重和偏置上。

前馈和反向传播的过程大家一定要看懂课程里的推导，接下来我们来看代码。

import torch
 
x_data = [1.0, 2.0, 3.0]
y_data = [2.0, 4.0, 6.0]
 
w = torch.Tensor([1.0])
#w需要计算梯度
w.requires_grad = True
 
"""
因为torch.Tensor函数，所以我们的w是一个张量；
下面我们计算时，返回的也是张量；这样就可以有一其他功能，便可直接函数取出我们想要的值；
item（）是从原来的精度上更加精确；
最外的两个print分别是，模型跑之前，和w值改变之后再跑的预测结果。
"""
 
def forward(x):
 
    return x * w
    #x和w都是Tensor
 
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()#反向传播函数，自动会计算并且w的值也会做相应的更新，并且最终会释放掉数据
        #前馈算损失，反馈算梯度
        print('\tgrad:', x, y, w.grad.item(), w.data)

        w.data = w.data - 0.01 * w.grad.data#取.data是用于计算，而不会构建计算图
        print(w.data)

        w.grad.data.zero_()#清零，即释放，用于下次计算

        print(w.data, w.grad.item())
 
    print('progress:', epoch, l.item())
 
print('predict (after taining)', 4, forward(4).item())

我们再做一个二次模型的例子来验证一下

import numpy as np
import matplotlib.pyplot as plt
import torch

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

w1 = torch.Tensor([1.0])#初始权值
w1.requires_grad = True#计算梯度，默认是不计算的
w2 = torch.Tensor([1.0])
w2.requires_grad = True
b = torch.Tensor([1.0])
b.requires_grad = True

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

def loss(x,y):#构建计算图
    y_pred = forward(x)
    return (y_pred-y) **2

print('Predict (befortraining)',4,forward(4))

for epoch in range(100):
    l = loss(1, 2)#为了在for循环之前定义l,以便之后的输出，无实际意义
    for x,y in zip(x_data,y_data):
        l = loss(x, y)
        l.backward()
        print('\tgrad:',x,y,w1.grad.item(),w2.grad.item(),b.grad.item())
        w1.data = w1.data - 0.01*w1.grad.data #注意这里的grad是一个tensor，所以要取他的data
        w2.data = w2.data - 0.01 * w2.grad.data
        b.data = b.data - 0.01 * b.grad.data
        w1.grad.data.zero_() #释放之前计算的梯度
        w2.grad.data.zero_()
        b.grad.data.zero_()
    print('Epoch:',epoch,l.item())

print('Predict(after training)',4,forward(4).item())

(4)用pytorch实现线性回归

本节我们主要用pytorch的工具来实现一个线性模型 

import torch

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):
    """
    模型从nn.Module继承下来
    """
    def __init__(self):#构造函数
        super(LinearModel, self).__init__()#调用父类的构造，必须有
        self.linear = torch.nn.Linear(1, 1)# w, b，并进行计算
 
    def forward(self, x):
        y_pred = self.linear(x)#可调用的对象 
 
        return y_pred
 
 #module构造出来的对象会自动根据你的计算图来计算backward

model = LinearModel()

criterion = torch.nn.MSELoss(size_average=False)#是否求均值
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()#对参数进行更新
 

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)

(5)处理多维输入

import numpy as np
import matplotlib.pyplot as py
import torch
 
"""
读取文件，以逗号未分割点
取所有行，和第一列到倒数第二列
取所有行，和最后一列
"""
xy = np.loadtxt('diabetes.csv', delimiter=',', dtype=np.float32)
x_data = torch.from_numpy(xy[:, :-1])#拿出所有行，但不包括最后一列
y_data = torch.from_numpy(xy[:, [-1]])#拿出最后一列，[-1]代表是个矩阵
 
 
class Model(torch.nn.Module):
    """
    super继承Module库
    线性变换；
    8维指的是8个特征-6维是6个特征；8个权重-6个权重
    这是线性网络
    输入数据为8维-输出6维
    输入数据为6维-输出4维
    输入数据为4维-输出1维
    调用sigmoid函数，对数据判断
    """
    def __init__(self):
        super(Model, self).__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()#本例是个模块，区别nn.Functional.Sigmod
 
    def forward(self, x):
        x = self.sigmoid(self.linear1(x))#x就是O1
        x = self.sigmoid(self.linear2(x))#x就是O2
        x = self.sigmoid(self.linear3(x))#x就是y_pred
        return x
 
 
model = Model()
criterion = torch.nn.BCELoss(size_average=False)
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)
 
for epoch in range(100):

    y_pred = model(x_data)#将所有数据传入
    loss = criterion(y_pred, y_data)
    print(epoch, loss.item())
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

(6)加载数据集

import numpy as np
import torch
from torch.utils.data import Dataset, DataLoader
#Dataset是一个抽象类，不能实例化，必须由其他子类去继承
 
class DiabetesDataset(Dataset):

    def __init__(self, filepath):
        xy = np.loadtxt(filepath, delimiter=',', dtype=np.float32)
        self.len = xy.shape[0]#去xy的shape[0],即取数据的数量，也即有多少行
        self.x_data = torch.from_numpy(xy[:, :-1])
        self.y_data = torch.from_numpy(xy[:, [-1]])#返回的  x_data  y_data 都是张量
 
    def __getitem__(self, index):#能够支持下标操作，返回索引
        return self.x_data[index], self.y_data[index]
 
    def __len__(self):#返回数据条数
        return self.len
 
dataset = DiabetesDataset('diabetes.csv')
train_loader = DataLoader(dataset=dataset,
                          batch_size=32,
                          shuffle=True,#打乱数据顺序，保证数据具有随机性
                          num_workers=2)#进程数量
 
 
class Model(torch.nn.Module):

    def __init__(self):
        super(Model, self).__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
 
model = Model()
criterion = torch.nn.BCELoss(size_average=True)
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)
 
if __name__ == '__main__':
    for epoch in range(100):
        for i, data in enumerate(train_loader, 0):
            """
            i：第几次循环；
            data：【张量每个数据集，张量每个数据对应的标签】
            张量每个数据集：一个列表，batch个元素；每个元素是由一条数据中的特征所构成的列表。
            张量每个数据对应的标签：一个列表，batch个元素，每个元素是由数据中的标签构成的列表。
            [tensor([[], [], []]), tensor([[], [], []])]
            """
            # print("i", i)
            # print("data", data)
            inputs, labels = data#x和y
            传出列表中的两个元素，都是张量
            inputs:tensor([[], [], []])
            labels:tensor([[], [], []])

            """
            # print("inputs", inputs)
            # print("labels", labels)
       
            """
            y_pred = model(inputs)
            loss = criterion(y_pred, labels)
            print(epoch, i, loss.item())
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

 (7)多分类问题

分类问题本质就是概率分布问题，我们要在神经网络的最后一层增加softmax层，来输出一个分布。

import numpy as np
 
y = np.array([1, 0, 0])
z = np.array([0.2, 0.1, -0.1])
y_pred = np.exp(z) / np.exp(z).sum()
loss = (-y * np.log(y_pred)).sum()
print(loss)
"""
这个是实现：我们给出真实的标签，和我们最后的输出层的数据；然后我们通过后softmax函数进行变换
softmax函数：e的x次方除以e的x次方的和，这个时候我们就可以保证每一个值是可以实现总和等于1的
同时我们就可以计算损失函数 -[ylog(y_^) + (1 - y)log(1 - y_^)]
我们可以直接写成 -ylog(y_^)
"""

import torch
 
y = torch.LongTensor([0])
z = torch.Tensor([[0.2, 0.1, -0.1]])
criterion = torch.nn.CrossEntropyLoss()
loss = criterion(z, y)
print(loss)
 
"""
Torch.nn.CrossEntropyLoss()
交叉验证熵求解
torch直接封装好了，我们直接用，但是我们需要把数据改为张量（tensor）
封装的包括：softmax 函数：（各个数据的：e^x / sum(e^x)）； 然后进行log变换，同时进行 loss函数(-ylogy_^)的求解
把后面的步骤都写出来了
所以我们的数据只需计算到最后一层，不需要进行softmax计算就好，
"""

import torch
 
criterion = torch.nn.CrossEntropyLoss()
Y = torch.LongTensor([2, 0, 1])
 
Y_pred1 = torch.Tensor([[0.1, 0.2, 0.9],
                        [1.1, 0.1, 0.2],
                        [0.2, 2.1, 0.1]])
Y_pred2 = torch.Tensor([[0.8, 0.2, 0.3],
                        [0.2, 0.3, 0.5],
                        [0.2, 0.2, 0.5]])
 
l1 = criterion(Y_pred1, Y)
l2 = criterion(Y_pred2, Y)
print("Batch Loss1 = ", l1.data, "\nBatch Loss2 = ", l2.data)
 
"""
两组数据我们通过【2， 0， 1】进行比较预测两组数据那个损失函数比较低
"""

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
 
 
"""
一次性组合64个数据为一个数据
对数据进行下载；
shuffle（数据是否打乱）
对数据进行打乱---在进行batch分割---train_dataset--train_loader
对数据直接进行分割 --- test_dataset--test_loader
"""
batch_size = 64
transform = transforms.Compose([transforms.ToTensor(),
                                transforms.Normalize((0.1307, ), (0.3081, ))
                                ])
 #totensor把拿到的图像转变成C*W*H的一个张量
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):
    """
    继承Module类，重写__init__和forward方法‘
    继承Net类
    书写神经网络层:(线性层)
    forward函数书写每一层的连接状态，进行的什么变化
    """
    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
判断使用gpu还是cpu
模型传到gpu上（.to(device)）
调用损失函数--criterion
选用优化器SGD 模型的数值初始化（权值，偏置，梯度函数等等），学习率设置为0.01， 
一般，神经网络在更新权值时，采用如下公式:
                                                   w = w - learning_rate * dw
    引入momentum后，采用如下公式：
                                                   v = mu * v - learning_rate * dw
                                                   w = w + v
    其中，v初始化为0，mu是设定的一个超变量，最常见的设定值是0.9。可以这样理解上式：如果上次的momentum()与这次的
"""
model = Net()
# model = model.cuda()
device = torch.device("cuda:0"if torch.cuda.is_available() else"cpu")
model.to(device)
 
criterion = torch.nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.5)
 
 
def train(epoch):
    """
    损失为0.0
    :param epoch: 迭代次数
    :return: 训练模型
    """
    running_loss = 0.0
    for batch_idx, data in enumerate(train_loader, 0):
        """
        取一个batch数据的索引值和相应的数据
        data = 特征数据 + 标签
        将数据上传到gpu上
        优化器将梯度设置为0
        将特征放入模型
        利用criterion函数计算损失函数
        利用.backward函数进行反向传播
        利用优化器进行更新权值和偏置值，更新的效果
        将损失函数的高精度进行叠加，传给running_loss变量
        判断对所有数据进行的第几组batch，如果是300组则输出....
        并且running损失为0
        """
 
        inputs, target = data
        # inputs.cuda()
        # target.cuda()
        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():#测试不需要反向传播
    """
    :return:预测结果
    定义两个变量
    """
    correct = 0
    total = 0
    with torch.no_grad():
        for data in test_loader:
            """
            取出测试数据
            将特征传给-images；标签传给-labels
            将数据放在gpu上
            将特征数据放入模型
            torch.max的返回值有两个，第一个是每一行的最大值是多少，第二个是每一行最大值的下标(索引)是多少。
            # dim = 1 列是第0个维度，行是第1个维度
               ---返回标签---
            如果预测值和标签一样，则加一
            total = 数据量
            # 张量之间的比较运算
            正确数据量除以总数据量  乘以  100 得到百分比
            """
            images, labels = data
            # images.cuda()
            # labels.cuda()
            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)
        test()

(8)卷积神经网络

其实在大模型中，卷积神经网络用的较少，但是我们还是需要了解其相关部件和原理。

卷积的过程通常涉及下采样，通道数不变，减少特征图的数据量，降低运算需求。

注意，卷积的计算是在做数乘，不是矩阵运算

大家一定要看懂卷积的过程，包括运算、特征维度的变化等

import torch
 
in_channels, out_channels = 5, 10
width, height = 100, 100
kernel_size = 3
batch_size = 1
 
"""
通道---可以想想成厚度
输入数据的设置：
batch_size = 一次集合多少数据
输入通道数
数据平面的维度，宽和高
构建卷积层：
利用卷积核进行卷积操作：nn.Conv2d ：
输入通道，输出通道 卷积核的大小，就卷积核的平面层；
"""
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)

import torch
 
input = [3, 4, 6, 5, 7,
         2, 4, 6, 8, 2,
         1, 6, 7, 8, 4,
         9, 7, 4, 6, 2,
         3, 7, 5, 4, 1]
 
"""
（1， 1， 5， 5）--- batch， 通道， width，height
卷积核：输入通道， 输出通道， 卷积核3*3， padding = 1， 不设置偏置
重写卷积核参数
数据训练
"""
input = torch.Tensor(input).view(1, 1, 5, 5)
 
conv_layer = torch.nn.Conv2d(1, 1, kernel_size=3, padding=1, bias=False)
 
kernel = torch.Tensor([1, 2, 3, 4, 5, 6, 7, 8, 9]).view(1, 1, 3, 3)
conv_layer.weight.data = kernel.data
 
output = conv_layer(input)
print(output)

import torch
 
"""
数据转换：batch，channel，width，height
卷积核2*2，MaxPool2d函数
搭建模型
"""
input = [3, 4, 6, 5,
         2, 4, 6, 8,
         1, 6, 7, 8,
         9, 7, 4, 6,
         ]
input = torch.Tensor(input).view(1, 1, 4, 4)
 
maxpooling_layer = torch.nn.MaxPool2d(kernel_size=2)
 
output = maxpooling_layer(input)
print(output)

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
 
"""
batch=64；
数据转换函数--变张量
下载数据，取出训练数据，dataset.MNIST函数
数据打断设置batch,DataLoader
下载数据，取出测试数据，dataset.MNIST函数
数据不打断 设置batch
"""
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):
        """
        搭建连接层
        Conv2d(输入通道，输出通道，卷积核是 几乘几的)
        MaxPool2d（2）   池化层   除以二；x y 维度上
        Linear线性变换
        """
        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)
        获取x的第一个参数，返回给batch，即获取batch的大小
        先通过 卷积核（conv1） 进行卷积变换 - 进行 池化层（pooling）  - 再 正则化（relu）
        通过（conv2）卷积核 进行卷积变换 - 进行 池化层（pooling） - 再 正则化（relu）
        将做变换好的数据，拉长，变成一个 一维的数据。 - view（batch， -1） 一次处理batch个数据
        进行线性变换 - Linear 函数
        :param x: 输入数据
        :return: 输出数据
        """
        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)
        x = self.fc(x)
        return x
 
 
"""
实例化Net类
判断在GPU还是cpu上跑，这里是在gpu上
将模型放在gpu上（to.(device)）
代价函数
优化器，初始化模型参数（权重，偏置），学习率0.01， 为了防止局部最优，在进行迭代时，再加一个函数，这里设置为0.5，和学习率一个等级
"""
model = Net()
device = torch.device("cuda:0"if torch.cuda.is_available() else"cpu")
model.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):
        """
        train_loader 传出 batch的索引，即第几次数据， data当前数据
        将数据的特征和标签传给 inputs，和 target
        将数据传入gpu上
        优化器中的梯度设置为零
        将数据放入模型中
        criterion计算损失值
        .backward - 反向传播
        优化器函数进行更新权重
        将损失值进行累加 传给 running_loss函数
        输出第300次及300的倍数次的测试数据  --  第几次迭代， 当前的第几组数据，即第几个batch，当前损失和
        每300次，损失值做一个归结
        """
        inputs, target = data
        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 / 2000))
            running_loss = 0.0
 
 
def test():
    correct = 0
    total = 0
    with torch.no_grad():
        """
        torch.no_grad()不用计算梯度
        对测试数据进行遍历
        将数据的 特征和标签 分别传给 images和labels
        数据传到gpu上
        带入模型进行预测 得到预测结果 - outputs
        这里是，我们取出概率最大的那个数作为输出
        取出标签的第一列， 进行累加，即计算总的数据集长度
        判断正确的数量，进行累加
        输出正确率
        """
        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))
 
 
if __name__ == '__main__':
    for epoch in range(10):
        train(epoch)
        test()

接下来我们做一些复杂的神经网络。

"""
构建数据处理函数 - 取出数据 - 进行数据处理（训练，测试）
搭建晓得模块的模型
再搭建整个模型 - 传入小模型，共同构建新的模型
实例化模型 - 选择损失函数，优化器
传入训练集对模型进行训练，传入测试集进行测试
写主函数
"""
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 torch.nn as nn
 
"""
设置batch
调用数据转换函数，transform函数
取出训练集，达到训练数据 - train_dataset
将训练集用DataLoader函数进行处理， shuffle数据打乱，batch个数据合成一个数据
测试集同样操作，只是没有打乱数据的操作
"""
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 InceptionA(nn.Module):
    def __init__(self, in_channels):
        """
        Inception类---继承Module类，
        重新写__init__ forward  方法
        搭建
        卷积层：branch1*1 = Conv2d（输入通达数，输出通道数，卷积核1*1）
        卷积层：branch5*5_1 = Conv2d(输入通道数，输出通道数，卷积核1*1)
        卷积层：branch5*5_2 = Conv2d(输入通道数，输出通道数，卷积核5*5,步长padding=2)
        卷积层：branch3*3_1 = Conv2d(输入通道数，输出通道数，卷积核1*1)
        卷积层：branch3*3_2 = Conv2d(输入通道数，输出通道数，卷积核3*3，步长padding=1)
        卷积层：branch3*3_3 = Conv2d(输入通道数，输出通道数，卷积核3*3，步长padding=1)
        卷积层：branch_pool = Conv2d(输入通道数，输出通道数，卷积核3*3，步长padding=1)
        :param 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):
        """
        1、将数据用branch1*1进行处理--branch1*1
        2、将数据用branch5*5_1进行处理，再将处理过的数据放进branch5*5_2卷积层进行训练--branch5*5_2
        3、将数据用branch3x3_1进行处理，再将处理过的数据放进branch3x3_2卷积层进行训练得到新的数据，再讲这些数据放入branch3x3_3层得到__branch3x3
        4、将数据用avg_pool2d进行卷积操作，再将得到的数据放进branch_pool卷积进行训练得到--branch_pool
        将得到的四组数据进行整合--outputs
        # b,c,w,h  c对应的是dim=1，通道数
        :param x:输入数据
        :return:训练后的通道数
        """
        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):
        """
        Net继承Module类
        重写__init__ 和 forward 方法
        搭建
        卷积层:Conv2d(输入通道，输出通道，卷积核大小)--conv1
        卷积层:Conv2d(输入通道，输出通道，卷积核大小)--conv2
        调用Inception方法--incep1
        调用Inception方法--incep2
        池化层 MaxPool2d函数--mp
        搭建线性层Linear（输入数据维度，输出维度）--fc
        """
        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):
        """
        取出x的size的第一位，获取batch的大小，这里可以根据模型实例化后，看传入的参数是什么，这里取得是参数的size 的 第一个数据，这里是一个数据有多少数据组成的
        x - 卷积层conv1 - 进行池化mp - 正则化relu（调用的F中的relu函数）- x
        将得到的x - 放入incep1中得到新的 - x
        x - 卷积层conv2 - 进行池化mp - 正则化relu（调用的F中的relu函数）- x
        将得到的x - 放入incep2中得到新的 - x
        将x用.view函数将二维数据转化成一维数据
        记性线性变换
        :param x:输入数据
        :return:返回结果
        """
        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
 
 
"""
实例化类 - model
判断是在什么上运行gpu or cpu
将模型放在GPU上
调用损失函数--criterion
悬着SGD优化器，模型参数初始化（权重，偏置，.parameters）， 学习率0.01，momentum = 0.5，更好的进行求最小值，减少局部最优
"""
model = Net()
device = torch.device("cuda:0"if torch.cuda.is_available() else"cpu")
model.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
        将数据放入gpu上
        优化器的梯度设置为0
        将数据的特征放入模型，进行训练
        计算损失函数 - loss
        反向传播， - .backward
        优化器函数进行梯度迭代，更新权重和偏置
        损失累加 - running_loss
        每300个数据进行输出一次（第几次迭代，第几个数据，损失是多少）
        并且损失running_loss置零
        """
        inputs, target = data
        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 / 2000))
            running_loss = 0.0
 
 
def test():
    correct = 0
    total = 0
    with torch.no_grad():
        """
        不用计算梯度
        遍历测试集
        将数据的 特征和标签 传给 - images labels
        将数据放在GPU上
        将特征放入模型进行训练 - outputs
        取出结果的最大值， max
        计算标签的长度，即计算测试了多少数量 通过累加 返回给 - total
        计算预测正确的函数进行累加 - correct
        输出正确率
        """
        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))
 
 
if __name__ == '__main__':
    for epoch in range(10):
        train(epoch)
        test()

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 torch.nn as nn
 
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 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)
 
 
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
 
 
model = Net()
device = torch.device("cuda:0"if torch.cuda.is_available() else"cpu")
model.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
        inputs, target = inputs.to(device), target.to(device)
        optimizer.zero_grad()
 
        #forward + bakward + iupdata
        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 / 2000))
            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))
 
 
if __name__ == '__main__':
    for epoch in range(10):
        train(epoch)
        test()

(9)循环神经网络

循环神经网络（Recurrent Neural Network, RNN）是一类专为处理序列数据设计的神经网络。其核心特点是利用循环结构（隐藏层的输出反馈到自身），使网络具备记忆能力，能够捕捉序列中的时间依赖关系。我们需要认真学习循环神经网络，因为它包含的组件和算子有助于理解后续大模型的基础架构。

import torch
 
batch_size = 1
seq_len = 3
input_size = 4
hidden_size = 2
 
"""
单个RNN
即循环神经网络的单个部分
多个部分组成的是RNN
"""
cell = torch.nn.RNNCell(input_size=input_size, hidden_size=hidden_size)
 
#(seq, batch, features)
"""
生成标准正态分布的（均值为零，方差为一，的高斯白噪声）的随机数；
.randn生成（3，1，4）  --  张量
生成为全为0.的数，
.zeros生成（1，2）  --  张量
"""
dataset = torch.randn(seq_len, batch_size, input_size)
hidden = torch.zeros(batch_size, hidden_size)
print(dataset)
 
for idx, input in enumerate(dataset):
    """
    idx：第几次遍历
    input：取出当前次数的数据，数据维度是（1，4）
    cell（输入数据（batch，特征的向量维度），记忆体维度（batch，hidden的维度））
    """
    print('=' * 20, idx, '=' * 20)
    hidden = cell(input, hidden)
    print('outputs size: ', hidden.shape)
    print(hidden)

​

import torch
 
batch_size = 1
seq_len = 3
input_size = 4
hidden_size = 2
num_layers = 1
 
"""
循环神经网络函数RNN
（输入数据（数据的时间维度x的个数，batch，单个数据的维度），记忆体数据（层数（也即第几层的hidden），hidden的维度），层数）
"""
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：（seq_len，batch，hidden数据维度）--最上边的那一行，横着的，也就是输出
hidden：（层数，batch，hidden数据维度）
"""
out, hidden = cell(inputs, hidden)
print('Output size: ', out.shape)
print('Output: ', out)
print('Hidden size: ', hidden.shape)
print('Hidden: ', hidden)

​

我们看一下num_layers的状态图

import torch
 
batch_size = 1
# seq_len = 3
input_size = 4
hidden_size = 4
# num_layers = 1
 
 
"""
创建字典
设置输入数据中，每一个x的索引位置构成一个列表
同理。构建一个列表y
构建独热向量（one-hot）
利用for循环，搭建出每一个x对应的向量
-----
将搭建好的数据通过.view方法
构建张量
inputs（seq_len，batch,输入数据）（5，1，4）
label（seq_len,batch）
"""
idx2char = ['e', 'h', 'l', 'o']
x_data = [1, 0, 2, 2, 3]
y_data = [3, 1, 2, 3, 2]
 
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)
 
 
class Model(torch.nn.Module):
    def __init__(self, input_size, hidden_size, batch_size):
        """
        Model继承Module类
        重写__init__;forward;init_hidden方法
        init（输入数据维度，记忆体维度，batch）
        super超类
        调用torch..nn.RNNCell(输入数据的维度，记忆体的维度)函数
        :param input_size:（batch，输入数据）
        :param hidden_size:（batch，hidden数据）
        :param batch_size:（batch）
        """
        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.rnncell = torch.nn.RNNCell(input_size=self.input_size,
                                        hidden_size=self.hidden_size)
 
    def forward(self, input, hidden):
        """
        方法，直接使用RNNCell函数，进行训练
        :param input: (batch,inputs)
        :param hidden: (batch,hidden)
        :return: (out（batch，hidden）)
        """
        hidden = self.rnncell(input, hidden)
 
        return hidden
 
    def init_hidden(self):
        """
        使用torch中的zeros函数，生成维度为（batch，hidden），数值为（0.）的张量
        :return:
        """
 
        return torch.zeros(self.batch_size, self.hidden_size)
 
 
"""
实例化Model类 -- net
使用CrossEntropyLoss函数计算损失 -- criterion
选择Adam优化器，初始化模型参数（W_hh,W_ih,B_ih,B_hh）,学习率为0.1
"""
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):
    """
    遍历15次
    损失值为更新为0
    梯度值为0
    取出训练集的特征和对应的标签 - 一次取一组，for循环来取
    将特征集填入模型，传出hidden-即输出值
    通过criterion函数计算损失值进行累加
    通过max函数取出，hidden中最大值，返回其索引位置 -- idx
    输出当前预测的结果
    
    反向传播
    优化器更新参数（权值，偏置，梯度）
    每次循环完，输出当前的循环的损失值
    """
    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)
        _, idx = hidden.max(dim=1)
        print(idx2char[idx.item()], end='')
 
    loss.backward()
    optimizer.step()
    print(', Epoch [%d/15] loss = %.4f' % (epoch + 1, loss.item()))

import torch
 
batch_size = 1
seq_len = 5
input_size = 4
hidden_size = 4
num_layers = 1
 
idx2char = ['e', 'h', 'l', 'o']
x_data = [1, 0, 2, 2, 3]
y_data = [3, 1, 2, 3, 2]
 
"""
inputs(seq_len 5,batch 1,input_size 4)
"""
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)
 
 
class Model(torch.nn.Module):
    def __init__(self, input_size, hidden_size, batch_size, num_layers=1):
        """
        继承Module类
        super超类
        重写__init__,forward方法
        调用RNN函数（输入数据维度，记忆层数据，层数）
        :param input_size:
        :param hidden_size:
        :param batch_size:
        :param num_layers:
        """
        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, input):
        """
        创建记忆层的维度（hidden：层数，batch，hidden维度）
        out：最后一层的输出值（即预测值）--(seq_len,batch,input_size)
        _ :最后时刻的hidden的值 -- (num_layers,batch,hidden_size)
        :param input: 输入数据
        :return:seq_len,hidden_size(每个时间的数据都会有个out，我们将每一个out组合起来，
        out中的一个数据和hidden是一样的维度，因为out_1 = g(W_oh * hidden))相当于我们自己定义了out的维度
        """
        hidden = torch.zeros(self.num_layers,
                             self.batch_size,
                             self.hidden_size)
        out, _ = self.rnn(input, hidden)
 
        return out.view(-1, self.hidden_size)
 
 
"""
实例化类--net  参数：输入数据，记忆层数据，batch，层数
调用损失函数
选择Adam优化器，模型初始化（W_ih,W_hh,B_ih,B_hh）学习率为0.05
"""
net = Model(input_size, hidden_size, batch_size, num_layers)
criterion = torch.nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(net.parameters(), lr=0.05)
 
for epoch in range(15):
    """
    优化器函数的梯度设置为0
    将数据放入net模型中
    计算损失函数
    反向传播
    优化器更新权值，偏置，梯度等
    期初每一个数据的最大值的索引，
    一次输出索引值对应的字母
    输出 迭代次数和损失函数
    """
    # loss = 0
    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()))

接下来我们来做一个应用，就是用RNN做一个分类器。这也能让我们初步了解在做NLP时大概是怎样的一个流程

import torch
from torch.utils.data import Dataset, DataLoader
from torch.nn.utils.rnn import pack_padded_sequence
import time
import matplotlib.pyplot as plt
import numpy as np
import gzip
import csv
import math
 
"""
hidden_size=100,batch_size=256,n_layer=2,n_epochs=100,n_chars=128,
记忆体维度100，batch维度为256，层数为2层，迭代次数100次，
"""
HIDDEN_SIZE = 100
BATCH_SIZE = 256
N_LAYER = 2
N_EPOCHS = 100
N_CHARS = 128
USE_GPU = True
 
 
class NameDataset(Dataset):
    """
    NameDataset类继承Dataset类
    重写__init__,__getitem__,__len__方法
    定义getCountryDict,idx2country,getCountriesNum函数
    实例化后，返回的有：
    countries:list,数据中的第二列即所有国家，无排序，无去重
    country_dict:dict:18,数据中所有国家，有排序，有去重，{键；键值}
    country_list:list:18,数据中所有国家，有排序，有去重
    country_num:int:18
    len:int,数据中名字的长度，即所有数据的数量
    names:list,数据所有名字，有排序
    """
    def __init__(self, is_train_set=True):
        """
        这个文件是第一列为名字，第二列为国家
        :param is_train_set: 用于判断文件类型
        使用gzip方法读取文件（解压读取）
            再用csv的方法将提取的文件读取，
            再用list方法将数据转换成列表 -- rows
        name = 取出rows列表中每个元素的第一个值；即文件的第一列 -- 名字
        len = 多少个名字
        counties = 取出rows列表中每个元素的第二个值，即文件的第二列 -- 国家
        set：去重，无序。将contries列表中的国家去重，随机存放  -- 假设返回 - A 列表
        将A列表进行排序，从小到大，按首字母再其次俺第二个字母（可以根据ASCII码）-返回列表 -- country_list
        country_dict = 调用getCountryDict函数  将列表转换为字典
        country_num = 取出country_list的长度，返回国家有多少个
        """
        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.getCountryDict()
        self.country_num = len(self.country_list)
 
    def __getitem__(self, index):
        """
        :param index:
        :return: 返回名字的索引，国家从小到大的索引位置，对应字典中的位置
        """
 
        return self.names[index], self.country_dict[self.countries[index]]
 
    def __len__(self):
        """
        :return: 求长度
        """
 
        return self.len
 
    def getCountryDict(self):
        """
        创建一个空字典 -- country_dict
        for循环，取出当前循环次数，和当前循环的名字
        存放字典，对应索引，
        :return:将列表转换为字典。（键，键值）
        """
        country_dict = dict()
        for idx, country_name in enumerate(self.country_list, 0):
            country_dict[country_name] = idx
 
        return country_dict
 
    def idx2country(self, index):
        """
        返回country_list的值
        """
 
        return self.country_list[index]
 
    def getCountriesNum(self):
        """
        :return: 国家的数量
        """
 
        return self.country_num
 
 
"""
实例化NameDataset类 -- trainset
利用DataLoader函数对数据进行处理，batch_size个数据为一组，shuffle打乱数据 -- trainloader
实例化NameDataset类 -- testset
利用DataLoader函数对数据进行处理，batch_size个数据为一组，shuffle不打乱数据 -- testloader
返回国家的数量（调用NameDataset类）
"""
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.getCountriesNum()
 
"""
^^^^^数据的处理阶段^^^^^
"""
 
def create_tensor(tensor):
    """
    传入张量
    判断USE_GPU:
    将数据传到cuda上
    """
    if USE_GPU:
        device = torch.device("cuda:0")
        tensor = tensor.to(device)
 
    return tensor
 
 
class RNNClassifier(torch.nn.Module):
    """
    RNNClassifier继承Module类
    重写__init__,_init_hidden,forward,方法。
    """
    def __init__(self, input_size, hidden_size, output_size, n_layers=1, bidirectional=True):
        """
        super超类
        :param input_size:输入数据维度
        :param hidden_size:记忆体维度
        :param output_size:输出数据维度
        :param n_layers:层数
        :param bidirectional:判断是双向还是单向传播
        Embedding:嵌入层，先对输入数据做一个预处理，从input_size维度到hidden维度
        GRU模型，hidden_size:,hidden_size:,n_layers:,
        fc:Linear:线性模型:hidden_size * n_directions(输入模型维度),输出模型维度
        """
        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)
        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):
        """
        搭建记忆体的维度（self.n_layers * self.n_directions，batch_size,hidden_size）
        返回：记忆体传到模型GPU上
        """
        hidden = torch.zeros(self.n_layers * self.n_directions,
                             batch_size, self.hidden_size)
 
        return create_tensor(hidden)
 
    def forward(self, input, seq_lengths):
        """
        数据流程
        batch_size = 输入数据的宽度，
        hidden的维度调用_init_hidden方法
        将数据进行embedding层；
        使用pack_padded_sequence函数得到 - gru_input
        将处理好的数据传入gru模型函数中返回 -- output,hidden
        判断是单向还是双向传播
        最后将记忆层进行星星层变换得到输出的结果，并返回
        forward:输入数据，时间长度
        :param input:输入数据
        :param seq_lengths:时间序列
        :return:对记忆体的参数进行fc全连接变换，输出，即out
        """
        #input shape: B x S --> S x B  转置
        input = input.t()
        batch_size = input.size(1)
 
        hidden = self._init_hidden(batch_size)
        embedding = self.embedding(input)
 
        #pack them up
        gru_input = pack_padded_sequence(embedding, seq_lengths)
 
        output, hidden = self.gru(gru_input, hidden)
        if self.n_directions == 2:
            hidden_cat = torch.cat([hidden[-1], hidden[-2]], dim=1)
        else:
            hidden_cat = hidden[-1]
        fc_output = self.fc(hidden_cat)
 
        return fc_output
 
 
def name2list(name):
    """
    :param name: 名字
    :return: 名字，长度
    """
    arr = [ord(c) for c in name]
 
    return arr, len(arr)
 
 
def make_tensors(names, countries):
    """
    for循环，调用name2list函数，传参数给--sequences_and_lengths
    取出名字
    取出长度，变成张量
    创建全是0.的维度为（len(name_sequences),seq_lengths.max()），并变为长整型 -- 行：名字数，列：最长的ascii名字
    遍历，并返回，最后构建seq_tensor[idx,:seq_len]的矩阵，  --- zip() 函数用于将可迭代的对象作为参数，将对象中对应的元素打包成一个个元组，然后返回由这些元组组成的列表。
    排序，依据序列长度降序
    """
    sequences_and_lengths = [name2list(name) for name in names]
    name_sequences = [s1[0] for s1 in sequences_and_lengths]
    seq_lengths = torch.LongTensor([s1[1] for s1 in sequences_and_lengths])
 
    #make rensor of name, BatchSize x SeqLen
    seq_tensor = torch.zeros(len(name_sequences), seq_lengths.max()).long()
    for idx, (seq, seq_len) in enumerate(zip(name_sequences, seq_lengths), 0):
        seq_tensor[idx, :seq_len] = torch.LongTensor(seq)
 
    #sort by length to use pack_padded_sequence
    seq_lengths, perm_idx = seq_lengths.sort(dim=0, descending=True)
    seq_tensor = seq_tensor[perm_idx]
    countries = countries[perm_idx]
 
    return create_tensor(seq_tensor), \
        create_tensor(seq_lengths), \
        create_tensor(countries)
 
 
def time_since(since):
    """
    时间计时模块
    """
    s = time.time() - since
    m = math.floor(s / 60)
    s -= m * 60
 
    return '%dm %ds' % (m, s)
 
 
def trainModel():
    """
    loss初始为0在循环之外；
    将训练集传入，取出数据，生成输入数据，长度，标签
    放入classifier模型训练
    利用criterion函数计算损失值
    梯度归零
    反向传播
    优化器迭代，更新参数（权重，偏置）
    损失值累加
    输出
    :return:所有数据一次的损失
    """
    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}', end='')
            print(f'[{i * len(inputs)} / {len(trainset)}]', end='')
            print(f'loss = {total_loss / (i * len(inputs))}')
 
    return total_loss
 
 
def testModel():
    """
    测试集
    不用计算梯度
    取出数据
    将特征放入，计算预测值
    求预测数据中的最大值，dim=1--每一行
    损失累加
    输出
    :return:错误率
    """
    correct = 0
    total = len(testset)
    print("evaluating trained model ...")
    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()
 
        percent = '%.2f' % (100 * correct / total)
        print(f'Test set: Accuracy {correct} / {total} {percent} %')
 
    return correct / total
 
 
if __name__ == '__main__':
    """
    模型RNNClassifier
    GPU运行
    损失函数criterion
    优化器Adam：初始化模型参数
    记录时间
    输出：一共输出多少次
    创建一个列表
        for循环对N_EPOCHS（1 - n_epochs+1）
        跑训练模型
        跑测试模型 -- acc
        将返回值依次存放在acc_list[]列表中
    画图部分
    x轴为acc_list的长度，间距为1
    y轴为acc_list的数值
    """
    classifier = RNNClassifier(N_CHARS, HIDDEN_SIZE, N_COUNTRY, N_LAYER)
    if USE_GPU:
        device = torch.device("cuda:0")
        classifier.to(device)
 
    criterion = torch.nn.CrossEntropyLoss()
    optimizer = torch.optim.Adam(classifier.parameters(), lr=0.001)
 
    start = time.time()
    print("Traning for %d epochs..." % N_EPOCHS)
    acc_list = []
    for epoch in range(1, N_EPOCHS + 1):
        #Train cycle
        trainModel()
        acc = testModel()
        acc_list.append(acc)
 
epoch = np.arange(1, len(acc_list) + 1, 1)
acc_list = np.array(acc_list)
plt.plot(epoch, acc_list)
plt.xlabel('Epoch')
plt.ylabel('Accuracy')
plt.grid()
plt.show()

S1.1总结

在本节课程中，我们要重点学习RNN的原理与应用，并通过代码实践掌握梯度下降、归一化和正则化等关键技术。大家需要熟练掌握dataset、dataloader、损失函数和优化器的使用。课后建议用PyTorch完成一个完整模型的训练和预测流程，并在GPU环境下运行，以巩固所学知识。

下一节我们直接开始做一个NLP的实战入门，从0到1完成一个文本分类任务，敬请期待~