Aktiv læring for dataannotering: Gemini + Adala-guide

Aktiv læring forvandler måten vi trener på AI modeller ved intelligent å velge de mest verdifulle dataene for annotering. Når det kombineres med kraftige LLM-er i likhet med Google Gemini, oppretter den effektive annoteringsrørledninger som reduserer manuell innsats samtidig som den opprettholder høy datakvalitet.

Denne veiledningen utforsker hvordan man bygger slike rørledninger ved hjelp av Adala-rammeverket – et kraftig, men underutnyttet verktøy for autonom datamerking.

Vi skal implementere en medisinsk symptomklassifikator som utnytter Gemini's evner gjennom en strukturert aktiv læringsarbeidsflyt.

Hva er Adala? En introduksjon til rammeverket

Adala (Autonom datamerkingsagent) er en rammeverk med åpen kildekode spesielt utviklet for implementering av spesialiserte agenter for databehandlingI motsetning til tradisjonelle annoteringsverktøy, benytter Adala seg av en agentbasert tilnærming som kombinerer:

Ser på Adala's hurtigstarteksempel, vi kan se hvordan det er strukturert følelsesklassifisering:

python

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)

For vår oppgave med å klassifisere medisinske symptomer, vil vi tilpasse denne arkitekturen for å integrere Google Gemini mens de implementerer en tilpasset aktiv læringsstrategi.

Sette opp miljøet ditt

La's begynn med å installere Adala og nødvendige avhengigheter:

python

# 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

Vi må også klone depotet for direkte tilgang til komponentene:

python

# 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

Integrering av Google Gemini som en tilpasset annotator

I motsetning til den opprinnelige implementeringen som brukte en grunnleggende innpakning rundt Google Gemini, skal vi bygge en mer robust annotator som følger Adala's designmønstre. Dette gjør løsningen vår mer vedlikeholdbar og utvidbar.

Først må vi sette opp Google Generative AI klient:

python

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)

Nå skal vi opprette en tilpasset annotator ved å utvide Adala's BaseAnnotator-klasse:

python

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

Denne implementeringen gir betydelige forbedringer i forhold til originalen:

1. Det følger riktig klassearv fra Adala's BaseAnnotator
2. Implementerer private hjelpemetoder for rask bygging og responsparsing
3. Bruker strukturert feilhåndtering og skriv hint
4. Gir fullstendig dokumentasjon

Bygge en pipeline for symptomklassifisering

La's lage et datasett av medisinske symptomer for klassifiseringsoppgaven vår. I motsetning til den opprinnelige implementeringen, vil vi bruke et mer mangfoldig datasett med balansert representasjon på tvers av kategorier:

python

# 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]

Implementering av avanserte aktive læringsstrategier

Den opprinnelige implementeringen brukte en enkel prioriteringsmekanisme. Vi vil forbedre denne med flere strategier for å demonstrere Adala.'s fleksibilitet:

python

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)

Nå, la's implementere vår forbedrede aktive læringsløkke:

python

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

Kjøre ende-til-ende-rørledningen

Nå kan vi kjøre hele vår aktive læringsprosess:

python

# 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}")

Praktiske anvendelser og utvidelser

Denne rørledningen har en rekke praktiske anvendelser utover medisinsk symptomklassifisering:

Du kan utvide denne implementeringen ved å:

Konklusjon

Integrasjonen av Adala og Google Gemini gir en kraftig rammeverk for å bygge intelligente annoteringsrørledninger. Ved å utnytte aktive læringsstrategier, kan vi dramatisk redusere den manuelle innsatsen som kreves samtidig som vi opprettholder annoteringer av høy kvalitet.

De modulære designmønstrene som er demonstrert i denne veiledningen, tillater enkel tilpasning til forskjellige domener og annoteringsoppgaver.