Реализация Active Learning для оптимизации разметки

Реализация Active Learning для оптимизации разметки

Разметка данных — самая дорогая часть ML-проекта. Active Learning позволяет модели самостоятельно выбирать, какие примеры размечать дальше, фокусируясь на наиболее неопределённых или информативных. Результат: 5-10× снижение стоимости разметки при той же точности.

Стратегии выбора примеров

Uncertainty Sampling — классика:

import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.base import BaseEstimator

class UncertaintySampler:
    def __init__(self, model: BaseEstimator, strategy='entropy'):
        self.model = model
        self.strategy = strategy

    def query(self, X_unlabeled: np.ndarray, n_instances: int = 10) -> np.ndarray:
        """
        Выбираем n наиболее неопределённых примеров для разметки.
        """
        proba = self.model.predict_proba(X_unlabeled)

        if self.strategy == 'entropy':
            # Максимальная энтропия — модель максимально неуверена
            scores = -np.sum(proba * np.log(proba + 1e-10), axis=1)

        elif self.strategy == 'margin':
            # Минимальный margin между топ-2 классами
            sorted_proba = np.sort(proba, axis=1)
            scores = 1 - (sorted_proba[:, -1] - sorted_proba[:, -2])

        elif self.strategy == 'least_confident':
            # Максимальная вероятность наиболее вероятного класса = минимальная уверенность
            scores = 1 - proba.max(axis=1)

        # Индексы n наиболее неопределённых примеров
        return np.argsort(scores)[-n_instances:]

Query by Committee — несогласие ансамбля:

from sklearn.base import clone

class CommitteeSampler:
    def __init__(self, base_estimator, n_members=5):
        self.committee = [clone(base_estimator) for _ in range(n_members)]

    def fit_committee(self, X_labeled: np.ndarray, y_labeled: np.ndarray):
        """
        Каждый член комитета обучается на bootstrap-выборке
        """
        n = len(X_labeled)
        for member in self.committee:
            bootstrap_idx = np.random.choice(n, n, replace=True)
            member.fit(X_labeled[bootstrap_idx], y_labeled[bootstrap_idx])

    def query(self, X_unlabeled: np.ndarray, n_instances: int = 10) -> np.ndarray:
        """
        Несогласие = Vote Entropy: чем больше членов расходятся — тем ценнее пример
        """
        predictions = np.array([
            member.predict(X_unlabeled) for member in self.committee
        ])  # (n_members, n_samples)

        vote_entropy = []
        for sample_idx in range(X_unlabeled.shape[0]):
            votes = predictions[:, sample_idx]
            unique, counts = np.unique(votes, return_counts=True)
            probs = counts / len(votes)
            entropy = -np.sum(probs * np.log(probs + 1e-10))
            vote_entropy.append(entropy)

        return np.argsort(vote_entropy)[-n_instances:]

Core-Set Sampling для разнообразия

Геометрическое покрытие признакового пространства:

from sklearn.metrics import pairwise_distances

def core_set_selection(X_labeled: np.ndarray,
                        X_unlabeled: np.ndarray,
                        n_instances: int) -> np.ndarray:
    """
    Core-Set: выбираем точки, максимально удалённые от уже размеченных.
    Обеспечивает разнообразие — не выбираем похожие неопределённые примеры.
    """
    selected_indices = []
    labeled_pool = X_labeled.copy()

    for _ in range(n_instances):
        # Расстояния от каждой несмеченной точки до ближайшей помеченной
        distances = pairwise_distances(X_unlabeled, labeled_pool)
        min_distances = distances.min(axis=1)

        # Выбираем точку с максимальным расстоянием до ближайшей помеченной
        best_idx = np.argmax(min_distances)
        selected_indices.append(best_idx)

        # Добавляем выбранную точку в помеченный пул
        labeled_pool = np.vstack([labeled_pool, X_unlabeled[best_idx]])

    return np.array(selected_indices)

Активное обучение для NLP (Sequence Labeling)

Token-level uncertainty для NER:

import torch
from transformers import AutoModelForTokenClassification, AutoTokenizer

def ner_uncertainty_sampling(texts: list,
                               model, tokenizer,
                               n_instances: int = 20) -> list:
    """
    Для NER: неопределённость на уровне токена.
    Aggregation: mean entropy по всем токенам предложения.
    """
    sentence_uncertainties = []

    for i, text in enumerate(texts):
        inputs = tokenizer(text, return_tensors='pt', truncation=True, max_length=512)

        with torch.no_grad():
            outputs = model(**inputs)

        # Softmax вероятности для каждого токена
        probs = torch.softmax(outputs.logits, dim=-1).squeeze()  # (seq_len, n_labels)

        # Энтропия каждого токена
        token_entropy = -(probs * torch.log(probs + 1e-10)).sum(dim=-1)

        # Агрегация: максимальная неопределённость токена в предложении
        sentence_uncertainty = token_entropy.max().item()
        sentence_uncertainties.append((i, sentence_uncertainty))

    # Топ-N наиболее неопределённых предложений
    sentence_uncertainties.sort(key=lambda x: x[1], reverse=True)
    return [idx for idx, _ in sentence_uncertainties[:n_instances]]

Active Learning Loop

Полный цикл с интеграцией в разметчик:

class ActiveLearningPipeline:
    def __init__(self, model, sampler, labeling_budget: int):
        self.model = model
        self.sampler = sampler
        self.budget = labeling_budget
        self.labeled_count = 0
        self.performance_history = []

    def run(self, X_initial: np.ndarray, y_initial: np.ndarray,
             X_pool: np.ndarray, batch_size: int = 20):
        """
        Цикл:
        1. Обучить на размеченных данных
        2. Выбрать наиболее информативные из пула
        3. Отправить на разметку
        4. Добавить к размеченным
        5. Повторить
        """
        X_labeled, y_labeled = X_initial.copy(), y_initial.copy()
        X_unlabeled = X_pool.copy()

        while self.labeled_count < self.budget and len(X_unlabeled) > 0:
            # Обучаем
            self.model.fit(X_labeled, y_labeled)

            # Оцениваем прогресс
            current_metric = self.evaluate(X_labeled, y_labeled)
            self.performance_history.append({
                'n_labeled': len(X_labeled),
                'metric': current_metric
            })

            # Выбираем следующий батч
            query_idx = self.sampler.query(X_unlabeled, n_instances=batch_size)

            # Симуляция разметки (в реальности — интерфейс аннотатора)
            new_y = get_labels_from_annotator(X_unlabeled[query_idx])

            X_labeled = np.vstack([X_labeled, X_unlabeled[query_idx]])
            y_labeled = np.concatenate([y_labeled, new_y])
            X_unlabeled = np.delete(X_unlabeled, query_idx, axis=0)
            self.labeled_count += batch_size

        return self.performance_history

Типичный результат: для задачи классификации текстов с базой 50 000 примеров — Active Learning достигает 90% качества Random Sampling при использовании 15-20% объёма разметки. Интеграция с платформами разметки: Label Studio, Prodigy, Scale AI.

Сроки: Uncertainty sampling + базовый AL loop + Label Studio интеграция — 2-3 недели. Committee sampling, Core-Set, NLP/NER active learning, cold start стратегия — 6-8 недель.