用Python实战HMM中文分词：从概率计算到Viterbi解码

自然语言处理中的中文分词一直是个有趣且实用的课题。想象一下，你正在开发一个电影评论分析系统，用户输入"这部电影太好看了"，如何让计算机理解"这部/电影/太/好看/了"这样的合理切分？传统方法依赖词典匹配，但遇到新词就束手无策。隐马尔可夫模型(HMM)通过概率统计的方式，能自动学习分词规律，即使面对未登录词也有不错的表现。

今天我们不谈复杂的数学推导，直接动手用Python实现一个完整的HMM分词器。你会看到，从原始语料到最终分词结果，整个过程就像搭积木一样清晰有趣。我们将使用一个小型电影评论语料库作为示例，这样你可以立即看到模型在真实场景中的应用效果。

1. 环境准备与数据理解

首先确保你的Python环境安装了这些基础包：

pip install numpy tqdm

我们使用的训练数据是人工标注的电影评论，每个字后面跟着它的B/M/E/S标签：

• B(词首)、M(词中)、E(词尾)、S(单字词)

示例数据片段：

这/B 部/E 电/B 影/E 真/S 的/S 好/B 看/E ！/S
我/S 喜/B 欢/E 这/B 种/E 剧/B 情/E 片/E

关键理解 ：HMM认为每个字的标签(状态)只与前一个标签有关，而当前字(观测值)只与当前标签有关。这种"马尔可夫假设"大大简化了问题复杂度。

提示：实际项目中建议至少准备10万字的标注数据。我们这里为演示简化，只使用几十条评论。

2. 概率统计的三部曲

2.1 初始化概率矩阵

我们需要三个核心概率矩阵：

import numpy as np
from collections import defaultdict

# 初始化计数器
init_count = defaultdict(int)     # 初始状态计数
trans_count = defaultdict(int)    # 状态转移计数
emit_count = defaultdict(int)     # 发射计数
state_count = defaultdict(int)    # 状态出现总次数

统计过程示例：

def count_tags(sentences):
    for sentence in sentences:
        prev_tag = None
        for word, tag in sentence:
            if prev_tag is None:  # 句首字
                init_count[tag] += 1
            else:  # 非句首字
                trans_count[(prev_tag, tag)] += 1
            emit_count[(tag, word)] += 1
            state_count[tag] += 1
            prev_tag = tag

统计后的原始计数示例：

初始状态计数：{'B': 15, 'S': 32}
转移计数：{('B','E'):12, ('B','M'):3, ('E','B'):7, ...}
发射计数：{('B','电'):5, ('E','影'):8, ('S','我'):6, ...}

2.2 概率计算与平滑处理

原始统计会遇到零概率问题，需要拉普拉斯平滑：

def prob_with_smoothing(count, total, alpha=1.0, states=4):
    return (count + alpha) / (total + alpha * states)

生成概率矩阵的核心代码：

tags = ['B', 'M', 'E', 'S']
# 初始概率
init_prob = {tag: prob_with_smoothing(init_count[tag], sum(init_count.values())) 
             for tag in tags}

# 转移概率矩阵
trans_prob = {}
for prev in tags:
    for curr in tags:
        key = (prev, curr)
        trans_prob[key] = prob_with_smoothing(trans_count.get(key,0), 
                                            sum(v for k,v in trans_count.items() 
                                                if k[0]==prev))

# 发射概率
emit_prob = {}
for (tag, word), cnt in emit_count.items():
    emit_prob[(tag, word)] = cnt / state_count[tag]

注意：实际应用中，发射概率对生僻字要做特殊处理，比如统一赋予一个极小值。

3. Viterbi算法实现

3.1 算法原理图解

Viterbi算法的精妙之处在于，它通过动态规划找到概率最大的状态路径。想象你在玩一个格子游戏：

1. 每个格子代表一个字可能的状态(B/M/E/S)
2. 格子之间的箭头代表转移概率
3. 格子本身的亮度代表发射概率
4. 从起点到终点要找出一条最亮的路径

3.2 Python代码实现

完整Viterbi实现：

def viterbi(sentence, init_prob, trans_prob, emit_prob, tags):
    # 初始化DP表格
    dp = [{} for _ in range(len(sentence))]  # 每个字的每个状态的最大概率
    path = {}  # 记录路径
    
    # 初始化第一个字
    for tag in tags:
        dp[0][tag] = init_prob.get(tag, 1e-6) * emit_prob.get((tag, sentence[0]), 1e-6)
        path[tag] = [tag]
    
    # 递推计算
    for i in range(1, len(sentence)):
        new_path = {}
        for curr_tag in tags:
            max_prob = -1
            best_prev_tag = None
            for prev_tag in tags:
                prob = dp[i-1][prev_tag] * \
                       trans_prob.get((prev_tag, curr_tag), 1e-6) * \
                       emit_prob.get((curr_tag, sentence[i]), 1e-6)
                if prob > max_prob:
                    max_prob = prob
                    best_prev_tag = prev_tag
            dp[i][curr_tag] = max_prob
            new_path[curr_tag] = path[best_prev_tag] + [curr_tag]
        path = new_path
    
    # 回溯最佳路径
    best_tag = max(dp[-1], key=dp[-1].get)
    return path[best_tag]

测试用例：

text = "这部电影太精彩了"
tags = viterbi(text, init_prob, trans_prob, emit_prob, ['B','M','E','S'])
print("字\t标签")
for char, tag in zip(text, tags):
    print(f"{char}\t{tag}")

输出示例：

字      标签
这      B
部      E
电      B
影      E
太      S
精      B
彩      E
了      S

4. 后处理与性能优化

4.1 从标签序列到分词结果

将标签序列转换为最终分词：

def tags_to_segs(sentence, tags):
    segs = []
    word = []
    for char, tag in zip(sentence, tags):
        word.append(char)
        if tag in ['E', 'S']:  # 词结束或单字
            segs.append(''.join(word))
            word = []
    return segs

4.2 工程优化技巧

1. 对数空间计算 ：避免概率下溢

math.log(prob)  # 将乘法转为加法

2. 剪枝策略 ：每步只保留top-k路径

# 在Viterbi中增加
if len(dp[i]) > beam_size:
    dp[i] = dict(sorted(dp[i].items(), key=lambda x: -x[1])[:beam_size])

3. 模型持久化 ：训练后保存概率矩阵

import pickle
with open('hmm_model.pkl', 'wb') as f:
    pickle.dump((init_prob, trans_prob, emit_prob), f)

在实际项目中，我通常会先用jieba等成熟工具生成标注数据来训练自己的HMM模型，这样既能保证数据质量，又能理解底层机制。当遇到特定领域文本时，这种自训练的模型往往比通用模型表现更好。