--- name: ai-annotation-workflow description: Эксперт по data annotation. Используй для ML labeling, annotation workflows и quality control. --- # AI Annotation Workflow Expert Эксперт по проектированию и управлению процессами разметки данных для машинного обучения. ## Основные принципы ### Цикл разметки данных 1. **Планирование**: Определение задачи, создание guidelines 2. **Пилот**: Тестирование с малой выборкой 3. **Масштабирование**: Расширение на полный датасет 4. **Контроль качества**: Постоянный мониторинг 5. **Итерация**: Улучшение на основе обратной связи ### Типы задач разметки - **Классификация**: Категоризация данных - **Детекция объектов**: Bounding boxes - **Сегментация**: Пиксельная разметка - **NER**: Named Entity Recognition - **Sentiment**: Анализ тональности - **Транскрипция**: Аудио в текст ## Создание Guidelines ### Структура документа ```markdown # Annotation Guidelines v1.0 ## Задача Разметка изображений товаров для e-commerce ## Категории 1. Одежда - Верхняя одежда - Нижняя одежда - Аксессуары 2. Электроника - Смартфоны - Ноутбуки - Аксессуары ## Правила разметки - Выбирайте наиболее специфичную категорию - При неопределённости используйте "Другое" - Один объект = одна категория ## Примеры [Примеры с изображениями и правильной разметкой] ## Edge Cases - Товар частично виден: размечайте если >50% видно - Несколько товаров: размечайте каждый отдельно ``` ## Метрики качества ### Inter-Annotator Agreement (IAA) ```python from sklearn.metrics import cohen_kappa_score, fleiss_kappa def calculate_agreement(annotations_a, annotations_b): """Расчёт согласованности между аннотаторами""" # Cohen's Kappa для двух аннотаторов kappa = cohen_kappa_score(annotations_a, annotations_b) # Интерпретация if kappa < 0.20: interpretation = "Poor" elif kappa < 0.40: interpretation = "Fair" elif kappa < 0.60: interpretation = "Moderate" elif kappa < 0.80: interpretation = "Good" else: interpretation = "Excellent" return { 'kappa': kappa, 'interpretation': interpretation } ``` ### Quality Metrics ```python class AnnotationQualityMonitor: def __init__(self): self.metrics = [] def calculate_metrics(self, annotations, gold_standard): """Расчёт метрик качества относительно эталона""" from sklearn.metrics import precision_score, recall_score, f1_score precision = precision_score(gold_standard, annotations, average='weighted') recall = recall_score(gold_standard, annotations, average='weighted') f1 = f1_score(gold_standard, annotations, average='weighted') return { 'precision': precision, 'recall': recall, 'f1': f1, 'accuracy': sum(a == g for a, g in zip(annotations, gold_standard)) / len(annotations) } def detect_drift(self, window_size=100): """Обнаружение дрифта качества""" if len(self.metrics) < window_size * 2: return False recent = self.metrics[-window_size:] previous = self.metrics[-window_size*2:-window_size] recent_avg = sum(m['f1'] for m in recent) / len(recent) previous_avg = sum(m['f1'] for m in previous) / len(previous) # Дрифт если падение > 5% return (previous_avg - recent_avg) / previous_avg > 0.05 ``` ## Workflow автоматизации ### Label Studio интеграция ```python from label_studio_sdk import Client class AnnotationPipeline: def __init__(self, api_key, url): self.client = Client(url=url, api_key=api_key) def create_project(self, name, label_config): """Создание проекта разметки""" project = self.client.create_project( title=name, label_config=label_config ) return project def import_tasks(self, project_id, data): """Импорт задач для разметки""" project = self.client.get_project(project_id) project.import_tasks(data) def export_annotations(self, project_id, format='JSON'): """Экспорт готовых аннотаций""" project = self.client.get_project(project_id) return project.export_tasks(format=format) ``` ### Active Learning Pipeline ```python class ActiveLearningAnnotation: def __init__(self, model, unlabeled_pool): self.model = model self.unlabeled_pool = unlabeled_pool self.labeled_data = [] def select_samples_for_annotation(self, n_samples=100, strategy='uncertainty'): """Выбор образцов для разметки""" if strategy == 'uncertainty': # Выбор образцов с наибольшей неопределённостью predictions = self.model.predict_proba(self.unlabeled_pool) uncertainties = -np.sum(predictions * np.log(predictions + 1e-10), axis=1) indices = np.argsort(uncertainties)[-n_samples:] elif strategy == 'diversity': # Выбор разнообразных образцов from sklearn.cluster import KMeans kmeans = KMeans(n_clusters=n_samples) kmeans.fit(self.unlabeled_pool) indices = [ np.argmin(np.linalg.norm(self.unlabeled_pool - center, axis=1)) for center in kmeans.cluster_centers_ ] return self.unlabeled_pool[indices] def update_model(self, new_annotations): """Обновление модели после получения аннотаций""" self.labeled_data.extend(new_annotations) X = [item['features'] for item in self.labeled_data] y = [item['label'] for item in self.labeled_data] self.model.fit(X, y) ``` ## Управление аннотаторами ### Онбординг ```python class AnnotatorOnboarding: def __init__(self, gold_standard_samples): self.gold_standard = gold_standard_samples self.passing_threshold = 0.85 def run_qualification_test(self, annotator_id, annotations): """Квалификационный тест для новых аннотаторов""" correct = sum( a == g for a, g in zip(annotations, self.gold_standard) ) accuracy = correct / len(self.gold_standard) return { 'annotator_id': annotator_id, 'accuracy': accuracy, 'passed': accuracy >= self.passing_threshold, 'errors': [ {'index': i, 'expected': g, 'actual': a} for i, (a, g) in enumerate(zip(annotations, self.gold_standard)) if a != g ] } ``` ### Мониторинг производительности ```python class AnnotatorPerformanceTracker: def __init__(self): self.annotator_stats = {} def log_annotation(self, annotator_id, task_id, time_spent, quality_score): if annotator_id not in self.annotator_stats: self.annotator_stats[annotator_id] = [] self.annotator_stats[annotator_id].append({ 'task_id': task_id, 'time_spent': time_spent, 'quality_score': quality_score, 'timestamp': datetime.now() }) def get_annotator_report(self, annotator_id): stats = self.annotator_stats.get(annotator_id, []) if not stats: return None return { 'total_tasks': len(stats), 'avg_time': sum(s['time_spent'] for s in stats) / len(stats), 'avg_quality': sum(s['quality_score'] for s in stats) / len(stats), 'tasks_per_hour': len(stats) / ( (stats[-1]['timestamp'] - stats[0]['timestamp']).total_seconds() / 3600 ) if len(stats) > 1 else 0 } ``` ## Инструменты разметки ### Популярные платформы - **Label Studio**: Open-source, гибкая конфигурация - **Labelbox**: Enterprise, ML-assisted labeling - **Scale AI**: Managed workforce - **Amazon SageMaker Ground Truth**: AWS интеграция - **Prodigy**: NLP-focused, active learning ### Выбор инструмента | Критерий | Label Studio | Labelbox | Scale AI | |----------|--------------|----------|----------| | Стоимость | Free/Open | $$ | $$$ | | Workforce | Self-managed | Optional | Included | | ML Assist | Basic | Advanced | Advanced | | Customization | High | Medium | Low | ## Лучшие практики 1. **Начинайте с пилота** — 100-200 образцов для калибровки 2. **Итерируйте guidelines** — обновляйте по мере обнаружения edge cases 3. **Используйте gold standard** — 5-10% данных для контроля качества 4. **Балансируйте скорость и качество** — не давите на аннотаторов 5. **Документируйте решения** — записывайте все разъяснения 6. **Автоматизируйте что можно** — pre-labeling, validation rules