当前位置：首页 > news >正文

从零训练LLM-1.训练BPE

news 来源：原创 2025/9/6 10:02:49

文章目录

- BPE 简介
- BPE (Byte-Pair Encoding) 算法训练流程
- BPE 编码流程
- BPE 评估
- 代码
参考

本文基于 HF -tokenizer 训练，更便捷

BPE 简介

分词器将单词从自然语言通过“词典”映射到0, 1, 36这样的数字，可以理解为数字就代表了单词在“词典”中的页码。可以选择自己构造词表训练一个“词典”或者选择比较出名的开源大模型分词器（直接将 tokenizer 的模型文件复制过来，然后 tokenizer.from_pretrained），正如同直接用新华/牛津词典的优点是token编码压缩率很好，缺点是页数太多，动辄数十万个词汇短语；自己训练的分词器，优点是词表长度和内容随意控制，缺点是压缩率很低（例如"hello"也许会被拆分为"h e l l o" 五个独立的token），且生僻词难以覆盖。 “词典”的选择固然很重要，LLM的输出本质上是SoftMax到词典N个词的多分类问题，然后通过“词典”解码到自然语言。因为模型体积需要严格控制，为了避免模型头重脚轻（词嵌入embedding层参数在LLM占比太高），所以词表长度短益善。

BPE (Byte-Pair Encoding) 算法训练流程

准备语料库，将文本拆分为基本单位(通常是字符或字节)
统计所有相邻单元对的频率
选择最高频率的单元对合并为新的单元
更新语料库，替换所有该单元对为新单元
重复步骤2-4直到达到预设的词表大小或合并次数

BPE 编码流程

把文本先拆成最小单位(比如单个字母或字节)
识别文本中的特殊token(如, )，这些特殊token会直接映射到对应ID
对于普通文本部分:

a. 初始化为基本单位序列

b. 扫描当前序列中的所有相邻单位对

c. 查找这些单位对是否在训练时学到的"合并规则表"中

d. 优先选择合并后ID值最小的单位对进行合并(即优先合并更基础、更短的单位)

e. 合并后继续重复b-d步骤，直到没有可合并的单位对
将特殊token和处理好的普通文本部分组合，得到完整的token序列
最后得到的单位序列就是文本的token编码结果

BPE 评估

评估 BPE 分词器质量的主要方法包括:

分词质量指标

词表利用率: 检查分词后的词表中各 token 使用频率分布，好的分词器应当避免出现大量低频或未使用的 token
平均 token 长度: 较好的分词器通常能将文本压缩为更少的 token
未知词率 (OOV率): 在测试集上计算未登录词（需用表示）的比例，越低越好

下游任务评估

困惑度 (Perplexity): 在语言模型上测试，较低的困惑度表示分词更有效
下游任务表现: 在翻译、分类等任务中比较不同分词方案的效率差异

def eval_tokenizer():# 加载预训练的tokenizertokenizer = AutoTokenizer.from_pretrained("./EmoLLM_tokenizer")# 准备测试数据test_texts = ["你好，我最近感到有些焦虑，不知道该怎么办。","我对未来感到迷茫，希望能得到一些建议。","最近工作压力很大，感觉睡不好觉。"]# 1. 计算平均token长度token_lengths = []for text in test_texts:tokens = tokenizer.encode(text)token_lengths.append(len(tokens))avg_token_length = sum(token_lengths) / len(token_lengths)print(f"平均token长度: {avg_token_length:.2f}")# 2. 检查词表利用率all_tokens = []for text in test_texts:tokens = tokenizer.encode(text)all_tokens.extend(tokens)unique_tokens = set(all_tokens)vocab_size = len(tokenizer)utilization_rate = len(unique_tokens) / vocab_size * 100print(f"词表利用率: {utilization_rate:.2f}%")# 3. 分析最常见的tokensfrom collections import Countertoken_counter = Counter(all_tokens)print("最常见的10个token:")for token_id, count in token_counter.most_common(10):token = tokenizer.decode([token_id])print(f"  Token: '{token}', ID: {token_id}, 出现次数: {count}")# 4. 检查长文本的编码/解码一致性long_text = " ".join(test_texts)encoded = tokenizer.encode(long_text)decoded = tokenizer.decode(encoded)print(f"编码/解码一致性: {long_text == decoded}")

也可以与其他 tokenizer 进行比较

def compare_tokenizers(test_text):# 加载不同的tokenizer进行比较tokenizers = {"EmoLLM": AutoTokenizer.from_pretrained("./EmoLLM_tokenizer"),"GPT2": AutoTokenizer.from_pretrained("gpt2"),"BERT": AutoTokenizer.from_pretrained("bert-base-uncased")}print(f"文本: {test_text}")for name, tok in tokenizers.items():tokens = tok.encode(test_text)print(f"{name} token数量: {len(tokens)}")# 可视化分词结果if hasattr(tok, "tokenize"):print(f"{name} 分词结果: {tok.tokenize(test_text)}")else:print(f"{name} 分词结果: {[tok.decode([t]) for t in tokens]}")

代码

import json
import os
import randomfrom tokenizers import (Tokenizer, decoders, models, normalizers,pre_tokenizers, processors, trainers)
from transformers import AutoTokenizer# 设置随机种子以确保结果可复现
random.seed(42)# BPE (Byte-Pair Encoding) 算法训练流程:
# 1. 准备语料库，将文本拆分为基本单位(通常是字符或字节)
# 2. 统计所有相邻单元对的频率
# 3. 选择最高频率的单元对合并为新的单元
# 4. 更新语料库，替换所有该单元对为新单元
# 5. 重复步骤2-4直到达到预设的词表大小或合并次数# BPE 编码流程:
# 1. 把文本先拆成最小单位(比如单个字母或字节)
# 2. 识别文本中的特殊token(如<s>, </s>)，这些特殊token会直接映射到对应ID
# 3. 对于普通文本部分:
#    a. 初始化为基本单位序列
#    b. 扫描当前序列中的所有相邻单位对
#    c. 查找这些单位对是否在训练时学到的"合并规则表"中
#    d. 优先选择合并后ID值最小的单位对进行合并(即优先合并更基础、更短的单位)
#    e. 合并后继续重复b-d步骤，直到没有可合并的单位对
# 4. 将特殊token和处理好的普通文本部分组合，得到完整的token序列
# 5. 最后得到的单位序列就是文本的token编码结果def train_tokenizer():# 读取JSONL文件并提取文本数据def read_texts_from_jsonl(file_path):with open(file_path, 'r', encoding='utf-8') as f:for line in f:data = json.loads(line)yield data['text']data_path = '../dataset/pretrain_hq.jsonl'# 初始化tokenizertokenizer = Tokenizer(models.BPE())# 预分词器负责将原始文本进行初步分割，为后续的 BPE 合并操作做准备add_prefix_space=False 表示不在每个序列前添加空格tokenizer.pre_tokenizer = pre_tokenizers.ByteLevel(add_prefix_space=False)# 定义特殊tokenspecial_tokens = ["<unk>", "<s>", "</s>"]# 设置训练器并添加特殊tokentrainer = trainers.BpeTrainer(vocab_size=6400,special_tokens=special_tokens,  # 确保这三个token被包含show_progress=True,# 使用字节级别的基础字符集作为初始词汇，它包含了256个可能的字节值（0-255）对应的Unicode字符表示initial_alphabet=pre_tokenizers.ByteLevel.alphabet())# 读取文本数据texts = read_texts_from_jsonl(data_path)# 训练tokenizertokenizer.train_from_iterator(texts, trainer=trainer)# 设置解码器tokenizer.decoder = decoders.ByteLevel()# 检查特殊token的索引# 确保三个特殊token被正确分配了预期的ID：# <unk>应该是ID 0 - 用于表示未知词汇assert tokenizer.token_to_id("<unk>") == 0# <s>应该是ID 1 - 用于表示文本开始assert tokenizer.token_to_id("<s>") == 1# </s>应该是ID 2 - 用于表示文本结束assert tokenizer.token_to_id("</s>") == 2# 保存tokenizertokenizer_dir = "./EmoLLM_tokenizer"os.makedirs(tokenizer_dir, exist_ok=True)tokenizer.save(os.path.join(tokenizer_dir, "tokenizer.json"))tokenizer.model.save("./EmoLLM_tokenizer")# 手动创建配置文件config = {"add_bos_token": False,"add_eos_token": False,"add_prefix_space": False,"added_tokens_decoder": {# 配置 <unk> 标记 (未知词标记)# 索引为0，用于表示词表外的词或字符"0": {"content": "<unk>",  # 标记的实际内容"lstrip": False,     # 是否从左侧移除空白"normalized": False,  # 是否规范化"rstrip": False,     # 是否从右侧移除空白"single_word": False,  # 是否作为单个词处理"special": True      # 标记为特殊标记，在解码时会特殊处理},# 配置 <s> 标记 (序列开始标记)# 索引为1，用于标识文本序列的开始"1": {"content": "<s>","lstrip": False,"normalized": False,"rstrip": False,"single_word": False,"special": True},# 配置 </s> 标记 (序列结束标记)# 索引为2，用于标识文本序列的结束"2": {"content": "</s>","lstrip": False,"normalized": False,"rstrip": False,"single_word": False,"special": True}},"additional_special_tokens": [],"bos_token": "<s>","clean_up_tokenization_spaces": False,"eos_token": "</s>","legacy": True,"model_max_length": 32768,"pad_token": "<unk>","sp_model_kwargs": {},"spaces_between_special_tokens": False,"tokenizer_class": "PreTrainedTokenizerFast","unk_token": "<unk>",# 配置聊天模板 - 使用Jinja2模板格式定义模型输入的格式化方式"chat_template": "{% if messages[0]['role'] == 'system' %}{% set system_message = messages[0]['content'] %}{{ '<s>system\\n' + system_message + '</s>\\n' }}{% else %}{{ '<s>system\\n你是 EmoLLM，是一个完全开源的心理健康大模型。</s>\\n' }}{% endif %}{% for message in messages %}{% set content = message['content'] %}{% if message['role'] == 'user' %}{{ '<s>user\\n' + content + '</s>\\n<s>assistant\\n' }}{% elif message['role'] == 'assistant' %}{{ content + '</s>' + '\\n' }}{% endif %}{% endfor %}"# 聊天模板说明:# 1. 如果第一条消息是系统消息，将其作为系统指令；否则使用默认系统消息# 2. 用户消息格式: <s>user\n{用户内容}</s>\n<s>assistant\n# 3. 助手消息格式: {助手内容}</s>\n# 4. 特殊标记<s>和</s>用于标记消息的开始和结束}# 保存配置文件with open(os.path.join(tokenizer_dir, "tokenizer_config.json"), "w", encoding="utf-8") as config_file:json.dump(config, config_file, ensure_ascii=False, indent=4)print("Tokenizer training completed and saved.")def eval_tokenizer():# 加载预训练的tokenizertokenizer = AutoTokenizer.from_pretrained("./EmoLLM_tokenizer")messages = [{"role": "system", "content": "你是一个优秀的聊天机器人，总是给我正确的回应！"},{"role": "user", "content": '你来自哪里？'},{"role": "assistant", "content": '我来自地球'}]new_prompt = tokenizer.apply_chat_template(messages,tokenize=False)print(new_prompt)# 获取实际词汇表长度（包括特殊符号）actual_vocab_size = len(tokenizer)print('tokenizer实际词表长度：', actual_vocab_size)model_inputs = tokenizer(new_prompt)print('encoder长度：', len(model_inputs['input_ids']))input_ids = model_inputs['input_ids']response = tokenizer.decode(input_ids, skip_special_tokens=False)print('decoder和原始文本是否一致：', response == new_prompt)def main():train_tokenizer()eval_tokenizer()if __name__ == '__main__':main()

参考

https://github.com/jingyaogong/minimind/blob/master/scripts/train_tokenizer.py
https://github.com/SmartFlowAI/EmoLLM
https://github.com/aJupyter/ThinkLLM

文章目录

BPE 简介

BPE (Byte-Pair Encoding) 算法训练流程

BPE 编码流程

BPE 评估

代码

参考

相关文章：