Кластерный анализ корпуса текстов

Иногда возникает необходимость провести анализ большого количества текстовых данных, не имея представления о содержании текстов. В таком случае можно попытаться разбить тексты на кластеры, и сгенерировать описание каждого кластера. Таким образом можно в первом приближении сделать выводы о содержании текстов.

Тестовые данные

В качестве тестовых данных был взят фрагмент новостного датасета от РИА, из которого в обработке участвовали только заголовки новостей.

Получение эмбеддингов

Для векторизации текста использовалась модель LaBSE от @cointegrated. Модель доступна на huggingface.

Код векторизации

import numpy as np
import torch
from transformers import AutoTokenizer, AutoModel
tokenizer = AutoTokenizer.from_pretrained("cointegrated/LaBSE-en-ru")
model = AutoModel.from_pretrained("cointegrated/LaBSE-en-ru")

sentenses = ['мама мыла раму']

embeddings_list = []

for s in sentences:
    encoded_input = tokenizer(s, padding=True, truncation=True, max_length=64, return_tensors='pt')
    with torch.no_grad():
        model_output = model(**encoded_input)
    embedding = model_output.pooler_output
    embeddings_list.append((embedding)[0].numpy())

embeddings = np.asarray(embeddings_list)

Кластеризация

В качестве алгоритма для кластеризации был выбран алгоритм k-means. Выбран он для наглядности, часто приходится поиграться с данными и алгоритмами для получения адекватных кластеров.

Для нахождения оптимального количества кластеров будем использовать функцию, реализующую "правило локтя":

Функция поиска оптимального количества кластеров:

from sklearn.cluster import KMeans
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error
from sklearn.metrics.pairwise import cosine_similarity

def determine_k(embeddings):
    k_min = 10
    clusters = [x for x in range(2, k_min * 11)]
    metrics = []
    for i in clusters:
        metrics.append((KMeans(n_clusters=i).fit(embeddings)).inertia_)
    k = elbow(k_min, clusters, metrics)
    return k

def elbow(k_min, clusters, metrics):
    score = []

    for i in range(k_min, clusters[-3]):
        y1 = np.array(metrics)[:i + 1]
        y2 = np.array(metrics)[i:]
    
        df1 = pd.DataFrame({'x': clusters[:i + 1], 'y': y1})
        df2 = pd.DataFrame({'x': clusters[i:], 'y': y2})
    
        reg1 = LinearRegression().fit(np.asarray(df1.x).reshape(-1, 1), df1.y)
        reg2 = LinearRegression().fit(np.asarray(df2.x).reshape(-1, 1), df2.y)

        y1_pred = reg1.predict(np.asarray(df1.x).reshape(-1, 1))
        y2_pred = reg2.predict(np.asarray(df2.x).reshape(-1, 1))    
        
        score.append(mean_squared_error(y1, y1_pred) + mean_squared_error(y2, y2_pred))

    return np.argmin(score) + k_min

k = determine_k(embeddings)

Выделение информации о полученных кластерах

После кластеризации текстов берем для каждого кластера по несколько текстов, расположенных максимально близко от центра кластера.

Функция поиска близких к центру кластера текстов:

from sklearn.metrics.pairwise import euclidean_distances

kmeans = KMeans(n_clusters = k_opt, random_state = 42).fit(embeddings)
kmeans_labels = kmeans.labels_

data = pd.DataFrame()
data['text'] = sentences
data['label'] = kmeans_labels
data['embedding'] = list(embeddings)

kmeans_centers = kmeans.cluster_centers_
top_texts_list = []
for i in range (0, k_opt):
    cluster = data[data['label'] == i]
    embeddings = list(cluster['embedding'])
    texts = list(cluster['text'])
    distances = [euclidean_distances(kmeans_centers[0].reshape(1, -1), e.reshape(1, -1))[0][0] for e in embeddings]
    scores = list(zip(texts, distances))
    top_3 = sorted(scores, key=lambda x: x[1])[:3]
    top_texts = list(zip(*top_3))[0]
    top_texts_list.append(top_texts)

Саммаризация центральных текстов

Полученные центральные тексты можно попробовать слепить в общее описание кластера с помощью модели для саммаризации текста. Я использовал для этого модель ruT5 за авторством @cointegrated. Модель доступна на huggingface.

Код саммаризации:

from transformers import T5ForConditionalGeneration, T5Tokenizer
MODEL_NAME = 'cointegrated/rut5-base-absum'
model = T5ForConditionalGeneration.from_pretrained(MODEL_NAME)
tokenizer = T5Tokenizer.from_pretrained(MODEL_NAME)

def summarize(
    text, n_words=None, compression=None,
    max_length=1000, num_beams=3, do_sample=False, repetition_penalty=10.0, 
    **kwargs
):
    """
    Summarize the text
    The following parameters are mutually exclusive:
    - n_words (int) is an approximate number of words to generate.
    - compression (float) is an approximate length ratio of summary and original text.
    """
    if n_words:
        text = '[{}] '.format(n_words) + text
    elif compression:
        text = '[{0:.1g}] '.format(compression) + text
    # x = tokenizer(text, return_tensors='pt', padding=True).to(model.device)
    x = tokenizer(text, return_tensors='pt', padding=True)
    with torch.inference_mode():
        out = model.generate(
            **x, 
            max_length=max_length, num_beams=num_beams, 
            do_sample=do_sample, repetition_penalty=repetition_penalty, 
            **kwargs
        )
    return tokenizer.decode(out[0], skip_special_tokens=True)

summ_list = []
for top in top_texts_list:
    summ_list.append(summarize(' '.join(list(top))))    

Заключение

Представленный подход работает не на всех доменах - новости тут приятное исключение, и тексты такого типа разделяются достаточно хорошо и обычными методами. А вот с условным твиттером придется повозиться - обилие грамматических ошибок, жаргона, и отсутствие пунктуации могут стать кошмаром для любого аналитика. Замечания исправления и дополнения приветствуются!

Ссылки

С ноутбуком можно поиграться в колабе, ссылка в репозитории на гитхаб.