自制深度学习框架一、深度学习数据结构设计1.1 Tensor数据结构二、基于 Tensor 的计算设计2.1 Function 的结构设计2.1.1 前向与反向传播2.2 单变量自动求导2.2.1 单变量计算图2.3 多变量自动求导2.3.1 可变参数2.3.2 重复变量2.3.3 导数重置2.3.4 传播顺序2.4 内存管理2.4.1 引用计数法2.4.2 分代垃圾回收2.4.3 弱引用2.4.4 清除无用导数2.5 模式切换2.6 属性扩展2.7 运算符重载三、Python打包3.1 概述3.2 项目配置一、深度学习数据结构设计在构建深度学习框架时数据结构是整个框架设计的起点。本章选择 ndarray 作为底层基础其原因在于它已经提供了成熟且高效的多维数组计算能力包括向量化运算、广播机制以及底层的高性能实现。这使得我们无需重新实现底层数值计算就可以直接获得接近工业级的计算效率从而将精力集中在框架层的设计上。然而深度学习并不仅仅是数值计算问题还涉及“计算过程的管理”。例如张量之间的依赖关系、梯度传播路径、以及模型参数的更新机制这些都是 ndarray 本身所不具备的能力。换言之ndarray 只解决了“算得快”的问题但没有解决“怎么学”的问题。因此本章在 ndarray 的基础上进行封装引入 Tensor 这一新的数据结构使其不仅保留高效的数值计算能力同时也具备进入深度学习计算系统所需的语义扩展能力。通过这一改造数据从单纯的数值载体升级为能够参与计算图与学习过程的基本单元。1.1 Tensor数据结构Tensor的最底层必须依赖ndarray因为所有的计算都由它完成。因此Tensor的第一步完成ndarray封装。class Tensor: def __init__(self, data): self.data data二、基于 Tensor 的计算设计深度学习框架的本质不仅在于“存储数据”更在于“定义计算过程”。因此仅有 Tensor 这一数据结构是不够的还需要在其之上进一步设计一套基于 Tensor 的计算机制使数据能够在不同操作之间流动、组合并逐步构建为可用于学习的计算结构。2.1 Function 的结构设计本章通过引入类来封装计算过程其核心目的在于将神经网络的前向传播与反向传播过程统一为可管理的对象结构使计算过程能够被保存、追踪与复用从而为后续的自动求导与模型优化提供基础。2.1.1 前向与反向传播class Function: def __call__(self, input): x input.data y self.forward(x) output Tensor(y) return output def forward(self, x): raise NotImplementedError def backward(self, dy): raise NotImplementedError class Exp(Function): def forward(self, x): return np.exp(x) def backward(self, dy): x self.input.data return np.exp(x)2.2 单变量自动求导2.2.1 单变量计算图从上图可以看出一个变量Variable需要存储三个部分ndarray、梯度以及生成该变量的函数它们各自的作用如下ndarray用于存储变量的实际数据是前向传播过程中参与各种数学运算的对象。梯度用于存储损失函数对该变量的偏导数。生成该变量的函数用于记录生成当前变量的函数对象。当执行反向传播时可以根据该函数找到当前变量的来源并调用相应的反向传播方法计算输入变量的梯度从而将梯度沿计算图逐层传递实现自动求导。class Tensor: def __init__(self, data): if data is not None: if not isinstance(data, np.ndarray): raise TypeError(f{type(data)} is not supported!) self.data data self.grad None self.creator None def set_creator(self, func): self.creator func def backward(self): if self.grad is None: self.grad np.ones_like(self.grad) func [self.creator] while func: f func.pop() x, y f.input, f.output x.grad f.backward(y.grad) if x.creator is not None: func.append(x.creator) class Function: def __call__(self, input): x input.data y self.forward(x) output Tensor(as_ndarray(y)) output.set_creator(self) self.input input self.output output return output def forward(self, x): raise NotImplementedError def backward(self, dy): raise NotImplementedError class Exp(Function): def forward(self, x): return np.exp(x) def backward(self, dy): x self.input.data return np.exp(x) def as_ndarray(x): if np.isscalar(x): return np.array(x) return x2.3 多变量自动求导前面介绍的自动求导仅适用于单变量计算而实际的深度学习模型通常包含多个输入变量各变量之间存在复杂的计算关系。为了能够处理更加复杂的计算图并实现神经网络中多输入的梯度计算需要将自动求导机制从单变量扩展到多变量。因此本节将在单变量自动求导的基础上引入多变量自动求导使其能够支持更加复杂的数学运算。2.3.1 可变参数本章节之前涉及的函数其输入、输出都只有一个变量但有些函数需要多个变量输入例如加法运算和乘法运算。可考虑到这些情况我们需要将单变量自动求导处理成可变输入和输出。可变意味着参数的数量是可以变化的数量可以是不小于1的整数值。class Tensor: def __init__(self, data): if data is not None: if not isinstance(data, np.ndarray): raise TypeError(f{type(data)} is not supported!) self.data data self.grad None self.creator None def set_creator(self, func): self.creator func def backward(self): if self.grad is None: self.grad np.ones_like(self.grad) func [self.creator] while func: f func.pop() dys [output.grad for output in f.outputs] dxs f.backward(*dys) if not isinstance(dxs, tuple): dxs (dxs,) for x, dx in zip(f.inputs, dxs): x.grad dx if x.creator is not None: func.append(x.creator) class Function: def __call__(self, *inputs): xs [input.data for input in inputs] ys self.forward(*xs) if not isinstance(ys, tuple): ys (ys,) outputs [Tensor(as_ndarray(y)) for y in ys] if output in outputs: output.set_creator(self) self.inputs inputs self.outputs outputs return outputs if len(outputs) 1 else outputs[0] def forward(self, *x): raise NotImplementedError def backward(self, dy): raise NotImplementedError class Exp(Function): def forward(self, x): return np.exp(x) def backward(self, dy): x self.inputs[0].data return np.exp(x) class Add(Function): def forward(self, x_0, x_1): y x_0 x_1 return y def backward(self, dy): return dy, dy def as_ndarray(x): if np.isscalar(x): return np.array(x) return x2.3.2 重复变量(1) 问题描述相同变量进行加法运算时不能正确求导。下面是测试结果if __name__ __main__: x Tensor(np.array(3.0)) y add(x, x) print(y.data) y.backward() print(x.grad) 测试结果 6.0 1(2) 优化方法class Tensor: def __init__(self, data): if data is not None: if not isinstance(data, np.ndarray): raise TypeError(f{type(data)} is not supported!) self.data data self.grad None self.creator None def set_creator(self, func): self.creator func def backward(self): if self.grad is None: self.grad np.ones_like(self.grad) func [self.creator] while func: f func.pop() dys [output.grad for output in f.outputs] dxs f.backward(*dys) if not isinstance(dxs, tuple): dxs (dxs,) for x, dx in zip(f.inputs, dxs): # 重复值优化 if x.grad is None: x.grad dx else: x.grad x.grad dx if x.creator is not None: func.append(x.creator) class Function: def __call__(self, *inputs): xs [input.data for input in inputs] ys self.forward(*xs) if not isinstance(ys, tuple): ys (ys,) outputs [Tensor(as_ndarray(y)) for y in ys] if output in outputs: output.set_creator(self) self.inputs inputs self.outputs outputs return outputs if len(outputs) 1 else outputs[0] def forward(self, *x): raise NotImplementedError def backward(self, dy): raise NotImplementedError class Exp(Function): def forward(self, x): return np.exp(x) def backward(self, dy): x self.inputs[0].data return np.exp(x) class Add(Function): def forward(self, x_0, x_1): y x_0 x_1 return y def backward(self, dy): return dy, dy def as_ndarray(x): if np.isscalar(x): return np.array(x) return x2.3.3 导数重置(1) 问题描述相同变量进行不同计算时不能正确求导。下面是测试结果if __name__ __main__: # 第一个计算 x Tensor(np.array(3.0)) y add(x, x) y.backward() print(x.grad) # 第二个计算 y add(add(x, x), x) y.backward() print(x.grad) 测试结果 2 5(2) 优化方法class Tensor: def __init__(self, data): if data is not None: if not isinstance(data, np.ndarray): raise TypeError(f{type(data)} is not supported!) self.data data self.grad None self.creator None def set_creator(self, func): self.creator func def zero_grad(self): self.grad None # 进行程序前对梯度清零 def backward(self): if self.grad is None: self.grad np.ones_like(self.grad) funcs [self.creator] while funcs: f funcs.pop() dys [output.grad for output in f.outputs] dxs f.backward(*dys) if not isinstance(dxs, tuple): dxs (dxs,) for x, dx in zip(f.inputs, dxs): # 重复值优化 if x.grad is None: x.grad dx else: x.grad x.grad dx if x.creator is not None: func.append(x.creator) class Function: def __call__(self, *inputs): xs [input.data for input in inputs] ys self.forward(*xs) if not isinstance(ys, tuple): ys (ys,) outputs [Tensor(as_ndarray(y)) for y in ys] for output in outputs: output.set_creator(self) self.inputs inputs self.outputs outputs return outputs if len(outputs) 1 else outputs[0] def forward(self, *x): raise NotImplementedError def backward(self, dy): raise NotImplementedError class Exp(Function): def forward(self, x): return np.exp(x) def backward(self, dy): x self.inputs[0].data return np.exp(x) class Add(Function): def forward(self, x_0, x_1): y x_0 x_1 return y def backward(self, dy): return dy, dy def as_ndarray(x): if np.isscalar(x): return np.array(x) return x2.3.4 传播顺序前面我们处理的都是笔直得计算图然而变量与函数并不局限于这种简单得连接方式。我们可以建立复杂的连接但是不能正确求出这种连接方式的导数即不能反向传播。从图2.2可以看出变量a是后续两个分支B和C的共同输入因此其梯度应当是两条传播路径梯度的累加。然而图中的导数传播顺序与神经网络反向传播的实际过程并不一致不能正确反映梯度沿计算图逐层反向传播的特点。本章节在正向传播中实现层级设置然后在反向传播中按照从最高层级到最低层级的顺序取函数。修改完成后无论计算图多么复杂反向传播都会按照正确的顺序进行。class Tensor: def __init__(self, data): if data is not None: if not isinstance(data, np.ndarray): raise TypeError(f{type(data)} is not supported!) self.data data self.grad None self.creator None self.layers 0 # 层级 def set_creator(self, func): self.creator func self.layers func.layers 1 def zero_grad(self): self.grad None # 进行程序前对梯度清零 def backward(self): if self.grad is None: self.grad np.ones_like(self.grad) # 优先队列法未实现 funcs, seen_set [], set() def add_func(f): if f not in seen_set: funcs.append(f) seen_set.add(f) # 防止重复调用 funcs.sort(keylambda x: x.layers) add_func(self.creator) while funcs: f funcs.pop() dys [output.grad for output in f.outputs] dxs f.backward(*dys) if not isinstance(dxs, tuple): dxs (dxs,) for x, dx in zip(f.inputs, dxs): # 重复值优化 if x.grad is None: x.grad dx else: x.grad x.grad dx if x.creator is not None: add_func(x.creator) class Function: def __call__(self, *inputs): xs [input.data for input in inputs] ys self.forward(*xs) if not isinstance(ys, tuple): ys (ys,) outputs [Tensor(as_ndarray(y)) for y in ys] # 通过输入变量层级记录函数层级 self.layers max([input.layers for input in inputs]) for output in outputs: output.set_creator(self) self.inputs inputs self.outputs outputs return outputs if len(outputs) 1 else outputs[0] def forward(self, *x): raise NotImplementedError def backward(self, dy): raise NotImplementedError class Exp(Function): def forward(self, x): return np.exp(x) def backward(self, dy): x self.inputs[0].data return np.exp(x) class Add(Function): def forward(self, x_0, x_1): y x_0 x_1 return y def backward(self, dy): return dy, dy def as_ndarray(x): if np.isscalar(x): return np.array(x) return x2.4 内存管理我们写的 Python 代码本身不能直接被计算机执行需要通过一个叫Python 解释器的程序来运行它。最常用的解释器是CPython它是用 C 语言实现的也是 Python 官方默认的实现方式。简单来说Python 负责让我们方便地编写程序而 CPython 负责帮助计算机理解并执行这些程序。Cpython使用两种方式管理内存一种引用计数另一种是分代垃圾回收。2.4.1 引用计数法GC 是一种“释放怎么都无法被引用的对象的机制”。显然可以让所有对象事先记录下“有多少程序引用自己”。让各对象知道自己的“人气指数”从而让没有人气的对象自己消失这就是引用计数法它是George E. Collins于1960年钻研出来的。2.4.2 分代垃圾回收分代垃圾回收在对象中导入“年龄”的概念通过优先回收容易成为垃圾的对象提高垃圾回收效率。人们从众多程序案例中总结出一个经验“大部对象在生成后马上变成垃圾”很少有对象活得很久。分代垃圾回收利用该经验在对象中导入年龄概念经历一次GC后活下来的对象年龄为1岁。Python的分代回收是一种垃圾回收策略其核心思想是对象存活时间越长以后存活的概率越大新创建的对象更容易成为垃圾。因此Python不会每次都扫描所有对象而把对象按年龄划分提高垃圾回收效率。2.4.3 弱引用对象的弱引用不能保证对象存活当所指对象的引用只剩弱引用时 垃圾回收 可以销毁所指对象并将其内存重新用于其它用途。但是在实际销毁对象之前即使没有强引用弱引用也能返回该对象。import weakref import numpy as np class Tensor: def __init__(self, data): if data is not None: if not isinstance(data, np.ndarray): raise TypeError(f{type(data)} is not supported!) self.data data self.grad None self.creator None self.layers 0 # 层级 def set_creator(self, func): self.creator func self.layers func.layers 1 def zero_grad(self): self.grad None # 进行程序前对梯度清零 def backward(self): if self.grad is None: self.grad np.ones_like(self.grad) # 优先队列法未实现 funcs, seen_set [], set() def add_func(f): if f not in seen_set: funcs.append(f) seen_set.add(f) # 防止重复调用 funcs.sort(keylambda x: x.layers) add_func(self.creator) while funcs: f funcs.pop() dys [output().grad for output in f.outputs] dxs f.backward(*dys) if not isinstance(dxs, tuple): dxs (dxs,) for x, dx in zip(f.inputs, dxs): # 重复值优化 if x.grad is None: x.grad dx else: x.grad x.grad dx if x.creator is not None: add_func(x.creator) class Function: def __call__(self, *inputs): xs [input.data for input in inputs] ys self.forward(*xs) if not isinstance(ys, tuple): ys (ys,) outputs [Tensor(as_ndarray(y)) for y in ys] # 通过输入变量层级记录函数层级 self.layers max([input.layers for input in inputs]) for output in outputs: output.set_creator(self) self.inputs inputs self.outputs [weakref.ref(output) for output in outputs] return outputs if len(outputs) 1 else outputs[0] def forward(self, *x): raise NotImplementedError def backward(self, dy): raise NotImplementedError class Exp(Function): def forward(self, x): return np.exp(x) def backward(self, dy): x self.inputs[0].data return np.exp(x) class Add(Function): def forward(self, x_0, x_1): y x_0 x_1 return y def backward(self, dy): return dy, dy def as_ndarray(x): if np.isscalar(x): return np.array(x) return x2.4.4 清除无用导数在深度学习中只有终端变量的导数需要通过反向传播求得。中间变量的导数几乎用不到。因此我们可以增加一种机制消除中间变量的导数。import weakref import numpy as np class Tensor: def __init__(self, data): if data is not None: if not isinstance(data, np.ndarray): raise TypeError(f{type(data)} is not supported!) self.data data self.grad None self.creator None self.layers 0 # 层级 def set_creator(self, func): self.creator func self.layers func.layers 1 def zero_grad(self): self.grad None # 进行程序前对梯度清零 def backward(self, retain_gradFalse): if self.grad is None: self.grad np.ones_like(self.grad) # 优先队列法未实现 funcs, seen_set [], set() def add_func(f): if f not in seen_set: funcs.append(f) seen_set.add(f) # 防止重复调用 funcs.sort(keylambda x: x.layers) add_func(self.creator) while funcs: f funcs.pop() dys [output().grad for output in f.outputs] dxs f.backward(*dys) if not isinstance(dxs, tuple): dxs (dxs,) for x, dx in zip(f.inputs, dxs): # 重复值优化 if x.grad is None: x.grad dx else: x.grad x.grad dx if x.creator is not None: add_func(x.creator) if not retain_grad: for output in f.outputs: output().grad None class Function: def __call__(self, *inputs): xs [input.data for input in inputs] ys self.forward(*xs) if not isinstance(ys, tuple): ys (ys,) outputs [Tensor(as_ndarray(y)) for y in ys] # 通过输入变量层级记录函数层级 self.layers max([input.layers for input in inputs]) for output in outputs: output.set_creator(self) self.inputs inputs self.outputs [weakref.ref(output) for output in outputs] return outputs if len(outputs) 1 else outputs[0] def forward(self, *x): raise NotImplementedError def backward(self, dy): raise NotImplementedError class Exp(Function): def forward(self, x): return np.exp(x) def backward(self, dy): x self.inputs[0].data return np.exp(x) class Add(Function): def forward(self, x_0, x_1): y x_0 x_1 return y def backward(self, dy): return dy, dy def add(x0, x1): return Add()(x0, x1) def as_ndarray(x): if np.isscalar(x): return np.array(x) return x if __name__ __main__: x_0 Tensor(np.array(1.0)) x_1 Tensor(np.array(1.0)) t add(x_0, x_1) y add(x_0, t) y.backward() print(y.grad, t.grad) print(x_0.grad, x_1.grad)2.5 模式切换神经网络训练阶段既需要执行正向传播也需要完成反向传播以计算参数梯度而模型推理阶段仅需前向计算无需构建梯度计算图、执行反向传播因此需要临时关闭自动微分功能。本文借助 Python 的with上下文语句实现 “禁用反向传播” 模式的临时切换。import weakref import contextlib import numpy as np class Tensor: def __init__(self, data, nameNone): if data is not None: if not isinstance(data, np.ndarray): raise TypeError(f{type(data)} is not supported!) self.data data self.grad None self.name name self.creator None self.layers 0 # 层级 def set_creator(self, func): self.creator func self.layers func.layers 1 def clear_grad(self): self.grad None def backward(self, retain_gradFalse): if self.grad is None: self.grad np.ones_like(self.grad) # 优先队列法未实现 funcs, seen_set [], set() def add_func(f): if f not in seen_set: funcs.append(f) seen_set.add(f) # 防止重复调用 funcs.sort(keylambda x: x.layers) add_func(self.creator) while funcs: f funcs.pop() dys [output().grad for output in f.outputs] dxs f.backward(*dys) if not isinstance(dxs, tuple): dxs (dxs,) for x, dx in zip(f.inputs, dxs): # 重复值优化 if x.grad is None: x.grad dx else: x.grad x.grad dx if x.creator is not None: add_func(x.creator) if not retain_grad: for output in f.outputs: output().grad None class Function: def __call__(self, *inputs): xs [input.data for input in inputs] ys self.forward(*xs) if not isinstance(ys, tuple): ys (ys,) outputs [Tensor(as_ndarray(y)) for y in ys] if Config.enable_backprop: # 通过输入变量层级记录函数层级 self.layers max([input.layers for input in inputs]) for output in outputs: output.set_creator(self) self.inputs inputs self.outputs [weakref.ref(output) for output in outputs] return outputs if len(outputs) 1 else outputs[0] def forward(self, *x): raise NotImplementedError def backward(self, dy): raise NotImplementedError class Exp(Function): def forward(self, x): return np.exp(x) def backward(self, dy): x self.inputs[0].data return np.exp(x) class Add(Function): def forward(self, x_0, x_1): y x_0 x_1 return y def backward(self, dy): return dy, dy def add(x0, x1): return Add()(x0, x1) def as_ndarray(x): if np.isscalar(x): return np.array(x) return x class Config: enable_backprop True contextlib.contextmanager def using_config(name, value): old_value getattr(Config, name) setattr(Config, name, value) try: yield finally: setattr(Config, name, old_value) def zero_grad(): return using_config(enable_backprop, False) if __name__ __main__: x_0 Tensor(np.array(1.0)) x_1 Tensor(np.array(1.0)) t add(x_0, x_1) y add(x_0, t) y.backward() print(y.grad, t.grad) print(x_0.grad, x_1.grad)2.6 属性扩展为了省去频繁书写.data 的冗余代码、对齐 PyTorch 的使用习惯我们基于底层存储的 numpy 数组借助 property 扩展了 shape、ndim、size、dtype 四个只读属性同时重载__len__、__repr__魔法方法。import weakref import contextlib import numpy as np class Tensor: def __init__(self, data, nameNone): if data is not None: if not isinstance(data, np.ndarray): raise TypeError(f{type(data)} is not supported!) self.data data self.grad None self.name name self.creator None self.layers 0 # 层级 property def shape(self): return self.data.shape property def ndim(self): return self.data.ndim property def size(self): return self.data.size property def dtype(self): return self.data.dtype def __len__(self): return len(self.data) def __repr__(self): if self.data is None: return Tensor(None) return fTensor({str(self.data)}, {self.data.dtype}) def set_creator(self, func): self.creator func self.layers func.layers 1 def clear_grad(self): self.grad None def backward(self, retain_gradFalse): if self.grad is None: self.grad np.ones_like(self.grad) # 优先队列法未实现 funcs, seen_set [], set() def add_func(f): if f not in seen_set: funcs.append(f) seen_set.add(f) # 防止重复调用 funcs.sort(keylambda x: x.layers) add_func(self.creator) while funcs: f funcs.pop() dys [output().grad for output in f.outputs] dxs f.backward(*dys) if not isinstance(dxs, tuple): dxs (dxs,) for x, dx in zip(f.inputs, dxs): # 重复值优化 if x.grad is None: x.grad dx else: x.grad x.grad dx if x.creator is not None: add_func(x.creator) if not retain_grad: for output in f.outputs: output().grad None class Function: def __call__(self, *inputs): xs [input.data for input in inputs] ys self.forward(*xs) if not isinstance(ys, tuple): ys (ys,) outputs [Tensor(as_ndarray(y)) for y in ys] if Config.enable_backprop: # 通过输入变量层级记录函数层级 self.layers max([input.layers for input in inputs]) for output in outputs: output.set_creator(self) self.inputs inputs self.outputs [weakref.ref(output) for output in outputs] return outputs if len(outputs) 1 else outputs[0] def forward(self, *x): raise NotImplementedError def backward(self, dy): raise NotImplementedError class Exp(Function): def forward(self, x): return np.exp(x) def backward(self, dy): x self.inputs[0].data return np.exp(x) class Add(Function): def forward(self, x_0, x_1): y x_0 x_1 return y def backward(self, dy): return dy, dy def add(x0, x1): return Add()(x0, x1) def as_ndarray(x): if np.isscalar(x): return np.array(x) return x class Config: enable_backprop True contextlib.contextmanager def using_config(name, value): old_value getattr(Config, name) setattr(Config, name, value) try: yield finally: setattr(Config, name, old_value) def zero_grad(): return using_config(enable_backprop, False) if __name__ __main__: x_0 Tensor(np.array([[1.0, 2.9], [2.0, 9.0]])) print(x_0)2.7 运算符重载运算符重载的目的是让自定义张量类像普通数字一样直接使用数学符号运算代码贴合数学公式、简洁易读同时统一封装前向、反向传播的计算图逻辑。import weakref import contextlib import numpy as np class Tensor: def __init__(self, data, nameNone): if data is not None: if not isinstance(data, np.ndarray): raise TypeError(f{type(data)} is not supported!) self.data data self.grad None self.name name self.creator None self.layers 0 # 层级 property def shape(self): return self.data.shape property def ndim(self): return self.data.ndim property def size(self): return self.data.size property def dtype(self): return self.data.dtype def __len__(self): return len(self.data) def __repr__(self): if self.data is None: return Tensor(None) return fTensor({str(self.data)}, {self.data.dtype}) def set_creator(self, func): self.creator func self.layers func.layers 1 def clear_grad(self): self.grad None def __mul__(self, other): return mul(self, other) def __add__(self, other): return add(self, other) def __neg__(self): return neg(self) def __sub__(self, other): return sub(self, other) def __truediv__(self, other): return div(self, other) def __pow__(self, c): return power(self, c) def backward(self, retain_gradFalse): if self.grad is None: self.grad np.ones_like(self.grad) # 优先队列法未实现 funcs, seen_set [], set() def add_func(f): if f not in seen_set: funcs.append(f) seen_set.add(f) # 防止重复调用 funcs.sort(keylambda x: x.layers) add_func(self.creator) while funcs: f funcs.pop() dys [output().grad for output in f.outputs] dxs f.backward(*dys) if not isinstance(dxs, tuple): dxs (dxs,) for x, dx in zip(f.inputs, dxs): # 重复值优化 if x.grad is None: x.grad dx else: x.grad x.grad dx if x.creator is not None: add_func(x.creator) if not retain_grad: for output in f.outputs: output().grad None class Function: def __call__(self, *inputs): inputs [as_tensor(input) for input in inputs] xs [input.data for input in inputs] ys self.forward(*xs) if not isinstance(ys, tuple): ys (ys,) outputs [Tensor(as_ndarray(y)) for y in ys] if Config.enable_backprop: # 通过输入变量层级记录函数层级 self.layers max([input.layers for input in inputs]) for output in outputs: output.set_creator(self) self.inputs inputs self.outputs [weakref.ref(output) for output in outputs] return outputs if len(outputs) 1 else outputs[0] def forward(self, *x): raise NotImplementedError def backward(self, dy): raise NotImplementedError class Exp(Function): def forward(self, x): return np.exp(x) def backward(self, dy): x self.inputs[0].data return np.exp(x) class Add(Function): def forward(self, x_0, x_1): y x_0 x_1 return y def backward(self, dy): return dy, dy def add(x0, x1): return Add()(x0, x1) class Mul(Function): def forward(self, x_0, x_1): y x_0 * x_1 return y def backward(self, dy): x_0, x_1 self.inputs[0].data, self.inputs[1].data return x_1 * dy, x_0 * dy def mul(x_0, x_1): return Mul()(x_0, x_1) class Neg(Function): def forward(self, x): return -x def backward(self, dy): return -dy def neg(x): return Neg()(x) class Sub(Function): def forward(self, x_0, x_1): y x_0 - x_1 return y def backward(self, dy): return dy, -dy def sub(x_0, x_1): return Sub()(x_0, x_1) class Div(Function): def forward(self, x_0, x_1): y x_0 / x_1 return y def backward(self, dy): x0, x1 self.inputs[0].data, self.inputs[1].data return dy/x1, (-x0/x1**2)*dy def div(x0, x1): return Div()(x0, x1) class Pow(Function): def __init__(self, c): self.c c def forward(self, x): y x ** self.c return y def backward(self, dy): x self.inputs[0].data return dy * self.c * x ** (self.c - 1) def power(x, c): return Pow(c)(x) def as_ndarray(x): if np.isscalar(x): return np.array(x) return x def as_tensor(tensor): if isinstance(tensor, Tensor): return tensor else: return Tensor(np.array(tensor)) class Config: enable_backprop True contextlib.contextmanager def using_config(name, value): old_value getattr(Config, name) setattr(Config, name, value) try: yield finally: setattr(Config, name, old_value) def zero_grad(): return using_config(enable_backprop, False) if __name__ __main__: x_0 Tensor(np.array([0.1, 0.1])) x_1 2 print(x_0**2)三、Python打包3.1 概述Python打包在很长一段时间内处于混乱不堪的状态人们花了很多年才使得这一主题重新变得有组织。一切都从1998年引入的distutils包开始随后在2003年setuptools对其进行改进。这两个项目开启了一段漫长而有纠结的故事故事包括派生fork、替代项目与完全重写编写都想要彻底修复Python打包生态。但是大部分尝试都没有成功。幸运的是这种状态正在逐渐被改变。有一个Python Packaging Authority组织将秩序和组织性带回到打包生态系统中。PyPA维护的Python打包用户指南Python Packaging User Guidehttps://packaging.python.org是关于最新打包工具和最佳实践的权威信息来源。由于PyPA的参与构建发行版已经逐渐弃用egg格式二十支持使用wheel格式。在Python打包用户指南中推荐包创建与分发的工具如下。使用setuptools来定义项目并创建源代码发行版。使用wheel而不是egg来创建发行版。使用twine向PyPI上传包的发行版。3.2 项目配置组织大型应用的代码最简单的方法是将其分成几个包。这使得代码更加简单也更容易理解、维护和修改。这样也使得每个包的可复用性最大化。它们的作用就像组件一样。setup.py对于一个需要被分发的包来说其根目录包含一个setup.py脚本。它定义了distutils模块中描述的所有元数据并将其合并为标准的setup函数调用的参数。虽然distutils 是一个标准库模块但建议你使用setuptools 包来代替它对标准的distutils 做了一些改进。from setuptools import setup setup( namePyTensor, version1.0.0, descriptionA lightweight deep learning tensor framework built from scratch based on NumPy, authorLee Wen-tsao, author_emailliwenchao36163.com, packages[PyTensor], install_requires[ numpy, ], python_requires3.11, )执行打包python setup.py bdist_wheel