BERTopic主题模型优化避坑指南：7个反常识技巧提升文本聚类效果

from bertopic import BERTopic from sklearn.cluster import HDBSCAN import numpy as np def find_optimal_clusters(embeddings, min_cluster_sizes=[5, 10, 15, 20]): """通过多组参数测试找到最优聚类结果""" results = [] for size in min_cluster_sizes: hdbscan_model = HDBSCAN( min_cluster_size=size, min_samples=5, metric='euclidean', cluster_selection_method='eom' ) topic_model = BERTopic( hdbscan_model=hdbscan_model, verbose=True ) topics, _ = topic_model.fit_transform(docs) topic_info = topic_model.get_topic_info() # 计算有效主题比例（排除-1噪声主题） valid_topics = len(topic_info) - 1 # 减1是排除-1主题 noise_ratio = np.sum(np.array(topics) == -1) / len(topics) results.append({ "min_cluster_size": size, "valid_topics": valid_topics, "noise_ratio": noise_ratio, "model": topic_model }) # 选择噪声比例在10%-15%之间的模型 for result in results: if 0.1 <= result["noise_ratio"] <= 0.15: return result["model"] # 如果没有合适的，返回噪声比例最低的 return min(results, key=lambda x: x["noise_ratio"])["model"] # 使用示例 # topic_model = find_optimal_clusters(embeddings)

from bertopic.vectorizers import ClassTfidfTransformer from sentence_transformers import SentenceTransformer import numpy as np class SemanticEnhancedCTFIDF(ClassTfidfTransformer): def __init__(self, model_name="all-MiniLM-L6-v2", top_n=10, **kwargs): super().__init__(**kwargs) self.embedding_model = SentenceTransformer(model_name) self.top_n = top_n def transform(self, X): # 先运行标准c-TF-IDF ctfidf = super().transform(X) # 获取词汇表 words = self.vectorizer.get_feature_names_out() # 为每个主题的关键词进行语义过滤 enhanced_keywords = [] for topic_idx in range(ctfidf.shape[0]): # 获取该主题的词权重 word_weights = dict(zip(words, ctfidf[topic_idx].toarray()[0])) # 按权重排序 sorted_words = sorted(word_weights.items(), key=lambda x: x[1], reverse=True)[:30] # 获取词向量 word_list = [word for word, _ in sorted_words] embeddings = self.embedding_model.encode(word_list) # 计算相似度矩阵 cos_sim = np.inner(embeddings, embeddings) # 过滤语义相似度过高的词 selected = [] for i, (word, weight) in enumerate(sorted_words[:self.top_n*2]): # 如果与已选词的相似度都低于阈值，则保留 if all(cos_sim[i][j] < 0.7 for j in selected): selected.append(i) if len(selected) >= self.top_n: break # 保留选中的关键词 enhanced_keywords.append([sorted_words[i][0] for i in selected]) return enhanced_keywords # 使用示例 # ctfidf_model = SemanticEnhancedCTFIDF(bm25_weighting=True) # topic_model = BERTopic(ctfidf_model=ctfidf_model)

from bertopic.representation import ZeroShotClassification from bertopic import BERTopic def create_interpretable_topics(docs, candidate_labels): """使用零样本分类为主题生成有意义的标签""" # 定义零样本分类器 zero_shot_model = ZeroShotClassification( model="facebook/bart-large-mnli", candidate_labels=candidate_labels, multi_label=True ) # 创建BERTopic模型 topic_model = BERTopic( representation_model=zero_shot_model, verbose=True ) # 训练模型 topics, probs = topic_model.fit_transform(docs) return topic_model # 使用示例 # candidate_labels = ["产品评价", "用户体验", "价格问题", "物流服务", "售后服务"] # topic_model = create_interpretable_topics(comments, candidate_labels)

from bertopic import BERTopic import numpy as np from tqdm import tqdm def incremental_topic_modeling(doc_batches, embedding_model="all-MiniLM-L6-v2"): """增量式主题建模，分批次处理文档""" # 初始化模型 topic_model = BERTopic( embedding_model=embedding_model, verbose=True ) # 处理第一批文档以初始化模型 first_batch = doc_batches[0] topics, probs = topic_model.fit_transform(first_batch) # 增量处理后续批次 for batch in tqdm(doc_batches[1:], desc="Processing batches"): # 部分拟合新文档 topics, probs = topic_model.partial_fit(batch) # 定期合并相似主题 if len(topic_model.get_topic_info()) > 50: # 当主题数超过50时 topic_model.merge_topics(batch, topics_to_merge="similar") return topic_model # 使用示例 # 将文档分成10个批次 # doc_batches = np.array_split(comments, 10) # topic_model = incremental_topic_modeling(doc_batches)

from bertopic import BERTopic from bertopic.dimensionality import BaseDimensionalityReduction import numpy as np import pandas as pd from sklearn.metrics import adjusted_rand_score def topic_stability_analysis(docs, timestamps, window_size=1000): """通过滑动窗口分析主题稳定性""" # 按时间排序 df = pd.DataFrame({"doc": docs, "timestamp": timestamps}).sort_values("timestamp") # 初始化模型 topic_model = BERTopic(verbose=True) # 存储结果 aris = [] topic_counts = [] # 滑动窗口处理 for i in range(0, len(df), window_size//2): end = min(i + window_size, len(df)) window_docs = df["doc"].iloc[i:end].tolist() # 训练模型 topics, _ = topic_model.fit_transform(window_docs) # 如果不是第一个窗口，计算ARI分数（主题稳定性指标） if i > 0: prev_topics = prev_model.transform(window_docs)[0] ari = adjusted_rand_score(topics, prev_topics) aris.append(ari) # 记录当前模型和主题数量 prev_model = topic_model topic_counts.append(len(set(topics))) return { "aris": aris, # ARI分数越高，主题稳定性越好（0-1） "topic_counts": topic_counts, "model": topic_model } # 使用示例 # 假设comments是文档列表，timestamps是对应的时间戳列表 # results = topic_stability_analysis(comments, timestamps)