hmm/hmm_tutorial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# python 实现 隐马尔可夫模型（HMM）\n",
    "\n",
    "本例子主要受 Michael Collins 教授的 Tagging Problems, and Hidden Markov Models 启发而编写，为了帮助大家理解，我在我的博客、公众号上发表了文章[一文读懂NLP中的HMM(公众号)](https://mp.weixin.qq.com/s?__biz=MzIwNDM1NjUzMA==&mid=2247483662&idx=1&sn=cf463dde9af1844a3fd1e3e4fec26f5c&chksm=96c02fd3a1b7a6c5cfabe53efbff54af33cd2f61d13064645fbff92ce1b024d82acb2375d9b0#rd)，欢迎大家阅读。当然强烈推荐Michael Collins 教授的 [Tagging Problems, and Hidden Markov Models](http://www.cs.columbia.edu/~mcollins/hmms-spring2013.pdf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 目录\n",
    "\n",
    "1. [项目结构](#项目结构)\n",
    "2. [环境要求](#环境要求)\n",
    "3. [代码分析](#代码分析)\n",
    "4. [结果分析](#结果分析)\n",
    "5. [项目后续](#项目后续)\n",
    "6. [联系作者](#联系作者)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 项目结构\n",
    "\n",
    "| - src\n",
    "\n",
    "    | - const.py      常量定义文件\n",
    "\n",
    "    | - corpus        语料库\n",
    "\n",
    "    | - dataset.py    加载语料\n",
    "\n",
    "    | - hmm.py        bigram hmm, trigram hmm, viterbi\n",
    "\n",
    "    | - main.py       例子程序\n",
    "\n",
    "    | - processing.py 字典的生成等处理方法"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 环境要求\n",
    "\n",
    "    python3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 代码分析"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### const.py\n",
    "\n",
    "在这里定义了三个常量"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# 未登录词\n",
    "UNK = None\n",
    "# 句子开始标记，代表句子的开头\n",
    "START_TOKEN = '<s>'\n",
    "# 句子结束标记，代表句子的结尾\n",
    "END_TOKEN = '</s>'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### processing.py\n",
    "\n",
    "字典的构建"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# !/usr/bin/env python3\n",
    "# -*- coding: utf-8 -*-\n",
    "\n",
    "'''\n",
    "@description: 句子的处理，字典的构建\n",
    "@author: Sean QQ: 929325776\n",
    "'''\n",
    "\n",
    "import const\n",
    "\n",
    "#加入起始标记\n",
    "def build_tags(tags):\n",
    "\tout = []\n",
    "\tfor sentence in tags:\n",
    "\t\titems = [x.lower() for x in sentence]\n",
    "\t\titems.insert(0, const.START_TOKEN)\n",
    "\t\titems.append(const.END_TOKEN)\n",
    "\t\tout.append(items)\n",
    "\treturn out\n",
    "\n",
    "# 构建ungram词频词典\n",
    "def build_undict(tags):\n",
    "\tundict = {}\n",
    "\tfor items in tags:\n",
    "\t\tfor word in items:\n",
    "\t\t\tif word == const.START_TOKEN or word == const.END_TOKEN:\n",
    "\t\t\t\tcontinue\n",
    "\t\t\tif word not in undict:\n",
    "\t\t\t\tundict[word] = 1\n",
    "\t\t\telse:\n",
    "\t\t\t\tundict[word] += 1\n",
    "\treturn undict\n",
    "\n",
    "\n",
    "# 构建bigram词频词典，其中以三元组(u, v)作为词典的键\n",
    "def build_bidict(tags):\n",
    "\tbidict = {}\n",
    "\tfor items in tags: \n",
    "\t\tfor i in range(len(items)-1):\n",
    "\t\t\ttup = (items[i], items[i+1])\n",
    "\t\t\tif tup not in bidict:\n",
    "\t\t\t\tbidict[tup] = 1\n",
    "\t\t\telse:\n",
    "\t\t\t\tbidict[tup] += 1\n",
    "\treturn bidict\n",
    "\n",
    "# 构建trigram词频词典，其中以三元组(u, v, w)作为词典的键\n",
    "def build_tridict(tags):\n",
    "\ttridict = {}\n",
    "\tfor items in tags:\n",
    "\t\titems.insert(0, const.START_TOKEN)\n",
    "\t\tfor i in range(len(items) -2):\n",
    "\t\t\ttup = (items[i], items[i+1], items[i+2])\n",
    "\t\t\tif tup not in tridict:\n",
    "\t\t\t\ttridict[tup] = 1\n",
    "\t\t\telse:\n",
    "\t\t\t\ttridict[tup] += 1\n",
    "\treturn tridict\n",
    "\n",
    "# 构建(词,词性)词频字典，以及统计词频\n",
    "def build_count_dict(datas, tags):\n",
    "\ttagword_dict = {}\n",
    "\twordcount = {}\n",
    "\ttagcount = {}\n",
    "\tfor i, data in enumerate(datas):\n",
    "\t\ttag = tags[i][1:-1]\n",
    "\t\tfor idx, d in enumerate(data):\n",
    "\t\t\ttup = (tag[idx], d)\n",
    "\t\t\tif tup not in tagword_dict:\n",
    "\t\t\t\ttagword_dict[tup] = 1\n",
    "\t\t\telse:\n",
    "\t\t\t\ttagword_dict[tup] += 1\n",
    "\n",
    "\t\t\tif d not in wordcount:\n",
    "\t\t\t\twordcount[d] = 1\n",
    "\t\t\telse:\n",
    "\t\t\t\twordcount[d] += 1\n",
    "\t\t\tif tag[idx] not in tagcount:\n",
    "\t\t\t\ttagcount[tag[idx]] = 1\n",
    "\t\t\telse:\n",
    "\t\t\t\ttagcount[tag[idx]] += 1\n",
    "\treturn tagword_dict, wordcount, tagcount"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### hmm.py\n",
    "\n",
    "基于bigram, trigram实现了hmm, 支持viterbi解码"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# !/usr/bin/env python3\n",
    "# -*- coding: utf-8 -*-\n",
    "\n",
    "'''\n",
    "@description: bigram hmm, trigram hmm, _viterbi\n",
    "@author: Sean QQ: 929325776\n",
    "'''\n",
    "\n",
    "import math\n",
    "import const\n",
    "from processing import *\n",
    "\n",
    "'''bigram hmm'''\n",
    "class BiHMM(object):\n",
    "\tdef __init__(self, datas, tags):\n",
    "\t\tself.datas = datas\n",
    "\t\tself.tags = build_tags(tags)\n",
    "\t\tself.undict = build_undict(self.tags) \n",
    "\t\tself.bidict = build_bidict(self.tags)\n",
    "\t\tself.tagword, self.wordcount, self.tagcount = build_count_dict(datas, self.tags)\n",
    "\t\tself.postags = [tag for tag in self.undict]\n",
    "\n",
    "\tdef calc_e_prob(self, *args):\n",
    "\t\tif len(args) != 2:\n",
    "\t\t\traise ValueError('two tags is required')\n",
    "\n",
    "\t\tn = 0.0\n",
    "\t\tm = 0.0\n",
    "\t\tif args in self.tagword:\n",
    "\t\t\tn = self.tagword[args]\n",
    "\t\tif args[0] in self.undict:\n",
    "\t\t\tm = self.undict[args[0]]\n",
    "\t\treturn (n + 1) * 1.0 / (m + len(self.wordcount)*len(self.undict))\n",
    "\n",
    "\tdef calc_prob(self, *args):\n",
    "\t\tif len(args) != 2:\n",
    "\t\t\traise ValueError('two tags is required')\n",
    "\n",
    "\t\tn = 0.0\n",
    "\t\tm = 0.0\n",
    "\t\tif args in self.bidict:\n",
    "\t\t\tn = self.bidict[args]\n",
    "\t\tif args[0] in self.undict:\n",
    "\t\t\tm = self.undict[args[0]]\n",
    "\t\treturn (n + 1) * 1.0 / (m + len(self.postags)**2)\n",
    "\n",
    "\tdef calc_tags_prob(self, tags):\n",
    "\t\tprob = 0\n",
    "\t\tprev_tag = const.START_TOKEN\n",
    "\t\tfor tag in tags:\n",
    "\t\t\ttag_prob = self.calc_prob(prev_tag, tag)\n",
    "\t\t\tprob += tag_prob\n",
    "\t\t\tprev_tag = tag\n",
    "\t\treturn prob\n",
    "\n",
    "\tdef calc_tagword_proba(self, tag, word):\n",
    "\t\tprob = 0.0\n",
    "\t\ttagword = (tag, word)\n",
    "\t\tif tagword in self.tagword:\n",
    "\t\t\tprob = float(self.tagword[tagword]) / self.tagcount[tag]\n",
    "\t\treturn prob\n",
    "\n",
    "\t# @param vb _viterbi\n",
    "\tdef pred(self, sentence, vb=False):\n",
    "\t\tif vb:\n",
    "\t\t\t# _viterbi\n",
    "\t\t\treturn self._viterbi(sentence)\n",
    "\n",
    "\t\twordtag = []\n",
    "\t\tmax_prob = 0.0\n",
    "\t\tmax_tag = None\n",
    "\t\t#total_prob = None\n",
    "\t\tfor word in sentence:\n",
    "\t\t\tfor tag1 in self.postags:\n",
    "\t\t\t\tfor tag2 in self.postags:\n",
    "\t\t\t\t\tq = self.calc_tags_prob((tag1, tag2))\n",
    "\t\t\t\t\te = self.calc_tagword_proba(tag2, word)\n",
    "\t\t\t\t\tprob = q*e*1.0\n",
    "\t\t\t\t\tif prob >= max_prob:\n",
    "\t\t\t\t\t\tmax_prob = prob\n",
    "\t\t\t\t\t\tmax_tag = tag2\n",
    "\t\t\twordtag.append((word, max_tag))\n",
    "\t\t\t'''\n",
    "\t\t\tif total_prob == None:\n",
    "\t\t\t\ttotal_prob = max_prob\n",
    "\t\t\telse:\n",
    "\t\t\t\ttotal_prob *= max_prob \n",
    "\t\t\t'''\n",
    "\t\t\tmax_prob = 0.0\t\t\n",
    "\t\treturn wordtag\n",
    "\n",
    "\n",
    "\tdef _viterbi_decode(self, sentence, score, trace):\n",
    "\t\tresult = []\n",
    "\t\ttmp = -float('inf')\n",
    "\t\tres_x = 0\n",
    "\t\tfor idx, val in enumerate(self.postags):\n",
    "\t\t\tif tmp < score[idx][len(sentence)-1]:\n",
    "\t\t\t\ttmp = score[idx][len(sentence)-1]\n",
    "\t\t\t\tres_x = idx\n",
    "\t\tresult.append(res_x)\n",
    "\t\tfor idx in range(len(sentence)-1, 0, -1):\n",
    "\t\t\tresult.append(trace[result[-1]][idx])\n",
    "\t\tresult.reverse()\n",
    "\t\tresult_pos = []\n",
    "\t\tresult_pos = [self.postags[k] for k in result]\n",
    "\t\twordtag = list(zip(sentence, result_pos))\n",
    "\t\treturn wordtag\n",
    "\n",
    "\tdef _viterbi(self, sentence):\n",
    "\t\trow = len(self.postags)\n",
    "\t\tcol = len(sentence)\n",
    "\n",
    "\t\ttrace = [[-1 for i in range(col)] for i in range(row)]\n",
    "\t\tscore = [[-1 for i in range(col)] for i in range(row)]\n",
    "\n",
    "\t\tfor idx, val in enumerate(sentence):\n",
    "\t\t\tif idx == 0:\n",
    "\t\t\t\tfor idx_pos, val_pos in enumerate(self.postags):\n",
    "\t\t\t\t\tscore[idx_pos][idx] = self.calc_e_prob(val_pos, sentence[idx]) # emit\n",
    "\t\t\telse:\n",
    "\t\t\t\tfor idx_pos, val_pos in enumerate(self.postags):\n",
    "\t\t\t\t\ttmp = -float('inf')\n",
    "\t\t\t\t\ttrace_tmp = -1\n",
    "\t\t\t\t\tfor idx_pos2, val_pos2 in enumerate(self.postags):\n",
    "\t\t\t\t\t\tr = score[idx_pos2][idx-1]*self.calc_prob(val_pos2, val_pos)\n",
    "\t\t\t\t\t\tif r > tmp:\n",
    "\t\t\t\t\t\t\ttmp = r\n",
    "\t\t\t\t\t\t\ttrace_tmp = idx_pos2\n",
    "\t\t\t\t\t\ttrace[idx_pos][idx] = trace_tmp\n",
    "\t\t\t\t\t\tscore[idx_pos][idx] = tmp*self.calc_e_prob(val_pos, val)\n",
    "\t\treturn self._viterbi_decode(sentence, score, trace)\n",
    "\n",
    "class TriHMM(BiHMM):\n",
    "\tdef __init__(self, datas, tags):\n",
    "\t\tBiHMM.__init__(self, datas, tags)\n",
    "\t\tself.tridict = build_tridict(self.tags)\n",
    "\n",
    "\tdef calc_prob(self, *args):\n",
    "\t\tif len(args) != 3:\n",
    "\t\t\traise ValueError('three tags is required')\n",
    "\n",
    "\t\tn = 0.0\n",
    "\t\tm = 0.0\n",
    "\t\tbitup = (args[0], args[1])\n",
    "\t\tif args in self.tridict:\n",
    "\t\t\tn = self.tridict[args]\n",
    "\t\tif bitup in self.bidict:\n",
    "\t\t\tm = self.bidict[bitup]\n",
    "\t\treturn (n + 1) * 1.0 / (m + len(self.postags)**2)\n",
    "\n",
    "\n",
    "\t\tprob = 0\n",
    "\t\tif self.smooth != None:\n",
    "\t\t\tprob = self.smooth(args[0], args[1], args[2], tridict=self.tridict, bidict=self.bidict, undict=self.undict)\n",
    "\t\telse:\n",
    "\t\t\tbitup = (args[0], args[1])\t\t\t\t\n",
    "\t\t\tif args in self.tridict and bitup in self.bidict:\n",
    "\t\t\t\treturn float(self.tridict[args]) / self.bidict[bitup]\n",
    "\t\treturn prob\n",
    "\n",
    "\tdef calc_tags_prob(self, tags):\n",
    "\t\tprob = 0\n",
    "\t\tprev_stack = [const.START_TOKEN, const.START_TOKEN]\n",
    "\t\tfor tag in tags:\n",
    "\t\t\ttag_prob = self.calc_prob(prev_stack[0], prev_stack[1], tag)\n",
    "\t\t\tprob += tag_prob\n",
    "\t\t\tprev_stack[0] = prev_stack[1]\n",
    "\t\t\tprev_stack[1] = tag\n",
    "\t\treturn prob\n",
    "\n",
    "\t# @param vb _viterbi\n",
    "\tdef pred(self, sentence, vb=False):\n",
    "\t\tif vb:\n",
    "\t\t\treturn self._viterbi(sentence)\n",
    "\t\twordtag = []\n",
    "\t\tmax_prob = 0.0\n",
    "\t\tmax_tag = None\n",
    "\t\t#total_prob = None\n",
    "\t\tfor word in sentence:\n",
    "\t\t\tfor tag1 in self.postags:\n",
    "\t\t\t\tfor tag2 in self.postags:\n",
    "\t\t\t\t\tfor tag3 in self.postags:\n",
    "\t\t\t\t\t\tq = self.calc_tags_prob((tag1, tag2, tag3))\n",
    "\t\t\t\t\t\te = self.calc_tagword_proba(tag3, word)\n",
    "\t\t\t\t\t\tprob = q*e*1.0\n",
    "\t\t\t\t\t\tif prob >= max_prob:\n",
    "\t\t\t\t\t\t\tmax_prob = prob\n",
    "\t\t\t\t\t\t\tmax_tag = tag3\n",
    "\t\t\twordtag.append((word, max_tag))\n",
    "\t\t\t'''\n",
    "\t\t\tif total_prob == None:\n",
    "\t\t\t\ttotal_prob = max_prob\n",
    "\t\t\telse:\n",
    "\t\t\t\ttotal_prob *= max_prob \n",
    "\t\t\t'''\n",
    "\t\t\tmax_prob = 0.0\t\t\n",
    "\t\treturn wordtag\n",
    "\n",
    "\tdef _viterbi_decode(self, sentence, score, trace):\n",
    "\t\tresult = []\n",
    "\t\ttmp = -float('inf')\n",
    "\t\tres_x = 0\n",
    "\t\tres_y = 0\n",
    "\t\tfor idx, val in enumerate(self.postags):\n",
    "\t\t\tfor idx_pos2, val_pos2 in enumerate(self.postags):\n",
    "\t\t\t\tif tmp < score[idx_pos2][idx][len(sentence)-1]:\n",
    "\t\t\t\t\ttmp = score[idx_pos2][idx][len(sentence)-1]\n",
    "\t\t\t\t\tres_x = idx\n",
    "\t\t\t\t\tres_y = idx_pos2\n",
    "\t\tresult.extend([res_x, res_y])\n",
    "\t\tfor idx in range(len(sentence)-1, 0, -1):\n",
    "\t\t\tresult.append(trace[result[-2]][result[-1]][idx])\n",
    "\t\tresult.reverse()\n",
    "\t\tresult_pos = []\n",
    "\t\tresult_pos = [self.postags[k] for k in result]\n",
    "\t\twordtag = list(zip(sentence, result_pos))\n",
    "\t\treturn wordtag\n",
    "\n",
    "\tdef _viterbi(self, sentence):\n",
    "\t\trow = len(self.postags)\n",
    "\t\tcol = len(sentence)\n",
    "\n",
    "\t\ttrace = [[[-1 for i in range(col)] for i in range(row)] for i in range(row)]\n",
    "\t\tscore = [[[-1 for i in range(col)] for i in range(row)] for i in range(row)]\n",
    "\n",
    "\t\tfor idx, val in enumerate(sentence):\n",
    "\t\t\tif idx == 0:\n",
    "\t\t\t\tfor idx_pos, val_pos in enumerate(self.postags):\n",
    "\t\t\t\t\tscore[idx_pos][0][idx] = self.calc_e_prob(val_pos, sentence[idx]) # emit\n",
    "\t\t\telse:\n",
    "\t\t\t\tfor idx_pos, val_pos in enumerate(self.postags):\n",
    "\t\t\t\t\ttmp = -float('inf')\n",
    "\t\t\t\t\ttrace_tmp = -1\n",
    "\t\t\t\t\tfor idx_pos2, val_pos2 in enumerate(self.postags):\n",
    "\t\t\t\t\t\tfor idx_pos3, val_pos3 in enumerate(self.postags):\n",
    "\t\t\t\t\t\t\tr = score[idx_pos3][idx_pos2][idx-1]*self.calc_prob(val_pos3, val_pos2 ,val_pos)\n",
    "\t\t\t\t\t\t\tif r > tmp:\n",
    "\t\t\t\t\t\t\t\ttmp = r\n",
    "\t\t\t\t\t\t\t\ttrace_tmp = idx_pos3\n",
    "\t\t\t\t\t\t\ttrace[idx_pos][idx_pos2][idx] = trace_tmp\n",
    "\t\t\t\t\t\t\tscore[idx_pos][idx_pos2][idx] = tmp*self.calc_e_prob(val_pos, val)\n",
    "\t\treturn self._viterbi_decode(sentence, score, trace)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 结果分析"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "** bigram hmm **\n",
    "\n",
    "bigram hmm\n",
    "\n",
    "[('小明', 'nr'), ('爱', 'v'), ('老鼠', 'n'), ('和', 'c'), ('狗', 'n')]\n",
    "\n",
    "bigram hmm with viterbi decode\n",
    "\n",
    "[('小明', 'nr'), ('爱', 'v'), ('老鼠', 'n'), ('和', 'v'), ('狗', 'n')]\n",
    "\n",
    "**trigram hmm**\n",
    "\n",
    "trigram hmm\n",
    "\n",
    "[('小明', 'nr'), ('爱', 'v'), ('老鼠', 'n'), ('和', 'c'), ('狗', 'n')]\n",
    "\n",
    "trigram hmm with viterbi decode\n",
    "\n",
    "[('小明', 'nr'), ('爱', 'v'), ('老鼠', 'n'), ('和', 'c'), ('狗', 'n')]\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 项目后续\n",
    "\n",
    "过段时间会加入深度学习在NLP上的应用，如果你感兴趣，可以关注我的公众号，或者star, watch 本项目哦"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 联系作者\n",
    "\n",
    "@author sean\n",
    "\n",
    "@qq 929325776\n",
    "\n",
    "有什么问题，可以联系我，一起讨论"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}