AI的工程基础2-反向传播

数据结构定义-标量scalar

class Scalar: def __init__(self, value, prevs=[], op=None, label='', requires_grad=True): # 节点的值 self.value = value # 节点的标识（label）和对应的运算（op），用于作图 self.label = label self.op = op # 节点的前节点，即当前节点是运算的结果，而前节点是参与运算的量 self.prevs = prevs # 是否需要计算该节点偏导数，即∂loss/∂self（loss表示最后的模型损失） self.requires_grad = requires_grad # 该节点偏导数，即∂loss/∂self self.grad = 0.0 # 如果该节点的prevs非空，存储所有的∂self/∂prev self.grad_wrt = dict() # 作图需要，实际上对计算没有作用 self.back_prop = dict() def __repr__(self): return f'Scalar(value={self.value:.2f}, grad={self.grad:.2f})' def __add__(self, other): ''' 定义加法，self + other将触发该函数 ''' if not isinstance(other, Scalar): other = Scalar(other, requires_grad=False) # output = self + other output = Scalar(self.value + other.value, [self, other], '+') output.requires_grad = self.requires_grad or other.requires_grad # 计算偏导数 ∂output/∂self = 1 output.grad_wrt[self] = 1 # 计算偏导数 ∂output/∂other = 1 output.grad_wrt[other] = 1 return output def __sub__(self, other): ''' 定义减法，self - other将触发该函数 ''' if not isinstance(other, Scalar): other = Scalar(other, requires_grad=False) # output = self - other output = Scalar(self.value - other.value, [self, other], '-') output.requires_grad = self.requires_grad or other.requires_grad # 计算偏导数 ∂output/∂self = 1 output.grad_wrt[self] = 1 # 计算偏导数 ∂output/∂other = -1 output.grad_wrt[other] = -1 return output def __mul__(self, other): ''' 定义乘法，self * other将触发该函数 ''' if not isinstance(other, Scalar): other = Scalar(other, requires_grad=False) # output = self * other output = Scalar(self.value * other.value, [self, other], '*') output.requires_grad = self.requires_grad or other.requires_grad # 计算偏导数 ∂output/∂self = other output.grad_wrt[self] = other.value # 计算偏导数 ∂output/∂other = self output.grad_wrt[other] = self.value return output def __pow__(self, other): ''' 定义乘方，self**other将触发该函数 ''' assert isinstance(other, (int, float)) # output = self ** other output = Scalar(self.value ** other, [self], f'^{other}') output.requires_grad = self.requires_grad # 计算偏导数 ∂output/∂self = other * self**(other-1) output.grad_wrt[self] = other * self.value**(other - 1) return output def sigmoid(self): ''' 定义sigmoid ''' s = 1 / (1 + math.exp(-1 * self.value)) output = Scalar(s, [self], 'sigmoid') output.requires_grad = self.requires_grad # 计算偏导数 ∂output/∂self = output * (1 - output) output.grad_wrt[self] = s * (1 - s) return output def __rsub__(self, other): ''' 定义右减法，other - self将触发该函数 ''' if not isinstance(other, Scalar): other = Scalar(other, requires_grad=False) output = Scalar(other.value - self.value, [self, other], '-') output.requires_grad = self.requires_grad or other.requires_grad # 计算偏导数 ∂output/∂self = -1 output.grad_wrt[self] = -1 # 计算偏导数 ∂output/∂other = 1 output.grad_wrt[other] = 1 return output def __radd__(self, other): ''' 定义右加法，other + self将触发该函数 ''' return self.__add__(other) def __rmul__(self, other): ''' 定义右乘法，other * self将触发该函数 ''' return self * other def backward(self, fn=None): ''' 由当前节点出发，求解以当前节点为顶点的计算图中每个节点的偏导数，i.e. ∂self/∂node 参数 ---- fn ：画图函数，如果该变量不等于None，则会返回向后传播每一步的计算的记录 返回 ---- re ：向后传播每一步的计算的记录 ''' def _topological_order(): ''' 利用深度优先算法，返回计算图的拓扑排序（topological sorting） ''' def _add_prevs(node): if node not in visited: visited.add(node) for prev in node.prevs: _add_prevs(prev) ordered.append(node) ordered, visited = [], set() _add_prevs(self) return ordered def _compute_grad_of_prevs(node): ''' 由node节点出发，向后传播 ''' # 作图需要，实际上对计算没有作用 node.back_prop = dict() # 得到当前节点在计算图中的梯度。由于一个节点可以在多个计算图中出现， # 使用cg_grad记录当前计算图的梯度 dnode = cg_grad[node] # 使用node.grad记录节点的累积梯度 node.grad += dnode for prev in node.prevs: # 由于node节点的偏导数已经计算完成，可以向后扩散（反向传播） # 需要注意的是，向后扩散到上游节点是累加关系 grad_spread = dnode * node.grad_wrt[prev] cg_grad[prev] = cg_grad.get(prev, 0.0) + grad_spread node.back_prop[prev] = node.back_prop.get(prev, 0.0) + grad_spread # 当前节点的偏导数等于1，因为∂self/∂self = 1。这是反向传播算法的起点 cg_grad = {self: 1} # 为了计算每个节点的偏导数，需要使用拓扑排序的倒序来遍历计算图 ordered = reversed(_topological_order()) re = [] for node in ordered: _compute_grad_of_prevs(node) # 作图需要，实际上对计算没有作用 if fn is not None: re.append(fn(self, 'backward')) return re

自定义运算符

在编程中，__add__是一种特殊方法（或称“魔术方法”），用于定义对象的加法操作（+）。以下是关于__add__方法的详细说明和示例。

基本语法

__add__方法需要至少一个参数（self除外），通常命名为other，表示参与加法运算的另一个对象。方法必须返回加法操作的结果。

def __add__(self, other): # 实现加法逻辑 return result

数据结构说明

Scalar类是构建自动微分计算图的核心数据结构，它封装了一个标量值及其在反向传播中所需的所有信息。其内部属性与功能如下：

核心属性

value：标量节点的数值，是前向传播的计算结果。

label：节点的标识符，主要用于可视化计算图。

op：生成该节点的运算符（如'+'、'*'、'sigmoid'），用于描述节点来源。

prevs：前驱节点列表，即参与当前运算的输入节点。它定义了计算图的依赖关系。

requires_grad：布尔值，指示该节点是否需要计算梯度（即∂loss/∂self）。

grad：累积梯度值，即损失函数对该节点的偏导数∂loss/∂self。在多次反向传播中会累加。

grad_wrt：字典，键为前驱节点，值为局部偏导数∂self/∂prev。它记录了当前节点对其每个输入的导数，用于链式法则。

back_prop：字典，用于记录反向传播过程中的中间值，主要服务于可视化，不影响数值计算。

在计算图中的角色

每个Scalar实例可视作计算图中的一个节点：

叶子节点：prevs为空，通常为输入变量或常数。

中间节点：由运算符（如__add__、__mul__）产生，其prevs指向参与运算的节点。

输出节点：计算图的最终输出，通过调用其backward()方法触发整个图的反向传播。

梯度计算与存储

为了支持反向传播，Scalar采用以下机制：

前向传播时记录局部导数：在每个运算符（如__add__）中，会计算并存储grad_wrt，即∂output/∂input。

梯度累积：grad属性用于累积从不同路径传播而来的梯度（因为一个节点可能被多个后续节点依赖）。

反向传播驱动：从输出节点开始，按照拓扑排序的逆序，利用grad_wrt和链式法则，将梯度逐节点向前传递。

这种设计使得Scalar既能保存数值，又能自动维护计算图的结构和梯度信息，是实现自动微分（Autograd）功能的基础单元。

简单的计算图

from ch07_autograd.utils import Scalar,draw_graph # 简单的计算图 a = Scalar(1.0, label='a') b = Scalar(2.0, label='b') c = a + b draw_graph(c) a = Scalar(1.0, label='a') b = Scalar(2.0, label='b') c = Scalar(4.0, label='c') d = a + b e = a * c f = d * e backward_process = f.backward(draw_graph) draw_graph(f) draw_graph(f, 'backward') # 将反向传播的过程展示出来（可能会有弹框） # for index, pic in enumerate(backward_process): # pic.view(str(index))

应用到线性回归

# 计算图膨胀 model = Linear() # 定义训练数据 x1 = Scalar(1.5, label='x1', requires_grad=False) y1 = Scalar(1.0, label='y1', requires_grad=False) x2 = Scalar(2.0, label='x2', requires_grad=False) y2 = Scalar(4.0, label='y2', requires_grad=False) loss = mse([model.error(x1, y1), model.error(x2, y2)]) draw_graph(loss) # 第一次触发方向传播 loss.backward() draw_graph(loss, 'backward') # 第二次触发方向传播 loss.backward() draw_graph(loss, 'backward')

完整训练代码修改

import torch # 固定随机种子，使得运行结果可以稳定复现 torch.manual_seed(1024) # 产生训练用的数据 x_origin = torch.linspace(100, 300, 200) # 将变量X归一化，否则梯度下降法很容易不稳定 x = (x_origin - torch.mean(x_origin)) / torch.std(x_origin) epsilon = torch.randn(x.shape) y = 10 * x + 5 + epsilon # 生成模型 model = Linear() # 定义每批次用到的数据量 batch_size = 20 learning_rate = 0.1 for t in range(20): # 选取当前批次的数据，用于训练模型 ix = (t * batch_size) % len(x) xx = x[ix: ix + batch_size] yy = y[ix: ix + batch_size] # 计算当前批次数据的损失 loss = mse([model.error(_x, _y) for _x, _y in zip(xx, yy)]) # 计算损失函数的梯度 loss.backward() # 迭代更新模型参数的估计值 model.a -= learning_rate * model.a.grad model.b -= learning_rate * model.b.grad # 将使用完的梯度清零 model.a.grad = 0.0 model.b.grad = 0.0 print(f'Step {t + 1}, Result: {model.string()}')