feat: question 2 finishes

2catycm · 2catycm · commit 3440d034639c · 2024-12-08T09:00:16.000+08:00
diff --git a/notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/00hidden_markov_model.ipynb b/notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/00hidden_markov_model.ipynb
@@ -111,22 +111,31 @@
     "我们的实验代码都是通过 jupyter notebook 进行编写，并通过 nbdev 进行自动化构建pypi库。\n",
     "复现我们实验代码时，建议使用支持jupyter notebook的IDE（如VSCode）运行我们的 `ipynb文件`。\n",
     "\n",
+    "由于本次作业较为简单，在安装了对应依赖库后，就可以直接运行我导出的 `viterbi.py`文件。但是仍然建议使用ipynb来运行。\n",
+    "由于进行了一些实验探索，我们引入了\n",
+    "```bash\n",
+    "pip install fastcore\n",
+    "pip install jax\n",
+    "pip install flax\n",
+    "```\n",
+    "\n",
     "::: {.callout-note}\n",
+    "\n",
+    "我们的代码导出为了python模块形式，如果你需要在别的项目中调用我们的代码，\n",
     "可以通过以下命令安装我们实验的代码：\n",
     "\n",
     "```shell\n",
     "git clone https://github.com/Open-Book-Studio/THU-Coursework-Machine-Learning-for-Big-Data.git\n",
     "cd THU-Coursework-Machine-Learning-for-Big-Data\n",
     "pip install -e .\n",
     "```\n",
-    "我们的代码导出为了python模块形式，如果你需要在别的项目中调用我们的代码，可以通过以下命令导入：\n",
+    "然后通过以下命令导入：\n",
     "```python\n",
     "from thu_big_data_ml.digital_processing_of_speech_signals.hidden_markov_model import *\n",
     "```\n",
     ":::\n",
     "\n",
-    "这次作业中，我开发的我们课题组的基础依赖库[ScholarlyInfrastructure]\\(https://github.com/THU-CVML/ScholarlyInfrastructure)也相应地进行了代码更新。这个库对标`fastcore`库，对AI科研经常会用到的一些基础性地、和Python语言的表达力有关的代码进行了整理，比如PyTorch模型检查、清晰的日志、实验参数管理、异常处理、argmax自动函数优化等。接下来我们也会用到这个项目中的一些代码来完成本次作业。\n",
-    "\n",
+    "这次作业中，我开发的我们课题组的基础依赖库[ScholarlyInfrastructure]\\(https://github.com/THU-CVML/ScholarlyInfrastructure)也相应地进行了代码更新。这个库对标`fastcore`库，对AI科研经常会用到的一些基础性地、和Python语言的表达力有关的代码进行了整理，比如PyTorch模型检查、清晰的日志、实验参数管理、异常处理、argmax自动函数优化等。具体而言，本次作业中我们更新了 `scholarly_infrastructure.nbscholar` 模块，加强了代码导出同步功能。\n",
     "::: {.callout-tip}\n",
     "\n",
     "通过以下命令安装 ScholarlyInfrastructure\n",
@@ -1428,11 +1437,14 @@
     "    model = HiddenMarkovModel()\n",
     "    observations = jnp.array([0, 1, 0, 1, 1])  # [THU, PKU, THU, PKU, PKU]\n",
     "    fwd, p = model.compute_likelihood_by_forward(observations)\n",
-    "    print(p, fwd)\n",
+    "    print(p)\n",
+    "    print(fwd)\n",
     "    bwd, p = model.compute_likelihood_by_backward(observations)\n",
-    "    print(p, bwd)\n",
+    "    print(p)\n",
+    "    print(bwd)\n",
     "    prob, path = model.decode_states_by_viterbi(observations)\n",
-    "    print(prob, path)"
+    "    print(prob)\n",
+    "    print(path)"
    ]
   },
   {
diff --git a/notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/viterbi.py b/notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/viterbi.py
@@ -0,0 +1,171 @@
+# AUTOGENERATED! DO NOT EDIT! File to edit: ../../notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/00hidden_markov_model.ipynb.
+
+# %% auto 0
+__all__ = ['HiddenMarkovModel']
+
+# %% ../../notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/00hidden_markov_model.ipynb 36
+from fastcore.all import patch
+
+# %% ../../notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/00hidden_markov_model.ipynb 38
+import jax.numpy as jnp
+from flax import nnx # 导入 nnx 库，里面包含了一些常用的网络层
+from fastcore.all import store_attr # 导入 fastcore 基础库的 store_attr 函数，用来方便地存储类的属性，这样Python面向对象写起来不那么冗长。 请 pip install fastcore。
+
+# %% ../../notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/00hidden_markov_model.ipynb 39
+class HiddenMarkovModel(nnx.Module):
+    """Hidden Markov Model
+
+    HMM with 3 states and 2 observation categories.
+
+    Attributes:
+        ob_category (list, with length 2): observation categories
+        total_states (int): number of states, default=3
+        pi (array, with shape (3,)): initial state probability
+        A (array, with shape (3, 3)): transition probability. A.sum(axis=1) must be all ones.
+                                      A[i, j] means transition prob from state i to state j.
+                                      A.T[i, j] means transition prob from state j to state i.
+        B (array, with shape (3, 2)): emitting probability, B.sum(axis=1) must be all ones.
+                                      B[i, k] means emitting prob from state i to observation k.
+
+    """
+
+    def __init__(self):
+        self.ob_category = ['THU', 'PKU']  # 0: THU, 1: PKU
+        self.total_states = 3
+        self.pi = nnx.Param(jnp.array([0.2, 0.4, 0.4]))
+        self.A = nnx.Param(jnp.array([[0.1, 0.6, 0.3],
+                           [0.3, 0.5, 0.2],
+                           [0.7, 0.2, 0.1]]))
+        self.B = nnx.Param(jnp.array([[0.5, 0.5],
+                           [0.4, 0.6],
+                           [0.7, 0.3]]))
+
+# %% ../../notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/00hidden_markov_model.ipynb 45
+@patch
+def compute_likelihood_by_forward(self: HiddenMarkovModel, ob):
+    """HMM Forward Algorithm.
+
+    Args:
+        ob (array, with shape(T,)): (o1, o2, ..., oT), observations
+
+    Returns:
+        fwd (array, with shape(T, 3)): fwd[t, s] means full-path forward probability torwards state s at
+                                        timestep t given the observation ob[0:t+1].
+                                        给定观察ob[0:t+1]情况下t时刻到达状态s的所有可能路径的概率和
+        prob: the probability of HMM model generating observations.
+
+    """
+    T = ob.shape[0]
+    fwd = jnp.zeros((T, self.total_states))
+
+    # Begin Assignment
+
+    # 初始化 t=0 时刻的前向概率
+    # fwd[0, :] = self.pi * self.B[:, ob[0]] # jax 不支持 in place 复制
+    fwd = fwd.at[0, :].set(self.pi * self.B[:, ob[0]])
+
+    # 根据前向概率的递推公式计算 t=1 到 T 时刻的前向概率
+    for t in range(1, T):
+        for j in range(self.total_states):
+            fwd = fwd.at[t, j].set(self.B[j, ob[t]] * jnp.dot(fwd[t - 1, :], self.A[:, j]))
+            
+    # End Assignment
+
+    prob = fwd[-1, :].sum()
+
+    return fwd, prob
+
+# %% ../../notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/00hidden_markov_model.ipynb 51
+@patch
+def compute_likelihood_by_backward(self:HiddenMarkovModel, ob):
+    """HMM Backward Algorithm.
+
+    Args:
+        ob (array, with shape(T,)): (o1, o2, ..., oT), observations
+
+    Returns:
+        bwd (array, with shape(T, 3)): bwd[t, s] means full-path backward probability torwards state s at
+                                        timestep t given the observation ob[t+1::]
+                                        给定观察ob[t+1::]情况下t时刻到达状态s的所有可能路径的概率和
+        prob: the probability of HMM model generating observations.
+
+    """
+    T = ob.shape[0]
+    bwd = jnp.zeros((T, self.total_states))
+
+    # Begin Assignment
+
+    # 初始化 t == T-1 时刻到达各个状态的概率
+    bwd = bwd.at[T - 1, :].set(1.0)
+
+    # Induction step
+    for t in reversed(range(T - 1)):
+        for i in range(self.total_states):
+            bwd = bwd.at[t, i].set(jnp.dot(bwd[t + 1, :] * self.B[:, ob[t + 1]], self.A[i, :]))
+
+    # End Assignment
+
+    prob = (bwd[0, :] * self.B[:, ob[0]] * self.pi).sum()
+
+    return bwd, prob
+
+# %% ../../notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/00hidden_markov_model.ipynb 56
+@patch
+def decode_states_by_viterbi(self:HiddenMarkovModel, ob):
+    """Viterbi Decoding Algorithm.
+
+    Args:
+        ob (array, with shape(T,)): (o1, o2, ..., oT), observations
+
+    Variables:
+        delta (array, with shape(T, 3)): delta[t, s] means max probability torwards state s at
+                                            timestep t given the observation ob[0:t+1]
+                                            给定观察ob[0:t+1]情况下t时刻到达状态s的概率最大的路径的概率
+        phi (array, with shape(T, 3)): phi[t, s] means prior state s' for delta[t, s]
+                                        给定观察ob[0:t+1]情况下t时刻到达状态s的概率最大的路径的t-1时刻的状态s'
+
+    Returns:
+        best_prob: the probability of the best state sequence
+        best_path: the best state sequence
+
+    """
+    T = ob.shape[0]
+    delta = jnp.zeros((T, self.total_states))
+    #update np.int32
+    phi = jnp.zeros((T, self.total_states), jnp.int32)
+    best_prob, best_path = 0.0, jnp.zeros(T, dtype=jnp.int32)
+
+    # Begin Assignment
+
+    # 从初始状态开始
+    delta = delta.at[0, :].set(self.pi * self.B[:, ob[0]])
+
+    # 根据动态规划的公式来更新delta和phi
+    for t in range(1, T):
+        for j in range(self.total_states):
+            d, p = max((delta[t - 1, i] * self.A[i, j] * self.B[j, ob[t]], i) for i in range(self.total_states))
+            delta = delta.at[t, j].set(d)
+            phi = phi.at[t, j].set(p)
+
+    # End Assignment
+
+    best_path = best_path.at[T-1].set(delta[T-1, :].argmax(0))
+    best_prob = delta[T-1, best_path[T-1]]
+    for t in reversed(range(T-1)):
+        best_path = best_path.at[t].set(phi[t+1, best_path[t+1]])
+
+    return best_prob, best_path
+
+# %% ../../notebooks/coding_projects/digital_processing_of_speech_signals/P2_HMM/00hidden_markov_model.ipynb 60
+if __name__ == "__main__":
+    model = HiddenMarkovModel()
+    observations = jnp.array([0, 1, 0, 1, 1])  # [THU, PKU, THU, PKU, PKU]
+    fwd, p = model.compute_likelihood_by_forward(observations)
+    print(p)
+    print(fwd)
+    bwd, p = model.compute_likelihood_by_backward(observations)
+    print(p)
+    print(bwd)
+    prob, path = model.decode_states_by_viterbi(observations)
+    print(prob)
+    print(path)
diff --git a/thu_big_data_ml/digital_processing_of_speech_signals/hidden_markov_model.py b/thu_big_data_ml/digital_processing_of_speech_signals/hidden_markov_model.py
@@ -161,8 +161,11 @@ def decode_states_by_viterbi(self:HiddenMarkovModel, ob):
     model = HiddenMarkovModel()
     observations = jnp.array([0, 1, 0, 1, 1])  # [THU, PKU, THU, PKU, PKU]
     fwd, p = model.compute_likelihood_by_forward(observations)
-    print(p, fwd)
+    print(p)
+    print(fwd)
     bwd, p = model.compute_likelihood_by_backward(observations)
-    print(p, bwd)
+    print(p)
+    print(bwd)
     prob, path = model.decode_states_by_viterbi(observations)
-    print(prob, path)
+    print(prob)
+    print(path)
