揭秘大模型的魔法：從零實現一個簡化版的GPT 模型

作者：寫代碼的中年人 2025-10-24 10:34:55

人工智能

GPT系列模型因其強大的生成能力和靈活性，成為了研究和應用的焦點。本文將帶你從零開始，揭開 GPT 模型的“魔法”面紗，通過理論講解和代碼示例，逐步實現一個簡化版的GPT 模型。

大家好，我是寫代碼的中年人！今天我們結合代碼從零實現一個簡化版 GPT 模型。

近年來，大語言模型席卷了人工智能領域，從 ChatGPT 到 LLaMA，它們以驚人的語言理解和生成能力改變了我們與機器交互的方式。

其中，GPT系列模型因其強大的生成能力和靈活性，成為了研究和應用的焦點。本文將帶你從零開始，揭開 GPT 模型的“魔法”面紗，通過理論講解和代碼示例，逐步實現一個簡化版的GPT 模型。

01、從零實現簡化版 GPT 模型的意義

近幾年，大模型火遍全球，成為推動人工智能發展的核心力量。我們每天都在使用這些模型，卻常常會好奇：

GPT 模型到底是怎么“理解”語言的？

分詞、注意力、Transformer 這些名詞具體意味著什么？

如果不用現成的 HuggingFace Transformers 庫，能否自己實現一個“微縮版 GPT”？

答案是：可以！

本文的目標是帶你揭開 GPT 的魔法，從數據處理到模型搭建，再到訓練與文本生成，完整走一遍。雖然我們的模型規模遠遠比不上真正的 GPT-2，但核心思想是一致的。只要你掌握了這里的思路，就能理解大模型的“秘密”。

02、GPT 模型的核心組件與結構

在動手實現 GPT 模型之前，我們先快速了解其核心結構，其所有核心結構我們在前面已經講過：

分詞器（Tokenizer）：

GPT 將文本切分為子詞單元（subword），以平衡詞表大小和語義表達能力。常用算法為 BPE（Byte Pair Encoding），如 GPT-2 和 GPT-3 使用的基于 BPE 的分詞器。本次我們使用SentencePiece。

嵌入層（Embedding）：

將離散的 token id 轉換為連續的向量表示。加入可學習的位置編碼（Learned Positional Encoding），以捕捉 token 在序列中的位置信息。

多頭自注意力（Multi-Head Self Attention）：

GPT 的核心機制，通過計算 Query、Key、Value 捕捉 token 之間的依賴關系。使用 masked self-attention，確保預測下一個 token 時只關注之前的上下文。

前饋層（Feed Forward Network）：

注意力層后的非線性變換，增強模型的表達能力。通常由兩層全連接網絡組成，中間加入激活函數（如 ReLU 或 GELU）。

Transformer Block：

由多頭自注意力層、前饋層、殘差連接和 LayerNorm 組成。殘差連接緩解梯度消失問題，LayerNorm 穩定訓練過程。通過堆疊多個 Transformer Block 構建深層網絡結構。

輸出層：

將隱狀態通過線性變換和 softmax 函數映射到詞表大小的概率分布，預測下一個 token。常結合 top-k 或 top-p 采樣策略處理大規模詞表。

訓練目標：

GPT 采用自回歸語言建模，目標是預測下一個 token。使用交叉熵損失函數，優化器通常為 Adam 或其變種。

03、代碼實現及訓練過程

數據準備

我們本次使用完整的《水滸傳》數據，一個章節一行，保存為: raw_shuihu.txt。如下圖：

清洗數據

清洗《水滸傳》原始文本，生成適合訓練的純凈語料，自動按中文標點（。！？）切分成句子，每句一行，保存為：shuihu.txt。(此處只做最容易的切割)

# path: tools/clean_shuihu.py
"""
清洗《水滸傳》原始文本，生成適合訓練的純凈語料
會自動按中文標點（。！？）切分成句子，每句一行
用法：
    python tools/clean_shuihu.py data/raw_shuihu.txt data/shuihu.txt
"""


import sys
import re


def clean_text(text: str) -> str:
    #  只保留中文和常見標點
    allowed = re.compile(r"[^\u4e00-\u9fff。，、！？：；（）《》——…\n ]+")
    text = allowed.sub(" ", text)


    #  去掉多余空格
    text = re.sub(r"\s+", " ", text)


    #  去掉章節標題 “第X回 …”
    text = re.sub(r"第[一二三四五六七八九十百千0-9]+回.*\n", "", text)


    #  按標點分句（句號、問號、感嘆號），切分后加回標點
    sentences = re.split(r"([。！？])", text)
    merged = []
    for i in range(0, len(sentences)-1, 2):
        s = sentences[i].strip()
        p = sentences[i+1].strip()
        if s:
            merged.append(s + p)


    # 去掉空白句子，避免太短
    merged = [s for s in merged if len(s) > 1]


    return "\n".join(merged)


def main():
    if len(sys.argv) != 2 and len(sys.argv) != 3:
        print("用法: python tools/clean_shuihu.py 輸入文件 [輸出文件]")
        sys.exit(1)


    in_path = sys.argv[1]
    out_path = sys.argv[2] if len(sys.argv) == 3 else "data/shuihu.txt"


    with open(in_path, "r", encoding="utf-8") as f:
        raw = f.read()


    clean = clean_text(raw)


    with open(out_path, "w", encoding="utf-8") as f:
        f.write(clean)


    print(f"清洗完成，輸出保存到 {out_path}")
    print(f"示例前5行：\n" + "\n".join(clean.splitlines()[:5]))


if __name__ == "__main__":
    main()

# 程序輸出
洗完成，輸出保存到 data/shuihu.txt
示例前5行：
水滸傳第一段 話說大宋仁宗天子在位，嘉祐三年三月三日五更三點，天子駕坐紫宸殿，受百官朝賀。
當有殿頭官喝道： 有事出班早奏，無事卷簾退朝。
只見班部叢中，宰相趙哲、參政文彥博出班奏曰： 目今京師瘟疫盛行，民不聊生，傷損軍民多矣。
伏望陛下釋罪寬恩，省刑薄稅，以禳天災，救濟萬民。
天子聽奏，急敕翰林院隨即草詔：一面降赦天下罪囚，應有民間稅賦悉皆赦免；一面命在京宮觀寺院，修設好事禳災。

訓練分詞器

此處使用SentencePiece分詞方法，SentencePiece 是一種開源的子詞分詞工具，由 Google 研發。它的主要特點是語言無關和可逆性，能將任何語言的文本序列無損地轉換成子詞序列。

# path: scripts/train_tokenizer.py
"""
訓練 SentencePiece 分詞器 (BPE)
用法：
    python scripts/train_tokenizer.py data/shuihu.txt workdir/spm_shuihu 8000
"""


import sys
import os
import sentencepiece as spm


def train_tokenizer(input_path: str, model_prefix: str, vocab_size: int = 8000):
    if not os.path.exists(input_path):
        raise FileNotFoundError(f"語料文件不存在: {input_path}")


    print(f"[INFO] 開始訓練分詞器: {input_path}")
    spm.SentencePieceTrainer.Train(
        f"--input={input_path} "
        f"--model_prefix={model_prefix} "
        f"--vocab_size={vocab_size} "
        f"--model_type=bpe "
        f"--character_coverage=0.9995 "
        f"--bos_id=1 --eos_id=2 --unk_id=3"
    )
    print(f"[INFO] 分詞器已保存: {model_prefix}.model / {model_prefix}.vocab")


def main():
    if len(sys.argv) < 3:
        print("用法: python scripts/train_tokenizer.py 輸入文本 模型前綴 [詞表大小]")
        sys.exit(1)


    input_path = sys.argv[1]
    model_prefix = sys.argv[2]
    vocab_size = int(sys.argv[3]) if len(sys.argv) >= 4 else 8000


    train_tokenizer(input_path, model_prefix, vocab_size)


if __name__ == "__main__":
    main()

# 輸出 
[INFO] 開始訓練分詞器: data/shuihu.txt
sentencepiece_trainer.cc(178) LOG(INFO) Running command: --input=data/shuihu.txt --model_prefix=workdir/spm_shuihu --vocab_size=8000 --model_type=bpe --character_coverage=0.9995 --bos_id=1 --eos_id=2 --unk_id=3
sentencepiece_trainer.cc(78) LOG(INFO) Starts training with : 
trainer_spec {
  input: data/shuihu.txt
  input_format: 
  model_prefix: workdir/spm_shuihu
  model_type: BPE
  vocab_size: 8000
  self_test_sample_size: 0
  character_coverage: 0.9995
  input_sentence_size: 0
  shuffle_input_sentence: 1
  seed_sentencepiece_size: 1000000
  shrinking_factor: 0.75
  max_sentence_length: 4192
  num_threads: 16
  num_sub_iterations: 2
  max_sentencepiece_length: 16
  split_by_unicode_script: 1
  split_by_number: 1
  split_by_whitespace: 1
  split_digits: 0
  pretokenization_delimiter: 
  treat_whitespace_as_suffix: 0
  allow_whitespace_only_pieces: 0
  required_chars: 
  byte_fallback: 0
  vocabulary_output_piece_score: 1
  train_extremely_large_corpus: 0
  seed_sentencepieces_file: 
  hard_vocab_limit: 1
  use_all_vocab: 0
  unk_id: 3
  bos_id: 1
  eos_id: 2
  pad_id: -1
  unk_piece: <unk>
  bos_piece: <s>
  eos_piece: </s>
  pad_piece: <pad>
  unk_surface:  ? 
  enable_differential_privacy: 0
  differential_privacy_noise_level: 0
  differential_privacy_clipping_threshold: 0
}
normalizer_spec {
  name: nmt_nfkc
  add_dummy_prefix: 1
  remove_extra_whitespaces: 1
  escape_whitespaces: 1
  normalization_rule_tsv: 
}

訓練簡化版 GPT 模型(使用絕對位置編碼)

此處我們使用絕對位置編碼進行訓練，模型參數為8M，在RTX4090 上進行訓練，平均占用顯存1G內，訓練時間：45分鐘左右。

# path: scripts/train_gpt.py
"""
訓練簡化版 GPT 模型
用法：
    python scripts/train_gpt.py workdir/spm_shuihu.model data/shuihu.txt workdir/gpt_shuihu.pth
"""


import sys
import re
import math
import torch
from torch import nn
from torch.utils.data import Dataset, DataLoader
import sentencepiece as spm
from tqdm import tqdm


DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
BLOCK_SIZE = 128
BATCH_SIZE = 16
EPOCHS = 5
LR = 3e-4


def clean_text(text: str) -> str:
    allowed = re.compile(r"[^\u4e00-\u9fff。，、！？：；（）《》——…\n ]+")
    text = allowed.sub(" ", text)
    text = re.sub(r"\s+", " ", text)
    return text.strip()


class TextDataset(Dataset):
    def __init__(self, token_ids, block_size):
        self.ids = token_ids
        self.block_size = block_size


    def __len__(self):
        return max(0, len(self.ids) - self.block_size)


    def __getitem__(self, idx):
        x = torch.tensor(self.ids[idx: idx + self.block_size], dtype=torch.long)
        y = torch.tensor(self.ids[idx + 1: idx + 1 + self.block_size], dtype=torch.long)
        return x, y


class MultiHeadSelfAttention(nn.Module):
    def __init__(self, embed_dim, num_heads, attn_dropout=0.1):
        super().__init__()
        assert embed_dim % num_heads == 0
        self.head_dim = embed_dim // num_heads
        self.num_heads = num_heads
        self.qkv = nn.Linear(embed_dim, embed_dim * 3)
        self.out = nn.Linear(embed_dim, embed_dim)
        self.drop = nn.Dropout(attn_dropout)


    def forward(self, x):
        B, T, C = x.shape
        qkv = self.qkv(x).chunk(3, dim=-1)
        q, k, v = [t.view(B, T, self.num_heads, self.head_dim).transpose(1, 2) for t in qkv]
        att = (q @ k.transpose(-2, -1)) / math.sqrt(self.head_dim)
        mask = torch.tril(torch.ones(T, T, device=x.device)).unsqueeze(0).unsqueeze(0)
        att = att.masked_fill(mask == 0, float("-inf"))
        att = torch.softmax(att, dim=-1)
        att = self.drop(att)
        out = att @ v
        out = out.transpose(1, 2).contiguous().view(B, T, C)
        return self.out(out)


class FeedForward(nn.Module):
    def __init__(self, dim, hidden_dim, dropout=0.1):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(dim, hidden_dim),
            nn.GELU(),
            nn.Linear(hidden_dim, dim),
            nn.Dropout(dropout)
        )
    def forward(self, x): return self.net(x)


class TransformerBlock(nn.Module):
    def __init__(self, dim, num_heads, dropout=0.1):
        super().__init__()
        self.ln1 = nn.LayerNorm(dim)
        self.attn = MultiHeadSelfAttention(dim, num_heads, dropout)
        self.ln2 = nn.LayerNorm(dim)
        self.ff = FeedForward(dim, dim*4, dropout)


    def forward(self, x):
        x = x + self.attn(self.ln1(x))
        x = x + self.ff(self.ln2(x))
        return x


class GPTLike(nn.Module):
    def __init__(self, vocab_size, block_size, n_layers=2, dim=128, num_heads=4):
        super().__init__()
        self.token_emb = nn.Embedding(vocab_size, dim)
        self.pos_emb = nn.Embedding(block_size, dim)
        self.blocks = nn.ModuleList([TransformerBlock(dim, num_heads) for _ in range(n_layers)])
        self.ln = nn.LayerNorm(dim)
        self.head = nn.Linear(dim, vocab_size)
        self.block_size = block_size


    def forward(self, idx):
        B, T = idx.shape
        x = self.token_emb(idx) + self.pos_emb(torch.arange(T, device=idx.device))
        for block in self.blocks:
            x = block(x)
        return self.head(self.ln(x))




def train(model, dataset, epochs=EPOCHS, batch_size=BATCH_SIZE, lr=LR):
    loader = DataLoader(dataset, batch_size=batch_size, shuffle=True)
    opt = torch.optim.AdamW(model.parameters(), lr=lr)
    loss_fn = nn.CrossEntropyLoss()
    model.train()
    step = 0
    for ep in range(epochs):
        total_loss = 0
        pbar = tqdm(loader, desc=f"Epoch {ep+1}/{epochs}")
        for xb, yb in pbar:
            xb, yb = xb.to(DEVICE), yb.to(DEVICE)
            logits = model(xb)
            loss = loss_fn(logits.view(-1, logits.size(-1)), yb.view(-1))
            opt.zero_grad()
            loss.backward()
            opt.step()
            total_loss += loss.item()
            step += 1
            if step % 100 == 0:
                pbar.set_postfix(loss=f"{loss.item():.4f}")
        avg_loss = total_loss / len(loader)
        ppl = math.exp(avg_loss)
        print(f"[Epoch {ep+1}] Avg Loss {avg_loss:.4f} | PPL {ppl:.2f}")




def main():
    if len(sys.argv) < 4:
        print("用法: python scripts/train_gpt.py 分詞器模型 輸入語料 輸出模型")
        sys.exit(1)


    sp_model, corpus_path, out_path = sys.argv[1], sys.argv[2], sys.argv[3]
    sp = spm.SentencePieceProcessor(model_file=sp_model)


    with open(corpus_path, encoding="utf-8") as f:
        text = clean_text(f.read())
    ids = sp.encode(text, out_type=int)


    dataset = TextDataset(ids, BLOCK_SIZE)
    model = GPTLike(sp.get_piece_size(), BLOCK_SIZE).to(DEVICE)
    train(model, dataset)
    torch.save(model.state_dict(), out_path)
    print(f"模型已保存: {out_path}")


if __name__ == "__main__":
    main()

# 輸出 
Epoch 1/5: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 38619/38619 [08:52<00:00, 72.47it/s, loss=3.5714]
[Epoch 1] Avg Loss 4.4607 | PPL 86.55
Epoch 2/5: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 38619/38619 [08:54<00:00, 72.23it/s, loss=2.9516]
[Epoch 2] Avg Loss 3.2717 | PPL 26.36
Epoch 3/5: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 38619/38619 [08:51<00:00, 72.63it/s, loss=2.6000]
[Epoch 3] Avg Loss 2.8272 | PPL 16.90
Epoch 4/5: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 38619/38619 [08:54<00:00, 72.28it/s, loss=2.5234]
[Epoch 4] Avg Loss 2.5424 | PPL 12.71
Epoch 5/5: 100% |██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 38619/38619 [08:54<00:00, 72.28it/s, loss=2.2621]
[Epoch 5] Avg Loss 2.3380 | PPL 10.36
模型已保存: workdir/gpt_shuihu.pth

訓練簡化版 GPT 模型(使用旋轉位置編碼RoPE)

此處我們使用旋轉位置編碼進行訓練，模型參數為8M，在RTX4090 上進行訓練，平均占用顯存1G內，訓練時間：45分鐘左右。

# path: scripts/train_gpt_RoPE.py
"""
訓練簡化版 GPT 模型，使用旋轉位置編碼（RoPE）
用法：
    python scripts/train_gpt_RoPE.py workdir/spm_shuihu.model data/shuihu.txt workdir/gpt_shuihu_RoPE.pth
"""


import sys
import re
import math
import torch
from torch import nn
from torch.utils.data import Dataset, DataLoader
import sentencepiece as spm
from tqdm import tqdm


DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
BLOCK_SIZE = 128
BATCH_SIZE = 16
EPOCHS = 5
LR = 3e-4


def clean_text(text: str) -> str:
    allowed = re.compile(r"[^\u4e00-\u9fff。，、！？：；（）《》——…\n ]+")
    text = allowed.sub(" ", text)
    text = re.sub(r"\s+", " ", text)
    return text.strip()


class TextDataset(Dataset):
    def __init__(self, token_ids, block_size):
        self.ids = token_ids
        self.block_size = block_size


    def __len__(self):
        return max(0, len(self.ids) - self.block_size)


    def __getitem__(self, idx):
        x = torch.tensor(self.ids[idx: idx + self.block_size], dtype=torch.long)
        y = torch.tensor(self.ids[idx + 1: idx + 1 + self.block_size], dtype=torch.long)
        return x, y


def apply_rope(x, freqs):
    # x: (B, T, n_heads, head_dim)
    # freqs: (T, head_dim)
    # 拆分 x 為 x_real 和 x_imag
    x_real, x_imag = x.float().view(*x.shape[:-1], -1, 2).unbind(-1)


    # 將 freqs 擴展到和 x_real 一樣的形狀
    freqs = freqs.unsqueeze(0).unsqueeze(0)
    freqs_real, freqs_imag = freqs.view(*freqs.shape[:-1], -1, 2).unbind(-1)


    # 應用旋轉
    x_rotated_real = x_real * freqs_real - x_imag * freqs_imag
    x_rotated_imag = x_real * freqs_imag + x_imag * freqs_real


    x_rotated = torch.stack((x_rotated_real, x_rotated_imag), dim=-1).flatten(start_dim=-2)
    return x_rotated.type_as(x)




class MultiHeadSelfAttention(nn.Module):
    def __init__(self, embed_dim, num_heads, attn_dropout=0.1):
        super().__init__()
        assert embed_dim % num_heads == 0
        self.head_dim = embed_dim // num_heads
        self.num_heads = num_heads
        self.qkv = nn.Linear(embed_dim, embed_dim * 3)
        self.out = nn.Linear(embed_dim, embed_dim)
        self.drop = nn.Dropout(attn_dropout)
        self.register_buffer("freqs", self._create_freqs_buffer())


    def _create_freqs_buffer(self):
        # 創建旋轉頻率
        head_dim = self.head_dim
        pos = torch.arange(BLOCK_SIZE, dtype=torch.float32)
        dim = torch.arange(0, head_dim, 2, dtype=torch.float32)
        inv_freq = 1.0 / (10000 ** (dim / head_dim))
        freqs = torch.outer(pos, inv_freq)
        return torch.stack([torch.cos(freqs), torch.sin(freqs)], dim=-1).flatten(-2)


    def forward(self, x):
        B, T, C = x.shape
        qkv = self.qkv(x).chunk(3, dim=-1)
        q, k, v = [t.view(B, T, self.num_heads, self.head_dim).transpose(1, 2) for t in qkv]


        # 應用 RoPE 到 q 和 k
        q = apply_rope(q, self.freqs[:T].to(q.device))
        k = apply_rope(k, self.freqs[:T].to(k.device))


        att = (q @ k.transpose(-2, -1)) / math.sqrt(self.head_dim)
        mask = torch.tril(torch.ones(T, T, device=x.device)).unsqueeze(0).unsqueeze(0)
        att = att.masked_fill(mask == 0, float("-inf"))
        att = torch.softmax(att, dim=-1)
        att = self.drop(att)
        out = att @ v
        out = out.transpose(1, 2).contiguous().view(B, T, C)
        return self.out(out)


class FeedForward(nn.Module):
    def __init__(self, dim, hidden_dim, dropout=0.1):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(dim, hidden_dim),
            nn.GELU(),
            nn.Linear(hidden_dim, dim),
            nn.Dropout(dropout)
        )
    def forward(self, x): return self.net(x)


class TransformerBlock(nn.Module):
    def __init__(self, dim, num_heads, dropout=0.1):
        super().__init__()
        self.ln1 = nn.LayerNorm(dim)
        self.attn = MultiHeadSelfAttention(dim, num_heads, dropout)
        self.ln2 = nn.LayerNorm(dim)
        self.ff = FeedForward(dim, dim*4, dropout)


    def forward(self, x):
        x = x + self.attn(self.ln1(x))
        x = x + self.ff(self.ln2(x))
        return x


class GPTLike(nn.Module):
    def __init__(self, vocab_size, block_size, n_layers=2, dim=128, num_heads=4):
        super().__init__()
        # 移除原有的位置嵌入層
        self.token_emb = nn.Embedding(vocab_size, dim)
        self.blocks = nn.ModuleList([TransformerBlock(dim, num_heads) for _ in range(n_layers)])
        self.ln = nn.LayerNorm(dim)
        self.head = nn.Linear(dim, vocab_size)
        self.block_size = block_size


    def forward(self, idx):
        B, T = idx.shape
        # 直接使用 token 嵌入，位置信息在注意力層中處理
        x = self.token_emb(idx)
        for block in self.blocks:
            x = block(x)
        return self.head(self.ln(x))


def train(model, dataset, epochs=EPOCHS, batch_size=BATCH_SIZE, lr=LR):
    loader = DataLoader(dataset, batch_size=batch_size, shuffle=True)
    opt = torch.optim.AdamW(model.parameters(), lr=lr)
    loss_fn = nn.CrossEntropyLoss()
    model.train()
    step = 0
    for ep in range(epochs):
        total_loss = 0
        pbar = tqdm(loader, desc=f"Epoch {ep+1}/{epochs}")
        for xb, yb in pbar:
            xb, yb = xb.to(DEVICE), yb.to(DEVICE)
            logits = model(xb)
            loss = loss_fn(logits.view(-1, logits.size(-1)), yb.view(-1))
            opt.zero_grad()
            loss.backward()
            opt.step()
            total_loss += loss.item()
            step += 1
            if step % 100 == 0:
                pbar.set_postfix(loss=f"{loss.item():.4f}")
        avg_loss = total_loss / len(loader)
        ppl = math.exp(avg_loss)
        print(f"[Epoch {ep+1}] Avg Loss {avg_loss:.4f} | PPL {ppl:.2f}")




def main():
    if len(sys.argv) < 4:
        print("用法: python scripts/train_gpt_RoPE.py 分詞器模型 輸入語料 輸出模型")
        sys.exit(1)


    sp_model, corpus_path, out_path = sys.argv[1], sys.argv[2], sys.argv[3]
    sp = spm.SentencePieceProcessor(model_file=sp_model)


    with open(corpus_path, encoding="utf-8") as f:
        text = clean_text(f.read())
    ids = sp.encode(text, out_type=int)


    dataset = TextDataset(ids, BLOCK_SIZE)
    model = GPTLike(sp.get_piece_size(), BLOCK_SIZE).to(DEVICE)
    train(model, dataset)
    torch.save(model.state_dict(), out_path)
    print(f"模型已保存: {out_path}")


if __name__ == "__main__":
    main()

# 輸出 
Epoch 1/5: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 38619/38619 [11:08<00:00, 57.79it/s, loss=3.0254]
[Epoch 1] Avg Loss 3.9948 | PPL 54.31
Epoch 2/5: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 38619/38619 [11:14<00:00, 57.27it/s, loss=2.7059]
[Epoch 2] Avg Loss 2.9947 | PPL 19.98
Epoch 3/5: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 38619/38619 [11:11<00:00, 57.47it/s, loss=2.4179]
[Epoch 3] Avg Loss 2.6470 | PPL 14.11
Epoch 4/5: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 38619/38619 [11:12<00:00, 57.39it/s, loss=2.2415]
[Epoch 4] Avg Loss 2.4296 | PPL 11.35
Epoch 5/5: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 38619/38619 [11:07<00:00, 57.88it/s, loss=2.2116]
[Epoch 5] Avg Loss 2.2804 | PPL 9.78
模型已保存: workdir/gpt_shuihu_RoPE.pth

使用訓練的模型進行推理(絕對位置編碼)

我們使用訓練好的模型gpt_shuihu.pth進行推理測試。

# path: scripts/predict_gpt.py
"""
使用訓練好的 GPT 模型生成文本
用法：
    python scripts/predict_gpt.py workdir/spm_shuihu.model workdir/gpt_shuihu.pth "宋江在梁山泊"
"""


import sys
import torch
import sentencepiece as spm
from train_gpt import GPTLike, DEVICE, BLOCK_SIZE


@torch.no_grad()
def generate(model, sp, prompt, max_new_tokens=100, temperature=1.0, top_k=50):
    idx = torch.tensor([sp.encode(prompt, out_type=int)], device=DEVICE)
    for _ in range(max_new_tokens):
        idx_cond = idx[:, -BLOCK_SIZE:]
        logits = model(idx_cond)[:, -1, :] / temperature
        if top_k:
            v, _ = torch.topk(logits, top_k)
            logits[logits < v[:, [-1]]] = -1e10
        probs = torch.softmax(logits, dim=-1)
        next_id = torch.multinomial(probs, 1)
        idx = torch.cat([idx, next_id], dim=1)
    return sp.decode(idx[0].tolist())


def main():
    if len(sys.argv) < 4:
        print("用法: python scripts/predict_gpt.py 分詞器模型 已訓練模型 輸入提示")
        sys.exit(1)


    sp_model, model_path, prompt = sys.argv[1], sys.argv[2], sys.argv[3]
    sp = spm.SentencePieceProcessor(model_file=sp_model)


    vocab_size = sp.get_piece_size()
    model = GPTLike(vocab_size, BLOCK_SIZE).to(DEVICE)
    model.load_state_dict(torch.load(model_path, map_locatinotallow=DEVICE))
    model.eval()


    result = generate(model, sp, prompt)
    print("=== 輸入提示 ===")
    print(prompt)
    print("=== 生成結果 ===")
    print(result)


if __name__ == "__main__":
    main()

ColinAI# python scripts/predict_gpt.py workdir/spm_shuihu.model workdir/gpt_shuihu.pth "武松在景陽岡"
=== 輸入提示 ===
武松在景陽岡
=== 生成結果 ===
武松在景陽岡子上,只見那張蒙管營手里道: 你認得這個賊人,不敢來投奔他。 那個是開娘的孩兒,也不回了,卻來胡亂尋死吃過! 蔣門神在地下,那里肯放。 何九叔道: 中了菜,放囊的說出武松面前,望后便倒了。 武松道: 嫂嫂在此? 小人埋怨我,又吃棒,不值得饑便說道: 好歹教你一個。 你好不曉事! 武松道: 都頭休多說
ColinAI# python scripts/predict_gpt.py workdir/spm_shuihu.model workdir/gpt_shuihu.pth "魯智深"
=== 輸入提示 ===
魯智深
=== 生成結果 ===
魯智深皂羅巾。 張清把槍拈起弓,搭上箭,把馬一拍,把郝思文后心。 雷橫這匹馬,望本陣便走。 雷橫把弓卷了一箭,射射來掃蕩,只一石子來。 雷橫大怒,把這一所眼睜春心頭,便帶著行枷呀,直走到縣住左腳。 宋江仰鞭梢一指,接著去了。 朱仝回到寨中,接著晁蓋,分賓主而坐。 小衙內中并訴舊音仰著。 吳學究
ColinAI# python scripts/predict_gpt.py workdir/spm_shuihu.model workdir/gpt_shuihu.pth "林教頭在東京"
=== 輸入提示 ===
林教頭在東京
=== 生成結果 ===
林教頭在東京住雕欄、布列做一盤,紫綬金章;銀朱紅甲章,前逢龍袍并首副座。 隨班列著一應果子、金字匠金大堅,絹一套絹帛縛朱戶。 三廂房中揀日,造些油紙剪獅。 那新官初時營后與王慶、段三娘。 次后前踅過東,一帶高彪將,向后一個頭領二員:呼保義宋江、王義公、婁
ColinAI#

使用訓練的模型進行推理(旋轉位置編碼RoPE)

我們使用訓練好的模型gpt_shuihu_RoPE.pth進行推理測試。

# path: scripts/predict_gpt_RoPE.py
"""
使用訓練好的 GPT 模型（RoPE版本）生成文本
用法：
    python scripts/predict_gpt_RoPE.py workdir/spm_shuihu.model workdir/gpt_shuihu_RoPE.pth "林教頭在東京"
"""


import sys
import torch
import sentencepiece as spm


DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
BLOCK_SIZE = 128


# ----------------- RoPE 模型定義開始 -----------------
import math
from torch import nn


def apply_rope(x, freqs):
    x_real, x_imag = x.float().view(*x.shape[:-1], -1, 2).unbind(-1)
    freqs = freqs.unsqueeze(0).unsqueeze(0)
    freqs_real, freqs_imag = freqs.view(*freqs.shape[:-1], -1, 2).unbind(-1)
    x_rotated_real = x_real * freqs_real - x_imag * freqs_imag
    x_rotated_imag = x_real * freqs_imag + x_imag * freqs_imag
    x_rotated = torch.stack((x_rotated_real, x_rotated_imag), dim=-1).flatten(start_dim=-2)
    return x_rotated.type_as(x)


class MultiHeadSelfAttention(nn.Module):
    def __init__(self, embed_dim, num_heads, attn_dropout=0.1):
        super().__init__()
        assert embed_dim % num_heads == 0
        self.head_dim = embed_dim // num_heads
        self.num_heads = num_heads
        self.qkv = nn.Linear(embed_dim, embed_dim * 3)
        self.out = nn.Linear(embed_dim, embed_dim)
        self.drop = nn.Dropout(attn_dropout)
        self.register_buffer("freqs", self._create_freqs_buffer())


    def _create_freqs_buffer(self):
        head_dim = self.head_dim
        pos = torch.arange(BLOCK_SIZE, dtype=torch.float32)
        dim = torch.arange(0, head_dim, 2, dtype=torch.float32)
        inv_freq = 1.0 / (10000 ** (dim / head_dim))
        freqs = torch.outer(pos, inv_freq)
        return torch.stack([torch.cos(freqs), torch.sin(freqs)], dim=-1).flatten(-2)


    def forward(self, x):
        B, T, C = x.shape
        qkv = self.qkv(x).chunk(3, dim=-1)
        q, k, v = [t.view(B, T, self.num_heads, self.head_dim).transpose(1, 2) for t in qkv]


        q = apply_rope(q, self.freqs[:T].to(q.device))
        k = apply_rope(k, self.freqs[:T].to(k.device))


        att = (q @ k.transpose(-2, -1)) / math.sqrt(self.head_dim)
        mask = torch.tril(torch.ones(T, T, device=x.device)).unsqueeze(0).unsqueeze(0)
        att = att.masked_fill(mask == 0, float("-inf"))
        att = torch.softmax(att, dim=-1)
        att = self.drop(att)
        out = att @ v
        out = out.transpose(1, 2).contiguous().view(B, T, C)
        return self.out(out)


class FeedForward(nn.Module):
    def __init__(self, dim, hidden_dim, dropout=0.1):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(dim, hidden_dim),
            nn.GELU(),
            nn.Linear(hidden_dim, dim),
            nn.Dropout(dropout)
        )
    def forward(self, x): return self.net(x)


class TransformerBlock(nn.Module):
    def __init__(self, dim, num_heads, dropout=0.1):
        super().__init__()
        self.ln1 = nn.LayerNorm(dim)
        self.attn = MultiHeadSelfAttention(dim, num_heads, dropout)
        self.ln2 = nn.LayerNorm(dim)
        self.ff = FeedForward(dim, dim*4, dropout)


    def forward(self, x):
        x = x + self.attn(self.ln1(x))
        x = x + self.ff(self.ln2(x))
        return x


class GPTLike(nn.Module):
    def __init__(self, vocab_size, block_size, n_layers=2, dim=128, num_heads=4):
        super().__init__()
        # 沒有位置嵌入層
        self.token_emb = nn.Embedding(vocab_size, dim)
        self.blocks = nn.ModuleList([TransformerBlock(dim, num_heads) for _ in range(n_layers)])
        self.ln = nn.LayerNorm(dim)
        self.head = nn.Linear(dim, vocab_size)
        self.block_size = block_size


    def forward(self, idx):
        B, T = idx.shape
        x = self.token_emb(idx)
        for block in self.blocks:
            x = block(x)
        return self.head(self.ln(x))
# ----------------- RoPE 模型定義結束 -----------------


@torch.no_grad()
def generate(model, sp, prompt, max_new_tokens=100, temperature=1.0, top_k=50):
    idx = torch.tensor([sp.encode(prompt, out_type=int)], device=DEVICE)
    for _ in range(max_new_tokens):
        idx_cond = idx[:, -BLOCK_SIZE:]
        logits = model(idx_cond)[:, -1, :] / temperature
        if top_k:
            v, _ = torch.topk(logits, top_k)
            logits[logits < v[:, [-1]]] = -1e10
        probs = torch.softmax(logits, dim=-1)
        next_id = torch.multinomial(probs, 1)
        idx = torch.cat([idx, next_id], dim=1)
    return sp.decode(idx[0].tolist())


def main():
    if len(sys.argv) < 4:
        print("用法: python scripts/predict_gpt_RoPE.py 分詞器模型 已訓練模型 輸入提示")
        sys.exit(1)


    sp_model, model_path, prompt = sys.argv[1], sys.argv[2], sys.argv[3]
    sp = spm.SentencePieceProcessor(model_file=sp_model)


    vocab_size = sp.get_piece_size()
    # 實例化 RoPE 版本的模型
    model = GPTLike(vocab_size, BLOCK_SIZE).to(DEVICE)
    model.load_state_dict(torch.load(model_path, map_locatinotallow=DEVICE))
    model.eval()


    result = generate(model, sp, prompt)
    print("=== 輸入提示 ===")
    print(prompt)
    print("=== 生成結果 ===")
    print(result)


if __name__ == "__main__":
    main()

ColinAI# python scripts/predict_gpt_RoPE.py workdir/spm_shuihu.model workdir/gpt_shuihu_RoPE.pth "林教頭在東京"
=== 輸入提示 ===
林教頭在東京
=== 生成結果 ===
林教頭在東京住了師父。 智深道: 師父休要有些尷尬一般。 走了馬,不提防語,且教庫吏軍漢,取了財帛,裝載上車,犒賞三軍盡做合后,仍送上梁山泊來使,見今引了十數個軍官武德大海軍州以殺之恩;梁山泊未有可出我等進身,到得深村便是殺大蟲,累蒙恩橋下。 更兼做甚么都在那里壞國家有何樂如何? 老丈道: 娘兒兩個新年進
ColinAI# python scripts/predict_gpt_RoPE.py workdir/spm_shuihu.model workdir/gpt_shuihu_RoPE.pth "魯智深"
=== 輸入提示 ===
魯智深
=== 生成結果 ===
魯智深跳將起來,提了禪杖大篦帳,一齊去那樹邊,高聲朗山勇在馬上,如清迎前來。 盧俊義喝采道: 好酒! 引李云有何不可。 宋江扯得地,不是肚皮,口里搖著,口里道: 阿呀,苦也。 哭罷,三人吃過口則藏寺,那漢子使做柴元只在包著,踏口鑿暗暗的耍的性命。 說時已有七八十斤兩眼也不敢殺
ColinAI# python scripts/predict_gpt_RoPE.py workdir/spm_shuihu.model workdir/gpt_shuihu_RoPE.pth "武松在景陽岡"
=== 輸入提示 ===
武松在景陽岡
=== 生成結果 ===
武松在景陽岡子上,推檐下坐地。 武松叫土兵篩來,一面篩酒。 那李鬼坐在廳上坐地。 武松就房里桌子上摸上樓開房門,叫聲你這盆時,被這雪只顧亂攛,將斧喝道: 你這兩個扛我吊在床邊一個,老爺去那嘴,把右手只一掣出青花滾與他出個躲一搠樸刀,一斧來。 眾人近前都饒,打得臉打碎那個鼓鬧黑旋風來。 兩邊眾莊客
ColinAI#

代碼已放在GitHub：

https://github.com/ColinAIAPP/MoiraiLM

結束語

經過訓練后，我們對這個小型類 GPT 模型進行了推理測試。結果顯示，模型往往會輸出一些“胡言亂語”式的文本(預測下一個token)，看起來并沒有連貫的語義。這種現象背后有幾個主要原因：

模型規模過小

我們的實驗模型只有很小的參數，而真實的模型起步就是上億參數，ChatGPT、GPT-4 更是數千億參數量級。過小的模型容量限制了它對語言規律的表達能力。

訓練數據有限

我們僅使用了《水滸傳》這一部作品作為訓練語料，而現代大模型的訓練數據規模往往是萬億級別 token，涵蓋多領域、多語言。數據多樣性不足，使得模型無法學到更廣泛的語言知識。

訓練時間不足

在有限硬件和時間下，我們只進行了少量 epoch 的訓練。這種“淺嘗輒止”的訓練無法讓模型充分收斂，也難以形成較強的生成能力。

因此，這個實驗模型只能算是原理性驗證 ——它向我們展示了 GPT 模型“預測下一個詞”的核心工作方式，卻無法產生高質量的文本。本次實驗主要是測試預訓練，驗證模型框架和訓練流程的可行性。

在下一步，我們計劃在更大的數據集上進行訓練，并嘗試增加模型的深度和寬度，提升參數規模。同時，將引入監督微調（Supervised Fine-Tuning）和 RLHF（Reinforcement Learning with Human Feedback）等訓練策略，以進一步提升模型在生成文本時的質量、連貫性和實用性。在這些改進下，模型的表現將更加接近我們熟悉的大語言模型，實現更高水平的文本生成能力。

責任編輯：龐桂玉來源：寫代碼的中年人

大模型 GPT模型 AI 人工智能