神经网络能否拟合x^2 + y^2 = 100在第一象限的曲线？
第一象限的曲线方程如下所示：
y
=
100
−
x
2
y = \sqrt{100-x^2}
y=100−x2
在[0, 10]中等距生成1000个点，划分训练集、开发集和测试集，构建神经网络训练。神经网络架构图如下，采用两层神经网络进行拟合，根据情况调节神经网络的深度以及隐藏层神经元个数。
import random
import matplotlib.pyplot as plt
import numpy as np
import torch
from torch import nn
from torch.utils.data import Dataset, DataLoader
import torch.nn.functional as F
def setup_seed(seed):
# 固定所有的随机数种子，使结果能够复现
torch.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
np.random.seed(seed)
random.seed(seed)
torch.backends.cudnn.deterministic = True
def generate_data(num=1000):
x = np.linspace(0, 10, num)
y = [np.sqrt(100 - i ** 2) for i in x]
return list(zip(x, y))
def distribute_dataset(data, rate):
total_num = len(data)
train_pos = int(total_num * rate[0])
dev_pos = int(total_num * (rate[0] + rate[1]))
train_data = data[: train_pos]
dev_data = data[train_pos: dev_pos]
test_data = data[dev_pos:]
return train_data, dev_data, test_data
class MyDataset(Dataset):
# 继承torch工具类的Dataset，进而构造DataLoader进行批量数据读取
# 继承Dataset需要实现以下3个函数
def __init__(self, dataset):
super().__init__()
data_x, data_y = zip(*dataset)
self.data_x = data_x
self.data_y = data_y
def __getitem__(self, item):
return self.data_x[item], self.data_y[item]
def __len__(self):
assert len(self.data_x) == len(self.data_y)
return len(self.data_x)
class MyModel(nn.Module):
# 继承nn.Module构建模型
def __init__(self, hidden_size):
super(MyModel, self).__init__()
self.h1 = nn.Linear(1, hidden_size)
self.h2 = nn.Linear(hidden_size, 1)
def forward(self, inputs):
out1 = self.h1(inputs)
out2 = self.h2(F.relu(out1))
return out2
if __name__ == '__main__':
# 可调超参数
data_num = 1000
tdt_rate = [0.7, 0.2, 0.1]
seed = 1
hidden_size = 128
batch_size = 4
lr = 1e-3
epoch = 50
# 固定随机数
setup_seed(seed)
# 生成数据
total_data = generate_data(data_num)
random.shuffle(total_data)
# 构建训练集、开发集、测试集
train_data, dev_data, test_data = distribute_dataset(total_data, tdt_rate)
train_ds = MyDataset(train_data)
dev_ds = MyDataset(dev_data)
test_ds = MyDataset(test_data)
# 构建模型
model = MyModel(hidden_size)
loss = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=lr)
# 训练集训练
train_loader = DataLoader(train_ds, batch_size=batch_size)
for e in range(epoch):
total_l = []
for x, y in train_loader:
# 调整数据形式 x -> [batch_size, 1], y -> [batch_size]
x = x.unsqueeze(-1).type(torch.float32)
y = y.type(torch.float32)
# 均方误差 (预测值 - 真实值)^2
l = loss(model(x).squeeze(-1), y)
# 邢将军法 |sign(预测值) * 预测值^2 - 真实值^2|
# y_ = model(x).squeeze(-1)
# sign = torch.sign(y_)
# l = torch.abs(sign * y_ ** 2 - y ** 2).sum()
# |预测值 - 真实值|
# l = torch.abs(model(x).squeeze(-1) - y).sum()
# |预测值^3 - 真实值^3|
# l = torch.abs(model(x).squeeze(-1) ** 3 - y ** 3).sum()
l.backward()
optimizer.step()
optimizer.zero_grad()
total_l.append(l.item())
print(f"epoch {e+1}, avg_loss {sum(total_l) / len(total_l)}")
# for e in range(epoch):
# # (x, y) -> (y, x) 增加靠近10的数据量，让模型再靠近10的地方拟合地更好
# total_l = []
# for x, y in train_loader:
# x, y = y, x
# # 调整数据形式 x -> [batch_size, 1], y -> [batch_size]
# x = x.unsqueeze(-1).type(torch.float32)
# y = y.type(torch.float32)
#
# # 均方误差 (预测值 - 真实值)^2
# # l = loss(model(x).squeeze(-1), y)
#
# # 邢将军法 |sign(预测值) * 预测值^2 - 真实值^2|
# y_ = model(x).squeeze(-1)
# sign = torch.sign(y_)
# l = torch.abs(sign * y_ ** 2 - y ** 2).sum()
#
# # |预测值 - 真实值|
# # l = torch.abs(model(x).squeeze(-1) - y).sum()
#
# # |预测值^3 - 真实值^3|
# # l = torch.abs(model(x).squeeze(-1) ** 3 - y ** 3).sum()
#
# l.backward()
# optimizer.step()
# optimizer.zero_grad()
# total_l.append(l.item())
# print(f"reverse epoch {e+1}, avg_loss {sum(total_l) / len(total_l)}")
# 开发集调超参数
dev_loader = DataLoader(dev_ds, batch_size=batch_size)
with torch.no_grad():
xs, ys, y_s = [], [], []
for x, y in dev_loader:
xs.extend(list(x.numpy()))
ys.extend(list(y.numpy()))
x = x.unsqueeze(-1).type(torch.float32)
y = y.type(torch.float32)
y_ = model(x).squeeze(-1)
y_s.extend(list(y_.numpy()))
# 排序
total = [(a, b, c) for a, b, c in zip(xs, ys, y_s)]
total.sort(key=lambda i: i[0])
xs, ys, y_s = zip(*total)
plt.plot(xs, ys, label='real')
plt.plot(xs, y_s, label='pred')
plt.legend()
plt.title('dev')
plt.show()
# 测试集测试
test_loader = DataLoader(test_ds, batch_size=batch_size)
with torch.no_grad():
xs, ys, y_s = [], [], []
for x, y in test_loader:
xs.extend(list(x.numpy()))
ys.extend(list(y.numpy()))
x = x.unsqueeze(-1).type(torch.float32)
y = y.type(torch.float32)
y_ = model(x).squeeze(-1)
y_s.extend(list(y_.numpy()))
# 排序
total = [(a, b, c) for a, b, c in zip(xs, ys, y_s)]
total.sort(key=lambda i: i[0])
xs, ys, y_s = zip(*total)
plt.plot(xs, ys, label='real')
plt.plot(xs, y_s, label='pred')
plt.legend()
plt.title('test')
plt.show()
神经网络的学习能力很强，获得不错的拟合效果。
上述代码中的loss选用的是均方误差——(预测值 - 真实值) ^2，邢将军提出了新的损失函数——|sign(预测值) * 预测值^2 - 真实值^2|，其中sign是符号函数，公式如下：
损失函数描述预测值和真实值之间的误差，当预测值趋近于真实值时，损失函数应越来越小。预测值^2 - 真实值^2 满足损失函数的基本定义。
损失函数不能为负数，网络的优化目标是最小化损失函数，如果损失可为负数，网络将把损失推向负无穷，所以加上绝对值，成为**|预测值^2 - 真实值^2|**。
在此次拟合中，真实值的取值范围是[0, 10]，永为正数。当预测值趋近于真实值的相反数时，损失依然在不断减小。如果初始梯度方向是向着真实值相反数的方向，最终就会导致出现关于x轴对称的拟合曲线出现。所以当预测值趋近于真实值的相反数时，要让损失变大。这里使用sign函数，当预测值为负值时，将产生更大的损失。最终的损失函数为**|sign(预测值) * 预测值^2 - 真实值^2|**。
代码实现如下：
y_ = model(x).squeeze(-1)
sign = torch.sign(y_)
l = torch.abs(sign * y_ ** 2 - y ** 2).sum()
效果图如下，除了尾部数据仍然有偏差，拟合效果大幅提升。
基于邢将军的启发，我们可以设计更多损失函数，这些损失函数都表现不错。
sr同学认为预拟合的曲线在越靠近10的地方，|斜率|越来越大，最终极限为无穷大。在数据生成阶段，x是等距采样的，而y值的变化并非均匀。越靠近10，y值的变化越明显，特征越稀疏，不利于模型学习。
sr同学认为 x^2 + y^2 = 100 中，x与y是对称的，也就是说x与y可以互换。如果将生成的数据(x, y)对调成(y, x)，则原来靠近0的地方数据更密集，现在靠近10的地方数据更密集。让模型再学习(y, x)，尾部的偏差就能够拟合地更好。
代码很简单，将原训练代码的x与y互换即可。
for e in range(epoch):
# (x, y) -> (y, x) 增加靠近10的数据量，让模型再靠近10的地方拟合地更好
total_l = []
for x, y in train_loader:
# 对调
x, y = y, x
# 调整数据形式 x -> [batch_size, 1], y -> [batch_size]
x = x.unsqueeze(-1).type(torch.float32)
y = y.type(torch.float32)
# 均方误差 (预测值 - 真实值)^2
# l = loss(model(x).squeeze(-1), y)
# 邢将军法 |sign(预测值) * 预测值^2 - 真实值^2|
y_ = model(x).squeeze(-1)
sign = torch.sign(y_)
l = torch.abs(sign * y_ ** 2 - y ** 2).sum()
# |预测值 - 真实值|
# l = torch.abs(model(x).squeeze(-1) - y).sum()
# |预测值^3 - 真实值^3|
# l = torch.abs(model(x).squeeze(-1) ** 3 - y ** 3).sum()
l.backward()
optimizer.step()
optimizer.zero_grad()
total_l.append(l.item())
print(f"reverse epoch {e+1}, avg_loss {sum(total_l) / len(total_l)}")
效果图如下，尾部数据也得到有效的拟合。
