פלטפורמת תיוג נתונים מותאמת אישית עם למידה אקטיבית

פלטפורמת תיוג נתונים מותאמת אישית: כשפתרונות מדף לא מתאימים

תיוג נתונים הוא צוואר הבקבוק של כל פרויקט למידת מכונה. כשבנינו פלטפורמה לסיווג דוחות רפואיים, לקוח אחד תייג ידנית 500 מסמכים בשבוע עם שלושה מומחים. צוות המהנדסים המנוסה שלנו סיפק יותר מתריסר פלטפורמות תיוג מותאמות אישית עם איכות מובטחת. הפלטפורמה המותאמת שלנו יעילה פי 6 מתיוג ידני, וחוסכת 12,000 דולר בחודש בעלויות עבודה — שיפור של פי 6 לעומת התהליך הידני.

פתרונות מדף כמו Label Studio או Supervisely לרוב לא מכסים צרכים ספציפיים: אינטגרציה עם המודל שלך, סוגי תיוג לא סטנדרטיים (סיווג היררכי עם יותר מ-10,000 מחלקות), בקרת איכות באמצעות IAA, תיוג מקדים עם מודלים חלשים, או למידה פעילה בלולאה סגורה. במעל 10 פרויקטים, נתקלנו בכל דבר מתורים שקרסו ב-50K משימות ועד בעיות סנכרון תיוג בזמן אמת.

איך למידה פעילה מפחיתה עלויות תיוג

בפרויקט NLP או ראייה ממוחשבת טיפוסי, תיוג נתונים צורך 60–80% מהזמן. תהליך ידני ללא צינורות נתונים מוביל לשלוש בעיות עיקריות: משימות כפולות (מסמך אחד שנשלח לשני מתעדים ללא צבירה), זמן השבתה של מתעדים עקב הקצאה ידנית, והשמטה שיטתית של דוגמאות קשות — המודל מתאמן על מקרים קלים ונכשל על נתוני ייצור. הפלטפורמה שלנו פותרת זאת עם API אחיד: קליטה → תיוג מקדים → תור → בקרת איכות → ייצוא → דגימה אדפטיבית. התפוקה גדלה פי 3–5 עם אותו מספר אנשים. בתעריף שעתי של מתעד של כ-15 דולר, זה חוסך 4,000–8,000 דולר בחודש לצוות של 5 אנשים.

למה פלטפורמה מותאמת אישית מנצחת פתרונות מדף

בקרת איכות ללא בדיקות חוזרות ידניות. תרחיש טיפוסי: שני מתעדים מתייגים את אותו טקסט אך חלוקים ב-30% מהמקרים. ללא IAA, אתה לא יודע מי צודק. אנו מיישמים את Cohen's Kappa (סיווג) והסכמת F1 (NER), ושולחים אוטומטית משימות שנויות במחלוקת לבדיקה. סף האיכות ניתן להגדרה לכל פרויקט — בדרך כלל 0.8–0.85.

תיוג מקדים חותך עלויות עבודה ב-40–70%. אנו משתמשים במודלים חלשים: zero-shot NLI של פייסבוק (bart-large-mnli) לסיווג או GLiNER ל-NER. אם הביטחון בתחזית הוא מעל 0.85, המשימה מתקבלת אוטומטית; המתעד רק מאשר. הבדיקות שלנו על מערך נתונים של 10K מסמכים הראו ש-60% מהמשימות עוברות אימות אוטומטי עם דיוק של 97%.

למידה פעילה — המודל בוחר מה לתייג. אסטרטגיה: אי-ודאות — בחירת דוגמאות עם האנטרופיה הגבוהה ביותר של תחזיות. זה מניב שיפור של 5–10% באיכות המודל לעומת דגימה אקראית. לייצור, אנו משתמשים בגישה היברידית: 70% אי-ודאות + 30% גיוון (core-set) כדי להימנע מתקיעה על דוגמאות דומות. למידע נוסף על למידה פעילה בויקיפדיה.

ארכיטקטורת הפלטפורמה

[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']]
        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 для >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

לולאת למידה פעילה

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))))

השוואת אסטרטגיות תיוג

השוואת פלטפורמות: מדף לעומת מותאמת אישית

איך ליישם פלטפורמת תיוג

1. בדיקת נתונים וסוגי תיוג — זיהוי סוגים מותרים, מורכבות, שיעורי שגיאות.
2. בחירת טכנולוגיה וארכיטקטורה — החלטה אילו רכיבים להתאים אישית (Label Studio או מאפס), אילו מודלי תיוג מקדים להשתמש.
3. פיתוח צד שרת — FastAPI + Celery + RabbitMQ לתורים, PostgreSQL לאחסון.
4. אינטגרציית תיוג מקדים ולמידה פעילה — חיבור מודלים חלשים ואסטרטגיית אי-ודאות.
5. הגדרת בקרת איכות — ספי IAA, תקן זהב, בוררות LLM.
6. בדיקות עם נתונים אמיתיים — בדיקות עומס לתורים, בדיקות עקביות.
7. פריסה והדרכת צוות — פריסה על התשתית שלך, מסירת תיעוד.

מה כלול בפיתוח וציר הזמן

• API לתזמור — קליטת נתונים, תור, עדיפויות, חלוקה.
• ממשק מתעדים — Label Studio מותאם או ממשק React.
• מודול תיוג מקדים — מודלים חלשים עם ספי ביטחון.
• בקרת איכות — IAA, תקן זהב, צינור סקירה.
• ייצוא — JSONL, COCO, YOLO, אינטגרציה עם HuggingFace Datasets.
• למידה פעילה — מחשבון אי-ודאות וגיוון.
• תיעוד והדרכת צוות.

פלטפורמה בסיסית מבוססת Label Studio — משבועיים. מלאה עם תיוג מקדים ולמידה פעילה — מ-3 עד 8 שבועות תלוי במורכבות. התמחור נקבע באופן אישי לאחר בדיקת הנתונים והדרישות שלך.

דון בפרויקט שלך עם המהנדסים שלנו — נעריך את הנתונים שלך ונציע ארכיטקטורה. צור קשר לקבלת ייעוץ.