環(huán)境
import torch
import numpy as np
import torch.nn as nn
import matplotlib.pyplot as plt
概要
使用 numpy 、pytorch和matplotlib,在線性回歸算法下實現(xiàn)曲線的擬合。
線性函數(shù)
使用高斯分布的隨機數(shù)噪音加擾。
數(shù)據(jù)生成
無噪音數(shù)據(jù)生成
因為我們的線性函數(shù)是 : 。所以先使用numpy生成x的數(shù)據(jù),在用x計算出y的數(shù)據(jù)即可。
函數(shù)如下:
def data_init_noise(start: int, step: float, end: int, plot: bool, noise: float = 0.45):
x = torch.from_numpy(np.arange(start, end, step).astype('float32').reshape(-1, 1))
# 加上 高斯 白噪音
y = x * 2 + torch.randn(x.shape) * noise
if plot:
plt.plot(x, y)
plt.scatter(x, y)
plt.draw()
return x, y
需要注意的點
對于
x = torch.from_numpy(np.arange(start, end, step).astype('float32').reshape(-1, 1))
我們先生成 np.arange(start, end, step) 數(shù)據(jù)序列。start,end對應開始和結束,step對應步幅。
將其數(shù)據(jù)類型轉化為 float32 ,再將其形狀從 一維行向量 轉成 二維列向量,再轉化為torch的tensor(張量)。
注意,這里如果你一開始初始化 x 成二維的,那么可以直接 x.T 進行逆置,但是我們初始化的是一維, 所以需要使用reshape(-1,1)。另一種解決方案是使用torch的unsqueeze函數(shù) torch.unsqueeze(-1)。
對于y,
y = x * 2 + torch.randn(x.shape) * noise
使用 torch.randd(x.shape)*noise 進行噪音加擾。
至此噪音加擾的數(shù)據(jù)集生成完畢。
網(wǎng)絡結構
自定義一個網(wǎng)絡結構如下:
class Linear_regression_net(nn.Module):
"""定義線性網(wǎng)絡模型類"""
def __init__(self, inputSize, outputSize):
super(Linear_regression_net, self).__init__()
self.net = torch.nn.Sequential()
self.net.add_module('Linear_1', nn.Linear(inputSize, outputSize, bias=True))
def forward(self, x):
"""實現(xiàn) `前向計算` """
out = self.net[0].forward(x)
return out
幾個需要注意的地方
1.創(chuàng)建自己的網(wǎng)絡類的時候都要繼承 nn.Module 類。
2.調用父類的初始化方法。
3.創(chuàng)建網(wǎng)絡結構時使用 Sequential 容器。
4.加入一個模塊 nn.Linear() , 添加不同網(wǎng)絡層的方法有很多種,這里不需要細說了。
forward前向傳播
def forward(self, x):
"""實現(xiàn) `前向計算` """
out = self.net[0].forward(x)
return out
剛開始學可能是看不懂的:前向傳播(foward propagation, FP)作用于每一層的輸入,通過逐層計算得到輸出結果;反向傳播(backward propagation, BP)作用于網(wǎng)絡的輸出,通過計算梯度由深到淺更新網(wǎng)絡參數(shù)。
在這里呢,前向傳播就相當于計算了一次 x*weight+bias,將得到的結果返回。
模型結構
這一段的代碼太長了,就不放出來了,參見文章最后的整體代碼。
次數(shù)模型的結構,我們定義了一個類 :class Linear_Model():
初始化
這里需要傳入的參數(shù)有:學習率learning_rate,迭代次數(shù)epoches,損失函數(shù) loss_func。
在初始化函數(shù)中,我們主要做了:初始化訓練所需的超參數(shù),選擇合適的損失函數(shù) 和 創(chuàng)建一個新的線性回歸 模型model。
注意:因為torch中的損失函數(shù)是一個具體的類,所以使用之前要先實例化 :
loss_func()。注意這個括號。

創(chuàng)建模型
定義線性模型
linear_mod 為一個 Linear_regression_net 的實例,傳入?yún)?shù) x_dim,y_dim。
初始化權重與偏執(zhí)
該網(wǎng)絡的第一層:->線性層,初始化權重為一個(0,0.1)的正態(tài)分布。
權重我們全部初始化為0。
self.linear_mod.net[0].weight.data.normal_(0, 0.01)
self.linear_mod.net[0].bias.data.fill_(0)

訓練的同時,實時畫圖
接收數(shù)據(jù)
這里data我們當成一個元組tuple來處理,先得到數(shù)據(jù)x和y。
在一個圖片中記錄訓練不同階段的成果
定義 num_picture 來記錄最大畫圖數(shù)目。
調整畫布大小為 (40,10)。
取其中某幾次的訓練數(shù)據(jù),使用利用epoch迭代的for循環(huán)來更新畫圖區(qū)域,并且使用plt.subplot()函數(shù)來進行畫子圖,在每幅上加上標題 epoch: xxx loss :xxx 。
注意matplotliab保存作圖時,應該在 plt.show()之前調用 plt.savefig()函數(shù)。圖片保存路徑需要存在,手動新建一個就行,或者 os.mkdir('Fig')
實時畫圖的細節(jié)
首先,對于使用Pycharm那些人來說,先按 Ctrl+Alt+S打開設置,找到 python scientific 模式,將這個選項關閉。

需要注意的地方就是:
使用plt的交互畫圖模式,
plt.ion() # 開啟一個畫圖的窗口
plt.pause(0.1) 時圖像停留,否則太快看不清。
plt.ioff() 之后調用 plt.show() 否則運行結束,窗口會自動關閉。
使用plt.show()之后,程序時阻塞狀態(tài)的,而使用plt.draw()程序會繼續(xù)運行,而不用手動取關閉窗口來使程序繼續(xù)執(zhí)行。
draw
訓練數(shù)據(jù)
prediction = self.linear_mod(x) 首先計算 前向傳播 的結果。
loss = self.loss_function(prediction, y) 使用均方誤差來計算損失值。
使用SDG隨機梯度下降來進行優(yōu)化。反向傳播,然后更新參數(shù)。
self.optimizer.zero_grad()
loss.backward()
with torch.no_grad():
for param in self.linear_mod.parameters():
param -= self.lr * param.grad
self.optimizer.step()
訓練結果
learning_rate=1e-6 epoch=1000

learning_rate=1e-6 epoch=2000

learning_rate=1e-6 epoch=5000

learning_rate=1e-4 epoch=500
可以看到,學習率調高兩個數(shù)量級之后,擬合效率大大提升。

learning_rate=5e-3 epoch=100
此時就說明學習率過高了,完全計算不出來了。


整體代碼
import torch
import numpy as np
import torch.nn as nn
import matplotlib.pyplot as plt
class Linear_regression_net(nn.Module):
"""定義線性網(wǎng)絡模型類"""
def __init__(self, inputSize, outputSize):
super(Linear_regression_net, self).__init__()
self.net = torch.nn.Sequential()
self.net.add_module('Linear_1', nn.Linear(inputSize, outputSize, bias=True))
def forward(self, x):
"""實現(xiàn) `前向計算` """
out = self.net[0].forward(x)
return out
class Linear_Model():
"""線性回歸模塊"""
def __init__(self, learning_rate: float, epoches: int, loss_func):
self.lr = learning_rate # 學習率
self.epoch = epoches # 迭代次數(shù)
# TODO:因為這個損失函數(shù)是一個類,所以調用之前要先實例化
self.loss_function = loss_func() # 損失函數(shù)
self.creeate_model(1, 1) # 新建模型
def creeate_model(self, x_dim, y_dim):
"""定義線性模型"""
self.linear_mod = Linear_regression_net(x_dim, y_dim) # y = 2*x 輸入輸出都是一維。
self.linear_mod.net[0].weight.data.normal_(0, 0.01)
self.linear_mod.net[0].bias.data.fill_(0)
# 隨機梯度下降優(yōu)化
self.optimizer = torch.optim.SGD(self.linear_mod.parameters(), lr=self.lr)
def train(self, data: tuple):
x = data[0]
y = data[1]
"""訓練數(shù)據(jù)"""
for epoch in range(self.epoch):
prediction = self.linear_mod(x)
loss = self.loss_function(prediction, y)
print(f"epoch:{epoch},loss:{loss}")
self.optimizer.zero_grad()
loss.backward()
with torch.no_grad():
for param in self.linear_mod.parameters():
param -= self.lr * param.grad
self.optimizer.step()
torch.save(self.linear_mod.state_dict(), "./Model_weight/線性回歸_合成數(shù)據(jù).pth")
def draw(self, data):
x = data[0]
y = data[1]
num_pictures = 8
fig = plt.figure(figsize=(40, 10))
current_fig = 0
"""訓練數(shù)據(jù)"""
for epoch in range(self.epoch):
prediction = self.linear_mod(x)
loss = self.loss_function(prediction, y)
print(f"epoch:{epoch},loss:{loss}")
self.optimizer.zero_grad()
loss.backward()
with torch.no_grad():
for param in self.linear_mod.parameters():
param -= self.lr * param.grad
self.optimizer.step()
"""畫多張圖片,保存為一張"""
# if epoch % (self.epoch / num_pictures) == 0:
# current_fig += 1
# plt.subplot(2, 4, current_fig)
# plt.title(f"epoch:{epoch} loss:{loss:^.3f}")
# plt.scatter(x, y)
# """擬合數(shù)據(jù)的線"""
# plt.plot(x, prediction.detach(), color="r")
# plt.savefig('Fig/線性回歸_合成數(shù)據(jù).png')
# plt.show()
"""實時顯示數(shù)據(jù)"""
plt.ion() # 開啟一個畫圖的窗口
plt.clf()
plt.title(f"epoch:{epoch} loss:{loss:^.3f}")
plt.scatter(x, y)
plt.plot(x, prediction.detach(), color="r")
plt.pause(0.1)
plt.ioff()
plt.show()
def dataset_init():
"""y = 2 x + 1 創(chuàng)建 dummy dataset (即 synthetic dataset)數(shù)據(jù)"""
# 行向量 -> 列向量
x = np.arange(1, 11, 0.2).astype('float32').reshape(-1, 1)
# 此時的輸出是不加 噪音干擾的
y = x * 2
return x, y
def data_init_noise(start: int, step: float, end: int, plot: bool, noise: float = 0.45):
x = torch.from_numpy(np.arange(start, end, step).astype('float32').reshape(-1, 1))
# 加上 高斯白噪音
y = x * 2 + torch.randn(x.shape) * noise
if plot:
plt.plot(x, y)
plt.scatter(x, y)
plt.draw()
return x, y
if __name__ == '__main__':
"""初始化數(shù)據(jù)集"""
Is_plot = True
x, y = data_init_noise(1, 0.3, 20, Is_plot, noise=2.23)
"""創(chuàng)建一個model"""
w = torch.randn(1, requires_grad=True)
b = torch.randn(1, requires_grad=True)
gyq_linear = Linear_Model(learning_rate=1e-4, epoches=100,
loss_func=nn.MSELoss) # MSELoss是均方誤差
gyq_linear.draw((x, y))
