Кастомная платформа разметки данных с Active Learning – разработка

Платформа разметки данных: когда типовые решения не подходят

Разметка данных — узкое горлышко любого ML-проекта. Строя платформу для классификации медицинских заключений, заказчик размечал 500 документов в неделю руками трёх экспертов. После внедрения кастомной платформы с предразметкой и Active Learning пропускная способность выросла до 3000 документов в неделю при тех же трёх аннотаторах, а экономия на оплате труда составила $12 000 в месяц — в 6 раз эффективнее ручного процесса.

Типовые решения вроде Label Studio или Supervisely не всегда закрывают специфику: нужна интеграция со своей моделью, нестандартные типы разметки (иерархическая классификация с 10 000+ классов), контроль качества через IAA, предразметка слабыми моделями или замкнутый Active Learning. За 10+ проектов мы набили шишек: от отваливающихся очередей на 50K задач до рассинхронизации аннотаций в реальном времени.

Как Active Learning сокращает затраты на разметку?

В типичном проекте NLP или Computer Vision разметка данных занимает 60-80% времени. Ручной процесс без пайплайнов приводит к трём главным проблемам: дублирование задач (один документ уходит двум аннотаторам без агрегации), простой разметчиков из-за ручного распределения и систематический пропуск сложных примеров — модель обучается на лёгких кейсах, а на боевых данных падает. Платформа решает это единым API: ингрест → предразметка → очередь → контроль → экспорт → Active Learning. Пропускная способность вырастает в 3-5 раз при том же числе людей. При стоимости часа аннотатора ~$15 это даёт экономию $4000–$8000 в месяц на команду из 5 человек.

Почему кастомная платформа лучше типовых решений?

Контроль качества без ручных перепроверок. Типичная ситуация: два аннотатора размечают один текст, но расходятся в 30% случаев. Без IAA вы не знаете, кто прав. Мы внедряем Cohen's Kappa (классификация) и F1-согласованность (NER), автоматически отправляем на ревью спорные задачи. Порог качества настраивается под проект — обычно 0.8-0.85.

Предразметка снижает трудозатраты на 40-70%. Используем слабые модели: zero-shot NLI от Facebook (bart-large-mnli) для классификации или GLiNER для NER. Если уверенность предсказания выше 0.85 — задача автоматически принимается, аннотатор только подтверждает. Наши тесты на датасете из 10K документов показали: 60% задач проходят авто-валидацию с точностью 97%.

Active Learning — модель сама выбирает, что размечать. Стратегия uncertainty: выбираем примеры с максимальной энтропией предсказаний. Это даёт прирост качества модели на 5-10% по сравнению со случайной выборкой. Для production используем гибрид: 70% неопределённость + 30% разнообразие (core-set), чтобы не застревать на похожих примерах. Подробнее о Active Learning можно прочитать в Wikipedia.

Архитектура платформы

[Raw Data Sources] ↓ [Ingestion & Preprocessing] ← конвертация форматов, deduplification
↓ [Pre-annotation (weak models)] ← экономия 40-70%
↓ [Task Queue Management] ← распределение
↓ [Annotation Interface] ← Label Studio / custom UI
↓ [Quality Control] ← IAA, gold standard
↓ [Export & Model Training] ← JSONL, COCO, YOLO
↓ [Active Learning Loop] ← сложные примеры

Ключевые модули платформы

Управление задачами и аннотаторами

from anthropic import Anthropic
import pandas as pd
from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime
import uuid
import numpy as np

class TaskStatus(Enum):
    PENDING = "pending"
    PRE_ANNOTATED = "pre_annotated"
    IN_REVIEW = "in_review"
    COMPLETED = "completed"
    DISPUTED = "disputed"

@dataclass
class AnnotationTask:
    task_id: str
    data: dict  # raw data (text, image_url, etc.)
    task_type: str  # classification, ner, segmentation
    annotations: list = field(default_factory=list)
    pre_annotation: dict = None
    status: TaskStatus = TaskStatus.PENDING
    assigned_to: list = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)
    difficulty_score: float = 0.5

class AnnotationPlatform:
    def __init__(self, db_connection):
        self.db = db_connection
        self.llm = Anthropic()
        self.quality_threshold = 0.8  # Minimum IAA
        self.annotators_per_task = 2

    def ingest_data(self, raw_data: list[dict], task_type: str) -> list[AnnotationTask]:
        """Прием данных и создание задач"""
        tasks = []
        for item in raw_data:
            task = AnnotationTask(
                task_id=str(uuid.uuid4()),
                data=item,
                task_type=task_type
            )
            tasks.append(task)
        # Предварительная оценка сложности
        tasks = self._estimate_difficulty(tasks)
        # Приоритизация: сначала лёгкие для быстрого старта
        tasks.sort(key=lambda t: t.difficulty_score)
        return tasks

    def _estimate_difficulty(self, tasks: list[AnnotationTask]) -> list[AnnotationTask]:
        """LLM-оценка сложности задач для приоритизации"""
        # Батч-оценка через LLM
        sample_texts = [t.data.get('text', '')[:200] for t in tasks[:20]]
        if not any(sample_texts):
            return tasks
        text_list = "\n".join([f"{i+1}. {t}" for i, t in enumerate(sample_texts)])
        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""Rate the annotation difficulty of these texts (0-1, where 1 is hardest). Consider: ambiguity, domain specificity, length complexity. Texts: {text_list} Return only comma-separated scores, e.g.: 0.3, 0.7, 0.5..."""
            }]
        )
        try:
            scores = [float(s.strip()) for s in response.content[0].text.split(',')]
            for i, task in enumerate(tasks[:len(scores)]):
                task.difficulty_score = scores[i]
        except Exception:
            pass
        return tasks

Контроль качества через IAA

def compute_iaa(self, annotations: list[dict], task_type: str) -> float:
    """ Inter-Annotator Agreement:
    - Classification: Cohen's Kappa
    - NER: F1 agreement
    - Segmentation: IoU agreement
    """
    if len(annotations) < 2:
        return 1.0
    if task_type == 'classification':
        return self._cohen_kappa(annotations)
    elif task_type == 'ner':
        return self._ner_agreement(annotations)
    else:
        return self._pairwise_agreement(annotations)

def _cohen_kappa(self, annotations: list[dict]) -> float:
    """Cohen's Kappa для классификации"""
    from sklearn.metrics import cohen_kappa_score
    if len(annotations) == 2:
        labels_a = [a['label'] for a in annotations[0]['items']]
        labels_b = [a['label'] for a in annotations[1]['items']]
        if len(labels_a) != len(labels_b):
            return 0.0
        try:
            return cohen_kappa_score(labels_a, labels_b)
        except Exception:
            return 0.0
    return 0.5 # Default для >2 аннотаторов (нужен Fleiss kappa)

def _ner_agreement(self, annotations: list[dict]) -> float:
    """F1 agreement для именованных сущностей"""
    if len(annotations) < 2:
        return 1.0
    spans_a = set(
        (e['start'], e['end'], e['label']) for e in annotations[0].get('entities', [])
    )
    spans_b = set(
        (e['start'], e['end'], e['label']) for e in annotations[1].get('entities', [])
    )
    if not spans_a and not spans_b:
        return 1.0
    intersection = spans_a & spans_b
    if not intersection:
        return 0.0
    precision = len(intersection) / len(spans_b)
    recall = len(intersection) / len(spans_a)
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0
    return f1

def review_disputed_task(self, task: AnnotationTask, annotations: list[dict]) -> dict:
    """Разбор спорных случаев через LLM"""
    response = self.llm.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=400,
        messages=[{
            "role": "user",
            "content": f"""You are a senior annotation expert. Resolve this labeling dispute.
Task type: {task.task_type}
Text: {task.data.get('text', '')[:500]}
Annotator A: {annotations[0]}
Annotator B: {annotations[1]}
Provide:
1. Correct annotation
2. Brief reasoning (1-2 sentences)
3. Guideline clarification needed (if any)"""
        }]
    )
    return {
        'resolution': response.content[0].text,
        'resolved_by': 'llm_arbitration',
        'task_id': task.task_id
    }

Автоматическая предразметка

class PreAnnotationEngine:
    """Предразметка для снижения нагрузки аннотаторов"""

    def __init__(self, task_type: str):
        self.task_type = task_type
        self.weak_model = None
        self.confidence_threshold = 0.85  # Только высококонфидентные принять без ревью

    def pre_annotate_classification(self, texts: list[str], labels: list[str]) -> list[dict]:
        """Zero-shot классификация через NLI"""
        from transformers import pipeline

        if self.weak_model is None:
            self.weak_model = pipeline(
                "zero-shot-classification",
                model="facebook/bart-large-mnli",
                device=0
            )

        results = []
        batch_size = 32
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            preds = self.weak_model(batch, candidate_labels=labels, batch_size=batch_size)
            for pred in preds:
                top_label = pred['labels'][0]
                confidence = pred['scores'][0]
                results.append({
                    'label': top_label,
                    'confidence': confidence,
                    'auto_accepted': confidence >= self.confidence_threshold
                })
        return results

    def pre_annotate_ner(self, texts: list[str]) -> list[dict]:
        """NER через GLiNER (general NER)"""
        from gliner import GLiNER

        if self.weak_model is None:
            self.weak_model = GLiNER.from_pretrained("urchade/gliner_multi-v2.1")

        entity_types = ["person", "organization", "location", "date", "product"]
        results = []
        for text in texts:
            entities = self.weak_model.predict_entities(text, entity_types)
            results.append({
                'entities': [
                    {'start': e['start'], 'end': e['end'], 'label': e['label'], 'confidence': e['score']}
                    for e in entities
                ],
                'auto_accepted': all(e['score'] >= self.confidence_threshold for e in entities)
            })
        return results

Active Learning цикл

class ActiveLearningLoop:
    """Умный выбор следующих задач для разметки"""

    def select_informative_samples(self, unlabeled_pool: list[dict], current_model, strategy: str = 'uncertainty', budget: int = 100) -> list[int]:
        """
        Стратегии:
        - uncertainty: наименее уверенные предсказания
        - diversity: наиболее разнообразные по feature space
        - hybrid: комбинация обеих
        """
        texts = [item.get('text', '') for item in unlabeled_pool]

        if strategy == 'uncertainty':
            probs = current_model.predict_proba(texts)
            # Наибольшая энтропия = наибольшая неопределённость
            entropy = -np.sum(probs * np.log(probs + 1e-10), axis=1)
            return np.argsort(entropy)[-budget:].tolist()

        elif strategy == 'diversity':
            # Core-set: максимально разнообразные примеры
            embeddings = current_model.encode(texts)  # если есть encoder
            selected = [np.random.randint(len(texts))]
            for _ in range(budget - 1):
                dists = np.min(
                    np.linalg.norm(
                        embeddings[:, None] - embeddings[selected], axis=2
                    ),
                    axis=1
                )
                selected.append(np.argmax(dists))
            return selected

        return list(range(min(budget, len(unlabeled_pool))))

Сравнение стратегий разметки

Сравнение платформ: типовые vs кастомная

Как внедрить платформу разметки?

1. Аудит данных и типов разметки — определяем допустимые типы, сложность, частоту ошибок.
2. Выбор стека и архитектуры — решаем, какие компоненты кастомизировать (Label Studio или с нуля), какие модели предразметки использовать.
3. Разработка бэкенда — FastAPI + Celery + RabbitMQ для очередей, PostgreSQL для хранения.
4. Интеграция предразметки и Active Learning — подключаем weak модели и стратегию неопределённости.
5. Настройка контроля качества — пороги IAA, золотой стандарт, LLM-арбитраж.
6. Тестирование с реальными данными — нагрузочное тестирование очередей, проверка согласованности.
7. Деплой и обучение команды — разворачиваем на вашей инфраструктуре, передаём документацию.

Что входит в разработку и сроки

• API оркестрации — приём данных, очередь, приоритизация, распределение.
• Интерфейс аннотатора — кастомизированный Label Studio или React UI.
• Модуль предразметки — слабые модели с порогами уверенности.
• Контроль качества — IAA, золотой стандарт, пайплайн ревью.
• Экспорт — JSONL, COCO, YOLO, интеграция с HuggingFace Datasets.
• Active Learning — вычислитель неопределённости и разнообразия.
• Документация и обучение команды.

Базовая платформа на основе Label Studio — от 2 недель. Полнофункциональная с предразметкой и Active Learning — от 3 до 8 недель в зависимости от сложности. Стоимость рассчитывается индивидуально после аудита ваших данных и требований.

Обсудите ваш проект с инженерами — мы оценим данные и предложим архитектуру. Напишите нам, чтобы получить консультацию.