mcpField
A
ai-annotation-workflow

ai-annotation-workflow

github.com/dengineproblem数据与数据库66 次安装
代码仓库

关于 ai-annotation-workflow

面向机器学习数据团队的 AI 标注工作流技能，涵盖流程设计、质量度量和标注人员管理。还介绍 Label Studio 自动化及主动学习样本筛选。

主要功能

  1. 01标注流程规划与迭代
  2. 02多种数据标注任务指南
  3. 03标注者一致性与质量度量
  4. 04Label Studio 项目自动化
  5. 05主动学习样本选择
  6. 06标注人员资格与表现跟踪

适用场景

  1. 01设计数据标注指南并开展试点
  2. 02比较标注者之间的一致性
  3. 03按黄金标准监控标注质量
  4. 04自动创建 Label Studio 标注项目

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/dengineproblem/agents-monorepo --skill ai-annotation-workflow

安装后在 Claude Code 中输入/ai-annotation-workflow确认已加载。

详细介绍

AI 标注工作流专家

机器学习数据标注流程的设计与管理专家。

核心原则

数据标注周期
  1. 规划：定义任务，创建指南
  2. 试点：使用小样本进行测试
  3. 规模化：扩展到完整数据集
  4. 质量控制：持续监控
  5. 迭代：根据反馈进行改进
标注任务类型
  • 分类：数据分类
  • 对象检测：边界框
  • 分割：像素级标注
  • NER：命名实体识别
  • 情感分析：情感倾向分析
  • 转录：将音频转为文本

创建指南

文档结构
# Annotation Guidelines v1.0

## Задача
Разметка изображений товаров для e-commerce

## Категории
1. Одежда
   - Верхняя одежда
   - Нижняя одежда
   - Аксессуары

2. Электроника
   - Смартфоны
   - Ноутбуки
   - Аксессуары

## Правила разметки
- Выбирайте наиболее специфичную категорию
- При неопределённости используйте "Другое"
- Один объект = одна категория

## Примеры
[Примеры с изображениями и правильной разметкой]

## Edge Cases
- Товар частично виден: размечайте если >50% видно
- Несколько товаров: размечайте каждый отдельно

质量指标

标注者间一致性（IAA）
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
    }
质量指标
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

自动化工作流

Label Studio 集成
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)
主动学习流水线
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)

标注人员管理

入职培训
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
            ]
        }
绩效监控
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：开源，配置灵活
  • Labelbox：面向企业，支持机器学习辅助标注
  • Scale AI：托管式人力团队
  • Amazon SageMaker Ground Truth：集成 AWS
  • Prodigy：专注 NLP，支持主动学习
工具选择
标准Label StudioLabelboxScale AI
成本免费/开源$$$$$
人力团队自行管理可选包含
机器学习辅助基础高级高级
自定义能力高中低

最佳实践

  1. 从试点开始 — 使用 100-200 个样本进行校准
  2. 迭代更新指南 — 发现边缘情况时及时更新
  3. 使用黄金标准 — 将 5-10% 的数据用于质量控制
  4. 平衡速度与质量 — 不要给标注人员施加压力
  5. 记录决策 — 记录所有说明
  6. 尽可能实现自动化 — 预标注、验证规则

评论

支持 Markdown