devbox
小蓝xlanll
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

【深度学习实战】一、Numpy手撸神经网络实现线性回归_def update(self,x,y,grad_step): y_pred=self.predic

作者：小蓝xlanll | 2024-03-26 03:24:48

赞

踩

def update(self,x,y,grad_step): y_pred=self.predict(x) grad=np.zeros_like(se

目录

一、引言

二、代码实战

1、Tensor和初始化类

2、全连接层

3、模型组网

4、SGD优化器

5、均方差损失函数

6、Dataset

三、线性回归实战

四、实验结果

五、总结

一、引言

深度学习理论相对简单，但是深度学习框架（tensorflow/torch/paddlepaddle）源码却比较复杂，对于初学者来说，将代码和理论相结合理解是一个巨大的问题，本文的目标是使用python的numpy库从零开始实现一个简单的神经网络模型，并解决一个简单的线性回归问题。本文不会过多介绍理论，直接从代码入手，如需要了解相关理论请参考下方链接。

相关资料参考：

1、全连接层前向传播和梯度计算

2、动量梯度下降

3、ReLU

二、代码实战

1、Tensor和初始化类

Tensor：包含数据和梯度信息，神经网络层的参数已Tensor保存。

初始化类（Constant/Normal）：参数初始化方法。

  1. # 因为层的参数需要保存值和对应的梯度，这里定义梯度，可训练的参数全部以Tensor的类别保存
  2.  
  3. import numpy as np
  4. np.random.seed(10001)
  5.  
  6. class Tensor:
  7.     def __init__(self, shape):
  8.         self.data = np.zeros(shape=shape, dtype=np.float32) # 存放数据
  9.         self.grad = np.zeros(shape=shape, dtype=np.float32) # 存放梯度
  10.  
  11.     def clear_grad(self):
  12.         self.grad = np.zeros_like(self.grad)
  13.  
  14.     def __str__(self):
  15.         return "Tensor shape: {}, data: {}".format(self.data.shape, self.data)
  16.  
  17.  
  18. # Tensor的初始化类，目前仅提供Normal初始化和Constant初始化
  19. class Initializer:
  20.     """
  21.     基类
  22.     """
  23.     def __init__(self, shape=None, name='initializer'):
  24.         self.shape = shape
  25.         self.name = name
  26.  
  27.     def __call__(self, *args, **kwargs):
  28.         raise NotImplementedError
  29.  
  30.     def __str__(self):
  31.         return self.name
  32.  
  33.  
  34. class Constant(Initializer):
  35.     def __init__(self, value=0., name='constant initializer', *args, **kwargs):
  36.         super().__init__(name=name, *args, **kwargs)
  37.         self.value = value
  38.  
  39.     def __call__(self, shape=None, *args, **kwargs):
  40.         if shape:
  41.             self.shape = shape
  42.         assert shape is not None, "the shape of initializer must not be None."
  43.         return self.value + np.zeros(shape=self.shape)
  44.  
  45.  
  46. class Normal(Initializer):
  47.     def __init__(self, mean=0., std=0.01, name='normal initializer', *args, **kwargs):
  48.         super().__init__(name=name, *args, **kwargs)
  49.         self.mean = mean
  50.         self.std = std
  51.  
  52.     def __call__(self, shape=None, *args, **kwargs):
  53.         if shape:
  54.             self.shape = shape
  55.         assert shape is not None, "the shape of initializer must not be None."
  56.         return np.random.normal(self.mean, self.std, size=self.shape)

2、全连接层

Layer：层的基类，主要包括前向传播和反向传播。

Linear：全连接层，继承自Layer，具体化前向传播和反向传播的参数计算过程。

  1. # 为了使层能够组建起来，实现前向传播和反向传播，首先定义层的基类Layer
  2. # Layer的几个主要方法说明：
  3. #   forward: 实现前向传播
  4. #   backward: 实现反向传播
  5. #   parameters: 返回该层的参数，传入优化器进行优化
  6.  
  7. class Layer:
  8.     def __init__(self, name='layer', *args, **kwargs):
  9.         self.name = name
  10.  
  11.     def forward(self, *args, **kwargs):
  12.         raise NotImplementedError
  13.  
  14.     def backward(self):
  15.         raise NotImplementedError
  16.  
  17.     def parameters(self):
  18.         return []
  19.  
  20.     def __call__(self, *args, **kwargs):
  21.         return self.forward(*args, **kwargs)
  22.  
  23.     def __str__(self):
  24.         return self.name
  25.  
  26.  
  27. class Linear(Layer):
  28.     """
  29.     input X, shape: [N, C]
  30.     output Y, shape: [N, O]
  31.     weight W, shape: [C, O]
  32.     bias b, shape: [1, O]
  33.     grad dY, shape: [N, O]
  34.     forward formula:
  35.         Y = X @ W + b   # @表示矩阵乘法
  36.     backward formula:
  37.         dW = X.T @ dY
  38.         db = sum(dY, axis=0)
  39.         dX = dY @ W.T
  40.     """
  41.     def __init__(
  42.         self,
  43.         in_features,
  44.         out_features,
  45.         name='linear',
  46.         weight_attr=Normal(),
  47.         bias_attr=Constant(),
  48.         *args,
  49.         **kwargs
  50.         ):
  51.         super().__init__(name=name, *args, **kwargs)
  52.         self.weights = Tensor((in_features, out_features))
  53.         self.weights.data = weight_attr(self.weights.data.shape)
  54.         self.bias = Tensor((1, out_features))
  55.         self.bias.data = bias_attr(self.bias.data.shape)
  56.         self.input = None
  57.  
  58.     def forward(self, x):
  59.         self.input = x
  60.         output = np.dot(x, self.weights.data) + self.bias.data
  61.         return output
  62.  
  63.     def backward(self, gradient):
  64.         self.weights.grad += np.dot(self.input.T, gradient)  # dy / dw
  65.         self.bias.grad += np.sum(gradient, axis=0, keepdims=True)  # dy / db 
  66.         input_grad = np.dot(gradient, self.weights.data.T)  # dy / dx
  67.         return input_grad
  68.  
  69.     def parameters(self):
  70.         return [self.weights, self.bias]
  71.  
  72.     def __str__(self):
  73.         string = "linear layer, weight shape: {}, bias shape: {}".format(self.weights.data.shape, self.bias.data.shape)
  74.         return string
  75.  
  76.  
  77. class ReLU(Layer):
  78.     """
  79.     forward formula:
  80.         relu = x if x >= 0
  81.              = 0 if x < 0
  82.     backwawrd formula:
  83.         grad = gradient * (x > 0)
  84.     """
  85.     def __init__(self, name='relu', *args, **kwargs):
  86.         super().__init__(name=name, *args, **kwargs)
  87.         self.activated = None
  88.  
  89.     def forward(self, x):
  90.         x[x < 0] = 0             
  91.         self.activated = x
  92.         return self.activated
  93.  
  94.     def backward(self, gradient):
  95.         return gradient * (self.activated > 0)

3、模型组网

Sequential：模型组网类，将神经网络的层按照顺序叠加起来，实现顺序前向传播和反向传播。

  1. # 模型组网的功能是将层串起来，实现数据的前向传播和梯度的反向传播
  2. # 添加层的时候，按照顺序添加层的参数
  3. # Sequential方法说明：
  4. #   add: 向组网中添加层
  5. #   forward: 按照组网构建的层顺序，依次前向传播
  6. #   backward: 接收损失函数的梯度，按照层的逆序反向传播
  7. class Sequential:
  8.     def __init__(self, *args, **kwargs):
  9.         self.graphs = []
  10.         self._parameters = []
  11.         for arg_layer in args:
  12.             if isinstance(arg_layer, Layer):
  13.                 self.graphs.append(arg_layer)
  14.                 self._parameters += arg_layer.parameters()
  15.  
  16.     def add(self, layer):
  17.         assert isinstance(layer, Layer), "The type of added layer must be Layer, but got {}.".format(type(layer))
  18.         self.graphs.append(layer)
  19.         self._parameters += layer.parameters()
  20.  
  21.     def forward(self, x):
  22.         for graph in self.graphs:
  23.             x = graph(x)
  24.         return x
  25.  
  26.     def backward(self, grad):
  27.         # grad backward in inverse order of graph
  28.         for graph in self.graphs[::-1]:
  29.             grad = graph.backward(grad)
  30.  
  31.     def __call__(self, *args, **kwargs):
  32.         return self.forward(*args, **kwargs)
  33.  
  34.     def __str__(self):
  35.         string = 'Sequential:\n'
  36.         for graph in self.graphs:
  37.             string += graph.__str__() + '\n'
  38.         return string
  39.  
  40.     def parameters(self):
  41.         return self._parameters

4、SGD优化器

Optimizer：优化器基类，包括参数优化方法（step）、清空梯度（clear_grad）、计算正则化（get_decay）。

SGD：随机梯度下降类，实现了动量梯度下降方法。

  1. # 优化器主要完成根据梯度来优化参数的任务,其主要参数有学习率和正则化类型和正则化系数
  2. # Optimizer主要方法：
  3. #   step: 梯度反向传播后调用，该方法根据计算出的梯度，对参数进行优化
  4. #   clear_grad: 模型调用backward后，梯度会进行累加，如果已经调用step优化过参数，需要将使用过的梯度清空
  5. #   get_decay: 根据不同的正则化方法，计算出正则化惩罚值
  6. class Optimizer:
  7.     """
  8.     optimizer base class.
  9.     Args:
  10.         parameters (Tensor): parameters to be optimized.
  11.         learning_rate (float): learning rate. Default: 0.001.
  12.         weight_decay (float): The decay weight of parameters. Defaylt: 0.0.
  13.         decay_type (str): The type of regularizer. Default: l2.
  14.     """
  15.     def __init__(self, parameters, learning_rate=0.001, weight_decay=0.0, decay_type='l2'):
  16.         assert decay_type in ['l1', 'l2'], "only support decay_type 'l1' and 'l2', but got {}.".format(decay_type)
  17.         self.parameters = parameters
  18.         self.learning_rate = learning_rate
  19.         self.weight_decay = weight_decay
  20.         self.decay_type = decay_type
  21.         
  22.     def step(self):
  23.         raise NotImplementedError
  24.  
  25.     def clear_grad(self):
  26.         for p in self.parameters:
  27.             p.clear_grad()
  28.  
  29.     def get_decay(self, g):
  30.         if self.decay_type == 'l1':
  31.             return self.weight_decay
  32.         elif self.decay_type == 'l2':
  33.             return self.weight_decay * g
  34.  
  35. # 基本的梯度下降法为（不带正则化）：
  36. # W = W - learn_rate * dW
  37. # 带动量的梯度计算方法（减弱的梯度的随机性）：
  38. # dW = (momentum * v) + (1 - momentum) * dW
  39. class SGD(Optimizer):
  40.     def __init__(self, momentum=0.9, *args, **kwargs):
  41.         super().__init__(*args, **kwargs)
  42.         self.momentum = momentum
  43.         self.velocity = []
  44.         for p in self.parameters:
  45.             self.velocity.append(np.zeros_like(p.grad))
  46.  
  47.     def step(self):
  48.         for p, v in zip(self.parameters, self.velocity):
  49.             decay = self.get_decay(p.grad)
  50.             v = self.momentum * v + p.grad + decay # 动量计算
  51.             p.data = p.data - self.learning_rate * v

5、均方差损失函数

MSE：均方差损失函数。

  1. # 损失函数的设计延续了Layer的模式，但是因为需要注意的是forward和backward部分有些不同
  2. # MSE_loss = (predict_value - label) ^ 2
  3. # MSE方法和Layer的区别：
  4. #   forward：y是组网输出的值，target是目标值（这里的输入是组网的输出和目标值），前向传播的同时把dloss / dy 计算出来
  5. #   backward: 没有参数，因为在forward的时候，计算出了dloss / dy，所以这里不需要输入参数
  6. class MSE(Layer):
  7.     """
  8.     Mean Square Error:
  9.         J = 0.5 * (y - target)^2
  10.     gradient formula:
  11.         dJ/dy = y - target
  12.     """
  13.     def __init__(self, name='mse', reduction='mean', *args, **kwargs):
  14.         super().__init__(name=name, *args, **kwargs)
  15.         assert reduction in ['mean', 'none', 'sum'], "reduction only support 'mean', 'none' and 'sum', but got {}.".format(reduction)
  16.         self.reduction = reduction
  17.         self.pred = None
  18.         self.target = None
  19.  
  20.     def forward(self, y, target):
  21.         assert y.shape == target.shape, "The shape of y and target is not same, y shape = {} but target shape = {}".format(y.shape, target.shape)
  22.         self.pred = y
  23.         self.target = target
  24.         loss = 0.5 * np.square(y - target)
  25.         if self.reduction is 'mean':
  26.             return loss.mean()
  27.         elif self.reduction is 'none':
  28.             return loss
  29.         else:
  30.             return loss.sum()
  31.  
  32.     def backward(self):
  33.         gradient = self.pred - self.target
  34.         return gradient

6、Dataset

tensorflow/pytorch/paddlepaddle都有Dataset类，这里也简单实现一个Dataset类。

  1. # 这里仿照PaddlePaddle，Dataset需要实现__getitem__和__len__方法
  2. class Dataset:
  3.     def __init__(self, *args, **kwargs):
  4.         pass
  5.  
  6.     def __getitem__(self, idx):
  7.         raise NotImplementedError("'{}' not implement in class {}"
  8.                                   .format('__getitem__', self.__class__.__name__))
  9.  
  10.     def __len__(self):
  11.         raise NotImplementedError("'{}' not implement in class {}"
  12.                                   .format('__len__', self.__class__.__name__))
  13.  
  14.  
  15. # 根据dataset和一些设置，生成每个batch在dataset中的索引
  16. class BatchSampler:
  17.     def __init__(self, dataset=None, shuffle=False, batch_size=1, drop_last=False):
  18.         self.batch_size = batch_size
  19.         self.drop_last = drop_last
  20.         self.shuffle = shuffle
  21.  
  22.         self.num_data = len(dataset)
  23.         if self.drop_last or (self.num_data % batch_size == 0):
  24.             self.num_samples = self.num_data // batch_size
  25.         else:
  26.             self.num_samples = self.num_data // batch_size + 1
  27.         indices = np.arange(self.num_data)
  28.         if shuffle:
  29.             np.random.shuffle(indices)
  30.         if drop_last:
  31.             indices = indices[:self.num_samples * batch_size]
  32.         self.indices = indices
  33.  
  34.     def __len__(self):
  35.         return self.num_samples
  36.  
  37.     def __iter__(self):
  38.         batch_indices = []
  39.         for i in range(self.num_samples):
  40.             if (i + 1) * self.batch_size <= self.num_data:
  41.                 for idx in range(i * self.batch_size, (i + 1) * self.batch_size):
  42.                     batch_indices.append(self.indices[idx])
  43.                 yield batch_indices
  44.                 batch_indices = []
  45.             else:
  46.                 for idx in range(i * self.batch_size, self.num_data):
  47.                     batch_indices.append(self.indices[idx])
  48.         if not self.drop_last and len(batch_indices) > 0:
  49.             yield batch_indices
  50.  
  51.  
  52. # 根据sampler生成的索引，从dataset中取数据，并组合成一个batch
  53. class DataLoader:
  54.     def __init__(self, dataset, sampler=BatchSampler, shuffle=False, batch_size=1, drop_last=False):
  55.         self.dataset = dataset
  56.         self.sampler = sampler(dataset, shuffle, batch_size, drop_last)
  57.  
  58.     def __len__(self):
  59.         return len(self.sampler)
  60.  
  61.     def __call__(self):
  62.         self.__iter__()
  63.  
  64.     def __iter__(self):
  65.         for sample_indices in self.sampler:
  66.             data_list = []
  67.             label_list = []
  68.             for indice in sample_indices:
  69.                 data, label = self.dataset[indice]
  70.                 data_list.append(data)
  71.                 label_list.append(label)
  72.             yield np.stack(data_list, axis=0), np.stack(label_list, axis=0)

三、线性回归实战

生成一组数据，利用上面完成的类，实现线性回归。

  1. import matplotlib.pyplot as plt
  2.  
  3.  
  4. class LinearDataset(Dataset):
  5.     def __init__(self, X, Y):
  6.         self.X = X
  7.         self.Y = Y
  8.  
  9.     def __len__(self):
  10.         return len(self.X)
  11.  
  12.     def __getitem__(self, idx):
  13.         return self.X[idx], self.Y[idx]
  14.  
  15.  
  16. num_data = 200  # 训练数据数量
  17. val_number = 500    # 验证数据数量
  18.  
  19. epoches = 500
  20. batch_size = 4
  21. learning_rate = 0.01
  22.  
  23. X = np.linspace(-np.pi, np.pi, num_data).reshape(num_data, 1)
  24. Y = np.sin(X) * 2 + (np.random.rand(*X.shape) - 0.5) * 0.1
  25. y_ = np.sin(X) * 2
  26.  
  27. model = Sequential(
  28.     Linear(1, 16, name='linear1'),
  29.     ReLU(name='relu1'),
  30.     Linear(16, 64, name='linear2'),
  31.     ReLU(name='relu1'),
  32.     Linear(64, 16, name='linear2'),
  33.     ReLU(name='relu1'),
  34.     Linear(16, 1, name='linear2'),
  35. )
  36. opt = SGD(parameters=model.parameters(), learning_rate=learning_rate, weight_decay=0.0, decay_type='l2')
  37. loss_fn = MSE()
  38.  
  39. train_dataset = LinearDataset(X, Y)
  40. train_dataloader = DataLoader(train_dataset, shuffle=True, batch_size=4, drop_last=True)
  41. for epoch in range(1, epoches):
  42.     for x, y in train_dataloader:
  43.         pred = model(x)
  44.  
  45.         loss = loss_fn(pred, y)
  46.  
  47.         grad = loss_fn.backward()
  48.         model.backward(grad)
  49.  
  50.         opt.step()
  51.         opt.clear_grad()
  52.     print("epoch: {}. loss: {}".format(epoch, loss))
  53.  
  54.  
  55. X_val = np.linspace(-np.pi, np.pi, val_number).reshape(val_number, 1)
  56. Y_val = np.sin(X_val) * 2
  57. val_dataset = LinearDataset(X_val, Y_val)
  58. val_dataloader = DataLoader(val_dataset, shuffle=False, batch_size=2, drop_last=False)
  59. all_pred = []
  60. for x, y in val_dataloader:
  61.     pred = model(x)
  62.     all_pred.append(pred)
  63. all_pred = np.vstack(all_pred)
  64.  
  65. plt.scatter(X, Y, marker='x')
  66. plt.plot(X_val, all_pred, color='red')
  67. plt.show()

四、实验结果

实验数据如图所示：

拟合结果如图所示，可以看出本文实现的神经网络模型拟合效果。

五、总结

深度学习理论简单，而且成熟的框架很多，大部分深度学习框架的使用者都不需要关注框架底层的实现。本文使用python的numpy库实现了一个非常简单的神经网络模型，并且验证了该模型在线性回归问题上的效果，相信你一定能有所收获。

代码运行不起来？在线notebook体验链接：【深度学习实战】一、Numpy手撸神经网络实现线性回归

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小蓝xlanll/article/detail/314359
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号