别再死记硬背了！用Python手撸一个HMM中文分词器（附完整代码与PKU语料）

weixin_30813225

288人浏览 · 2026-04-24 02:07:32

weixin_30813225 · 2026-04-24 02:07:32 发布

从零构建HMM中文分词器：原理剖析与工程实践

在自然语言处理领域，中文分词一直是基础而关键的环节。市面上成熟的工具如jieba、HanLP等虽然开箱即用，但真正想深入理解其背后的统计学习原理，亲手实现一个分词器才是最佳途径。本文将带您用Python从零实现基于隐马尔可夫模型(HMM)的中文分词系统，不仅包含完整代码实现，还会深入解析统计分词的核心数学原理。

1. 隐马尔可夫模型基础

隐马尔可夫模型是处理序列标注问题的经典概率图模型，特别适合中文分词这种需要考虑上下文关系的任务。HMM假设系统是一个马尔可夫过程，即当前状态只依赖于前一个状态，而观测值只依赖于当前状态。

在中文分词场景中，我们采用B/M/E/S四种标签：

B（Begin）：词语的开始字
M（Middle）：词语的中间字
E（End）：词语的结束字
S（Single）：单字成词

例如句子"自然语言处理"会被标注为：

自/B 然/E 语/B 言/E 处/B 理/E

HMM需要解决三个核心问题：

评估问题：给定模型参数和观测序列，计算该序列出现的概率
解码问题：给定模型参数和观测序列，找出最可能的状态序列
学习问题：给定观测序列，估计模型参数

对于中文分词，我们主要关注解码问题——即找到最可能的标签序列。

2. 数据准备与预处理

高质量的训练数据是模型效果的基础。我们使用北京大学计算语言学研究所提供的PKU语料库，这是中文分词领域的标准数据集之一。

2.1 语料格式处理

原始PKU语料采用空格分隔词语的格式：

迈向 充满 希望 的 新 世纪

我们需要将其转换为逐字标注的形式：

迈/B 向/E 充/B 满/E 希/B 望/E 的/S 新/S 世/B 纪/E

处理脚本如下：

def convert_to_bmes(input_file, output_file):
    with open(input_file, 'r', encoding='utf-8') as fin, \
         open(output_file, 'w', encoding='utf-8') as fout:
        for line in fin:
            line = line.strip()
            if not line:
                continue
            words = line.split()
            for word in words:
                if len(word) == 1:
                    fout.write(f"{word}/S ")
                else:
                    fout.write(f"{word[0]}/B ")
                    for char in word[1:-1]:
                        fout.write(f"{char}/M ")
                    fout.write(f"{word[-1]}/E ")
            fout.write("\n")

2.2 统计特征提取

从标注好的语料中，我们需要统计三类关键概率：

初始概率：每个标签作为序列开头的概率
转移概率：从一个标签转移到另一个标签的概率
发射概率：某个标签生成特定汉字的概率

统计过程需要考虑数据稀疏问题，采用加一平滑(Add-One Smoothing)处理未登录词：

def calculate_probabilities(tagged_corpus):
    # 初始化统计字典
    start_count = {'B':0, 'M':0, 'E':0, 'S':0}
    trans_count = {from_tag: {to_tag:0 for to_tag in ['B','M','E','S']} 
                  for from_tag in ['B','M','E','S']}
    emit_count = {tag: {} for tag in ['B','M','E','S']}
    state_count = {'B':0, 'M':0, 'E':0, 'S':0}
    
    # 遍历语料统计频次
    for sentence in tagged_corpus:
        prev_tag = None
        for char_tag in sentence.split():
            char, tag = char_tag.split('/')
            state_count[tag] += 1
            emit_count[tag][char] = emit_count[tag].get(char, 0) + 1
            
            if prev_tag is None:
                start_count[tag] += 1
            else:
                trans_count[prev_tag][tag] += 1
            prev_tag = tag
    
    # 计算概率（加一平滑）
    total_starts = sum(start_count.values())
    start_prob = {tag: (count+1)/(total_starts+4) for tag, count in start_count.items()}
    
    trans_prob = {}
    for from_tag, counts in trans_count.items():
        total = sum(counts.values())
        trans_prob[from_tag] = {to_tag: (count+1)/(total+4) 
                               for to_tag, count in counts.items()}
    
    emit_prob = {}
    for tag, counts in emit_count.items():
        total = state_count[tag]
        emit_prob[tag] = {char: (count+1)/(total+len(counts)) 
                         for char, count in counts.items()}
    
    return start_prob, trans_prob, emit_prob

3. 模型训练与参数估计

基于统计得到的概率参数，我们可以构建完整的HMM模型类。这里采用面向对象的方式封装核心功能：

import pickle
from collections import defaultdict

class HMMSegmenter:
    def __init__(self):
        self.states = ['B', 'M', 'E', 'S']
        self.start_p = {}  # 初始概率
        self.trans_p = {}  # 转移概率
        self.emit_p = {}   # 发射概率
        self.trained = False
    
    def train(self, corpus_path, model_path='hmm_model.pkl'):
        # 初始化统计字典
        start_count = defaultdict(int)
        trans_count = {s: defaultdict(int) for s in self.states}
        emit_count = {s: defaultdict(int) for s in self.states}
        state_count = defaultdict(int)
        
        # 读取并处理语料
        with open(corpus_path, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                
                prev_state = None
                for char_state in line.split():
                    char, state = char_state.rsplit('/', 1)
                    state_count[state] += 1
                    emit_count[state][char] += 1
                    
                    if prev_state is None:
                        start_count[state] += 1
                    else:
                        trans_count[prev_state][state] += 1
                    prev_state = state
        
        # 计算概率（加一平滑）
        total_start = sum(start_count.values())
        self.start_p = {s: (start_count.get(s, 0)+1)/(total_start+4) 
                       for s in self.states}
        
        self.trans_p = {}
        for s1 in self.states:
            total = sum(trans_count[s1].values())
            self.trans_p[s1] = {s2: (trans_count[s1].get(s2, 0)+1)/(total+4)
                              for s2 in self.states}
        
        self.emit_p = {}
        for s in self.states:
            total = state_count[s]
            vocab_size = len(emit_count[s])
            self.emit_p[s] = {char: (emit_count[s].get(char, 0)+1)/(total+vocab_size)
                            for char in emit_count[s]}
        
        # 保存模型
        with open(model_path, 'wb') as f:
            pickle.dump({
                'start_p': self.start_p,
                'trans_p': self.trans_p,
                'emit_p': self.emit_p
            }, f)
        
        self.trained = True
        print(f"Model trained and saved to {model_path}")

4. 维特比算法实现

维特比(Viterbi)算法是HMM解码问题的经典解决方案，它使用动态规划高效地找出最可能的状态序列。算法分为前向传播和路径回溯两个阶段：

def viterbi(self, text):
    if not self.trained:
        raise ValueError("Model not trained. Please train or load a model first.")
    
    # 初始化维特比矩阵和路径记录
    V = [{}]
    path = {}
    
    # 初始化第一个时间步
    for state in self.states:
        V[0][state] = self.start_p[state] * self.emit_p[state].get(text[0], 1e-10)
        path[state] = [state]
    
    # 递推计算后续时间步
    for t in range(1, len(text)):
        V.append({})
        new_path = {}
        
        for curr_state in self.states:
            max_prob = -1
            best_prev_state = None
            
            # 寻找最可能的前驱状态
            for prev_state in self.states:
                prob = V[t-1][prev_state] * \
                       self.trans_p[prev_state].get(curr_state, 1e-10) * \
                       self.emit_p[curr_state].get(text[t], 1e-10)
                
                if prob > max_prob:
                    max_prob = prob
                    best_prev_state = prev_state
            
            V[t][curr_state] = max_prob
            new_path[curr_state] = path[best_prev_state] + [curr_state]
        
        path = new_path
    
    # 回溯找到最优路径
    last_char = text[-1]
    if self.emit_p['E'].get(last_char, 0) > self.emit_p['S'].get(last_char, 0):
        final_states = ['E', 'M']
    else:
        final_states = ['S']
    
    max_prob = -1
    best_path = None
    for state in final_states:
        if V[-1][state] > max_prob:
            max_prob = V[-1][state]
            best_path = path[state]
    
    return best_path

5. 分词结果生成与评估

获得最优标签序列后，我们需要将其转换为实际的分词结果：

def segment(self, text):
    if not text:
        return []
    
    best_path = self.viterbi(text)
    result = []
    begin = 0
    
    for i in range(len(best_path)):
        state = best_path[i]
        if state == 'B':
            begin = i
        elif state == 'E':
            result.append(text[begin:i+1])
        elif state == 'S':
            result.append(text[i])
    
    # 处理最后一个词未结束的情况
    if begin < len(text):
        result.append(text[begin:])
    
    return result

为了评估分词效果，我们采用标准的精确率(Precision)、召回率(Recall)和F1值：

def evaluate(self, test_file):
    with open(test_file, 'r', encoding='utf-8') as f:
        total_gold = 0
        total_pred = 0
        correct = 0
        
        for line in f:
            line = line.strip()
            if not line:
                continue
            
            # 标准分词结果
            gold_words = line.split()
            gold_seg = list(''.join(gold_words))
            total_gold += len(gold_words)
            
            # 模型预测结果
            raw_text = ''.join(gold_words)
            pred_words = self.segment(raw_text)
            total_pred += len(pred_words)
            
            # 统计正确切分的词语
            gold_set = set()
            pos = 0
            for word in gold_words:
                gold_set.add((pos, pos+len(word)-1))
                pos += len(word)
            
            pos = 0
            for word in pred_words:
                if (pos, pos+len(word)-1) in gold_set:
                    correct += 1
                pos += len(word)
        
        precision = correct / total_pred
        recall = correct / total_gold
        f1 = 2 * precision * recall / (precision + recall)
        
        print(f"Precision: {precision:.4f}")
        print(f"Recall: {recall:.4f}")
        print(f"F1 Score: {f1:.4f}")

6. 工程优化与实践技巧

在实际应用中，我们还需要考虑以下优化点：

6.1 未登录词处理

对于训练集中未出现的汉字，可以采用以下策略：

均匀分布：给所有标签相同的发射概率
基于汉字特性：根据汉字结构、偏旁等特征分配概率
回退平滑：使用更粗粒度的统计信息

def get_emission_prob(self, state, char):
    # 优先使用统计得到的概率
    if char in self.emit_p[state]:
        return self.emit_p[state][char]
    
    # 未登录词处理策略
    if self.unk_handler == "uniform":
        return 1.0 / len(self.emit_p[state])
    elif self.unk_handler == "character":
        # 基于汉字特征的启发式规则
        if len(char) > 1:  # 非中文字符
            return 1e-10
        # 根据汉字结构判断
        if is_common_character(char):
            return 0.1 if state in ['S', 'E'] else 0.01
        else:
            return 1e-5
    else:
        return 1e-10

6.2 模型压缩与加速

当词典规模较大时，可以采用以下优化：

概率取对数，将乘法转为加法
剪枝策略，提前淘汰低概率路径
使用更高效的数据结构存储概率矩阵

def initialize_log_probabilities(self):
    self.log_start_p = {s: math.log(p) for s, p in self.start_p.items()}
    self.log_trans_p = {s1: {s2: math.log(p) for s2, p in s1_p.items()}
                       for s1, s1_p in self.trans_p.items()}
    self.log_emit_p = {s: {char: math.log(p) for char, p in char_p.items()}
                      for s, char_p in self.emit_p.items()}

6.3 混合分词策略

纯HMM分词在长词识别上可能存在不足，可以结合以下方法：

词典匹配：优先匹配已知词表
N-gram语言模型：考虑更长的上下文依赖
规则修正：处理数字、英文等特殊模式

def hybrid_segment(self, text):
    # 先用词典匹配处理已知词语
    known_words = set()  # 加载词典
    result = []
    i = 0
    n = len(text)
    
    while i < n:
        # 查找最长词典词
        matched = None
        for j in range(min(i+self.max_word_len, n), i, -1):
            word = text[i:j]
            if word in known_words:
                matched = word
                break
        
        if matched:
            result.append(matched)
            i += len(matched)
        else:
            # 对剩余部分使用HMM分词
            remaining = text[i:]
            hmm_segs = self.segment(remaining)
            result.extend(hmm_segs)
            break
    
    return result