Auto-Data Labeling Pipeline with LLM and Snorkel

Auto-Data Labeling with LLM and Snorkel

Teams spend many days manually labeling datasets for NLP or Computer Vision. The bottleneck is not the model architecture, but high-quality labeled data. Auto-labeling pipelines reduce manual work by 60–80% while maintaining accuracy above the training threshold. We implement custom pipelines with LLM, Snorkel and ensemble strategies—turnkey, with quality guarantee. Snorkel — a framework for programmatic data labeling (Wikipedia)

How to Choose an Auto-Labeling Strategy?

Each pipeline starts with analyzing the data distribution, label schema, and accuracy requirements. We select the optimal strategy: LLM labeling for nuanced tasks, weak labeling (Snorkel) for large volumes, or a hybrid ensemble of models. Our engineers have implemented 30+ auto-labeling projects—from text sentiment to object detection in images. We will evaluate your dataset and propose a solution within 2-3 days.

Why Does an Ensemble of Models Yield Better Accuracy?

Combining Snorkel rules and neural networks improves recall without losing precision. The ensemble approach is 15–20% more accurate than any single model, without significant speed loss. When the weak model and LLM disagree (ensemble_disagree), such examples are automatically sent to a human. This catch-check captures 100% of ambiguous cases.

Technical Implementation of Auto-Labeling

Labeling via LLM and Zero-Shot

from anthropic import Anthropic
import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import Optional

@dataclass
class AutoLabelResult:
    text: str
    predicted_label: str
    confidence: float
    auto_accepted: bool
    method: str  # 'weak_model', 'llm', 'rules', 'ensemble'

class AutoLabelingPipeline:
    def __init__(self, task_type: str, confidence_threshold: float = 0.85):
        self.task_type = task_type
        self.threshold = confidence_threshold
        self.llm = Anthropic()
        self.stats = {'auto_accepted': 0, 'sent_to_review': 0}

    def label_batch(self, texts: list[str], label_schema: list[str], method: str = 'ensemble') -> list[AutoLabelResult]:
        """Auto-label a batch of texts"""
        if method == 'llm':
            return self._llm_labeling(texts, label_schema)
        elif method == 'weak_model':
            return self._weak_model_labeling(texts, label_schema)
        elif method == 'ensemble':
            return self._ensemble_labeling(texts, label_schema)
        else:
            raise ValueError(f"Unknown method: {method}")

    def _llm_labeling(self, texts: list[str], label_schema: list[str]) -> list[AutoLabelResult]:
        """LLM labeling with confidence estimation"""
        results = []
        batch_size = 10
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            texts_formatted = "\n".join([f"{j+1}. {t[:300]}" for j, t in enumerate(batch)])
            labels_str = ", ".join(label_schema)
            response = self.llm.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=400,
                messages=[{
                    "role": "user",
                    "content": f"""Classify each text. Labels: {labels_str}
Texts: {texts_formatted}
Return JSON array: [{{"label": "...", "confidence": 0.0-1.0}}]
confidence = how certain you are (0.9+ for obvious cases, 0.5-0.7 for ambiguous)."""
                }]
            )
            try:
                import json
                preds = json.loads(response.content[0].text)
                for text, pred in zip(batch, preds):
                    confidence = pred.get('confidence', 0.5)
                    results.append(AutoLabelResult(
                        text=text,
                        predicted_label=pred['label'],
                        confidence=confidence,
                        auto_accepted=confidence >= self.threshold,
                        method='llm'
                    ))
            except Exception:
                # Fallback: send to manual labeling
                for text in batch:
                    results.append(AutoLabelResult(
                        text=text,
                        predicted_label='unknown',
                        confidence=0.0,
                        auto_accepted=False,
                        method='llm_failed'
                    ))
        return results

    def _weak_model_labeling(self, texts: list[str], label_schema: list[str]) -> list[AutoLabelResult]:
        """Fast labeling via zero-shot model"""
        from transformers import pipeline
        classifier = pipeline(
            "zero-shot-classification",
            model="facebook/bart-large-mnli",
            device=0
        )
        results = []
        predictions = classifier(texts, candidate_labels=label_schema, batch_size=32)
        for text, pred in zip(texts, predictions):
            confidence = pred['scores'][0]
            # Penalty for close scores (uncertainty between labels)
            if len(pred['scores']) > 1 and pred['scores'][1] > 0.3:
                confidence *= 0.9
            results.append(AutoLabelResult(
                text=text,
                predicted_label=pred['labels'][0],
                confidence=confidence,
                auto_accepted=confidence >= self.threshold,
                method='weak_model'
            ))
        return results

    def _ensemble_labeling(self, texts: list[str], label_schema: list[str]) -> list[AutoLabelResult]:
        """Combination: fast model + LLM for uncertain cases"""
        # Step 1: Fast labeling
        weak_results = self._weak_model_labeling(texts, label_schema)
        # Step 2: LLM for uncertain ones
        uncertain_indices = [
            i for i, r in enumerate(weak_results)
            if not r.auto_accepted and r.confidence > 0.5  # Not complete failure
        ]
        uncertain_texts = [texts[i] for i in uncertain_indices]
        if uncertain_texts:
            llm_results = self._llm_labeling(uncertain_texts, label_schema)
            for idx, llm_result in zip(uncertain_indices, llm_results):
                # If models agree—increase confidence
                if llm_result.predicted_label == weak_results[idx].predicted_label:
                    combined_confidence = (weak_results[idx].confidence + llm_result.confidence) / 2 + 0.1
                    weak_results[idx].confidence = min(combined_confidence, 1.0)
                    weak_results[idx].auto_accepted = combined_confidence >= self.threshold
                    weak_results[idx].method = 'ensemble_agree'
                else:
                    # Disagreement—send to human
                    weak_results[idx].auto_accepted = False
                    weak_results[idx].method = 'ensemble_disagree'
        return weak_results

Weak Labeling with Snorkel

from snorkel.labeling import labeling_function, PandasLFApplier
from snorkel.labeling.model import LabelModel
import re

# Label constants
NEGATIVE, ABSTAIN, POSITIVE = -1, -2, 0

@labeling_function()
def lf_contains_positive_words(x):
    positive_words = ['excellent', 'great', 'amazing', 'love', 'perfect', 'отлично', 'супер', 'замечательно']
    return POSITIVE if any(w in x.text.lower() for w in positive_words) else ABSTAIN

@labeling_function()
def lf_contains_negative_words(x):
    negative_words = ['terrible', 'awful', 'worst', 'hate', 'horrible', 'ужасно', 'плохо', 'отстой']
    return NEGATIVE if any(w in x.text.lower() for w in negative_words) else ABSTAIN

@labeling_function()
def lf_rating_pattern(x):
    match = re.search(r'(\d)[/из]\s*5', x.text)
    if match:
        rating = int(match.group(1))
        if rating >= 4:
            return POSITIVE
        elif rating <= 2:
            return NEGATIVE
    return ABSTAIN

@labeling_function()
def lf_exclamation_positive(x):
    if x.text.count('!') >= 2 and len(x.text) < 100:
        return POSITIVE
    return ABSTAIN

def train_label_model(df: pd.DataFrame) -> pd.Series:
    """Snorkel: combine weak labeling functions"""
    lfs = [lf_contains_positive_words, lf_contains_negative_words, lf_rating_pattern, lf_exclamation_positive]
    applier = PandasLFApplier(lfs=lfs)
    L_train = applier.apply(df=df)
    # Train generative model
    label_model = LabelModel(cardinality=2, verbose=True)
    label_model.fit(L_train=L_train, n_epochs=500, lr=0.001)
    return label_model.predict(L=L_train)

Quality Monitoring and Threshold Tuning

How to Control Auto-Labeling Accuracy?

For data verification we use gold samples (up to 5% of the dataset), allowing continuous monitoring of auto-labeling accuracy and timely adjustment of thresholds or rules. Monitoring via gold samples is a standard practice that reduces the risk of error accumulation.

class AutoLabelQualityMonitor:
    """Quality control via gold samples"""

    def __init__(self, gold_samples: list[dict]):
        """gold_samples: [{text, true_label}]"""
        self.gold = gold_samples

    def evaluate_accuracy(self, pipeline: AutoLabelingPipeline) -> dict:
        """Accuracy of auto-labeling on gold samples"""
        texts = [g['text'] for g in self.gold]
        true_labels = [g['true_label'] for g in self.gold]
        label_schema = list(set(true_labels))
        results = pipeline.label_batch(texts, label_schema, method='ensemble')
        correct = sum(
            1 for r, true in zip(results, true_labels) if r.predicted_label == true
        )
        auto_accepted_correct = sum(
            1 for r, true in zip(results, true_labels) if r.auto_accepted and r.predicted_label == true
        )
        auto_accepted_total = sum(1 for r in results if r.auto_accepted)
        return {
            'overall_accuracy': correct / len(results),
            'auto_accepted_accuracy': (
                auto_accepted_correct / auto_accepted_total if auto_accepted_total > 0 else 0
            ),
            'auto_acceptance_rate': auto_accepted_total / len(results),
            'review_queue_size': len(results) - auto_accepted_total
        }

Comparison of Auto-Labeling Methods

Resource Savings and Confidence Threshold Selection

The optimal threshold for most classification tasks is 0.85–0.90. Reduces manual work by 65–70% with auto-accepted examples accuracy of 95–97%. Saves labeling budget up to 80% through automation. ROI less than two weeks.

Choosing the confidence threshold depends on the cost of error. If false classification is critical (medical diagnosis) — set 0.95, sacrificing speed. For mass tasks (review sentiment) — 0.85 gives the best balance. We help select the threshold experimentally within 1-2 days on your data — we guarantee the auto-labeling accuracy will not be lower than agreed.

Implementation Process and Typical Mistakes

Step-by-Step Pipeline Setup

1. Dataset analysis: evaluate label distribution, volume, noise level.
2. Model selection: LLM (Claude 3.5) for complex, zero-shot for simple.
3. Create Snorkel rules: from 10 to 50+ labeling functions.
4. Computation integration: code combines weak labels into a single dataset.
5. Pilot run: label 1000 examples, verify against gold.
6. Threshold adjustment: tune confidence threshold via ROC curve.
7. Production run: full pipeline with monitoring.

What Usually Goes Wrong?

• Blind trust in threshold without considering class difficulty: accuracy for rare classes may be lower.
• Using only one model: an ensemble is always more reliable.
• Lack of gold examples: without them you won't know quality.
• Too low threshold for the sake of savings: leads to error accumulation.

Results and Economic Efficiency

Case Study

For a client with a dataset of 50,000 reviews (sentiment task), we implemented an ensemble pipeline with threshold 0.85. Result: 95% accuracy on auto-labeled examples, manual work reduced from 40 to 12 person-days — a 3.3x speedup. ROI less than two weeks.

Contact us for an evaluation of your dataset — we will select the optimal auto-labeling strategy. We'll assess your project within 2-3 days. Get a consultation on pipeline implementation and learn how to automate labeling of your data.