Custom Data Labeling Platform with Active Learning

Custom Data Labeling Platform: When Off-the-Shelf Solutions Don't Fit

Data labeling is the bottleneck of any ML project. When building a platform for classifying medical reports, one client was manually labeling 500 documents per week with three experts. Our team of experienced engineers has delivered over a dozen custom labeling platforms with guaranteed quality. Our custom platform is 6 times more efficient than manual labeling, saving $12,000 per month on labor costs — a 6x improvement over the manual process.

Off-the-shelf solutions like Label Studio or Supervisely often don't cover specific needs: integration with your own model, non-standard labeling types (hierarchical classification with 10,000+ classes), quality control via IAA, pre-annotation with weak models, or closed-loop Active Learning. Over 10+ projects, we've encountered everything from queues breaking at 50K tasks to real-time annotation sync issues.

How Active Learning Reduces Labeling Costs

In a typical NLP or Computer Vision project, data labeling consumes 60–80% of the time. A manual process without pipelines leads to three main problems: duplicate tasks (one document sent to two annotators without aggregation), annotator idle time due to manual assignment, and systematic omission of hard examples — the model trains on easy cases and fails on production data. Our platform solves this with a unified API: ingest → pre-annotation → queue → quality control → export → adaptive sampling. Throughput increases 3–5x with the same number of people. At an annotator's hourly rate of ~$15, this saves $4,000–$8,000 per month for a 5-person team.

Why a Custom Platform Beats Off-the-Shelf Solutions

Quality control without manual re-checks. A typical scenario: two annotators label the same text but disagree in 30% of cases. Without IAA, you don't know who is correct. We implement Cohen's Kappa (classification) and F1 agreement (NER), automatically sending disputed tasks for review. The quality threshold is configurable per project — typically 0.8–0.85.

Pre-annotation cuts labor costs by 40–70%. We use weak models: zero-shot NLI from Facebook (bart-large-mnli) for classification or GLiNER for NER. If the confidence of the prediction is above 0.85, the task is automatically accepted; the annotator only confirms. Our tests on a 10K document dataset showed that 60% of tasks pass auto-validation with 97% accuracy.

Active Learning — the model chooses what to label. Strategy: uncertainty — select examples with the highest entropy of predictions. This yields a 5–10% improvement in model quality compared to random sampling. For production, we use a hybrid: 70% uncertainty + 30% diversity (core-set) to avoid getting stuck on similar examples. Learn more about Active Learning on Wikipedia.

Platform Architecture

[Raw Data Sources] ↓ [Ingestion & Preprocessing] ← format conversion, deduplication ↓ [Pre-annotation (weak models)] ← saves 40-70% ↓ [Task Queue Management] ← distribution ↓ [Annotation Interface] ← Label Studio / custom UI ↓ [Quality Control] ← IAA, gold standard ↓ [Export & Model Training] ← JSONL, COCO, YOLO ↓ [Active Learning Loop] ← complex examples 

Key Platform Modules

Task and Annotator Management

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]: """Ingest data and create tasks""" tasks = [] for item in raw_data: task = AnnotationTask( task_id=str(uuid.uuid4()), data=item, task_type=task_type ) tasks.append(task) # Pre-estimate difficulty tasks = self._estimate_difficulty(tasks) # Prioritize: easy tasks first for quick start tasks.sort(key=lambda t: t.difficulty_score) return tasks def _estimate_difficulty(self, tasks: list[AnnotationTask]) -> list[AnnotationTask]: """LLM-based difficulty estimation for prioritization""" # Batch evaluation via 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 

Quality Control via 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 for classification""" 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 for >2 annotators (needs Fleiss kappa) def _ner_agreement(self, annotations: list[dict]) -> float: """F1 agreement for named entities""" 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: """Resolve disputed cases via 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 } 

Automatic Pre-annotation

class PreAnnotationEngine: """Pre-annotation to reduce annotator workload""" def __init__(self, task_type: str): self.task_type = task_type self.weak_model = None self.confidence_threshold = 0.85 # Only high-confidence accepted without review def pre_annotate_classification(self, texts: list[str], labels: list[str]) -> list[dict]: """Zero-shot classification via 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 via 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 Loop

class ActiveLearningLoop: """Smart selection of next tasks for labeling""" def select_informative_samples(self, unlabeled_pool: list[dict], current_model, strategy: str = 'uncertainty', budget: int = 100) -> list[int]: """ Strategies: - uncertainty: least confident predictions - diversity: most diverse in feature space - hybrid: combination of both """ texts = [item.get('text', '') for item in unlabeled_pool] if strategy == 'uncertainty': probs = current_model.predict_proba(texts) # Highest entropy = highest uncertainty entropy = -np.sum(probs * np.log(probs + 1e-10), axis=1) return np.argsort(entropy)[-budget:].tolist() elif strategy == 'diversity': # Core-set: maximally diverse examples embeddings = current_model.encode(texts) # if encoder available 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)))) 

Labeling Strategy Comparison

Platform Comparison: Off-the-Shelf vs Custom

How to Implement a Labeling Platform

1. Data and labeling type audit — identify permissible types, complexity, error rates.
2. Stack and architecture selection — decide which components to customize (Label Studio or from scratch), which pre-annotation models to use.
3. Backend development — FastAPI + Celery + RabbitMQ for queues, PostgreSQL for storage.
4. Pre-annotation and Active Learning integration — connect weak models and uncertainty strategy.
5. Quality control setup — IAA thresholds, gold standard, LLM arbitration.
6. Testing with real data — load testing of queues, consistency checks.
7. Deployment and team training — deploy on your infrastructure, hand over documentation.

What's Included in Development and Timeline

• Orchestration API — data ingestion, queue, prioritization, distribution.
• Annotator interface — customized Label Studio or React UI.
• Pre-annotation module — weak models with confidence thresholds.
• Quality control — IAA, gold standard, review pipeline.
• Export — JSONL, COCO, YOLO, integration with HuggingFace Datasets.
• Active Learning — uncertainty and diversity calculator.
• Documentation and team training.

Basic platform based on Label Studio — from 2 weeks. Full-featured with pre-annotation and Active Learning — from 3 to 8 weeks depending on complexity. Pricing is determined individually after auditing your data and requirements.

Discuss your project with our engineers — we'll assess your data and propose an architecture. Contact us to get a consultation.