基于Hidden Markov Model的主题文本分割代码报错求助
2026-7-19
基于HMM的主题文本分割报错排查与解决
问题背景
编写函数实现长文档按主题分割，作为向量搜索前的文本预处理步骤，采用pomegranate v1.0的DenseHMM实现，运行时出现索引错误。
原始代码
# import pomegranate library and numpy import pomegranate from pomegranate.hmm import DenseHMM import numpy import torch # import LabelEncoder from sklearn from sklearn.preprocessing import LabelEncoder def segment_document(document): # define the range of possible topics min_topics = 2 max_topics = 20 # initialize a list to store the models and their scores models = [] # create an encoder object encoder = LabelEncoder() # fit the encoder to the vocabulary vocab = numpy.unique(numpy.char.split(document))[0] encoder.fit(vocab) # transform the document into numeric values document = encoder.transform(document.split()) document = document.reshape(1, -1) # reshape into a 2D array with one row # add another dimension to document for DenseHMM input document = numpy.expand_dims(document, axis=-1) # convert document to torch tensor of float type document = torch.from_numpy(document).float() vocab_size = len(vocab) # loop over the number of topics for num_topics in range(min_topics, max_topics + 1): # initialize state priors and transition probabilities state_priors = numpy.random.dirichlet(numpy.ones(num_topics)) trans_probs = numpy.random.dirichlet(numpy.ones(num_topics), size=num_topics) # initialize emission factors with random or uniform probabilities emission_probs = [] for i in range(num_topics): # create a categorical distribution from random integers and their counts ints = numpy.random.randint(0, vocab_size, size=100) # generate 100 random integers in [0, vocab_size) counts = numpy.bincount(ints) # count the occurrences of each integer probs = counts / counts.sum() # normalize the counts to get probabilities print(f"probs before adjustment: {probs}, sum: {numpy.sum(probs)}") assert numpy.isclose(numpy.sum(probs), 1) # check if the sum is close to 1 within a tolerance print(f"probs after adjustment: {probs}, sum: {numpy.sum(probs)}") probs = probs.reshape(1, -1) # reshape into a 2D array with one row emission_probs.append(pomegranate.distributions.Categorical(probs)) # create a categorical distribution # create model object model = DenseHMM() # add distributions to model model.add_distributions(emission_probs) # set starts and ends probabilities for model model.starts = state_priors model.ends = numpy.zeros(num_topics) #train model using Baum-Welch algorithm model.fit([document]) # calculate the log probability of the document under the model logprob = model.log_probability(document) # calculate the number of parameters of the model n_params = num_topics * (num_topics - 1) + num_topics * vocab_size # calculate the BIC score of the model bic = -2 * logprob + n_params * numpy.log(len(document)) # calculate the AIC score of the model aic = -2 * logprob + 2 * n_params # append the model and its scores to the list models.append((model, bic, aic)) # sort the models by their BIC scores in ascending order models.sort(key=lambda x: x[1]) # get the best model by BIC and its viterbi states best_model_bic, best_bic, _ = models[0] viterbi_states_bic, viterbi_logprob_bic = best_model_bic.viterbi(document) # sort the models by their AIC scores in ascending order models.sort(key=lambda x: x[2]) # get the best model by AIC and its viterbi states best_model_aic, _, best_aic = models[0] viterbi_states_aic, viterbi_logprob_aic = best_model_aic.viterbi(document) # compare the BIC and AIC scores and choose the best one if best_bic < best_aic: # use BIC as the criterion and segment document based on viterbi states segments = [] current_segment = [] current_state = viterbi_states_bic[0][1].name for i in range(1, len(viterbi_states_bic)): word = document[i-1] state = viterbi_states_bic[i][1].name if state == current_state: # add word to current segment current_segment.append(word) else: # start a new segment segments.append((current_state, current_segment)) current_segment = [word] current_state = state # add last segment segments.append((current_state, current_segment)) # return segments and BIC score return segments, best_bic else: # use AIC as the criterion and segment document based on viterbi states segments = [] current_segment = [] current_state = viterbi_states_aic[0][1].name for i in range(1, len(viterbi_states_aic)): word = document[i-1] state = viterbi_states_aic[i][1].name if state == current_state: # add word to current segment current_segment.append(word) else: # start a new segment segments.append((current_state, current_segment)) current_segment = [word] current_state = state # add last segment segments.append((current_state, current_segment)) # return segments and AIC score return segments, best_aic if __name__ == "__main__": # Get the document as input from the user. document = input("Enter a document: ") # Segment the document into distinct segments. segments = segment_document(document) # Print the segments. for segment in segments: print("Segment:", segment)
报错回溯
--------------------------------------------------------------------------- IndexError Traceback (most recent call last) <ipython-input-70-38334fe21de7> in <cell line: 144>() 147 148 # Segment the document into distinct segments. --> 149 segments = segment_document(document) 150 151 # Print the segments. 6 frames <ipython-input-70-38334fe21de7> in segment_document(document) 63 64 #train model using Baum-Welch algorithm --> 65 model.fit([document]) 66 67 # calculate the log probability of the document under the model /usr/local/lib/python3.10/dist-packages/pomegranate/hmm/_base.py in fit(self, X, sample_weight, priors) 604 p_ = None if priors is None else priors[j] 605 --> 606 logp += self.summarize(X_, sample_weight=w_, priors=p_).sum() 607 608 # Calculate and check improvement and optionally print it /usr/local/lib/python3.10/dist-packages/pomegranate/hmm/dense_hmm.py in summarize(self, X, sample_weight, emissions, priors) 541 """ 542 --> 543 X, emissions, sample_weight = super().summarize(X, 544 sample_weight=sample_weight, emissions=emissions, priors=priors) 545 /usr/local/lib/python3.10/dist-packages/pomegranate/hmm/_base.py in summarize(self, X, sample_weight, emissions, priors) 681 X = _check_parameter(_cast_as_tensor(X), "X", ndim=3, 682 shape=(-1, -1, self.d), check_parameter=self.check_data) --> 683 emissions = _check_inputs(self, X, emissions, priors) 684 685 if sample_weight is None: /usr/local/lib/python3.10/dist-packages/pomegranate/hmm/_base.py in _check_inputs(model, X, emissions, priors) 26 ndim=3) 27 if emissions is None: --> 28 emissions = model._emission_matrix(X, priors=priors) 29 30 return emissions /usr/local/lib/python3.10/dist-packages/pomegranate/hmm/_base.py in _emission_matrix(self, X, priors) 285 286 for i, node in enumerate(self.distributions): --> 287 logp = node.log_probability(X) 288 if isinstance(logp, torch.masked.MaskedTensor): 289 logp = logp._masked_data /usr/local/lib/python3.10/dist-packages/pomegranate/distributions/categorical.py in log_probability(self, X) 173 logps = torch.zeros(X.shape[0], dtype=self.probs.dtype) 174 for i in range(self.d): --> 175 logps += self._log_probs[i][X[:, i]] 176 177 return logps IndexError: tensors used as indices must be long, int, byte or bool tensors
问题根源
报错核心是张量类型不匹配：
- LabelEncoder将文本转换为整数类型的索引，但代码中把这个整数数组转成了
float类型的tensor - pomegranate的Categorical分布在计算log概率时，需要用整数类型的tensor作为索引来查找对应类别的概率，float类型无法作为索引使用，因此抛出IndexError。
另外还有两个潜在问题：
- 词汇表提取逻辑错误：
vocab = numpy.unique(numpy.char.split(document))[0]会只取第一个分词后的词汇，应该直接用document.split()提取所有单词后去重。 - 结果还原错误：生成segment时直接保存tensor对象，没有还原为原始文本，导致输出结果不可读。
修复方案
1. 修正张量类型
将document = torch.from_numpy(document).float()改为：
document = torch.from_numpy(document).long()
2. 修正词汇表提取逻辑
将vocab = numpy.unique(numpy.char.split(document))[0]改为：
vocab = numpy.unique(document.split())
3. 还原原始文本
在生成segment的循环中，将word = document[i-1]改为：
word = encoder.inverse_transform([document[i-1].item()])[0]
修改后的完整代码
# import pomegranate library and numpy import pomegranate from pomegranate.hmm import DenseHMM import numpy import torch # import LabelEncoder from sklearn from sklearn.preprocessing import LabelEncoder def segment_document(document): # define the range of possible topics min_topics = 2 max_topics = 20 # initialize a list to store the models and their scores models = [] # create an encoder object encoder = LabelEncoder() # fit the encoder to the vocabulary vocab = numpy.unique(document.split()) encoder.fit(vocab) # transform the document into numeric values doc_numeric = encoder.transform(document.split()) doc_numeric = doc_numeric.reshape(1, -1) # reshape into a 2D array with one row # add another dimension to document for DenseHMM input doc_numeric = numpy.expand_dims(doc_numeric, axis=-1) # convert document to torch tensor of long type (integer indices) doc_tensor = torch.from_numpy(doc_numeric).long() vocab_size = len(vocab) # loop over the number of topics for num_topics in range(min_topics, max_topics + 1): # initialize state priors and transition probabilities state_priors = numpy.random.dirichlet(numpy.ones(num_topics)) trans_probs = numpy.random.dirichlet(numpy.ones(num_topics), size=num_topics) # initialize emission factors with random or uniform probabilities emission_probs = [] for i in range(num_topics): # create a categorical distribution from random integers and their counts ints = numpy.random.randint(0, vocab_size, size=100) # generate 100 random integers in [0, vocab_size) counts = numpy.bincount(ints) # count the occurrences of each integer # pad counts to vocab_size in case some indices are missing if len(counts) < vocab_size: counts = numpy.pad(counts, (0, vocab_size - len(counts)), mode='constant') probs = counts / counts.sum() # normalize the counts to get probabilities probs = probs.reshape(1, -1) # reshape into a 2D array with one row emission_probs.append(pomegranate.distributions.Categorical(probs)) # create a categorical distribution # create model object model = DenseHMM() # add distributions to model model.add_distributions(emission_probs) # set starts and ends probabilities for model model.starts = state_priors model.ends = numpy.zeros(num_topics) #train model using Baum-Welch algorithm model.fit([doc_tensor]) # calculate the log probability of the document under the model logprob = model.log_probability(doc_tensor) # calculate the number of parameters of the model n_params = num_topics * (num_topics - 1) + num_topics * vocab_size # calculate the BIC score of the model bic = -2 * logprob + n_params * numpy.log(len(doc_numeric.flatten())) # calculate the AIC score of the model aic = -2 * logprob + 2 * n_params # append the model and its scores to the list models.append((model, bic, aic)) # sort the models by their BIC scores in ascending order models.sort(key=lambda x: x[1]) # get the best model by BIC and its viterbi states best_model_bic, best_bic, _ = models[0] viterbi_states_bic, viterbi_logprob_bic = best_model_bic.viterbi(doc_tensor) # sort the models by their AIC scores in ascending order models.sort(key=lambda x: x[2]) # get the best model by AIC and its viterbi states best_model_aic, _, best_aic = models[0] viterbi_states_aic, viterbi_logprob_aic = best_model_aic.viterbi(doc_tensor) # compare the BIC and AIC scores and choose the best one if best_bic < best_aic: # use BIC as the criterion and segment document based on viterbi states segments = [] current_segment = [] current_state = viterbi_states_bic[0][1].name for i in range(1, len(viterbi_states_bic)): word_idx = doc_tensor[i-1].item() word = encoder.inverse_transform([word_idx])[0] state = viterbi_states_bic[i][1].name if state == current_state: # add word to current segment current_segment.append(word) else: # start a new segment segments.append((current_state, ' '.join(current_segment))) current_segment = [word] current_state = state # add last segment segments.append((current_state, ' '.join(current_segment))) # return segments and BIC score return segments, best_bic else: # use AIC as the criterion and segment document based on viterbi states segments = [] current_segment = [] current_state = viterbi_states_aic[0][1].name for i in range(1, len(viterbi_states_aic)): word_idx = doc_tensor[i-1].item() word = encoder.inverse_transform([word_idx])[0] state = viterbi_states_aic[i][1].name if state == current_state: # add word to current segment current_segment.append(word) else: # start a new segment segments.append((current_state, ' '.join(current_segment))) current_segment = [word] current_state = state # add last segment segments.append((current_state, ' '.join(current_segment))) # return segments and AIC score return segments, best_aic if __name__ == "__main__": # Get the document as input from the user. document = input("Enter a document: ") # Segment the document into distinct segments. segments, score = segment_document(document
相关产品推荐
相关产品推荐
