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

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