natural language toolkit hidden markov model c 20012023 nltk project trevor cohn tacohncsse unimelb edu au philip blunsom pcblcsse unimelb edu au tiago tresoldi tiagotresoldi pro br fixes steven bird stevenbird1gmail com fixes joseph frazee jfrazeemail utexas edu fixes steven xu xxustudent unimelb edu au fixes url https www nltk org for license information see license txt hidden markov models hmms largely used to assign the correct label sequence to sequential data or assess the probability of a given label and data sequence these models are finite state machines characterised by a number of states transitions between these states and output symbols emitted while in each state the hmm is an extension to the markov chain where each state corresponds deterministically to a given event in the hmm the observation is a probabilistic function of the state hmms share the markov chain s assumption being that the probability of transition from one state to another only depends on the current state i e the series of states that led to the current state are not used they are also time invariant the hmm is a directed graph with probability weighted edges representing the probability of a transition between the source and sink states where each vertex emits an output symbol when entered the symbol or observation is nondeterministically generated for this reason knowing that a sequence of output observations was generated by a given hmm does not mean that the corresponding sequence of states and what the current state is is known this is the hidden in the hidden markov model formally a hmm can be characterised by the output observation alphabet this is the set of symbols which may be observed as output of the system the set of states the transition probabilities aij pst j st1 i these represent the probability of transition to each state from a given state the output probability matrix bik pxt ok st i these represent the probability of observing each symbol in a given state the initial state distribution this gives the probability of starting in each state to ground this discussion take a common nlp application partofspeech pos tagging an hmm is desirable for this task as the highest probability tag sequence can be calculated for a given sequence of word forms this differs from other tagging techniques which often tag each word individually seeking to optimise each individual tagging greedily without regard to the optimal combination of tags for a larger unit such as a sentence the hmm does this with the viterbi algorithm which efficiently computes the optimal path through the graph given the sequence of words forms in pos tagging the states usually have a 1 1 correspondence with the tag alphabet i e each state represents a single tag the output observation alphabet is the set of word forms the lexicon and the remaining three parameters are derived by a training regime with this information the probability of a given sentence can be easily derived by simply summing the probability of each distinct path through the model similarly the highest probability tagging sequence can be derived with the viterbi algorithm yielding a state sequence which can be mapped into a tag sequence this discussion assumes that the hmm has been trained this is probably the most difficult task with the model and requires either mle estimates of the parameters or unsupervised learning using the baumwelch algorithm a variant of em for more information please consult the source code for this module which includes extensive demonstration code hidden markov model class a generative model for labelling sequence data these models define the joint probability of a sequence of symbols and their labels state transitions as the product of the starting state probability the probability of each state transition and the probability of each observation being generated from each state this is described in more detail in the module documentation this implementation is based on the hmm description in chapter 8 huang acero and hon spoken language processing and includes an extension for training shallow hmm parsers or specialized hmms as in molina et al 2002 a specialized hmm modifies training data by applying a specialization function to create a new training set that is more appropriate for sequential tagging with an hmm a typical use case is chunking param symbols the set of output symbols alphabet type symbols seq of any param states a set of states representing state space type states seq of any param transitions transition probabilities prsi sj is the probability of transition from state i given the model is in statej type transitions conditionalprobdisti param outputs output probabilities prok si is the probability of emitting symbol k when entering state i type outputs conditionalprobdisti param priors initial state distribution prsi is the probability of starting in state i type priors probdisti param transform an optional function for transforming training instances defaults to the identity function type transform callable train a new hiddenmarkovmodeltagger using the given labeled and unlabeled training instances testing will be performed if test instances are provided return a hidden markov model tagger rtype hiddenmarkovmodeltagger param labeledsequence a sequence of labeled training instances i e a list of sentences represented as tuples type labeledsequence listlist param testsequence a sequence of labeled test instances type testsequence listlist param unlabeledsequence a sequence of unlabeled training instances i e a list of sentences represented as words type unlabeledsequence listlist param transform an optional function for transforming training instances defaults to the identity function see transform type transform function param estimator an optional function or class that maps a condition s frequency distribution to its probability distribution defaults to a lidstone distribution with gamma 0 1 type estimator class or function param verbose boolean flag indicating whether training should be verbose or include printed output type verbose bool param maxiterations number of baumwelch iterations to perform type maxiterations int returns the probability of the given symbol sequence if the sequence is labelled then returns the joint probability of the symbol state sequence otherwise uses the forward algorithm to find the probability over all label sequences return the probability of the sequence rtype float param sequence the sequence of symbols which must contain the text property and optionally the tag property type sequence token returns the logprobability of the given symbol sequence if the sequence is labelled then returns the joint logprobability of the symbol state sequence otherwise uses the forward algorithm to find the logprobability over all label sequences return the logprobability of the sequence rtype float param sequence the sequence of symbols which must contain the text property and optionally the tag property type sequence token tags the sequence with the highest probability state sequence this uses the bestpath method to find the viterbi path return a labelled sequence of symbols rtype list param unlabeledsequence the sequence of unlabeled symbols type unlabeledsequence list return the log probability of the symbol being observed in the given state rtype float the cache is a tuple p o x s where s maps symbols to integers i e it is the inverse mapping from self symbols for each symbol s in self symbols the following is true self symbolsss s o is the log output probabilities oi k log ptokentsymktagtstatei x is the log transition probabilities xi j log ptagtstatejtagt1statei p is the log prior probabilities pi log ptag0statei add new symbols to the symbol table and repopulate the output probabilities and symbol table mapping don t bother with the work if there aren t any new symbols add new columns to the output probability table without destroying the old probabilities only calculate probabilities for new symbols only create symbol mappings for new symbols returns the state sequence of the optimal most probable path through the hmm uses the viterbi algorithm to calculate this part by dynamic programming return the state sequence rtype sequence of any param unlabeledsequence the sequence of unlabeled symbols type unlabeledsequence list returns the state sequence of the optimal most probable path through the hmm uses the viterbi algorithm to calculate this part by dynamic programming this uses a simple direct method and is included for teaching purposes return the state sequence rtype sequence of any param unlabeledsequence the sequence of unlabeled symbols type unlabeledsequence list find the starting log probabilities for each state find the maximum log probabilities for reaching each state at time t find the highest probability final state traverse the backpointers b to find the state sequence randomly sample the hmm to generate a sentence of a given length this samples the prior distribution then the observation distribution and transition distribution for each subsequent observation and state this will mostly generate unintelligible garbage but can provide some amusement return the randomly created stateobservation sequence generated according to the hmm s probability distributions the subtokens have text and tag properties containing the observation and state respectively rtype list param rng random number generator type rng random or any object with a random method param length desired output length type length int sample the starting state and symbol prob dists sample the state transition and symbol prob dists returns the entropy over labellings of the given sequence this is given by ho sums prs o log prs o where the summation ranges over all state sequences s let z pro sums prs o where the summation ranges over all state sequences and o is the observation sequence as such the entropy can be reexpressed as h sums prs o log prs o z log z sums prs o log prs 0 log z sums prs o log prs0 sumt prst st1 sumt prot st the order of summation for the log terms can be flipped allowing dynamic programming to be used to calculate the entropy specifically we use the forward and backward probabilities alpha beta giving h log z sums0 alpha0s0 beta0s0 z log prs0 sumt si sj alphatsi prsj si prot1 sj betatsj z log prsj si sumt st alphatst betatst z log prot st this simply uses alpha and beta to find the probabilities of partial sequences constrained to include the given states at some point in time starting state t 0 print ps0 s state p state transitions print psd s sd s t0 s0 t1 s1 p symbol emissions print psd s t state p returns the pointwise entropy over the possible states at each position in the chain given the observation sequence return a matrix of transition log probabilities transiter self transitionssj logprobsi for sj in self states for si in self states transitionslogprob np fromitertransiter dtypenp float64 n lenself states return transitionslogprob reshapen n t def outputsvectorself symbol outiter self outputlogprobsj symbol for sj in self states return np fromiteroutiter dtypenp float64 def forwardprobabilityself unlabeledsequence t lenunlabeledsequence n lenself states alpha ninfarrayt n transitionslogprob self transitionsmatrix initialization symbol unlabeledsequence0text for i state in enumerateself states alpha0 i self priors logprobstate self outputlogprob state symbol induction for t in range1 t symbol unlabeledsequencettext outputlogprob self outputsvectorsymbol for i in rangen summand alphat 1 transitionslogprobi alphat i logsumexp2summand outputlogprobi return alpha def backwardprobabilityself unlabeledsequence t lenunlabeledsequence n lenself states beta ninfarrayt n transitionslogprob self transitionsmatrix t initialise the backward values 1 is an arbitrarily chosen value from rabiner tutorial betat 1 np log21 inductively calculate remaining backward values for t in ranget 2 1 1 symbol unlabeledsequencet 1text outputs self outputsvectorsymbol for i in rangen summand transitionslogprobi betat 1 outputs betat i logsumexp2summand return beta def testself testsequence verbosefalse kwargs def wordssent return word for word tag in sent def tagssent return tag for word tag in sent def flattenseq return listitertools chainseq testsequence self transformtestsequence predictedsequence listmapself tag mapwords testsequence if verbose for testsent predictedsent in ziptestsequence predictedsequence print test joinftokentag for token tag in testsent print printuntagged joins token for token tag in testsent print print hmmtagged joinftokentag for token tag in predictedsent print print entropy self entropytoken none for token tag in predictedsent print print 60 testtags flattenmaptags testsequence predictedtags flattenmaptags predictedsequence acc accuracytesttags predictedtags count sumlensent for sent in testsequence printaccuracy over d tokens 2f count acc 100 def reprself return hiddenmarkovmodeltagger d states and d output symbols lenself states lenself symbols class hiddenmarkovmodeltrainer def initself statesnone symbolsnone self states states if states else self symbols symbols if symbols else def trainself labeledsequencesnone unlabeledsequencesnone kwargs assert labeledsequences or unlabeledsequences model none if labeledsequences model self trainsupervisedlabeledsequences kwargs if unlabeledsequences if model kwargsmodel model model self trainunsupervisedunlabeledsequences kwargs return model def baumwelchstepself sequence model symboltonumber n lenmodel states m lenmodel symbols t lensequence compute forward and backward probabilities alpha model forwardprobabilitysequence beta model backwardprobabilitysequence find the log probability of the sequence lpk logsumexp2alphat 1 anumer ninfarrayn n bnumer ninfarrayn m adenom ninfarrayn bdenom ninfarrayn transitionslogprob model transitionsmatrix t for t in ranget symbol sequencettext not found fixme nextsymbol none if t t 1 nextsymbol sequencet 1text not found fixme xi symboltonumbersymbol nextoutputslogprob model outputsvectornextsymbol alphaplusbeta alphat betat if t t 1 numeradd transitionslogprob nextoutputslogprob betat 1 alphat reshapen 1 anumer np logaddexp2anumer numeradd adenom np logaddexp2adenom alphaplusbeta else bdenom np logaddexp2adenom alphaplusbeta bnumer xi np logaddexp2bnumer xi alphaplusbeta return lpk anumer adenom bnumer bdenom def trainunsupervisedself unlabeledsequences updateoutputstrue kwargs create a uniform hmm which will be iteratively refined unless given an existing model model kwargs getmodel if not model priors randomprobdistself states transitions dictionaryconditionalprobdist state randomprobdistself states for state in self states outputs dictionaryconditionalprobdist state randomprobdistself symbols for state in self states model hiddenmarkovmodeltagger self symbols self states transitions outputs priors self states model states self symbols model symbols n lenself states m lenself symbols symbolnumbers sym i for i sym in enumerateself symbols update model prob dists so that they can be modified model priors mutableprobdistmodel priors self states model transitions dictionaryconditionalprobdist s mutableprobdistmodel transitionss self states for s in self states if updateoutputs model outputs dictionaryconditionalprobdist s mutableprobdistmodel outputss self symbols for s in self states model resetcache iterate until convergence converged false lastlogprob none iteration 0 maxiterations kwargs getmaxiterations 1000 epsilon kwargs getconvergencelogprob 1e6 while not converged and iteration maxiterations anumer ninfarrayn n bnumer ninfarrayn m adenom ninfarrayn bdenom ninfarrayn logprob 0 for sequence in unlabeledsequences sequence listsequence if not sequence continue lpk seqanumer seqadenom seqbnumer seqbdenom self baumwelchstepsequence model symbolnumbers add these sums to the global a and b values for i in rangen anumeri np logaddexp2anumeri seqanumeri lpk bnumeri np logaddexp2bnumeri seqbnumeri lpk adenom np logaddexp2adenom seqadenom lpk bdenom np logaddexp2bdenom seqbdenom lpk logprob lpk use the calculated values to update the transition and output probability values for i in rangen logprobai anumeri adenomi logprobbi bnumeri bdenomi we should normalize all probabilities see p 391 huang et al let sump be k we can divide each pi by k to make sump 1 pi pik log2pi log2pi log2k logprobai logsumexp2logprobai logprobbi logsumexp2logprobbi update output and transition probabilities si self statesi for j in rangen sj self statesj model transitionssi updatesj logprobaij if updateoutputs for k in rangem ok self symbolsk model outputssi updateok logprobbik rabiner says the priors don t need to be updated i don t believe him fixme test for convergence if iteration 0 and abslogprob lastlogprob epsilon converged true printiteration iteration logprob logprob iteration 1 lastlogprob logprob return model def trainsupervisedself labelledsequences estimatornone default to the mle estimate if estimator is none estimator lambda fdist bins mleprobdistfdist count occurrences of starting states transitions out of each state and output symbols observed in each state knownsymbols setself symbols knownstates setself states starting freqdist transitions conditionalfreqdist outputs conditionalfreqdist for sequence in labelledsequences lasts none for token in sequence state tokentag symbol tokentext if lasts is none startingstate 1 else transitionslastsstate 1 outputsstatesymbol 1 lasts state update the state and symbol lists if state not in knownstates self states appendstate knownstates addstate if symbol not in knownsymbols self symbols appendsymbol knownsymbols addsymbol create probability distributions with smoothing n lenself states pi estimatorstarting n a conditionalprobdisttransitions estimator n b conditionalprobdistoutputs estimator lenself symbols return hiddenmarkovmodeltaggerself symbols self states a b pi def ninfarrayshape res np emptyshape np float64 res fillnp inf return res def logsumexp2arr max arr max return np log2np sum2 arr max max def logaddvalues x maxvalues if x np inf sumdiffs 0 for value in values sumdiffs 2 value x return x np log2sumdiffs else return x def createhmmtaggerstates symbols a b pi def pdvalues samples d dictzipsamples values return dictionaryprobdistd def cpdarray conditions samples d for values condition in ziparray conditions dcondition pdvalues samples return dictionaryconditionalprobdistd a cpda states states b cpdb states symbols pi pdpi states return hiddenmarkovmodeltagger symbolssymbols statesstates transitionsa outputsb priorspi def markethmmexample states bull bear static symbols up down unchanged a np array0 6 0 2 0 2 0 5 0 3 0 2 0 4 0 1 0 5 np float64 b np array0 7 0 1 0 2 0 1 0 6 0 3 0 3 0 3 0 4 np float64 pi np array0 5 0 2 0 3 np float64 model createhmmtaggerstates symbols a b pi return model states symbols def demo demonstrates hmm probability calculation print printhmm probability calculation demo print model states symbols markethmmexample printtesting model for test in up up up down up down 5 unchanged 5 up sequence t none for t in test printtesting with state sequence test printprobability model probabilitysequence printtagging model tagword for word tag in sequence printptagged model probabilitysequence printh model entropysequence printhexh model exhaustiveentropysequence printhpoint model pointentropysequence printhexhpoint model exhaustivepointentropysequence print def loadposnumsents from nltk corpus import brown sentences brown taggedsentscategoriesnews numsents tagre re compiler tagset set symbols set cleanedsentences for sentence in sentences for i in rangelensentence word tag sentencei word word lower normalize symbols addword log this word clean up the tag tag tagre matchtag group tagset addtag sentencei word tag store cleanedup tagged token cleanedsentences sentence return cleanedsentences listtagset listsymbols def demopos demonstrates pos tagging using supervised training print printhmm pos tagging demo print printtraining hmm labelledsequences tagset symbols loadpos20000 trainer hiddenmarkovmodeltrainertagset symbols hmm trainer trainsupervised labelledsequences10 estimatorlambda fd bins lidstoneprobdistfd 0 1 bins printtesting hmm testlabelledsequences 10 verbosetrue def untagsentences unlabeled for sentence in sentences unlabeled appendtokentext none for token in sentence return unlabeled def demoposbw test10 supervised20 unsupervised10 verbosetrue maxiterations5 demonstrates the baumwelch algorithm in pos tagging print printbaumwelch demo for pos tagging print printtraining hmm supervised d sentences supervised sentences tagset symbols loadpostest supervised unsupervised symbols set for sentence in sentences for token in sentence symbols addtokentext trainer hiddenmarkovmodeltrainertagset listsymbols hmm trainer trainsupervised sentencestest test supervised estimatorlambda fd bins lidstoneprobdistfd 0 1 bins hmm testsentences test verboseverbose printtraining unsupervised d sentences unsupervised it s rather slow so only use 10 samples by default unlabeled untagsentencestest supervised hmm trainer trainunsupervised unlabeled modelhmm maxiterationsmaxiterations hmm testsentences test verboseverbose def demobw demo baum welch by generating some sequences and then performing unsupervised training on them print printbaumwelch demo for market example print model states symbols markethmmexample generate some random sequences training import random rng random random rng seed0 for i in range10 item model randomsamplerng 5 training appendi0 none for i in item train on those examples starting with the model that generated them trainer hiddenmarkovmodeltrainerstates symbols hmm trainer trainunsupervisedtraining modelmodel maxiterations1000 natural language toolkit hidden markov model c 2001 2023 nltk project trevor cohn tacohn csse unimelb edu au philip blunsom pcbl csse unimelb edu au tiago tresoldi tiago tresoldi pro br fixes steven bird stevenbird1 gmail com fixes joseph frazee jfrazee mail utexas edu fixes steven xu xxu student unimelb edu au fixes url https www nltk org for license information see license txt hidden markov models hmms largely used to assign the correct label sequence to sequential data or assess the probability of a given label and data sequence these models are finite state machines characterised by a number of states transitions between these states and output symbols emitted while in each state the hmm is an extension to the markov chain where each state corresponds deterministically to a given event in the hmm the observation is a probabilistic function of the state hmms share the markov chain s assumption being that the probability of transition from one state to another only depends on the current state i e the series of states that led to the current state are not used they are also time invariant the hmm is a directed graph with probability weighted edges representing the probability of a transition between the source and sink states where each vertex emits an output symbol when entered the symbol or observation is non deterministically generated for this reason knowing that a sequence of output observations was generated by a given hmm does not mean that the corresponding sequence of states and what the current state is is known this is the hidden in the hidden markov model formally a hmm can be characterised by the output observation alphabet this is the set of symbols which may be observed as output of the system the set of states the transition probabilities a_ ij p s_t j s_ t 1 i these represent the probability of transition to each state from a given state the output probability matrix b_i k p x_t o_k s_t i these represent the probability of observing each symbol in a given state the initial state distribution this gives the probability of starting in each state to ground this discussion take a common nlp application part of speech pos tagging an hmm is desirable for this task as the highest probability tag sequence can be calculated for a given sequence of word forms this differs from other tagging techniques which often tag each word individually seeking to optimise each individual tagging greedily without regard to the optimal combination of tags for a larger unit such as a sentence the hmm does this with the viterbi algorithm which efficiently computes the optimal path through the graph given the sequence of words forms in pos tagging the states usually have a 1 1 correspondence with the tag alphabet i e each state represents a single tag the output observation alphabet is the set of word forms the lexicon and the remaining three parameters are derived by a training regime with this information the probability of a given sentence can be easily derived by simply summing the probability of each distinct path through the model similarly the highest probability tagging sequence can be derived with the viterbi algorithm yielding a state sequence which can be mapped into a tag sequence this discussion assumes that the hmm has been trained this is probably the most difficult task with the model and requires either mle estimates of the parameters or unsupervised learning using the baum welch algorithm a variant of em for more information please consult the source code for this module which includes extensive demonstration code index of text in a tuple index of tag in a tuple hidden markov model class a generative model for labelling sequence data these models define the joint probability of a sequence of symbols and their labels state transitions as the product of the starting state probability the probability of each state transition and the probability of each observation being generated from each state this is described in more detail in the module documentation this implementation is based on the hmm description in chapter 8 huang acero and hon spoken language processing and includes an extension for training shallow hmm parsers or specialized hmms as in molina et al 2002 a specialized hmm modifies training data by applying a specialization function to create a new training set that is more appropriate for sequential tagging with an hmm a typical use case is chunking param symbols the set of output symbols alphabet type symbols seq of any param states a set of states representing state space type states seq of any param transitions transition probabilities pr s_i s_j is the probability of transition from state i given the model is in state_j type transitions conditionalprobdisti param outputs output probabilities pr o_k s_i is the probability of emitting symbol k when entering state i type outputs conditionalprobdisti param priors initial state distribution pr s_i is the probability of starting in state i type priors probdisti param transform an optional function for transforming training instances defaults to the identity function type transform callable train a new hiddenmarkovmodeltagger using the given labeled and unlabeled training instances testing will be performed if test instances are provided return a hidden markov model tagger rtype hiddenmarkovmodeltagger param labeled_sequence a sequence of labeled training instances i e a list of sentences represented as tuples type labeled_sequence list list param test_sequence a sequence of labeled test instances type test_sequence list list param unlabeled_sequence a sequence of unlabeled training instances i e a list of sentences represented as words type unlabeled_sequence list list param transform an optional function for transforming training instances defaults to the identity function see transform type transform function param estimator an optional function or class that maps a condition s frequency distribution to its probability distribution defaults to a lidstone distribution with gamma 0 1 type estimator class or function param verbose boolean flag indicating whether training should be verbose or include printed output type verbose bool param max_iterations number of baum welch iterations to perform type max_iterations int returns the probability of the given symbol sequence if the sequence is labelled then returns the joint probability of the symbol state sequence otherwise uses the forward algorithm to find the probability over all label sequences return the probability of the sequence rtype float param sequence the sequence of symbols which must contain the text property and optionally the tag property type sequence token returns the log probability of the given symbol sequence if the sequence is labelled then returns the joint log probability of the symbol state sequence otherwise uses the forward algorithm to find the log probability over all label sequences return the log probability of the sequence rtype float param sequence the sequence of symbols which must contain the text property and optionally the tag property type sequence token tags the sequence with the highest probability state sequence this uses the best_path method to find the viterbi path return a labelled sequence of symbols rtype list param unlabeled_sequence the sequence of unlabeled symbols type unlabeled_sequence list return the log probability of the symbol being observed in the given state rtype float the cache is a tuple p o x s where s maps symbols to integers i e it is the inverse mapping from self _symbols for each symbol s in self _symbols the following is true self _symbols s s s o is the log output probabilities o i k log p token t sym k tag t state i x is the log transition probabilities x i j log p tag t state j tag t 1 state i p is the log prior probabilities p i log p tag 0 state i add new symbols to the symbol table and repopulate the output probabilities and symbol table mapping don t bother with the work if there aren t any new symbols add new columns to the output probability table without destroying the old probabilities only calculate probabilities for new symbols only create symbol mappings for new symbols returns the state sequence of the optimal most probable path through the hmm uses the viterbi algorithm to calculate this part by dynamic programming return the state sequence rtype sequence of any param unlabeled_sequence the sequence of unlabeled symbols type unlabeled_sequence list returns the state sequence of the optimal most probable path through the hmm uses the viterbi algorithm to calculate this part by dynamic programming this uses a simple direct method and is included for teaching purposes return the state sequence rtype sequence of any param unlabeled_sequence the sequence of unlabeled symbols type unlabeled_sequence list find the starting log probabilities for each state find the maximum log probabilities for reaching each state at time t find the highest probability final state traverse the back pointers b to find the state sequence randomly sample the hmm to generate a sentence of a given length this samples the prior distribution then the observation distribution and transition distribution for each subsequent observation and state this will mostly generate unintelligible garbage but can provide some amusement return the randomly created state observation sequence generated according to the hmm s probability distributions the subtokens have text and tag properties containing the observation and state respectively rtype list param rng random number generator type rng random or any object with a random method param length desired output length type length int sample the starting state and symbol prob dists sample the state transition and symbol prob dists returns the entropy over labellings of the given sequence this is given by h o sum_s pr s o log pr s o where the summation ranges over all state sequences s let z pr o sum_s pr s o where the summation ranges over all state sequences and o is the observation sequence as such the entropy can be re expressed as h sum_s pr s o log pr s o z log z sum_s pr s o log pr s 0 log z sum_s pr s o log pr s_0 sum_t pr s_t s_ t 1 sum_t pr o_t s_t the order of summation for the log terms can be flipped allowing dynamic programming to be used to calculate the entropy specifically we use the forward and backward probabilities alpha beta giving h log z sum_s0 alpha_0 s0 beta_0 s0 z log pr s0 sum_t si sj alpha_t si pr sj si pr o_t 1 sj beta_t sj z log pr sj si sum_t st alpha_t st beta_t st z log pr o_t st this simply uses alpha and beta to find the probabilities of partial sequences constrained to include the given state s at some point in time starting state t 0 print p s_0 s state p state transitions print p s_ d s s_ d s t0 s0 t1 s1 p symbol emissions print p s_ d s t state p returns the pointwise entropy over the possible states at each position in the chain given the observation sequence return a matrix of transition log probabilities return a vector with log probabilities of emitting a symbol when entering states return the forward probability matrix a t by n array of log probabilities where t is the length of the sequence and n is the number of states each entry t s gives the probability of being in state s at time t after observing the partial symbol sequence up to and including t param unlabeled_sequence the sequence of unlabeled symbols type unlabeled_sequence list return the forward log probability matrix rtype array initialization induction return the backward probability matrix a t by n array of log probabilities where t is the length of the sequence and n is the number of states each entry t s gives the probability of being in state s at time t after observing the partial symbol sequence from t t return the backward log probability matrix rtype array param unlabeled_sequence the sequence of unlabeled symbols type unlabeled_sequence list initialise the backward values 1 is an arbitrarily chosen value from rabiner tutorial inductively calculate remaining backward values tests the hiddenmarkovmodeltagger instance param test_sequence a sequence of labeled test instances type test_sequence list list param verbose boolean flag indicating whether training should be verbose or include printed output type verbose bool algorithms for learning hmm parameters from training data these include both supervised learning mle and unsupervised learning baum welch creates an hmm trainer to induce an hmm with the given states and output symbol alphabet a supervised and unsupervised training method may be used if either of the states or symbols are not given these may be derived from supervised training param states the set of state labels type states sequence of any param symbols the set of observation symbols type symbols sequence of any trains the hmm using both or either of supervised and unsupervised techniques return the trained model rtype hiddenmarkovmodeltagger param labelled_sequences the supervised training data a set of labelled sequences of observations ex word_1 tag_1 word_n tag_n type labelled_sequences list param unlabeled_sequences the unsupervised training data a set of sequences of observations ex word_1 word_n type unlabeled_sequences list param kwargs additional arguments to pass to the training methods compute forward and backward probabilities find the log probability of the sequence not found fixme not found fixme trains the hmm using the baum welch algorithm to maximise the probability of the data sequence this is a variant of the em algorithm and is unsupervised in that it doesn t need the state sequences for the symbols the code is based on a tutorial on hidden markov models and selected applications in speech recognition lawrence rabiner ieee 1989 return the trained model rtype hiddenmarkovmodeltagger param unlabeled_sequences the training data a set of sequences of observations type unlabeled_sequences list kwargs may include following parameters param model a hiddenmarkovmodeltagger instance used to begin the baum welch algorithm param max_iterations the maximum number of em iterations param convergence_logprob the maximum change in log probability to allow convergence create a uniform hmm which will be iteratively refined unless given an existing model update model prob dists so that they can be modified model _priors mutableprobdist model _priors self _states iterate until convergence add these sums to the global a and b values use the calculated values to update the transition and output probability values we should normalize all probabilities see p 391 huang et al let sum p be k we can divide each pi by k to make sum p 1 pi pi k log2 pi log2 pi log2 k update output and transition probabilities rabiner says the priors don t need to be updated i don t believe him fixme test for convergence supervised training maximising the joint probability of the symbol and state sequences this is done via collecting frequencies of transitions between states symbol observations while within each state and which states start a sentence these frequency distributions are then normalised into probability estimates which can be smoothed if desired return the trained model rtype hiddenmarkovmodeltagger param labelled_sequences the training data a set of labelled sequences of observations type labelled_sequences list param estimator a function taking a freqdist and a number of bins and returning a cprobdisti otherwise a mle estimate is used default to the mle estimate count occurrences of starting states transitions out of each state and output symbols observed in each state update the state and symbol lists create probability distributions with smoothing adds the logged values returning the logarithm of the addition return an example hmm described at page 381 huang et al demonstrates hmm probability calculation normalize log this word clean up the tag store cleaned up tagged token demonstrates pos tagging using supervised training demonstrates the baum welch algorithm in pos tagging it s rather slow so only use 10 samples by default demo baum welch by generating some sequences and then performing unsupervised training on them generate some random sequences train on those examples starting with the model that generated them
import itertools
import re

try:
    import numpy as np
except ImportError:
    pass

from nltk.metrics import accuracy
from nltk.probability import (
    ConditionalFreqDist,
    ConditionalProbDist,
    DictionaryConditionalProbDist,
    DictionaryProbDist,
    FreqDist,
    LidstoneProbDist,
    MLEProbDist,
    MutableProbDist,
    RandomProbDist,
)
from nltk.tag.api import TaggerI
from nltk.util import LazyMap, unique_list

_TEXT = 0  
_TAG = 1  


def _identity(labeled_symbols):
    return labeled_symbols


class HiddenMarkovModelTagger(TaggerI):
    

    def __init__(
        self, symbols, states, transitions, outputs, priors, transform=_identity
    ):
        self._symbols = unique_list(symbols)
        self._states = unique_list(states)
        self._transitions = transitions
        self._outputs = outputs
        self._priors = priors
        self._cache = None
        self._transform = transform

    @classmethod
    def _train(
        cls,
        labeled_sequence,
        test_sequence=None,
        unlabeled_sequence=None,
        transform=_identity,
        estimator=None,
        **kwargs,
    ):

        if estimator is None:

            def estimator(fd, bins):
                return LidstoneProbDist(fd, 0.1, bins)

        labeled_sequence = LazyMap(transform, labeled_sequence)
        symbols = unique_list(word for sent in labeled_sequence for word, tag in sent)
        tag_set = unique_list(tag for sent in labeled_sequence for word, tag in sent)

        trainer = HiddenMarkovModelTrainer(tag_set, symbols)
        hmm = trainer.train_supervised(labeled_sequence, estimator=estimator)
        hmm = cls(
            hmm._symbols,
            hmm._states,
            hmm._transitions,
            hmm._outputs,
            hmm._priors,
            transform=transform,
        )

        if test_sequence:
            hmm.test(test_sequence, verbose=kwargs.get("verbose", False))

        if unlabeled_sequence:
            max_iterations = kwargs.get("max_iterations", 5)
            hmm = trainer.train_unsupervised(
                unlabeled_sequence, model=hmm, max_iterations=max_iterations
            )
            if test_sequence:
                hmm.test(test_sequence, verbose=kwargs.get("verbose", False))

        return hmm

    @classmethod
    def train(
        cls, labeled_sequence, test_sequence=None, unlabeled_sequence=None, **kwargs
    ):
        
        return cls._train(labeled_sequence, test_sequence, unlabeled_sequence, **kwargs)

    def probability(self, sequence):
        
        return 2 ** (self.log_probability(self._transform(sequence)))

    def log_probability(self, sequence):
        
        sequence = self._transform(sequence)

        T = len(sequence)

        if T > 0 and sequence[0][_TAG]:
            last_state = sequence[0][_TAG]
            p = self._priors.logprob(last_state) + self._output_logprob(
                last_state, sequence[0][_TEXT]
            )
            for t in range(1, T):
                state = sequence[t][_TAG]
                p += self._transitions[last_state].logprob(
                    state
                ) + self._output_logprob(state, sequence[t][_TEXT])
                last_state = state
            return p
        else:
            alpha = self._forward_probability(sequence)
            p = logsumexp2(alpha[T - 1])
            return p

    def tag(self, unlabeled_sequence):
        
        unlabeled_sequence = self._transform(unlabeled_sequence)
        return self._tag(unlabeled_sequence)

    def _tag(self, unlabeled_sequence):
        path = self._best_path(unlabeled_sequence)
        return list(zip(unlabeled_sequence, path))

    def _output_logprob(self, state, symbol):
        
        return self._outputs[state].logprob(symbol)

    def _create_cache(self):
        
        if not self._cache:
            N = len(self._states)
            M = len(self._symbols)
            P = np.zeros(N, np.float32)
            X = np.zeros((N, N), np.float32)
            O = np.zeros((N, M), np.float32)
            for i in range(N):
                si = self._states[i]
                P[i] = self._priors.logprob(si)
                for j in range(N):
                    X[i, j] = self._transitions[si].logprob(self._states[j])
                for k in range(M):
                    O[i, k] = self._output_logprob(si, self._symbols[k])
            S = {}
            for k in range(M):
                S[self._symbols[k]] = k
            self._cache = (P, O, X, S)

    def _update_cache(self, symbols):
        
        
        if symbols:
            self._create_cache()
            P, O, X, S = self._cache
            for symbol in symbols:
                if symbol not in self._symbols:
                    self._cache = None
                    self._symbols.append(symbol)
            
            if not self._cache:
                N = len(self._states)
                M = len(self._symbols)
                Q = O.shape[1]
                
                
                O = np.hstack([O, np.zeros((N, M - Q), np.float32)])
                for i in range(N):
                    si = self._states[i]
                    
                    for k in range(Q, M):
                        O[i, k] = self._output_logprob(si, self._symbols[k])
                
                for k in range(Q, M):
                    S[self._symbols[k]] = k
                self._cache = (P, O, X, S)

    def reset_cache(self):
        self._cache = None

    def best_path(self, unlabeled_sequence):
        
        unlabeled_sequence = self._transform(unlabeled_sequence)
        return self._best_path(unlabeled_sequence)

    def _best_path(self, unlabeled_sequence):
        T = len(unlabeled_sequence)
        N = len(self._states)
        self._create_cache()
        self._update_cache(unlabeled_sequence)
        P, O, X, S = self._cache

        V = np.zeros((T, N), np.float32)
        B = -np.ones((T, N), int)

        V[0] = P + O[:, S[unlabeled_sequence[0]]]
        for t in range(1, T):
            for j in range(N):
                vs = V[t - 1, :] + X[:, j]
                best = np.argmax(vs)
                V[t, j] = vs[best] + O[j, S[unlabeled_sequence[t]]]
                B[t, j] = best

        current = np.argmax(V[T - 1, :])
        sequence = [current]
        for t in range(T - 1, 0, -1):
            last = B[t, current]
            sequence.append(last)
            current = last

        sequence.reverse()
        return list(map(self._states.__getitem__, sequence))

    def best_path_simple(self, unlabeled_sequence):
        
        unlabeled_sequence = self._transform(unlabeled_sequence)
        return self._best_path_simple(unlabeled_sequence)

    def _best_path_simple(self, unlabeled_sequence):
        T = len(unlabeled_sequence)
        N = len(self._states)
        V = np.zeros((T, N), np.float64)
        B = {}

        
        symbol = unlabeled_sequence[0]
        for i, state in enumerate(self._states):
            V[0, i] = self._priors.logprob(state) + self._output_logprob(state, symbol)
            B[0, state] = None

        
        for t in range(1, T):
            symbol = unlabeled_sequence[t]
            for j in range(N):
                sj = self._states[j]
                best = None
                for i in range(N):
                    si = self._states[i]
                    va = V[t - 1, i] + self._transitions[si].logprob(sj)
                    if not best or va > best[0]:
                        best = (va, si)
                V[t, j] = best[0] + self._output_logprob(sj, symbol)
                B[t, sj] = best[1]

        
        best = None
        for i in range(N):
            val = V[T - 1, i]
            if not best or val > best[0]:
                best = (val, self._states[i])

        
        current = best[1]
        sequence = [current]
        for t in range(T - 1, 0, -1):
            last = B[t, current]
            sequence.append(last)
            current = last

        sequence.reverse()
        return sequence

    def random_sample(self, rng, length):
        

        
        tokens = []
        state = self._sample_probdist(self._priors, rng.random(), self._states)
        symbol = self._sample_probdist(
            self._outputs[state], rng.random(), self._symbols
        )
        tokens.append((symbol, state))

        for i in range(1, length):
            
            state = self._sample_probdist(
                self._transitions[state], rng.random(), self._states
            )
            symbol = self._sample_probdist(
                self._outputs[state], rng.random(), self._symbols
            )
            tokens.append((symbol, state))

        return tokens

    def _sample_probdist(self, probdist, p, samples):
        cum_p = 0
        for sample in samples:
            add_p = probdist.prob(sample)
            if cum_p <= p <= cum_p + add_p:
                return sample
            cum_p += add_p
        raise Exception("Invalid probability distribution - " "does not sum to one")

    def entropy(self, unlabeled_sequence):
        
        unlabeled_sequence = self._transform(unlabeled_sequence)

        T = len(unlabeled_sequence)
        N = len(self._states)

        alpha = self._forward_probability(unlabeled_sequence)
        beta = self._backward_probability(unlabeled_sequence)
        normalisation = logsumexp2(alpha[T - 1])

        entropy = normalisation

        
        for i, state in enumerate(self._states):
            p = 2 ** (alpha[0, i] + beta[0, i] - normalisation)
            entropy -= p * self._priors.logprob(state)
            

        
        for t0 in range(T - 1):
            t1 = t0 + 1
            for i0, s0 in enumerate(self._states):
                for i1, s1 in enumerate(self._states):
                    p = 2 ** (
                        alpha[t0, i0]
                        + self._transitions[s0].logprob(s1)
                        + self._outputs[s1].logprob(unlabeled_sequence[t1][_TEXT])
                        + beta[t1, i1]
                        - normalisation
                    )
                    entropy -= p * self._transitions[s0].logprob(s1)
                    

        
        for t in range(T):
            for i, state in enumerate(self._states):
                p = 2 ** (alpha[t, i] + beta[t, i] - normalisation)
                entropy -= p * self._outputs[state].logprob(
                    unlabeled_sequence[t][_TEXT]
                )
                

        return entropy

    def point_entropy(self, unlabeled_sequence):
        
        unlabeled_sequence = self._transform(unlabeled_sequence)

        T = len(unlabeled_sequence)
        N = len(self._states)

        alpha = self._forward_probability(unlabeled_sequence)
        beta = self._backward_probability(unlabeled_sequence)
        normalisation = logsumexp2(alpha[T - 1])

        entropies = np.zeros(T, np.float64)
        probs = np.zeros(N, np.float64)
        for t in range(T):
            for s in range(N):
                probs[s] = alpha[t, s] + beta[t, s] - normalisation

            for s in range(N):
                entropies[t] -= 2 ** (probs[s]) * probs[s]

        return entropies

    def _exhaustive_entropy(self, unlabeled_sequence):
        unlabeled_sequence = self._transform(unlabeled_sequence)

        T = len(unlabeled_sequence)
        N = len(self._states)

        labellings = [[state] for state in self._states]
        for t in range(T - 1):
            current = labellings
            labellings = []
            for labelling in current:
                for state in self._states:
                    labellings.append(labelling + [state])

        log_probs = []
        for labelling in labellings:
            labeled_sequence = unlabeled_sequence[:]
            for t, label in enumerate(labelling):
                labeled_sequence[t] = (labeled_sequence[t][_TEXT], label)
            lp = self.log_probability(labeled_sequence)
            log_probs.append(lp)
        normalisation = _log_add(*log_probs)

        entropy = 0
        for lp in log_probs:
            lp -= normalisation
            entropy -= 2 ** (lp) * lp

        return entropy

    def _exhaustive_point_entropy(self, unlabeled_sequence):
        unlabeled_sequence = self._transform(unlabeled_sequence)

        T = len(unlabeled_sequence)
        N = len(self._states)

        labellings = [[state] for state in self._states]
        for t in range(T - 1):
            current = labellings
            labellings = []
            for labelling in current:
                for state in self._states:
                    labellings.append(labelling + [state])

        log_probs = []
        for labelling in labellings:
            labelled_sequence = unlabeled_sequence[:]
            for t, label in enumerate(labelling):
                labelled_sequence[t] = (labelled_sequence[t][_TEXT], label)
            lp = self.log_probability(labelled_sequence)
            log_probs.append(lp)

        normalisation = _log_add(*log_probs)

        probabilities = _ninf_array((T, N))

        for labelling, lp in zip(labellings, log_probs):
            lp -= normalisation
            for t, label in enumerate(labelling):
                index = self._states.index(label)
                probabilities[t, index] = _log_add(probabilities[t, index], lp)

        entropies = np.zeros(T, np.float64)
        for t in range(T):
            for s in range(N):
                entropies[t] -= 2 ** (probabilities[t, s]) * probabilities[t, s]

        return entropies

    def _transitions_matrix(self):
        
        trans_iter = (
            self._transitions[sj].logprob(si)
            for sj in self._states
            for si in self._states
        )

        transitions_logprob = np.fromiter(trans_iter, dtype=np.float64)
        N = len(self._states)
        return transitions_logprob.reshape((N, N)).T

    def _outputs_vector(self, symbol):
        
        out_iter = (self._output_logprob(sj, symbol) for sj in self._states)
        return np.fromiter(out_iter, dtype=np.float64)

    def _forward_probability(self, unlabeled_sequence):
        
        T = len(unlabeled_sequence)
        N = len(self._states)
        alpha = _ninf_array((T, N))

        transitions_logprob = self._transitions_matrix()

        
        symbol = unlabeled_sequence[0][_TEXT]
        for i, state in enumerate(self._states):
            alpha[0, i] = self._priors.logprob(state) + self._output_logprob(
                state, symbol
            )

        
        for t in range(1, T):
            symbol = unlabeled_sequence[t][_TEXT]
            output_logprob = self._outputs_vector(symbol)

            for i in range(N):
                summand = alpha[t - 1] + transitions_logprob[i]
                alpha[t, i] = logsumexp2(summand) + output_logprob[i]

        return alpha

    def _backward_probability(self, unlabeled_sequence):
        
        T = len(unlabeled_sequence)
        N = len(self._states)
        beta = _ninf_array((T, N))

        transitions_logprob = self._transitions_matrix().T

        
        
        beta[T - 1, :] = np.log2(1)

        
        for t in range(T - 2, -1, -1):
            symbol = unlabeled_sequence[t + 1][_TEXT]
            outputs = self._outputs_vector(symbol)

            for i in range(N):
                summand = transitions_logprob[i] + beta[t + 1] + outputs
                beta[t, i] = logsumexp2(summand)

        return beta

    def test(self, test_sequence, verbose=False, **kwargs):
        

        def words(sent):
            return [word for (word, tag) in sent]

        def tags(sent):
            return [tag for (word, tag) in sent]

        def flatten(seq):
            return list(itertools.chain(*seq))

        test_sequence = self._transform(test_sequence)
        predicted_sequence = list(map(self._tag, map(words, test_sequence)))

        if verbose:
            for test_sent, predicted_sent in zip(test_sequence, predicted_sequence):
                print(
                    "Test:",
                    " ".join(f"{token}/{tag}" for (token, tag) in test_sent),
                )
                print()
                print("Untagged:", " ".join("%s" % token for (token, tag) in test_sent))
                print()
                print(
                    "HMM-tagged:",
                    " ".join(f"{token}/{tag}" for (token, tag) in predicted_sent),
                )
                print()
                print(
                    "Entropy:",
                    self.entropy([(token, None) for (token, tag) in predicted_sent]),
                )
                print()
                print("-" * 60)

        test_tags = flatten(map(tags, test_sequence))
        predicted_tags = flatten(map(tags, predicted_sequence))

        acc = accuracy(test_tags, predicted_tags)
        count = sum(len(sent) for sent in test_sequence)
        print("accuracy over %d tokens: %.2f" % (count, acc * 100))

    def __repr__(self):
        return "<HiddenMarkovModelTagger %d states and %d output symbols>" % (
            len(self._states),
            len(self._symbols),
        )


class HiddenMarkovModelTrainer:
    

    def __init__(self, states=None, symbols=None):
        self._states = states if states else []
        self._symbols = symbols if symbols else []

    def train(self, labeled_sequences=None, unlabeled_sequences=None, **kwargs):
        
        assert labeled_sequences or unlabeled_sequences
        model = None
        if labeled_sequences:
            model = self.train_supervised(labeled_sequences, **kwargs)
        if unlabeled_sequences:
            if model:
                kwargs["model"] = model
            model = self.train_unsupervised(unlabeled_sequences, **kwargs)
        return model

    def _baum_welch_step(self, sequence, model, symbol_to_number):

        N = len(model._states)
        M = len(model._symbols)
        T = len(sequence)

        
        alpha = model._forward_probability(sequence)
        beta = model._backward_probability(sequence)

        
        lpk = logsumexp2(alpha[T - 1])

        A_numer = _ninf_array((N, N))
        B_numer = _ninf_array((N, M))
        A_denom = _ninf_array(N)
        B_denom = _ninf_array(N)

        transitions_logprob = model._transitions_matrix().T

        for t in range(T):
            symbol = sequence[t][_TEXT]  
            next_symbol = None
            if t < T - 1:
                next_symbol = sequence[t + 1][_TEXT]  
            xi = symbol_to_number[symbol]

            next_outputs_logprob = model._outputs_vector(next_symbol)
            alpha_plus_beta = alpha[t] + beta[t]

            if t < T - 1:
                numer_add = (
                    transitions_logprob
                    + next_outputs_logprob
                    + beta[t + 1]
                    + alpha[t].reshape(N, 1)
                )
                A_numer = np.logaddexp2(A_numer, numer_add)
                A_denom = np.logaddexp2(A_denom, alpha_plus_beta)
            else:
                B_denom = np.logaddexp2(A_denom, alpha_plus_beta)

            B_numer[:, xi] = np.logaddexp2(B_numer[:, xi], alpha_plus_beta)

        return lpk, A_numer, A_denom, B_numer, B_denom

    def train_unsupervised(self, unlabeled_sequences, update_outputs=True, **kwargs):
        

        
        
        model = kwargs.get("model")
        if not model:
            priors = RandomProbDist(self._states)
            transitions = DictionaryConditionalProbDist(
                {state: RandomProbDist(self._states) for state in self._states}
            )
            outputs = DictionaryConditionalProbDist(
                {state: RandomProbDist(self._symbols) for state in self._states}
            )
            model = HiddenMarkovModelTagger(
                self._symbols, self._states, transitions, outputs, priors
            )

        self._states = model._states
        self._symbols = model._symbols

        N = len(self._states)
        M = len(self._symbols)
        symbol_numbers = {sym: i for i, sym in enumerate(self._symbols)}

        
        

        model._transitions = DictionaryConditionalProbDist(
            {
                s: MutableProbDist(model._transitions[s], self._states)
                for s in self._states
            }
        )

        if update_outputs:
            model._outputs = DictionaryConditionalProbDist(
                {
                    s: MutableProbDist(model._outputs[s], self._symbols)
                    for s in self._states
                }
            )

        model.reset_cache()

        
        converged = False
        last_logprob = None
        iteration = 0
        max_iterations = kwargs.get("max_iterations", 1000)
        epsilon = kwargs.get("convergence_logprob", 1e-6)

        while not converged and iteration < max_iterations:
            A_numer = _ninf_array((N, N))
            B_numer = _ninf_array((N, M))
            A_denom = _ninf_array(N)
            B_denom = _ninf_array(N)

            logprob = 0
            for sequence in unlabeled_sequences:
                sequence = list(sequence)
                if not sequence:
                    continue

                (
                    lpk,
                    seq_A_numer,
                    seq_A_denom,
                    seq_B_numer,
                    seq_B_denom,
                ) = self._baum_welch_step(sequence, model, symbol_numbers)

                
                for i in range(N):
                    A_numer[i] = np.logaddexp2(A_numer[i], seq_A_numer[i] - lpk)
                    B_numer[i] = np.logaddexp2(B_numer[i], seq_B_numer[i] - lpk)

                A_denom = np.logaddexp2(A_denom, seq_A_denom - lpk)
                B_denom = np.logaddexp2(B_denom, seq_B_denom - lpk)

                logprob += lpk

            
            
            for i in range(N):
                logprob_Ai = A_numer[i] - A_denom[i]
                logprob_Bi = B_numer[i] - B_denom[i]

                
                
                
                
                
                logprob_Ai -= logsumexp2(logprob_Ai)
                logprob_Bi -= logsumexp2(logprob_Bi)

                
                si = self._states[i]

                for j in range(N):
                    sj = self._states[j]
                    model._transitions[si].update(sj, logprob_Ai[j])

                if update_outputs:
                    for k in range(M):
                        ok = self._symbols[k]
                        model._outputs[si].update(ok, logprob_Bi[k])

                
                

            
            if iteration > 0 and abs(logprob - last_logprob) < epsilon:
                converged = True

            print("iteration", iteration, "logprob", logprob)
            iteration += 1
            last_logprob = logprob

        return model

    def train_supervised(self, labelled_sequences, estimator=None):
        

        
        if estimator is None:
            estimator = lambda fdist, bins: MLEProbDist(fdist)

        
        
        known_symbols = set(self._symbols)
        known_states = set(self._states)

        starting = FreqDist()
        transitions = ConditionalFreqDist()
        outputs = ConditionalFreqDist()
        for sequence in labelled_sequences:
            lasts = None
            for token in sequence:
                state = token[_TAG]
                symbol = token[_TEXT]
                if lasts is None:
                    starting[state] += 1
                else:
                    transitions[lasts][state] += 1
                outputs[state][symbol] += 1
                lasts = state

                
                if state not in known_states:
                    self._states.append(state)
                    known_states.add(state)

                if symbol not in known_symbols:
                    self._symbols.append(symbol)
                    known_symbols.add(symbol)

        
        N = len(self._states)
        pi = estimator(starting, N)
        A = ConditionalProbDist(transitions, estimator, N)
        B = ConditionalProbDist(outputs, estimator, len(self._symbols))

        return HiddenMarkovModelTagger(self._symbols, self._states, A, B, pi)


def _ninf_array(shape):
    res = np.empty(shape, np.float64)
    res.fill(-np.inf)
    return res


def logsumexp2(arr):
    max_ = arr.max()
    return np.log2(np.sum(2 ** (arr - max_))) + max_


def _log_add(*values):
    
    x = max(values)
    if x > -np.inf:
        sum_diffs = 0
        for value in values:
            sum_diffs += 2 ** (value - x)
        return x + np.log2(sum_diffs)
    else:
        return x


def _create_hmm_tagger(states, symbols, A, B, pi):
    def pd(values, samples):
        d = dict(zip(samples, values))
        return DictionaryProbDist(d)

    def cpd(array, conditions, samples):
        d = {}
        for values, condition in zip(array, conditions):
            d[condition] = pd(values, samples)
        return DictionaryConditionalProbDist(d)

    A = cpd(A, states, states)
    B = cpd(B, states, symbols)
    pi = pd(pi, states)
    return HiddenMarkovModelTagger(
        symbols=symbols, states=states, transitions=A, outputs=B, priors=pi
    )


def _market_hmm_example():
    
    states = ["bull", "bear", "static"]
    symbols = ["up", "down", "unchanged"]
    A = np.array([[0.6, 0.2, 0.2], [0.5, 0.3, 0.2], [0.4, 0.1, 0.5]], np.float64)
    B = np.array([[0.7, 0.1, 0.2], [0.1, 0.6, 0.3], [0.3, 0.3, 0.4]], np.float64)
    pi = np.array([0.5, 0.2, 0.3], np.float64)

    model = _create_hmm_tagger(states, symbols, A, B, pi)
    return model, states, symbols


def demo():
    

    print()
    print("HMM probability calculation demo")
    print()

    model, states, symbols = _market_hmm_example()

    print("Testing", model)

    for test in [
        ["up", "up"],
        ["up", "down", "up"],
        ["down"] * 5,
        ["unchanged"] * 5 + ["up"],
    ]:

        sequence = [(t, None) for t in test]

        print("Testing with state sequence", test)
        print("probability =", model.probability(sequence))
        print("tagging =    ", model.tag([word for (word, tag) in sequence]))
        print("p(tagged) =  ", model.probability(sequence))
        print("H =          ", model.entropy(sequence))
        print("H_exh =      ", model._exhaustive_entropy(sequence))
        print("H(point) =   ", model.point_entropy(sequence))
        print("H_exh(point)=", model._exhaustive_point_entropy(sequence))
        print()


def load_pos(num_sents):
    from nltk.corpus import brown

    sentences = brown.tagged_sents(categories="news")[:num_sents]

    tag_re = re.compile(r"[*]|--|[^+*-]+")
    tag_set = set()
    symbols = set()

    cleaned_sentences = []
    for sentence in sentences:
        for i in range(len(sentence)):
            word, tag = sentence[i]
            word = word.lower()  
            symbols.add(word)  
            
            tag = tag_re.match(tag).group()
            tag_set.add(tag)
            sentence[i] = (word, tag)  
        cleaned_sentences += [sentence]

    return cleaned_sentences, list(tag_set), list(symbols)


def demo_pos():
    

    print()
    print("HMM POS tagging demo")
    print()

    print("Training HMM...")
    labelled_sequences, tag_set, symbols = load_pos(20000)
    trainer = HiddenMarkovModelTrainer(tag_set, symbols)
    hmm = trainer.train_supervised(
        labelled_sequences[10:],
        estimator=lambda fd, bins: LidstoneProbDist(fd, 0.1, bins),
    )

    print("Testing...")
    hmm.test(labelled_sequences[:10], verbose=True)


def _untag(sentences):
    unlabeled = []
    for sentence in sentences:
        unlabeled.append([(token[_TEXT], None) for token in sentence])
    return unlabeled


def demo_pos_bw(
    test=10, supervised=20, unsupervised=10, verbose=True, max_iterations=5
):
    

    print()
    print("Baum-Welch demo for POS tagging")
    print()

    print("Training HMM (supervised, %d sentences)..." % supervised)

    sentences, tag_set, symbols = load_pos(test + supervised + unsupervised)

    symbols = set()
    for sentence in sentences:
        for token in sentence:
            symbols.add(token[_TEXT])

    trainer = HiddenMarkovModelTrainer(tag_set, list(symbols))
    hmm = trainer.train_supervised(
        sentences[test : test + supervised],
        estimator=lambda fd, bins: LidstoneProbDist(fd, 0.1, bins),
    )

    hmm.test(sentences[:test], verbose=verbose)

    print("Training (unsupervised, %d sentences)..." % unsupervised)
    
    unlabeled = _untag(sentences[test + supervised :])
    hmm = trainer.train_unsupervised(
        unlabeled, model=hmm, max_iterations=max_iterations
    )
    hmm.test(sentences[:test], verbose=verbose)


def demo_bw():
    
    

    print()
    print("Baum-Welch demo for market example")
    print()

    model, states, symbols = _market_hmm_example()

    
    training = []
    import random

    rng = random.Random()
    rng.seed(0)
    for i in range(10):
        item = model.random_sample(rng, 5)
        training.append([(i[0], None) for i in item])

    
    trainer = HiddenMarkovModelTrainer(states, symbols)
    hmm = trainer.train_unsupervised(training, model=model, max_iterations=1000)