计算图
- 计算图：用来描述运算的有向无环图。
- 计算图有两个主要元素：节点(Node)和边(Edge)。
- 节点：表示数据，如向量、矩阵、张量等。
- 边：表示运算，如加、减、乘、除、矩阵乘法等。
- 根据计算图的搭建方式，可分为静态图和动态图。
- 静态图：先搭建图，然后运算。
- 优点：容易在图上做优化，图的效率更高。
- 缺点：不灵活。
- 代表：Tensorflow 1.x
- 动态图：运算与搭建同时进行。
- 优点：可以根据需求进行调整，更灵活，容易debug。
- 缺点：不容易对图做优化图，计算速度慢些。
- 代表：PyTorch、Tensorflow 2.x
- 静态图：先搭建图，然后运算。
- 计算图有两个主要元素：节点(Node)和边(Edge)。
静态图
手动实现代码：
import numpy as np
class OP:
def __init__(self):
self.name = self.__class__.__name__
# 静态图，所以OP操作后用Placeholder占位符表示
# Placeholder也是一个Tensor，初始化为0的Tensor，表示未计算的
def __call__(self, *args):
self.input = args # save for backward
# 静态图：操作后的输出定义为一个Placeholder占位符
# 同时在Placeholder中存储操作op
self.output = Placeholder(self)
return self.output
def compute(self):
# 递归调用Tensor.op.compute()计算结果
new_input = [] # 将未计算的Tensor转换为计算后的Tensor，用于前向推理
for item in self.input:
if isinstance(item, Tensor):
if item.op is not None:
# 将Placeholder转变为确定的Tensor值
item = item.op.compute()
new_input.append(item)
# 进行前向过程
self.input = new_input # save for backward
output = self.forward(*new_input)
output.op = self # 将操作OP保存到输出的Tensor中，反向传播时需要知道op操作
return output
def forward(self, *args):
raise NotImplementedError()
def backward(self, *args):
raise NotImplementedError()
def backward_native(self, grad):
# 上层传递过来的传递归来的梯度，即为对OP的输出output的梯度。这里保存起来。
self.output.grad = grad
# 首先调用backward(grad)，根据上层传递过来的grad，计算"误差项对OP的输入input"的梯度
input_grads = self.backward(grad)
# input可以有多个，比如AddOP操作a+b，输入有a和b两个。
# input也可能是一个，比如e^a操作，输入只有一个a。
# 将input_grads转换为tuple形式
if not isinstance(input_grads, tuple):
input_grads = (input_grads, )
# 断言: assert (condition[, "...error info..."])
# 若不满足condition条件，则跑出AssertionError("...error info...")异常
# input_grads的形状应该是与input一致。这里判断是会否一致，如果不一致抛出异常
assert len(input_grads) == len(self.input), "Number grads mismatch number input"
# 遍历input的每个Tensor元素，递归调用backward计算对应的梯度
for input_grad, ip in zip(input_grads, self.input):
if isinstance(ip, Tensor):
ip.backward(input_grad)
# 获取item的值：判断item是否是Tensor，如果是返回item.data，否则直接返回item。
def get_data(self, item):
if isinstance(item, Tensor):
return item.data
else:
return item
# Add操作：a + b
class AddOP(OP):
def __init__(self):
super().__init__()
def forward(self, a, b):
return Tensor(self.get_data(a) + self.get_data(b))
def backward(self, grad):
# 对于a+b=c操作，grad_a=1*grad_c =grad_c。同理grad_b=grad_c
# grad是上级传递过来的梯度，即是grad_c
return grad, grad
# Sub操作：a - b
class SubOP(OP):
def __init__(self):
super().__init__()
def forward(self, a, b):
return Tensor(self.get_data(a) - self.get_data(b))
def backward(self, grad):
return grad, -1 * grad
# Mul操作：a * b
class MulOP(OP):
def __init__(self):
super().__init__()
def forward(self, a, b):
return Tensor(self.get_data(a) * self.get_data(b))
def backward(self, grad):
a, b = self.input
return grad * self.get_data(b), grad * self.get_data(a)
# Div操作：a * b
class DivOP(OP):
def __init__(self):
super().__init__()
def forward(self, a, b):
return Tensor(self.get_data(a) / self.get_data(b))
def backward(self, grad):
a, b = self.input
return grad / self.get_data(b), grad * self.get_data(a) / (self.get_data(b) ** 2) * (-1)
# Exp操作：e^a
class ExpOP(OP):
def __init__(self):
super().__init__()
def forward(self, a):
return Tensor(np.exp(self.get_data(a)))
def backward(self, grad):
a = self.input[0]
return grad * np.exp(self.get_data(a))
# Log操作：loga
class LogOP(OP):
def __init__(self):
super().__init__()
def forward(self, a):
return Tensor(np.log(self.get_data(a)))
def backward(self, grad):
a = self.input[0]
return grad / self.get_data(a)
# 矩阵乘法操作：a @ b
class MatMulOP(OP):
def __init__(self):
super().__init__()
def forward(self, a, b):
return Tensor(self.get_data(a) @ self.get_data(b))
def backward(self, grad):
a, b = self.input
return grad @ self.get_data(b).T, self.get_data(a).T @ grad
# SumOP：对Tensor/Placeholder的求和操作，可以将矩阵的值相加。
class SumOP(OP):
def __init__(self):
super().__init__()
def forward(self, a):
return Tensor(np.sum(self.get_data(a)))
def backward(self, grad):
a = self.input[0] # SumOP求和操作的输入input只有一个。
# 误差对SumOP的输入的梯度等于误差对SumOP的输出的梯度(即传递过来的梯度)
# self.get_data(a)返回numpy类型数据。
return np.full_like(self.get_data(a), grad)
# MeanOP：对Tensor/Placeholder的求平均操作，可以将矩阵的值相加并除以总个数。
class MeanOP(OP):
def __init__(self):
super().__init__()
def forward(self, a):
return Tensor(np.mean(self.get_data(a)))
def backward(self, grad):
a = self.input[0] # SumOP求和操作的输入input只有一个。
# 误差对SumOP的输入的梯度等于误差对SumOP的输出的梯度(即传递过来的梯度)
# self.get_data(a)返回numpy类型数据。
d = self.get_data(a)
return np.full_like(d, grad / d.size)
# 自定义Tensor张量: data存数据，grad存梯度，op存操作。
class Tensor:
def __init__(self, data, op=None):
self.data = data
self.grad = None
self.op = op
def __radd__(self, other):
return AddOP()(other, self)
def __add__(self, other):
return AddOP()(self, other)
def __rsub__(self, other):
return SubOP()(other, self)
def __sub__(self, other):
return SubOP()(self, other)
def __rmul__(self, other):
return MulOP()(other, self)
def __mul__(self, other):
return MulOP()(self, other)
def __rtruediv__(self, other):
return DivOP()(other, self)
def __truediv__(self, other):
return DivOP()(self, other)
def __neg__(self):
return MulOP()(self, -1)
def __matmul__(self, other):
return MatMulOP()(self, other)
def __repr__(self): # print对象时触发
# 如果self.op不为None，说明当前Tensor是通过op操作得到的。
if self.op is not None:
return f"tensor({self.data}, grad_fn=<{self.op.name}>)"
else:
return f"{self.data}"
# Tensor的反向传播
def backward(self, grad=1):
# 梯度累加操作: 对于同一个Tensor，若参与多次运算，梯度应该是累加的。
# grad是上级传递过来的对当前Tensor c的梯度。
self.grad = (self.grad if self.grad else 0) + grad
# 每个Tensor c同时保存了操作OP，根据操作OP我们可以知道 Tensor c是如何计算的来的。
# 比如：c中保存了AddOP，而AddOP中又存有输入input(即a、b)，我们就知道c=a+b得到的。
# 因此，我们知道了grad_c，就可以递归计算grad_a和grad_b
if self.op is not None:
self.op.backward_native(grad) # 通过op，递归计算
# 占位符(继承Tensor)：初始化值为0，操作op为None
class Placeholder(Tensor):
def __init__(self, op=None):
super().__init__(0, op)
# SessionRun 将输入值带进静态图中，计算复合操作OPS的输出Tensor c的结果
# feed_dict将输入值以字典的形式传递，形如: {a: 1, b: 2}
def SessionRun(var, feed_dict):
for key in feed_dict:
key.data = feed_dict[key]
return var.op.compute()
# 模拟包的形式
class morch:
@staticmethod
def exp(value):
return ExpOP()(value)
@staticmethod
def log(value):
return LogOP()(value)
@staticmethod
def sum(value):
return SumOP()(value)
@staticmethod
def mean(value):
return MeanOP()(value)
与PyTorch的动态图做对比：
import numpy as np
import torch
torch.set_printoptions(precision=10)
np.set_printoptions(precision=10)
value = np.arange(9).reshape(3, 3).astype(np.float32)
mul_value = np.linspace(0, 1, 9).reshape(3, 3)
print("=============基于PyTorch的自动微分-动态图版本：=============")
a = torch.tensor(value, dtype=torch.float32, requires_grad=True)
b = torch.tensor(mul_value, dtype=torch.float32, requires_grad=True)
t = torch.sum(1 / (1 + torch.exp(-a)) @ b)
print("计算结果是：", t)
print("a的导数是：", a.grad)
print("b的导数是：", b.grad)
t.backward() # 反向传播
print("a的导数是：", a.grad.numpy())
print("b的导数是：", b.grad.numpy())
print("\n")
print("=============手动实现的自动微分-静态图版本：=============")
# 当你执行完表达式时，就等同于构建了一个计算图。通过计算图反推即可得到梯度
a = Placeholder() # shape None, H, W, C
b = Tensor(mul_value) # 可训练的参数，比如初始化为 高斯初始化，凯明初始化，常量初始化
t = morch.sum(1 / (1 + morch.exp(-a)) @ b)
# 构建计算图
out = SessionRun(t, {a: value})
print("计算结果是：", out)
print("a的导数是：", a.grad)
print("b的导数是：", b.grad)
out.backward() # 反向传播
print("a的导数是：", a.grad)
print("b的导数是：", b.grad)
对比结果如下，结果一致：
动态图
手动实现代码：
import numpy as np
class OP:
def __init__(self):
self.name = self.__class__.__name__
def __call__(self, *args):
self.input = args # save for backward
self.output = self.forward(*args)
self.output.op = self
return self.output
def forward(self, *args):
raise NotImplementedError()
def backward(self, *args):
raise NotImplementedError()
def backward_native(self, grad):
# 上层传递过来的传递归来的梯度，即为对OP的输出output的梯度。这里保存起来。
self.output.grad = grad
# 首先调用backward(grad)，根据上层传递过来的grad，计算"误差项对OP的输入input"的梯度
input_grads = self.backward(grad)
# input可以有多个，比如AddOP操作a+b，输入有a和b两个。
# input也可能是一个，比如e^a操作，输入只有一个a。
# 将input_grads转换为tuple形式
if not isinstance(input_grads, tuple):
input_grads = (input_grads, )
# 断言: assert (condition[, "...error info..."])
# 若不满足condition条件，则跑出AssertionError("...error info...")异常
# input_grads的形状应该是与input一致。这里判断是会否一致，如果不一致抛出异常
assert len(input_grads) == len(self.input), "Number grads mismatch number input"
# 遍历input的每个Tensor元素，递归调用backward计算对应的梯度
for input_grad, ip in zip(input_grads, self.input):
if isinstance(ip, Tensor):
ip.backward(input_grad)
# 获取item的值：判断item是否是Tensor，如果是返回item.data，否则直接返回item。
def get_data(self, item):
if isinstance(item, Tensor):
return item.data
else:
return item
# Add操作：a + b
class AddOP(OP):
def __init__(self):
super().__init__()
def forward(self, a, b):
return Tensor(self.get_data(a) + self.get_data(b))
def backward(self, grad):
# 对于a+b=c操作，grad_a=1*grad_c =grad_c。同理grad_b=grad_c
# grad是上级传递过来的梯度，即是grad_c
return grad, grad
# Sub操作：a - b
class SubOP(OP):
def __init__(self):
super().__init__()
def forward(self, a, b):
return Tensor(self.get_data(a) - self.get_data(b))
def backward(self, grad):
return grad, -1 * grad
# Mul操作：a * b
class MulOP(OP):
def __init__(self):
super().__init__()
def forward(self, a, b):
return Tensor(self.get_data(a) * self.get_data(b))
def backward(self, grad):
a, b = self.input
return grad * self.get_data(b), grad * self.get_data(a)
# Div操作：a * b
class DivOP(OP):
def __init__(self):
super().__init__()
def forward(self, a, b):
return Tensor(self.get_data(a) / self.get_data(b))
def backward(self, grad):
a, b = self.input
return grad / self.get_data(b), grad * self.get_data(a) / (self.get_data(b) ** 2) * (-1)
# Exp操作：e^a
class ExpOP(OP):
def __init__(self):
super().__init__()
def forward(self, a):
return Tensor(np.exp(self.get_data(a)))
def backward(self, grad):
a = self.input[0]
return grad * np.exp(self.get_data(a))
# Log操作：loga
class LogOP(OP):
def __init__(self):
super().__init__()
def forward(self, a):
return Tensor(np.log(self.get_data(a)))
def backward(self, grad):
a = self.input[0]
return grad / self.get_data(a)
# 矩阵乘法操作：a @ b
class MatMulOP(OP):
def __init__(self):
super().__init__()
def forward(self, a, b):
return Tensor(self.get_data(a) @ self.get_data(b))
def backward(self, grad):
a, b = self.input
return grad @ self.get_data(b).T, self.get_data(a).T @ grad
# SumOP：对Tensor/Placeholder的求和操作，可以将矩阵的值相加。
class SumOP(OP):
def __init__(self):
super().__init__()
def forward(self, a):
return Tensor(np.sum(self.get_data(a)))
def backward(self, grad):
a = self.input[0] # SumOP求和操作的输入input只有一个。
# 误差对SumOP的输入的梯度等于误差对SumOP的输出的梯度(即传递过来的梯度)
# self.get_data(a)返回numpy类型数据。
return np.full_like(self.get_data(a), grad)
# MeanOP：对Tensor/Placeholder的求平均操作，可以将矩阵的值相加并除以总个数。
class MeanOP(OP):
def __init__(self):
super().__init__()
def forward(self, a):
return Tensor(np.mean(self.get_data(a)))
def backward(self, grad):
a = self.input[0] # SumOP求和操作的输入input只有一个。
# 误差对SumOP的输入的梯度等于误差对SumOP的输出的梯度(即传递过来的梯度)
# self.get_data(a)返回numpy类型数据。
d = self.get_data(a)
return np.full_like(d, grad / d.size)
# 自定义Tensor张量: data存数据，grad存梯度，op存操作。
class Tensor:
def __init__(self, data, op=None):
self.data = data
self.grad = None
self.op = op
def __radd__(self, other):
return AddOP()(other, self)
def __add__(self, other):
return AddOP()(self, other)
def __rsub__(self, other):
return SubOP()(other, self)
def __sub__(self, other):
return SubOP()(self, other)
def __rmul__(self, other):
return MulOP()(other, self)
def __mul__(self, other):
return MulOP()(self, other)
def __rtruediv__(self, other):
return DivOP()(other, self)
def __truediv__(self, other):
return DivOP()(self, other)
def __neg__(self):
return MulOP()(self, -1)
def __matmul__(self, other):
return MatMulOP()(self, other)
def __repr__(self): # print对象时触发
# 如果self.op不为None，说明当前Tensor是通过op操作得到的。
if self.op is not None:
return f"tensor({self.data}, grad_fn=<{self.op.name}>)"
else:
return f"{self.data}"
# Tensor的反向传播
def backward(self, grad=1):
# 梯度累加操作: 对于同一个Tensor，若参与多次运算，梯度应该是累加的。
# grad是上级传递过来的对当前Tensor c的梯度。
self.grad = (self.grad if self.grad else 0) + grad
# 每个Tensor c同时保存了操作OP，根据操作OP我们可以知道 Tensor c是如何计算的来的。
# 比如：c中保存了AddOP，而AddOP中又存有输入input(即a、b)，我们就知道c=a+b得到的。
# 因此，我们知道了grad_c，就可以递归计算grad_a和grad_b
if self.op is not None:
self.op.backward_native(grad) # 通过op，递归计算
# SessionRun 将输入值带进静态图中，计算复合操作OPS的输出Tensor c的结果
# feed_dict将输入值以字典的形式传递，形如: {a: 1, b: 2}
def SessionRun(var, feed_dict):
for key in feed_dict:
key.data = feed_dict[key]
return var.op.compute()
# 模拟包的形式
class morch:
@staticmethod
def exp(value):
return ExpOP()(value)
@staticmethod
def log(value):
return LogOP()(value)
@staticmethod
def sum(value):
return SumOP()(value)
@staticmethod
def mean(value):
return MeanOP()(value)
与PyTorch的动态图做对比：
import numpy as np
import torch
torch.set_printoptions(precision=10)
np.set_printoptions(precision=10)
value = np.arange(9).reshape(3, 3).astype(np.float32)
mul_value = np.linspace(0, 1, 9).reshape(3, 3)
print("=============基于PyTorch的自动微分-动态图版本：=============")
a = torch.tensor(value, dtype=torch.float32, requires_grad=True)
b = torch.tensor(mul_value, dtype=torch.float32, requires_grad=True)
t = torch.sum(1 / (1 + torch.exp(-a)) @ b)
print("计算结果是：", t)
print("a的导数是：", a.grad)
print("b的导数是：", b.grad)
t.backward() # 反向传播
print("a的导数是：", a.grad.numpy())
print("b的导数是：", b.grad.numpy())
print("\n")
print("=============手动实现的自动微分-动态图版本：=============")
# 当你执行完表达式时，就等同于构建了一个计算图。通过计算图反推即可得到梯度
a = Tensor(value)
b = Tensor(mul_value)
t = morch.sum(1 / (1 + morch.exp(-a)) @ b)
print("计算结果是：", t)
print("a的导数是：", a.grad)
print("b的导数是：", b.grad)
t.backward() # 反向传播
print("a的导数是：", a.grad)
print("b的导数是：", b.grad)
对比结果如下，结果一致：
