自動微分 | DIY 實現自己的 PyTorch
在機器學習問題中,模型訓練的核心是梯度下降:
- 計算損失函數(Loss Function)
- 計算損失對參數的導數(Gradient)
- 根據梯度更新參數
- 重復直到收斂
我們需要找到一個高效的梯度計算方法。
梯度計算方法包括:數值微分、符號微分、自動微分
數值微分的基本思想,是使用一個非常小的 $h$ 去近似計算。
但是,這種計算方法存在舍入誤差,而且計算復雜度高。
如果我們有 $n$ 個參數,需要執行 $O(n)$ 次函數計算。在參數量達百萬級的神經網絡中,這種計算成本是不可接受的。
符號微分的思路是通過鏈式法則得到一個完整的解析表達式。在 Python 中,我們可以使用 sympy 去實現。
但是,一旦函數變得復雜,表達式長度增長極快,同一個子表達式在求導過程中會被多次重復計算。這種方法都理論上很優雅,在工程上卻不可行。
自動微分將復雜函數拆解成一系列簡單的「基本運算」,並在計算過程中同步或反向計算導數。
自動微分通過記錄這些基礎運算的執行軌跡(即構建「計算圖」),並在計算過程中系統化地應用微積分中的鏈式法則##,從而極其精確且高效地計算出復雜函數對各個變量的導數。
要實現自動微分,首先要將復雜的算式轉化為「計算圖」。在這張有向無環圖(DAG)中:葉子節點代表輸入變量或模型參數。內部節點代表基本運算(如加法、乘法、$sin$ 等)。邊代表數據流向。
我們可以在計算圖上進行反向傳播,這一過程分為兩個階段:
- 前向階段: 從輸入到輸出正常計算,但會保存所有的中間變量(計算圖)。
- 反向階段: 從最終的標量輸出(如 Loss 損失值)開始,從輸出向輸入反向遍歷計算圖。它通過鏈式法則,將輸出節點對當前節點的導數(稱為伴隨值 Adjoint,用 $\bar{v}_i$ 表示)一步步往回傳遞。
工程實現上,我們可以使用一個簡單的拓撲排序,按照拓撲順序,從輸出節點反向計算梯度,完成梯度下降。
筆者自己完成了一自動微分的簡單實現,包括了一些常用的神經網絡 Loss Function 和 Optimizer,在 GitHub 上開源:https://github.com/aeilot/simplegrad
我實現的是類似 PyTorch 的動態圖。
下面簡單介紹一些核心模塊。
1
2
3
4
5
6
7
8
9
10
11
12
class Tensor:
def __init__(self, data, requires_grad: bool = False):
if not isinstance(data, np.ndarray):
data = np.array(data, dtype=np.float32)
self.data = data
self.requires_grad = requires_grad
self.grad = None
self.parents = []
self.grad_fn = None
Tensor 不僅僅是簡單的張量,還存儲了梯度、父節點等信息。通過運算符重載,我們可以隱式地構建計算圖,實現反向傳播。
對於數學操作,我們需要定義前向運算和反向運算。
1
2
3
4
5
6
7
8
9
10
class Function:
def __init__(self):
pass
def forward(self):
raise NotImplementedError
def backward(self, grad_output):
raise NotImplementedError
定義了接口,我們就可以實現各種各樣的算子。
以矩陣乘法為例。
眾所周知,矩陣乘法求導有如下公式:
我們很容易寫出代碼:
1
2
3
4
5
6
7
8
9
10
class MatMul(Function):
def __init__(self, a, b):
super().__init__()
self.a = a
self.b = b
def backward(self, grad):
grad_a = grad @ self.b.data.T
grad_b = self.a.data.T @ grad
return [grad_a, grad_b]
然後進行運算符重載:
1
2
3
4
5
6
7
8
9
10
def __matmul__(self, other):
other = ensure_tensor(other)
out = Tensor(
self.data @ other.data,
requires_grad=self.requires_grad or other.requires_grad,
)
if out.requires_grad and GradMode.enabled:
out.grad_fn = ops.MatMul(self, other)
out.parents = [self, other]
return out
運算符重載的時候,我們儲存了父節點,用戶不需要自己手動建圖,我們自動構建了計算圖。
用這種方法,我實現了加法、減法、Hadamard 積、對數、Softmax、求和、平均數等運算,可以在 GitHub 倉庫 查看(不要白嫖 給個 star)
細心的讀者可能註意到了,在前面 MatMul 的重載代碼中,有一個判斷條件 if out.requires_grad and GradMode.enabled:。
這對應了 PyTorch 中的 torch.no_grad()。在模型訓練好後進行推理(Inference)或評估時,我們不需要更新參數,也就不需要計算梯度。如果此時框架還在後臺默默「建圖」(保存父節點和 grad_fn),會白白浪費大量內存。
我們可以用一個簡單的上下文管理器(Context Manager)來實現這個開關:
1
2
3
4
5
6
7
8
9
10
11
class GradMode:
enabled = True
@contextmanager
def no_grad():
prev = GradMode.enabled
GradMode.enabled = False
try:
yield
finally:
GradMode.enabled = prev
使用時非常優雅:
1
2
3
with no_grad():
y_pred = model(x)
反向傳播的任務是從輸出節點出發,一路沿著圖的邊,把梯度精準地傳回給所有的輸入節點。
我們基於 DFS + 拓撲排序實現,保證在計算某個節點的梯度之前,依賴於它的所有「下遊」節點的梯度都已經被完全計算出來。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def backward(output):
def toposort(tensor):
visited = set()
order = []
def dfs(node):
node_id = id(node)
if node_id in visited:
return
visited.add(node_id)
for pa in node.parents:
dfs(pa)
order.append(node)
dfs(tensor)
order.reverse()
return order
order = toposort(output)
output.grad = np.ones_like(output.data)
for node in order:
if node.grad_fn is None:
continue
grads = node.grad_fn.backward(node.grad)
for p, g in zip(node.parents, grads):
if p.grad is None:
p.grad = unbroadcast(g, p.shape)
else:
p.grad += unbroadcast(g, p.shape)
當一個變量在計算圖中被多次使用時(例如 $y = x * x$),它的梯度必須是各個分支回傳梯度的總和。
這裏註意一點細節,有的時候 numpy 運算會出現 broadcasting,導致父節點和子節點 shape 不符。
在 NumPy 中,如果我們把一個形狀為 (3, 1) 的張量與一個形狀為 (1, 3) 的張量相加,NumPy 會極其聰明地將它們自動擴展(Broadcast)成 (3, 3) 並計算出結果。然而,在反向傳播時,上遊傳下來的梯度形狀是 (3, 3),但我們原本的節點形狀是 (3, 1),直接將梯度傳回去會導致形狀不匹配而報錯。
因此,在把梯度賦值給父節點之前,我們必須進行一次反廣播(Unbroadcasting),把多出來的維度通過求和操作「擠壓」回原本的形狀:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def unbroadcast(grad, target_shape):
if grad.shape == target_shape:
return grad
ndims_added = grad.ndim - len(target_shape)
for _ in range(ndims_added):
grad = grad.sum(axis=0)
for i, dim in enumerate(target_shape):
if dim == 1:
grad = grad.sum(axis=i, keepdims=True)
return grad
unbroadcast 函數的邏輯非常清晰,它分兩步解決了形狀還原的問題:
首先,如果正向計算時由於維度不對齊導致新增了維度,我們將這些多余的維度全部按軸求和並抹除。
其次,逐一對比目標形狀。如果發現原本該維度的大小為 1(意味著它被 NumPy 復製擴展了),我們就將傳回來的梯度沿著這個維度進行聚合求和,並使用
keepdims=True維持其(..., 1, ...)的骨架。
我實現了一個帶動量的隨機梯度下降優化器:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import numpy as np
class SGD:
def __init__(self, parameters, lr=0.01, momentum=0.0):
self.parameters = list(parameters)
self.lr = lr
self.momentum = momentum
self.velocities = {id(p): np.zeros_like(p.data) for p in self.parameters}
def step(self):
for p in self.parameters:
if p.grad is None:
continue
pid = id(p)
grad = p.grad
if self.momentum != 0.0:
self.velocities[pid] = self.momentum * self.velocities[pid] + grad
update_dir = self.velocities[pid]
else:
update_dir = grad
p.data -= self.lr * update_dir
def zero_grad(self):
for p in self.parameters:
p.zero_grad()
在我們的 backward 實現中,有一行代碼是 p.grad += unbroadcast(...)。我們使用的是梯度累加(+=),而不是直接賦值(=)。這意味著,如果不手動清空梯度,第二輪叠代的梯度就會疊加上第一輪的梯度,導致參數更新完全錯誤!因此,每次叠代前必須調用 optimizer.zero_grad()。
我在 SimpleGrad 中還實現了如 Adam 優化器等其他模塊,大家可以自行查看。