LLM-Assisted Labeling & Active Learning

# Structure annotation tasks as JSON with labeling guidelines
# Consistent formatting enables both human and LLM annotators
import json
from openai import OpenAI
from dataclasses import dataclass

client = OpenAI()

@dataclass
class PreLabel:
 text: str
 label: str
 confidence: float
 reasoning: str

def llm_prelabel(
 texts: list[str],
 label_options: list[str],
 task_description: str,
 model: str = "gpt-4o-mini"
) -> list[PreLabel]:
 """Pre-label a batch of texts using an LLM with confidence scores."""
 labels_str = ", ".join(f'"{l}"' for l in label_options)
 results = []

 for text in texts:
 prompt = f"""Task: {task_description}

Text: "{text}"

Available labels: [{labels_str}]

Classify this text. Provide:
1. The label (must be one of the available options)
2. Your confidence (0.0 to 1.0)
3. Brief reasoning

Respond as JSON:
{{
 "label": "chosen_label",
 "confidence": 0.95,
 "reasoning": "why this label"
}}"""

 # Send chat completion request to the API
 response = client.chat.completions.create(
 model=model,
 messages=[{"role": "user", "content": prompt}],
 temperature=0.1,
 response_format={"type": "json_object"}
 )

 # Extract the generated message from the API response
 data = json.loads(response.choices[0].message.content)
 results.append(PreLabel(
 text=text,
 label=data["label"],
 confidence=data["confidence"],
 reasoning=data.get("reasoning", "")
 ))

 return results

# Example: Sentiment classification pre-labeling
texts = [
 "This product exceeded all my expectations. Highly recommend!",
 "The delivery was okay but the packaging was damaged.",
 "Worst purchase I have ever made. Complete waste of money.",
 "It does what it says. Nothing special, nothing terrible.",
 "I cannot figure out how to set this up. Instructions are unclear.",
]

prelabels = llm_prelabel(
 texts=texts,
 label_options=["positive", "negative", "neutral", "mixed"],
 task_description="Classify the sentiment of this product review."
)

for pl in prelabels:
 route = "AUTO" if pl.confidence >= 0.85 else "HUMAN"
 print(f"[{route}] {pl.label} ({pl.confidence:.2f}): "
 f"{pl.text[:50]}...")

# Calculate annotation quality metrics using numpy
# Track agreement rates, label distributions, and annotator reliability
import numpy as np
from sklearn.metrics.pairwise import cosine_distances

def uncertainty_sampling(
 predictions: np.ndarray,
 n_select: int = 50
) -> np.ndarray:
 """Select examples where the model is most uncertain.

 Args:
 predictions: Array of shape (n_samples, n_classes) with
 predicted probabilities
 n_select: Number of examples to select

 Returns:
 Indices of selected examples
 """
 # Entropy-based uncertainty
 entropy = -np.sum(
 predictions * np.log(predictions + 1e-10), axis=1
 )
 # Select top-k most uncertain
 return np.argsort(entropy)[-n_select:]

def diversity_sampling(
 embeddings: np.ndarray,
 labeled_embeddings: np.ndarray,
 n_select: int = 50
) -> np.ndarray:
 """Select examples most different from the already-labeled set.

 Uses maximum distance to nearest labeled example (core-set approach).
 """
 # Distance from each unlabeled example to nearest labeled example
 distances = cosine_distances(embeddings, labeled_embeddings)
 min_distances = distances.min(axis=1)
 # Select the most distant (most different from labeled set)
 return np.argsort(min_distances)[-n_select:]

def hybrid_acquisition(
 predictions: np.ndarray,
 embeddings: np.ndarray,
 labeled_embeddings: np.ndarray,
 n_select: int = 50,
 uncertainty_weight: float = 0.6
) -> np.ndarray:
 """Hybrid strategy: weighted combination of uncertainty and diversity."""
 # Normalize uncertainty scores to [0, 1]
 entropy = -np.sum(predictions * np.log(predictions + 1e-10), axis=1)
 max_entropy = np.log(predictions.shape[1])
 uncertainty_scores = entropy / max_entropy

 # Normalize diversity scores to [0, 1]
 distances = cosine_distances(embeddings, labeled_embeddings)
 min_distances = distances.min(axis=1)
 diversity_scores = min_distances / max(min_distances.max(), 1e-10)

 # Weighted combination
 combined = (
 uncertainty_weight * uncertainty_scores +
 (1 - uncertainty_weight) * diversity_scores
 )
 return np.argsort(combined)[-n_select:]

# Simulate an active learning scenario
np.random.seed(42)
n_unlabeled = 1000
n_classes = 4

# Simulated model predictions (some confident, some uncertain)
predictions = np.random.dirichlet(np.ones(n_classes) * 2, n_unlabeled)
embeddings = np.random.randn(n_unlabeled, 128)
labeled_embeddings = np.random.randn(100, 128)

# Select using each strategy
uncertain_idx = uncertainty_sampling(predictions, n_select=50)
diverse_idx = diversity_sampling(embeddings, labeled_embeddings, n_select=50)
hybrid_idx = hybrid_acquisition(
 predictions, embeddings, labeled_embeddings, n_select=50
)

# Check overlap between strategies
overlap_u_d = len(set(uncertain_idx) & set(diverse_idx))
overlap_u_h = len(set(uncertain_idx) & set(hybrid_idx))
print(f"Uncertainty vs Diversity overlap: {overlap_u_d}/50 examples")
print(f"Uncertainty vs Hybrid overlap: {overlap_u_h}/50 examples")
print(f"Hybrid captures both uncertain AND diverse examples")

# Label Studio: Setting up a pre-labeling backend with LLM
# This creates a backend service that Label Studio calls for predictions

from label_studio_ml.model import LabelStudioMLBase
from openai import OpenAI

class LLMPreLabeler(LabelStudioMLBase):
 """Label Studio ML backend that uses an LLM for pre-labeling."""

 def setup(self):
 self.client = OpenAI()
 self.model = "gpt-4o-mini"

 def predict(self, tasks, **kwargs):
 """Generate pre-labels for a batch of tasks."""
 predictions = []

 for task in tasks:
 text = task["data"].get("text", "")

 response = self.client.chat.completions.create(
 model=self.model,
 messages=[{
 "role": "user",
 "content": f"Classify the sentiment of this text as "
 f"'positive', 'negative', or 'neutral'.\n\n"
 f"Text: {text}\n\nLabel:"
 }],
 temperature=0.1,
 max_tokens=10
 )

 label = response.choices[0].message.content.strip().lower()

 predictions.append({
 "result": [{
 "from_name": "sentiment",
 "to_name": "text",
 "type": "choices",
 "value": {"choices": [label]}
 }],
 "score": 0.85 # Confidence score
 })

 return predictions

# To run: label-studio-ml start ./llm_backend
# Then connect in Label Studio: Settings > Machine Learning > Add Model
print("LLM pre-labeling backend configured for Label Studio")

# Compute inter-annotator agreement using Cohen's kappa
# Kappa measures agreement beyond chance between two raters
import numpy as np
from itertools import combinations

def cohens_kappa(labels_a: list, labels_b: list) -> float:
 """Compute Cohen's Kappa between two annotators."""
 assert len(labels_a) == len(labels_b)
 n = len(labels_a)

 # Observed agreement
 observed = sum(a == b for a, b in zip(labels_a, labels_b)) / n

 # Expected agreement (by chance)
 unique_labels = set(labels_a) | set(labels_b)
 expected = 0
 for label in unique_labels:
 freq_a = labels_a.count(label) / n
 freq_b = labels_b.count(label) / n
 expected += freq_a * freq_b

 if expected == 1.0:
 return 1.0
 return (observed - expected) / (1 - expected)

def fleiss_kappa(ratings_matrix: np.ndarray) -> float:
 """Compute Fleiss' Kappa for multiple annotators.

 Args:
 ratings_matrix: Shape (n_subjects, n_categories).
 Each cell is the count of raters who assigned
 that category to that subject.
 """
 n_subjects, n_categories = ratings_matrix.shape
 n_raters = ratings_matrix.sum(axis=1)[0] # Assume same per subject

 # Proportion of assignments to each category
 p_j = ratings_matrix.sum(axis=0) / (n_subjects * n_raters)

 # Per-subject agreement
 p_i = (
 (ratings_matrix ** 2).sum(axis=1) - n_raters
 ) / (n_raters * (n_raters - 1))

 p_bar = p_i.mean()
 p_e = (p_j ** 2).sum()

 if p_e == 1.0:
 return 1.0
 return (p_bar - p_e) / (1 - p_e)

# Example: Compare LLM labels with two human annotators
human_a = ["pos", "neg", "neu", "pos", "neg", "pos", "neu", "neg", "pos", "neu"]
human_b = ["pos", "neg", "pos", "pos", "neg", "pos", "neg", "neg", "pos", "neu"]
llm = ["pos", "neg", "neu", "pos", "neg", "pos", "neu", "neg", "pos", "pos"]

kappa_humans = cohens_kappa(human_a, human_b)
kappa_llm_a = cohens_kappa(llm, human_a)
kappa_llm_b = cohens_kappa(llm, human_b)

print(f"Human A vs Human B (Kappa): {kappa_humans:.3f}")
print(f"LLM vs Human A (Kappa): {kappa_llm_a:.3f}")
print(f"LLM vs Human B (Kappa): {kappa_llm_b:.3f}")
print()
print("Interpretation:")
print(" 0.81-1.00: Almost perfect agreement")
print(" 0.61-0.80: Substantial agreement")
print(" 0.41-0.60: Moderate agreement")
print(" 0.21-0.40: Fair agreement")
print(" < 0.20: Slight/poor agreement")