Я участвую в разработке продукта, одна из фич которго – классификация текстов документов по их содержимому. Например, входящий поток требуется разделить на разные папки: отедилить мух от котлет договоры от счетов-фактур.

Кратко одну из последних задач можно описать следующим образом: мы поставляем “в коробке” классификатор с пятью стандартными классами документов, а клиент хочет иметь возможность разделять поток документов на шесть классов: пять наших и один свой.

В этой статье я покажу один из возможных вариантов решения этой задачи на примере классического набора данных 20 newsgroup dataset:

  1. Создадам новый классификатор новостей из категорий “rec.sport.baseball”, “talk.politics.guns” и “comp.sys.ibm.pc.hardware” (далее, модель_1)
  2. Расширю этот классификатор таким образом, чтобы он мог определять новости по теме “sci.electronics” (далее, модель_2).

Считаем, что модель_1 поставляется “в коробке”, а клиенту требуется помимо новостей про бейсбол, оружие и PC иметь возможность классифицировать новости про электронику.

Есть несколько вариантов решения этой проблемы.

Клиент передаёт нам статьи про электронику

Сейчас это работает именно так, но это не масштабируется – отвлекаться на каждого клиента нет ни времени, ни желания.

Мы передаём клиенту статьи про бейсбол, оружие и PC

Считаем, что эти статьи конфицециальные и передавать их кому-либо в открытом виде нам запрещает NDA (с реальными документами так и есть).

Шифровать статьи про бейсбол, оружие и PC и передавать их клиенту

Считаем, что у нас есть быстрый алгоритм шифрования (хэширования) текстов. Здесь возможны два варианта.

Расшифровать тексты в процессе работы приложения и обучить на расшифрованных данных и данных клиента классификатор

Здесь основная пробема в том, что в памяти эти статьи будут в незашифрованном виде и в теории их можно получить в виде простого текста – не годится.

Зашифровать статьи клиента тем же алгоритмом, как наш набор данных и обучать классификатор на зашифрованных данных

Этот подход выглядит неплохо, но имея большой набор данных и некоторое количество времени можно сгенерировать достаточно много хешей и расшифровать датасет.

Обучать классификатор на сгенерированных на существующем классификаторе текстах

Здесь так же возможны два варианта с использованием векторизатора (в моём случае TfidfVectorizer) и классификатора (SGDClassifier).

Генерация текстов для каждого класса в существующей модели и добавление к ним новых текстов для нового класса

Упрощённый алгоритм выглядит следующим образом:

train_samples = []

for document_class in classifier:
    words := get_specific_words_from_vectorizer_and_classifier()
    samples := generate_N_samples_for_class()
    train_samples.append(samples)
train_samples.append(new_samples)

model.fit(train_samples)

Проверил этот вариант и получил F1-меру ниже, чем при обучении на “реальных” данных. Но в этом случае нет необходимости передавать чужие данные клиентам. В принципе, этот вариант вполне неплох.

Генерация текстов из данных модели и обучить бинарный классификатор: существующие тексты vs новые тексты

В этом случае потребуется создать копию существующего классификатора и переписать некоторые свойства:

train_samples := []
words := get_all_words_from_vectorizer_and_model
samples := generate_samples_for_existing_classes_and_save_them_as_an_other_class()
train_samples.append(samples)
train_samples.append(new_samples_with_new_class_name)
new_model.fit(train_samples)
clone_model := copy(model)
clone_model.update_params_from_model(new_model)

Далее в этой статье будет описание этого варианта.

Генерация текстов из существующей модели и обучение бинарного классификатора: существующий класс vs новый класс

Дальше буду работать с набором 20-newsgroup из scikit-learn. Для начала несколько импортов:

from typing import List, Tuple

from sklearn.linear_model import SGDClassifier
from sklearn.datasets import fetch_20newsgroups
from sklearn.metrics import roc_auc_score, classification_report, plot_confusion_matrix
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.base import clone as clone_model

import pandas as pd
import numpy as np
from matplotlib import pyplot as plt

from stop_words import get_stop_words

np.random.seed = 0

Требуется скачать и подготовить данные:

twenty_newsgroup = fetch_20newsgroups('./', subset='all')
subj_id_subj_title = dict(zip(range(len(twenty_newsgroup.target_names)), twenty_newsgroup.target_names))

twenty_newsgroup_df = pd.DataFrame()
twenty_newsgroup_df['Text'] = twenty_newsgroup.data
twenty_newsgroup_df['Target'] = twenty_newsgroup.target

# заменим ИД классов на их имена
twenty_newsgroup_df['Target'] = twenty_newsgroup_df['Target'].replace(subj_id_subj_title)

# оставим только интересные классы
source_cat = ['rec.sport.baseball', 'talk.politics.guns', 'comp.sys.ibm.pc.hardware']
added_cat = 'sci.electronics'

filtered = twenty_newsgroup_df.copy()
filtered = filtered[filtered['Target'].isin(source_cat + added_cat)]

# подготовим исходных и "дополнительный" наборы данных
initial_df = filtered[filtered['Target'].isin(source_cat)]
additional_df = filtered[filtered['Target'].isin(added_cat)]

initial_train, initial_test = train_test_split(initial_df, test_size=0.3, random_state=0)
additional_train, additional_test = train_test_split(additional_df, test_size=0.3, random_state=0)

На этот момент есть четыре набора данных: исходный набор разделённый на два (для обучения и для тестирования) и “дополнительный” набор разделённый на два (для обучения и для тестирования).

Создади пайплайн с TfidfVectorizer и SGDClassifier:

vectorizer = TfidfVectorizer(
    use_idf=True, 
    min_df=10,
    max_features=100000,
    ngram_range=(1, 3),
    stop_words=get_stop_words('en'),
    norm='l2')
clf = SGDClassifier(loss='modified_huber', alpha=0.0001, penalty='l2', max_iter=500, random_state=0)

initial_pipeline = Pipeline([
        ('vect', vectorizer),
        ('clf', clf)
    ])

initial_pipeline.fit(initial_train['Text'], initial_train['Target'])

print('Classification report:')
print(classification_report(initial_pipeline.predict(initial_test['Text']), initial_test['Target']))

print('\nROC-AUC Score: {}'.format(roc_auc_score(
            initial_test['Target'], 
            initial_pipeline.predict_proba(initial_test['Text']), 
            multi_class='ovo')))

Выхлоп кода выше будет выглядеть примерно так:

Classification report:
                          precision    recall  f1-score   support

comp.sys.ibm.pc.hardware       0.99      0.98      0.99       294
      rec.sport.baseball       0.98      0.98      0.98       308
      talk.politics.guns       0.98      0.99      0.98       264

                accuracy                           0.98       866
               macro avg       0.98      0.99      0.99       866
            weighted avg       0.99      0.98      0.98       866


ROC-AUC Score: 0.9994522360572858

И теперь переходим к главной части статьи: добавим новый класс к существующему классификатору:

def generate_texts(tokens: List[str]) -> List[str]:
    texts = [' '.join(x) for x in np.random.choice(tokens, (2000, 100))]
    return texts

def extend_classifer(new_df: pd.DataFrame, initial_pipline: Pipeline) -> Pipeline:
    initial_vect = initial_pipeline.named_steps['vect']
    initial_model = initial_pipeline.named_steps['clf']
    
    # генерация новых примеров для существующих классов из обученного ранее TfidfVectorizer
    generated_texts = generate_texts(initial_vect.get_feature_names())
    
    # Создадим новый SGDClassifier с параметрами как в классификаторе оригинального набора
    clf = SGDClassifier(loss='modified_huber', alpha=0.0001, penalty='l2', max_iter=500, random_state=0)
    
    # Создадим новый набор данных из сгенерированных текстов и добавим к нему новые тексты
    # Все "старые" классы помечаем классом "Other", т.к. классификатор будет бинарным.
    df = pd.DataFrame({'Text': generated_texts, 'Target': ['Other'] * len(generated_texts)})
    new_df = pd.concat([new_df, df])
    
    # Преобразуем тексты используя обученный TfidfVectorizer
    X = initial_vect.transform(new_df['Text'])
    y = new_df['Target']

    clf.fit(X, y)
    
    # Создаём копию обученной модели
    new_model = clone_model(initial_model)
    
    # Обновим атрибуты копии исходого классификатора на новые значения
    new_model.classes_ = np.append(initial_model.classes_, clf.classes_[1])
    new_model.coef_ = np.append(initial_model.coef_, clf.coef_, axis=0)
    new_model.intercept_ = np.append(initial_model.intercept_, clf.intercept_)
    
    # Вернём новый обученный пайплайн с классификатором, который содержит новый класс.
    return Pipeline([
            ('vect', initial_vect),
            ('clf', new_model)
        ])

Теперь необходимо проверить полученный пайплайн:

p = extend_classifer(additional_train, initial_pipeline)

test = pd.concat([initial_test, additional_test])
print('Classification report:')
print(classification_report(p.predict(test['Text']), test['Target']))

print('\nROC-AUC Score: {}'.format(roc_auc_score(
            test['Target'],
            p.predict_proba(test['Text']),
            multi_class='ovo')))

Приблизительный выхлоп:

Classification report:
                          precision    recall  f1-score   support

comp.sys.ibm.pc.hardware       0.78      0.87      0.82       261
      rec.sport.baseball       0.83      1.00      0.91       258
         sci.electronics       0.88      0.64      0.74       409
      talk.politics.guns       0.88      1.00      0.93       234

                accuracy                           0.84      1162
               macro avg       0.84      0.88      0.85      1162
            weighted avg       0.85      0.84      0.84      1162


ROC-AUC Score: 0.6682616497756867

Результаты не очень?

Меня это слегка удивило, т.к. на реальных данных результат был значительно выше: F1-мера отличалась от “эталонного” классификатора приблизительно на 1-2%. Так что перед тем как отвергать это решение имеет смысл попробовать на своих данных.

Jupyter Notebook со всем кодом можно найти на gihub.