Skip to content

Wiki Analyzer

Not all users have access to an LLM API or can afford to run LLMs on their own hardware. This is why we've added a light-weight analyzer that is retrieval-based, rather than relying on text generation. This allows the topic analyzer to work with the same language model that was used for fitting the topic model.

The WikiAnalyzer works in the following steps:

  1. It searches Wikipedia for articles using the top N keywords from a topic model.
  2. For each topic it produces a topic embedding from the average of top 10 keywords and documents.
  3. It retrieves the most similar articles to each topic.
  4. If the similarity crosses a certain threshold, it assigns the article's name to the topic.
from sklearn.datasets import fetch_20newsgroups

from turftopic import SensTopic
from turftopic.analyzers.wiki import WikiAnalyzer

dataset = fetch_20newsgroups(subset="all", categories=["alt.atheism"])
corpus = dataset.data

t_model = SensTopic(
    random_state=42,
    encode_kwargs=dict(show_progress_bar=True),
    sparsity=5.0,
)
embeddings = t_model.encode_documents(corpus)
t_model.fit(corpus, embeddings=embeddings)

analyzer = WikiAnalyzer(t_model, similarity_threshold=0.3)
t_model.rename_topics(analyzer)
t_model.print_topics()
Topic Name Highest Ranking
0 Omnipotence contradictions, contradiction, creationism, creation, omnipotent, belief, believing, contradictory, deity, believed
1 Capital punishment genocide, punishments, murder, punishment, killing, punish, deaths, executed, kills, penalty
2 Morality morality, morals, moral, morally, ethical, immoral, societal, societally, objectively, justified
3 amusing, responses, discussions, discussing, funny, disclaimer, newsgroups, policy, isn, offensive
4 Agnostic atheism atheism, atheist, atheists, atheistic, agnostics, agnostic, agnosticism, theists, secular, religious
5 Gospel testament, gospel, theological, biblical, bible, verses, revelation, theology, christianity, verses_
6 Quran islamic, muslim, islam, qur, muslims, koran, quran, allah, rushdie, rashid

API Reference

turftopic.analyzers.wiki.WikiAnalyzer

Bases: Analyzer

Analyze topic model with a page titles and summaries from Wikipedia's API. The analyzer searches wikipedia with the highest rankning N keywords from a topic and then ranks pages based on their semantic proximity to example keywords and documents from the topic using the topic model's encoder.

Parameters:

Name Type Description Default
topic_model

Topic model to use for embedding keywords and documents.

 required
language_code str

Wikipedia language code for the language of the documents.

 'en'
n_keywords int

Number of search words to use when searching Wikipedia.

 5
similarity_threshold float

Cosine similarity threshold between page titles and topic representations to consider the page a match.

 0.5
limit int

Maximum number of pages to return in each search.

 10
prune_summaries

Indicates whether only the first sentence should be used from the page summaries.

 True
default

Indicates whether only the first sentence should be used from the page summaries.

 True
Source code in turftopic/analyzers/wiki.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
class WikiAnalyzer(Analyzer):
    """Analyze topic model with a page titles and summaries from Wikipedia's API.
    The analyzer searches wikipedia with the highest rankning N keywords from a topic
    and then ranks pages based on their semantic proximity to example keywords and documents
    from the topic using the topic model's encoder.

    Parameters
    ----------
    topic_model: ContextualModel
        Topic model to use for embedding keywords and documents.
    language_code: str, default "en"
        Wikipedia language code for the language of the documents.
    n_keywords: int, default 5
        Number of search words to use when searching Wikipedia.
    similarity_threshold: float = 0.7
        Cosine similarity threshold between page titles and topic representations
        to consider the page a match.
    limit: int, default 10
        Maximum number of pages to return in each search.
    prune_summaries, default True,
        Indicates whether only the first sentence should be used from the page summaries.
    """

    use_summaries = False

    def __init__(
        self,
        topic_model,
        language_code: str = "en",
        n_keywords: int = 5,
        similarity_threshold: float = 0.5,
        limit: int = 10,
        prune_summaries=True,
        penalize_length=True,
    ):
        import requests

        self.session = requests.Session()
        self.topic_model = topic_model
        self.n_keywords = n_keywords
        self.similarity_threshold = similarity_threshold
        self.limit = limit
        self.prune_summaries = prune_summaries
        self.language_code = language_code
        self.penalize_length = penalize_length

    def summarize_document(self, document: str) -> str:
        raise NotImplementedError

    def generate_text(self, prompt: str) -> str:
        raise NotImplementedError

    def _search_page(self, keywords: list[str]):
        query = " ".join(keywords[: self.n_keywords])
        params = {
            "action": "query",
            "format": "json",
            "list": "search",
            "srsearch": query,
            "srlimit": self.limit,
        }
        results = self.session.get(
            url=URL.format(language_code=self.language_code),
            params=params,
            headers=HEADERS,
        )
        try:
            data = results.json()
            return data["query"]["search"]
        except Exception:
            return []

    def _get_summary(self, pageid):
        params = {
            "action": "query",
            "format": "json",
            "prop": "extracts",
            "explaintext": 1,
            "exsectionformat": "wiki",
            "exintro": 1,
            "pageids": pageid,
        }
        results = self.session.get(
            url=URL.format(language_code=self.language_code),
            params=params,
            headers=HEADERS,
        )
        try:
            data = results.json()
            pages = data["query"]["pages"]
            summary = pages[str(pageid)]["extract"]
            if self.prune_summaries:
                summary = summary.split(".")[0] + "."
            return summary
        except Exception:
            return None

    def _get_topic_embedding(
        self, keywords: list[str], documents: list[str] = None
    ):
        repr_str = list(keywords)
        if documents is not None:
            repr_str.extend(documents)
        embeddings = self.topic_model.encode_documents(repr_str)
        return np.mean(embeddings, axis=0)

    def _get_best_match(
        self, keywords: list[str], documents: list[str] = None
    ):
        search_results = self._search_page(keywords)
        search_results = [
            entry
            for entry in search_results
            if len(remove_parens(entry["title"]).split()) < 5
        ]
        if not search_results:
            return None
        titles = [entry["title"] for entry in search_results]
        snippets = [remove_html(entry["snippet"]) for entry in search_results]
        repr_str = titles
        topic_embedding = self._get_topic_embedding(keywords, documents)
        page_embeddings = self.topic_model.encode_documents(repr_str)
        sim = cosine_similarity([topic_embedding], page_embeddings)[0]
        threshold = self.similarity_threshold
        if self.penalize_length:
            lengths = np.array([len(title.split()) for title in titles])
            sim = sim / lengths
            threshold = threshold / np.max(lengths)
        i_best_page = np.argmax(sim)
        if sim[i_best_page] < self.similarity_threshold:
            return None
        return dict(
            name=remove_parens(titles[i_best_page]),
            snippet=snippets[i_best_page],
            pageid=search_results[i_best_page]["pageid"],
        )

    def describe_topic(
        self,
        keywords: list[str],
        documents=None,
    ):
        """Gives abstract summarization of topic content."""
        best_match = self._get_best_match(keywords)
        if best_match is None:
            return None
        return self._get_summary(best_match["pageid"])

    def name_topic(
        self,
        keywords: list[str],
        documents=None,
    ) -> str:
        """Names one topic based on top descriptive aspects."""
        best_match = self._get_best_match(keywords, documents)
        if best_match is None:
            return None
        return best_match["name"]

    def analyze_topics(
        self,
        keywords: list[list[str]],
        documents: list[list[str]] = None,
        use_summaries=None,
    ) -> AnalysisResults:
        """
        Parameters
        ----------
        keywords: list[list[str]]
            Keywords for each topic.
        documents: list[list[str]], default None
            Top documents for each topic.
        use_summaries: None
            Ignored.

        Returns
        -------
        dict
            Dictionary containing `topic_names`, `topic_descriptions` and `document_summaries` if relevant.
        """
        output = {"topic_names": [], "topic_descriptions": []}
        if documents is None:
            for keys in track(keywords, description="Analyzing topics..."):
                best_match = self._get_best_match(keys)
                if best_match is None:
                    output["topic_names"].append(None)
                    output["topic_descriptions"].append(None)
                    continue
                output["topic_names"].append(best_match["name"])
                summary = self._get_summary(best_match["pageid"])
                output["topic_descriptions"].append(summary)
        else:
            for keys, docs in track(
                zip_longest(keywords, documents),
                description="Analyzing topics...",
                total=len(keywords),
            ):
                best_match = self._get_best_match(keys, docs)
                if best_match is None:
                    output["topic_names"].append(None)
                    output["topic_descriptions"].append(None)
                    continue
                output["topic_names"].append(best_match["name"])
                summary = self._get_summary(best_match["pageid"])
                output["topic_descriptions"].append(summary)
        return AnalysisResults(**output)

analyze_topics(keywords, documents=None, use_summaries=None)

Parameters:

Name Type Description Default
keywords list[list[str]]

Keywords for each topic.

 required
documents list[list[str]]

Top documents for each topic.

 None
use_summaries

Ignored.

 None

Returns:

Type Description
dict

Dictionary containing topic_names, topic_descriptions and document_summaries if relevant.
Source code in turftopic/analyzers/wiki.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def analyze_topics(
    self,
    keywords: list[list[str]],
    documents: list[list[str]] = None,
    use_summaries=None,
) -> AnalysisResults:
    """
    Parameters
    ----------
    keywords: list[list[str]]
        Keywords for each topic.
    documents: list[list[str]], default None
        Top documents for each topic.
    use_summaries: None
        Ignored.

    Returns
    -------
    dict
        Dictionary containing `topic_names`, `topic_descriptions` and `document_summaries` if relevant.
    """
    output = {"topic_names": [], "topic_descriptions": []}
    if documents is None:
        for keys in track(keywords, description="Analyzing topics..."):
            best_match = self._get_best_match(keys)
            if best_match is None:
                output["topic_names"].append(None)
                output["topic_descriptions"].append(None)
                continue
            output["topic_names"].append(best_match["name"])
            summary = self._get_summary(best_match["pageid"])
            output["topic_descriptions"].append(summary)
    else:
        for keys, docs in track(
            zip_longest(keywords, documents),
            description="Analyzing topics...",
            total=len(keywords),
        ):
            best_match = self._get_best_match(keys, docs)
            if best_match is None:
                output["topic_names"].append(None)
                output["topic_descriptions"].append(None)
                continue
            output["topic_names"].append(best_match["name"])
            summary = self._get_summary(best_match["pageid"])
            output["topic_descriptions"].append(summary)
    return AnalysisResults(**output)

describe_topic(keywords, documents=None)

Gives abstract summarization of topic content.

Source code in turftopic/analyzers/wiki.py 
166
167
168
169
170
171
172
173
174
175
def describe_topic(
    self,
    keywords: list[str],
    documents=None,
):
    """Gives abstract summarization of topic content."""
    best_match = self._get_best_match(keywords)
    if best_match is None:
        return None
    return self._get_summary(best_match["pageid"])

name_topic(keywords, documents=None)

Names one topic based on top descriptive aspects.

Source code in turftopic/analyzers/wiki.py 
177
178
179
180
181
182
183
184
185
186
def name_topic(
    self,
    keywords: list[str],
    documents=None,
) -> str:
    """Names one topic based on top descriptive aspects."""
    best_match = self._get_best_match(keywords, documents)
    if best_match is None:
        return None
    return best_match["name"]