Tiny Transformer

OWPETER Lv4

简单的字符预测模型

数据集

1
2
3
# wget https://raw.githubusercontent.com/karpathy/char-rnn/master/data/tinyshakespeare/input.txt
with open('input.txt', 'r', encoding='utf-8') as f:
text = f.read()

我们的模型使用的数据集是莎士比亚诗歌集合,其中包含了莎士比亚所有的诗,大小大约1M

为了防止模型过拟合,需要将数据集划分为训练集与验证集

1
2
3
4
data = torch.tensor(encode(text), dtype=torch.long)
n = int(0.9*len(data)) # first 90% will be train, rest val
train_data = data[:n]
val_data = data[n:]

这里我们将前90%的内容划分到训练集,后10%的内容划分到验证集。在训练模型时,只使用训练集中的数据,在验证模型准确性时,只使用验证集的数据。

解码器编码器

我们将制作一个最简单的字符集encoder和decoder。首先统计数据集中所有出现过的字符并排序,encoder将字符映射到整形的序号,decoder将序号映射到字符

1
2
3
4
5
6
7
8
# here are all the unique characters that occur in this text
chars = sorted(list(set(text)))
vocab_size = len(chars)
# create a mapping from characters to integers
stoi = { ch:i for i,ch in enumerate(chars) }
itos = { i:ch for i,ch in enumerate(chars) }
encode = lambda s: [stoi[c] for c in s] # encoder: take a string, output a list of integers
decode = lambda l: ''.join([itos[i] for i in l]) # decoder: take a list of integers, output a string

数据加载

在训练模型时,我们不会把整个训练集数据一股脑全部喂给模型,主要的原因是你的GPU显存是有限的,将整个数据集同时加载到显存中会导致溢出。当然也有训练稳定性和收敛性的原因…

正确的做法是每次从训练集中随机抽出一定量的数据组,这个量称为batch_size,将这batch_size个数据组送给你的GPU去并行计算。为什么是数据“组”呢,因为这个模型是个字符预测模型,需要根据前面几个字符是什么去预测下一个字符可能是什么,你不能每次只给模型传一个字符吧,那人家学习什么呢。

顺带一提,每个batch之间是相互独立的,这得益于CUDA的线程块机制以及GPU的SIMD架构

1
2
3
4
5
6
7
8
def get_batch(split):
# generate a small batch of data of inputs x and targets y
data = train_data if split == 'train' else val_data
ix = torch.randint(len(data) - block_size, (batch_size,))
x = torch.stack([data[i:i+block_size] for i in ix])
y = torch.stack([data[i+1:i+block_size+1] for i in ix])
x, y = x.to(device), y.to(device)
return x, y

借助这个get_batch函数,我们可以每次获得batch_size个训练数据xbatch_size训练数据对应的标签。这个标签的作用是告诉模型“某个字符后面跟着的真实的字符是哪个”,从而与模型自己预测的字符进行比较。标签的获取也非常简单,就是训练用的字符后面跟着的那个字符。

模型搭建

数据准备好了,就可以开始搭模型了

1
2
3
4
def __init__(self, vocab_size):
super().__init__()
# each token directly reads off the logits for the next token from a lookup table
self.token_embedding_table = nn.Embedding(vocab_size, vocab_size)

首先定义模型的嵌入层,第一个vocab_size代表输入词汇表的大小,即有多少个不同的字符,第二个vocab_size代表每个token的嵌入向量维度。这个模型较为特殊的一点是其嵌入向量维度正好等于token的个数,因此可以直接将嵌入向量视作每个token的概率分布,你将在forward函数中更直观的看到这一点。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def forward(self, idx, targets=None):

# idx and targets are both (B,T) tensor of integers
logits = self.token_embedding_table(idx) # (B,T,C)

if targets is None:
loss = None
else:
B, T, C = logits.shape
logits = logits.view(B*T, C)
targets = targets.view(B*T)
loss = F.cross_entropy(logits, targets)

return logits, loss

然后是模型的前向传播函数,idx对应get_batch得到的xtargets对应y。该函数先将通过embedding表将输入的token转换为logits,其实就是未归一化的预测概率,然后使用cross_entropy计算模型预测与真实值的差距

这里最重要的是要理解logits张量各个维度代表的含义:B, Batch size,代表批次大小,指一次并行处理的样本数量;T, Time steps, 代表每个序列包含的token数量,也就是上下文长度;C, Channels,代表每个token嵌入向量的维度

另外,为什么要view一下,因为cross_entropy对于输入形状有要求:

最后编写推理函数。推理函数是在模型训练完成后,用来运行模型进行生成任务的函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def generate(self, idx, max_new_tokens):
# idx is (B, T) array of indices in the current context
for _ in range(max_new_tokens):
# get the predictions
logits, loss = self(idx)
# focus only on the last time step
logits = logits[:, -1, :] # becomes (B, C)
# apply softmax to get probabilities
probs = F.softmax(logits, dim=-1) # (B, C)
# sample from the distribution
idx_next = torch.multinomial(probs, num_samples=1) # (B, 1)
# append sampled index to the running sequence
idx = torch.cat((idx, idx_next), dim=1) # (B, T+1)
return idx

推理的过程:将当前的序列输入模型,获得每个位置对所有可能下一个token的预测分数,但我们只关注于最后一个位置,因此将最后一个位置的预测概率抽出、softmax、然后加权随机选择得到新的token,最后将新的token拼接到原序列末尾,完成一次生成。由于返回的结果是整形向量,因此需要由decoder解码才能转换为可读内容

训练与推理

完成了模型之后,就可以编写训练与推理的代码了

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
device = 'cuda' if torch.cuda.is_available() else 'cpu'

model = BigramLanguageModel(vocab_size)
m = model.to(device)

# create a PyTorch optimizer
optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate)

for iter in range(max_iters):

# every once in a while evaluate the loss on train and val sets
if iter % eval_interval == 0:
losses = estimate_loss()
print(f"step {iter}: train loss {losses['train']:.4f}, val loss {losses['val']:.4f}")

# sample a batch of data
xb, yb = get_batch('train')

# evaluate the loss
logits, loss = model(xb, yb)
optimizer.zero_grad(set_to_none=True)
loss.backward()
optimizer.step()

# generate from the model
context = torch.zeros((1, 1), dtype=torch.long, device=device)
print(decode(m.generate(context, max_new_tokens=500)[0].tolist()))

值得一提的是backward函数,你会发现自己并没有编写这个函数,但却可以直接调用它进行反向传播,这是因为现代主流机器学习框架都具有自动微分技术,它可以根据你编写的前向传播函数自动构建计算图,在反向传播时遍历计算图并自动计算每个结点的梯度


完整代码:点击查看


这个简单的模型已经能够输出具有诗歌格式的火星文了!

Self-attention

Q K and V

注意力机制的核心其实就三个张量:Q,K,V,全名是Query,Key,Value。每个token都会发出一个query,代表“我要查询的内容”,同时发出一个key,代表“我具有的内容”,那么就可以通过矩阵点积 来匹配两个token的之间的相关度,而点积出来的结果是一个值,这个值越大说明两个token之间的关联程度越高,反之亦然。

QKV张量的计算方法是完全一样的。对于1个批次,T步长的一组token来说,它们经过嵌入后组成的张量大小为T,C,我们通过nn.Linear(C, head_size, bias = False)得到大小为T, head_size的QKV张量。在经过Q和K的点积,得到大小为T,T的权重张量,该张量中的第i,j个元素的值就代表第i个token与第j个token的相关性。

计算相关性的目的在于对token的嵌入张量进行更新。我们需要将与某个token具有高相关度的token的内容赋予这个token。这时候就需要V张量了,它代表了“我能给别的token提供什么信息”。这一更新过程也是由点积实现的。

1
2
3
4
5
6
7
8
9
10
11
12
13
B, T, C = 4, 8, 32
x = torch.randn(B, T, C)

head_size = 16
key = nn.Linear(C, head_size, bias=False)
query = nn.Linear(C, head_size, bias=False)
k = key(x) # (B, T, 16)
q = query(x) # (B, T, 16)
v = value(x) # (B, T, 16)

wei = q @ k.transpose(-2, -1) # (B, T, 16) @ (B, 16, T) -> (B, T, T)

out = wei @ v # (B, T, T) @ (B, T, 16) -> (B, T, 16)

那为什么叫自注意力呢?因为QKV的来源都是同一个x,如果来源于不同的模块,那么就叫cross-attention

掩码

在模型推理过程中,只能依靠已有的词推测下一个词,但是如果在预测第t个词时,能够“看到”第t+1个词或更后面的词,就会导致信息泄露(data leakage),这会使模型在训练时学到不正确的依赖关系,导致在实际应用中表现不佳。因此我们使用掩码,通过让当前位置之后的token权重为负无穷,从而在计算softmax时,使这些位置的权重变为0。这样,在计算某个位置的注意力时,模型就只能得到该位置前面的信息了。

具体的代码是这样的:

1
2
3
tril = torch.tril(torch.ones(T, T))
wei = wei.masked_fill(tril == 0, float('-inf'))
wei = F.softmax(wei, dim=-1)

这就是完整的单头注意力机制的实现啦

加入自注意力模块

现在我们需要把自注意力模块加入到上面那个小模型中

首先创建一个新的Head类,作为自注意力的头

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
class Head(nn.Module):
""" one head of self-attention """

def __init__(self, head_size):
super().__init__()
self.key = nn.Linear(n_embd, head_size, bias=False)
self.query = nn.Linear(n_embd, head_size, bias=False)
self.value = nn.Linear(n_embd, head_size, bias=False)
self.register_buffer('tril', torch.tril(torch.ones(block_size, block_size)))

self.dropout = nn.Dropout(dropout)

def forward(self, x):
# input of size (batch, time-step, channels)
# output of size (batch, time-step, head size)
B,T,C = x.shape
k = self.key(x) # (B,T,hs)
q = self.query(x) # (B,T,hs)
# compute attention scores ("affinities")
wei = q @ k.transpose(-2,-1) * k.shape[-1]**-0.5 # (B, T, hs) @ (B, hs, T) -> (B, T, T)
wei = wei.masked_fill(self.tril[:T, :T] == 0, float('-inf')) # (B, T, T)
wei = F.softmax(wei, dim=-1) # (B, T, T)
wei = self.dropout(wei)
# perform the weighted aggregation of the values
v = self.value(x) # (B,T,hs)
out = wei @ v # (B, T, T) @ (B, T, hs) -> (B, T, hs)
return out

然后为模型添加一个sa_head(全称为self attention head)。同时,设置一个超参数n_embd,代表嵌入向量维度,之后,增加一个记录每个token位置信息的嵌入向量,称之为pos_embedding,该嵌入向量与token_embedding共同构成嵌入向量。由于嵌入向量维度不再是vocab_size,我们就需要一个n_embdvocab_size的线性变换层

1
2
3
4
5
6
7
# BigramLanguageModel
def __init__(self, vocab_size):
super().__init__()
self.token_embedding_table = nn.Embedding(vocab_size, n_embd)
self.position_embedding_table = nn.Embedding(block_size, n_embd)
self.lm_head = nn.Linear(n_embd, vocab_size)
self.sa_head = Head(n_embd)

这里传给Head类的head_sizen_embd,因为我们希望经过一个注意力头的处理后,嵌入向量的维度依然保持n_embd

相应的,forward函数中logtis的计算过程也需要修改

1
2
3
4
5
6
# forward
tok_emb = self.token_embedding_table(idx) # (B,T,C)
pos_emb = self.position_embedding_table(torch.arange(T, device=device)) # (T,C)
x = tok_emb + pos_emb # (B,T,C)
x = self.sa_head(x)
logits = self.lm_head(x) # (B,T,vocab_size)

最后有一个小的点需要注意,由于pos_emb的存在,在generate函数中,只能将block_size个idx送入forward中,因此需要对idx进行裁剪

1
2
idx_corp = idx[:, -block_size:]
logits, loss = self(idx_corp)

包含单头自注意力的代码:点击查看


相较于没有注意力机制的模型,损失从2.49下降到2.43,稍有下降,但离我们希望的效果还有很长一段距离

多头注意力

在写好注意力头后,很容易的就能得到多头注意力。多头顾名思义就是多个单头,它们并行的进行计算,以提升计算效率。我们会将完整的数据分发给每个头,让每个头通过不同的权重矩阵去学习不同的模式。但为了最终得到的输出张量维度依然为B, T, C(这里的C是n_embd),需要将每个单头的head_size设置为n_embd // num_heads,最终将每个头的结果使用torch.cat进行拼接

1
2
3
4
5
6
7
8
9
class MultiHeadAttention(nn.Module):
""" multiple heads of self-attention in parallel """

def __init__(self, num_heads, head_size):
super().__init__()
self.heads = nn.ModuleList([Head(head_size) for _ in range(num_heads)])

def forward(self, x):
return torch.cat([h(x) for h in self.heads], dim=-1)

相应对模型进行修改

1
2
3
4
5
6
7
8
9
10
11
12
class BigramLanguageModel(nn.Module):

def __init__(self, vocab_size):
super().__init__()
# ...
self.sa_heads = MultiHeadAttention(4, n_embd//4)

def forward(self, idx, targets=None):
# ...
x = tok_emb + pos_emb # (B,T,C)
x = self.sa_heads(x)
# ...

损失进一步下降,证明多头注意力确实学到了一些单头没有学到的特征

这么小的修改就不用放完整代码了吧

Feed Forward

注意力机制的核心是一个加权求和的过程,这导致它是线性的,为了增强模型的表示能力,引入Feed Forward非线性层,让模型具备学习非线性模式的能力。另外,注意力机制的另一个特征是“token与token之间的交互”,而Feed Forward只对输入张量的最后一个维度进行变换,相当于每个token都有了自己独立的“思考”时间。

具体来说,Feed Forward通过线性变换将每个token的嵌入向量从n_embd映射到n * n_embd的高维空间,在高维空间使用非线性的nn.ReLU激活函数增强使模型学习更复杂的模式,最后再使用线性变换降至n_embd维。

1
2
3
4
5
6
7
8
9
10
11
12
13
class FeedForward(nn.Module):
""" a simple linear layer followed by a non-linearity """

def __init__(self, n_embd):
super().__init__()
self.net = nn.Sequential(
nn.Linear(n_embd, 4 * n_embd),
nn.ReLU(),
nn.Linear(4 * n_embd, n_embd),
)

def forward(self, x):
return self.net(x)

然后在模型中进行相应修改

1
2
3
4
5
6
7
8
9
10
11
12
class BigramLanguageModel(nn.Module):
def __init__(self, vocab_size):
super().__init__()
# ...
self.ffwd = FeedForward(n_embd)

def forward(self, idx, targets=None):
# ...
x = tok_emb + pos_emb # (B,T,C)
x = self.sa_heads(x)
x = self.ffwd(x)
# ...

完整代码:点击查看


又有一点小进步

增加深度与优化

为了进一步提升模型的学习能力,我们希望模型具有多个self-attention层与ffd层,那么不妨将这两部分放入一个Block类,并在模型中设置多个Block

1
2
3
4
5
6
7
8
9
10
11
12
13
class Block(nn.Module):

def __init__(self, n_embd, n_head):
# n_embd: embedding dimension, n_head: the number of heads we'd like
super().__init__()
head_size = n_embd // n_head
self.sa = MultiHeadAttention(n_head, head_size)
self.ffwd = FeedForward(n_embd)

def forward(self, x):
x = self.sa(x)
x = self.ffwd(x)
return x

然后在模型中设置多个Block

1
2
3
4
5
6
7
8
9
10
11
class BigramLanguageModel(nn.Module):

def __init__(self, vocab_size):
# ...
self.blocks = nn.Sequential(*[Block(n_embd, n_head) for _ in range(n_block)])

def forward(self, idx, targets=None):
# ...
x = tok_emb + pos_emb # (B,T,C)
x = self.blocks(x)
# ...

但这样一来,模型深度过深会导致梯度消失,因此必须引入一些优化来避免这一问题

  1. 残差链接
1
2
3
4
5
6
7
class Block(nn.Module):
# ...

def forward(self, x):
x = x + self.sa(x)
x = x + self.ffwd(x)
return x
  1. 使用LayerNorm进行归一化
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
class Block(nn.Module):

def __init__(self, n_embd, n_head):
super().__init__()
head_size = n_embd // n_head
self.sa = MultiHeadAttention(n_head, head_size)
self.ffwd = FeedForward(n_embd)
self.ln1 = nn.LayerNorm(n_embd)
self.ln2 = nn.LayerNorm(n_embd)


def forward(self, x):
x = x + self.sa(self.ln1(x))
x = x + self.ffwd(self.ln2(x))
return x

class BigramLanguageModel(nn.Module):
def __init__(self, vocab_size):
# ...
self.ln_f = nn.LayerNorm(n_embd)

def forward(self, idx, targets=None):
# ...
x = self.blocks(x)
x = self.ln_f(x)
logits = self.lm_head(x) # (B,T,vocab_size)

# ...

Scaling up!

到目前为止,我们已经基本完成了这张图右边的部分(除了cross-attention部分),也就是一个只有decoder的transformer。没有写encoder的原因是,我们的模型目前的功能仅局限于自己产生文本,不受任何条件约束,而encoder的作用就是用来接收并理解外部约束条件的(例如用户的输入)

按照Attention is all you need增加一些dropout之后,就可以开始大规模训练了


完整代码:点击查看


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
First Servant:
You tell might, Aufidius, in descending your
and
accordering to waiting: dost you to me, my chamber, and
prologation again, and I drived think.
Happine! E should how margar me hear to a
custimar?
Is By the time forber Saintman: but what I
will she undo report; and there have threat of the please.
Oposite, sir, here that pikes to the pude woo
to flow last; and your knowers and loves; bear me heavenly
as will commanded, as you will do.

QUEEN MARGARET:
Read upon, and thou art come of them for the
shepherds fall benefit.

Second MENERSE:
Thou art done, and for three once, of thou art dove.;
For any thing to mad, an oathless of rute,
But thou first, to shrewdst which word-blood I spright,
Upon the reason, fury, their deaths: iffiance,
will I look that like triophet the weak,
but chief, let their five blackles; from exclamation
meet or than thou dost. To unachild,
Have done in this day, and make thee out of mine;
first's those our bosoms advance to yourselves;
For thought yet too return back the state; and
Be stand to please.
For this indeeds have rofining and with love:
As there, like one fall, to it scall.

现在已经有明确的角色划分了,格式完全是莎士比亚的格式,语言上也能看出来是英语了。


最后,可以通过以下代码查看这个小模型有多少参数:

1
2
3
4
# model = BigramLanguageModel(vocab_size)
# m = model.to(device)

print(sum(p.numel() for p in m.parameters()) / 1e6, "M")

打印出来大约是10M,也就是1000万

  • Title: Tiny Transformer
  • Author: OWPETER
  • Created at : 2025-08-16 13:45:54
  • Updated at : 2025-08-19 16:30:07
  • Link: https://owpeter.github.io/2025/08/16/ML/LLM/
  • License: This work is licensed under CC BY-NC-SA 4.0.
Comments