Замовна платформа розмітки даних з Active Learning

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

Анотування даних часто стає вузьким горлечком ML-проєкту. Будуючи платформу для класифікації медичних висновків, замовник розмічав 500 документів на тиждень руками трьох експертів. Після впровадження кастомної платформи з автоматичною передрозміткою та Active Learning пропускна здатність зросла до 3000 документів на тиждень при тих же трьох анотаторах — у 6 разів ефективніше, ніж ручний процес. Економія на оплаті праці склала $12 000 на місяць.

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

Як Active Learning скорочує витрати на розмітку?

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

Чому кастомна платформа краща за типові рішення?

Контроль якості без ручних перевірок. Типова ситуація: два анотатори розмічають один текст, але розходяться в 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']] если 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 кастомна

Приклад розрахунку економії на команді з 5 анотаторів

Без платформи: 5 анотаторів × 40 годин/тиждень × $15/година = $3,000/тиждень. За місяць — $12,000. З Active Learning: обсяг розмітки скорочується в 3–5 разів, трудовитрати падають до $150–$300 на 1K документів. Підсумкова економія — $4,000–$8,000/міс.

Як впровадити платформу розмітки?

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 тижнів залежно від складності. Вартість розраховується індивідуально після аудиту ваших даних та вимог.

Обговоріть ваш проєкт з інженерами — ми оцінимо дані та запропонуємо архітектуру. Напишіть нам, щоб отримати консультацію.