上一章，我们用独热编码把字符变成了向量，再用词袋把一段文字压缩进一个向量（向量和），从而第一次让网络模型读起了莎士比亚。
但是用独热编码表示语言有一个明显的短板：太占用空间。每个独热编码的长度和词表长度一致，在真正的 LLM 模型中，一个独热编码向量的长度经常可以达到几万、甚至十几万。
每个独热编码向量只使用一个位置（词袋也只使用极少数位置），其他都是 0，称为稀疏向量（Sparse Vector）。这种表达方式用于表示分类问题的标签是有意义的，但是用于表示文字特征却是巨大的浪费，也会造成权重参数“爆表”。
那么是否可以参考数据压缩技术，把每个独热编码向量压缩进一个相对较短的向量？
比如：如果我们的词表包括 10,000 个单词，那么每个单词的独热编码向量的长度也是 10,000。如果我们想把它压缩到 128：
首先，我们可以使用线性层来完成这个任务。线性层的输入特征维度
in_size设置为 10,000，输出特征维度out_size为 128，权重矩阵的形状就是(128, 10000)，每行代表一个神经元，产生一个输出值；再进一步的话，每个输入数据都是一个独热编码，总是只有一个位置是 1。这相当于，计算结果就是权重矩阵中对应位置的那一列。而如果输入数据是一段文字的独热编码向量，结果就是对应位置的那几列；
所以，我们事实上可以不要独热编码。直接使用索引编码作为索引，从权重矩阵中截取对应位置的一列（或者几列），而无需任何矢量计算。我们称这种方法为嵌入（Embedding）。
from abc import ABC, abstractmethod
import numpy as npnp.random.seed(42)张量¶
class Tensor:
def __init__(self, data):
self.data = np.array(data)
self.grad = np.zeros_like(self.data)
self.gradient_fn = None
self.parents = set()
def backward(self):
if self.gradient_fn is not None:
self.gradient_fn()
for p in self.parents:
p.backward()
def __str__(self):
return f'Tensor({self.data})'class Dataset(ABC):
def __init__(self, batch_size=1):
self.batch_size = batch_size
self.load()
self.train()
@abstractmethod
def load(self):
pass
def train(self):
self.data = self.train_data
def eval(self):
self.data = self.test_data
def all(self):
x, y = self.data
return Tensor(x), Tensor(y)
def __len__(self):
x, *_ = self.data
return len(x) // self.batch_size
def __getitem__(self, index):
s = slice(index * self.batch_size, (index + 1) * self.batch_size)
x, y = self.data
return Tensor(x[s]), Tensor(y[s])class CharDataset(Dataset):
def __init__(self, filename, batch_size=1, context_size=64, stride=None, split=0.9):
self.filename = filename
self.context_size = context_size
self.stride = stride if stride is not None else context_size // 2
self.split = split
super().__init__(batch_size)
def load(self):
with open(self.filename, encoding="utf-8") as f:
text = f.read()
self.vocab = sorted(set(text))
self.vocab_size = len(self.vocab)
self.stoi = {ch: i for i, ch in enumerate(self.vocab)}
self.itos = {i: ch for i, ch in enumerate(self.vocab)}
self.tokens = self.encode(text)
split = int(len(self.tokens) * self.split)
self.train_data = self._pack(self.tokens[:split])
self.test_data = self._pack(self.tokens[split:])
def _pack(self, tokens):
onehot = np.eye(self.vocab_size)
x, y = [], []
for i in range(0, len(tokens) - self.context_size - 1, self.stride):
# 特征：索引编码数组
x.append(tokens[i: i + self.context_size])
# 标签：独热编码
y.append(onehot[tokens[i + self.context_size]])
return x, y
def encode(self, symbols):
return [self.stoi[s] for s in symbols]
def decode(self, tokens):
return "".join(self.itos[t] for t in tokens)class Layer(ABC):
def __call__(self, x: Tensor):
return self.forward(x)
@abstractmethod
def forward(self, x: Tensor):
pass
@property
def parameters(self):
return []线性层¶
class Linear(Layer):
def __init__(self, in_size, out_size):
self.weight = Tensor(np.random.randn(out_size, in_size) * np.sqrt(2 / in_size))
self.bias = Tensor(np.zeros(out_size))
def forward(self, x: Tensor):
p = Tensor(x.data @ self.weight.data.T + self.bias.data)
def gradient_fn():
self.weight.grad += p.grad.T @ x.data
self.bias.grad += np.sum(p.grad, axis=0)
x.grad += p.grad @ self.weight.data
p.gradient_fn = gradient_fn
p.parents = [x]
return p
@property
def parameters(self):
return [self.weight, self.bias]顺序层¶
class Sequential(Layer):
def __init__(self, layers):
self.layers = layers
def forward(self, x: Tensor):
for l in self.layers:
x = l(x)
return x
@property
def parameters(self):
return [p for l in self.layers for p in l.parameters]嵌入层¶
本质上，嵌入层（Embedding Layer）是针对编码向量特别设计的线性层。它的设计思路是用一个比较小的稠密向量（Dense Vector），来取代巨大但是稀疏的独热编码向量。
前向传播（查表）：嵌入层的输入数据是输入文字的词元数组（索引编码数组）。前向传播的过程就是用这个索引编码数组来检索权重矩阵，取出对应位置的几行权重。每行权重就是我们希望获得的稠密向量，称为词向量（Word Vector）；
反向传播：和线性层一样，嵌入层的权重矩阵也需要在模型训练中被优化。区别在于，前向传播中只有几行特定的权重被使用，因此在反向传播中也只需要把梯度累加到这几行权重上（同一个词元重复出现时，梯度会累加多次）。
💡 提示：前面用线性层推导时，权重矩阵的形状是
(128, 10000)，取出的是列。嵌入层的权重矩阵则存放成转置的形式(词表大小, 嵌入尺寸)，这样按索引查表取出的就是行。两种做法是等价的。
💡 提示：嵌入层区别于线性层的另一点在于，它没有偏置参数。因为嵌入层的前向传播是一个查表的动作，没有加权求和的计算步骤。
class Embedding(Layer):
def __init__(self, vocab_size, embedding_size, std=0.02):
super().__init__()
self.weight = Tensor(np.random.randn(vocab_size, embedding_size) * std)
def forward(self, x: Tensor):
# 查表，创建嵌入值张量
p = Tensor(self.weight.data[x.data])
# 嵌入值张量的梯度（闭包）函数
def gradient_fn():
np.add.at(self.weight.grad, x.data, p.grad)
p.gradient_fn = gradient_fn
# 嵌入值张量的父节点列表
p.parents = [self.weight]
return p
@property
def parameters(self):
return [self.weight]平均池化层¶
嵌入层查完表以后，一段文字就被转换成一组词向量（每个字符对应一行）。
平均池化层（Mean Pooling Layer）做的，就是把所有取出的词向量求均值，合并成一个向量。这个向量包含了整段文字的信息。这样做的目的，是为了在数据维度上对接多层网络模型，另一个好处是可以处理不同长度的文字。
💡 提示：平均池化层合并数据的结果，和词袋一样，会丢失文字的词序信息。
class MeanPool(Layer):
def forward(self, x: Tensor):
# 创建输出值张量
p = Tensor(np.mean(x.data, axis=1))
# 输出值张量的梯度（闭包）函数
def gradient_fn():
x.grad += p.grad[:, None, :] / x.data.shape[1]
p.gradient_fn = gradient_fn
# 输出值张量的父节点列表
p.parents = [x]
return pclass ReLU(Layer):
def forward(self, x: Tensor):
a = Tensor(np.maximum(0, x.data))
def gradient_fn():
x.grad += a.grad * (a.data > 0)
a.gradient_fn = gradient_fn
a.parents = [x]
return a归一化指数函数（Softmax）¶
class Softmax(Layer):
def __init__(self, axis=-1):
super().__init__()
self.axis = axis
def forward(self, x: Tensor):
exp = np.exp(x.data - np.max(x.data, axis=self.axis, keepdims=True))
a = Tensor(exp / np.sum(exp, axis=self.axis, keepdims=True))
def gradient_fn():
grad = np.sum(a.data * a.grad, axis=self.axis, keepdims=True)
x.grad += a.data * (a.grad - grad)
a.gradient_fn = gradient_fn
a.parents = [x]
return a损失函数（均方误差）¶
class MSELoss:
def __call__(self, p: Tensor, y: Tensor):
return self.loss(p, y)
def loss(self, p: Tensor, y: Tensor):
mse = Tensor(np.mean(np.square(y.data - p.data)))
def gradient_fn():
p.grad += -2 * (y.data - p.data) / y.data.size
mse.gradient_fn = gradient_fn
mse.parents = [p]
return mse优化器（随机梯度下降）¶
class SGDOptimizer:
def __init__(self, parameters, lr):
self.parameters = parameters
self.lr = lr
def zero_grad(self):
for p in self.parameters:
p.grad = np.zeros_like(p.data)
def step(self):
for p in self.parameters:
p.data -= p.grad * self.lr训练器¶
class Trainer:
def __init__(self, model, loss_fn, optimizer):
self.model = model
self.loss_fn = loss_fn
self.optimizer = optimizer
def train(self, dataset, epochs):
dataset.train()
for epoch in range(epochs):
for i in range(len(dataset)):
feature, label = dataset[i]
self.optimizer.zero_grad()
prediction = self.model(feature)
loss = self.loss_fn(prediction, label)
loss.backward()
self.optimizer.step()
def test(self, dataset):
dataset.eval()
feature, label = dataset.all()
prediction = self.model(feature)
loss = self.loss_fn(prediction, label)
return prediction, loss
def generate(self, dataset, prompt, steps=256):
tokens = dataset.encode(prompt)
for _ in range(steps):
window = tokens[-dataset.context_size:]
feature = Tensor([window])
prediction = self.model(feature)
probs = prediction.data[0] / np.sum(prediction.data[0])
token = np.random.choice(len(probs), p=probs)
tokens.append(token)
return dataset.decode(tokens)LEARNING_RATE = 0.01批大小¶
BATCH_SIZE = 4轮数¶
EPOCHS = 10上下文尺寸¶
CONTEXT_SIZE = 32嵌入尺寸¶
嵌入尺寸（Embedding Size）是嵌入层的超参数，代表我们想把数据向量的尺寸从词表大小压缩到多大。
实际的大模型，在 GPT-3 时期，词表大小约为 50K，嵌入尺寸从几百到一万出头不等（最大的模型是 12,288）。目前最新的大模型，词表已经扩充到 100K 到 200K，而嵌入尺寸大约在 10K 左右。
EMBEDDING_SIZE = 16建模¶
数据集提供的特征，从词袋的独热向量变成了词元数组（索引编码集合）。
而我们的网络模型也增加了嵌入层和平均池化层，目的在于将特征的索引编码集合，转换合并成一个嵌入权重均值向量。
之后再连接到常规的两层网络模型：
Embedding：嵌入层，将索引编码集合转换成一组嵌入向量；MeanPool：平均池化层，将这些嵌入向量压缩进一个向量；Linear：隐藏层；ReLU：隐藏层激活函数；Linear：输出层；Softmax：输出层激活函数，输出下一个字符的概率分布。
数据在模型中的形状依次是：特征 (批大小, 上下文尺寸)，嵌入层输出 (批大小, 上下文尺寸, 嵌入尺寸)，平均池化层输出 (批大小, 嵌入尺寸)，最终的预测 (批大小, 词表大小)。
DATA_FILE = "../../../../data/tinyshakespeare.txt"
dataset = CharDataset(DATA_FILE, BATCH_SIZE, CONTEXT_SIZE)
model = Sequential([
# 嵌入层
Embedding(dataset.vocab_size, EMBEDDING_SIZE),
# 平均池化层
MeanPool(),
# 隐藏层
Linear(EMBEDDING_SIZE, EMBEDDING_SIZE * 2),
# 隐藏层激活函数
ReLU(),
# 输出层
Linear(EMBEDDING_SIZE * 2, dataset.vocab_size),
# 输出层激活函数
Softmax()
])
loss_fn = MSELoss()
optimizer = SGDOptimizer(model.parameters, lr=LEARNING_RATE)
trainer = Trainer(model, loss_fn, optimizer)训练¶
trainer.train(dataset, EPOCHS)评估¶
prediction, loss = trainer.test(dataset)
print(f'prediction:\t{prediction.data.shape}\nloss:\t{loss}')prediction: (6970, 65)
loss: Tensor(0.015120499995513398)
推理¶
print(trainer.generate(dataset, prompt="ROMEO:"))ROMEO:tLjtp;xLJ$S,DHnrGqqk3gwfITdU-Hnq.wIeknwHq.gjfDr:$hSqM:pX?hUaRooh,pegKsi-Ns!xfcEDwGlM.
WyLKJVPsUU.yD&eaOOm?odOQDkUHazPuJ$AOB
upj:-&ivP-UJLbh,!zC$?!MnJQeUqr
lRzSzkDWzq,hJ
NCoIo
GCC&?XFOLrtTI,w$GGH:m-c-&t?oW&xCq:q
Ulvtos?FFLMW:g:HUmjd3W'LO!ExwK!;Mw.k,yLlJc?3v
这一章，我们将文字的表示方式从独热编码升级成了嵌入向量。
这样做的目的，是降低数据量。但是并不能明显地提升推理质量：因为平均池化和词袋一样，仍然没有利用文字的词序。这一点从推理结果就可以看出，依旧是没有任何意义的乱码。
课后练习¶
尝试调整嵌入尺寸，比如改小到
8，或者改大到128，观察对训练和推理的速度和结果有什么影响？