文章目录
- 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