diff --git a/bertopic/representation/__init__.py b/bertopic/representation/__init__.py index f1502982..46b4171c 100644 --- a/bertopic/representation/__init__.py +++ b/bertopic/representation/__init__.py @@ -1,6 +1,7 @@ from bertopic._utils import NotInstalled from bertopic.representation._cohere import Cohere from bertopic.representation._base import BaseRepresentation +from bertopic.representation._feature_importance import FeatureImportance from bertopic.representation._keybert import KeyBERTInspired from bertopic.representation._mmr import MaximalMarginalRelevance @@ -63,6 +64,7 @@ __all__ = [ "BaseRepresentation", "Cohere", + "FeatureImportance", "KeyBERTInspired", "LangChain", "LiteLLM", diff --git a/bertopic/representation/_feature_importance.py b/bertopic/representation/_feature_importance.py new file mode 100644 index 00000000..cb75a973 --- /dev/null +++ b/bertopic/representation/_feature_importance.py @@ -0,0 +1,199 @@ +"""Feature importance representation model for BERTopic. + +New file: bertopic/representation/_feature_importance.py + +Provides two methods for computing term importance per topic: +- "fighting_words": Bayesian log-odds with Dirichlet priors (Monroe et al. 2008) +- "centroid_distance": Cosine similarity between topic centroid and term embeddings +""" + +from typing import List, Mapping, Tuple + +import numpy as np +import pandas as pd +from scipy.sparse import csr_matrix +from sklearn.metrics.pairwise import cosine_similarity + +from bertopic.representation._base import BaseRepresentation + + +class FeatureImportance(BaseRepresentation): + """Compute alternative feature importance scores for topic terms. + + This representation model re-ranks the top words per topic using either + contrastive (fighting words) or embedding-based (centroid distance) methods. + + Arguments: + method: The importance method to use. Options: + * ``"fighting_words"`` — Bayesian log-odds with Dirichlet priors. + Highlights terms that distinguish a topic from the rest. + * ``"centroid_distance"`` — Cosine similarity between the topic + centroid embedding and term embeddings. Highlights terms that + are semantically central to the topic. + top_n_words: Number of top words to return per topic. + prior: Dirichlet prior for fighting words. Use ``"corpus"`` to derive + priors from corpus word frequencies, or a float for a symmetric prior. + + Examples: + ```python + from bertopic import BERTopic + from bertopic.representation import FeatureImportance + + model = BERTopic( + representation_model={ + "Main": some_model, + "FightingWords": FeatureImportance(method="fighting_words"), + "CentroidDist": FeatureImportance(method="centroid_distance"), + } + ) + topics, probs = model.fit_transform(docs) + model.get_topic(0, aspect="FightingWords") + ``` + """ + + def __init__( + self, + method: str = "fighting_words", + top_n_words: int = 10, + prior: float | str = "corpus", + ): + self.method = method + self.top_n_words = top_n_words + self.prior = prior + + def extract_topics( + self, + topic_model, + documents: pd.DataFrame, + c_tf_idf: csr_matrix, + topics: Mapping[str, List[Tuple[str, float]]], + ) -> Mapping[str, List[Tuple[str, float]]]: + """Extract topics with alternative importance scores. + + Arguments: + topic_model: The fitted BERTopic model. + documents: DataFrame with "Document" and "Topic" columns. + c_tf_idf: The c-TF-IDF matrix for the topics. + topics: Default topic representations from c-TF-IDF. + + Returns: + Updated topic representations with re-ranked words. + """ + # Get feature names from the vectorizer + try: + words = topic_model.vectorizer_model.get_feature_names_out() + except AttributeError: + words = topic_model.vectorizer_model.get_feature_names() + + # Get unique topics (excluding outliers) + unique_topics = sorted([t for t in documents.Topic.unique() if t != -1]) + + if self.method == "fighting_words": + importance = self._fighting_words(topic_model, documents, unique_topics, words) + elif self.method == "centroid_distance": + importance = self._centroid_distance(topic_model, unique_topics, words) + else: + raise ValueError(f"Unknown method: {self.method}. Use 'fighting_words' or 'centroid_distance'.") + + # Convert importance matrix to topic representations + updated_topics = {} + for idx, topic_id in enumerate(unique_topics): + scores = importance[idx] + top_indices = np.argsort(scores)[::-1][: self.top_n_words] + updated_topics[topic_id] = [(words[i], float(scores[i])) for i in top_indices] + + return updated_topics + + def _fighting_words( + self, + topic_model, + documents: pd.DataFrame, + unique_topics: list, + words: np.ndarray, + ) -> np.ndarray: + """Compute fighting words (Bayesian log-odds) importance. + + Arguments: + topic_model: The fitted BERTopic model. + documents: DataFrame with "Document" and "Topic" columns. + unique_topics: Sorted list of non-outlier topic IDs. + words: Feature names from the vectorizer. + + Returns: + Importance matrix of shape (n_topics, vocab_size). + """ + # Build document-term matrix from the documents + clean_docs = topic_model._preprocess_text(documents.Document.values) + doc_term_matrix = topic_model.vectorizer_model.transform(clean_docs) + + n_topics = len(unique_topics) + n_vocab = doc_term_matrix.shape[1] + + # Map document topics to sequential indices + topic_to_idx = {t: i for i, t in enumerate(unique_topics)} + labels = np.array([topic_to_idx.get(t, -1) for t in documents.Topic.to_numpy()]) + + # Compute priors + if self.prior == "corpus": + priors = np.ravel(np.asarray(doc_term_matrix.sum(axis=0))) + else: + priors = np.full(n_vocab, float(self.prior)) + a0 = np.sum(priors) + + components = [] + for i_topic in range(n_topics): + mask = labels == i_topic + topic_freq = np.ravel(np.asarray(doc_term_matrix[mask].sum(axis=0))) + rest_freq = np.ravel(np.asarray(doc_term_matrix[~mask].sum(axis=0))) + n1 = np.sum(topic_freq) + n2 = np.sum(rest_freq) + topic_logodds = np.log((topic_freq + priors) / (n1 + a0 - topic_freq - priors)) + rest_logodds = np.log((rest_freq + priors) / (n2 + a0 - rest_freq - priors)) + delta = topic_logodds - rest_logodds + delta_var = 1 / (topic_freq + priors) + 1 / (rest_freq + priors) + zscore = delta / np.sqrt(delta_var) + components.append(zscore) + + return np.stack(components) + + def _centroid_distance( + self, + topic_model, + unique_topics: list, + words: np.ndarray, + ) -> np.ndarray: + """Compute centroid distance importance. + + Arguments: + topic_model: The fitted BERTopic model. + unique_topics: Sorted list of non-outlier topic IDs. + words: Feature names from the vectorizer. + + Returns: + Importance matrix of shape (n_topics, vocab_size). + """ + if topic_model.embedding_model is None: + raise ValueError( + "centroid_distance requires an embedding model. Pass embedding_model when instantiating BERTopic." + ) + + # Get topic centroids (skip outlier topic if present) + topic_embeddings = topic_model.topic_embeddings_[topic_model._outliers :] + + # Embed vocabulary terms + vocab_embeddings = topic_model.embedding_model.embed_words(list(words)) + if vocab_embeddings is None: + # Fallback: embed as documents + vocab_embeddings = topic_model.embedding_model.embed_documents(list(words)) + + vocab_embeddings = np.array(vocab_embeddings) + + # Compute cosine similarity between topic centroids and term embeddings + n_topics = len(unique_topics) + n_vocab = len(words) + components = np.full((n_topics, n_vocab), np.nan) + valid = np.all(np.isfinite(topic_embeddings), axis=1) + similarities = cosine_similarity(topic_embeddings[valid], vocab_embeddings) + components[valid, :] = similarities + + return components diff --git a/docs/getting_started/multiaspect/multiaspect.md b/docs/getting_started/multiaspect/multiaspect.md index f4c06c63..dd97928a 100644 --- a/docs/getting_started/multiaspect/multiaspect.md +++ b/docs/getting_started/multiaspect/multiaspect.md @@ -44,3 +44,31 @@ After we have fitted our model, we can access all representations with `topic_mo

As you can see, there are a number of different representations for our topics that we can inspect. All aspects are found in `topic_model.topic_aspects_`. + + +## **Feature Importance** + +The `FeatureImportance` representation model re-ranks topic words using statistical +methods rather than c-TF-IDF scores. Currently supports the "fighting words" method +(Monroe et al. 2008), which computes log-odds ratios with informative Dirichlet priors: + +```python +from bertopic.representation import FeatureImportance + +fi = FeatureImportance(method="fighting_words", top_n_words=10) +topic_model = BERTopic(representation_model=fi) +``` + +You can combine it with other models using multi-aspect representations: + +```python +from bertopic.representation import KeyBERTInspired, FeatureImportance + +representation_model = { + "Main": KeyBERTInspired(), + "Fighting Words": FeatureImportance(method="fighting_words"), +} +topic_model = BERTopic(representation_model=representation_model) +``` + +After fitting, access results via `topic_model.topic_aspects_["Fighting Words"]`. diff --git a/tests/test_feature_importance.py b/tests/test_feature_importance.py new file mode 100644 index 00000000..44dd2c43 --- /dev/null +++ b/tests/test_feature_importance.py @@ -0,0 +1,204 @@ +"""Tests for FeatureImportance representation model.""" + +from unittest.mock import MagicMock + +import numpy as np +import pandas as pd +import pytest +from bertopic.representation._feature_importance import FeatureImportance +from scipy.sparse import csr_matrix + + +@pytest.fixture +def mock_topic_model(): + """Create a mock BERTopic model with basic attributes.""" + model = MagicMock() + model._outliers = 1 + model.topic_embeddings_ = np.random.rand(4, 10) # 1 outlier + 3 topics + model.vectorizer_model.get_feature_names_out.return_value = np.array(["alpha", "beta", "gamma", "delta", "epsilon"]) + model._preprocess_text.side_effect = lambda x: x + # Create a sparse document-term matrix + dtm = csr_matrix( + np.array( + [ + [2, 1, 0, 0, 1], + [3, 0, 1, 0, 0], + [0, 0, 2, 1, 0], + [0, 1, 3, 0, 1], + [1, 0, 0, 2, 0], + [0, 0, 0, 3, 1], + ] + ) + ) + model.vectorizer_model.transform.return_value = dtm + # Embedding model for centroid_distance + model.embedding_model.embed_words.return_value = np.random.rand(5, 10) + return model + + +@pytest.fixture +def sample_documents(): + """Sample documents DataFrame.""" + return pd.DataFrame( + { + "Document": ["doc1", "doc2", "doc3", "doc4", "doc5", "doc6"], + "Topic": [0, 0, 1, 1, 2, 2], + } + ) + + +@pytest.fixture +def c_tf_idf(): + """Sample c-TF-IDF matrix.""" + return csr_matrix(np.random.rand(3, 5)) + + +@pytest.fixture +def default_topics(): + """Default topic representations.""" + return { + 0: [("alpha", 0.5), ("beta", 0.3), ("epsilon", 0.2)], + 1: [("gamma", 0.6), ("delta", 0.2), ("beta", 0.1)], + 2: [("delta", 0.7), ("epsilon", 0.2), ("alpha", 0.1)], + } + + +class TestFeatureImportanceInit: + def test_default_params(self): + fi = FeatureImportance() + assert fi.method == "fighting_words" + assert fi.top_n_words == 10 + assert fi.prior == "corpus" + + def test_custom_params(self): + fi = FeatureImportance(method="centroid_distance", top_n_words=5, prior=0.1) + assert fi.method == "centroid_distance" + assert fi.top_n_words == 5 + assert fi.prior == 0.1 + + +class TestFightingWords: + def test_returns_correct_format(self, mock_topic_model, sample_documents, c_tf_idf, default_topics): + fi = FeatureImportance(method="fighting_words", top_n_words=3) + result = fi.extract_topics(mock_topic_model, sample_documents, c_tf_idf, default_topics) + + assert isinstance(result, dict) + assert set(result.keys()) == {0, 1, 2} + for topic_id, words in result.items(): + assert len(words) == 3 + for word, score in words: + assert isinstance(word, str) + assert isinstance(score, float) + + def test_corpus_prior(self, mock_topic_model, sample_documents, c_tf_idf, default_topics): + fi = FeatureImportance(method="fighting_words", top_n_words=3, prior="corpus") + result = fi.extract_topics(mock_topic_model, sample_documents, c_tf_idf, default_topics) + assert len(result) == 3 + + def test_numeric_prior(self, mock_topic_model, sample_documents, c_tf_idf, default_topics): + fi = FeatureImportance(method="fighting_words", top_n_words=3, prior=0.01) + result = fi.extract_topics(mock_topic_model, sample_documents, c_tf_idf, default_topics) + assert len(result) == 3 + + def test_scores_are_zscores(self, mock_topic_model, sample_documents, c_tf_idf, default_topics): + """Fighting words produces z-scores that can be positive or negative.""" + fi = FeatureImportance(method="fighting_words", top_n_words=5) + result = fi.extract_topics(mock_topic_model, sample_documents, c_tf_idf, default_topics) + # Top words should have positive z-scores (distinguishing) + for topic_id, words in result.items(): + top_score = words[0][1] + assert top_score > 0 + + +class TestCentroidDistance: + def test_returns_correct_format(self, mock_topic_model, sample_documents, c_tf_idf, default_topics): + fi = FeatureImportance(method="centroid_distance", top_n_words=3) + result = fi.extract_topics(mock_topic_model, sample_documents, c_tf_idf, default_topics) + + assert isinstance(result, dict) + assert set(result.keys()) == {0, 1, 2} + for topic_id, words in result.items(): + assert len(words) == 3 + + def test_no_embedding_model_raises(self, mock_topic_model, sample_documents, c_tf_idf, default_topics): + mock_topic_model.embedding_model = None + fi = FeatureImportance(method="centroid_distance") + with pytest.raises(ValueError, match="centroid_distance requires an embedding model"): + fi.extract_topics(mock_topic_model, sample_documents, c_tf_idf, default_topics) + + def test_fallback_to_embed_documents(self, mock_topic_model, sample_documents, c_tf_idf, default_topics): + mock_topic_model.embedding_model.embed_words.return_value = None + mock_topic_model.embedding_model.embed_documents.return_value = np.random.rand(5, 10) + fi = FeatureImportance(method="centroid_distance", top_n_words=3) + result = fi.extract_topics(mock_topic_model, sample_documents, c_tf_idf, default_topics) + assert len(result) == 3 + + +class TestInvalidMethod: + def test_unknown_method_raises(self, mock_topic_model, sample_documents, c_tf_idf, default_topics): + fi = FeatureImportance(method="nonexistent") + with pytest.raises(ValueError, match="Unknown method"): + fi.extract_topics(mock_topic_model, sample_documents, c_tf_idf, default_topics) + + +class TestEdgeCases: + def test_single_topic(self, mock_topic_model, c_tf_idf, default_topics): + documents = pd.DataFrame( + { + "Document": ["doc1", "doc2"], + "Topic": [0, 0], + } + ) + dtm = csr_matrix(np.array([[2, 1, 0, 0, 1], [3, 0, 1, 0, 0]])) + mock_topic_model.vectorizer_model.transform.return_value = dtm + + fi = FeatureImportance(method="fighting_words", top_n_words=3) + result = fi.extract_topics(mock_topic_model, documents, c_tf_idf, default_topics) + assert 0 in result + + def test_top_n_words_exceeds_vocab(self, mock_topic_model, sample_documents, c_tf_idf, default_topics): + fi = FeatureImportance(method="fighting_words", top_n_words=100) + result = fi.extract_topics(mock_topic_model, sample_documents, c_tf_idf, default_topics) + # Should return all 5 words (vocab size) + for topic_id, words in result.items(): + assert len(words) == 5 + + +class TestFeatureImportanceIntegration: + """Integration test with a real fitted model.""" + + def test_as_representation_model(self, base_topic_model, documents, document_embeddings, embedding_model): + """FeatureImportance should work as a representation model in the BERTopic pipeline.""" + from bertopic import BERTopic + + fi = FeatureImportance(method="fighting_words", top_n_words=5) + model = BERTopic( + embedding_model=embedding_model, + representation_model={"Main": fi}, + ) + model.umap_model.random_state = 42 + model.hdbscan_model.min_cluster_size = 3 + model.fit(documents, embeddings=document_embeddings) + + # Should produce topic representations + assert hasattr(model, "topic_representations_") + for topic_id, words in model.topic_representations_.items(): + if topic_id != -1: + assert len(words) > 0 + + def test_as_aspect_model(self, base_topic_model, documents, document_embeddings): + """FeatureImportance should integrate with topic_aspects_ via multi-aspect.""" + import copy + + model = copy.deepcopy(base_topic_model) + + fi = FeatureImportance(method="fighting_words", top_n_words=5) + result = fi.extract_topics( + model, + pd.DataFrame({"Document": documents, "Topic": model.topics_}), + model.c_tf_idf_, + model.topic_representations_, + ) + + assert isinstance(result, dict) + assert len(result) > 0