自定义神经网络四之编写自定义神经网络

文章目录

- 前言
- 神经网络组件代码
- - 整体的项目结构
  - Tensor张量
  - Layers层
  - NeuralNet神经网络
  - Loss损失函数
  - Optim优化器
  - data数据处理
  - train训练
- 神经网络解决实际问题
- - 实际问题
  - 训练和推理代码
- 总结

前言

自定义神经网络一之Tensor和神经网络
自定义神经网络二之模型训练推理
自定义神经网络三之梯度和损失函数激活函数

经过前几篇的铺垫，终于到了自定义神经网络的部分。这也是博主一开始的目标，不使用任何框架，跟着大佬手撸一个神经网络，然后run起来。
目标如下：

实践神经网络中的基础组件，例如Tensor，神经网络层，优化器，损失函数，激活函数等
手动实践模型的训练，推理，解决实际的问题
保存模型到本地，并且通过加载本地模型实现推理任务。

整篇文章的代码来源：(需要科学上网)
https://www.youtube.com/watch?v=o64FV-ez6Gw
https://github.com/joelgrus/joelnet

神经网络组件代码

整体的项目结构

joelnet就是大佬封装的神经网络的类。麻雀虽小，五脏俱全。其他的神经网络框架，比如pytorch等，相对来说是丰富了主干，但万变不离其宗。大佬没有使用其他的第三方库，纯python手撸的神经网络类，值得学习。

Tensor张量

from numpy import ndarray as Tensor

我们这里不需要使用GPU来进行并行加速，因此使用python的ndarray就足够了，表示多维数组。

Layers层

from typing import Dict, Callable

import numpy as np

from joelnet.tensor import Tensor


class Layer:
    def __init__(self) -> None:
        self.params: Dict[str, Tensor] = {}
        self.grads: Dict[str, Tensor] = {}

    def forward(self, inputs: Tensor) -> Tensor:
        """
        Produce the outputs corresponding to these inputs
        """
        raise NotImplementedError

    def backward(self, grad: Tensor) -> Tensor:
        """
        Backpropagate this gradient thought the layer
        """
        raise NotImplementedError


class Linear(Layer):
    """
    computes output = input @ w + b
    这是一个线性层，也叫全连接层，在神经网络中非常常见。
    """
    def __init__(self, input_size: int, output_size: int) -> None:
        # inputs will be (bitch_size, input_size)
        # outputs will be (bitch_size, output_size)
        super().__init__()
        self.params["w"] = np.random.randn(input_size, output_size)
        self.params["b"] = np.random.randn(output_size)
    # 线性变换，其中@代表矩阵乘法，W是权重矩阵，b是偏差项（或称为偏置）
    def forward(self, inputs: Tensor) -> Tensor:
        """
        outputs = inputs @ w + b
        """
        self.inputs = inputs
        return inputs @ self.params["w"] + self.params["b"]

    # 计算误差，计算梯度，后续就可以使用优化算法（比如梯度下降）来更新参数W和b
    def backward(self, grad: Tensor) -> Tensor:
        # 当前层输出的梯度对偏差b的梯度，也就是误差项对偏差b的偏导数
        self.grads["b"] = np.sum(grad, axis=0)
        # 当前层输出的梯度对权重w的梯度，也就是误差项对权重w的偏导数
        self.grads["w"] = self.inputs.T @ grad
        # 计算损失对该层输入的梯度，这个梯度会被反向传播到前一层。
        return grad @ self.params["w"].T


"""
定义了一个类型 F。
F 类型的对象是一个函数，该函数接受一个 Tensor 类型的参数，并且返回一个 Tensor 类型的结果。
任何接受一个 Tensor 类型参数并返回一个 Tensor 类型结果的函数都可以被认为是 F 类型的函数。
"""
F = Callable[[Tensor], Tensor]


class Activation(Layer):
    """
    An activation layer just applies a function
    elementwise to its inputs
    激活层，构造函数的参数是：函数f和函数f的导数f_prime
    """
    def __init__(self, f: F, f_prime: F) -> None:
        super().__init__()
        self.f = f
        self.f_prime = f_prime

    def forward(self, inputs: Tensor) -> Tensor:
        self.inputs = inputs
        return self.f(inputs)

    def backward(self, grad: Tensor) -> Tensor:
        """
        if y = f(x) and x = g(z)
        then dy/dz = f'(x) * g'(z)
        """
        return self.f_prime(self.inputs) * grad


def tanh(x: Tensor) -> Tensor:
    return np.tanh(x)


def tanh_prime(x: Tensor) -> Tensor:
    y = tanh(x)
    return 1 - y ** 2


"""
tanh 是一种常见的激活函数，它的输出范围是 (-1, 1)。这样的输出范围通常
可能使得训练更加稳定，因为它可以把输入的数据标准化到一个比较小的范围。
"""


# Tanh 类是一个具体的激活层实现，它使用了 tanh 函数以及其导函数 tanh_prime。
class Tanh(Activation):
    def __init__(self) -> None:
        super().__init__(tanh, tanh_prime)

定义层代码，包括每层的参数，如权重，偏置等。定义线性层，实现前向传播和反向传播的逻辑。
同时定义层的激活函数，给神经元添加一些非线性因素，使得神经网络可以逼近任何复杂函数，提高神经网络模型的表达能力。

NeuralNet神经网络

from typing import Sequence, Iterator, Tuple

from joelnet.tensor import Tensor
from joelnet.layers import Layer


class NeuralNet:
    def __init__(self, layers: Sequence[Layer]) -> None:
        self.layers = layers

    def forward(self, inputs: Tensor) -> Tensor:
        for layer in self.layers:
            inputs = layer.forward(inputs)
        return inputs

    def backward(self, grad: Tensor) -> Tensor:
        for layer in reversed(self.layers):
            grad = layer.backward(grad)
        return grad

    def params_and_grads(self) -> Iterator[Tuple[Tensor, Tensor]]:
        for layer in self.layers:
            for name, param in layer.params.items():
                grad = layer.grads[name]
                yield param, grad

定义神经网络，这里主要是构造神经网络结构，把上面定义的层给加进去。 params_and_grads函数主要是输出每层的权重参数和梯度，方便后续的优化器使用。

Loss损失函数

import numpy as np

from joelnet.tensor import Tensor


class Loss:
    def loss(self, predicted: Tensor, actual: Tensor) -> float:
        raise NotImplementedError

    def grad(self, predicted: Tensor, actual: Tensor) -> Tensor:
        raise NotImplementedError


class MSE(Loss):
    """
    MSE is mean squared error, although we're just
    going to do total squared error
    """
    def loss(self, predicted: Tensor, actual: Tensor) -> float:
        return np.sum((predicted - actual) ** 2)

    def grad(self, predicted: Tensor, actual: Tensor) -> Tensor:
        return 2 * (predicted - actual)

定义**均方误差损失函数。 **计算方法为预测值与真实值之差的平方和的均值。MSE对于大的误差值具有很高的惩罚程度，因为差值会被平方。

Optim优化器

"""
We use an optimizer to adjust the parameters
of our network based on the gradients computed
during backprepagation
实现了一个名为SGD（随机梯度下降）的优化器，这是一种用于神经网络训练的常见优化算法。
在每个训练步骤里，它会调整神经网络的参数以达到减小损失函数的目的。
也就是一直说的梯度下降算法
"""
from joelnet.nn import NeuralNet


class Optimizer:
    def step(self, net: NeuralNet) -> None:
        raise NotImplementedError


class SGD(Optimizer):
    def __init__(self, lr: float = 0.01) -> None:
        self.lr = lr

    def step(self, net: NeuralNet) -> None:
        for param, grad in net.params_and_grads():
            param -= self.lr * grad

这里的学习率可以调整，学习率小的话，训练时间会变长，收敛的慢。学习率过大的话，收敛慢，甚至可能会导致不收敛。

data数据处理

from typing import Iterator, NamedTuple

import numpy as np

from joelnet.tensor import Tensor

Batch = NamedTuple("Batch", [("inputs", Tensor), ("targets", Tensor)])


class DataIterator:
    def __call__(self, inputs: Tensor, targets: Tensor) -> Iterator[Batch]:
        raise NotImplementedError


class BatchIterator(DataIterator):
    def __init__(self, batch_size: int = 32, shuffle: bool = True) -> None:
        self.batch_size = batch_size
        self.shuffle = shuffle

    def __call__(self, inputs: Tensor, targets: Tensor) -> Iterator[Batch]:
        starts = np.arange(0, len(inputs), self.batch_size)
        # 打乱数据
        if self.shuffle:
            np.random.shuffle(starts)
        # 根据batch_size对数据分批次
        for start in starts:
            end = start + self.batch_size
            batch_inputs = inputs[start:end]
            batch_targets = targets[start:end]
            yield Batch(batch_inputs, batch_targets)

这个文件的目的是在训练神经网络时，按批次获取输入数据和目标数据，以方便神经网络进行分批训练

train训练

from joelnet.tensor import Tensor
from joelnet.nn import NeuralNet
from joelnet.loss import Loss, MSE
from joelnet.optim import Optimizer, SGD
from joelnet.data import DataIterator, BatchIterator


def train(net: NeuralNet,
          inputs: Tensor,
          targets: Tensor,
          num_epochs: int = 5000,
          iterator: DataIterator = BatchIterator(),
          loss: Loss = MSE(),
          optimizer: Optimizer = SGD()) -> None:
    for epoch in range(num_epochs):
        epoch_loss = 0.0
        for batch in iterator(inputs, targets):
            predicted = net.forward(batch.inputs)
            epoch_loss += loss.loss(predicted, batch.targets)
            grad = loss.grad(predicted, batch.targets)
            net.backward(grad)
            optimizer.step(net)
        print(epoch, epoch_loss)

参数分别是：

输出和输出数据
训练5000代
数据迭代器
损失函数和优化器

训练逻辑如下：

迭代数据，获取输入，传入到神经网络
根据预测值和期望结果，计算神经网络的误差
获取神经网络梯度，进行反向传播，更新参数
优化器调整根据学习率调整梯度

神经网络解决实际问题

实际问题

输入1到100的数值，期望：

数值可以整除3，输出fizz
数值可以整除5，输出buzz
数值可以整除3和5，输出fizzbuzz
不符合条件的数值原样输出

训练和推理代码

from typing import List

import numpy as np
import pickle

from joelnet.train import train
from joelnet.nn import NeuralNet
from joelnet.layers import Linear, Tanh
from joelnet.optim import SGD


def fizz_buzz_encode(x: int) -> List[int]:
    if x % 15 == 0:
        return [0, 0, 0, 1]
    elif x % 5 == 0:
        return [0, 0, 1, 0]
    elif x % 3 == 0:
        return [0, 1, 0, 0]
    else:
        return [1, 0, 0, 0]


# 整数转换成二进制编码，可以减小输入数据规模
# 神经网络不能直接理解整数，转换成二进制编码可以为
# 神经网络提供了一种更有效的信息表达方式，使得网络能从中学习到更多有用的信息。
def binary_encode(x: int) -> List[int]:
    """
    10 digit binary encoding of x
    """
    return [x >> i & 1 for i in range(10)]


# 训练数据是从101-1024之间的数字
inputs = np.array([
    binary_encode(x)
    for x in range(101, 1024)
])

# 列表生成式语法：[表达式 for 元素 in 可迭代对象]
targets = np.array([
    fizz_buzz_encode(x)
    for x in range(101, 1024)
])

# 两层线性变换神经网络
net = NeuralNet([
    Linear(input_size=10, output_size=50),
    Tanh(),
    Linear(input_size=50, output_size=4)
])

train(net,
      inputs,
      targets,
      num_epochs=50000,
      optimizer=SGD(lr=0.001))

print("save model")
# 保存模型。模型的大小和神经元的个数，层数等有关系
# 神经网络output_size=50的时候，模型大小是41k
# 神经网络output_size=60的时候，模型大小是48k
# 神经网络output_size=50，且设置为3层的时候，模型大小51k
with open('fizzbuzz.pkl', 'wb') as f:
    pickle.dump(net, f)


# 读取模型，进行推理
with open('fizzbuzz.pkl', 'rb') as f:
    loaded_net = pickle.load(f)

# 传入输入数据，获取模型推理结果
for x in range(1, 101):
    predicted = loaded_net.forward(binary_encode(x))
    predicted_idx = np.argmax(predicted)
    actual_idx = np.argmax(fizz_buzz_encode(x))
    labels = [str(x), "fizz", "buzz", "fizzbuzz"]
    # 输出预测值和实际应该返回的值
    print(x, labels[predicted_idx], labels[actual_idx])

注意，这里的训练epoch和lr学习率都可以自己调整的，博主这里分别调整到了训练5w次和学习率0.001.
模型大小和神经网络的层数和参数量有关系，具体可以看注释。
以上代码不依赖特殊的库，理论上来说可以直接运行的。

模型输出如下：

90 fizzbuzz fizzbuzz
91 91 91
92 92 92
93 fizz fizz
94 94 94
95 buzz buzz
96 fizz fizz
97 97 97
98 98 98
99 fizz fizz
100 buzz buzz

可以看到，模型推理出来的预测值和实际的值是一致的，说明训练有效果。

总结

本博客是在大佬代码的基础上，实现了自定义神经网络的训练和推理。外网上的优秀文章和视频太多了，可惜限于网络和语言，能被我们看到的太少了。这个大佬40多分钟就手撸了简单的神经网络类，并且实现了训练和推理，博主只能说，牛逼。

本系列文章到这里就结束了。本来只是想分享一下大佬的视频和代码，但直接输出难免会没有上下文，因此只能把以前的一些笔记梳理下，期望读者能先有一些基础概念，然后再手撸代码实现一个自己的神经网络。

不得不感概，现在的网络资料太多了。依稀记得18年尝试学习一下TensorFlow的痛苦，上来就是各种公式和专业术语轰炸，直接劝退。近几年随着大量的工程师涌入人工智能领域，网上的教程也越来越通俗易懂了，站在工程的角度去讲概念，从实用性角度去学习可简单太多了。

end