下面的范例使用Pytorch的低阶API实现线性回归模型和DNN二分类模型。
低阶API主要包括张量操作，计算图和自动微分。

  1. import os
  2. import datetime
  4. #打印时间
  5. def printbar():
  6.     nowtime = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
  7.     print("\n"+"=========="*8 + "%s"%nowtime)
  9. #mac系统上pytorch和matplotlib在jupyter中同时跑需要更改环境变量
  10. os.environ["KMP_DUPLICATE_LIB_OK"]="TRUE"

线性回归模型

准备数据

  1. import numpy as np 
  2. import pandas as pd
  3. from matplotlib import pyplot as plt 
  4. import torch
  5. from torch import nn
  8. #样本数量
  9. n = 400
  11. # 生成测试用数据集
  12. X = 10*torch.rand([n,2])-5.0  #torch.rand是均匀分布 
  13. w0 = torch.tensor([[2.0],[-3.0]])
  14. b0 = torch.tensor([[10.0]])
  15. Y = X@w0 + b0 + torch.normal( 0.0,2.0,size = [n,1])  # @表示矩阵乘法,增加正态扰动
  1. # 数据可视化
  3. %matplotlib inline
  4. %config InlineBackend.figure_format = 'svg'
  6. plt.figure(figsize = (12,5))
  7. ax1 = plt.subplot(121)
  8. ax1.scatter(X[:,0].numpy(),Y[:,0].numpy(), c = "b",label = "samples")
  9. ax1.legend()
  10. plt.xlabel("x1")
  11. plt.ylabel("y",rotation = 0)
  13. ax2 = plt.subplot(122)
  14. ax2.scatter(X[:,1].numpy(),Y[:,0].numpy(), c = "g",label = "samples")
  15. ax2.legend()
  16. plt.xlabel("x2")
  17. plt.ylabel("y",rotation = 0)
  18. plt.show()

image.png

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

image.png

定义模型

  1. # 定义模型
  2. class LinearRegression: 
  4.     def __init__(self):
  5.         self.w = torch.randn_like(w0,requires_grad=True)
  6.         self.b = torch.zeros_like(b0,requires_grad=True)
  8.     #正向传播
  9.     def forward(self,x): 
  10.         return x@self.w + self.b
  12.     # 损失函数
  13.     def loss_func(self,y_pred,y_true):  
  14.         return torch.mean((y_pred - y_true)**2/2)
  16. model = LinearRegression()

训练模型

  1. def train_step(model, features, labels):
  3.     predictions = model.forward(features)
  4.     loss = model.loss_func(predictions,labels)
  6.     # 反向传播求梯度
  7.     loss.backward()
  9.     # 使用torch.no_grad()避免梯度记录，也可以通过操作 model.w.data 实现避免梯度记录 
  10.     with torch.no_grad():
  11.         # 梯度下降法更新参数
  12.         model.w -= 0.001*model.w.grad
  13.         model.b -= 0.001*model.b.grad
  15.         # 梯度清零
  16.         model.w.grad.zero_()
  17.         model.b.grad.zero_()
  18.     return loss
  1. # 测试train_step效果
  2. batch_size = 10
  3. (features,labels) = next(data_iter(X,Y,batch_size))
  4. train_step(model,features,labels)

image.png

  1. def train_model(model,epochs):
  2.     for epoch in range(1,epochs+1):
  3.         for features, labels in data_iter(X,Y,10):
  4.             loss = train_step(model,features,labels)
  6.         if epoch%200==0:
  7.             printbar()
  8.             print("epoch =",epoch,"loss = ",loss.item())
  9.             print("model.w =",model.w.data)
  10.             print("model.b =",model.b.data)
  12. train_model(model,epochs = 1000)

image.png

  1. # 结果可视化
  3. %matplotlib inline
  4. %config InlineBackend.figure_format = 'svg'
  6. plt.figure(figsize = (12,5))
  7. ax1 = plt.subplot(121)
  8. ax1.scatter(X[:,0].numpy(),Y[:,0].numpy(), c = "b",label = "samples")
  9. ax1.plot(X[:,0].numpy(),(model.w[0].data*X[:,0]+model.b[0].data).numpy(),"-r",linewidth = 5.0,label = "model")
  10. ax1.legend()
  11. plt.xlabel("x1")
  12. plt.ylabel("y",rotation = 0)
  15. ax2 = plt.subplot(122)
  16. ax2.scatter(X[:,1].numpy(),Y[:,0].numpy(), c = "g",label = "samples")
  17. ax2.plot(X[:,1].numpy(),(model.w[1].data*X[:,1]+model.b[0].data).numpy(),"-r",linewidth = 5.0,label = "model")
  18. ax2.legend()
  19. plt.xlabel("x2")
  20. plt.ylabel("y",rotation = 0)
  22. plt.show()

image.png

DNN二分类模型

准备数据

  1. import numpy as np 
  2. import pandas as pd 
  3. from matplotlib import pyplot as plt
  4. import torch
  5. from torch import nn
  6. %matplotlib inline
  7. %config InlineBackend.figure_format = 'svg'
  9. #正负样本数量
  10. n_positive,n_negative = 2000,2000
  12. #生成正样本, 小圆环分布
  13. r_p = 5.0 + torch.normal(0.0,1.0,size = [n_positive,1]) 
  14. theta_p = 2*np.pi*torch.rand([n_positive,1])
  15. Xp = torch.cat([r_p*torch.cos(theta_p),r_p*torch.sin(theta_p)],axis = 1)
  16. Yp = torch.ones_like(r_p)
  18. #生成负样本, 大圆环分布
  19. r_n = 8.0 + torch.normal(0.0,1.0,size = [n_negative,1]) 
  20. theta_n = 2*np.pi*torch.rand([n_negative,1])
  21. Xn = torch.cat([r_n*torch.cos(theta_n),r_n*torch.sin(theta_n)],axis = 1)
  22. Yn = torch.zeros_like(r_n)
  24. #汇总样本
  25. X = torch.cat([Xp,Xn],axis = 0)
  26. Y = torch.cat([Yp,Yn],axis = 0)
  29. #可视化
  30. plt.figure(figsize = (6,6))
  31. plt.scatter(Xp[:,0].numpy(),Xp[:,1].numpy(),c = "r")
  32. plt.scatter(Xn[:,0].numpy(),Xn[:,1].numpy(),c = "g")
  33. plt.legend(["positive","negative"]);

image.png

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

image.png

定义模型

此处范例我们利用nn.Module来组织模型变量。

  1. class DNNModel(nn.Module):
  2.     def __init__(self):
  3.         super(DNNModel, self).__init__()
  4.         self.w1 = nn.Parameter(torch.randn(2,4))
  5.         self.b1 = nn.Parameter(torch.zeros(1,4))
  6.         self.w2 = nn.Parameter(torch.randn(4,8))
  7.         self.b2 = nn.Parameter(torch.zeros(1,8))
  8.         self.w3 = nn.Parameter(torch.randn(8,1))
  9.         self.b3 = nn.Parameter(torch.zeros(1,1))
  11.     # 正向传播
  12.     def forward(self,x):
  13.         x = torch.relu(x@self.w1 + self.b1)
  14.         x = torch.relu(x@self.w2 + self.b2)
  15.         y = torch.sigmoid(x@self.w3 + self.b3)
  16.         return y
  18.     # 损失函数(二元交叉熵)
  19.     def loss_func(self,y_pred,y_true):  
  20.         #将预测值限制在1e-7以上, 1- (1e-7)以下，避免log(0)错误
  21.         eps = 1e-7
  22.         y_pred = torch.clamp(y_pred,eps,1.0-eps)
  23.         bce = - y_true*torch.log(y_pred) - (1-y_true)*torch.log(1-y_pred)
  24.         return torch.mean(bce)
  26.     # 评估指标(准确率)
  27.     def metric_func(self,y_pred,y_true):
  28.         y_pred = torch.where(y_pred>0.5,torch.ones_like(y_pred,dtype = torch.float32),torch.zeros_like(y_pred,dtype = torch.float32))
  29.         acc = torch.mean(1-torch.abs(y_true-y_pred))
  30.         return acc
  32. model = DNNModel()
  1. # 测试模型结构
  2. batch_size = 10
  3. (features,labels) = next(data_iter(X,Y,batch_size))
  5. predictions = model(features)
  7. loss = model.loss_func(labels,predictions)
  8. metric = model.metric_func(labels,predictions)
  10. print("init loss:", loss.item())
  11. print("init metric:", metric.item())

image.png

  1. len(list(model.parameters()))

image.png

训练模型

  1. def train_step(model, features, labels):   
  3.     # 正向传播求损失
  4.     predictions = model.forward(features)
  5.     loss = model.loss_func(predictions,labels)
  6.     metric = model.metric_func(predictions,labels)
  8.     # 反向传播求梯度
  9.     loss.backward()
  11.     # 梯度下降法更新参数
  12.     for param in model.parameters():
  13.         #注意是对param.data进行重新赋值,避免此处操作引起梯度记录
  14.         param.data = (param.data - 0.01*param.grad.data) 
  16.     # 梯度清零
  17.     model.zero_grad()
  19.     return loss.item(),metric.item()
  22. def train_model(model,epochs):
  23.     for epoch in range(1,epochs+1):
  24.         loss_list,metric_list = [],[]
  25.         for features, labels in data_iter(X,Y,20):
  26.             lossi,metrici = train_step(model,features,labels)
  27.             loss_list.append(lossi)
  28.             metric_list.append(metrici)
  29.         loss = np.mean(loss_list)
  30.         metric = np.mean(metric_list)
  32.         if epoch%100==0:
  33.             printbar()
  34.             print("epoch =",epoch,"loss = ",loss,"metric = ",metric)
  36. train_model(model,epochs = 1000)

image.png

  1. # 结果可视化
  2. fig, (ax1,ax2) = plt.subplots(nrows=1,ncols=2,figsize = (12,5))
  3. ax1.scatter(Xp[:,0],Xp[:,1], c="r")
  4. ax1.scatter(Xn[:,0],Xn[:,1],c = "g")
  5. ax1.legend(["positive","negative"]);
  6. ax1.set_title("y_true");
  8. Xp_pred = X[torch.squeeze(model.forward(X)>=0.5)]
  9. Xn_pred = X[torch.squeeze(model.forward(X)<0.5)]
  11. ax2.scatter(Xp_pred[:,0],Xp_pred[:,1],c = "r")
  12. ax2.scatter(Xn_pred[:,0],Xn_pred[:,1],c = "g")
  13. ax2.legend(["positive","negative"]);
  14. ax2.set_title("y_pred");

image.png