1. 先搞清楚我们要复现的到底是什么很多人一提到“从零构建大模型”脑子里立刻浮现出几千张 GPU 和几千万美元预算。其实真正卡住个人开发者的从来不是算力而是不清楚一条完整流水线到底分几步、每一步的输入输出是什么、哪一步最容易翻车。GPT、Claude、Llama 这些模型底层走的都是同一套五阶段流程数据清洗、分词、预训练、对齐、评测。架构只是第三阶段里的一个模块真正拉开差距的是另外四步。这篇面向想动手复现的开发者目标不是训出 GPT-4 的克隆体而是跑通一条最小可用流水线用 WikiText 这类公开语料在单卡甚至 Colab 上训出一个千万参数级、能生成连贯文本的小模型再通过 TaoToken 的统一 Key/API 通道把推理环节接进来省去自己搭服务、管多套密钥的麻烦。全程给出可复制的config.toml骨架和逐阶段验证动作你照着敲就能看到困惑度从几百掉到几十的过程。适合谁有 Python 基础、想真正理解大模型内部机制、又不满足于只调 API 的开发者。下面按流水线顺序拆每一步都配可运行代码和“怎么判断这步做对了”的验证动作。2. 前置准备环境、依赖与 TaoToken 通道2.1 环境与依赖清单先把地基打好。我建议用 Python 3.10CUDA 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.tomltokenizers负责 BPE 分词datasets拉公开语料transformers主要用来做评测时的对照。别一上来就装一堆训练框架最小依赖反而更容易定位问题。2.2 为什么推理环节要接 TaoToken训练和推理是两件事。你辛苦训完一个小模型想验证它生成效果或者想拿 GPT、Claude 这类成熟模型做对照评测就得有稳定的 API 通道。自己维护多套密钥、处理不同厂商的接口差异很耗精力。TaoToken 提供统一的 Key 和 API 入口把模型对话、编码计划、密钥管理都收在一个控制台里接入时只改 base_url 和 key 两个字段。官网入口在这里https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。API 地址是 https://taotoken.net/api 注意这个不带 UTM 参数直接填进配置即可。2.3 拿 Key 与选通道登录后进控制台创建 API Key然后根据你的用途选通道只是验证模型生成效果、做对照评测 → 用模型对话通道deep linkhttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite长期做编码、跑 Agent 类任务 → 用 Coding Plandeep linkhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite管理多把密钥、看用量 → 控制台deep linkhttps://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 遇到参数问题先翻这里。3. 五阶段流水线的可复制配置骨架3.1 阶段一数据清洗原始网页文本没法直接训。Common Crawl 那种量级里重复页眉、垃圾内容、低质量短页占大头。最小可用版本我们用 WikiText-103它已经做过基础清洗适合练手。但清洗逻辑你要理解因为换成自己的语料时全靠它。from datasets import load_dataset dataset load_dataset(wikitext, wikitext-103-v1, splittrain) 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, encodingutf-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_size8000, special_tokens[PAD, BOS, EOS, UNK] ) tokenizer.train([corpus.txt], trainertrainer) 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, biasFalse) self.proj nn.Linear(embed_dim, embed_dim, biasFalse) 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, devicex.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, dropout0.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, dropout0.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, biasFalse) self.dropout nn.Dropout(dropout) def forward(self, token_ids): B, T token_ids.shape positions torch.arange(T, devicetoken_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_size8000, embed_dim256, num_heads8, ff_dim1024, num_layers4, max_seq_len256).to(device) print(f参数量: {sum(p.numel() for p in model.parameters()):,}) optimizer optim.AdamW(model.parameters(), lr3e-4, weight_decay0.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_norm1.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(reductionsum) 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 约 800Epoch 5 约 120Epoch 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, modelgpt-4o-mini): resp requests.post( f{BASE_URL}/v1/chat/completions, headers{ Authorization: fBearer {API_KEY}, Content-Type: application/json, }, json{ model: model, messages: [{role: user, content: prompt}], temperature: 0.8, }, timeout60, ) return resp.json()[choices][0][message][content] print(chat(用一句话解释什么是困惑度。))验证动作同一个 prompt 分别喂给你的 MiniLLM 和 TaoToken 通道里的成熟模型对比输出。你的模型可能语法通顺但事实错误这正是预训练数据规模差距的体现。4.2 本地生成函数你自己的模型生成文本核心是温度采样def generate(model, tokenizer, prompt, max_new_tokens100, temperature0.8, devicecuda): model.eval() token_ids tokenizer.encode(prompt).ids input_tensor torch.tensor(token_ids, dtypetorch.long, devicedevice).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_samples1) input_tensor torch.cat([input_tensor, next_token], dim1) 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_norm1.0必须开。5.2 困惑度不下降八成是数据问题。检查清洗后的语料是不是重复度太高或者分词后 token 分布过于集中。另一个可能是模型规模和数据量不匹配参数太多数据太少会欠拟合。经验比例每个参数约需 20 个 token 的训练数据。5.3 生成结果全是重复词温度太低或模型欠训练。先把 temperature 调到 0.8再确认训练 epoch 够不够。如果还是重复检查位置编码有没有正确加上。5.4 TaoToken 请求 401Key 没填对或没带Bearer前缀。确认 header 是Authorization: Bearer key。如果还报错去接入文档核对参数https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。5.5 显存不够把 batch_size 降到 16 或 8block_size 从 256 降到 128。模型参数量也可以从 1500 万降到 800 万先跑通再放大。6. 继续往下走跑通这条最小流水线后你会发现真正决定模型好坏的是数据质量、scaling 数学、对齐策略和诚实的评测架构只是其中一段。想把这套流程用到垂直领域比如代码助手、SQL 生成、法律文书摘要换的只是训练数据分词配置、Transformer 结构、训练循环、评测方法全部复用。如果你要长期做编码类任务或跑 Agent建议直接上 Coding Plan省去自己管密钥和通道的麻烦https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。需要管理多把 Key 或看用量进控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。接入过程中遇到报错先翻接入文档大部分参数问题那里都有答案。