計(jì)算圖
- 計(jì)算圖:用來描述運(yùn)算的有向無環(huán)圖。
- 計(jì)算圖有兩個(gè)主要元素:節(jié)點(diǎn)(Node)和邊(Edge)。
- 節(jié)點(diǎn):表示數(shù)據(jù),如向量、矩陣、張量等。
- 邊:表示運(yùn)算,如加、減、乘、除、矩陣乘法等。
- 根據(jù)計(jì)算圖的搭建方式,可分為靜態(tài)圖和動(dòng)態(tài)圖。
- 靜態(tài)圖:先搭建圖,然后運(yùn)算。
- 優(yōu)點(diǎn):容易在圖上做優(yōu)化,圖的效率更高。
- 缺點(diǎn):不靈活。
- 代表:Tensorflow 1.x
- 動(dòng)態(tài)圖:運(yùn)算與搭建同時(shí)進(jìn)行。
- 優(yōu)點(diǎn):可以根據(jù)需求進(jìn)行調(diào)整,更靈活,容易debug。
- 缺點(diǎn):不容易對圖做優(yōu)化圖,計(jì)算速度慢些。
- 代表:PyTorch、Tensorflow 2.x
- 靜態(tài)圖:先搭建圖,然后運(yùn)算。
- 計(jì)算圖有兩個(gè)主要元素:節(jié)點(diǎn)(Node)和邊(Edge)。
靜態(tài)圖
手動(dòng)實(shí)現(xiàn)代碼:
import numpy as np
class OP:
def __init__(self):
self.name = self.__class__.__name__
# 靜態(tài)圖,所以O(shè)P操作后用Placeholder占位符表示
# Placeholder也是一個(gè)Tensor,初始化為0的Tensor,表示未計(jì)算的
def __call__(self, *args):
self.input = args # save for backward
# 靜態(tài)圖:操作后的輸出定義為一個(gè)Placeholder占位符
# 同時(shí)在Placeholder中存儲(chǔ)操作op
self.output = Placeholder(self)
return self.output
def compute(self):
# 遞歸調(diào)用Tensor.op.compute()計(jì)算結(jié)果
new_input = [] # 將未計(jì)算的Tensor轉(zhuǎn)換為計(jì)算后的Tensor,用于前向推理
for item in self.input:
if isinstance(item, Tensor):
if item.op is not None:
# 將Placeholder轉(zhuǎn)變?yōu)榇_定的Tensor值
item = item.op.compute()
new_input.append(item)
# 進(jìn)行前向過程
self.input = new_input # save for backward
output = self.forward(*new_input)
output.op = self # 將操作OP保存到輸出的Tensor中,反向傳播時(shí)需要知道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
# 首先調(diào)用backward(grad),根據(jù)上層傳遞過來的grad,計(jì)算"誤差項(xiàng)對OP的輸入input"的梯度
input_grads = self.backward(grad)
# input可以有多個(gè),比如AddOP操作a+b,輸入有a和b兩個(gè)。
# input也可能是一個(gè),比如e^a操作,輸入只有一個(gè)a。
# 將input_grads轉(zhuǎn)換為tuple形式
if not isinstance(input_grads, tuple):
input_grads = (input_grads, )
# 斷言: assert (condition[, "...error info..."])
# 若不滿足condition條件,則跑出AssertionError("...error info...")異常
# input_grads的形狀應(yīng)該是與input一致。這里判斷是會(huì)否一致,如果不一致拋出異常
assert len(input_grads) == len(self.input), "Number grads mismatch number input"
# 遍歷input的每個(gè)Tensor元素,遞歸調(diào)用backward計(jì)算對應(yīng)的梯度
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是上級(jí)傳遞過來的梯度,即是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只有一個(gè)。
# 誤差對SumOP的輸入的梯度等于誤差對SumOP的輸出的梯度(即傳遞過來的梯度)
# self.get_data(a)返回numpy類型數(shù)據(jù)。
return np.full_like(self.get_data(a), grad)
# MeanOP:對Tensor/Placeholder的求平均操作,可以將矩陣的值相加并除以總個(gè)數(shù)。
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只有一個(gè)。
# 誤差對SumOP的輸入的梯度等于誤差對SumOP的輸出的梯度(即傳遞過來的梯度)
# self.get_data(a)返回numpy類型數(shù)據(jù)。
d = self.get_data(a)
return np.full_like(d, grad / d.size)
# 自定義Tensor張量: data存數(shù)據(jù),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對象時(shí)觸發(fā)
# 如果self.op不為None,說明當(dāng)前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):
# 梯度累加操作: 對于同一個(gè)Tensor,若參與多次運(yùn)算,梯度應(yīng)該是累加的。
# grad是上級(jí)傳遞過來的對當(dāng)前Tensor c的梯度。
self.grad = (self.grad if self.grad else 0) + grad
# 每個(gè)Tensor c同時(shí)保存了操作OP,根據(jù)操作OP我們可以知道 Tensor c是如何計(jì)算的來的。
# 比如:c中保存了AddOP,而AddOP中又存有輸入input(即a、b),我們就知道c=a+b得到的。
# 因此,我們知道了grad_c,就可以遞歸計(jì)算grad_a和grad_b
if self.op is not None:
self.op.backward_native(grad) # 通過op,遞歸計(jì)算
# 占位符(繼承Tensor):初始化值為0,操作op為None
class Placeholder(Tensor):
def __init__(self, op=None):
super().__init__(0, op)
# SessionRun 將輸入值帶進(jìn)靜態(tài)圖中,計(jì)算復(fù)合操作OPS的輸出Tensor c的結(jié)果
# 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的動(dòng)態(tài)圖做對比:
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的自動(dòng)微分-動(dòng)態(tài)圖版本:=============")
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("計(jì)算結(jié)果是:", t)
print("a的導(dǎo)數(shù)是:", a.grad)
print("b的導(dǎo)數(shù)是:", b.grad)
t.backward() # 反向傳播
print("a的導(dǎo)數(shù)是:", a.grad.numpy())
print("b的導(dǎo)數(shù)是:", b.grad.numpy())
print("\n")
print("=============手動(dòng)實(shí)現(xiàn)的自動(dòng)微分-靜態(tài)圖版本:=============")
# 當(dāng)你執(zhí)行完表達(dá)式時(shí),就等同于構(gòu)建了一個(gè)計(jì)算圖。通過計(jì)算圖反推即可得到梯度
a = Placeholder() # shape None, H, W, C
b = Tensor(mul_value) # 可訓(xùn)練的參數(shù),比如初始化為 高斯初始化,凱明初始化,常量初始化
t = morch.sum(1 / (1 + morch.exp(-a)) @ b)
# 構(gòu)建計(jì)算圖
out = SessionRun(t, {a: value})
print("計(jì)算結(jié)果是:", out)
print("a的導(dǎo)數(shù)是:", a.grad)
print("b的導(dǎo)數(shù)是:", b.grad)
out.backward() # 反向傳播
print("a的導(dǎo)數(shù)是:", a.grad)
print("b的導(dǎo)數(shù)是:", b.grad)
對比結(jié)果如下,結(jié)果一致:

動(dòng)態(tài)圖
手動(dòng)實(shí)現(xiàn)代碼:
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
# 首先調(diào)用backward(grad),根據(jù)上層傳遞過來的grad,計(jì)算"誤差項(xiàng)對OP的輸入input"的梯度
input_grads = self.backward(grad)
# input可以有多個(gè),比如AddOP操作a+b,輸入有a和b兩個(gè)。
# input也可能是一個(gè),比如e^a操作,輸入只有一個(gè)a。
# 將input_grads轉(zhuǎn)換為tuple形式
if not isinstance(input_grads, tuple):
input_grads = (input_grads, )
# 斷言: assert (condition[, "...error info..."])
# 若不滿足condition條件,則跑出AssertionError("...error info...")異常
# input_grads的形狀應(yīng)該是與input一致。這里判斷是會(huì)否一致,如果不一致拋出異常
assert len(input_grads) == len(self.input), "Number grads mismatch number input"
# 遍歷input的每個(gè)Tensor元素,遞歸調(diào)用backward計(jì)算對應(yīng)的梯度
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是上級(jí)傳遞過來的梯度,即是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只有一個(gè)。
# 誤差對SumOP的輸入的梯度等于誤差對SumOP的輸出的梯度(即傳遞過來的梯度)
# self.get_data(a)返回numpy類型數(shù)據(jù)。
return np.full_like(self.get_data(a), grad)
# MeanOP:對Tensor/Placeholder的求平均操作,可以將矩陣的值相加并除以總個(gè)數(shù)。
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只有一個(gè)。
# 誤差對SumOP的輸入的梯度等于誤差對SumOP的輸出的梯度(即傳遞過來的梯度)
# self.get_data(a)返回numpy類型數(shù)據(jù)。
d = self.get_data(a)
return np.full_like(d, grad / d.size)
# 自定義Tensor張量: data存數(shù)據(jù),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對象時(shí)觸發(fā)
# 如果self.op不為None,說明當(dāng)前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):
# 梯度累加操作: 對于同一個(gè)Tensor,若參與多次運(yùn)算,梯度應(yīng)該是累加的。
# grad是上級(jí)傳遞過來的對當(dāng)前Tensor c的梯度。
self.grad = (self.grad if self.grad else 0) + grad
# 每個(gè)Tensor c同時(shí)保存了操作OP,根據(jù)操作OP我們可以知道 Tensor c是如何計(jì)算的來的。
# 比如:c中保存了AddOP,而AddOP中又存有輸入input(即a、b),我們就知道c=a+b得到的。
# 因此,我們知道了grad_c,就可以遞歸計(jì)算grad_a和grad_b
if self.op is not None:
self.op.backward_native(grad) # 通過op,遞歸計(jì)算
# SessionRun 將輸入值帶進(jìn)靜態(tài)圖中,計(jì)算復(fù)合操作OPS的輸出Tensor c的結(jié)果
# 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的動(dòng)態(tài)圖做對比:
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的自動(dòng)微分-動(dòng)態(tài)圖版本:=============")
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("計(jì)算結(jié)果是:", t)
print("a的導(dǎo)數(shù)是:", a.grad)
print("b的導(dǎo)數(shù)是:", b.grad)
t.backward() # 反向傳播
print("a的導(dǎo)數(shù)是:", a.grad.numpy())
print("b的導(dǎo)數(shù)是:", b.grad.numpy())
print("\n")
print("=============手動(dòng)實(shí)現(xiàn)的自動(dòng)微分-動(dòng)態(tài)圖版本:=============")
# 當(dāng)你執(zhí)行完表達(dá)式時(shí),就等同于構(gòu)建了一個(gè)計(jì)算圖。通過計(jì)算圖反推即可得到梯度
a = Tensor(value)
b = Tensor(mul_value)
t = morch.sum(1 / (1 + morch.exp(-a)) @ b)
print("計(jì)算結(jié)果是:", t)
print("a的導(dǎo)數(shù)是:", a.grad)
print("b的導(dǎo)數(shù)是:", b.grad)
t.backward() # 反向傳播
print("a的導(dǎo)數(shù)是:", a.grad)
print("b的導(dǎo)數(shù)是:", b.grad)
對比結(jié)果如下,結(jié)果一致:
