本文基于 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():
# 加载预训练的tokenizer
tokenizer = 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 * 100
print(f"词表利用率: {utilization_rate:.2f}%")
# 3. 分析最常见的tokens
from collections import Counter
token_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 random
from 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'
# 初始化tokenizer
tokenizer = Tokenizer(models.BPE())
# 预分词器负责将原始文本进行初步分割，为后续的 BPE 合并操作做准备add_prefix_space=False 表示不在每个序列前添加空格
tokenizer.pre_tokenizer = pre_tokenizers.ByteLevel(add_prefix_space=False)
# 定义特殊token
special_tokens = ["<unk>", "<s>", "</s>"]
# 设置训练器并添加特殊token
trainer = 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)
# 训练tokenizer
tokenizer.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
# 保存tokenizer
tokenizer_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():
# 加载预训练的tokenizer
tokenizer = 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