Pembelajaran Aktif untuk Anotasi Data: Panduan Gemini + Adala

Pembelajaran aktif mengubah cara kita berlatih AI model dengan secara cerdas memilih data yang paling berharga untuk anotasi. Ketika dipasangkan dengan LLM yang kuat 'like' Google Gemini, ia menciptakan jalur anotasi efisien yang mengurangi upaya manual sambil mempertahankan kualitas data yang tinggi.

Panduan ini membahas cara membangun jaringan pipa tersebut menggunakan Kerangka kerja Adala – sebuah alat yang kuat namun kurang dimanfaatkan untuk pelabelan data otonom.

Kami akan menerapkan pengklasifikasi gejala medis yang memanfaatkan Gemini's kemampuan melalui alur kerja pembelajaran aktif yang terstruktur.

Apa itu Adala? Pengenalan Kerangka Kerja

Adala (Autonomous Data Labeling Agent) adalah kerangka kerja sumber terbuka dirancang khusus untuk menerapkan agen khusus untuk pengolahan dataTidak seperti alat anotasi tradisional, Adala menggunakan pendekatan berbasis agen yang menggabungkan:

Melihat Adala's contoh quickstart, kita bisa melihat bagaimana strukturnya klasifikasi sentimen:

ular sanca

import pandas as pd
from adala.agents import Agent
from adala.environments import StaticEnvironment
from adala.skills import ClassificationSkill
from adala.runtimes import OpenAIChatRuntime
from rich import print

# Train dataset
train_df = pd.DataFrame([
    ["It was the negative first impressions, and then it started working.", "Positive"],
    ["Not loud enough and doesn't turn on like it should.", "Negative"],
    ["I don't know what to say.", "Neutral"],
    ["Manager was rude, but the most important that mic shows very flat frequency response.", "Positive"],
    ["The phone doesn't seem to accept anything except CBR mp3s.", "Negative"],
    ["I tried it before, I bought this device for my son.", "Neutral"],
], columns=["text", "sentiment"])

# Test dataset
test_df = pd.DataFrame([
    "All three broke within two months of use.",
    "The device worked for a long time, can't say anything bad.",
    "Just a random line of text."
], columns=["text"])

agent = Agent(
    # connect to a dataset
    environment=StaticEnvironment(df=train_df),
    # define a skill
    skills=ClassificationSkill(
        name='sentiment',
        instructions="Label text as positive, negative or neutral.",
        labels=["Positive", "Negative", "Neutral"],
        input_template="Text: {text}",
        output_template="Sentiment: {sentiment}"
    ),
    # define runtimes
    runtimes = {
        'openai': OpenAIChatRuntime(model='gpt-4o'),
    },
    teacher_runtimes = {
        'default': OpenAIChatRuntime(model='gpt-4o'),
    },
    default_runtime='openai',
)

agent.learn(learning_iterations=3, accuracy_threshold=0.95)
predictions = agent.run(test_df)

Untuk tugas klasifikasi gejala medis kami, kami akan mengadaptasi arsitektur ini untuk mengintegrasikan Google Gemini sambil menerapkan strategi pembelajaran aktif khusus.

Menyiapkan Lingkungan Anda

membiarkan's mulai dengan menginstal Adala dan dependensi yang diperlukan:

ular sanca

# Install Adala directly from GitHub
!pip install -q git+https://github.com/HumanSignal/Adala.git

# Verify installation
!pip list | grep adala

# Install additional dependencies
!pip install -q google-generativeai pandas matplotlib numpy

Kita juga perlu mengkloning repositori untuk akses langsung ke komponen-komponennya:

ular sanca

# Clone the repository for access to source files
!git clone https://github.com/HumanSignal/Adala.git

# Ensure the package is in our Python path
import sys
sys.path.append('./Adala')

# Import key components
from Adala.adala.annotators.base import BaseAnnotator
from Adala.adala.strategies.random_strategy import RandomStrategy
from Adala.adala.utils.custom_types import TextSample, LabeledSample

Mengintegrasikan Google Gemini sebagai Anotator Kustom

Tidak seperti implementasi asli yang menggunakan pembungkus dasar di sekitar Google Gemini, kami akan membangun lebih banyak pencatat yang kuat yang mengikuti Adala's pola desain. Hal ini membuat solusi kami lebih dapat dipelihara dan dapat diperluas.

Pertama, kita perlu mengatur Google Generative AI klien:

ular sanca

import google.generativeai as genai
import os

# Set API key from environment or enter manually
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") or getpass("Enter your Gemini API Key: ")
genai.configure(api_key=GEMINI_API_KEY)

Sekarang, kita akan membuat anotator khusus dengan memperluas Adala's Kelas BaseAnnotator:

ular sanca

import json
import re
from typing import List, Dict, Any, Optional

class GeminiAnnotator(BaseAnnotator):
    """Custom annotator using Google Gemini for medical symptom classification."""
    
    def __init__(self, 
                 model_name: str = "models/gemini-2.0-flash-lite", 
                 categories: List[str] = None,
                 temperature: float = 0.1):
        """Initialize the Gemini annotator.
        
        Args:
            model_name: The Gemini model to use
            categories: List of valid classification categories
            temperature: Controls randomness in generation (lower = more deterministic)
        """
        self.model = genai.GenerativeModel(
            model_name=model_name,
            generation_config={"temperature": temperature}
        )
        self.categories = categories or ["Cardiovascular", "Respiratory", 
                                         "Gastrointestinal", "Neurological"]
    
    def _build_prompt(self, text: str) -> str:
        """Create a structured prompt for the model.
        
        Args:
            text: The symptom text to classify
            
        Returns:
            A formatted prompt string
        """
        return f"""Classify this medical symptom into one of these categories:
        {', '.join(self.categories)}.
        
        Return JSON format: {{"category": "selected_category", 
        "confidence": 0.XX, "explanation": "brief_reason"}}
        
        SYMPTOM: {text}"""
    
    def _parse_response(self, response: str) -> Dict[str, Any]:
        """Extract structured data from model response.
        
        Args:
            response: Raw text response from Gemini
            
        Returns:
            Dictionary containing parsed fields
        """
        try:
            # Extract JSON from response even if surrounded by text
            json_match = re.search(r'(\{.*\})', response, re.DOTALL)
            result = json.loads(json_match.group(1) if json_match else response)
            return {
                "category": result.get("category", "Unknown"),
                "confidence": result.get("confidence", 0.0),
                "explanation": result.get("explanation", "")
            }
        except Exception as e:
            return {
                "category": "Unknown",
                "confidence": 0.0,
                "explanation": f"Error parsing response: {str(e)}"
            }
    
    def annotate(self, samples: List[TextSample]) -> List[LabeledSample]:
        """Annotate a batch of text samples.
        
        Args:
            samples: List of TextSample objects
            
        Returns:
            List of LabeledSample objects with annotations
        """
        results = []
        for sample in samples:
            prompt = self._build_prompt(sample.text)
            try:
                response = self.model.generate_content(prompt).text
                parsed = self._parse_response(response)
                
                # Create labeled sample with metadata
                labeled_sample = LabeledSample(
                    text=sample.text,
                    labels=parsed["category"],
                    metadata={
                        "confidence": parsed["confidence"],
                        "explanation": parsed["explanation"]
                    }
                )
            except Exception as e:
                # Graceful error handling
                labeled_sample = LabeledSample(
                    text=sample.text,
                    labels="Unknown",
                    metadata={"error": str(e)}
                )
            
            # Store reference to original sample
            labeled_sample._sample = sample
            results.append(labeled_sample)
            
        return results

Implementasi ini memberikan perbaikan signifikan dibandingkan versi aslinya:

1. Ini mengikuti pewarisan kelas yang tepat dari Adala's Anotator Dasar
2. Menerapkan metode pembantu pribadi untuk membangun prompt dan penguraian respons
3. Menggunakan terstruktur penanganan kesalahan dan ketik petunjuk
4. Menyediakan dokumentasi lengkap

Membangun Alur Klasifikasi Gejala

membiarkan's membuat kumpulan data gejala medis untuk tugas klasifikasi kami. Tidak seperti implementasi asli, kami akan menggunakan kumpulan data yang lebih beragam dengan representasi seimbang di seluruh kategori:

ular sanca

# Create a more comprehensive dataset
symptom_data = [
    # Cardiovascular symptoms
    "Chest pain radiating to left arm during exercise",
    "Heart palpitations when lying down",
    "Swollen ankles and shortness of breath",
    "Dizziness when standing up quickly",
    
    # Respiratory symptoms
    "Persistent dry cough with occasional wheezing",
    "Shortness of breath when climbing stairs",
    "Coughing up yellow or green mucus",
    "Rapid breathing with chest tightness",
    
    # Gastrointestinal symptoms
    "Stomach cramps and nausea after eating",
    "Burning sensation in upper abdomen",
    "Frequent loose stools with abdominal pain",
    "Yellowing of skin and eyes",
    
    # Neurological symptoms
    "Severe headache with sensitivity to light",
    "Numbness in fingers of right hand",
    "Memory loss and confusion",
    "Tremors in hands when reaching for objects"
]

# Convert to TextSample objects
text_samples = [TextSample(text=text) for text in symptom_data]

Menerapkan Strategi Pembelajaran Aktif Tingkat Lanjut

Implementasi awal menggunakan mekanisme penilaian prioritas sederhana. Kami akan menyempurnakannya dengan beberapa strategi untuk mendemonstrasikan Adala's fleksibilitas:

ular sanca

import numpy as np
from typing import List, Callable

class PrioritizationStrategy:
    """Base class for sample prioritization strategies."""
    
    def score_samples(self, samples: List[TextSample]) -> np.ndarray:
        """Assign priority scores to samples.
        
        Args:
            samples: List of samples to score
            
        Returns:
            Array of scores, higher values indicate higher priority
        """
        raise NotImplementedError("Subclasses must implement this method")
    
    def select(self, samples: List[TextSample], n: int = 1) -> List[TextSample]:
        """Select the top n highest scoring samples.
        
        Args:
            samples: List of samples to select from
            n: Number of samples to select
            
        Returns:
            List of selected samples
        """
        if not samples:
            return []
        
        scores = self.score_samples(samples)
        indices = np.argsort(-scores)[:n]  # Descending order
        return [samples[i] for i in indices]

class KeywordPriority(PrioritizationStrategy):
    """Prioritize samples based on medical urgency keywords."""
    
    def __init__(self, keyword_weights: Dict[str, float]):
        """Initialize with keyword weights.
        
        Args:
            keyword_weights: Dictionary mapping keywords to priority weights
        """
        self.keyword_weights = keyword_weights
    
    def score_samples(self, samples: List[TextSample]) -> np.ndarray:
        scores = np.zeros(len(samples))
        for i, sample in enumerate(samples):
            # Base score
            scores[i] = 0.1
            
            # Add weights for each keyword found
            text_lower = sample.text.lower()
            for keyword, weight in self.keyword_weights.items():
                if keyword in text_lower:
                    scores[i] += weight
        
        return scores

class UncertaintyPriority(PrioritizationStrategy):
    """Prioritize samples based on model uncertainty."""
    
    def __init__(self, model_fn: Callable[[List[TextSample]], List[float]]):
        """Initialize with uncertainty model function.
        
        Args:
            model_fn: Function that returns uncertainty scores for samples
        """
        self.model_fn = model_fn
    
    def score_samples(self, samples: List[TextSample]) -> np.ndarray:
        # Higher uncertainty = higher priority
        return np.array(self.model_fn(samples))

# Create a combined strategy
keyword_weights = {
    "chest": 0.5,
    "pain": 0.4,
    "breathing": 0.4, 
    "dizz": 0.3,
    "head": 0.2,
    "numb": 0.2
}

keyword_strategy = KeywordPriority(keyword_weights)

Sekarang, mari's menerapkan siklus pembelajaran aktif kami yang telah ditingkatkan:

ular sanca

from matplotlib import pyplot as plt
from IPython.display import clear_output
import time

def run_active_learning_loop(
    samples: List[TextSample],
    annotator: GeminiAnnotator,
    strategy: PrioritizationStrategy,
    iterations: int = 5,
    batch_size: int = 1,
    visualization_interval: int = 1
):
    """Run an active learning loop with visualization.
    
    Args:
        samples: Pool of unlabeled samples
        annotator: Annotation system
        strategy: Sample selection strategy
        iterations: Number of learning iterations
        batch_size: Samples to annotate per iteration
        visualization_interval: How often to update visualizations
    
    Returns:
        List of labeled samples
    """
    labeled_samples = []
    remaining_samples = list(samples)
    
    print("\nStarting Active Learning Loop:")
    
    for i in range(iterations):
        print(f"\n--- Iteration {i+1}/{iterations} ---")
        
        # Filter out already labeled samples
        remaining_samples = [
            s for s in remaining_samples 
            if s not in [getattr(l, '_sample', l) for l in labeled_samples]
        ]
        
        if not remaining_samples:
            print("No more samples to label. Stopping.")
            break
        
        # Select most important samples
        selected = strategy.select(remaining_samples, n=batch_size)
        
        # Annotate selected samples
        newly_labeled = annotator.annotate(selected)
        labeled_samples.extend(newly_labeled)
        
        # Display annotation results
        for sample in newly_labeled:
            print(f"Text: {sample.text}")
            print(f"Category: {sample.labels}")
            print(f"Confidence: {sample.metadata.get('confidence', 0):.2f}")
            explanation = sample.metadata.get('explanation', '')
            print(f"Explanation: {explanation[:100]}..." if len(explanation) > 100 else explanation)
            print()
        
        # Visualize results periodically
        if (i + 1) % visualization_interval == 0:
            visualize_results(labeled_samples)
            
    return labeled_samples

def visualize_results(labeled_samples: List[LabeledSample]):
    """Create visualizations of annotation results.
    
    Args:
        labeled_samples: List of labeled samples to visualize
    """
    if not labeled_samples:
        return
        
    # Extract data
    categories = [s.labels for s in labeled_samples]
    confidence = [s.metadata.get("confidence", 0) for s in labeled_samples]
    texts = [s.text[:30] + "..." for s in labeled_samples]
    
    # Set up plots
    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
    
    # Plot 1: Confidence by category
    category_counts = {}
    category_confidence = {}
    
    for cat, conf in zip(categories, confidence):
        if cat not in category_counts:
            category_counts[cat] = 0
            category_confidence[cat] = 0
        category_counts[cat] += 1
        category_confidence[cat] += conf
    
    for cat in category_confidence:
        category_confidence[cat] /= category_counts[cat]
    
    cats = list(category_counts.keys())
    counts = list(category_counts.values())
    avg_conf = list(category_confidence.values())
    
    x = np.arange(len(cats))
    width = 0.35
    
    ax1.bar(x - width/2, counts, width, label='Count')
    ax1.bar(x + width/2, avg_conf, width, label='Avg Confidence')
    ax1.set_xticks(x)
    ax1.set_xticklabels(cats, rotation=45)
    ax1.set_title('Category Distribution and Confidence')
    ax1.legend()
    
    # Plot 2: Individual sample confidence
    sorted_indices = np.argsort(confidence)
    ax2.barh(range(len(texts)), [confidence[i] for i in sorted_indices])
    ax2.set_yticks(range(len(texts)))
    ax2.set_yticklabels([texts[i] for i in sorted_indices])
    ax2.set_title('Sample Confidence')
    ax2.set_xlabel('Confidence')
    
    plt.tight_layout()
    plt.show()

Menjalankan Pipeline End-to-End

Sekarang kita dapat menjalankan alur pembelajaran aktif kita secara lengkap:

ular sanca

# Initialize components
categories = ["Cardiovascular", "Respiratory", "Gastrointestinal", "Neurological"]
annotator = GeminiAnnotator(categories=categories)
strategy = keyword_strategy

# Run the active learning loop
labeled_data = run_active_learning_loop(
    samples=text_samples,
    annotator=annotator,
    strategy=strategy,
    iterations=5,
    visualization_interval=2
)

# Final visualization and analysis
visualize_results(labeled_data)

# Print summary statistics
print("\nAnnotation Summary:")
print(f"Total samples annotated: {len(labeled_data)}")

categories = [s.labels for s in labeled_data]
unique_categories = set(categories)
print(f"Categories found: {len(unique_categories)}")
for category in unique_categories:
    count = categories.count(category)
    print(f"  - {category}: {count} samples ({count/len(labeled_data):.1%})")

avg_confidence = sum(s.metadata.get("confidence", 0) for s in labeled_data) / len(labeled_data)
print(f"Average confidence: {avg_confidence:.2f}")

Aplikasi Praktis dan Ekstensi

Saluran ini memiliki banyak aplikasi praktis di luar klasifikasi gejala medis:

Anda dapat memperluas implementasi ini dengan:

Kesimpulan

Integrasi Adala dan Google Gemini menyediakan kerangka kerja yang kuat untuk membangun jalur anotasi cerdas. Dengan memanfaatkan belajar strategi, kita dapat secara drastis mengurangi upaya manual yang diperlukan sambil tetap mempertahankan anotasi berkualitas tinggi.

Pola desain modular yang ditunjukkan dalam tutorial ini memungkinkan adaptasi mudah ke berbagai domain dan tugas anotasi.