Python之Viterbi算法介绍与理解

什么是Viterbi算法？

求助度娘后得知：Viterbi算法中文名字叫做维特比算法，听上去好像比特币算法。它是一种动态规划算法用于寻找最有可能产生观测事件序列的-维特比路径-隐含状态序列。通常我们叫做最优路径算法。

什么是最优路径？

最优路径意思就是计算一个节点到另一个节点的最短路径，请看下图

我们可以直观的看到，s要想到达e，有四种显而易见的方式

1. s->x1->y1-e，路径为10
2. s->x1->y2->e，路径为25
3. s->x2->y1->e，路径为13
4. s->x2->y2->e，路径为17

由此可见，最优路径就是第一条，因为它的路径最短。

Viterbi算法的中心思想

1.如果概率最大的路径p(或者说最短路径)经过某个点，比如途中的y1，那么这条路径上的起始点S到y1的这段子路径Q(s->x1->y1)，一定是S到y1之间的最短路径。否则，用S到y1的最短路径R(s->x2-y1)替代Q，便构成一条比P更短的路径，这显然是矛盾的。证明了满足最优性原理。

2.从S到E的路径必定经过第i个时刻的某个状态，假定第i个时刻有k个状态（例如上图必定经过某个x和某个y），那么如果记录了从S到第i个状态的所有k个节点的最短路径，最终的最短路径必经过其中一条，这样，在任意时刻，只要考虑非常有限的最短路即可。

3. 结合以上两点，假定当我们从状态i进入状态i+1时，从S到状态i上各个节的最短路径已经找到，并且记录在这些节点上，那么在计算从起点S到第i+1状态的某个节点X_i+1的最短路径时，只要考虑从S到前一个状态i所有的k个节点的最短路径，以及从这个节点到X_i+1，j的距离即可。

官方代码的理解

import sys
import operator
MIN_FLOAT = -3.14e100
MIN_INF = float("-inf")

if sys.version_info[0] > 2:
    xrange = range


def get_top_states(t_state_v, K=4):
    return sorted(t_state_v, key=t_state_v.__getitem__, reverse=True)[:K]


'''
obs:观测序列（已知)
states:隐含状态，通过隐含状态我们要找到最佳的观测序列，观测序列里面的元素从states里面寻找
start_p:初始概率，即obs[0]对应的隐含状态中每个state能够出现的状态的概率
trans_p:转换概率，从一个隐含状态到下一个隐含状态的概率
emit_p:发射概率,某种隐含状态产生某种观测现象的概率
'''


def viterbi(obs, states, start_p, trans_p, emit_p):
    # 用来记录当前观测序列的所有隐含状态能够构成的最大值，key表示状态value表示到达该点的最优。
    V = [{}]  # tabular
    # 字典列，字典的key为
    mem_path = [{}]
    all_states = trans_p.keys()
    for y in states.get(obs[0], all_states):  # init
        V[0][y] = start_p[y] + emit_p[y].get(obs[0], MIN_FLOAT)
        # print(V[0])
        mem_path[0][y] = ''
    for t in xrange(1, len(obs)):
        V.append({})
        mem_path.append({})
        #   prev_states = get_top_states(V[t-1])
        #   获取前一个状态
        prev_states = [
            x for x in mem_path[t - 1].keys() if len(trans_p[x]) > 0]
        #   根据前面一个状态推断出后面可能出现的状态有哪些
        prev_states_expect_next = set(
            (y for x in prev_states for y in trans_p[x].keys()))
        #   当前观测序列对应的状态与前面所推断出的状态求交集
        obs_states = set(
            states.get(obs[t], all_states)) & prev_states_expect_next
        #   如果交集为空就取推断，如果推断也为空就取全部下来
        if not obs_states:
            obs_states = prev_states_expect_next if prev_states_expect_next else all_states
        #   算出当前t的最优解并记录
        for y in obs_states:
            #   当前这个点的最大值，以及当前最大值对应的state
            prob, state = max((V[t - 1][y0] + trans_p[y0].get(y, MIN_INF) +
                               emit_p[y].get(obs[t], MIN_FLOAT), y0) for y0 in prev_states)
            V[t][y] = prob
            mem_path[t][y] = state

    last = [(V[-1][y], y) for y in mem_path[-1].keys()]
    # if len(last)==0:
    #     print obs
    prob, state = max(last)
    route = [None] * len(obs)
    # 由于mem_path的类型所以需要用回溯的方法去找出最优路径
    i = len(obs) - 1
    while i >= 0:
        route[i] = state
        state = mem_path[i][state]
        i -= 1
    return (prob, route)