აქტიური სწავლება მონაცემთა ანოტაციისთვის: Gemini + Adala-ს სახელმძღვანელო

აქტიური სწავლა ცვლის ჩვენს ვარჯიშს AI მოდელები ანოტაციისთვის ყველაზე ღირებული მონაცემების ინტელექტუალურად შერჩევით. როდესაც დაწყვილებულია ძლიერი LLM-ები ისევე როგორც Google Gemini, ის ქმნის ეფექტურ ანოტაციების არხებს, რომლებიც ამცირებს ხელით დატვირთვას და ამავდროულად ინარჩუნებს მონაცემთა მაღალ ხარისხს.

ეს სახელმძღვანელო იკვლევს, თუ როგორ უნდა ავაშენოთ ასეთი მილსადენები Adala-ს ჩარჩოს გამოყენებით - ეს არის ძლიერი, მაგრამ ნაკლებად გამოყენებული ინსტრუმენტი ავტონომიური მონაცემთა მარკირებისთვის.

ჩვენ განვახორციელებთ სამედიცინო სიმპტომების კლასიფიკატორს, რომელიც იყენებს Gemini-ს.'s შესაძლებლობები სტრუქტურირებული აქტიური სწავლების სამუშაო პროცესის მეშვეობით.

რა არის ადალა? შესავალი ჩარჩოში

Adala (ავტონომიური მონაცემთა მარკირების აგენტი) არის ღია კოდის ჩარჩო, რომელიც სპეციალურად შექმნილია მონაცემთა დამუშავების სპეციალიზებული აგენტების დანერგვისთვის . ტრადიციული ანოტაციის ინსტრუმენტებისგან განსხვავებით, Adala იყენებს აგენტებზე დაფუძნებულ მიდგომას, რომელიც აერთიანებს:

Adala-ს სწრაფი დაწყების მაგალითის გათვალისწინებით, შეგვიძლია დავინახოთ, თუ როგორ აყალიბებს ის განწყობის კლასიფიკაციას :

პითონი

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)

ჩვენი სამედიცინო სიმპტომების კლასიფიკაციის ამოცანისთვის, ჩვენ ამ არქიტექტურას ადაპტირებთ Google Gemini-ის ინტეგრირებისთვის, ამავდროულად, განვახორციელებთ მორგებულ აქტიური სწავლების სტრატეგიას.

თქვენი გარემოს დაყენება

მიადევნე's დაიწყეთ Adala-ს და საჭირო დამოკიდებულებების ინსტალაციით:

პითონი

# 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

ასევე დაგვჭირდება საცავის კლონირება მის კომპონენტებზე პირდაპირი წვდომისთვის:

პითონი

# 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

Google Gemini-ს, როგორც მორგებული ანოტატორის ინტეგრირება

ორიგინალური იმპლემენტაციისგან განსხვავებით, რომელიც Google Gemini-ს გარშემო საბაზისო შეფუთვას იყენებდა, ჩვენ შევქმნით უფრო მძლავრ ანოტატორს , რომელიც Adala-ს დიზაინის ნიმუშებს მიჰყვება. ეს ჩვენს გადაწყვეტას უფრო მდგრადს და გაფართოებადს ხდის.

პირველ რიგში, ჩვენ უნდა დავაყენოთ Google Generative AI კლიენტი:

პითონი

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)

ახლა, Adala-ს გაფართოებით შევქმნით მორგებულ ანოტატორს's BaseAnnotator კლასი:

პითონი

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

ეს განხორციელება ორიგინალთან შედარებით მნიშვნელოვან გაუმჯობესებას იძლევა:

1. ეს ადალადან კლასის მემკვიდრეობის შესაბამისობას მიჰყვება.'s BaseAnnotator
2. ახორციელებს კერძო დამხმარე მეთოდებს სწრაფი აგებისა და რეაგირების ანალიზისთვის.
3. იყენებს სტრუქტურირებულს შეცდომების გატარება და აკრიფეთ მინიშნებები
4. გთავაზობთ სრულ დოკუმენტაციას

სიმპტომების კლასიფიკაციის მილსადენის შექმნა

მოდით, შევქმნათ სამედიცინო სიმპტომების მონაცემთა ნაკრები ჩვენი კლასიფიკაციის ამოცანისთვის. ორიგინალური იმპლემენტაციისგან განსხვავებით, ჩვენ გამოვიყენებთ უფრო მრავალფეროვან მონაცემთა ნაკრებებს კატეგორიებში დაბალანსებული წარმოდგენით :

პითონი

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

აქტიური სწავლების მოწინავე სტრატეგიების დანერგვა

თავდაპირველი იმპლემენტაცია იყენებდა პრიორიტეტულობის შეფასების მარტივ მექანიზმს. ჩვენ ამას გავაუმჯობესებთ მრავალი სტრატეგიით, რათა ადალას დემონსტრირება მოვახდინოთ.'s მოქნილობა:

პითონი

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)

ახლა, მოდით's განახორციელეთ ჩვენი გაძლიერებული აქტიური სწავლების ციკლი:

პითონი

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

End-to-End Pipeline-ის გაშვება

ახლა ჩვენ შეგვიძლია გავუშვათ ჩვენი სრული აქტიური სწავლების პროცესი:

პითონი

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

პრაქტიკული გამოყენება და გაფართოებები

ამ მილსადენს სამედიცინო სიმპტომების კლასიფიკაციის გარდა მრავალი პრაქტიკული გამოყენება აქვს:

თქვენ შეგიძლიათ გააფართოვოთ ეს განხორციელება შემდეგნაირად:

დასკვნა

Adala-სა და Google Gemini-ს ინტეგრაცია ინტელექტუალური ანოტაციების სისტემების შესაქმნელად ძლიერ ჩარჩოს ქმნის. აქტიური სწავლების სტრატეგიების გამოყენებით , ჩვენ შეგვიძლია მნიშვნელოვნად შევამციროთ საჭირო ხელით მუშაობის ძალისხმევა მაღალი ხარისხის ანოტაციების შენარჩუნებით.

ამ სახელმძღვანელოში წარმოდგენილი მოდულური დიზაინის ნიმუშები სხვადასხვა დომენებსა და ანოტაციების ამოცანებთან მარტივად ადაპტირების საშუალებას იძლევა.