从零手搓MiniGPT预训练与监督微调完整流程详解

GPT这类模型的核心逻辑是“根据上文预测下文”，因此需要一种机制，让当前token只能看到自己和它前面的内容，不能提前“偷看”后面的信息。这就是Attention Mask的作用。

4. miniGPT的实现过程

整个训练分为两步：先进行预训练，让模型学会“接话”；再执行SFT（监督微调），使模型的回答更接近人类语言。

4.1. 预训练

4.1.1. 选择数据集

预训练环节选用了minimind项目中的pretrain_t2t_mini.jsonl数据集。数据格式如下：

{"text": "给我生成一首有关秋天的诗歌。秋日早晨，清风拂面。n金色的叶子，似火在燃烧。n露珠晶莹，如珍珠般美丽。n秋的气息，弥漫在空气中。n余音袅袅，如鸟儿的歌唱。n美丽的秋天，是大自然的馈赠。帮我想一些创意，给即将到来的公司年会准备节目。一些节目比如能否请一位表演者为我们表演一曲钢琴曲，或者请一位小提琴手为大家演奏一首古典曲目。如果想要画面更具有视觉冲击力，可以安排一个魔术师或者杂技演员的表演。另外，也可以设计一些小游戏或者有奖竞猜来增添活动的趣味性，这些小游戏可以和公司的文化、发展历程等相关。请问给我讲一个清净的法则。无为而治是一项清净的法则。即在处理问题时，不要强行干预，反而尽可能地减少干预，坚持自然的发展趋势。比如，让植物自然生长，照顾它们就行了，不需要过度地修剪和整齐地排列。类似地，让人们自由发展，而不是通过繁琐的管制、政策干涉等方式，去要求、指导人们的行为。这种无为而治的法则旨在维持一个有秩序的、和平的治理状态，却不需要大规模的干预和指挥。"}

4.1.2. 分词器处理文本

选择的分词器是Qwen2-0.5B-Instruct。原始词表大小为151643，官方额外添加了三个特殊字符。在预训练阶段，每个句子的末尾会加上"<|endoftext|>"。然后按 seq_len + 1 = 257 切块（每块包含input和target的右移关系），最后按95/5比例划分为训练集和验证集。

4.1.3. 实现模型

多头自注意力机制的代码实现如下：

class CausalSelfAttention(nn.Module):"""带mask的多头自注意力机制"""def __init__(self, d_model: int, n_heads: int, dropout: float = 0.1):super().__init__()assert d_model % n_heads == 0self.n_heads = n_headsself.d_head = d_model // n_headsself.qkv_proj = nn.Linear(d_model, 3* d_model, bias=False) # qkv合并self.out_proj = nn.Linear(d_model, d_model, bias=False)self.attn_dropout = nn.Dropout(dropout)self.resid_dropout = nn.Dropout(dropout)def forward(self, x: torch.Tensor) -> torch.Tensor:B, T, C = x.shape # batch, seq_len, d_modelqkv = self.qkv_proj(x) # (B, T, 3*C)q, k, v = qkv.split(C, dim=-1) # 各 (B, T, C)# 多头，(B, T, C) -> (B, n_heads, T, d_head)q = q.view(B, T, self.n_heads, self.d_head).transpose(1, 2)k = k.view(B, T, self.n_heads, self.d_head).transpose(1, 2)v = v.view(B, T, self.n_heads, self.d_head).transpose(1, 2)# QK^T / sqrt(d_head)scale = math.sqrt(self.d_head)attn = (q @ k.transpose(-2, -1)) / scale # (B, n_heads, T, T)# 上三角矩阵，不含对角线。把上三角置为-inf，softmax后趋近于0causal_mask = torch.triu(torch.ones(T, T, device=x.device), diagonal=1).bool()attn = attn.masked_fill(causal_mask, float("-inf"))# softmaxattn = F.softmax(attn, dim=-1)attn = self.attn_dropout(attn)# *Vout = (attn @ v).transpose(1, 2).reshape(B, T, C)return self.resid_dropout(self.out_proj(out))

前馈神经网络的代码：

class FeedForward(nn.Module):"""前馈网络Linear -> GELU -> Linear"""def __init__(self, d_model: int, d_ff: int, dropout: float = 0.1):super().__init__()self.net = nn.Sequential(nn.Linear(d_model, d_ff),nn.GELU(),nn.Linear(d_ff, d_model),nn.Dropout(dropout))def forward(self, x: torch.Tensor) -> torch.Tensor:return self.net(x)

单个Decoder块的代码：

class TransformerBlock(nn.Module):"""单个block块，transformer decoder"""def __init__(self, d_model: int, n_heads: int, d_ff: int, dropout: float = 0.1):super().__init__()self.ln1 = nn.LayerNorm(d_model) # 归一化，防止计算爆炸self.attn = CausalSelfAttention(d_model, n_heads, dropout)self.ln2 = nn.LayerNorm(d_model)self.ffn = FeedForward(d_model, d_ff, dropout)def forward(self, x: torch.Tensor) -> torch.Tensor:x = x + self.attn(self.ln1(x)) # 残差x = x + self.ffn(self.ln2(x))return x

整体的模型结构如下图所示：

对应的代码实现：

class MiniGPT(nn.Module):"""Decoder-only GPT 模型"""def __init__(self,vocab_size: int,seq_len: int,d_model: int,n_heads: int,n_layers: int,d_ff: int,dropout: float,):super().__init__()self.seq_len = seq_lenself.d_model = d_modelself.token_emb = nn.Embedding(vocab_size, d_model)self.pos_emb = nn.Embedding(seq_len, d_model)self.emb_dropout = nn.Dropout(dropout)self.blocks = nn.ModuleList([TransformerBlock(d_model, n_heads, d_ff, dropout)for _ in range(n_layers)])self.ln_final = nn.LayerNorm(d_model)self.lm_head = nn.Linear(d_model, vocab_size, bias=False)self.apply(self._init_weights)def _init_weights(self, module: nn.Module): # 不用自带的初始值if isinstance(module, nn.Linear):nn.init.normal_(module.weight, mean=0.0, std=0.02)if module.bias is not None:nn.init.zeros_(module.bias)elif isinstance(module, nn.Embedding):nn.init.normal_(module.weight, mean=0.0, std=0.02)def forward(self, input_ids: torch.Tensor) -> torch.Tensor:B, T = input_ids.shape # batch, seq_lenassert T <= self.seq_len, f"输入长度 {T} 超过最大序列长度 {self.seq_len}"positions = torch.arange(T, device=input_ids.device)x = self.emb_dropout(self.token_emb(input_ids) + self.pos_emb(positions))for block in self.blocks:x = block(x)x = self.ln_final(x)return self.lm_head(x) # (B, T, vocab_size)def count_parameters(self) -> int:return sum(p.numel() for p in self.parameters() if p.requires_grad)

4.1.4. 训练

假设输入序列是 abcd，训练目标就是让模型根据 abc 预测出 bcd。首先加载处理好的预训练文本：

class TokenDataset(Dataset):"""从 .pt 文件加载加载数据集每个样本是 (seq_len+1, ) 的token序列input = sample[:-1]target = sample[1:]"""def __init__(self, pt_path: Path):ckpt = torch.load(pt_path, weights_only=True)self.data = ckpt["data"]self.vocab_size = ckpt["vocab_size"]self.seq_len = ckpt["seq_len"]def __len__(self):return len(self.data)def __getitem__(self, idx):chunk = self.data[idx]return chunk[:-1], chunk[1:]

然后将输入喂给模型，计算输出与目标的损失，再更新梯度：

input_ids = input_ids.to(device)targets = targets.to(device)logits = model(input_ids) # transformer计算后的结果B, T, V = logits.shape # (batch_size, seq_len, vocab_size)# 自回归损失：每个位置预测下一个tokenloss = loss_fn(logits.view(B*T, V), targets.view(B*T))optimizer.zero_grad()loss.backward()# Gradient clipping：将梯度的 L2 范数截断到 grad_clipgrad_norm = torch.nn.utils.clip_grad_norm_(model.parameters(), grad_clip)optimizer.step()scheduler.step()

4.2. 采样策略

常见的采样策略有四种：

• Greedy（贪心）：每一步直接取 argmax，选择概率最高的token。
• Temperature（温度采样）：通过temperature参数控制分布的平滑度，温度越低，输出越保守。
• Top-K：从概率最高的K个token中随机采样。
• Top-P（Nucleus采样）：从累积概率达到阈值p的最小集合中采样。

for _ in range(max_new_tokens):# 如果序列超过 seq_len，截取最后 seq_len 个 token 作为上下文context = generated[:, -seq_len:]logits = model(context)# (1, T, V)next_logits = logits[0, -1, :] # (V,) 只取最后一步的预测if strategy == "greedy":# 贪心：直接取最高分next_token = next_logits.argmax(dim=-1, keepdim=True)# (1,)else:# 采样前统一做 temperature 缩放next_logits = next_logits / max(temperature, 1e-8)if strategy == "top_k":# Top-K：将低于第 K 大的 logits 置为 -infvalues, _ = torch.topk(next_logits, top_k)threshold = values[-1]next_logits = next_logits.masked_fill(next_logits < threshold, float("-inf"))elif strategy == "top_p":# Top-P（Nucleus）：按概率从大到小排序，累积到 ≥ p 时截断sorted_logits, sorted_indices = torch.sort(next_logits, descending=True)cumprobs = torch.cumsum(F.softmax(sorted_logits, dim=-1), dim=-1)# 找到刚好超过 p 的位置，将该位置之后的 token 排除sorted_indices_to_remove = cumprobs > top_p# 保留至少 1 个 token：右移一位，使超过阈值的第一个位置不被移除sorted_indices_to_remove[1:] = sorted_indices_to_remove[:-1].clone()sorted_indices_to_remove[0] = Falseindices_to_remove = sorted_indices[sorted_indices_to_remove]next_logits[indices_to_remove] = float("-inf")elif strategy == "temperature":pass# 只做 temperature 缩放，不做 top-k/p 截断probs = F.softmax(next_logits, dim=-1)next_token = torch.multinomial(probs, num_samples=1)# (1,)if next_token.item() == 151643:breakgenerated = torch.cat([generated, next_token.unsqueeze(0)], dim=1)# (1, T+1)return generated[0]# (total_len,)

预训练的效果如下：