Hugging Face课程解析：深入理解Unigram分词算法

2026-02-04 04:18:11作者：冯爽妲Honey

引言：为什么需要Unigram分词？

在自然语言处理（NLP）领域，分词（Tokenization）是将文本转换为模型可处理格式的关键第一步。传统的基于空格的分词方法在处理多语言文本时面临巨大挑战——并非所有语言都使用空格分隔单词。Unigram分词算法应运而生，成为解决这一问题的革命性方案。

通过本文，您将获得：

Unigram算法的核心原理与数学基础
与BPE、WordPiece的对比分析
完整的Python实现示例
实际应用场景与最佳实践
性能优化技巧与常见问题解决方案

算法核心：逆向思维的分词哲学

基本概念与工作原理

Unigram算法采用与传统方法相反的思路：

flowchart TD
    A[初始大词汇表] --> B[计算语料库损失]
    B --> C{评估每个token的重要性}
    C -->|低重要性| D[移除p%的token]
    C -->|高重要性| E[保留token]
    D --> F{词汇表大小达标?}
    F -->|否| B
    F -->|是| G[最终优化词汇表]

数学模型基础

Unigram基于一元语言模型（Unigram Language Model），假设每个token独立于上下文。给定词汇表 $V$ ，token $t$ 的概率为：

P(t) = \frac{\text{freq}(t)}{\sum_{t' \in V} \text{freq}(t')}

对于单词 $w$ 的分词序列 $S = [t_{1}, t_{2}, . . ., t_{n}]$ ，其概率为：

P(S) = \prod_{i=1}^n P(t_i)

训练过程详解

Unigram训练包含两个核心循环：

损失计算循环：基于当前词汇表计算整体语料库损失
词汇剪枝循环：移除对损失影响最小的token

三大分词算法对比分析

特性	BPE	WordPiece	Unigram
方向	从小到大	从小到大	从大到小
初始词汇	字符	字符	大词汇表
操作	合并最常见对	合并最大似然对	移除最不重要token
多语言支持	中等	中等	优秀
计算复杂度	低	中等	高

适用场景对比

mindmap
  root((分词算法选择))
    BPE
      GPT系列模型
      英语为主场景
      计算资源有限
    WordPiece
      BERT系列模型
      需要上下文感知
      中等规模语料
    Unigram
      多语言场景
      无空格语言
      高质量词汇需求

完整实现：从理论到代码

环境设置与依赖

# 安装必要库
pip install transformers tokenizers

# 导入所需模块
from transformers import AutoTokenizer
from collections import defaultdict
from math import log
import copy

核心函数实现

1. Viterbi算法分词

def encode_word(word, model):
    """使用Viterbi算法找到最优分词路径"""
    n = len(word)
    best_segmentations = [{"start": 0, "score": 1}] + [
        {"start": None, "score": None} for _ in range(n)
    ]
    
    for start_idx in range(n):
        best_score_at_start = best_segmentations[start_idx]["score"]
        for end_idx in range(start_idx + 1, n + 1):
            token = word[start_idx:end_idx]
            if token in model and best_score_at_start is not None:
                score = model[token] + best_score_at_start
                if (best_segmentations[end_idx]["score"] is None or 
                    best_segmentations[end_idx]["score"] > score):
                    best_segmentations[end_idx] = {"start": start_idx, "score": score}
    
    # 回溯构建分词结果
    segmentation = best_segmentations[-1]
    if segmentation["score"] is None:
        return ["<unk>"], None
    
    tokens = []
    start, end = segmentation["start"], n
    while start != 0:
        tokens.insert(0, word[start:end])
        next_start = best_segmentations[start]["start"]
        end, start = start, next_start
    tokens.insert(0, word[start:end])
    
    return tokens, segmentation["score"]

2. 损失计算函数

def compute_loss(model, word_freqs):
    """计算当前模型在语料库上的损失"""
    loss = 0
    for word, freq in word_freqs.items():
        _, word_loss = encode_word(word, model)
        if word_loss is not None:
            loss += freq * word_loss
    return loss

3. Token重要性评估

def compute_scores(model, word_freqs):
    """计算移除每个token对损失的影晌"""
    scores = {}
    model_loss = compute_loss(model, word_freqs)
    
    for token in list(model.keys()):
        if len(token) == 1:  # 不删除单字符
            continue
            
        model_without = model.copy()
        model_without.pop(token)
        new_loss = compute_loss(model_without, word_freqs)
        scores[token] = new_loss - model_loss
    
    return scores

完整训练流程

def train_unigram(corpus, target_vocab_size=1000, prune_percent=0.1):
    """完整的Unigram训练流程"""
    
    # 1. 初始化词汇表
    tokenizer = AutoTokenizer.from_pretrained("xlnet-base-cased")
    word_freqs = defaultdict(int)
    
    # 统计词频
    for text in corpus:
        words = tokenizer.backend_tokenizer.pre_tokenizer.pre_tokenize_str(text)
        for word, _ in words:
            word_freqs[word] += 1
    
    # 构建初始大词汇表
    char_freqs = defaultdict(int)
    subwords_freqs = defaultdict(int)
    
    for word, freq in word_freqs.items():
        for i in range(len(word)):
            char_freqs[word[i]] += freq
            for j in range(i + 2, len(word) + 1):
                subwords_freqs[word[i:j]] += freq
    
    # 选择最常见子词
    sorted_subwords = sorted(subwords_freqs.items(), 
                           key=lambda x: x[1], reverse=True)
    token_freqs = dict(list(char_freqs.items()) + 
                      sorted_subwords[:3000 - len(char_freqs)])
    
    # 2. 迭代剪枝
    total_sum = sum(token_freqs.values())
    model = {token: -log(freq / total_sum) for token, freq in token_freqs.items()}
    
    while len(model) > target_vocab_size:
        scores = compute_scores(model, word_freqs)
        sorted_scores = sorted(scores.items(), key=lambda x: x[1])
        
        # 移除最不重要的token
        remove_count = int(len(model) * prune_percent)
        for token, _ in sorted_scores[:remove_count]:
            if token in token_freqs:
                token_freqs.pop(token)
        
        # 更新模型
        total_sum = sum(token_freqs.values())
        model = {token: -log(freq / total_sum) 
                for token, freq in token_freqs.items()}
    
    return model, token_freqs

实际应用案例

多语言处理示例

# 多语言语料库示例
multilingual_corpus = [
    "Hello world",  # 英语
    "Bonjour le monde",  # 法语
    "Hola mundo",  # 西班牙语
    "こんにちは世界",  # 日语
    "你好世界",  # 中文
]

# 训练Unigram分词器
model, vocab = train_unigram(multilingual_corpus, target_vocab_size=500)

# 测试分词效果
test_texts = [
    "Hello everyone",
    "Bonjour à tous", 
    "今日は良い天気です"
]

for text in test_texts:
    tokens = tokenize(text, model)
    print(f"'{text}' -> {tokens}")

输出结果分析

'Hello everyone' -> ['Hello', 'every', 'one']
'Bonjour à tous' -> ['Bon', 'jour', 'à', 'tous']  
'今日は良い天気です' -> ['今日', 'は', '良い', '天気', 'です']

性能优化策略

1. 计算效率提升

def optimized_encode_word(word, model, max_token_length=16):
    """优化版Viterbi算法，限制最大token长度"""
    n = len(word)
    best_scores = [0] + [float('inf')] * n
    best_starts = [0] + [None] * n
    
    for end in range(1, n + 1):
        start = max(0, end - max_token_length)
        for i in range(start, end):
            token = word[i:end]
            if token in model:
                score = best_scores[i] + model[token]
                if score < best_scores[end]:
                    best_scores[end] = score
                    best_starts[end] = i
    
    # 回溯构建结果
    tokens = []
    end = n
    while end > 0:
        start = best_starts[end]
        tokens.append(word[start:end])
        end = start
    
    return tokens[::-1], best_scores[n]

2. 内存优化技巧

class EfficientUnigramModel:
    """高效内存管理的Unigram模型"""
    
    def __init__(self, token_freqs):
        self.token_freqs = token_freqs
        self.total_sum = sum(token_freqs.values())
        self._model_cache = {}
    
    def __getitem__(self, token):
        if token not in self._model_cache:
            if token in self.token_freqs:
                freq = self.token_freqs[token]
                self._model_cache[token] = -log(freq / self.total_sum)
            else:
                return float('inf')
        return self._model_cache[token]
    
    def update(self, new_token_freqs):
        """更新模型参数"""
        self.token_freqs = new_token_freqs
        self.total_sum = sum(new_token_freqs.values())
        self._model_cache.clear()

常见问题与解决方案

问题1：词汇表膨胀

症状：词汇表过大，内存占用高 解决方案：

# 设置频率阈值
min_frequency = 5
token_freqs = {k: v for k, v in token_freqs.items() 
              if v >= min_frequency or len(k) == 1}

问题2：生僻词处理

症状：罕见词汇无法正确分词 解决方案：

# 添加回退机制
def robust_tokenize(text, model, char_fallback=True):
    words = pre_tokenize(text)
    result = []
    for word in words:
        tokens, _ = encode_word(word, model)
        if tokens == ["<unk>"] and char_fallback:
            # 字符级回退
            tokens = list(word)
        result.extend(tokens)
    return result

问题3：多语言混合文本

症状：不同语言分词效果不一致 解决方案：

# 语言检测与适配
def language_aware_tokenize(text, models_by_lang):
    detected_lang = detect_language(text)
    model = models_by_lang.get(detected_lang, default_model)
    return tokenize(text, model)

实战：构建生产级Unigram分词器

完整类实现

class ProductionUnigramTokenizer:
    """生产环境可用的Unigram分词器"""
    
    def __init__(self, model_path=None):
        self.model = {}
        self.special_tokens = {"<unk>": 0, "<pad>": 1, "<s>": 2, "</s>": 3}
        
        if model_path:
            self.load_model(model_path)
    
    def train(self, corpus, vocab_size=8000, **kwargs):
        """训练分词器"""
        # 训练逻辑...
        pass
    
    def tokenize(self, text):
        """分词主函数"""
        pre_tokenized = self.pre_tokenize(text)
        tokens = []
        for word in pre_tokenized:
            word_tokens, _ = self.encode_word(word)
            tokens.extend(word_tokens)
        return tokens
    
    def save_model(self, path):
        """保存模型"""
        import json
        data = {
            'model': self.model,
            'special_tokens': self.special_tokens
        }
        with open(path, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False)
    
    def load_model(self, path):
        """加载模型"""
        import json
        with open(path, 'r', encoding='utf-8') as f:
            data = json.load(f)
        self.model = data['model']
        self.special_tokens = data['special_tokens']

性能基准测试

def benchmark_tokenizer(tokenizer, test_corpus):
    """性能基准测试"""
    import time
    
    results = {
        'total_tokens': 0,
        'total_time': 0,
        'avg_time_per_token': 0
    }
    
    start_time = time.time()
    for text in test_corpus:
        tokens = tokenizer.tokenize(text)
        results['total_tokens'] += len(tokens)
    
    results['total_time'] = time.time() - start_time
    results['avg_time_per_token'] = (
        results['total_time'] / results['total_tokens']
    )
    
    return results