本文并不是为了造轮子，只是通过手动实现来介绍建根本深度学习框架所需组件和步骤

Numpy 曾经提供了基本上所有须要的计算操作，咱们须要的是一个反对主动微分（autograd）的框架来计算多个操作的梯度，这是模块化办法构建神经网络层的标准化办法，通过主动微分的框架，咱们能够将优化器、激活函数等组合在一起用于训练神经网络。

所以一个根本的深度学习框架的组件总结如下：

  • 一个autograd零碎
  • 神经网络层
  • 神经网络模型
  • 优化器
  • 激活函数
  • 数据集

接下来，咱们将逐个介绍这些组件，看看它们的作用以及如何应用他们，这里将应用 gradflow（这是一个集体开源教育 autograd 零碎）因为它反对深度神经网络，并且和 PyTorch API基本一致。

Autograd零碎

这是最重要的组成部分，它是每个深度学习框架的根底，因为零碎将跟踪利用于输出张量的操作，并应用损失函数针对于每个参数的梯度来更新模型的权重。这里的一个必要条件是这些操作必须是可微的。

咱们的 autograd 零碎的根底是变量，通过为咱们须要的操作实现 dunder 办法（dunder 办法 ：Python中以双下划线结尾的非凡办法），咱们将可能跟踪每个实例的父实例是什么以及如何为它们计算梯度。为了帮忙进行一些操作，咱们将应用一个 numpy 数组来保留理论数据。

变量的另一个重要局部是反向流传办法，这将计算以后实例绝对于计算图中每个父类先人的梯度。在具体步骤中，咱们将应用父级的援用和原始操作中嵌入的梯度函数来更新 grad 成员字段。

以下代码片段蕴含主变量类初始化函数、增加操作的 dunder 办法和反向流传办法：

class Variable:    def __init__(        self,         data: np.ndarray,         parents: Tuple[Variable] = None,         requires_grad: bool = False    ) -> None:        self.data = data        self.grad: Union[int, None] = .0 if requires_grad else None        self.parents = parents or ()        self._requires_grad = requires_grad        self._back_grad_fn = lambda: None    def __add__(self, other: Variable) -> Variable:        if not isinstance(other, Variable):            raise TypeError("The second operator must be a Variable type")        result = self.data + other.data        variable = Variable(result, parents=(self, other))        if any((parent.requires_grad for parent in variable.parents)):            variable.requires_grad = True            def _back_grad_fn():                self.grad += variable.grad                other.grad += variable.grad            variable._back_grad_fn = _back_grad_fn        return variable          def backward(self, grad: Variable | np.ndarray = None) -> None:        if grad is None:            grad = np.array([1])                  self.grad = grad        variable_queue = [self]        while len(variable_queue):            variable = variable_queue.pop(0)            variable._back_grad_fn()            variable_queue.extend(list(variable.parents))s

在 _back_grad_fn 中还须要留神两件事，1、咱们须要将梯度增加到现有值，因为须要累积它们以防计算图中有多个门路达到该变量，2、还须要利用子节点的梯度，如果您想理解无关主动微分和矩阵微积分的更多详细信息，咱们会在后续的文章中具体介绍。

神经网络模块

对于理论的神经网络模块，咱们心愿灵便地实现新的层和现有模块的重用。所以这里hi用PyTorch API 相似的架构，创立一个须要实现 init 和 forward 办法的基类 Module。除了这两个办法，咱们还须要几个基于实用程序的办法来拜访参数和子模块。

class Module(ABC):    def __init__(self, training = True) -> None:        self._parameters: List[Variable] = []        self._modules: List[Module] = []        self._training = training        self._module_name = DEFAULT_MODULE_NAME    @abstractmethod    def forward(self, input: Variable) -> Variable:        raise NotImplemented    def add_parameter(self, parameter: Variable) -> Variable:        self._parameters.append(parameter)        return parameter    def add_module(self, module: Module) -> Module:        self._modules.append(module)        return module    def __call__(self, input: Variable) -> Variable:        return self.forward(input)    @property    def modules(self) -> List[Module]:        return self._modules    @property    def parameters(self) -> List[Variable]:        modules_parameters = []        modules = [self]        visited_modules: Set[Module] = set([])        while len(modules) != 0:            module = modules.pop()            if module in visited_modules:                raise RecursionError("Module already visited, cycle detected.")            modules_parameters.extend(module._parameters)            modules.extend(module.modules)            visited_modules.add(module)        return modules_parameters    def module_name(self) -> str:        return self._module_name

线性层

线形层是神经网络模型中应用的最多，也是最简略的层，咱们应用上一节中的形象模块实现一个简略的线性层。线形层的数学运算非常简单：

咱们将应用之前实现的变量来主动计算操作的理论后果和梯度，所以实现很简略：

class Linear(Module):    def __init__(self, in_size, out_size) -> None:        super().__init__()        weights_data: np.ndarray = np.random.uniform(size=in_size * out_size).reshape((in_size, out_size))        self.weights = Variable(weights_data, requires_grad=True)        self.b = Variable(np.random.uniform(size=out_size), requires_grad=True)        self.add_parameter(self.weights)        self.add_parameter(self.b)    def forward(self, input: Variable):        tmp = input @ self.weights        out = tmp + self.b        return out

激活函数

事实世界中的大多数数据在自变量和因变量之间存在非线性关系，咱们也心愿咱们的模型可能学习这种关系。如果咱们不在线性层上增加非线性激活函数，那么无论咱们增加多少线性层，最初咱们都能够只用一层（一个权重矩阵）来示意它们。

所以这里实现最简略也是最常见的激活函数ReLu

当实现 relu 函数时，还须要指定反向流传函数：

def relu(input: Variable) -> Variable:    result = np.maximum(input.data, 0)    variable = Variable(result, parents=(input,))    if input.requires_grad:        def _back_grad_fn():            input.grad += np.transpose((variable.data > 0)) * variable.grad        variable._back_grad_fn = _back_grad_fn        variable.requires_grad = True    return variable

优化器

在通过咱们的模型执行前向流传并通过咱们自定义的层进行梯度的反向流传之后，咱们须要理论更新参数使损失函数变得更小。

最简略的优化器之一是 SGD（随机梯度降落），在本文的实现中，咱们还是应用最简略的实现办法，仅应用梯度和学习率裁剪变动值增量并更新权重：

class BaseOptimizer(ABC):    def __init__(self, parameters: List[Variable], lr=0.0001) -> None:        super().__init__()        self._parameters = parameters        self._lr = lr    def zero_grad(self):        for parameter in self._parameters:            if parameter.requires_grad == False:                continue            if isinstance(parameter.grad, np.ndarray):                parameter.grad = np.zeros_like(parameter.grad)            else:                parameter.grad = np.array([0], dtype=np.float)    @abstractmethod    def step(self):        raise NotImplementedErrorclass NaiveSGD(BaseOptimizer):    def __init__(self, parameters: List[Variable], lr=0.001) -> None:        super().__init__(parameters=parameters, lr=lr)    def step(self):        for parameter in self._parameters:            clipped_grad = np.clip(parameter.grad, -1000, 1000)            delta = -self._lr * clipped_grad            delta = np.transpose(delta)            parameter.data = parameter.data + delta

数据集

最初组件就是数据集了，数据集尽管并不是外围组件然而它却十分的重要，一位内它能够帮忙咱们组织数据集，并将其集成到训练过程中。咱们也应用Pytorch的办法创立一个Dataset类，实现迭代器的dunder办法，并将特色X和标签Y转换为Variable类型:

class Dataset:    def __init__(self, features: np.ndarray, labels: np.ndarray, batch_size=16) -> None:        self._features = features        self._labels = labels        self._batch_size = batch_size        self._cur_index = 0        def __iter__(self):        self._cur_index = 0        return self    def __next__(self) -> Tuple[Variable, Variable]:        if self._cur_index >= len(self):            raise StopIteration                sample_batch, label_batch = self[self._cur_index]        self._cur_index += self._batch_size                return sample_batch, label_batch        def __getitem__(self, idx) -> Tuple[Variable, Variable]:        if idx >= len(self):            raise IndexError        sample_batch = Variable(self._features[self._cur_index: self._cur_index + self._batch_size])        label_batch = Variable(self._labels[self._cur_index : self._cur_index + self._batch_size])        return sample_batch, label_batch        def __len__(self) -> int:        return int(len(self._features) / self._batch_size)

训练

最初把所有货色放在一起，并应用 sklearn.datasets 应用人工生成的数据集训练一个简略的线性模型：

  X, y = datasets.make_regression(        n_samples=100,        n_features=1,        n_informative=1,        noise=10,        coef=False,        random_state=0    )    y = np.power(y, 2)    dataset = Dataset(X, y, batch_size=1)    return dataset      optimizer = NaiveSGD(model.parameters, lr=config["lr"])    training_loss = []    for epoch in range(epochs):        epoch_loss = .0        for X, y in dataset:            pred_y = model(X)            loss = (pred_y - y) @ (pred_y - y)            epoch_loss += loss.data.item()            loss.backward()            optimizer.step()            optimizer.zero_grad()        epoch_loss /= len(dataset)        training_loss.append(epoch_loss)        print(f"Epoch {epoch} | Loss: {epoch_loss}")

总结

就像结尾说的那样本文所展现的实现绝不是生产级的并且十分无限，然而能够让咱们更好地了解在其余风行框架的底层产生的一些操作，这是咱们学习和应用深度学习框架必不可少的局部。

https://avoid.overfit.cn/post/06fddd582b27492cae9a00e9f600dce4

作者：Tudor Surdoiu