簡單理解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()

運行結果：