维特比算法的Python实现

Question

我在做一个Python项目，我想使用维特比算法。有谁知道维特比算法的完整Python实现吗？维基百科的正确性似乎在谈话页面上有问题。有人有指针吗？

Answer 1

嗯我可以发布我的。尽管如此，请您告诉我是否需要澄清。我最近写了这个专门用于词性标注的部分。

class Trellis: 
    trell = [] 
    def __init__(self, hmm, words): 
     self.trell = [] 
     temp = {} 
     for label in hmm.labels: 
      temp[label] = [0,None] 
     for word in words: 
      self.trell.append([word,copy.deepcopy(temp)]) 
     self.fill_in(hmm) 

    def fill_in(self,hmm): 
     for i in range(len(self.trell)): 
      for token in self.trell[i][1]: 
       word = self.trell[i][0] 
       if i == 0: 
        self.trell[i][1][token][0] = hmm.e(token,word) 
       else: 
        max = None 
        guess = None 
        c = None 
        for k in self.trell[i-1][1]: 
         c = self.trell[i-1][1][k][0] + hmm.t(k,token) 
         if max == None or c > max: 
          max = c 
          guess = k 
        max += hmm.e(token,word) 
        self.trell[i][1][token][0] = max 
        self.trell[i][1][token][1] = guess 

    def return_max(self): 
     tokens = [] 
     token = None 
     for i in range(len(self.trell)-1,-1,-1): 
      if token == None: 
       max = None 
       guess = None 
       for k in self.trell[i][1]: 
        if max == None or self.trell[i][1][k][0] > max: 
         max = self.trell[i][1][k][0] 
         token = self.trell[i][1][k][1] 
         guess = k 
       tokens.append(guess) 
      else: 
       tokens.append(token) 
       token = self.trell[i][1][token][1] 
     tokens.reverse() 
     return tokens 

Answer 2

NLTK has one。 http://nltk.github.com/_modules/nltk/parse/viterbi.html#ViterbiParser

Answer 3

我的Artificial Intelligence: A Modern Approach例子存储库中找到以下code。是这样的东西你在找什么？

def viterbi_segment(text, P): 
    """Find the best segmentation of the string of characters, given the 
    UnigramTextModel P.""" 
    # best[i] = best probability for text[0:i] 
    # words[i] = best word ending at position i 
    n = len(text) 
    words = [''] + list(text) 
    best = [1.0] + [0.0] * n 
    ## Fill in the vectors best, words via dynamic programming 
    for i in range(n+1): 
     for j in range(0, i): 
      w = text[j:i] 
      if P[w] * best[i - len(w)] >= best[i]: 
       best[i] = P[w] * best[i - len(w)] 
       words[i] = w 
    ## Now recover the sequence of best words 
    sequence = []; i = len(words)-1 
    while i > 0: 
     sequence[0:0] = [words[i]] 
     i = i - len(words[i]) 
    ## Return sequence of best words and overall probability 
    return sequence, best[-1] 

Answer 4

我刚纠正了Viterbi in Wikipedia的伪执行。从最初的（不正确的）版本开始，我花了一段时间才弄清楚自己出错的地方，但是我终于成功了，这要归功于Kevin Murphy在MatLab HMM toolbox中实现viterbi_path.m。

在具有变量的HMM对象的上下文，如下所示：

hmm = HMM() 
hmm.priors = np.array([0.5, 0.5]) # pi = prior probs 
hmm.transition = np.array([[0.75, 0.25], # A = transition probs./2 states 
          [0.32, 0.68]]) 
hmm.emission = np.array([[0.8, 0.1, 0.1], # B = emission (observation) probs./3 obs modes 
         [0.1, 0.2, 0.7]]) 

Python的函数来运行维特比（最佳路径）的算法是如下：

def viterbi (self,observations): 
    """Return the best path, given an HMM model and a sequence of observations""" 
    # A - initialise stuff 
    nSamples = len(observations[0]) 
    nStates = self.transition.shape[0] # number of states 
    c = np.zeros(nSamples) #scale factors (necessary to prevent underflow) 
    viterbi = np.zeros((nStates,nSamples)) # initialise viterbi table 
    psi = np.zeros((nStates,nSamples)) # initialise the best path table 
    best_path = np.zeros(nSamples); # this will be your output 

    # B- appoint initial values for viterbi and best path (bp) tables - Eq (32a-32b) 
    viterbi[:,0] = self.priors.T * self.emission[:,observations(0)] 
    c[0] = 1.0/np.sum(viterbi[:,0]) 
    viterbi[:,0] = c[0] * viterbi[:,0] # apply the scaling factor 
    psi[0] = 0; 

    # C- Do the iterations for viterbi and psi for time>0 until T 
    for t in range(1,nSamples): # loop through time 
     for s in range (0,nStates): # loop through the states @(t-1) 
      trans_p = viterbi[:,t-1] * self.transition[:,s] 
      psi[s,t], viterbi[s,t] = max(enumerate(trans_p), key=operator.itemgetter(1)) 
      viterbi[s,t] = viterbi[s,t]*self.emission[s,observations(t)] 

     c[t] = 1.0/np.sum(viterbi[:,t]) # scaling factor 
     viterbi[:,t] = c[t] * viterbi[:,t] 

    # D - Back-tracking 
    best_path[nSamples-1] = viterbi[:,nSamples-1].argmax() # last state 
    for t in range(nSamples-1,0,-1): # states of (last-1)th to 0th time step 
     best_path[t-1] = psi[best_path[t],t] 

    return best_path 

Answer 5

这是一个老问题，但没有其他答案是相当我需要的是因为我的应用程序没有特定的观察状态。

接过@Rhubarb之后，我也重新实现了Kevin Murphey的Matlab implementation（请参阅viterbi_path.m），但我已经更接近原来的了。我还包含一个简单的测试用例。

import numpy as np 


def viterbi_path(prior, transmat, obslik, scaled=True, ret_loglik=False): 
    '''Finds the most-probable (Viterbi) path through the HMM state trellis 
    Notation: 
     Z[t] := Observation at time t 
     Q[t] := Hidden state at time t 
    Inputs: 
     prior: np.array(num_hid) 
      prior[i] := Pr(Q[0] == i) 
     transmat: np.ndarray((num_hid,num_hid)) 
      transmat[i,j] := Pr(Q[t+1] == j | Q[t] == i) 
     obslik: np.ndarray((num_hid,num_obs)) 
      obslik[i,t] := Pr(Z[t] | Q[t] == i) 
     scaled: bool 
      whether or not to normalize the probability trellis along the way 
      doing so prevents underflow by repeated multiplications of probabilities 
     ret_loglik: bool 
      whether or not to return the log-likelihood of the best path 
    Outputs: 
     path: np.array(num_obs) 
      path[t] := Q[t] 
    ''' 
    num_hid = obslik.shape[0] # number of hidden states 
    num_obs = obslik.shape[1] # number of observations (not observation *states*) 

    # trellis_prob[i,t] := Pr((best sequence of length t-1 goes to state i), Z[1:(t+1)]) 
    trellis_prob = np.zeros((num_hid,num_obs)) 
    # trellis_state[i,t] := best predecessor state given that we ended up in state i at t 
    trellis_state = np.zeros((num_hid,num_obs), dtype=int) # int because its elements will be used as indicies 
    path = np.zeros(num_obs, dtype=int) # int because its elements will be used as indicies 

    trellis_prob[:,0] = prior * obslik[:,0] # element-wise mult 
    if scaled: 
     scale = np.ones(num_obs) # only instantiated if necessary to save memory 
     scale[0] = 1.0/np.sum(trellis_prob[:,0]) 
     trellis_prob[:,0] *= scale[0] 

    trellis_state[:,0] = 0 # arbitrary value since t == 0 has no predecessor 
    for t in xrange(1, num_obs): 
     for j in xrange(num_hid): 
      trans_probs = trellis_prob[:,t-1] * transmat[:,j] # element-wise mult 
      trellis_state[j,t] = trans_probs.argmax() 
      trellis_prob[j,t] = trans_probs[trellis_state[j,t]] # max of trans_probs 
      trellis_prob[j,t] *= obslik[j,t] 
     if scaled: 
      scale[t] = 1.0/np.sum(trellis_prob[:,t]) 
      trellis_prob[:,t] *= scale[t] 

    path[-1] = trellis_prob[:,-1].argmax() 
    for t in range(num_obs-2, -1, -1): 
     path[t] = trellis_state[(path[t+1]), t+1] 

    if not ret_loglik: 
     return path 
    else: 
     if scaled: 
      loglik = -np.sum(np.log(scale)) 
     else: 
      p = trellis_prob[path[-1],-1] 
      loglik = np.log(p) 
     return path, loglik 


if __name__=='__main__': 
    # Assume there are 3 observation states, 2 hidden states, and 5 observations 
    priors = np.array([0.5, 0.5]) 
    transmat = np.array([ 
     [0.75, 0.25], 
     [0.32, 0.68]]) 
    emmat = np.array([ 
     [0.8, 0.1, 0.1], 
     [0.1, 0.2, 0.7]]) 
    observations = np.array([0, 1, 2, 1, 0], dtype=int) 
    obslik = np.array([emmat[:,z] for z in observations]).T 
    print viterbi_path(priors, transmat, obslik)        #=> [0 1 1 1 0] 
    print viterbi_path(priors, transmat, obslik, scaled=False)     #=> [0 1 1 1 0] 
    print viterbi_path(priors, transmat, obslik, ret_loglik=True)    #=> (array([0, 1, 1, 1, 0]), -7.776472586614755) 
    print viterbi_path(priors, transmat, obslik, scaled=False, ret_loglik=True) #=> (array([0, 1, 1, 1, 0]), -8.0120386579275227) 

注意，此实现不使用直接发射概率，但使用可变obslik。通常，emissions[i,j] := Pr(observed_state == j | hidden_state == i)针对特定的观察状态i，使得emissions.shape == (num_hidden_states, num_obs_states)。

但是，给定一个序列observations[t] := observation at time t，所有维特比算法要求的是对每个隐藏状态进行观察的可能性。因此，obslik[i,t] := Pr(observations[t] | hidden_state == i)。观察状态的实际值不是必需的。

Answer 6

我已经修改了@大黄的答案，其中的边际概率已知（例如通过计算Forward Forwardward算法）。

def viterbi (transition_probabilities, conditional_probabilities): 
    # Initialise everything 
    num_samples = conditional_probabilities.shape[1] 
    num_states = transition_probabilities.shape[0] # number of states 

    c = np.zeros(num_samples) #scale factors (necessary to prevent underflow) 
    viterbi = np.zeros((num_states,num_samples)) # initialise viterbi table 
    best_path_table = np.zeros((num_states,num_samples)) # initialise the best path table 
    best_path = np.zeros(num_samples).astype(np.int32) # this will be your output 

    # B- appoint initial values for viterbi and best path (bp) tables - Eq (32a-32b) 
    viterbi[:,0] = conditional_probabilities[:,0] 
    c[0] = 1.0/np.sum(viterbi[:,0]) 
    viterbi[:,0] = c[0] * viterbi[:,0] # apply the scaling factor 

    # C- Do the iterations for viterbi and psi for time>0 until T 
    for t in range(1, num_samples): # loop through time 
     for s in range (0,num_states): # loop through the states @(t-1) 
      trans_p = viterbi[:, t-1] * transition_probabilities[:,s] # transition probs of each state transitioning 
      best_path_table[s,t], viterbi[s,t] = max(enumerate(trans_p), key=operator.itemgetter(1)) 
      viterbi[s,t] = viterbi[s,t] * conditional_probabilities[s][t] 

     c[t] = 1.0/np.sum(viterbi[:,t]) # scaling factor 
     viterbi[:,t] = c[t] * viterbi[:,t] 

    ## D - Back-tracking 
    best_path[num_samples-1] = viterbi[:,num_samples-1].argmax() # last state 
    for t in range(num_samples-1,0,-1): # states of (last-1)th to 0th time step 
     best_path[t-1] = best_path_table[best_path[t],t] 
    return best_path