从零构建你自己的大模型：GPT 与 Claude 背后的 5 阶段流水线实战拆解

1. 先搞清楚：我们要复现的到底是什么

很多人一提到“从零构建大模型”，脑子里立刻浮现出几千张 GPU 和几千万美元预算。其实真正卡住个人开发者的，从来不是算力，而是不清楚一条完整流水线到底分几步、每一步的输入输出是什么、哪一步最容易翻车。GPT、Claude、Llama 这些模型，底层走的都是同一套五阶段流程：数据清洗、分词、预训练、对齐、评测。架构只是第三阶段里的一个模块，真正拉开差距的是另外四步。

这篇面向想动手复现的开发者，目标不是训出 GPT-4 的克隆体，而是跑通一条最小可用流水线：用 WikiText 这类公开语料，在单卡甚至 Colab 上训出一个千万参数级、能生成连贯文本的小模型，再通过 TaoToken 的统一 Key/API 通道把推理环节接进来，省去自己搭服务、管多套密钥的麻烦。全程给出可复制的config.toml骨架和逐阶段验证动作，你照着敲就能看到困惑度从几百掉到几十的过程。

适合谁：有 Python 基础、想真正理解大模型内部机制、又不满足于只调 API 的开发者。下面按流水线顺序拆，每一步都配可运行代码和“怎么判断这步做对了”的验证动作。

2. 前置准备：环境、依赖与 TaoToken 通道

2.1 环境与依赖清单

先把地基打好。我建议用 Python 3.10+，CUDA 11.8 或 12.1 都行，单卡 8GB 显存足够跑本文的 1500 万参数模型。依赖装这几个就够：

pip install torch --index-url https://download.pytorch.org/whl/cu121 pip install tokenizers datasets transformers pip install tomli # 读取 config.toml

tokenizers负责 BPE 分词，datasets拉公开语料，transformers主要用来做评测时的对照。别一上来就装一堆训练框架，最小依赖反而更容易定位问题。

2.2 为什么推理环节要接 TaoToken

训练和推理是两件事。你辛苦训完一个小模型，想验证它生成效果，或者想拿 GPT、Claude 这类成熟模型做对照评测，就得有稳定的 API 通道。自己维护多套密钥、处理不同厂商的接口差异，很耗精力。TaoToken 提供统一的 Key 和 API 入口，把模型对话、编码计划、密钥管理都收在一个控制台里，接入时只改 base_url 和 key 两个字段。

官网入口在这里：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。API 地址是 https://taotoken.net/api ，注意这个不带 UTM 参数，直接填进配置即可。

2.3 拿 Key 与选通道

登录后进控制台创建 API Key，然后根据你的用途选通道：

• 只是验证模型生成效果、做对照评测 → 用模型对话通道，deep link：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite
• 长期做编码、跑 Agent 类任务 → 用 Coding Plan，deep link：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite
• 管理多把密钥、看用量 → 控制台，deep link：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite

接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，遇到参数问题先翻这里。

3. 五阶段流水线的可复制配置骨架

3.1 阶段一：数据清洗

原始网页文本没法直接训。Common Crawl 那种量级里，重复页眉、垃圾内容、低质量短页占大头。最小可用版本我们用 WikiText-103，它已经做过基础清洗，适合练手。但清洗逻辑你要理解，因为换成自己的语料时全靠它。

from datasets import load_dataset dataset = load_dataset("wikitext", "wikitext-103-v1", split="train") texts = [t.strip() for t in dataset["text"] if len(t.strip().split()) > 20] # 简单去重：按行哈希 seen = set() clean = [] for t in texts: h = hash(t[:200]) if h not in seen: seen.add(h) clean.append(t) with open("corpus.txt", "w", encoding="utf-8") as f: f.write("\n\n".join(clean[:200000])) # 先用 20 万段启动 print(f"清洗后段落数: {len(clean)}")

验证动作：打印前 5 段，确认没有 HTML 标签残留、没有连续空行、没有超短碎片。数据质量永远胜过数量，这一步偷懒，后面训多久都白搭。

3.2 阶段二：分词

模型不读原始文本，读 token。BPE 从单字符开始，反复合并高频字符对，直到词汇表达到设定大小。

from tokenizers import Tokenizer, models, trainers, pre_tokenizers tokenizer = Tokenizer(models.BPE()) tokenizer.pre_tokenizer = pre_tokenizers.Whitespace() trainer = trainers.BpeTrainer( vocab_size=8000, special_tokens=["<PAD>", "<BOS>", "<EOS>", "<UNK>"] ) tokenizer.train(["corpus.txt"], trainer=trainer) tokenizer.save("tokenizer.json") out = tokenizer.encode("Building an LLM from scratch is powerful") print(out.tokens) print(out.ids)

验证动作：编码一句英文，看 token 列表是否合理切分。经验法则：1 token ≈ 0.75 个英文单词。如果你的词汇表里出现大量单字符 token，说明语料太小或 vocab_size 设太大。

3.3 阶段三：预训练

整个训练目标简单到出人意料：预测下一个 token。给定 "The cat sat on the"，预测 "mat"。在足够多数据上重复这件事，语法、事实、推理会逐步涌现。

先写config.toml，把超参集中管理：

[model] vocab_size = 8000 embed_dim = 256 num_heads = 8 ff_dim = 1024 num_layers = 4 max_seq_len = 256 dropout = 0.1 [training] batch_size = 32 learning_rate = 3e-4 weight_decay = 0.01 max_epochs = 20 grad_clip = 1.0 warmup_steps = 500 [data] corpus_path = "corpus.txt" tokenizer_path = "tokenizer.json" block_size = 256 [checkpoint] save_dir = "checkpoints" save_every = 2

模型定义用纯解码器 Transformer，因果掩码保证只能看过去：

import torch import torch.nn as nn import math class CausalSelfAttention(nn.Module): def __init__(self, embed_dim, num_heads): super().__init__() self.num_heads = num_heads self.head_dim = embed_dim // num_heads self.qkv = nn.Linear(embed_dim, 3 * embed_dim, bias=False) self.proj = nn.Linear(embed_dim, embed_dim, bias=False) def forward(self, x): B, T, C = x.shape q, k, v = self.qkv(x).chunk(3, dim=-1) q = q.view(B, T, self.num_heads, self.head_dim).transpose(1, 2) k = k.view(B, T, self.num_heads, self.head_dim).transpose(1, 2) v = v.view(B, T, self.num_heads, self.head_dim).transpose(1, 2) scale = math.sqrt(self.head_dim) attn = (q @ k.transpose(-2, -1)) / scale mask = torch.tril(torch.ones(T, T, device=x.device)) attn = attn.masked_fill(mask == 0, float("-inf")) attn = torch.softmax(attn, dim=-1) out = (attn @ v).transpose(1, 2).contiguous().view(B, T, C) return self.proj(out) class TransformerBlock(nn.Module): def __init__(self, embed_dim, num_heads, ff_dim, dropout=0.1): super().__init__() self.attn = CausalSelfAttention(embed_dim, num_heads) self.ff = nn.Sequential( nn.Linear(embed_dim, ff_dim), nn.GELU(), nn.Linear(ff_dim, embed_dim), nn.Dropout(dropout), ) self.ln1 = nn.LayerNorm(embed_dim) self.ln2 = nn.LayerNorm(embed_dim) def forward(self, x): x = x + self.attn(self.ln1(x)) x = x + self.ff(self.ln2(x)) return x class MiniLLM(nn.Module): def __init__(self, vocab_size, embed_dim, num_heads, ff_dim, num_layers, max_seq_len, dropout=0.1): super().__init__() self.token_emb = nn.Embedding(vocab_size, embed_dim) self.pos_emb = nn.Embedding(max_seq_len, embed_dim) self.blocks = nn.ModuleList([ TransformerBlock(embed_dim, num_heads, ff_dim, dropout) for _ in range(num_layers) ]) self.ln_final = nn.LayerNorm(embed_dim) self.output = nn.Linear(embed_dim, vocab_size, bias=False) self.dropout = nn.Dropout(dropout) def forward(self, token_ids): B, T = token_ids.shape positions = torch.arange(T, device=token_ids.device).unsqueeze(0) x = self.dropout(self.token_emb(token_ids) + self.pos_emb(positions)) for block in self.blocks: x = block(x) x = self.ln_final(x) return self.output(x)

训练循环，注意梯度裁剪防爆炸：

import torch.optim as optim from torch.nn.utils import clip_grad_norm_ device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model = MiniLLM(vocab_size=8000, embed_dim=256, num_heads=8, ff_dim=1024, num_layers=4, max_seq_len=256).to(device) print(f"参数量: {sum(p.numel() for p in model.parameters()):,}") optimizer = optim.AdamW(model.parameters(), lr=3e-4, weight_decay=0.01) criterion = nn.CrossEntropyLoss() def train_epoch(model, dataloader): model.train() total_loss = 0 for input_ids, target_ids in dataloader: input_ids = input_ids.to(device) target_ids = target_ids.to(device) logits = model(input_ids) loss = criterion(logits.view(-1, logits.size(-1)), target_ids.view(-1)) optimizer.zero_grad() loss.backward() clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() total_loss += loss.item() return total_loss / len(dataloader)

验证动作：跑一个 epoch，看 loss 是否稳定下降。如果 loss 变成 NaN，先查学习率是不是太大，再查数据里有没有空样本。

3.4 阶段四：对齐

预训练完的模型是个“文本预测器”，你问它问题，它可能用三个问题回你。对齐分两步：SFT 教格式，RLHF 教偏好。最小版本先做 SFT，几千条样本就够，因为知识已经在预训练权重里了。

sft_examples = [ { "prompt": "用简单的语言解释什么是 API。", "response": "API 就像餐厅里的服务员。你告诉服务员想要什么，服务员去厨房取来再端给你。" }, { "prompt": "法国的首都是哪里？", "response": "法国的首都是巴黎。" }, ] # 格式化为 <prompt> [SEP] <response> <EOS> def format_sft(ex): return f"{ex['prompt']} [SEP] {ex['response']} <EOS>" sft_texts = [format_sft(e) for e in sft_examples]

验证动作：SFT 后拿几个训练里没出现的 prompt 测，看模型是否按“回答”格式输出，而不是继续续写问题。

3.5 阶段五：评测

预训练阶段看困惑度，它衡量模型对真实文本的“惊讶程度”，越低越好。

import math def calculate_perplexity(model, dataloader, device): model.eval() total_loss = 0 total_tokens = 0 criterion = nn.CrossEntropyLoss(reduction="sum") with torch.no_grad(): for input_ids, target_ids in dataloader: input_ids = input_ids.to(device) target_ids = target_ids.to(device) logits = model(input_ids) loss = criterion(logits.view(-1, logits.size(-1)), target_ids.view(-1)) total_loss += loss.item() total_tokens += target_ids.numel() return math.exp(total_loss / total_tokens)

验证动作：记录每个 epoch 的困惑度。参考区间：Epoch 1 约 800，Epoch 5 约 120，Epoch 20 约 23。如果一直卡在几百不降，回去查数据清洗和分词。

注意：对齐之后困惑度会失效，微调后的模型困惑度反而更差但更有用。这时要换人类 benchmark，比如 MMLU 测知识广度、Chatbot Arena 做盲评。同一个模型换个 prompt 格式，benchmark 分数可能从 0.637 掉到 0.488，评测本身就是难题。

4. 接入 TaoToken 验证推理与对照

4.1 用统一通道做生成对照

训完模型，你想拿 GPT、Claude 做对照，看自己的小模型差在哪。用 TaoToken 的模型对话通道，一个 base_url 搞定：

import requests API_KEY = "你的_taotoken_key" BASE_URL = "https://taotoken.net/api" def chat(prompt, model="gpt-4o-mini"): resp = requests.post( f"{BASE_URL}/v1/chat/completions", headers={ "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", }, json={ "model": model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.8, }, timeout=60, ) return resp.json()["choices"][0]["message"]["content"] print(chat("用一句话解释什么是困惑度。"))

验证动作：同一个 prompt 分别喂给你的 MiniLLM 和 TaoToken 通道里的成熟模型，对比输出。你的模型可能语法通顺但事实错误，这正是预训练数据规模差距的体现。

4.2 本地生成函数

你自己的模型生成文本，核心是温度采样：

def generate(model, tokenizer, prompt, max_new_tokens=100, temperature=0.8, device="cuda"): model.eval() token_ids = tokenizer.encode(prompt).ids input_tensor = torch.tensor(token_ids, dtype=torch.long, device=device).unsqueeze(0) with torch.no_grad(): for _ in range(max_new_tokens): context = input_tensor[:, -256:] logits = model(context) next_logits = logits[:, -1, :] / temperature probs = torch.softmax(next_logits, dim=-1) next_token = torch.multinomial(probs, num_samples=1) input_tensor = torch.cat([input_tensor, next_token], dim=1) if next_token.item() == tokenizer.token_to_id("<EOS>"): break return tokenizer.decode(input_tensor[0].tolist())

温度参数控制创造力：0.1 安全可预测，0.8 自然多样，1.5 有创意但可能语无伦次。默认用 0.8。

5. 本篇常见错排查

5.1 loss 变 NaN

最常见原因是学习率过大或数据里有空样本。先把 lr 降到 1e-4 试，再检查corpus.txt里有没有长度为 0 的行。梯度裁剪max_norm=1.0必须开。

5.2 困惑度不下降

八成是数据问题。检查清洗后的语料是不是重复度太高，或者分词后 token 分布过于集中。另一个可能是模型规模和数据量不匹配，参数太多数据太少会欠拟合。经验比例：每个参数约需 20 个 token 的训练数据。

5.3 生成结果全是重复词

温度太低或模型欠训练。先把 temperature 调到 0.8，再确认训练 epoch 够不够。如果还是重复，检查位置编码有没有正确加上。

5.4 TaoToken 请求 401

Key 没填对或没带Bearer前缀。确认 header 是Authorization: Bearer <key>。如果还报错，去接入文档核对参数：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。

5.5 显存不够

把 batch_size 降到 16 或 8，block_size 从 256 降到 128。模型参数量也可以从 1500 万降到 800 万，先跑通再放大。

6. 继续往下走

跑通这条最小流水线后，你会发现真正决定模型好坏的，是数据质量、scaling 数学、对齐策略和诚实的评测，架构只是其中一段。想把这套流程用到垂直领域，比如代码助手、SQL 生成、法律文书摘要，换的只是训练数据，分词配置、Transformer 结构、训练循环、评测方法全部复用。

如果你要长期做编码类任务或跑 Agent，建议直接上 Coding Plan，省去自己管密钥和通道的麻烦：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。需要管理多把 Key 或看用量，进控制台：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite 。接入过程中遇到报错，先翻接入文档，大部分参数问题那里都有答案。