BERTopic优化实战：5个行业验证的进阶技巧

from bertopic import BERTopic from umap import UMAP from hdbscan import HDBSCAN import numpy as np def create_optimal_reduction_pipeline(sample_size): """根据样本量动态调整降维参数""" # 小样本数据（<10k）采用激进降维 if sample_size < 10000: umap_model = UMAP( n_neighbors=15, # 减少邻居数，增强局部结构识别 n_components=10, # 较低维度，强化主题分离 min_dist=0.1, # 允许点之间更紧密，适合小样本 metric='cosine' ) hdbscan_model = HDBSCAN( min_cluster_size=15, # 减小聚类大小阈值 min_samples=5, cluster_selection_method='leaf' # 更敏感的聚类方式 ) # 中等样本数据（10k-100k）平衡降维 elif 10000 <= sample_size < 100000: umap_model = UMAP( n_neighbors=30, n_components=15, min_dist=0.05, metric='cosine' ) hdbscan_model = HDBSCAN( min_cluster_size=30, min_samples=10, cluster_selection_method='eom' ) # 大样本数据（>100k）保守降维 else: umap_model = UMAP( n_neighbors=50, n_components=20, min_dist=0.01, metric='cosine' ) hdbscan_model = HDBSCAN( min_cluster_size=50, min_samples=20, cluster_selection_method='eom' ) return umap_model, hdbscan_model # 使用示例 # docs = ["用户评论1", "用户评论2", ...] # umap_model, hdbscan_model = create_optimal_reduction_pipeline(len(docs)) # topic_model = BERTopic(umap_model=umap_model, hdbscan_model=hdbscan_model) # topics, _ = topic_model.fit_transform(docs)

from bertopic.vectorizers import ClassTfidfTransformer from sentence_transformers import SentenceTransformer from sklearn.feature_extraction.text import CountVectorizer import numpy as np from gensim.summarization.bm25 import BM25Okapi class SemanticWeightedCTFIDF(ClassTfidfTransformer): def __init__(self, model_name="all-MiniLM-L6-v2", top_n=10, bm25_weight=True, semantic_threshold=0.75,** kwargs): super().__init__(**kwargs) self.embedding_model = SentenceTransformer(model_name) self.top_n = top_n self.bm25_weight = bm25_weight self.semantic_threshold = semantic_threshold def fit(self, X, y=None): super().fit(X) if self.bm25_weight: # 训练BM25模型 tokenized_corpus = [doc.split() for doc in X] self.bm25 = BM25Okapi(tokenized_corpus) return self 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])) # 如果启用BM25加权 if self.bm25_weight: topic_docs = [doc for doc, topic in zip(X, self.labels_) if topic == topic_idx] if topic_docs: tokenized_docs = [doc.split() for doc in topic_docs] bm25 = BM25Okapi(tokenized_docs) avg_idf = sum(bm25.idf.values()) / len(bm25.idf) # 合并BM25权重 for word in word_weights: if word in bm25.idf: word_weights[word] *= (bm25.idf[word] / avg_idf) # 按权重排序 sorted_words = sorted(word_weights.items(), key=lambda x: x[1], reverse=True)[:30] # 语义过滤 if len(sorted_words) >= self.top_n: 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] < self.semantic_threshold 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]) else: enhanced_keywords.append([word for word, _ in sorted_words[:self.top_n]]) return enhanced_keywords # 使用示例 # vectorizer_model = CountVectorizer(stop_words="english") # ctfidf_model = SemanticWeightedCTFIDF(bm25_weight=True, semantic_threshold=0.7) # topic_model = BERTopic(vectorizer_model=vectorizer_model, ctfidf_model=ctfidf_model)

from bertopic.representation import ZeroShotClassification, KeyBERTInspired from bertopic import BERTopic from sklearn.ensemble import VotingClassifier from sklearn.linear_model import LogisticRegression import numpy as np def create_semantic_topic_labels(docs, domain_labels, multi_label=True): """ 创建具有业务含义的主题标签 参数: docs: 文档列表 domain_labels: 领域相关的候选标签列表 multi_label: 是否允许一个主题对应多个标签 """ # 定义多种标签生成模型 zero_shot_model = ZeroShotClassification( model="facebook/bart-large-mnli", candidate_labels=domain_labels, multi_label=multi_label ) keybert_model = KeyBERTInspired() # 创建BERTopic模型 topic_model = BERTopic( representation_model=[zero_shot_model, keybert_model], verbose=True ) # 训练模型 topics, probs = topic_model.fit_transform(docs) # 融合多个模型的结果 def custom_label_generator(topic_model, topic): # 获取零样本分类结果 zero_shot_labels = topic_model.get_topic(topic)[0] if topic != -1 else ["噪声"] # 获取KeyBERT关键词 keybert_keywords = [word for word, _ in topic_model.get_topic(topic)[:3]] if topic != -1 else ["无意义"] # 合并结果 if topic == -1: return "噪声主题" return f"{zero_shot_labels}: {'-'.join(keybert_keywords)}" # 应用自定义标签生成器 topic_model.set_topic_labels(custom_label_generator) return topic_model # 使用示例 # 电商评论领域标签 # domain_labels = ["产品质量", "物流服务", "价格问题", "售后服务", "用户体验", "包装设计"] # topic_model = create_semantic_topic_labels(comments, domain_labels) # topic_info = topic_model.get_topic_info() # print(topic_info[["Topic", "Name"]])

from bertopic import BERTopic from sentence_transformers import SentenceTransformer import numpy as np import pandas as pd from tqdm import tqdm import gc def memory_efficient_topic_modeling(documents, batch_size=5000, embedding_model="all-MiniLM-L6-v2", min_cluster_size=20, merge_threshold=0.9): """ 内存高效的大规模文档主题建模 参数: documents: 文档列表或DataFrame batch_size: 每批处理的文档数量 embedding_model: 嵌入模型名称 min_cluster_size: 最小聚类大小 merge_threshold: 主题合并阈值 """ # 如果是DataFrame，提取文档列 if isinstance(documents, pd.DataFrame): docs = documents[documents.columns[0]].tolist() else: docs = documents # 初始化嵌入模型 embedder = SentenceTransformer(embedding_model) # 分批次处理文档 num_batches = (len(docs) + batch_size - 1) // batch_size embeddings = [] print(f"分{num_batches}批处理文档...") for i in tqdm(range(num_batches)): start = i * batch_size end = min((i+1) * batch_size, len(docs)) batch_docs = docs[start:end] # 计算嵌入 batch_embeddings = embedder.encode(batch_docs, show_progress_bar=False) embeddings.append(batch_embeddings) # 释放内存 del batch_docs, batch_embeddings gc.collect() # 合并嵌入 embeddings = np.vstack(embeddings) # 初始化模型 topic_model = BERTopic( min_cluster_size=min_cluster_size, verbose=True ) # 分阶段拟合 print("开始增量主题建模...") for i in tqdm(range(num_batches)): start = i * batch_size end = min((i+1) * batch_size, len(docs)) batch_docs = docs[start:end] batch_embeddings = embeddings[start:end] if i == 0: # 第一批文档完全拟合 topics, probs = topic_model.fit_transform(batch_docs, batch_embeddings) else: # 后续批次增量拟合 topics, probs = topic_model.partial_fit(batch_docs, batch_embeddings) # 定期合并相似主题 if i > 0 and i % 5 == 0: topic_model.merge_topics(batch_docs, topics_to_merge="similar", merge_threshold=merge_threshold) # 释放内存 del batch_docs, batch_embeddings, topics, probs gc.collect() return topic_model # 使用示例 # 加载大规模文档数据 # df = pd.read_csv("large_document_corpus.csv") # topic_model = memory_efficient_topic_modeling(df, batch_size=10000)

from bertopic import BERTopic import numpy as np import pandas as pd from sklearn.metrics import adjusted_rand_score, normalized_mutual_info_score import matplotlib.pyplot as plt from tqdm import tqdm def topic_stability_analysis(docs, timestamps, window_sizes=[500, 1000, 2000], n_runs=5, random_seeds=[42, 123, 456, 789, 1011]): """ 主题稳定性多维度评估 参数: docs: 文档列表 timestamps: 对应的时间戳列表 window_sizes: 滑动窗口大小列表 n_runs: 重复运行次数 random_seeds: 随机种子列表 """ # 按时间排序 df = pd.DataFrame({"doc": docs, "timestamp": timestamps}).sort_values("timestamp") results = {"ari_scores": [], "nmi_scores": [], "topic_counts": []} # 1. 时间稳定性分析（滑动窗口） print("进行时间稳定性分析...") for window_size in window_sizes: ari_scores = [] prev_topics = None for i in tqdm(range(0, len(df), window_size//2)): end = min(i + window_size, len(df)) window_docs = df["doc"].iloc[i:end].tolist() # 训练模型 topic_model = BERTopic(random_state=42, verbose=False) topics, _ = topic_model.fit_transform(window_docs) # 计算与前一窗口的ARI分数 if prev_topics is not None: # 确保两个窗口有相同数量的文档 min_len = min(len(topics), len(prev_topics)) ari = adjusted_rand_score(topics[:min_len], prev_topics[:min_len]) ari_scores.append(ari) prev_topics = topics results["ari_scores"].append({ "window_size": window_size, "mean_ari": np.mean(ari_scores), "std_ari": np.std(ari_scores) }) # 2. 随机种子稳定性分析 print("进行随机种子稳定性分析...") seed_ari = [] seed_nmi = [] topic_counts = [] for seed in random_seeds: topic_model = BERTopic(random_state=seed, verbose=False) topics, _ = topic_model.fit_transform(df["doc"].tolist()) topic_counts.append(len(set(topics))) if seed != random_seeds[0]: # 与第一个种子的结果比较 base_model = BERTopic(random_state=random_seeds[0], verbose=False) base_topics, _ = base_model.fit_transform(df["doc"].tolist()) ari = adjusted_rand_score(topics, base_topics) nmi = normalized_mutual_info_score(topics, base_topics) seed_ari.append(ari) seed_nmi.append(nmi) results["nmi_scores"] = { "mean_nmi": np.mean(seed_nmi), "std_nmi": np.std(seed_nmi) } results["topic_counts"] = { "mean_count": np.mean(topic_counts), "std_count": np.std(topic_counts), "counts": topic_counts } # 3. 可视化结果 plt.figure(figsize=(12, 5)) # ARI分数图 plt.subplot(1, 2, 1) window_sizes = [res["window_size"] for res in results["ari_scores"]] mean_aris = [res["mean_ari"] for res in results["ari_scores"]] plt.bar([str(w) for w in window_sizes], mean_aris) plt.title("不同窗口大小的主题稳定性 (ARI分数)") plt.ylabel("平均ARI分数") plt.xlabel("窗口大小") plt.axhline(y=0.5, color='r', linestyle='--', label='稳定性阈值') plt.legend() # 主题数量稳定性 plt.subplot(1, 2, 2) plt.boxplot(results["topic_counts"]["counts"]) plt.title("不同随机种子的主题数量分布") plt.ylabel("主题数量") plt.xticks([1], ["随机种子"]) plt.tight_layout() plt.savefig("topic_stability_analysis.png") plt.close() return results # 使用示例 # 假设docs是文档列表，timestamps是对应的时间戳列表 # stability_results = topic_stability_analysis(docs, timestamps) # print(f"平均NMI分数: {stability_results['nmi_scores']['mean_nmi']:.3f}") # print(f"主题数量变异系数: {stability_results['topic_counts']['std_count']/stability_results['topic_counts']['mean_count']:.3f}")