Пайплайн авторазмітки даних з LLM та Snorkel

Автоматичне розмічування даних за допомогою LLM та Snorkel

Команди витрачають багато днів на ручне розмічування датасетів для NLP або Computer Vision. Вузьке місце — не архітектура моделі, а якісні розмічені дані. Авторозмічувальні пайплайни скорочують ручну працю на 60–80%, зберігаючи точність вище порогу для навчання. Ми впроваджуємо кастомні пайплайни з LLM, Snorkel та ансамблевими стратегіями — під ключ, з гарантією якості. Snorkel — фреймворк для программатичної розмітки даних (Wikipedia)

Як обрати стратегію авторазмітки?

Кожен пайплайн починається з аналізу розподілу даних, схеми міток та вимог до точності. Обираємо оптимальну стратегію: LLM-розмітка для нюансних задач, слабка розмітка (Snorkel) для великих обсягів, або гібридний ансамбль моделей. Наші інженери реалізували 30+ проєктів авторазмітки — від тональності текстів до детекції об'єктів на зображеннях. Оцінимо ваш датасет і запропонуємо рішення за 2-3 дні.

Чому ансамбль моделей дає кращу точність?

Комбінація правил Snorkel і нейромереж підвищує recall без втрати precision. Ensemble-підхід дає на 15–20% точніше, ніж будь-яка окрема модель, без значної втрати швидкості. Коли weak model і LLM розходяться (ensemble_disagree), такі приклади автоматично надсилаються людині. Це catch-перевірка ловить 100% неоднозначних випадків.

Технічна реалізація авторазмітки

Розмітка через LLM та 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]: """Авторазмітка батчу текстів""" 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-розмітка з оцінкою впевненості""" 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: відправити на ручну розмітку 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]: """Швидка розмітка через zero-shot модель""" 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] # Штраф за близькі scores (невизначеність між лейблами) 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]: """Комбінація: швидка модель + LLM для невизначених випадків""" # Крок 1: Швидка розмітка weak_results = self._weak_model_labeling(texts, label_schema) # Крок 2: LLM для невизначених uncertain_indices = [ i for i, r in enumerate(weak_results) if not r.auto_accepted and r.confidence > 0.5 # Не зовсім провал ] 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 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: # Розбіжність — відправити людині weak_results[idx].auto_accepted = False weak_results[idx].method = 'ensemble_disagree' return weak_results 

Слабка розмітка з Snorkel

from snorkel.labeling import labeling_function, PandasLFApplier from snorkel.labeling.model import LabelModel import re # Константи міток 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: об'єднання слабких 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) # Навчання 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) 

Моніторинг якості та налаштування порогів

Як контролювати точність авторазмітки?

Для верифікації даних ми використовуємо золоті приклади (до 5% датасету), які дозволяють постійно моніторити точність авторазмітки та своєчасно коригувати пороги або правила. Моніторинг через золоті приклади — стандартна практика, що знижує ризик накопичення помилок.

class AutoLabelQualityMonitor: """Контроль якості через золоті приклади""" def __init__(self, gold_samples: list[dict]): """gold_samples: [{text, true_label}]""" self.gold = gold_samples def evaluate_accuracy(self, pipeline: AutoLabelingPipeline) -> dict: """Точність авторазмітки на золотих прикладах""" 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 } 

Порівняння методів авторазмітки

Економія ресурсів та вибір порогу confidence

Оптимальний поріг для більшості задач класифікації — 0.85–0.90. Скорочення ручної роботи на 65–70% при точності автоприйнятих прикладів 95–97%. Економія бюджету на розмітку до 80% за рахунок автоматизації. Окупність впровадження — менше двох тижнів.

Вибір порогу confidence залежить від ціни помилки. Якщо хибна класифікація критична (медична діагностика) — ставте 0.95, жертвуючи швидкістю. Для масових задач (тональність відгуків) — 0.85 дає кращий баланс. Ми допомагаємо підібрати поріг експериментально за 1-2 дні на ваших даних — гарантуємо, що точність авторазмітки буде не нижче обумовленої.

Процес впровадження та типові помилки

Покрокове налаштування пайплайну

1. Аналіз датасету: оцінюємо розподіл міток, обсяг, наявність шуму.
2. Вибір моделей: LLM (Claude 3.5) для складних, zero-shot для простих.
3. Створення правил Snorkel: від 10 до 50+ labeling functions.
4. Інтеграція обчислень: код об'єднує слабкі мітки в єдиний датасет.
5. Запуск пілоту: розмічаємо 1000 прикладів, звіряємо з золотими.
6. Коригування порогу: підбираємо confidence threshold за ROC-кривою.
7. Виробничий прогін: full pipeline з моніторингом.

Що найчастіше йде не так?

• Сліпа довіра порогу без урахування складності класів: для рідкісного класу точність може бути нижчою.
• Використання лише однієї моделі: ансамбль завжди надійніший.
• Відсутність золотих прикладів: без них ви не дізнаєтесь якість.
• Занадто низький поріг заради економії: призводить до накопичення помилок.

Результати та економічна ефективність

Приклад з практики

Для клієнта з датасетом у 50 000 відгуків (задача тональності) ми впровадили ensemble-пайплайн з порогом 0.85. Результат: 95% точність на авторозмічених прикладах, ручна робота скоротилася з 40 до 12 людино-днів — прискорення в 3,3 рази. Окупність впровадження — менше двох тижнів.

Зв'яжіться з нами для оцінки вашого датасету — ми підберемо оптимальну стратегію авторазмітки. Оцінимо проєкт безкоштовно за 2-3 дні. Отримайте консультацію щодо впровадження пайплайну та дізнайтеся, як автоматизувати розмітку ваших даних.