大家好,我是阿浪

说实话,几个月前我还觉得“大模型”这三个字离我很远。什么Transformer、注意力机制、千亿参数,听起来就像是天才才能碰的东西。

直到有一天,我硬着头皮把《Attention Is All You Need》那篇论文打印出来,关掉所有浏览器标签页,拿着笔一点点推公式。然后我发现了一个让我既震惊又兴奋的事实——

大模型的底层,全是我考研时背烂了的数学。

线性代数、高等数学、概率统计,这三门课几乎覆盖了Transformer的全部核心计算。矩阵乘法、复合函数、链式求导、Softmax归一化……这些东西我们当年刷了成百上千道题,结果它们就是大模型的骨架。

于是我决定做一件事:从零写一个大模型,一行预训练代码都不调,全部手写。

三天之后,我跑通了一个最简版本。今天就把这个过程分享出来,代码全附上,不藏私。

大模型到底是什么?
先别急着看代码,咱们先把事情说透。

如果你把大模型抽象到最顶层,它其实就是一个超级复合函数:

输入(文字) → 一系列数学变换 → 输出(下一个文字)

这个函数有多层嵌套,每一层都是一个数学变换。你给它一段话,它预测下一个词最可能是什么。

就这么简单。

那这个“超级复合函数”由什么组成?拆开看就是三样东西:

  1. 注意力机制(Attention) ——让模型知道一句话里谁跟谁关系最密切

  2. 前馈网络(Feed-Forward) ——对每个位置独立做非线性变换

  3. 位置编码(Position Encoding) ——告诉模型词语的顺序

这三样东西堆叠起来,就是Transformer。堆得越多,模型越大。

第一步:输入层——把文字变成数字(线性代数)
计算机不认识“你好世界”这四个字,它只认识数字。所以第一步就是把文字变成矩阵。

这个过程分两步:词嵌入 + 位置编码。

词嵌入:离散→连续的线性映射
词嵌入的本质就是一个矩阵乘法。我们有一个词表(比如1000个词),每个词对应一个128维的向量。这个向量就是模型对这个词的“理解”。

import numpy as np

# 词表大小和向量维度
vocab_size = 1000
hidden_dim = 128

# 随机初始化词嵌入矩阵(可学习参数)
embedding_matrix = np.random.randn(vocab_size, hidden_dim) * 0.01

# 模拟输入:假设输入了4个token,索引分别是10, 20, 30, 40
input_tokens = np.array([10, 20, 30, 40])

# 查表得到每个token的向量表示
input_embeds = embedding_matrix[input_tokens]

print("输入特征矩阵形状:", input_embeds.shape)  # (4, 128)

这里我想多说一句:很多人觉得词嵌入很神秘,其实它就是一张大表格。每个词占一行,每行是一个向量。模型训练的过程,就是不断调整这张表格里的数字。

位置编码:用三角函数告诉模型顺序(高等数学)
词嵌入解决了“每个词是什么”,但没解决“每个词在什么位置”。Transformer是并行计算的,它不像RNN那样一个一个读,所以需要主动告诉它位置信息。

位置编码用的是正弦和余弦函数:

def get_positional_encoding(seq_len, d_model):
    """生成位置编码矩阵"""
    pos_encoding = np.zeros((seq_len, d_model))
    
    for pos in range(seq_len):
        for i in range(0, d_model, 2):
            # 偶数维度用sin
            pos_encoding[pos, i] = np.sin(pos / (10000 ** (i / d_model)))
            # 奇数维度用cos
            if i + 1 < d_model:
                pos_encoding[pos, i + 1] = np.cos(pos / (10000 ** (i / d_model)))
    
    return pos_encoding

# 假设序列长度是4,维度是128
pos_encoding = get_positional_encoding(4, 128)

# 最终输入 = 词嵌入 + 位置编码
final_input = input_embeds + pos_encoding[:4, :]

说句心里话:我第一次看到这个公式的时候也是一脸懵,为什么要用sin和cos?后来我想明白了——因为sin和cos能产生不同频率的波形,不同的位置就有不同的“指纹”,模型就能区分谁在前谁在后。这就是高数里周期函数的实际应用。

第二步:自注意力——模型怎么知道谁跟谁有关系(概率统计)
这是Transformer最核心的部分,也是很多人觉得最难的部分。

但其实它的逻辑特别朴素:模型在处理一个词的时候,会去“看”句子里的其他词,判断谁跟当前词关系最紧密。

举个例子:“小明喜欢吃苹果,因为它很甜。”这里的“它”指谁?模型通过注意力机制会发现“苹果”和“它”的关系最密切。

数学公式(别怕,我拆开讲)
自注意力的公式长这样:

Attention(Q, K, V) = softmax(Q × K^T / √d_k) × V

拆成三步:

  1. 算相关性:Q和K做矩阵乘法,得到一个“相关性分数”矩阵

  2. Softmax归一化:把分数变成概率(加起来等于1)

  3. 加权求和:用这些概率去加权V,得到最终输出

手写代码

import torch
import torch.nn as nn
import torch.nn.functional as F

class SelfAttention(nn.Module):
    def __init__(self, embed_size):
        super().__init__()
        self.embed_size = embed_size
        
        # 三个线性层,分别生成Q、K、V
        self.Wq = nn.Linear(embed_size, embed_size, bias=False)
        self.Wk = nn.Linear(embed_size, embed_size, bias=False)
        self.Wv = nn.Linear(embed_size, embed_size, bias=False)
    
    def forward(self, x):
        # x的形状: (batch_size, seq_len, embed_size)
        
        # 1. 生成Q、K、V
        Q = self.Wq(x)  # (batch, seq_len, embed_size)
        K = self.Wk(x)  # (batch, seq_len, embed_size)
        V = self.Wv(x)  # (batch, seq_len, embed_size)
        
        # 2. 计算注意力分数:Q × K^T
        # K.transpose(-2, -1) 把最后两维转置
        scores = torch.matmul(Q, K.transpose(-2, -1))  # (batch, seq_len, seq_len)
        
        # 3. 缩放(除以√d_k),防止softmax进入梯度极小区域
        scores = scores / (self.embed_size ** 0.5)
        
        # 4. Softmax归一化成概率
        attention_weights = F.softmax(scores, dim=-1)  # (batch, seq_len, seq_len)
        
        # 5. 加权求和
        out = torch.matmul(attention_weights, V)  # (batch, seq_len, embed_size)
        
        return out

这里有个细节我想特别说一下:为什么要除以√d_k?因为如果向量维度很大,Q和K的点积结果会很大,导致Softmax之后梯度极小,模型学不动。除以√d_k就是把数值拉回到一个合适的范围。这就是为什么考研数学里的“量纲分析”在实际工程中这么重要。

第三步:多头注意力——让模型从多个角度理解
一个注意力头只能从一个角度“看”句子。Transformer的做法是:同时用多个头,每个头关注不同的东西。

比如:

  1. 头1可能关注语法结构

  2. 头2可能关注语义关联

  3. 头3可能关注指代关系

class MultiHeadAttention(nn.Module):
    def __init__(self, embed_size, num_heads):
        super().__init__()
        self.num_heads = num_heads
        self.head_dim = embed_size // num_heads
        
        # 用一个线性层同时生成所有头的Q、K、V
        self.qkv_proj = nn.Linear(embed_size, embed_size * 3, bias=False)
        self.out_proj = nn.Linear(embed_size, embed_size, bias=False)
    
    def forward(self, x):
        batch_size, seq_len, embed_size = x.shape
        
        # 1. 生成Q、K、V
        qkv = self.qkv_proj(x)  # (batch, seq_len, 3 * embed_size)
        
        # 2. 拆分成多个头
        qkv = qkv.reshape(batch_size, seq_len, 3, self.num_heads, self.head_dim)
        qkv = qkv.permute(2, 0, 3, 1, 4)  # (3, batch, num_heads, seq_len, head_dim)
        Q, K, V = qkv[0], qkv[1], qkv[2]
        
        # 3. 每个头独立计算注意力
        scores = torch.matmul(Q, K.transpose(-2, -1)) / (self.head_dim ** 0.5)
        attention_weights = F.softmax(scores, dim=-1)
        out = torch.matmul(attention_weights, V)  # (batch, num_heads, seq_len, head_dim)
        
        # 4. 合并所有头
        out = out.permute(0, 2, 1, 3).reshape(batch_size, seq_len, embed_size)
        
        # 5. 输出投影
        out = self.out_proj(out)
        return out

第四步:组装Transformer层
有了注意力机制,再加上前馈网络和残差连接,就是一个完整的Transformer层了。

class TransformerBlock(nn.Module):
    def __init__(self, embed_size, num_heads, ff_hidden_size):
        super().__init__()
        self.attention = MultiHeadAttention(embed_size, num_heads)
        self.feed_forward = nn.Sequential(
            nn.Linear(embed_size, ff_hidden_size),
            nn.ReLU(),
            nn.Linear(ff_hidden_size, embed_size)
        )
        self.norm1 = nn.LayerNorm(embed_size)
        self.norm2 = nn.LayerNorm(embed_size)
    
    def forward(self, x):
        # 残差连接 + 层归一化
        attn_out = self.attention(x)
        x = self.norm1(x + attn_out)  # 残差连接:输入+输出
        
        ff_out = self.feed_forward(x)
        x = self.norm2(x + ff_out)    # 残差连接:输入+输出
        
        return x

残差连接(就是代码里的 x + attn_out)是我觉得特别巧妙的设计。它的作用是让梯度能直接传回浅层,避免深层网络梯度消失。通俗说就是给信息开了一条“高速公路”,不管中间经过多少层变换,原始信息都能直接传到后面。

第五步:堆叠多层,形成完整模型
把多个Transformer层堆叠起来,再加上输入层和输出层,就是一个完整的大模型了。

class MiniGPT(nn.Module):
    def __init__(self, vocab_size, embed_size, num_layers, num_heads, ff_hidden_size, max_seq_len):
        super().__init__()
        self.embed_size = embed_size
        
        # 词嵌入
        self.token_embedding = nn.Embedding(vocab_size, embed_size)
        
        # 位置编码(用可学习的位置编码,简单起见)
        self.pos_embedding = nn.Embedding(max_seq_len, embed_size)
        
        # Transformer层
        self.layers = nn.ModuleList([
            TransformerBlock(embed_size, num_heads, ff_hidden_size)
            for _ in range(num_layers)
        ])
        
        # 输出层
        self.ln_final = nn.LayerNorm(embed_size)
        self.lm_head = nn.Linear(embed_size, vocab_size)
    
    def forward(self, input_ids):
        seq_len = input_ids.shape[1]
        
        # 词嵌入
        x = self.token_embedding(input_ids)  # (batch, seq_len, embed_size)
        
        # 位置编码
        positions = torch.arange(seq_len, device=input_ids.device).unsqueeze(0)
        x = x + self.pos_embedding(positions)
        
        # 通过所有Transformer层
        for layer in self.layers:
            x = layer(x)
        
        # 输出层
        x = self.ln_final(x)
        logits = self.lm_head(x)  # (batch, seq_len, vocab_size)
        
        return logits

训练:让模型学会说话(链式求导 + 梯度下降)
模型搭好了,但它现在是个“傻子”——参数全是随机的,输出全是乱码。

怎么让它学会说话?训练。

训练的本质就是链式求导 + 梯度下降:

  1. 给模型输入一段文字,它预测下一个词

  2. 计算预测和真实答案之间的差距(损失函数)

  3. 用链式求导算出每个参数对差距的贡献(反向传播)

  4. 用梯度下降调整参数,缩小差距

import torch.optim as optim

# 初始化模型
model = MiniGPT(
    vocab_size=1000,
    embed_size=128,
    num_layers=4,
    num_heads=4,
    ff_hidden_size=512,
    max_seq_len=128
)

# 损失函数和优化器
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=3e-4)

# 模拟一次训练
def train_step(input_ids, target_ids):
    model.train()
    optimizer.zero_grad()
    
    # 前向传播
    logits = model(input_ids)  # (batch, seq_len, vocab_size)
    
    # 计算损失(预测下一个词)
    loss = criterion(logits.view(-1, logits.shape[-1]), target_ids.view(-1))
    
    # 反向传播(链式求导)
    loss.backward()
    
    # 梯度下降(更新参数)
    optimizer.step()
    
    return loss.item()

# 模拟数据
batch_size = 4
seq_len = 32
input_ids = torch.randint(0, 1000, (batch_size, seq_len))
target_ids = torch.randint(0, 1000, (batch_size, seq_len))

loss = train_step(input_ids, target_ids)
print(f"训练损失: {loss:.4f}")

你想过没有:我们考研时背得滚瓜烂熟的链式法则,在这里就是让模型变聪明的核心引擎。每一次 loss.backward(),PyTorch都在做你当年手算过的那些求导。

完整代码

我把上面所有的代码整合成了一个完整的文件

import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim


# ============ 位置编码 ============
def get_positional_encoding(seq_len, d_model):
    pos_encoding = np.zeros((seq_len, d_model))
    for pos in range(seq_len):
        for i in range(0, d_model, 2):
            pos_encoding[pos, i] = np.sin(pos / (10000 ** (i / d_model)))
            if i + 1 < d_model:
                pos_encoding[pos, i + 1] = np.cos(pos / (10000 ** (i / d_model)))
    return torch.tensor(pos_encoding, dtype=torch.float32)


# ============ 多头注意力 ============
class MultiHeadAttention(nn.Module):
    def __init__(self, embed_size, num_heads):
        super().__init__()
        assert embed_size % num_heads == 0
        self.num_heads = num_heads
        self.head_dim = embed_size // num_heads
        
        self.qkv_proj = nn.Linear(embed_size, embed_size * 3, bias=False)
        self.out_proj = nn.Linear(embed_size, embed_size, bias=False)
    
    def forward(self, x):
        batch_size, seq_len, embed_size = x.shape
        
        qkv = self.qkv_proj(x)
        qkv = qkv.reshape(batch_size, seq_len, 3, self.num_heads, self.head_dim)
        qkv = qkv.permute(2, 0, 3, 1, 4)
        Q, K, V = qkv[0], qkv[1], qkv[2]
        
        scores = torch.matmul(Q, K.transpose(-2, -1)) / (self.head_dim ** 0.5)
        attention_weights = F.softmax(scores, dim=-1)
        out = torch.matmul(attention_weights, V)
        
        out = out.permute(0, 2, 1, 3).reshape(batch_size, seq_len, embed_size)
        out = self.out_proj(out)
        return out


# ============ Transformer层 ============
class TransformerBlock(nn.Module):
    def __init__(self, embed_size, num_heads, ff_hidden_size):
        super().__init__()
        self.attention = MultiHeadAttention(embed_size, num_heads)
        self.feed_forward = nn.Sequential(
            nn.Linear(embed_size, ff_hidden_size),
            nn.ReLU(),
            nn.Linear(ff_hidden_size, embed_size)
        )
        self.norm1 = nn.LayerNorm(embed_size)
        self.norm2 = nn.LayerNorm(embed_size)
    
    def forward(self, x):
        attn_out = self.attention(x)
        x = self.norm1(x + attn_out)
        ff_out = self.feed_forward(x)
        x = self.norm2(x + ff_out)
        return x


# ============ 完整模型 ============
class MiniGPT(nn.Module):
    def __init__(self, vocab_size, embed_size, num_layers, num_heads, ff_hidden_size, max_seq_len):
        super().__init__()
        self.token_embedding = nn.Embedding(vocab_size, embed_size)
        self.pos_embedding = nn.Embedding(max_seq_len, embed_size)
        
        self.layers = nn.ModuleList([
            TransformerBlock(embed_size, num_heads, ff_hidden_size)
            for _ in range(num_layers)
        ])
        
        self.ln_final = nn.LayerNorm(embed_size)
        self.lm_head = nn.Linear(embed_size, vocab_size)
    
    def forward(self, input_ids):
        seq_len = input_ids.shape[1]
        x = self.token_embedding(input_ids)
        positions = torch.arange(seq_len, device=input_ids.device).unsqueeze(0)
        x = x + self.pos_embedding(positions)
        
        for layer in self.layers:
            x = layer(x)
        
        x = self.ln_final(x)
        logits = self.lm_head(x)
        return logits


# ============ 训练 ============
def train_step(model, input_ids, target_ids, optimizer, criterion):
    model.train()
    optimizer.zero_grad()
    logits = model(input_ids)
    loss = criterion(logits.view(-1, logits.shape[-1]), target_ids.view(-1))
    loss.backward()
    optimizer.step()
    return loss.item()


# ============ 主程序 ============
if __name__ == "__main__":
    # 超参数
    vocab_size = 1000
    embed_size = 128
    num_layers = 4
    num_heads = 4
    ff_hidden_size = 512
    max_seq_len = 128
    batch_size = 4
    seq_len = 32
    epochs = 100
    
    # 初始化
    model = MiniGPT(vocab_size, embed_size, num_layers, num_heads, ff_hidden_size, max_seq_len)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=3e-4)
    
    # 模拟训练
    print("开始训练...")
    for epoch in range(epochs):
        input_ids = torch.randint(0, vocab_size, (batch_size, seq_len))
        target_ids = torch.randint(0, vocab_size, (batch_size, seq_len))
        
        loss = train_step(model, input_ids, target_ids, optimizer, criterion)
        
        if (epoch + 1) % 10 == 0:
            print(f"Epoch {epoch+1}/{epochs}, Loss: {loss:.4f}")
    
    print("训练完成!")

把上面的代码复制下来,跑一遍。改改参数,看看效果。把每一行代码都搞清楚它在做什么。

你会发现,大模型真的没有那么神秘。

它就是一个超级复合函数。而你,早就学过怎么拆解它。

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐