简单理解viterbi算法

简介：
viterbi算法其实就是多步骤每步多选择模型的最优选择问题，其在每一步的全部选择都保存了前续全部步骤到当前步骤当前选择的最小总代价（或者最大价值）以及当前代价的状况下前继步骤的选择。依次计算完全部步骤后，经过回溯的方法找到最优选择路径。符合这个模型的均可以用viterbi算法解决。

用如下例子加以说明：
1.题目背景：
从前有个村儿，村里的人的身体状况只有两种可能：健康或者发烧。假设这个村儿的人没有体温计或者百度这种神奇东西，他惟一判断他身体状况的途径就是到村头个人偶像金正月的小诊所询问。月儿经过询问村民的感受，判断她的病情，再假设村民只会回答正常、头晕或冷。有一天村里奥巴驴就去月儿那去询问了。第一天她告诉月儿她感受正常。次日她告诉月儿感受有点冷。第三天她告诉月儿感受有点头晕。那么问题来了，月儿如何根据阿驴的描述的状况，推断出这三天中阿驴的一个身体状态呢?为此月儿上百度搜 google ，一番狂搜，发现维特比算法正好能解决这个问题。月儿乐了。

2.已知状况：
隐含的身体状态 = { 健康 , 发烧 }
可观察的感受状态 = { 正常 , 冷 , 头晕 }
月儿预判的阿驴身体状态的几率分布 = { 健康：0.6 , 发烧： 0.4 }这就是初始状态序列。
月儿认为的阿驴身体健康状态的转换几率分布 = {健康->健康： 0.7 ,健康->发烧： 0.3 ,发烧->健康：0.4 ,发烧->发烧： 0.6}
这样就能够列出相应的状态转移矩阵。
月儿认为的在相应健康情况条件下，阿驴的感受的几率分布 = {健康，正常：0.5 ，冷 ：0.4 ，头晕： 0.1 ；发烧，正常：0.1 ，冷 ：0.3 ，头晕： 0.6 }这样就能够列出相应的观测矩阵。
下面就是解决问题了。阿驴连续三天的身体感受依次是： 正常、冷、头晕 。

3.题目：
已知如上，求：阿驴这三天的身体健康状态变化的过程是怎么样的？即已知观测序列和HMM模型的状况下，求状态序列。

4.过程：

最后一天的状态几率分布即为最优路径的几率，即P(最优)=0.01512，这样咱们能够获得最优路径的终点，是发烧。这样，咱们的状态序列逆推出来了。即为：健康，健康，发烧。

简要的画个图：

代码实现：

import math
# 状态的样本空间
states = ('健康', '发热')
# 观测的样本空间
observations = ('正常', '发冷', '发晕')
# 起始个状态几率
start_probability = {'健康': 0.6, '发热': 0.4}
# 状态转移几率
transition_probability = {
  '健康': {'健康': 0.7, '发热': 0.3},
  '发热': {'健康': 0.4, '发热': 0.6},
}
# 状态->观测的发散几率
emission_probability = {
  '健康': {'正常': 0.5, '发冷': 0.4, '发晕': 0.1},
  '发热': {'正常': 0.1, '发冷': 0.3, '发晕': 0.6},
}
# 计算以E为底的幂
def E(x):
  #return math.pow(math.e,x)
  return x

def display_result(observations,result_m):
  """
  较为友好清晰的显示结果
  :param result_m:
  :return:
  """
  # 从结果中找出最佳路径
  infered_states = []
  final = len(result_m) - 1
  (p, pre_state), final_state = max(zip(result_m[final].values(), result_m[final].keys()))
  infered_states.insert(0, final_state)
  infered_states.insert(0, pre_state)
  for t in range(final - 1, 0, -1):
    _, pre_state = result_m[t][pre_state]
    infered_states.insert(0, pre_state)
  print(format("Viterbi Result", "=^59s"))
  head = format("obs", " ^10s")
  head += format("Infered state", " ^18s")
  for s in states:
    head += format(s, " ^15s")
  print(head)
  print(format("", "-^59s"))

  for obs,result,infered_state in zip(observations,result_m,infered_states):
    item = format(obs," ^10s")
    item += format(infered_state," ^18s")
    for s in states:
      item += format(result[s][0]," >12.8f")
      if infered_state == s:
        item += "(*)"
      else:
        item +="   "

    print(item)
  print(format("", "=^59s"))

def viterbi(obs, states, start_p, trans_p, emit_p):

  result_m = [{}] # 存放结果,每个元素是一个字典，每个字典的形式是 state:(p,pre_state)
                  # 其中state,p分别是当前状态下的几率值，pre_state表示该值由上一次的那个状态计算获得
  for s in states:  # 对于每个状态
    result_m[0][s] = (E(start_p[s]*emit_p[s][obs[0]]),None) # 把第一个观测节点对应的各状态值计算出来

  for t in range(1,len(obs)):
    result_m.append({})  # 准备t时刻的结果存放字典，形式同上

    for s in states: # 对于每个t时刻状态s,获取t-1时刻每一个状态s0的p,结合由s0转化为s的转移几率和s状态至obs的发散几率
                     # 计算t时刻s状态的最大几率，并记录该几率的来源状态s0
                     # max()内部比较的是一个tuple:(p,s0),max比较tuple内的第一个元素值
      result_m[t][s] = max([(E(result_m[t-1][s0][0]*trans_p[s0][s]*emit_p[s][obs[t]]),s0) for s0 in states])
  return result_m    # 全部结果（包括最佳路径）都在这里，但直观的最佳路径还须要依此结果单独生成，在显示的时候生成

def example():
  """
  一个能够交互的示例
  """
  result_m = viterbi(observations,
                 states,
                 start_probability,
                 transition_probability,
                 emission_probability)
  display_result(observations,result_m)
  while True:
    user_obs = input("轮到你来输入观测,计算机来推断可能状态\n"
                "使用 'N' 表明'正常', 'C' 表明'发冷','D'表明'发晕'\n"
                "您输入：('q'将退出):")

    if len(user_obs) ==0 or 'q' in user_obs or 'Q' in user_obs:
      break
    else:
      obs = []
      for o in user_obs:
        if o.upper() == 'N':
          obs.append("正常")
        elif o.upper() == 'C':
          obs.append("发冷")
        elif o.upper() == 'D':
          obs.append("发晕")
        else:
          pass
      result_m = viterbi(obs,
                       states,
                       start_probability,
                       transition_probability,
                       emission_probability)
      display_result(obs,result_m)

if __name__ == "__main__":
  example()

运行结果：