from model import GPT
m = GPT(40, 8, 12)
print('token 5 :', [round(v.data, 3) for v in m.weights['token'][5]])
print('position 0:', [round(v.data, 3) for v in m.weights['position'][0]])
for name, matrix in m.weights.items():
rows, cols = len(matrix), len(matrix[0])
print(name, rows, 'x', cols, '=', rows * cols)
print('total', len(m.parameters))
cd ~/xslm
mkdir -p lab
micro lab/embedding.py
python3 -m lab.embedding
运行结果
token 5 : [-0.081, 0.024, 0.043, -0.149, 0.005, 0.131, -0.201, -0.032]
position 0: [-0.102, -0.036, 0.089, 0.01, -0.085, -0.084, 0.267, 0.114]
token 40 x 8 = 320
position 12 x 8 = 96
query 8 x 8 = 64
key 8 x 8 = 64
value 8 x 8 = 64
project 8 x 8 = 64
expand 16 x 8 = 128
contract 8 x 16 = 128
head 40 x 8 = 320
total 1248
XSLM 的 linear 没有相加的(bias),norm 没有可学习的缩放和平移值(γ、β)。 的 nn.Linear、nn.LayerNorm 两者都有。
动手验证 lab/linear.py
from engine import linear, norm
x = [1.0, 2.0]
W = [[1.0, 0.0], [0.5, 0.5], [-1.0, 1.0]]
print('linear:', linear(x, W))
print('norm :', [round(v, 3) for v in norm([2.0, 4.0, 6.0])])
print('relu :', [max(0.0, v) for v in [-2.0, 0.5, 3.0]])
cd ~/xslm
mkdir -p lab
micro lab/linear.py
python3 -m lab.linear
例如在 "the cat sat because it was tired" 中,it 位置的注意力可以学会给 cat 位置很大的权重。真实模型有多个头(GPT-2 Small 是 12 个),每个头看不同的关系(指代词、紧邻的前一个词、格式标记等)。
动手验证 lab/attention.py
from engine import Value, softmax
keys = [[1, 0], [0, 1], [1, 1]]
values = [[10, 0], [0, 10], [5, 5]]
query = [2, 0]
scores = [sum(q * k for q, k in zip(query, key)) / 2 ** 0.5 for key in keys]
weights = softmax([Value(s) for s in scores])
mixed = [sum(w.data * v[i] for w, v in zip(weights, values)) for i in range(2)]
print('scores ', [round(s, 3) for s in scores])
print('weights', [round(w.data, 3) for w in weights])
print('mixed ', [round(m, 3) for m in mixed])
cd ~/xslm
mkdir -p lab
micro lab/attention.py
python3 -m lab.attention
from model import GPT
m = GPT(10, 8, 12)
short = m.forward([1, 2])
long = m.forward([1, 2, 3])
same = [[v.data for v in r] for r in short] == [[v.data for v in r] for r in long[:2]]
print('first two positions unchanged:', same)
print('scores at position 3:', [round(v.data, 3) for v in long[2][:4]], '...')
cd ~/xslm
mkdir -p lab
micro lab/causal.py
python3 -m lab.causal
运行结果
first two positions unchanged: True
scores at position 3: [0.159, 0.011, -0.239, 0.428] ...
问题是 exp 增长得非常快。exp(1000) 超出计算机能表示的范围,会出现 OverflowError(实验中的 math range error)。所以 engine.py 的 softmax 先从所有分数中减去最大的分数(peak)。所有分数减去同一个数,概率不变,而最大值变成 exp(0) = 1,不会溢出。[1000, 999] 也能正确得到 [0.731, 0.269]。
softmax 用在两处:求注意力权重时,以及把输出层分数变成下一个词元概率时。
动手验证 lab/softmax.py
import math
from engine import Value, softmax
probs = softmax([Value(s) for s in [2.0, 1.0, 0.1]])
print([round(p.data, 3) for p in probs], round(sum(p.data for p in probs), 3))
try:
print([math.exp(s) for s in [1000.0, 999.0]])
except OverflowError as error:
print('without max:', error)
print('with max:', [round(p.data, 3) for p in softmax([Value(1000.0), Value(999.0)])])
cd ~/xslm
mkdir -p lab
micro lab/softmax.py
python3 -m lab.softmax
运行结果
[0.659, 0.242, 0.099] 1.0
without max: math range error
with max: [0.731, 0.269]
from engine import Value
x = Value(3)
y = x * x + 2 * x
y.backward()
f = lambda v: v * v + 2 * v
h = 1e-6
print('autograd:', x.grad)
print('numeric :', round((f(3 + h) - f(3 - h)) / (2 * h), 6))
a, b = Value(2), Value(5)
c = a * b + a
c.backward()
print('dc/da =', a.grad, ' dc/db =', b.grad)
cd ~/xslm
mkdir -p lab
micro lab/grad.py
python3 -m lab.grad
import math
class Value:
def __init__(self, data, parents=()):
self.data = float(data)
self.grad = 0.0
self.parents = parents
def __add__(self, other):
other = other if isinstance(other, Value) else Value(other)
return Value(self.data + other.data, ((self, 1), (other, 1)))
def __mul__(self, other):
other = other if isinstance(other, Value) else Value(other)
return Value(self.data * other.data,
((self, other.data), (other, self.data)))
def __pow__(self, power):
return Value(self.data ** power,
((self, power * self.data ** (power - 1)),))
def __neg__(self):
return self * -1
def __sub__(self, other):
return self + (-other)
def __truediv__(self, other):
return self * (other ** -1)
__radd__ = __add__
__rmul__ = __mul__
def exp(self):
result = math.exp(self.data)
return Value(result, ((self, result),))
def log(self):
return Value(math.log(self.data), ((self, 1 / self.data),))
def relu(self):
return Value(max(0, self.data), ((self, float(self.data > 0)),))
def backward(self):
order, seen, stack = [], set(), [(self, False)]
while stack:
node, expanded = stack.pop()
if expanded:
order.append(node)
elif node not in seen:
seen.add(node)
stack.append((node, True))
stack.extend((parent, False) for parent, _ in node.parents)
for node in order:
node.grad = 0.0
self.grad = 1.0
for node in reversed(order):
for parent, derivative in node.parents:
parent.grad += derivative * node.grad
def softmax(values):
peak = max(v.data for v in values)
exps = [(v - peak).exp() for v in values]
total = sum(exps)
return [v / total for v in exps]
def linear(x, weights):
return [sum(a * b for a, b in zip(x, row)) for row in weights]
def norm(x):
mean = sum(x) / len(x)
variance = sum((v - mean) ** 2 for v in x) / len(x)
return [(v - mean) * (variance + 1e-5) ** -0.5 for v in x]
if __name__ == '__main__':
x = Value(3)
y = x * x + 2 * x
y.backward()
print('y =', y.data, 'dy/dx =', x.grad)
assert y.data == 15 and x.grad == 8
model.py需要亲手编写的完整源码
先读shapes再读forward。全部权重参与学习,state与load保存数字。
import random
from engine import Value, linear, norm, softmax
class GPT:
def __init__(self, vocab_size, width, context, seed=7):
self.width, self.context = width, context
rng = random.Random(seed)
shapes = {
'token': (vocab_size, width), 'position': (context, width),
'query': (width, width), 'key': (width, width),
'value': (width, width), 'project': (width, width),
'expand': (2 * width, width), 'contract': (width, 2 * width),
'head': (vocab_size, width),
}
self.weights = {
name: [[Value(rng.gauss(0, 0.1)) for _ in range(cols)]
for _ in range(rows)] for name, (rows, cols) in shapes.items()
}
self.parameters = [v for matrix in self.weights.values()
for row in matrix for v in row]
def forward(self, ids):
assert 0 < len(ids) <= self.context
w, keys, values, output = self.weights, [], [], []
for position, token in enumerate(ids):
x = [a + b for a, b in zip(w['token'][token], w['position'][position])]
h = norm(x)
q = linear(h, w['query'])
keys.append(linear(h, w['key']))
values.append(linear(h, w['value']))
scores = [sum(a * b for a, b in zip(q, k)) / self.width ** 0.5
for k in keys]
attention = softmax(scores)
mixed = [sum(p * v[i] for p, v in zip(attention, values))
for i in range(self.width)]
x = [a + b for a, b in zip(x, linear(mixed, w['project']))]
hidden = [v.relu() for v in linear(norm(x), w['expand'])]
x = [a + b for a, b in zip(x, linear(hidden, w['contract']))]
output.append(linear(norm(x), w['head']))
return output
def loss(self, inputs, targets):
logits = self.forward(inputs)
losses = [-softmax(row)[target].log()
for row, target in zip(logits, targets)]
return sum(losses) / len(losses)
def state(self):
return {name: [[v.data for v in row] for row in matrix]
for name, matrix in self.weights.items()}
def load(self, state):
assert set(state) == set(self.weights), 'checkpoint keys'
for name, matrix in self.weights.items():
assert len(state[name]) == len(matrix), 'checkpoint rows'
for row, saved in zip(matrix, state[name]):
assert len(saved) == len(row), 'checkpoint columns'
for value, number in zip(row, saved):
value.data = float(number)
def backward(self): order, seen, stack = [], set(), [(self, False)] while stack: node, expanded = stack.pop() if expanded: order.append(node) elif node not in seen: seen.add(node) stack.append((node, True)) stack.extend((parent, False) for parent, _ in node.parents) for node in order: node.grad = 0.0 self.grad = 1.0 for node in reversed(order): for parent, derivative in node.parents: parent.grad += derivative * node.grad
分两步。
建立顺序(第46~54行):用 stack 深度优先遍历图,生成 order。节点以 (节点, False) 第一次被取出时,在 seen 中标记,重新压入 (节点, True),再压入各父节点。所有父节点处理完后,(节点, True) 被取出时才放进 order,所以在 order 中输入总是排在结果前面(拓扑排序)。
传播梯度(第55~60行):把图中所有节点的 grad 清为 0,损失自身设为 1.0(dL/dL = 1)。倒着走 order(从结果到输入),给每个父节点加上"局部导数 × 我的 grad"。因为倒着走,一个节点的 grad 从所有用处收集完毕之后,才会传给它的父节点。
engine.py softmax——不会溢出的概率
def softmax(values): peak = max(v.data for v in values) exps = [(v - peak).exp() for v in values] total = sum(exps) return [v / total for v in exps]
def linear(x, weights): return [sum(a * b for a, b in zip(x, row)) for row in weights]def norm(x): mean = sum(x) / len(x) variance = sum((v - mean) ** 2 for v in x) / len(x) return [(v - mean) * (variance + 1e-5) ** -0.5 for v in x]
linear 对 weights 的每一行(row),与 x 同一格相乘再相加,输出结果。所以 weights 是(输出格数)行 × (输入格数)格,结果的格数等于行数。PyTorch 的 nn.Linear 也按(输出, 输入)的形状保存权重。norm 求出平均 mean 和方差 variance 后,换成(值 − 平均) ÷ √(方差 + 0.00001)。取 -0.5 次方等于除以平方根。
engine.py 自我检查
if __name__ == '__main__': x = Value(3) y = x * x + 2 * x y.backward() print('y =', y.data, 'dy/dx =', x.grad) assert y.data == 15 and x.grad == 8
def forward(self, ids): assert 0 < len(ids) <= self.context w, keys, values, output = self.weights, [], [], [] for position, token in enumerate(ids): x = [a + b for a, b in zip(w['token'][token], w['position'][position])] h = norm(x)
q = linear(h, w['query']) keys.append(linear(h, w['key'])) values.append(linear(h, w['value'])) scores = [sum(a * b for a, b in zip(q, k)) / self.width ** 0.5 for k in keys] attention = softmax(scores) mixed = [sum(p * v[i] for p, v in zip(attention, values)) for i in range(self.width)] x = [a + b for a, b in zip(x, linear(mixed, w['project']))]
def state(self): return {name: [[v.data for v in row] for row in matrix] for name, matrix in self.weights.items()} def load(self, state): assert set(state) == set(self.weights), 'checkpoint keys' for name, matrix in self.weights.items(): assert len(state[name]) == len(matrix), 'checkpoint rows' for row, saved in zip(matrix, state[name]): assert len(saved) == len(row), 'checkpoint columns' for value, number in zip(row, saved): value.data = float(number)
state 返回一个形状相同、只取出普通数(.data)而不是 Value 的字典。因为 Value 对象带有计算图的连接,无法存成 JSON。load 反过来,把保存的数字放进 Value 的 data。在此之前用 assert 确认矩阵名称、行数、格数与当前模型相同,避免悄悄读错不同大小的检查点。