7个颠覆认知的文本聚类优化指南：从理论到实战的效果提升策略

from bertopic import BERTopic from sklearn.cluster import HDBSCAN import numpy as np def adaptive_density_clustering(embeddings): # 动态计算聚类参数 cluster_sizes = [5, 10, 15, 20] best_model = None best_noise_ratio = 1.0 for size in cluster_sizes: hdbscan = HDBSCAN(min_cluster_size=size, min_samples=3) model = BERTopic(hdbscan_model=hdbscan) topics, _ = model.fit_transform(docs) noise_ratio = np.sum(np.array(topics) == -1) / len(topics) # 寻找噪声比例在10%-15%之间的模型 if 0.1 <= noise_ratio <= 0.15 and noise_ratio < best_noise_ratio: best_model = model best_noise_ratio = noise_ratio return best_model

from bertopic.vectorizers import ClassTfidfTransformer from sentence_transformers import SentenceTransformer import numpy as np class SemanticNoiseReducer(ClassTfidfTransformer): def __init__(self, model_name="all-MiniLM-L6-v2", **kwargs): super().__init__(**kwargs) self.embedding_model = SentenceTransformer(model_name) def transform(self, X): ctfidf = super().transform(X) words = self.vectorizer.get_feature_names_out() enhanced = [] for topic in ctfidf: word_weights = dict(zip(words, topic.toarray()[0])) sorted_words = sorted(word_weights.items(), key=lambda x: x[1], reverse=True)[:30] # 过滤语义相似词 embeddings = self.embedding_model.encode([w for w, _ in sorted_words]) selected = [] for i, (word, _) in enumerate(sorted_words): if all(np.inner(embeddings[i], embeddings[j]) < 0.7 for j in selected): selected.append(i) if len(selected) >= 10: break enhanced.append([sorted_words[i][0] for i in selected]) return enhanced

from bertopic.representation import ZeroShotClassification def context_aware_naming(docs, candidate_labels): # 使用零样本分类生成有意义的主题名称 zero_shot_model = ZeroShotClassification( model="facebook/bart-large-mnli", candidate_labels=candidate_labels, multi_label=True ) topic_model = BERTopic(representation_model=zero_shot_model) topics, _ = topic_model.fit_transform(docs) return topic_model

from bertopic import BERTopic import numpy as np def hierarchical_incremental_learning(doc_batches): # 初始化模型 topic_model = BERTopic(verbose=True) # 处理第一批文档 topics, _ = topic_model.fit_transform(doc_batches[0]) # 增量处理后续批次 for batch in doc_batches[1:]: topics, _ = topic_model.partial_fit(batch) # 定期合并相似主题 if len(topic_model.get_topic_info()) > 50: topic_model.merge_topics(batch, topics_to_merge="similar") return topic_model

from sklearn.metrics import adjusted_rand_score import pandas as pd 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 = [] # 滑动窗口分析 for i in range(0, len(df), window_size//2): window_docs = df["doc"].iloc[i:i+window_size].tolist() topics, _ = topic_model.fit_transform(window_docs) if i > 0: prev_topics = prev_model.transform(window_docs)[0] aris.append(adjusted_rand_score(topics, prev_topics)) prev_model = topic_model return {"aris": aris, "model": topic_model}

def domain_adaptive_topic_model(docs, domain_type): # 根据领域类型选择不同参数 if domain_type == "news": # 新闻文本：长文本，主题多样 embedding_model = "all-mpnet-base-v2" min_cluster_size = 15 n_gram_range = (1, 3) elif domain_type == "ecommerce": # 电商评论：短文本，情感丰富 embedding_model = "all-MiniLM-L6-v2" min_cluster_size = 20 n_gram_range = (1, 2) elif domain_type == "scientific": # 科研文献：专业术语多 embedding_model = "allenai-specter" min_cluster_size = 10 n_gram_range = (2, 4) return BERTopic( embedding_model=embedding_model, min_cluster_size=min_cluster_size, n_gram_range=n_gram_range )