3-1 低阶API示范

下面的范例使用Pytorch的低阶API实现线性回归模型和DNN二分类模型。
低阶API主要包括张量操作，计算图和自动微分。
线性模型以及简单的DNN二分类模型的实现，很简单，但是是最基础的 构造数据、构建数据迭代器（原生，不用DataLoader）、定义模型、定义评估指标、定义损失函数、训练等部分

一、线性回归模型

准备数据

import numpy as np 
import pandas as pd
from matplotlib import pyplot as plt 
import torch
from torch import nn


#样本数量
n = 400
# 生成测试用数据集
X = 10*torch.rand([n,2])-5.0  #torch.rand是均匀分布 （0,1）
w0 = torch.tensor([[2.0],[-3.0]])
b0 = torch.tensor([[10.0]])
Y = X@w0 + b0 + torch.normal(0.0,2.0,size = [n,1])  # @表示矩阵乘法,增加正态扰动
1
2
3
4
5
6
7
8
9
10
11
12
13
14

构建数据管道迭代器 模拟DataLoader

# 构建数据管道迭代器 模拟DataLoader
def data_iter(features, labels, batch_size=8):
    num_examples = len(features)
    indices = list(range(num_examples)) # 构建一个列表
    np.random.shuffle(indices)  #样本的读取顺序是随机的
    for i in range(0, num_examples, batch_size): 
        indexs = torch.LongTensor(indices[i: min(i + batch_size, num_examples)]) # 构建索引切片
        yield  features.index_select(0, indexs), labels.index_select(0, indexs) # yield用于生成
        
# 测试数据管道效果   输出一个batch数据
batch_size = 8
(features,labels) = next(data_iter(X, Y, batch_size))  # 刚刚用了 yield 所以现在用 next 就可以输出每个batch的数据（下面循环）
print(features)
print(labels)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

上述代码相当于原始实现/模拟 DataLoader 的原理。（如何打乱？如何迭代？如何获取每个batch数据？）

定义模型

# 定义模型
class LinearRegression: 
    
    def __init__(self):
        # 初始化 形状一致
        self.w = torch.randn_like(w0,requires_grad=True)
        self.b = torch.zeros_like(b0,requires_grad=True) # 偏置初始化为0
        
    #正向传播
    def forward(self,x): 
        return x@self.w + self.b

    # 损失函数
    def loss_fn(self,y_pred,y_true):  
        return torch.mean((y_pred - y_true)**2/2)

model = LinearRegression()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

训练模型

def train_step(model, features, labels):
    
    predictions = model.forward(features)
    loss = model.loss_fn(predictions,labels)
        
    # 反向传播求梯度
    loss.backward()
    
    # 使用torch.no_grad()避免梯度记录（避免梯度累加），也可以通过操作 model.w.data 实现避免梯度记录 
    with torch.no_grad():
        # 梯度下降法更新参数
        model.w -= 0.001*model.w.grad
        model.b -= 0.001*model.b.grad

        # 梯度清零 避免累加
        model.w.grad.zero_()
        model.b.grad.zero_()
    return loss 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

从上述代码可以看到，在梯度更新时都是用的梯度下降原理，还没有使用优化器。（优化器属于中阶API）

# 测试train_step效果
batch_size = 10
(features,labels) = next(data_iter(X,Y,batch_size)) # 一个batch
train_step(model,features,labels) 
1
2
3
4

def train_model(model,epochs):
    for epoch in range(1,epochs+1):
        for features, labels in data_iter(X,Y,10): # 刚刚有next 所以循环的时候可以输出每个batch来训练
            loss = train_step(model,features,labels)

        if epoch%20==0:
            printbar()
            print("epoch =",epoch,"loss = ",loss.item())
            print("model.w =",model.w.data)
            print("model.b =",model.b.data)

train_model(model,epochs = 200)
1
2
3
4
5
6
7
8
9
10
11
12

二、DNN二分类模型

准备数据

import numpy as np 
import pandas as pd 
from matplotlib import pyplot as plt
import torch
from torch import nn
%matplotlib inline
%config InlineBackend.figure_format = 'svg'

#正负样本数量
n_positive,n_negative = 2000,2000

#生成正样本, 小圆环分布
r_p = 5.0 + torch.normal(0.0,1.0,size = [n_positive,1]) 
theta_p = 2*np.pi*torch.rand([n_positive,1])
Xp = torch.cat([r_p*torch.cos(theta_p),r_p*torch.sin(theta_p)],axis = 1)
Yp = torch.ones_like(r_p)

#生成负样本, 大圆环分布
r_n = 8.0 + torch.normal(0.0,1.0,size = [n_negative,1]) 
theta_n = 2*np.pi*torch.rand([n_negative,1])
Xn = torch.cat([r_n*torch.cos(theta_n),r_n*torch.sin(theta_n)],axis = 1)
Yn = torch.zeros_like(r_n)

#汇总样本
X = torch.cat([Xp,Xn],axis = 0)
Y = torch.cat([Yp,Yn],axis = 0)


#可视化
plt.figure(figsize = (6,6))
plt.scatter(Xp[:,0].numpy(),Xp[:,1].numpy(),c = "r")
plt.scatter(Xn[:,0].numpy(),Xn[:,1].numpy(),c = "g")
plt.legend(["positive","negative"]);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# 构建数据管道迭代器
def data_iter(features, labels, batch_size=8):
    num_examples = len(features)
    indices = list(range(num_examples))
    np.random.shuffle(indices)  #样本的读取顺序是随机的
    for i in range(0, num_examples, batch_size):
        indexs = torch.LongTensor(indices[i: min(i + batch_size, num_examples)])
        yield  features.index_select(0, indexs), labels.index_select(0, indexs)
        
# 测试数据管道效果   
batch_size = 8
(features,labels) = next(data_iter(X,Y,batch_size))
print(features)
print(labels)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

定义模型

class DNNModel(nn.Module): # 深度神经网络 继承自 nn.Module
    def __init__(self):
        super(DNNModel, self).__init__()
        self.w1 = nn.Parameter(torch.randn(2,4))
        self.b1 = nn.Parameter(torch.zeros(1,4))
        self.w2 = nn.Parameter(torch.randn(4,8))
        self.b2 = nn.Parameter(torch.zeros(1,8))
        self.w3 = nn.Parameter(torch.randn(8,1))
        self.b3 = nn.Parameter(torch.zeros(1,1))

    # 正向传播
    def forward(self,x):
        x = torch.relu(x@self.w1 + self.b1)
        x = torch.relu(x@self.w2 + self.b2)
        y = torch.sigmoid(x@self.w3 + self.b3)
        return y
    
    # 损失函数(二元交叉熵)
    def loss_fn(self,y_pred,y_true):  
        #将预测值限制在1e-7以上, 1- (1e-7)以下，避免log(0)错误
        eps = 1e-7
        y_pred = torch.clamp(y_pred,eps,1.0-eps) # 将输入input张量每个元素的值压缩到区间 [min,max]，并返回结果到一个新张量
        bce = - y_true*torch.log(y_pred) - (1-y_true)*torch.log(1-y_pred) # bce loss 公式
        return torch.mean(bce)
    
    # 评估指标(准确率)
    def metric_fn(self,y_pred,y_true):
        y_pred = torch.where(y_pred>0.5,torch.ones_like(y_pred,dtype = torch.float32), # 条件是 >0.5 那么就是返回1 否则返回0
                          torch.zeros_like(y_pred,dtype = torch.float32))
        acc = torch.mean(1-torch.abs(y_true-y_pred))
        return acc
    
model = DNNModel()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

可以看到，损失函数，评估指标都是底层实现。

# 测试模型结构
batch_size = 10
(features,labels) = next(data_iter(X,Y,batch_size))

predictions = model(features)

loss = model.loss_fn(labels,predictions)
metric = model.metric_fn(labels,predictions)

print("init loss:", loss.item())
print("init metric:", metric.item())

1
2
3
4
5
6
7
8
9
10
11
12

训练模型

def train_step(model, features, labels):   
    
    # 正向传播求损失
    predictions = model.forward(features)
    loss = model.loss_fn(predictions,labels)
    metric = model.metric_fn(predictions,labels)
        
    # 反向传播求梯度
    loss.backward()
    
    # 梯度下降法更新参数
    for param in model.parameters(): # 6个参数
        #注意是对param.data进行重新赋值,避免此处操作引起梯度记录
        param.data = (param.data - 0.01*param.grad.data) 
        
    # 梯度清零
    model.zero_grad()
        
    return loss.item(),metric.item()
 

def train_model(model,epochs):
    for epoch in range(1,epochs+1):
        loss_list,metric_list = [],[]
        for features, labels in data_iter(X,Y,20):
            lossi,metrici = train_step(model,features,labels)
            loss_list.append(lossi)
            metric_list.append(metrici)
        loss = np.mean(loss_list)
        metric = np.mean(metric_list)

        if epoch%10==0:
            printbar()
            print("epoch =",epoch,"loss = ",loss,"metric = ",metric)
        
train_model(model,epochs = 100)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

参考：https://github.com/lyhue1991/eat_pytorch_in_20_days