pytorch结构化封装：线性与逻辑回归

Posted 2021-05-02 AI量化实验室

尽管pytorch已经足够易用，但是仍然有一个模板化的代码，比如循环迭代训练网络。参考keras的api封装理念，我们可以把常用的pytorch的代码片断，也封装起来。

BaseModel的实现

• 继承自nn.Module

• compile传入优化器，损失函数。注意，我们在调用compile的时候再初始化

• fit函数是sklearn风格的训练接口。 loss = self.loss_fn(y_pred, y_train)，这里必须先传y_pred,后传y_train，损失函数会检查后者没有梯度函数

• 如下三步是训练的核心： 每个迭代，需要把导数清零，否则导数是累计的；然后对loss进行求导loss.backward()，最后往导数的方向迭代一步self.optimizer.step()

• predict函数，在训练完成，使用模型时，需要把model标记为测试模式，即调用.eval方法。

self.optimizer.zero_grad()
            loss.backward()
            self.optimizer.step()

import torch
from torch import nn
from torch import optim

class BaseModel(nn.Module):
    def __init__(self):
        super(BaseModel,self).__init__()

    def predict(self,x_test):
        self.eval()
        y_pred = self(x_test)
        return y_pred

    def save(self,path):
        torch.save(self.state_dict(), path)

    def compile(self,optimizer,loss,metrics=None):
        for p in self.parameters():
            print(p)
        self.optimizer = optimizer(self.parameters(), lr=1e-4)
        self.loss_fn = loss()
        if metrics is not None:
            self.metrics = metrics()

    def fit(self,x_train,y_train):
        # 开始训练
        num_epochs = 1000
        for epoch in range(num_epochs):
            y_pred = self(x_train)
            #这里顺序不能反，前者是预测值out，后者是target
            
            # backward
            self.optimizer.zero_grad()
            loss = self.loss_fn(y_pred, y_train)
            loss.backward()
            self.optimizer.step()

            if (epoch + 1) % 20 == 0:
                print('Epoch[{}/{}], loss: {:.6f}'
                      .format(epoch + 1, num_epochs, loss.data[0]))

线性模型的改造

在这个BaseModel的基础上对前文的线性模型进行改造，代码就非常简洁了。

定义模型的代码不需要变：

# 线性模型
class LinearRegression(BaseModel):
    def __init__(self):
        super(LinearRegression, self).__init__()
        self.linear = nn.Linear(1, 1)  # input and output is 1 dimension

    def forward(self, x):
        out = self.linear(x)
        return out

模型初始化，只需要两行代码就够了：

model = LinearRegression()
model.compile(optimizer=optim.SGD,loss=nn.MSELoss)

准备数据并训练，训练只需要调用一下fit就好了，相当的简洁：

import numpy as np
x_train = np.array([[3.3], [4.4], [5.5], [6.71], [6.93], [4.168],
                    [9.779], [6.182], [7.59], [2.167], [7.042],
                    [10.791], [5.313], [7.997], [3.1]])

y_train = np.array([[1.7], [2.76], [2.09], [3.19], [1.694], [1.573],
                    [3.366], [2.596], [2.53], [1.221], [2.827],
                    [3.465], [1.65], [2.904], [1.3]])

model.fit(torch.Tensor(x_train),torch.Tensor(y_train))

最后是模型的应用，对结果进行绘制：

y_pred = model.predict(torch.Tensor(x_train))

data = y_pred.data

import matplotlib.pyplot as plt
plt.plot(x_train, y_pred.data.numpy(), label='Fitting Line')
plt.show()

model.save('./mymodel.pth')

逻辑回归的实现

logistic回归是一种广义线性回归（generalized linear model），因此与多重线性回归分析有很多相同之处。它们的模型形式基本上相同，都具有 wx+b，其中w和b是待求参数，其区别在于他们的因变量不同，多重线性回归直接将wx+b作为因变量，即y =wx+b，而logistic回归则通过函数L将wx+b对应一个隐状态p，p=L(wx+b),然后根据p 与1-p的大小决定因变量的值。如果L是logistic函数，就是logistic回归，如果L是多项式函数就是多项式回归。

逻辑回归模型，虽然模型名字里有“回归”两字，但它是一个分类算法，我们用mnist的图片集来验证它的分类效果。

模型的实现与线性模型几乎一致，区别在于损失函数上：loss=nn.CrossEntropyLoss交叉熵损失。

输入参数：in_dim是输入N*784,输出分类n_class为10类

# 定义 Logistic Regression 模型
class Logstic_Regression(BaseModel):
    def __init__(self, in_dim, n_class):
        super(Logstic_Regression, self).__init__()
        self.logstic = nn.Linear(in_dim, n_class)

    def forward(self, x):
        out = self.logstic(x)
        return out

mnist数据集准备，torsh.util.data下的dataloader为数据预处理提供了很好的模型，当然我们也可以自己实现和扩展。

from torchvision import datasets,transforms
    from torch.utils.data import DataLoader

    train_dataset = datasets.MNIST(
        root='./data', train=True, transform=transforms.ToTensor(), download=True)

    test_dataset = datasets.MNIST(
        root='./data', train=False, transform=transforms.ToTensor())

    train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
    test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)

为使用dataloader进行模型训练，我们需要新增一个fit_loader的函数。主要是针对一批数据较大的情况，比如minist训练集有6万多张图片，不可能一次在内存里计算。我们会设定batch_size=32，那一次读取32张，作为一批训练的样本。

def fit_dataloader(self,loader):
        num_epochs = 3
        for epoch in range(num_epochs):

            for i,data in enumerate(loader):
                img,y_train = data
                y_pred = self(img.view(img.size()[0],-1))
                #y_train = y_train.view(y_train.size()[0],-1)
                print(y_train.size())

                # backward
                self.optimizer.zero_grad()
                loss = self.loss_fn(y_pred, y_train)
                loss.backward()
                self.optimizer.step()

                if i % 300 == 0:
                    print('Epoch[{}/{}], loss: {:.6f}'
                          .format(epoch + 1, num_epochs, loss.data[0]))

多层感知器MLP的实现

MLP（Multi-Layer Perceptron），即多层感知器，是一种前向结构的人工神经网络，映射一组输入向量到一组输出向量。MLP可以被看做是一个有向图，由多个节点层组成，每一层全连接到下一层。除了输入节点，每个节点都是一个带有非线性激活函数的神经元（或称处理单元）。一种被称为反向传播算法的监督学习方法中间一个隐层，克服了感知器不能对线性不可分数据进行识别的弱点。

模型实现：

• n_input是输入的维度batch_size*n_input

• n_hidden_1是隐层的输入维度

• n_hidden_2是隐层的输出维度

• n_output是输出层的分类数

• 三个层都是全连接层

• 使用SGD随机梯度下降训练网络

• 使用交叉熵损失计算多分类问题

import torch
from torch import nn
from torch import optim

class MLP(BaseModel):
    def __init__(self,n_input,n_hidden_1,n_hidden_2,n_output):
        super(MLP,self).__init__()
        self.layer1 = nn.Linear(n_input,n_hidden_1)
        self.layer2 = nn.Linear(n_hidden_1,n_hidden_2)
        self.layer3 = nn.Linear(n_hidden_2,n_output)

    def forward(self, x):
        out =  self.layer1(x)
        out = self.layer2(out)
        out = self.layer3(out)
        return out

模型调用：

from torchvision import datasets,transforms
from torch.utils.data import DataLoader

train_dataset = datasets.MNIST(
    root='./data', train=True, transform=transforms.ToTensor(), download=True)

test_dataset = datasets.MNIST(
    root='./data', train=False, transform=transforms.ToTensor())

train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)

model = MLP(28*28,300,100,10)
model.compile(optimizer=optim.SGD,loss=nn.CrossEntropyLoss)
model.fit_dataloader(loader)

优化fit_dataloader函数，增加每一个batch_size的losss,计算准确性的统计。

def fit_dataloader(self,loader):
        num_epochs = 10
        for epoch in range(num_epochs):

            batch_loss = 0.0
            batch_acc = 0.0
            step = 300
            for i,data in enumerate(loader):
                img,y_train = data
                y_pred = self(img.view(img.size()[0],-1))

                #计算当前批次的准确率（max:返回每列最大的，第二个返回值是对应的下标)
                _, pred = torch.max(y_pred, 1)
                num_correct = (pred == y_train).sum()
                batch_acc += num_correct.data.item()

                # 反向传播
                self.optimizer.zero_grad()
                loss = self.loss_fn(y_pred, y_train)

                #计算当前批次的损失
                batch_loss += loss.data.item()

                loss.backward()
                self.optimizer.step()

                if (i+1) % step == 0:
                    print('Epoch[{}/{}],batch:{}, avg loss: {:.6f},train acc:{:.6f}'
                          .format(epoch + 1, num_epochs,i+1, batch_loss/step,batch_acc/(step*32)))
                    batch_loss = 0.0
                    batch_acc = 0.0

这样可以清楚看出，训练过程中的损失在逐渐背叛以及准确性在稳步上升。

Epoch[1/10],batch:300, avg loss: 2.286891,train acc:0.163438
Epoch[1/10],batch:600, avg loss: 2.283830,train acc:0.160000
Epoch[1/10],batch:900, avg loss: 2.276441,train acc:0.180312
......
Epoch[10/10],batch:300, avg loss: 1.938921,train acc:0.677188
Epoch[10/10],batch:600, avg loss: 1.928737,train acc:0.679896
Epoch[10/10],batch:900, avg loss: 1.920588,train acc:0.690208
Epoch[10/10],batch:1200, avg loss: 1.912090,train acc:0.691562
Epoch[10/10],batch:1500, avg loss: 1.908464,train acc:0.689583
Epoch[10/10],batch:1800, avg loss: 1.892407,train acc:0.696667

对模型进行评估，扩展BaseModel，新增predict_loader函数：

def predict_dataloader(self,loader):
        self.eval()
        total_loss = 0.0
        acc = 0.0
        for data in test_loader:
            img, y_train = data
            img = img.view(img.size(0), -1)
            y_pred = self(img)

            loss = self.loss_fn(y_pred, y_train)
            total_loss += loss.data.item()

            _, pred = torch.max(y_pred, 1)
            num_correct = (pred == y_train).sum()
            acc += num_correct.data.item()

        print(total_loss/len(loader),acc/32/len(loader))

从结果上看出来，经过10轮的迭代,在测试集上得到总体67.16%的准确率。

1.9051747440149227 0.6716253993610224

https://github.com/ailabx/ailabx

扫描下方二维码，关注：AI量化实验室（ailabx），了解AI量化最前沿技术、资讯。