机器学习之经典算法：隐马尔可夫模型（HMM）原理、手动计算与Python/Java双代码实战

import numpy as np

class HMM:
    def __init__(self, states, observations, pi, A, B):
        self.states = states
        self.observations = observations
        self.pi = np.array(pi)
        self.A = np.array(A)
        self.B = np.array(B)
        self.state_to_idx = {s: i for i, s in enumerate(states)}
        self.obs_to_idx = {o: i for i, o in enumerate(observations)}
    
    def viterbi(self, obs_seq):
        T = len(obs_seq)
        N = len(self.states)
        
        # 初始化
        delta = np.zeros((T, N))
        psi = np.zeros((T, N), dtype=int)
        
        o0 = self.obs_to_idx[obs_seq[0]]
        delta[0] = self.pi * self.B[:, o0]
        
        # 递推
        for t in range(1, T):
            ot = self.obs_to_idx[obs_seq[t]]
            for j in range(N):
                probs = delta[t-1] * self.A[:, j]
                psi[t, j] = np.argmax(probs)
                delta[t, j] = np.max(probs) * self.B[j, ot]
        
        # 终止
        path = [np.argmax(delta[T-1])]
        
        # 回溯
        for t in range(T-1, 0, -1):
            path.insert(0, psi[t, path[0]])
        
        return [self.states[i] for i in path], np.max(delta[T-1])
    
    def forward(self, obs_seq):
        """计算 P(O|λ)"""
        T = len(obs_seq)
        N = len(self.states)
        alpha = np.zeros((T, N))
        
        # 初始化
        o0 = self.obs_to_idx[obs_seq[0]]
        alpha[0] = self.pi * self.B[:, o0]
        
        # 递推
        for t in range(1, T):
            ot = self.obs_to_idx[obs_seq[t]]
            for j in range(N):
                alpha[t, j] = np.sum(alpha[t-1] * self.A[:, j]) * self.B[j, ot]
        
        return np.sum(alpha[T-1])

# 模型参数
states = ["Sunny", "Rainy"]
observations = ["Walk", "Shop", "Clean"]

pi = [0.6, 0.4]
A = [[0.7, 0.3],
     [0.4, 0.6]]
B = [[0.6, 0.3, 0.1],
     [0.1, 0.4, 0.5]]

hmm = HMM(states, observations, pi, A, B)

# 解码
obs_seq = ["Walk", "Shop", "Clean"]
best_path, prob = hmm.viterbi(obs_seq)
print("最优状态序列:", best_path)  # ['Sunny', 'Rainy', 'Rainy']

# 评估
p_obs = hmm.forward(obs_seq)
print("P(O|λ) =", p_obs)  # ≈ 0.0234

import java.util.*;

public class HMM {
    private String[] states;
    private String[] observations;
    private double[] pi;
    private double[][] A;
    private double[][] B;
    private Map<String, Integer> stateToIdx;
    private Map<String, Integer> obsToIdx;

    public HMM(String[] states, String[] observations, double[] pi, double[][] A, double[][] B) {
        this.states = states;
        this.observations = observations;
        this.pi = pi;
        this.A = A;
        this.B = B;
        this.stateToIdx = new HashMap<>();
        this.obsToIdx = new HashMap<>();
        for (int i = 0; i < states.length; i++) stateToIdx.put(states[i], i);
        for (int i = 0; i < observations.length; i++) obsToIdx.put(observations[i], i);
    }

    public Result viterbi(String[] obsSeq) {
        int T = obsSeq.length;
        int N = states.length;
        double[][] delta = new double[T][N];
        int[][] psi = new int[T][N];

        // 初始化
        int o0 = obsToIdx.get(obsSeq[0]);
        for (int i = 0; i < N; i++) {
            delta[0][i] = pi[i] * B[i][o0];
        }

        // 递推
        for (int t = 1; t < T; t++) {
            int ot = obsToIdx.get(obsSeq[t]);
            for (int j = 0; j < N; j++) {
                double maxProb = -1;
                int bestPrev = 0;
                for (int i = 0; i < N; i++) {
                    double prob = delta[t-1][i] * A[i][j];
                    if (prob > maxProb) {
                        maxProb = prob;
                        bestPrev = i;
                    }
                }
                psi[t][j] = bestPrev;
                delta[t][j] = maxProb * B[j][ot];
            }
        }

        // 终止
        double maxProb = -1;
        int lastState = 0;
        for (int i = 0; i < N; i++) {
            if (delta[T-1][i] > maxProb) {
                maxProb = delta[T-1][i];
                lastState = i;
            }
        }

        // 回溯
        int[] pathIdx = new int[T];
        pathIdx[T-1] = lastState;
        for (int t = T-2; t >= 0; t--) {
            pathIdx[t] = psi[t+1][pathIdx[t+1]];
        }

        String[] path = new String[T];
        for (int i = 0; i < T; i++) {
            path[i] = states[pathIdx[i]];
        }

        return new Result(path, maxProb);
    }

    public static class Result {
        public String[] path;
        public double probability;
        public Result(String[] path, double prob) {
            this.path = path;
            this.probability = prob;
        }
    }

    // 测试
    public static void main(String[] args) {
        String[] states = {"Sunny", "Rainy"};
        String[] observations = {"Walk", "Shop", "Clean"};
        double[] pi = {0.6, 0.4};
        double[][] A = {{0.7, 0.3}, {0.4, 0.6}};
        double[][] B = {{0.6, 0.3, 0.1}, {0.1, 0.4, 0.5}};

        HMM hmm = new HMM(states, observations, pi, A, B);
        String[] obsSeq = {"Walk", "Shop", "Clean"};
        Result res = hmm.viterbi(obsSeq);

        System.out.println("最优状态序列: " + Arrays.toString(res.path));
        // 输出: [Sunny, Rainy, Rainy]
        System.out.printf("最大概率: %.5f\n", res.probability); // ≈0.01296
    }
}