HMM中文分词

hmm.py

# encoding: utf8

"""
Created on 2017.03.06

@author: yalei
"""

import itertools


def word2state(string):
    nlen = len(string)
    if nlen == 1:
        return ['S']
    else:
        return ['B'] + ['M'] * (nlen - 2) + ['E']


def state2word(string, path):
    words = []
    word_cache = []
    for i, s in enumerate(string):
        state = path[i]
        if state == 'S':
            words.append(s)
            continue
        word_cache.append(s)
        if state == 'E':
            words.append(''.join(word_cache))
            word_cache = []
    print ' / '.join(words)


class HMM():
    def __init__(self, state_list):
        # state_list = ['B', 'M', 'E', 'S']
        self.states = state_list
        self.init_state = {state: 0 for state in state_list}   # initial state probbility
        self.count_state = {state: 0 for state in state_list}
        self.move_state = {state: {} for state in state_list}  # transform matrix
        self.emit_state = {state: {} for state in state_list}  # emit matrix
        self.word_set = set()
        self.__new_feed = False
        self.data = {}

    def feed(self, lines, verbose=True):
        self.__new_feed = True
        for line_count, line in enumerate(lines):
            line = line.strip().decode("utf-8", "ignore")
            if not line:
                continue
            char_list = [char for char in line if char != ' ']
            self.word_set = self.word_set | set(char_list)
            # words to states
            line_state = [word2state(word) for word in line.split(' ')]
            line_state = list(itertools.chain(*line_state))
            assert len(line_state) == len(char_list)
            for i, state in enumerate(line_state):
                self.count_state[state] += 1
                if i == 0:
                    self.init_state[state] += 1
                    continue
                self.move_state[line_state[i-1]][state] = self.move_state[line_state[i-1]].get(state, 0) + 1
                char = char_list[i]
                self.emit_state[state][char] = self.emit_state[state].get(char, 0) + 1
            if verbose and line_count >= 1000 and line_count % 1000 == 0:
                print 'process %s lines' % line_count
        if verbose:
            print 'process %s chars: done.' % len(self.word_set)

    @classmethod
    def viterbi(self, obs, states, start_p, trans_p, emit_p):
        """
        :param obs: observation
        :param states: states
        :param start_p: initial distribution of states
        :param trans_p: transform matrix between states
        :param emit_p: emit matrix
        :return: path
        """
        if len(obs) == 0:
            return
        V = [{}]
        path = {}
        # first step
        for s in states:
            V[0][s] = start_p[s] * emit_p[s].get(obs[0], 0)
            path[s] = [s]
        # 2->end steps
        for i in range(1, len(obs)):
            char = obs[i]
            V.append({})
            newPath = {}
            for s in states:
                prob, state = max([(V[i-1].get(s0, 0) * trans_p[s0].get(s, 0) * emit_p[s].get(char, 0), s0) for s0 in states])
                V[i][s] = prob
                newPath[s] = path[state] + [s]
            path = newPath
        prob, state = max([(V[len(obs) - 1][s], s) for s in states])
        return prob, path[state]

    def matrix(self):
        if self.__new_feed:
            # initial distribution of states
            state_count = sum(self.init_state.values())
            self.data['start_p'] = {k: 1.0 * v / state_count for k, v in self.init_state.iteritems()}
            # trans matrix
            self.data['prob_trans'] = {}
            for key, dic in self.move_state.items():
                self.data['prob_trans'][key] = {k: 1.0 * v / self.count_state[key] for k,v in dic.items()}
            # emit matrix
            self.data['prob_emit'] = {}
            for key, dic in self.emit_state.items():
                self.data['prob_emit'][key] = {k: 1.0 * v / self.count_state[key] for k,v in dic.items()}
        return self.data

    def infer(self, sentence):
        data = self.matrix()
        prob, path_list = self.viterbi(sentence, self.states,
                                       data['start_p'],
                                       data['prob_trans'],
                                       data['prob_emit'])
        return (prob, path_list)


if __name__ == '__main__':
    hmm = HMM(['B', 'M', 'E', 'S'])
    with open('trainCorpus.txt_utf8', 'r') as f:
        hmm.feed(f)

    def cut(sen):
        prob, path = hmm.infer(sen)
        print 'max prob: ', prob
        print path
        state2word(sen, path)

    cut(u"你好我亲爱的祖国")

trainCorpus.txt_utf8

HMM (隐马尔可夫)

1. 观测值集合 ObservedSet
2. 初始状态分布 InitStatus
3. 转移概率矩阵 TransProbMatrix
4. 发射概率矩阵 EmitRobMatrix
5. 状态值集合

将HMM应用在分词上，要解决的问题是：
参数(ObservedSet, TransProbMatrix, EmitRobMatrix, InitStatus)已知的情况下，求解状态值序列。