Кастомная платформа разметки данных с 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 недель в зависимости от сложности. Стоимость рассчитывается индивидуально после аудита ваших данных и требований.

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

Data Engineering для ML: пайплайны, разметка и качество данных

«У нас много данных» — фраза, которая на деле часто означает «у нас много сырых логов в S3, которые никто не трогал два года». Перед тем как обучить модель, нужно понять, что вообще есть: какова структура, есть ли дубли, как часто меняется схема, насколько репрезентативна выборка.

Data Engineering для ML — не просто ETL. Это построение воспроизводимой инфраструктуры данных, которая делает обучение моделей надёжным, а переобучение — предсказуемым. По опыту нашей команды (8 лет в дата-инжиниринге, более 30 проектов в ML) каждая вторая проблема в продакшене связана не с архитектурой модели, а с качеством данных.

ETЛ-пайплайны для ML: чем отличаются от BI

ETL для аналитики и ETL для ML — разные задачи. В аналитике важна агрегация, в ML — индивидуальные записи с историей. В аналитике train/val/test split не нужен, в ML — критичен. В аналитике skew данных мешает интерпретации, в ML — напрямую влияет на качество модели.

Инструменты. Apache Spark (Wikipedia) для больших объёмов (10GB+): PySpark с DataFrames, оптимизации через partitioning и caching. dbt для трансформаций поверх DWH (Snowflake, BigQuery, Redshift) — декларативно, версионируется, тестируется. Pandas + Polars для объёмов до нескольких GB — Polars в 5-10x быстрее Pandas на типичных трансформациях.

Temporal splits. Для ML важно, что split по времени, а не случайный. Если данные временные (транзакции, события пользователей), случайный split даёт data leakage: модель видит «будущие» данные при обучении. Правило: train на периоде T1-T2, validation на T2-T3 (с gap для предотвращения leakage), test на T3-T4. Неправильный split может стоить 10–15% качества модели на валидации.

Инкрементальные пайплайны. Модель переобучается еженедельно на новых данных. Нужен пайплайн, который инкрементально добавляет новые записи к обучающей выборке, не перегружая всё с нуля. Delta Lake или Apache Iceberg — форматы с ACID-транзакциями, Change Data Capture, time travel.

Как избежать training-serving skew с помощью Feature Store

Feature Store решает проблему рассинхронизации между обучением и инференсом. Самая коварная ошибка в ML-инфраструктуре — training-serving skew: признак считается по-разному в обучении и в продакшене. Модель учится на «правильных» данных, а инференс получает другие.

Feast (open source) — офлайн store на Parquet/Delta в S3 для обучения, онлайн store на Redis для low-latency инференса (<10ms). Feature definitions как Python-код:

from feast import FeatureView, Field
from feast.types import Float32, Int64

user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    schema=[
        Field(name="purchase_count_7d", dtype=Int64),
        Field(name="avg_session_duration", dtype=Float32),
    ],
    ttl=timedelta(days=7),
    source=user_features_source,
)

Один definition, используется везде. Нет расхождений.

Потоковые признаки. Когда признак должен обновляться в реальном времени (количество транзакций за последние 10 минут), нужна потоковая обработка. Apache Kafka + Apache Flink или Kafka Streams для вычисления признаков в реальном времени → запись в онлайн store. Сложнее, дороже, нужно только когда staleness признаков критична для качества.

Разметка данных: как не потратить бюджет впустую

Разметка — самая трудоёмкая и недооцениваемая часть ML-проекта. Плохо размеченные данные не исправит никакая архитектура.

Label Studio — open source, поддерживает разметку изображений (bounding box, polygon, segmentation), текста (NER, классификация), аудио, видео. Поднимается за 10 минут через Docker. Для небольших команд — первый выбор.

Оценка качества разметки. Inter-annotator agreement — насколько согласны разметчики между собой. Cohen's Kappa > 0.8 — хорошо, 0.6-0.8 — приемлемо, < 0.6 — задача неоднозначна или инструкция плохая. Пересечение разметок (10-20% примеров размечают два независимых аннотатора) — обязательная практика.

Active learning. Не размечать случайные примеры, а выбирать те, на которых модель наиболее неуверена (low confidence, high uncertainty). Позволяет добиться того же качества при 50-70% объёма разметки. Modals, Prodigy, Label Studio поддерживают active learning workflows. На одном из проектов для NLP мы сократили бюджет на разметку в 2,5 раза за счёт active learning.

Синтетические данные. Когда реальных данных мало или получить их дорого. Для CV: рендеринг в Blender/Unity с реалистичными текстурами (domain randomization). Для NLP: parafrase через LLM, backtranslation. Риск: модель обучается на distribution синтетических данных, а не реальных — нужна осторожность и проверка на реальном holdout.

Качество данных: валидация и мониторинг

Great Expectations — de facto стандарт для data validation в ML-пайплайнах. Expectations — это декларативные утверждения о данных: «колонка age содержит значения от 0 до 120», «колонка user_id не содержит null», «распределение amount не отклоняется более чем на 20% от baseline». Запускается в пайплайне, при провале — блокирует прохождение.

Pandera — Pythonic alternative для pandas/polars DataFrames. Schema-based validation с type hints:

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int, nullable=False),
    "score": pa.Column(float, pa.Check.between(0, 1)),
    "label": pa.Column(str, pa.Check.isin(["positive", "negative", "neutral"])),
})

Data freshness. Модель ожидает данные за последние N дней. ETL упал, данные не обновились — модель использует устаревшие признаки. Мониторинг свежести данных: timestamp последней записи в каждой таблице, алерт при задержке > порога.

Дедупликация. Дубликаты в обучающей выборке завышают метрики (одни и те же примеры в train и val) и искажают веса модели. MinHash LSH для приближённой дедупликации больших датасетов. Для точной — хэш по нормализованному контенту.

Инструменты валидации: сравнение

Хранилища и форматы

Для большинства ML-проектов на старте: Parquet в S3 + DVC для версионирования. Delta Lake или Iceberg — когда появляется потребность в инкрементальных обновлениях или time travel.

Что входит в проект по дата-инжинирингу для ML

Мы предоставляем полный цикл:

• Аудит существующих данных и пайплайнов (1 неделя).
• Проектирование архитектуры: выбор инструментов, форматов, способов разметки.
• Реализация ETL/ELT пайплайна с валидацией и мониторингом.
• Документация кода и процессов (model card, data card).
• Обучение вашей команды работе с пайплайном.
• SLA на сопровождение и поддержку.

Как мы строим пайплайн: пошагово

1. Аудит существующих данных. Профилирование: ydata-profiling (бывший pandas-profiling) генерирует HTML-репорт со статистиками, дистрибуциями, корреляциями, missing values за минуты.
2. Проектирование пайплайна. Определяем источники данных, частоту обновления, требования к latency признаков, объёмы.
3. Реализация и тестирование. Unit-тесты на трансформации, integration-тесты на пайплайн, data validation через Great Expectations.
4. Деплой и мониторинг. Алерты на freshness, quality checks, аномалии в объёмах данных.

Почему стоит доверить это нам

Мы занимаемся дата-инжинирингом и ML с 2016 года. За это время реализовали более 40 проектов — от построения пайплайнов для NLP-моделей до разметки датасетов для компьютерного зрения. Гарантируем воспроизводимость пайплайнов и полную прозрачность процессов. В каждом проекте используем инструменты с открытым исходным кодом, чтобы вы не были привязаны к вендору.

Свяжитесь с нами для бесплатного аудита ваших данных — оценим текущий пайплайн и предложим roadmap. Закажите построение ML-пайплайна под ключ.