数据结构定义-标量scalarclass Scalar: def __init__(self, value, prevs[], opNone, label, requires_gradTrue): # 节点的值 self.value value # 节点的标识label和对应的运算op用于作图 self.label label self.op op # 节点的前节点即当前节点是运算的结果而前节点是参与运算的量 self.prevs prevs # 是否需要计算该节点偏导数即∂loss/∂selfloss表示最后的模型损失 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 fScalar(value{self.value:.2f}, grad{self.grad:.2f}) def __add__(self, other): 定义加法self other将触发该函数 if not isinstance(other, Scalar): other Scalar(other, requires_gradFalse) # 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_gradFalse) # 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_gradFalse) # 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_gradFalse) 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, fnNone): 由当前节点出发求解以当前节点为顶点的计算图中每个节点的偏导数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, labela) b Scalar(2.0, labelb) c a b draw_graph(c) a Scalar(1.0, labela) b Scalar(2.0, labelb) c Scalar(4.0, labelc) 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, labelx1, requires_gradFalse) y1 Scalar(1.0, labely1, requires_gradFalse) x2 Scalar(2.0, labelx2, requires_gradFalse) y2 Scalar(4.0, labely2, requires_gradFalse) 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(fStep {t 1}, Result: {model.string()})