ai-annotation-workflow

仓库: dengineproblem/agents-monorepo
安装量: 49
排名: #15158

安装

npx skills add https://github.com/dengineproblem/agents-monorepo --skill ai-annotation-workflow
AI Annotation Workflow Expert
Эксперт по проектированию и управлению процессами разметки данных для машинного обучения.
Основные принципы
Цикл разметки данных
Планирование
Определение задачи, создание guidelines
Пилот
Тестирование с малой выборкой
Масштабирование
Расширение на полный датасет
Контроль качества
Постоянный мониторинг
Итерация
Улучшение на основе обратной связи
Типы задач разметки
Классификация
Категоризация данных
Детекция объектов
Bounding boxes
Сегментация
Пиксельная разметка
NER
Named Entity Recognition
Sentiment
Анализ тональности
Транскрипция
Аудио в текст Создание Guidelines Структура документа

Annotation Guidelines v1.0

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

Категории 1. Одежда - Верхняя одежда - Нижняя одежда - Аксессуары 2. Электроника - Смартфоны - Ноутбуки - Аксессуары

Правила разметки

Выбирайте наиболее специфичную категорию

При неопределённости используйте "Другое"

Один объект = одна категория

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

Edge Cases

Товар частично виден: размечайте если >50% видно

Несколько товаров: размечайте каждый отдельно Метрики качества Inter-Annotator Agreement (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 } Quality Metrics 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 интеграция 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 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
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 Лучшие практики Начинайте с пилота — 100-200 образцов для калибровки Итерируйте guidelines — обновляйте по мере обнаружения edge cases Используйте gold standard — 5-10% данных для контроля качества Балансируйте скорость и качество — не давите на аннотаторов Документируйте решения — записывайте все разъяснения Автоматизируйте что можно — pre-labeling, validation rules
← 返回排行榜