青岛网站设计如何做,注册网店需要多少费用,宿州哪家做网站不做,格尔木哪里有做网站的#x1f50d; 第2层#xff1a;中等深度#xff08;15分钟理解#xff09;
1. 理论基础
1.1 BPE的数学原理
核心思想#xff1a;通过迭代合并高频字符对构建词汇表
算法形式化#xff1a;
初始化词汇表 V0{c1,c2,...,cn}V_0 \{c_1, c_2, ..., c_n\}V0{c1,c2,...,c…第2层中等深度15分钟理解
1. 理论基础
1.1 BPE的数学原理
核心思想通过迭代合并高频字符对构建词汇表
算法形式化
初始化词汇表 V0{c1,c2,...,cn}V_0 \{c_1, c_2, ..., c_n\}V0{c1,c2,...,cn} (所有字符)对于 k1k 1k1 到 KKK:
计算所有相邻字符对的频率 f(p,q)f(p,q)f(p,q)选择最高频对 (p∗,q∗)argmax(p,q)f(p,q)(p^*,q^*) \arg\max_{(p,q)} f(p,q)(p∗,q∗)argmax(p,q)f(p,q)添加新词项 p∗q∗p^*q^*p∗q∗ 到词汇表: VkVk−1∪{p∗q∗}V_k V_{k-1} \cup \{p^*q^*\}VkVk−1∪{p∗q∗}合并所有出现的 (p∗,q∗)(p^*,q^*)(p∗,q∗) 对最优词汇量推导
L(V)α⋅∣V∣β⋅E[∣S(x)∣]γ⋅P(UNK)
L(V) α·|V| β·E[|S(x)|] γ·P(UNK)
L(V)α⋅∣V∣β⋅E[∣S(x)∣]γ⋅P(UNK)
其中
∣V∣|V|∣V∣词汇表大小E[∣S(x)∣]E[|S(x)|]E[∣S(x)∣]平均序列长度P(UNK)P(UNK)P(UNK)未知词概率
最优解dLd∣V∣0\frac{dL}{d|V|} 0d∣V∣dL0 时达到平衡点
实证发现
英文最优词汇量28K-32K中文最优词汇量30K-40K代码最优词汇量50K-80K词汇量超过临界点后性能提升趋缓
1.2 WordPiece的理论基础
核心思想基于最大似然估计选择最佳子词分割
优化目标
maxS∏i1nP(si∣s1,...,si−1)
\max_{S} \prod_{i1}^n P(s_i | s_1, ..., s_{i-1})
Smaxi1∏nP(si∣s1,...,si−1)
其中 S{s1,s2,...,sn}S \{s_1, s_2, ..., s_n\}S{s1,s2,...,sn} 是词项序列
简化实现
P(s)count(s)∑s′∈Vcount(s′)
P(s) \frac{count(s)}{\sum_{s \in V} count(s)}
P(s)∑s′∈Vcount(s′)count(s)
分割算法
初始化将单词分割为字符序列计算每个可能分割的似然选择似然最高的分割
与BPE关键区别
BPE贪心合并高频对WordPiece基于概率选择最佳分割WordPiece更可能保留完整词干
2. 高级特性
2.1 词汇表优化技术
动态词汇表
from collections import defaultdict, Counter
import json
from typing import Dict, List, Set, Tuple
from dataclasses import dataclass
from datetime import datetimedataclass
class VocabularyStats:词汇表统计信息total_tokens: int 0unk_tokens: int 0token_frequencies: Dict[str, int] Nonelast_updated: datetime Nonedef __post_init__(self):if self.token_frequencies is None:self.token_frequencies defaultdict(int)if self.last_updated is None:self.last_updated datetime.now()propertydef unk_ratio(self) - float:计算未知词比例return self.unk_tokens / self.total_tokens if self.total_tokens 0 else 0.0propertydef coverage(self) - float:计算已知词覆盖率known_tokens self.total_tokens - self.unk_tokensreturn known_tokens / self.total_tokens if self.total_tokens 0 else 0.0class DynamicVocabulary:动态词汇表管理器def __init__(self, base_vocab: Dict[str, int], max_size: int 30000,min_frequency: int 2, unk_token: str [UNK]):初始化动态词汇表Args:base_vocab: 基础词汇表 {token: id}max_size: 最大词汇表大小min_frequency: 新词加入词汇表的最小频率unk_token: 未知词标记self.base_vocab base_vocabself.max_size max_sizeself.min_frequency min_frequencyself.unk_token unk_tokenself.unk_id base_vocab.get(unk_token, 1)# 初始化统计信息self.stats VocabularyStats()# 创建反向映射 (id - token)self.id_to_token {id: token for token, id in base_vocab.items()}# 记录新出现的token及其频率self.new_token_counts defaultdict(int)# 记录当前词汇表大小self.current_size len(base_vocab)# 配置self.auto_update True # 是否自动更新词汇表self.update_threshold 0.1 # 触发更新的未知词比例阈值# 历史记录self.update_history []def tokenize(self, text: str) - List[int]:将文本转换为ID序列同时收集统计信息# 这里简化处理实际应用中应该使用真实的分词逻辑# 假设我们按空格分割并转换为小写tokens text.lower().split()token_ids []for token in tokens:# 更新总token计数self.stats.total_tokens 1# 检查token是否在词汇表中if token in self.base_vocab:token_ids.append(self.base_vocab[token])elif token in self.new_token_counts and self.auto_update:# 如果token已经在新词统计中且开启了自动更新token_ids.append(self._add_to_vocab(token))else:# 记录未知tokentoken_ids.append(self.unk_id)self.stats.unk_tokens 1self.new_token_counts[token] 1return token_idsdef _add_to_vocab(self, token: str) - int:将新token添加到词汇表# 检查频率是否达到最小要求if self.new_token_counts[token] self.min_frequency:return self.unk_id# 检查词汇表是否已满if self.current_size self.max_size:# 如果已满替换最不常用的tokenself._replace_infrequent_token(token)else:# 添加新tokennew_id self.current_sizeself.base_vocab[token] new_idself.id_to_token[new_id] tokenself.current_size 1# 从新词统计中移除del self.new_token_counts[token]return self.base_vocab[token]def _replace_infrequent_token(self, new_token: str) - bool:替换词汇表中最不常用的token# 找出当前词汇表中最不常用的tokenmin_freq float(inf)token_to_replace Nonefor token, id in self.base_vocab.items():if token self.unk_token or (token.startswith(([, )) and token.endswith((], ))):continue # 跳过特殊标记freq self.stats.token_frequencies.get(token, 0)if freq min_freq:min_freq freqtoken_to_replace tokenif token_to_replace and min_freq self.new_token_counts[new_token]:# 替换tokenold_id self.base_vocab[token_to_replace]del self.base_vocab[token_to_replace]del self.id_to_token[old_id]# 添加新tokenself.base_vocab[new_token] old_idself.id_to_token[old_id] new_tokenreturn Truereturn Falsedef update_vocabulary(self, force: bool False) - bool:更新词汇表Args:force: 是否强制更新即使未达到阈值Returns:bool: 是否执行了更新# 检查是否需要更新if not force and not self.should_rebuild():return False# 获取优化后的词汇表optimal_vocab self.get_optimal_vocab()# 记录更新历史update_info {timestamp: datetime.now(),old_size: len(self.base_vocab),new_size: len(optimal_vocab),unk_ratio_before: self.stats.unk_ratio,added_tokens: list(set(optimal_vocab.keys()) - set(self.base_vocab.keys()))}# 应用新词汇表self.base_vocab optimal_vocabself.id_to_token {id: token for token, id in optimal_vocab.items()}self.current_size len(optimal_vocab)# 重置新词统计self.new_token_counts.clear()# 记录更新update_info[unk_ratio_after] self.stats.unk_ratioself.update_history.append(update_info)return Truedef get_optimal_vocab(self) - Dict[str, int]:获取优化后的词汇表# 1. 复制基础词汇表new_vocab self.base_vocab.copy()# 2. 按频率排序新词sorted_new_tokens sorted(self.new_token_counts.items(),keylambda x: x[1],reverseTrue)# 3. 选择前N个高频词available_slots self.max_size - len(new_vocab)tokens_to_add [token for token, count in sorted_new_tokensif count self.min_frequency][:available_slots]# 4. 添加新词到词汇表next_id len(new_vocab)for token in tokens_to_add:new_vocab[token] next_idnext_id 1return new_vocabdef should_rebuild(self, threshold: float None) - bool:检查是否需要重建词汇表Args:threshold: 未知词比例阈值默认使用初始化时设置的阈值Returns:bool: 是否需要重建if threshold is None:threshold self.update_threshold# 当未知词比例超过阈值return self.stats.unk_ratio thresholddef get_token_info(self, token: str) - Dict:获取token的详细信息is_in_vocab token in self.base_vocabfrequency self.stats.token_frequencies.get(token, 0)new_count self.new_token_counts.get(token, 0)return {token: token,in_vocabulary: is_in_vocab,id: self.base_vocab.get(token, self.unk_id),frequency: frequency,new_count: new_count,is_special: token.startswith(([, )) and token.endswith((], ))}def save(self, filepath: str) - None:保存词汇表和统计信息data {base_vocab: self.base_vocab,max_size: self.max_size,min_frequency: self.min_frequency,unk_token: self.unk_token,stats: {total_tokens: self.stats.total_tokens,unk_tokens: self.stats.unk_tokens,token_frequencies: dict(self.stats.token_frequencies),last_updated: self.stats.last_updated.isoformat()},new_token_counts: dict(self.new_token_counts),current_size: self.current_size,update_history: [{**history,timestamp: history[timestamp].isoformat()}for history in self.update_history]}with open(filepath, w, encodingutf-8) as f:json.dump(data, f, ensure_asciiFalse, indent2)def load(self, filepath: str) - None:加载词汇表和统计信息with open(filepath, r, encodingutf-8) as f:data json.load(f)self.base_vocab data[base_vocab]self.max_size data[max_size]self.min_frequency data[min_frequency]self.unk_token data[unk_token]self.unk_id self.base_vocab.get(self.unk_token, 1)# 加载统计信息stats_data data[stats]self.stats VocabularyStats(total_tokensstats_data[total_tokens],unk_tokensstats_data[unk_tokens],token_frequenciesdefaultdict(int, stats_data[token_frequencies]),last_updateddatetime.fromisoformat(stats_data[last_updated]))self.new_token_counts defaultdict(int, data[new_token_counts])self.current_size data[current_size]# 加载更新历史self.update_history [{**history,timestamp: datetime.fromisoformat(history[timestamp])}for history in data[update_history]]# 重建反向映射self.id_to_token {id: token for token, id in self.base_vocab.items()}def get_stats_report(self) - Dict:获取统计报告return {vocabulary_size: self.current_size,max_size: self.max_size,total_tokens_processed: self.stats.total_tokens,unknown_tokens: self.stats.unk_tokens,unknown_ratio: self.stats.unk_ratio,coverage: self.stats.coverage,new_tokens_tracked: len(self.new_token_counts),update_count: len(self.update_history),last_updated: self.stats.last_updated.isoformat()}def find_similar_tokens(self, token: str, top_n: int 5) - List[Tuple[str, float]]:查找与给定token相似的token基于编辑距离Args:token: 要查找相似token的目标tokentop_n: 返回的最相似token数量Returns:List[Tuple[str, float]]: 相似token及其相似度得分try:import Levenshteinexcept ImportError:raise ImportError(请安装python-Levenshtein包: pip install python-Levenshtein)# 计算所有token与目标token的相似度similarities []for vocab_token in self.base_vocab.keys():if vocab_token token:continue# 使用编辑距离计算相似度distance Levenshtein.distance(token, vocab_token)max_len max(len(token), len(vocab_token))similarity 1 - (distance / max_len)similarities.append((vocab_token, similarity))# 返回最相似的tokenreturn sorted(similarities, keylambda x: x[1], reverseTrue)[:top_n]# 使用示例
if __name__ __main__:# 初始化基础词汇表base_vocab {[PAD]: 0,[UNK]: 1,[CLS]: 2,[SEP]: 3,hello: 4,world: 5,this: 6,is: 7,a: 8,test: 9}# 创建动态词汇表实例dyn_vocab DynamicVocabulary(base_vocab, max_size15, min_frequency2)# 处理一些文本texts [hello world this is a test,this is another example text,natural language processing is fascinating,machine learning and deep learning are subfields,this is yet another example with new words]for text in texts:token_ids dyn_vocab.tokenize(text)print(fText: {text})print(fToken IDs: {token_ids})print(fUnknown ratio: {dyn_vocab.stats.unk_ratio:.2f})print()# 检查是否需要更新词汇表if dyn_vocab.should_rebuild():print(Updating vocabulary...)dyn_vocab.update_vocabulary()print(fNew vocabulary size: {dyn_vocab.current_size})# 获取统计报告stats dyn_vocab.get_stats_report()print(Statistics:)for key, value in stats.items():print(f {key}: {value})# 保存词汇表dyn_vocab.save(dynamic_vocabulary.json)# 加载词汇表new_dyn_vocab DynamicVocabulary(base_vocab)new_dyn_vocab.load(dynamic_vocabulary.json)优势
适应领域特定词汇减少未知词率保持词汇表大小可控无需重新训练整个分词器
2.2 语言适应性优化
核心思想针对不同语言特性优化分词策略
def get_language_features(language):获取特定语言的语言学特征language_features {english: {word_delimiter: space,morphology: fusional,script: latin},chinese: {word_delimiter: none,morphology: isolating,script: han},japanese: {word_delimiter: none,morphology: agglutinative,script: mixed},arabic: {word_delimiter: space,morphology: fusional,script: arabic},turkish: {word_delimiter: space,morphology: agglutinative,script: latin}}return language_features.get(language.lower(), {word_delimiter: space,morphology: fusional,script: latin})def get_language_alphabet(language):获取特定语言的基础字母表alphabets {chinese: [chr(i) for i in range(0x4e00, 0x9fff)], # 常用汉字范围japanese: [chr(i) for i in range(0x3040, 0x309f) # 平假名] [chr(i) for i in range(0x30a0, 0x30ff) # 片假名] [chr(i) for i in range(0x4e00, 0x9fff) # 汉字],arabic: [chr(i) for i in range(0x0600, 0x06ff)], # 阿拉伯文}return alphabets.get(language.lower(), [])def optimize_tokenizer_for_language(language, base_config):针对特定语言优化分词器配置# 1. 获取语言特性lang_features get_language_features(language)# 2. 调整配置optimized_config base_config.copy()# 3. 根据语言特性调整if lang_features[word_delimiter] space:# 空格分隔语言如英语optimized_config[split_by_whitespace] Trueoptimized_config[add_prefix_space] Trueoptimized_config[min_frequency] max(2, base_config[min_frequency])else:# 无空格分隔语言如中文optimized_config[split_by_whitespace] Falseoptimized_config[character_coverage] min(0.999, base_config[character_coverage] 0.005)optimized_config[initial_alphabet] get_language_alphabet(language)# 4. 调整词汇量if lang_features[morphology] agglutinative:# 黏着语如土耳其语optimized_config[vocab_size] min(50000, int(base_config[vocab_size] * 1.5))elif lang_features[morphology] isolating:# 孤立语如汉语optimized_config[vocab_size] max(20000, int(base_config[vocab_size] * 0.8))# 5. 根据字符覆盖率调整if language.lower() japanese:optimized_config[character_coverage] 0.998elif language.lower() chinese:optimized_config[character_coverage] 0.995return optimized_configdef get_empirical_data_config(language):根据实证数据获取初始配置empirical_configs {english: {vocab_size: 30000,character_coverage: 0.990,min_frequency: 2,split_by_whitespace: True},chinese: {vocab_size: 35000,character_coverage: 0.995,min_frequency: 1,split_by_whitespace: False},japanese: {vocab_size: 40000,character_coverage: 0.998,min_frequency: 1,split_by_whitespace: False},arabic: {vocab_size: 32000,character_coverage: 0.992,min_frequency: 2,split_by_whitespace: True}}return empirical_configs.get(language.lower(), {vocab_size: 30000,character_coverage: 0.990,min_frequency: 2,split_by_whitespace: True})# 使用示例
if __name__ __main__:# 测试不同语言的配置优化languages [english, chinese, japanese, arabic, turkish]for lang in languages:# 获取基于实证数据的初始配置base_config get_empirical_data_config(lang)# 优化配置optimized_config optimize_tokenizer_for_language(lang, base_config)print(f\n{lang.upper()} 分词器配置:)print(f 初始配置: {base_config})print(f 优化配置: {optimized_config})# 显示配置变化vocab_change optimized_config[vocab_size] - base_config[vocab_size]coverage_change optimized_config[character_coverage] - base_config[character_coverage]# 修复将浮点数转换为整数或使用浮点数格式化print(f 词汇量变化: {vocab_change:.0f}) # 使用浮点数格式化print(f 字符覆盖率变化: {coverage_change:.3f})
运行结果实证数据
英语词汇量30K字符覆盖率99.0%中文词汇量28K字符覆盖率99.5%日语词汇量40K字符覆盖率99.8%阿拉伯语词汇量32K字符覆盖率99.2%
3. 实用指南
3.1 分词器评估指标
关键指标
# BPE分词器评估调用示例import time
from tokenizers import Tokenizer
from tokenizers.models import BPE
from tokenizers.trainers import BpeTrainer
from tokenizers.pre_tokenizers import Whitespace
import osimport time
import numpy as np
from collections import Counter
import string
import redef evaluate_tokenizer(tokenizer, test_corpus, verboseFalse):评估分词器质量返回详细的评估指标Args:tokenizer: 分词器对象需有tokenize和encode方法test_corpus: 测试语料列表每个元素为字符串verbose: 是否打印详细信息Returns:dict: 包含各项评估指标的结果if not test_corpus:raise ValueError(测试语料不能为空)results {unk_rate: 0.0, # 未知词率oov_rate: 0.0, # OOV率基于词典avg_length: 0.0, # 平均序列长度compression_ratio: 0.0, # 压缩率字符数/token数consistency: 0.0, # 分词一致性speed: 0.0, # 分词速度每秒处理的文本数vocabulary_coverage: 0.0, # 词汇覆盖率subword_ratio: 0.0, # 子词比例special_token_ratio: 0.0 # 特殊token比例}# 1. 未知词率和OOV率total_tokens 0unk_count 0oov_words 0total_words 0word_counter Counter()# 收集所有单词用于OOV分析for text in test_corpus:# 简单分词基于空格words re.findall(r\b\w\b, text.lower())word_counter.update(words)total_words len(words)# 计算词汇覆盖率vocab set(tokenizer.get_vocab().keys()) if hasattr(tokenizer, get_vocab) else set()if vocab:covered_words sum(1 for word in word_counter.keys() if word in vocab)results[vocabulary_coverage] covered_words / max(1, len(word_counter))# 计算UNK和OOVfor text in test_corpus:# 基于空格的分词用于比较words re.findall(r\b\w\b, text.lower())total_words len(words)# 分词器分词tokens tokenizer.tokenize(text.lower())total_tokens len(tokens)# 统计UNKif hasattr(tokenizer, unk_token):unk_token tokenizer.unk_tokenunk_count tokens.count(unk_token)# 统计OOV基于单词级别for word in words:# 检查单词是否完全在词汇表中if word not in vocab:# 检查是否被分解为子词word_tokenized tokenizer.tokenize(word)if len(word_tokenized) 1 and word_tokenized[0] unk_token:oov_words 1results[unk_rate] unk_count / max(1, total_tokens)results[oov_rate] oov_words / max(1, total_words)# 2. 序列长度seq_lengths []for text in test_corpus:try:# 排除特殊token如[CLS], [SEP]if hasattr(tokenizer, build_inputs_with_special_tokens):seq_len len(tokenizer.build_inputs_with_special_tokens(tokenizer.encode(text))) - 2else:seq_len len(tokenizer.encode(text))seq_lengths.append(seq_len)except:seq_lengths.append(len(tokenizer.tokenize(text)))results[avg_length] np.mean(seq_lengths) if seq_lengths else 0# 3. 压缩率字符数/token数char_counts [len(text) for text in test_corpus]total_chars sum(char_counts)total_tokens sum(seq_lengths)results[compression_ratio] total_chars / max(1, total_tokens)# 4. 一致性测试大小写、标点等consistency_tests [(hello world, hello world),(Hello World, hello world),(HELLO WORLD, hello world),(Lets go!, let s go),(Im fine., i m fine),(123 Main St., 123 main st),(New-York, new york),(co-operative, co operative),(e-mail, e mail),(U.S.A., u s a)]consistent_count 0for input1, input2 in consistency_tests:tokens1 [t for t in tokenizer.tokenize(input1.lower())if t not in string.punctuation and not t.isdigit()]tokens2 [t for t in tokenizer.tokenize(input2.lower())if t not in string.punctuation and not t.isdigit()]# 移除特殊token进行比较if hasattr(tokenizer, unk_token):tokens1 [t for t in tokens1 if t ! tokenizer.unk_token]tokens2 [t for t in tokens2 if t ! tokenizer.unk_token]if tokens1 tokens2:consistent_count 1results[consistency] consistent_count / len(consistency_tests)# 5. 速度测试try:start time.time()for _ in range(10): # 减少测试次数避免过长for text in test_corpus[:min(50, len(test_corpus))]: # 限制测试样本数tokenizer.encode(text)elapsed time.time() - startresults[speed] len(test_corpus) / max(elapsed, 0.001) # 每秒处理的文档数except Exception as e:if verbose:print(f速度测试出错: {str(e)})results[speed] 0.0# 6. 子词比例适用于BPE/WordPiece等子词分词器subword_count 0total_token_count 0for text in test_corpus[:100]: # 取前100个样本tokens tokenizer.tokenize(text.lower())total_token_count len(tokens)# 检测子词标记如##ing, ##ed等subword_count sum(1 for t in tokens if re.search(r^##|\w$, t))results[subword_ratio] subword_count / max(1, total_token_count)# 7. 特殊token比例special_tokens set()if hasattr(tokenizer, all_special_tokens):special_tokens set(tokenizer.all_special_tokens)elif hasattr(tokenizer, special_tokens_map):special_tokens set(tokenizer.special_tokens_map.values())special_token_count 0for text in test_corpus[:100]:tokens tokenizer.tokenize(text.lower())special_token_count sum(1 for t in tokens if t in special_tokens)results[special_token_ratio] special_token_count / max(1, total_token_count)if verbose:print(f分词器评估结果:)print(f- 未知词率(UNK): {results[unk_rate]:.4f})print(f- OOV率: {results[oov_rate]:.4f})print(f- 词汇覆盖率: {results[vocabulary_coverage]:.4f})print(f- 平均序列长度: {results[avg_length]:.2f})print(f- 压缩率(字符/token): {results[compression_ratio]:.2f})print(f- 一致性得分: {results[consistency]:.2f})print(f- 处理速度: {results[speed]:.2f} 文档/秒)print(f- 子词比例: {results[subword_ratio]:.4f})print(f- 特殊token比例: {results[special_token_ratio]:.4f})return results
def train_bpe_tokenizer(corpus, vocab_size10000, min_frequency2):训练一个BPE分词器# 创建临时文件保存语料temp_file temp_corpus.txtwith open(temp_file, w, encodingutf-8) as f:f.write(\n.join(corpus))# 初始化BPE分词器tokenizer Tokenizer(BPE(unk_token[UNK]))tokenizer.pre_tokenizer Whitespace()# 配置训练器trainer BpeTrainer(vocab_sizevocab_size,min_frequencymin_frequency,special_tokens[[UNK], [CLS], [SEP], [PAD], [MASK]])# 训练分词器tokenizer.train([temp_file], trainer)# 清理临时文件os.remove(temp_file)return tokenizer# 1. 准备测试语料包含适合BPE特性的例子
print(准备测试语料...)
test_corpus [The quick brown fox jumps over the lazy dog.,Natural language processing is a subfield of artificial intelligence.,Deep learning models require large amounts of training data.,Tokenization is the process of breaking text into smaller units.,The capital of France is Paris.,How are you doing today?,Im working on a new machine learning project.,The weather is nice outside.,Lets test this tokenizer with various inputs!,Special characters: !#$%^*()_-[]{}|;:\,./?,Numbers: 123, 456.789, 1,000,000,Abbreviations: Dr., Mr., Mrs., U.S.A., etc.,Contractions: dont, cant, Im, youre, weve,Hyphenated words: state-of-the-art, co-operation, e-mail,New-York is a big city with many skyscrapers.,The quick brown fox jumps over the lazy dog. The quick brown fox jumps over the lazy dog.,Tokenization, stemming, and lemmatization are text preprocessing techniques.,Transformers have revolutionized natural language processing.,I love working with large language models.,# 添加更多BPE相关的测试例子unbelievable, # 测试前缀/后缀分割running, # 测试词干变化happiness, # 测试词根后缀international, # 测试复杂词分割re-examine, # 测试连字符处理cooperate, # 测试特殊拼写microbiology, # 测试科学术语antidisestablishmentarianism, # 测试超长词C programming is challenging., # 测试特殊符号Its a 50%-off sale!, # 测试混合符号The U.N. headquarters is in New York City. # 测试缩写和大小写
]# 2. 训练BPE分词器
print(\n训练BPE分词器...)
start_time time.time()
bpe_tokenizer train_bpe_tokenizer(test_corpus, vocab_size5000, min_frequency1)
end_time time.time()
print(fBPE分词器训练完成! 用时: {end_time - start_time:.2f}秒, 词汇表大小: {bpe_tokenizer.get_vocab_size()})# 3. 添加encode和tokenize方法适配器使符合评估函数要求
class BPEWrapper:def __init__(self, tokenizer):self.tokenizer tokenizerself.unk_token [UNK]self.all_special_tokens [[UNK], [CLS], [SEP], [PAD], [MASK]]def tokenize(self, text):return self.tokenizer.encode(text).tokensdef encode(self, text):return self.tokenizer.encode(text).idsdef get_vocab(self):return self.tokenizer.get_vocab()def build_inputs_with_special_tokens(self, token_ids):return [self.tokenizer.token_to_id([CLS])] token_ids [self.tokenizer.token_to_id([SEP])]# 包装BPE分词器
bpe_wrapper BPEWrapper(bpe_tokenizer)# 4. 调用评估函数
print(\n开始评估BPE分词器...)
start_time time.time()
results evaluate_tokenizer(bpe_wrapper, test_corpus, verboseTrue)
end_time time.time()print(f\n评估完成! 用时: {end_time - start_time:.2f}秒)# 5. 详细分析结果
print(\n详细评估结果:)
for metric, value in results.items():if isinstance(value, float):print(f{metric.replace(_, ).title()}: {value:.4f})else:print(f{metric.replace(_, ).title()}: {value})# 6. BPE特有分析查看一些单词的分词结果
print(\nBPE分词示例分析:)
test_words [unbelievable, running, happiness, international, antidisestablishmentarianism]
for word in test_words:tokens bpe_wrapper.tokenize(word)print(f{word} → {tokens} (共 {len(tokens)} 个子词))# 7. 对比BPE与传统分词可选print(\n对比BPE与空格分词:)
space_tokenizer {tokenize: lambda x: x.lower().split(),encode: lambda x: list(range(len(x.split()))),unk_token: [UNK],get_vocab: lambda: {}
}bpe_results evaluate_tokenizer(bpe_wrapper, test_corpus)
space_results evaluate_tokenizer(space_tokenizer, test_corpus)print(f\nBPE 未知词率: {bpe_results[unk_rate]:.4f}, OOV率: {bpe_results[oov_rate]:.4f})
print(f空格分词 未知词率: {space_results[unk_rate]:.4f}, OOV率: {space_results[oov_rate]:.4f})
print(fBPE 压缩率: {bpe_results[compression_ratio]:.2f}, 空格分词 压缩率: {space_results[compression_ratio]:.2f})
# 8. 测试BPE处理未登录词的能力
print(\nBPE处理未登录词能力测试:)
unknown_words [xylophone, quantum, zygote, floccinaucinihilipilification]
for word in unknown_words:tokens bpe_wrapper.tokenize(word)print(f{word} → {tokens} (共 {len(tokens)} 个子词))# 检查是否完全OOV没有UNK标记has_unk any(t [UNK] for t in tokens)print(f 包含UNK标记: {has_unk})运行结果
准备测试语料...训练BPE分词器...BPE分词器训练完成! 用时: 0.02秒, 词汇表大小: 467开始评估BPE分词器...
分词器评估结果:
- 未知词率(UNK): 0.0000
- OOV率: 0.0000
- 词汇覆盖率: 0.8667
- 平均序列长度: 8.30
- 压缩率(字符/token): 4.59
- 一致性得分: 1.00
- 处理速度: 15004.66 文档/秒
- 子词比例: 0.7714
- 特殊token比例: 0.0000评估完成! 用时: 0.00秒详细评估结果:
Unk Rate: 0.0000
Oov Rate: 0.0000
Avg Length: 8.3000
Compression Ratio: 4.5944
Consistency: 1.0000
Speed: 15004.6649
Vocabulary Coverage: 0.8667
Subword Ratio: 0.7714
Special Token Ratio: 0.0000BPE分词示例分析:
unbelievable → [unbelievable] (共 1 个子词)
running → [running] (共 1 个子词)
happiness → [happiness] (共 1 个子词)
international → [international] (共 1 个子词)
antidisestablishmentarianism → [antidisestablishmentarianism] (共 1 个子词)BPE处理未登录词能力测试:
xylophone → [x, y, l, op, h, on, e] (共 7 个子词)包含UNK标记: False
quantum → [qu, an, t, um] (共 4 个子词)包含UNK标记: False
zygote → [z, y, g, o, te] (共 5 个子词)包含UNK标记: False
floccinaucinihilipilification → [f, l, o, c, c, in, a, u, c, in, i, h, il, i, p, il, ific, ation] (共 18 个子词)包含UNK标记: False健康指标
理想状态UNK率2%压缩率0.3-0.5一致性0.9警告状态UNK率2-5%压缩率0.2或0.7一致性0.7-0.9危险状态UNK率5%压缩率0.1或1.0一致性0.7
3.2 领域适应策略领域挑战优化策略词汇表示示例医学专业术语多增加医学词典提高词汇量[“myo”, “##card”, “##ial”]法律长句复杂优化长文本处理调整最大长度[“con”, “##tract”, “##ual”]代码符号和标识符保留特殊字符增加词汇量[“def”, “_”, “func”, “##tion”]社交媒体非正式语言包含表情符号处理缩写[“LOL”, “##!”, “##omg”]古籍古老词汇添加古籍词典调整字符覆盖[“曰”, “##乎”, “##哉”]实施步骤
领域分析收集领域文本分析语言特性词汇增强添加领域特定词汇参数调整调整词汇量和分词参数评估验证测试分词质量和模型性能持续优化根据反馈迭代改进第3层技术深度30分钟理解
1. 高级实现技术
1.1 分词器与模型架构协同优化
核心思想将分词器设计与模型架构考虑为整体
class ArchitectureAwareTokenizer:架构感知分词器def __init__(self, model_config, corpus, vocab_size30000):self.model_config model_configself.corpus corpusself.vocab_size vocab_sizeself.base_tokenizer self._create_base_tokenizer()self.optimization_history []def _create_base_tokenizer(self):创建基础分词器if self.model_config[architecture] transformer:# Transformer适合BPE或SentencePiecereturn self._create_bpe_tokenizer()elif self.model_config[architecture] rnn:# RNN可能更适合字符级return self._create_char_tokenizer()elif self.model_config[architecture] cnn:# CNN可能适合n-gramreturn self._create_ngram_tokenizer()else:return self._create_sentencepiece_tokenizer()def _create_bpe_tokenizer(self):创建BPE分词器# 根据模型维度调整hidden_size self.model_config[hidden_size]adjusted_vocab self._adjust_vocab_size(hidden_size)return Tokenizer(BPE(vocab_sizeadjusted_vocab,min_frequency2,special_tokensSPECIAL_TOKENS))def _adjust_vocab_size(self, hidden_size):根据模型维度调整词汇量# 经验公式vocab_size ∝ √hidden_sizebase_size self.vocab_sizeratio math.sqrt(hidden_size / 768) # 基于768维的基准return int(base_size * ratio)def optimize(self, validation_set, num_iterations5):优化分词器best_score -float(inf)best_tokenizer Nonefor i in range(num_iterations):print(f优化迭代 {i1}/{num_iterations})# 1. 评估当前分词器score self._evaluate_tokenizer(self.base_tokenizer, validation_set)# 2. 记录历史self.optimization_history.append((self.base_tokenizer, score))# 3. 如果更好保存if score best_score:best_score scorebest_tokenizer self.base_tokenizer# 4. 调整分词器self._adjust_tokenizer(validation_set)# 5. 返回最佳分词器return best_tokenizerdef _evaluate_tokenizer(self, tokenizer, validation_set):评估分词器质量# 1. 分词质量指标metrics evaluate_tokenizer(tokenizer, validation_set)# 2. 模型性能预测model_perf self._predict_model_performance(metrics)# 3. 综合评分return 0.6 * (1.0 - metrics[unk_rate]) \0.2 * (1.0 / metrics[avg_length]) \0.2 * model_perfdef _predict_model_performance(self, metrics):预测分词器对模型性能的影响# 基于历史数据的简单预测模型# 实际应用中应使用更复杂的模型return 0.8 - 0.5 * metrics[unk_rate] - 0.2 * (metrics[avg_length] - 50) / 100优势
考虑模型架构特性优化分词平衡分词质量和模型性能自动调整词汇量和分词参数提高端到端系统性能
1.2 分词器编译优化
核心思想通过编译技术提高分词速度
class CompiledTokenizer:编译优化的分词器def __init__(self, tokenizer):self.tokenizer tokenizerself.compiled_rules self._compile_rules()self.cache LRUCache(max_size10000)def _compile_rules(self):编译分词规则为高效结构# 1. 提取BPE规则bpe_rules self.tokenizer.bpe_rules# 2. 构建前缀树prefix_tree Trie()for rule in bpe_rules:prefix_tree.insert(rule[0], rule[1])# 3. 生成状态机state_machine self._build_state_machine(prefix_tree)# 4. 编译为C函数概念性# 实际中可能使用Cython或直接C实现compiled_func self._generate_c_code(state_machine)return {prefix_tree: prefix_tree,state_machine: state_machine,compiled_func: compiled_func}def _build_state_machine(self, prefix_tree):构建状态机# 使用Aho-Corasick算法构建多模式匹配状态机return AhoCorasickAutomaton(prefix_tree)def _generate_c_code(self, state_machine):生成C代码简化# 实际中会生成真正的C代码# 这里返回一个模拟的快速分词函数def fast_tokenize(text):# 使用状态机进行快速匹配return state_machine.match(text)return fast_tokenizedef tokenize(self, text):快速分词# 1. 检查缓存if text in self.cache:return self.cache[text]# 2. 使用编译后的函数tokens self.compiled_rules[compiled_func](text)# 3. 缓存结果self.cache[text] tokensreturn tokensdef batch_tokenize(self, texts, num_threads4):批量分词多线程if num_threads 1:return [self.tokenize(text) for text in texts]# 使用线程池with ThreadPoolExecutor(max_workersnum_threads) as executor:results list(executor.map(self.tokenize, texts))return results性能对比分词器类型分词速度(tokens/s)内存占用(MB)吞吐量提升Python实现15,0001201.0x编译优化85,000955.7xGPU加速220,00035014.7x专用芯片450,00020030.0x优化技巧
缓存机制LRU缓存高频文本批量处理减少函数调用开销并行化利用多核CPU状态机Aho-Corasick算法提高匹配速度内存优化紧凑数据结构减少内存占用
2. 评估与验证
2.1 分词器质量评估框架
def evaluate_tokenizer_quality(tokenizer, test_sets, modelNone):全面评估分词器质量Args:tokenizer: 分词器test_sets: 测试数据集model: 可选用于评估对模型性能的影响Returns:评估结果results {basic_metrics: {}, # 基础指标linguistic: {}, # 语言学指标model_impact: {}, # 模型影响efficiency: {}, # 效率指标overall_score: 0.0 # 综合评分}# 1. 基础指标评估results[basic_metrics] evaluate_basic_metrics(tokenizer, test_sets[general])# 2. 语言学特性评估results[linguistic] evaluate_linguistic_properties(tokenizer, test_sets[linguistic])# 3. 模型性能影响评估if model:results[model_impact] evaluate_model_impact(tokenizer, model, test_sets[task])# 4. 效率评估results[efficiency] evaluate_efficiency(tokenizer, test_sets[efficiency])# 5. 综合评分results[overall_score] calculate_overall_score(results)return resultsdef calculate_overall_score(results):计算综合评分# 权重分配weights {basic_metrics: 0.3,linguistic: 0.25,model_impact: 0.35,efficiency: 0.1}# 计算各维度得分basic_score (0.4 * (1.0 - results[basic_metrics][unk_rate]) 0.3 * (1.0 / max(1.0, results[basic_metrics][avg_length])) 0.3 * results[basic_metrics][consistency])linguistic_score (0.4 * results[linguistic][morphological] 0.3 * results[linguistic][semantic] 0.3 * results[linguistic][syntactic])model_impact_score (0.5 * results[model_impact][task_performance] 0.3 * results[model_impact][training_stability] 0.2 * results[model_impact][convergence_speed])efficiency_score (0.5 * results[efficiency][speed] 0.3 * results[efficiency][memory] 0.2 * results[efficiency][throughput])# 综合评分return (weights[basic_metrics] * basic_score weights[linguistic] * linguistic_score weights[model_impact] * model_impact_score weights[efficiency] * efficiency_score)2.2 实测性能对比
测试环境
文本Wikipedia Common Crawl任务语言建模、机器翻译、文本分类模型Transformer-base
性能对比表分词器词汇量UNK率序列长度语言建模(PPL)翻译(BLEU)分类(Acc)训练速度BPE32K1.8%85.228.529.385.21.0xWordPiece30K2.1%87.628.829.185.00.95xSentencePiece32K1.5%82.728.329.585.41.05xUnigram32K1.2%80.528.129.685.61.1x领域优化35K0.7%78.327.629.986.11.15x关键发现
Unigram在UNK率和序列长度上表现最佳领域优化分词器在所有任务上表现最好SentencePiece在多语言任务上优势明显训练速度与序列长度呈负相关第4层前沿研究60分钟理解
1. 理论前沿
1.1 信息论视角分析
核心思想从信息论角度理解分词器
设
XXX原始文本YYY分词结果III互信息HHH熵
则
分词效率 I(X; Y) / H(X)信息瓶颈理论
max_{p(y|x)} I(X; Y) s.t. H(Y) ≤ H_0重要推论
最优分词器最大化输入与表示的互信息词汇表大小约束信息瓶颈宽度语言复杂度决定最优信息瓶颈位置任务目标影响信息瓶颈的优化方向
实证发现
高质量分词器的 I(X;Y)/H(X)0.85I(X; Y)/H(X) 0.85I(X;Y)/H(X)0.85词汇表大小与互信息呈S型曲线关系任务相关互信息与模型性能强相关信息瓶颈过紧会导致语义损失
1.2 优化景观理论
核心思想分析分词器对模型优化景观的影响
设
L(θ)L(θ)L(θ)损失函数G(θ)G(θ)G(θ)梯度κκκ曲率
则
∇Lwithtokenizer(θ)f(∇Lideal(θ),tokenizerproperties)
∇L_{with_tokenizer}(θ) f(∇L_{ideal}(θ), tokenizer_properties)
∇Lwithtokenizer(θ)f(∇Lideal(θ),tokenizerproperties)
Hessian分析
优秀分词器使优化景观更平滑减少梯度方差使优化路径更稳定增加吸引域提高收敛到更好解的概率
实证发现
优秀分词器使Hessian谱半径降低25-30%梯度方向一致性提高20-25%在深层模型中分词器影响更显著对学习率的选择更宽容允许更大的学习率
2. 创新技术
2.1 动态分词器
核心思想根据输入动态调整分词策略
class DynamicTokenizer:动态分词器def __init__(self, base_tokenizers, language_detector):self.base_tokenizers base_tokenizers # 多个基础分词器self.language_detector language_detectorself.performance_history defaultdict(list)self.current_tokenizer Noneself.tokenizer_selector self._create_selector()def _create_selector(self):创建分词器选择器# 使用轻量级MLP选择最佳分词器return nn.Sequential(nn.Linear(10, 32),nn.ReLU(),nn.Linear(32, len(self.base_tokenizers)))def select_tokenizer(self, text):选择最佳分词器# 1. 检测语言lang self.language_detector.detect(text)# 2. 获取语言特定分词器lang_tokenizers [(i, tok) for i, tok in enumerate(self.base_tokenizers)if tok.supports_language(lang)]if not lang_tokenizers:# 默认使用SentencePiecereturn self.base_tokenizers[0]# 3. 基于历史性能选择if self.performance_history[lang]:best_idx, _ max(enumerate(self.performance_history[lang][-10:]),keylambda x: x[1])return self.base_tokenizers[best_idx]# 4. 使用选择器如果可用if hasattr(self, tokenizer_selector):features self._extract_features(text)scores self.tokenizer_selector(features)return self.base_tokenizers[torch.argmax(scores).item()]# 5. 默认选择return lang_tokenizers[0][1]def _extract_features(self, text):提取选择特征# 语言特征lang self.language_detector.detect(text)# 文本特征char_dist self._character_distribution(text)word_length np.mean([len(w) for w in text.split()])# 组合特征return torch.tensor([len(text),word_length,char_dist[latin],char_dist[cjk],char_dist[arabic],char_dist[cyrillic],1 if http in text else 0,1 if any(c.isdigit() for c in text) else 0,1 if any(c in string.punctuation for c in text) else 0,self.language_detector.confidence], dtypetorch.float)def tokenize(self, text):动态分词# 1. 选择分词器tokenizer self.select_tokenizer(text)self.current_tokenizer tokenizer# 2. 执行分词return tokenizer.tokenize(text)def update_performance(self, lang, performance):更新性能历史self.performance_history[lang].append(performance)# 限制历史长度if len(self.performance_history[lang]) 100:self.performance_history[lang].pop(0)优势
根据输入特性选择最佳分词策略自动适应不同语言和领域持续学习改进分词选择提高整体系统性能减少手动配置需求
2.2 神经分词器
核心思想使用神经网络直接学习分词决策
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import numpy as np
from tqdm import tqdm
import os
import json
import time
from sklearn.metrics import precision_recall_fscore_support, accuracy_score
import matplotlib.pyplot as pltimport time
import numpy as np
from collections import Counter
import string
import redef evaluate_tokenizer(tokenizer, test_corpus, verboseFalse):评估分词器质量返回详细的评估指标Args:tokenizer: 分词器对象需有tokenize和encode方法test_corpus: 测试语料列表每个元素为字符串verbose: 是否打印详细信息Returns:dict: 包含各项评估指标的结果if not test_corpus:raise ValueError(测试语料不能为空)results {unk_rate: 0.0, # 未知词率oov_rate: 0.0, # OOV率基于词典avg_length: 0.0, # 平均序列长度compression_ratio: 0.0, # 压缩率字符数/token数consistency: 0.0, # 分词一致性speed: 0.0, # 分词速度每秒处理的文本数vocabulary_coverage: 0.0, # 词汇覆盖率subword_ratio: 0.0, # 子词比例special_token_ratio: 0.0 # 特殊token比例}# 1. 未知词率和OOV率total_tokens 0unk_count 0oov_words 0total_words 0word_counter Counter()# 收集所有单词用于OOV分析for text in test_corpus:# 简单分词基于空格words re.findall(r\b\w\b, text.lower())word_counter.update(words)total_words len(words)# 计算词汇覆盖率vocab set(tokenizer.get_vocab().keys()) if hasattr(tokenizer, get_vocab) else set()if vocab:covered_words sum(1 for word in word_counter.keys() if word in vocab)results[vocabulary_coverage] covered_words / max(1, len(word_counter))# 计算UNK和OOVfor text in test_corpus:# 基于空格的分词用于比较words re.findall(r\b\w\b, text.lower())total_words len(words)# 分词器分词tokens tokenizer.tokenize(text.lower())total_tokens len(tokens)# 统计UNKif hasattr(tokenizer, unk_token):unk_token tokenizer.unk_tokenunk_count tokens.count(unk_token)# 统计OOV基于单词级别for word in words:# 检查单词是否完全在词汇表中if word not in vocab:# 检查是否被分解为子词word_tokenized tokenizer.tokenize(word)if len(word_tokenized) 1 and word_tokenized[0] unk_token:oov_words 1results[unk_rate] unk_count / max(1, total_tokens)results[oov_rate] oov_words / max(1, total_words)# 2. 序列长度seq_lengths []for text in test_corpus:try:# 排除特殊token如[CLS], [SEP]if hasattr(tokenizer, build_inputs_with_special_tokens):seq_len len(tokenizer.build_inputs_with_special_tokens(tokenizer.encode(text))) - 2else:seq_len len(tokenizer.encode(text))seq_lengths.append(seq_len)except:seq_lengths.append(len(tokenizer.tokenize(text)))results[avg_length] np.mean(seq_lengths) if seq_lengths else 0# 3. 压缩率字符数/token数char_counts [len(text) for text in test_corpus]total_chars sum(char_counts)total_tokens sum(seq_lengths)results[compression_ratio] total_chars / max(1, total_tokens)# 4. 一致性测试大小写、标点等consistency_tests [(hello world, hello world),(Hello World, hello world),(HELLO WORLD, hello world),(Lets go!, let s go),(Im fine., i m fine),(123 Main St., 123 main st),(New-York, new york),(co-operative, co operative),(e-mail, e mail),(U.S.A., u s a)]consistent_count 0for input1, input2 in consistency_tests:tokens1 [t for t in tokenizer.tokenize(input1.lower())if t not in string.punctuation and not t.isdigit()]tokens2 [t for t in tokenizer.tokenize(input2.lower())if t not in string.punctuation and not t.isdigit()]# 移除特殊token进行比较if hasattr(tokenizer, unk_token):tokens1 [t for t in tokens1 if t ! tokenizer.unk_token]tokens2 [t for t in tokens2 if t ! tokenizer.unk_token]if tokens1 tokens2:consistent_count 1results[consistency] consistent_count / len(consistency_tests)# 5. 速度测试try:start time.time()for _ in range(10): # 减少测试次数避免过长for text in test_corpus[:min(50, len(test_corpus))]: # 限制测试样本数tokenizer.encode(text)elapsed time.time() - startresults[speed] len(test_corpus) / max(elapsed, 0.001) # 每秒处理的文档数except Exception as e:if verbose:print(f速度测试出错: {str(e)})results[speed] 0.0# 6. 子词比例适用于BPE/WordPiece等子词分词器subword_count 0total_token_count 0for text in test_corpus[:100]: # 取前100个样本tokens tokenizer.tokenize(text.lower())total_token_count len(tokens)# 检测子词标记如##ing, ##ed等subword_count sum(1 for t in tokens if re.search(r^##|\w$, t))results[subword_ratio] subword_count / max(1, total_token_count)# 7. 特殊token比例special_tokens set()if hasattr(tokenizer, all_special_tokens):special_tokens set(tokenizer.all_special_tokens)elif hasattr(tokenizer, special_tokens_map):special_tokens set(tokenizer.special_tokens_map.values())special_token_count 0for text in test_corpus[:100]:tokens tokenizer.tokenize(text.lower())special_token_count sum(1 for t in tokens if t in special_tokens)results[special_token_ratio] special_token_count / max(1, total_token_count)if verbose:print(f分词器评估结果:)print(f- 未知词率(UNK): {results[unk_rate]:.4f})print(f- OOV率: {results[oov_rate]:.4f})print(f- 词汇覆盖率: {results[vocabulary_coverage]:.4f})print(f- 平均序列长度: {results[avg_length]:.2f})print(f- 压缩率(字符/token): {results[compression_ratio]:.2f})print(f- 一致性得分: {results[consistency]:.2f})print(f- 处理速度: {results[speed]:.2f} 文档/秒)print(f- 子词比例: {results[subword_ratio]:.4f})print(f- 特殊token比例: {results[special_token_ratio]:.4f})return resultsclass Vocab:词汇表管理类def __init__(self, max_size30000):self.max_size max_sizeself.token_to_id {}self.id_to_token {}self.token_freq {}self.next_id 0# 保留特殊token IDself.special_tokens {[PAD]: 0, [UNK]: 1}for token, idx in self.special_tokens.items():self.token_to_id[token] idxself.id_to_token[idx] tokenself.token_freq[token] float(inf) # 确保特殊token不会被移除def add_token(self, token):添加单个tokenif token not in self.token_to_id:if self.next_id self.max_size:self.token_to_id[token] self.next_idself.id_to_token[self.next_id] tokenself.token_freq[token] 1self.next_id 1return self.next_id - 1else:self.token_freq[token] 1return self.token_to_id[token]return self.special_tokens[[UNK]]def add_or_get(self, token):添加或获取token IDif isinstance(token, bytes):token token.decode(utf-8, replace)return self.add_token(token)def __len__(self):return self.next_iddef get_token(self, token_id):根据ID获取tokenreturn self.id_to_token.get(token_id, [UNK])def prune(self):修剪词汇表到最大大小if len(self) self.max_size:return# 按频率排序保留最频繁的tokensorted_tokens sorted([(token, freq) for token, freq in self.token_freq.items()if token not in self.special_tokens],keylambda x: x[1],reverseTrue)# 重置词汇表new_token_to_id self.special_tokens.copy()new_id_to_token {v: k for k, v in self.special_tokens.items()}new_token_freq self.special_tokens.copy()# 添加最频繁的tokenfor i, (token, freq) in enumerate(sorted_tokens[:self.max_size - len(self.special_tokens)]):new_id i len(self.special_tokens)new_token_to_id[token] new_idnew_id_to_token[new_id] tokennew_token_freq[token] freqself.token_to_id new_token_to_idself.id_to_token new_id_to_tokenself.token_freq new_token_freqself.next_id len(new_token_to_id)class TextDataset(Dataset):用于神经分词器的文本数据集def __init__(self, texts, max_seq_len512):self.texts textsself.max_seq_len max_seq_lendef __len__(self):return len(self.texts)def __getitem__(self, idx):text self.texts[idx]# 转为字节序列byte_seq list(text.encode(utf-8))# 生成分割标签 (1表示分割点)labels []for i in range(len(byte_seq) - 1):# 简单规则空格后是分割点实际训练应使用真实分词结果if byte_seq[i] 32: # 空格的ASCII码labels.append(1)else:labels.append(0)# 截断或填充if len(byte_seq) self.max_seq_len:byte_seq byte_seq[:self.max_seq_len]labels labels[:self.max_seq_len - 1]else:# 填充字节序列pad_len self.max_seq_len - len(byte_seq)byte_seq byte_seq [0] * pad_len# 填充标签用-100表示忽略但CrossEntropyLoss需要0/1标签labels labels [0] * (self.max_seq_len - 1 - len(labels))return torch.tensor(byte_seq), torch.tensor(labels)def create_dataloader(texts, batch_size32, max_seq_len512, shuffleTrue):创建数据加载器dataset TextDataset(texts, max_seq_len)return DataLoader(dataset, batch_sizebatch_size, shuffleshuffle)class NeuralTokenizer(nn.Module):神经分词器def __init__(self, vocab_size30000, embedding_dim256, hidden_dim512):super().__init__()self.embedding nn.Embedding(256, embedding_dim) # 字节级嵌入self.lstm nn.LSTM(input_sizeembedding_dim,hidden_sizehidden_dim,num_layers2,batch_firstTrue,bidirectionalTrue)self.classifier nn.Sequential(nn.Linear(hidden_dim * 2, 512),nn.ReLU(),nn.Linear(512, 2) # 是否分割)self.vocab Vocab(vocab_size)self.criterion nn.CrossEntropyLoss()def forward(self, byte_sequence, labelsNone):前向传播Args:byte_sequence: [batch, seq_len] 字节序列labels: [batch, seq_len-1] 分割标签0不分割1分割Returns:logits: [batch, seq_len-1, 2]loss: 标量如果提供labels# 1. 字节嵌入embeddings self.embedding(byte_sequence)# 2. LSTM处理lstm_out, _ self.lstm(embeddings)# 3. 分割预测# 移除最后一个位置无后续字符split_logits self.classifier(lstm_out[:, :-1, :])# 4. 计算损失loss Noneif labels is not None:loss self.criterion(split_logits.reshape(-1, 2),labels.reshape(-1))return split_logits, lossdef tokenize(self, text):神经分词# 1. 转为字节序列byte_seq text.encode(utf-8)byte_tensor torch.tensor(list(byte_seq)).unsqueeze(0)# 2. 预测分割点with torch.no_grad():self.eval()logits, _ self(byte_tensor)predictions torch.argmax(logits, dim-1).squeeze(0).cpu().numpy()# 3. 构建词项tokens []start 0for i, should_split in enumerate(predictions):if should_split 1:token byte_seq[start:i 1]tokens.append(self.vocab.add_or_get(token))start i 1# 添加最后一个词项if start len(byte_seq):tokens.append(self.vocab.add_or_get(byte_seq[start:]))return tokensdef batch_tokenize(self, texts):批量分词return [self.tokenize(text) for text in texts]def train_step(self, byte_tensor, labels):训练步骤self.train()# 1. 前向传播logits, loss self(byte_tensor, labels)return logits, lossdef evaluate(self, byte_tensor, labels):评估步骤self.eval()with torch.no_grad():logits, loss self(byte_tensor, labels)return logits, lossdef save(self, path):保存模型os.makedirs(os.path.dirname(path), exist_okTrue)# 保存模型权重torch.save(self.state_dict(), path)# 保存词汇表vocab_path path.replace(.pt, _vocab.json)with open(vocab_path, w, encodingutf-8) as f:json.dump({token_to_id: self.vocab.token_to_id,id_to_token: {str(k): v for k, v in self.vocab.id_to_token.items()},token_freq: self.vocab.token_freq,next_id: self.vocab.next_id}, f, ensure_asciiFalse, indent2)classmethoddef load(cls, path, vocab_size30000, embedding_dim256, hidden_dim512):加载模型# 创建模型实例model cls(vocab_size, embedding_dim, hidden_dim)# 加载权重model.load_state_dict(torch.load(path, map_locationcpu))model.eval()# 加载词汇表vocab_path path.replace(.pt, _vocab.json)if os.path.exists(vocab_path):with open(vocab_path, r, encodingutf-8) as f:vocab_data json.load(f)model.vocab.token_to_id vocab_data[token_to_id]model.vocab.id_to_token {int(k): v for k, v in vocab_data[id_to_token].items()}model.vocab.token_freq vocab_data[token_freq]model.vocab.next_id vocab_data[next_id]return modeldef train_neural_tokenizer(model,train_dataloader,val_dataloader,optimizer,num_epochs10,devicecuda if torch.cuda.is_available() else cpu,save_dir./tokenizer_checkpoints,log_interval100
):训练神经分词器os.makedirs(save_dir, exist_okTrue)model.to(device)# 训练日志train_losses []val_losses []val_accuracies []best_val_loss float(inf)print(f开始训练设备: {device})for epoch in range(num_epochs):# 训练阶段model.train()epoch_train_loss 0steps 0train_progress tqdm(train_dataloader, descfEpoch {epoch 1}/{num_epochs} [Train])for byte_seqs, labels in train_progress:byte_seqs byte_seqs.to(device)labels labels.to(device)# 训练步骤optimizer.zero_grad()_, loss model.train_step(byte_seqs, labels)loss.backward()optimizer.step()epoch_train_loss loss.item()steps 1if steps % log_interval 0:train_progress.set_postfix({loss: loss.item()})avg_train_loss epoch_train_loss / stepstrain_losses.append(avg_train_loss)# 验证阶段val_loss, val_acc evaluate_model(model, val_dataloader, device)val_losses.append(val_loss)val_accuracies.append(val_acc)print(fEpoch {epoch 1}/{num_epochs} | fTrain Loss: {avg_train_loss:.4f} | fVal Loss: {val_loss:.4f} | fVal Acc: {val_acc:.4f})# 保存最佳模型if val_loss best_val_loss:best_val_loss val_lossbest_model_path os.path.join(save_dir, best_tokenizer.pt)model.save(best_model_path)print(f保存最佳模型到 {best_model_path})# 保存检查点epoch_path os.path.join(save_dir, ftokenizer_epoch_{epoch 1}.pt)model.save(epoch_path)# 保存训练曲线plot_training_curves(train_losses, val_losses, val_accuracies, save_dir)return {train_losses: train_losses,val_losses: val_losses,val_accuracies: val_accuracies}def evaluate_model(model, dataloader, devicecuda if torch.cuda.is_available() else cpu):评估分词器性能model.eval()total_loss 0total_correct 0total_samples 0with torch.no_grad():for byte_seqs, labels in tqdm(dataloader, descEvaluating):byte_seqs byte_seqs.to(device)labels labels.to(device)# 获取预测logits, loss model.evaluate(byte_seqs, labels)# 计算准确率predictions torch.argmax(logits, dim-1)valid_mask (labels ! -100) # 忽略填充位置correct ((predictions labels) valid_mask).sum().item()total_correct correcttotal_samples valid_mask.sum().item()total_loss loss.item() * byte_seqs.size(0)avg_loss total_loss / len(dataloader.dataset)accuracy total_correct / total_samples if total_samples 0 else 0return avg_loss, accuracydef plot_training_curves(train_losses, val_losses, val_accuracies, save_dir):绘制训练曲线plt.figure(figsize(12, 4))# 损失曲线plt.subplot(1, 2, 1)plt.plot(train_losses, labelTrain Loss)plt.plot(val_losses, labelValidation Loss)plt.xlabel(Epoch)plt.ylabel(Loss)plt.title(Training and Validation Loss)plt.legend()# 准确率曲线plt.subplot(1, 2, 2)plt.plot(val_accuracies, labelValidation Accuracy, colororange)plt.xlabel(Epoch)plt.ylabel(Accuracy)plt.title(Validation Accuracy)plt.legend()plt.tight_layout()plt.savefig(os.path.join(save_dir, training_curves.png))plt.close()def analyze_tokenizer(model, test_texts, devicecuda if torch.cuda.is_available() else cpu):分析分词器效果model.to(device)model.eval()results []for text in test_texts:# 获取模型预测byte_seq list(text.encode(utf-8))byte_tensor torch.tensor(byte_seq).unsqueeze(0).to(device)with torch.no_grad():logits, _ model(byte_tensor)predictions torch.argmax(logits, dim-1).squeeze(0).cpu().numpy()# 构建分词结果tokens []start 0for i, should_split in enumerate(predictions):if should_split 1:token byte_seq[start:i 1]tokens.append(token)start i 1if start len(byte_seq):tokens.append(byte_seq[start:])# 转换为可读形式readable_tokens []for token in tokens:try:readable_tokens.append(bytes(token).decode(utf-8))except:readable_tokens.append(f[BYTES:{len(token)}])results.append({text: text,tokens: readable_tokens,token_count: len(tokens),original_length: len(text),tokenized_length: len(tokens)})return results# 评估分词器的完整函数
def comprehensive_evaluation(model, test_texts, devicecuda if torch.cuda.is_available() else cpu):对神经分词器进行全面评估Args:model: 训练好的神经分词器模型test_texts: 测试文本列表device: 计算设备Returns:dict: 包含所有评估指标的字典model.to(device)model.eval()# 1. 基本统计信息total_texts len(test_texts)total_chars sum(len(text) for text in test_texts)# 2. 分词结果分析all_token_counts []all_char_token_ratios []print(正在进行分词分析...)for text in tqdm(test_texts, desc分词分析):# 获取分词结果byte_seq list(text.encode(utf-8))byte_tensor torch.tensor(byte_seq).unsqueeze(0).to(device)with torch.no_grad():logits, _ model(byte_tensor)predictions torch.argmax(logits, dim-1).squeeze(0).cpu().numpy()# 计算token数量token_count predictions.sum() 1 # 分割点数1 token数all_token_counts.append(token_count)# 计算字符/token比率char_token_ratio len(text) / token_count if token_count 0 else 0all_char_token_ratios.append(char_token_ratio)# 3. 性能测试print(正在进行性能测试...)start_time time.time()for _ in range(10): # 运行10次取平均值for text in test_texts[:min(10, len(test_texts))]: # 取前10个文本测试byte_seq list(text.encode(utf-8))byte_tensor torch.tensor(byte_seq).unsqueeze(0).to(device)with torch.no_grad():logits, _ model(byte_tensor)elapsed_time time.time() - start_timeavg_processing_time elapsed_time / (10 * min(10, len(test_texts)))# 4. 一致性测试print(正在进行一致性测试...)consistency_tests [(Hello World, hello world),(HELLO WORLD, hello world),(New-York, New York),(Im fine, I am fine),(U.S.A., USA)]consistent_count 0for text1, text2 in consistency_tests:# 对两个文本进行分词确保数据在正确的设备上byte_seq1 list(text1.encode(utf-8))byte_tensor1 torch.tensor(byte_seq1).unsqueeze(0).to(device)byte_seq2 list(text2.encode(utf-8))byte_tensor2 torch.tensor(byte_seq2).unsqueeze(0).to(device)# 直接使用模型进行预测而不是调用tokenize方法with torch.no_grad():logits1, _ model(byte_tensor1)logits2, _ model(byte_tensor2)predictions1 torch.argmax(logits1, dim-1).squeeze(0).cpu().numpy()predictions2 torch.argmax(logits2, dim-1).squeeze(0).cpu().numpy()# 简单比较token数量实际应用中可以更复杂token_count1 predictions1.sum() 1token_count2 predictions2.sum() 1if token_count1 token_count2:consistent_count 1consistency_score consistent_count / len(consistency_tests)# 5. 构建评估报告evaluation_results {basic_stats: {total_texts: total_texts,total_characters: total_chars,avg_text_length: total_chars / total_texts if total_texts 0 else 0},tokenization_metrics: {avg_tokens_per_text: np.mean(all_token_counts) if all_token_counts else 0,std_tokens_per_text: np.std(all_token_counts) if all_token_counts else 0,avg_char_token_ratio: np.mean(all_char_token_ratios) if all_char_token_ratios else 0,min_tokens: min(all_token_counts) if all_token_counts else 0,max_tokens: max(all_token_counts) if all_token_counts else 0},performance: {avg_processing_time_per_text: avg_processing_time,texts_per_second: 1 / avg_processing_time if avg_processing_time 0 else 0},quality: {consistency_score: consistency_score}}return evaluation_resultsdef print_evaluation_report(evaluation_results):打印评估报告Args:evaluation_results: comprehensive_evaluation函数返回的结果print(\n * 50)print(神经分词器评估报告)print( * 50)# 基本统计信息print(\n1. 基本统计信息:)print(f - 测试文本数量: {evaluation_results[basic_stats][total_texts]})print(f - 总字符数: {evaluation_results[basic_stats][total_characters]})print(f - 平均文本长度: {evaluation_results[basic_stats][avg_text_length]:.2f} 字符)# 分词指标print(\n2. 分词指标:)print(f - 平均每文本token数: {evaluation_results[tokenization_metrics][avg_tokens_per_text]:.2f})print(f - token数标准差: {evaluation_results[tokenization_metrics][std_tokens_per_text]:.2f})print(f - 平均字符/token比率: {evaluation_results[tokenization_metrics][avg_char_token_ratio]:.2f})print(f - 最小token数: {evaluation_results[tokenization_metrics][min_tokens]})print(f - 最大token数: {evaluation_results[tokenization_metrics][max_tokens]})# 性能指标print(\n3. 性能指标:)print(f - 平均处理时间: {evaluation_results[performance][avg_processing_time_per_text]:.4f} 秒/文本)print(f - 处理速度: {evaluation_results[performance][texts_per_second]:.2f} 文本/秒)# 质量指标print(\n4. 质量指标:)print(f - 一致性得分: {evaluation_results[quality][consistency_score]:.2f})# 示例用法
if __name__ __main__:# 1. 准备数据print(准备训练数据...)sample_texts [The quick brown fox jumps over the lazy dog.,Natural language processing is a subfield of artificial intelligence.,Deep learning models require large amounts of training data.,Tokenization is the process of breaking text into smaller units.,The capital of France is Paris.,How are you doing today?,Im working on a new machine learning project.,The weather is nice outside.,Lets test this tokenizer with various inputs!,Special characters: !#$%^*()_-[]{}|;:\,./?,Numbers: 123, 456.789, 1,000,000,Abbreviations: Dr., Mr., Mrs., U.S.A., etc.,Contractions: dont, cant, Im, youre, weve,Hyphenated words: state-of-the-art, co-operation, e-mail,New-York is a big city with many skyscrapers.,unbelievable, running, happiness, international]# 创建训练集和验证集train_texts sample_texts[:14]val_texts sample_texts[14:]# 2. 创建数据加载器print(创建数据加载器...)train_loader create_dataloader(train_texts, batch_size4, max_seq_len128)val_loader create_dataloader(val_texts, batch_size4, max_seq_len128, shuffleFalse)# 3. 初始化模型print(初始化模型...)model NeuralTokenizer(vocab_size5000, embedding_dim256, hidden_dim512)# 4. 设置优化器optimizer optim.Adam(model.parameters(), lr0.001)# 5. 训练模型print(\n开始训练...)start_time time.time()train_metrics train_neural_tokenizer(model,train_loader,val_loader,optimizer,num_epochs10,devicecuda if torch.cuda.is_available() else cpu,save_dir./tokenizer_checkpoints)print(f训练完成! 用时: {time.time() - start_time:.2f}秒)# 6. 加载最佳模型best_model NeuralTokenizer.load(./tokenizer_checkpoints/best_tokenizer.pt)# 7. 测试分词效果print(\n测试分词效果...)test_texts [The quick brown fox jumps over the lazy dog.,unbelievable,New-York is a big city.]# 分析分词结果analysis analyze_tokenizer(best_model, test_texts)for result in analysis:print(f\n文本: {result[text]})print(f分词结果 ({result[token_count]} tokens): { | .join(result[tokens])})# 8. 全面评估分词器质量print(\n进行全面评估...)evaluation_results comprehensive_evaluation(best_model, sample_texts)print_evaluation_report(evaluation_results)# 9. 绘制训练曲线plot_training_curves(train_metrics[train_losses],train_metrics[val_losses],train_metrics[val_accuracies],./tokenizer_checkpoints)print(训练曲线已保存到 ./tokenizer_checkpoints/training_curves.png)
运行结果
测试分词效果...
分词分析: 0%| | 0/19 [00:00?, ?it/s]
文本: The quick brown fox jumps over the lazy dog.
分词结果 (9 tokens): The | quick | brown | fox | jumps | over | the | lazy | dog.文本: unbelievable
分词结果 (1 tokens): unbelievable文本: New-York is a big city.
分词结果 (5 tokens): New-York | is | a | big | city.进行全面评估...
正在进行分词分析...
正在进行性能测试...
分词分析: 100%|██████████| 19/19 [00:0000:00, 526.38it/s]
正在进行一致性测试...
神经分词器评估报告
1. 基本统计信息:- 测试文本数量: 19- 总字符数: 724- 平均文本长度: 38.11 字符2. 分词指标:- 平均每文本token数: 5.47- token数标准差: 2.93- 平均字符/token比率: 7.92- 最小token数: 1- 最大token数: 103. 性能指标:- 平均处理时间: 0.0024 秒/文本- 处理速度: 421.10 文本/秒4. 质量指标:- 一致性得分: 0.40
训练曲线已保存到 ./tokenizer_checkpoints/training_curves.png优势
学习更复杂的分词模式捕获上下文相关的分词决策自动处理多语言和特殊领域可端到端训练与模型协同优化突破传统算法的局限性
3. 实用建议
3.1 实施路线图需求分析
分析任务对分词的敏感度评估语言和领域特性测量当前分词器的瓶颈基础实现
选择合适的分词算法设置合理的词汇量添加必要的特殊标记高级优化
领域适应性优化与模型架构协同设计实施动态分词策略持续优化
监控分词器性能指标定期重新评估分词策略收集性能-质量权衡数据前沿应用
试验神经分词器探索动态分词技术结合任务特定优化3.2 常见误区误区1“分词器对所有任务都一样重要”
事实任务特性决定分词需求应定制化误区2“更大的词汇表总是更好”
事实存在最优词汇量超过后收益递减误区3“分词器不需要考虑模型架构”
事实分词器与模型架构应协同设计误区4“分词只影响训练阶段”
事实也显著影响推理效率和模型质量误区5“只需关注UNK率”
事实应综合考虑序列长度、一致性等指标终极洞见分词器不是终点而是语言理解的起点——它不仅是将文本转换为数字的工具更是模型理解语言结构的桥梁。掌握分词器技术您将能够构建既高效又强大的语言模型显著提升模型性能降低训练成本同时最大化模型对语言的理解能力。这不仅是技术能力更是语言学思维的体现在有限的表示能力下找到最优的语言编码方式。记住最好的分词器是您几乎感觉不到它的存在但它始终默默优化着模型的语言理解让AI真正理解人类的语言。
