Blablablab/audio-classification
0
1"""2Enhanced Active Learning Manager with Database Persistence3 4This module provides a comprehensive active learning system with optional5database persistence, model saving, LLM integration, and multiple query6strategies including uncertainty sampling, diversity sampling, BADGE, BALD,7and hybrid combinations.8 9References:10 [1] Ash et al. (2020) "Deep Batch Active Learning by Diverse, Uncertain11 Gradient Lower Bounds" (BADGE). ICLR 2020.12 [2] Houlsby et al. (2011) "Bayesian Active Learning for Classification13 and Preference Learning" (BALD).14 [3] Bayer et al. (2024) "ActiveLLM: Large Language Model-Based Active15 Learning for Textual Few-Shot Scenarios". TACL.16 [4] Yuan et al. (2024) "Hide and Seek in Noise Labels: Noise-Robust17 Collaborative Active Learning" (NoiseAL). ACL 2024.18 [5] Mavromatis et al. (2024) "CoverICL: Selective Annotation for19 In-Context Learning via Active Graph Coverage". EMNLP 2024.20"""21 22import threading23import logging24import time25import os26import pickle27import json28from typing import Dict, List, Optional, Tuple, Any, Union29from collections import defaultdict, Counter30import dataclasses31from dataclasses import dataclass, field, asdict32from enum import Enum33import random34import queue35from datetime import datetime36from abc import ABC, abstractmethod37 38from sklearn.pipeline import Pipeline39from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer40from sklearn.linear_model import LogisticRegression41from sklearn.ensemble import RandomForestClassifier42from sklearn.svm import SVC43from sklearn.metrics import accuracy_score, classification_report44import numpy as np45 46from potato.item_state_management import ItemStateManager, get_item_state_manager47from potato.user_state_management import get_user_state_manager48 49 50logger = logging.getLogger(__name__)51 52 53class ResolutionStrategy(Enum):54 """Strategies for resolving multiple annotations per instance."""55 MAJORITY_VOTE = "majority_vote"56 RANDOM = "random"57 CONSENSUS = "consensus"58 WEIGHTED_AVERAGE = "weighted_average"59 60 61# ---------------------------------------------------------------------------62# SentenceTransformerVectorizer63# ---------------------------------------------------------------------------64 65class SentenceTransformerVectorizer:66 """sklearn-compatible wrapper for sentence-transformers.67 68 Uses dense embeddings from pre-trained transformer models instead of69 bag-of-words features. Produces 384-dim vectors (for default model)70 that capture semantic meaning, enabling better classification with71 fewer training examples.72 73 The ``sentence-transformers`` package is an **optional** dependency and74 is only imported when this vectorizer is actually used.75 """76 77 def __init__(self, model_name: str = "all-MiniLM-L6-v2"):78 self.model_name = model_name79 self._model = None80 81 def fit(self, X, y=None):82 from sentence_transformers import SentenceTransformer83 self._model = SentenceTransformer(self.model_name)84 return self85 86 def transform(self, X):87 if self._model is None:88 raise RuntimeError("SentenceTransformerVectorizer has not been fitted yet")89 return self._model.encode(list(X), show_progress_bar=False)90 91 def fit_transform(self, X, y=None):92 self.fit(X, y)93 return self.transform(X)94 95 96# ---------------------------------------------------------------------------97# Query Strategies98# ---------------------------------------------------------------------------99 100class QueryStrategy(ABC):101 """Base class for active learning query strategies."""102 103 @abstractmethod104 def rank(self, texts: List[str], model, vectorizer,105 annotated_texts: Optional[List[str]] = None) -> List[Tuple[int, float]]:106 """Return list of (index, score) sorted by selection priority (highest first)."""107 108 109class UncertaintySampling(QueryStrategy):110 """Select instances where classifier is least confident.111 112 Selects x* = argmax_x (1 - max_y P(y|x)), i.e., instances where the113 model's best guess has lowest confidence.114 """115 116 def rank(self, texts, model, vectorizer, annotated_texts=None):117 try:118 features = vectorizer.transform(texts)119 probas = model.predict_proba(features)120 # Score = 1 - max_prob (higher = more uncertain = higher priority)121 scores = 1.0 - np.max(probas, axis=1)122 ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)123 return ranked124 except Exception as e:125 logger.warning(f"UncertaintySampling failed: {e}")126 return [(i, 0.5) for i in range(len(texts))]127 128 129class DiversitySampling(QueryStrategy):130 """Select instances that maximize feature-space coverage.131 132 Uses cosine distance from already-annotated instances in the vectorized133 feature space. Ensures the training set covers the full data distribution134 rather than over-sampling one region.135 """136 137 def rank(self, texts, model, vectorizer, annotated_texts=None):138 from sklearn.metrics.pairwise import cosine_distances139 140 try:141 features = vectorizer.transform(texts)142 if hasattr(features, 'toarray'):143 features = features.toarray()144 145 if annotated_texts:146 annotated_features = vectorizer.transform(annotated_texts)147 if hasattr(annotated_features, 'toarray'):148 annotated_features = annotated_features.toarray()149 # Score = min cosine distance to any annotated instance150 distances = cosine_distances(features, annotated_features)151 scores = np.min(distances, axis=1)152 else:153 # No annotated texts yet: use distance from centroid154 centroid = np.mean(features, axis=0, keepdims=True)155 scores = cosine_distances(features, centroid).ravel()156 157 ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)158 return ranked159 except Exception as e:160 logger.warning(f"DiversitySampling failed: {e}")161 return [(i, 0.5) for i in range(len(texts))]162 163 164class BadgeStrategy(QueryStrategy):165 """BADGE approximation: uncertainty-weighted diversity.166 167 Inspired by Ash et al. (2020) [Ref 1]. Full BADGE uses gradient embeddings168 from neural networks. Our approximation:169 1. Weight feature vectors by (1 - max_prob) as uncertainty proxy170 2. Run k-means++ initialization on weighted vectors to select171 diverse-uncertain instances.172 """173 174 def rank(self, texts, model, vectorizer, annotated_texts=None):175 try:176 features = vectorizer.transform(texts)177 if hasattr(features, 'toarray'):178 features = features.toarray()179 180 probas = model.predict_proba(features)181 uncertainty = 1.0 - np.max(probas, axis=1)182 183 # Weight features by uncertainty184 weighted = features * uncertainty[:, np.newaxis]185 186 # Use k-means++ initialization to select diverse-uncertain points187 from sklearn.cluster import kmeans_plusplus188 n_clusters = min(len(texts), max(1, len(texts) // 2))189 _, indices = kmeans_plusplus(weighted, n_clusters=n_clusters,190 random_state=42)191 192 # Build score: selected centroids get highest scores193 scores = np.zeros(len(texts))194 for rank_pos, idx in enumerate(indices):195 scores[idx] = len(indices) - rank_pos # highest for first-selected196 197 # For non-selected, use uncertainty as tiebreaker198 for i in range(len(texts)):199 if scores[i] == 0:200 scores[i] = uncertainty[i] * 0.01201 202 ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)203 return ranked204 except Exception as e:205 logger.warning(f"BadgeStrategy failed, falling back to uncertainty: {e}")206 return UncertaintySampling().rank(texts, model, vectorizer, annotated_texts)207 208 209class BaldStrategy(QueryStrategy):210 """BALD: Bayesian Active Learning by Disagreement.211 212 Based on Houlsby et al. (2011) [Ref 2]. Trains an ensemble of classifiers213 with different random seeds/bootstrap samples. Selects instances with214 highest mutual information: H[y|x] - E_theta[H[y|x,theta]], i.e.,215 where the ensemble disagrees most.216 """217 218 def __init__(self, n_estimators: int = 5, bootstrap_fraction: float = 0.8):219 self.n_estimators = n_estimators220 self.bootstrap_fraction = bootstrap_fraction221 222 def rank(self, texts, model, vectorizer, annotated_texts=None):223 try:224 features = vectorizer.transform(texts)225 if hasattr(features, 'toarray'):226 features = features.toarray()227 228 probas = model.predict_proba(features)229 # Average entropy230 avg_proba = probas231 entropy_avg = -np.sum(avg_proba * np.log(avg_proba + 1e-10), axis=1)232 233 # For a single model, we approximate BALD by using dropout-like noise234 # or by comparing with uniform. Since we store the ensemble models235 # on the manager, we just use the single model's entropy here and236 # the ensemble version is handled in ActiveLearningManager._train_bald_ensemble237 scores = entropy_avg238 ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)239 return ranked240 except Exception as e:241 logger.warning(f"BaldStrategy failed: {e}")242 return [(i, 0.5) for i in range(len(texts))]243 244 def rank_with_ensemble(self, texts, ensemble_models, vectorizer):245 """Rank using actual ensemble disagreement (mutual information)."""246 try:247 features = vectorizer.transform(texts)248 if hasattr(features, 'toarray'):249 features = features.toarray()250 251 all_probas = []252 for m in ensemble_models:253 all_probas.append(m.predict_proba(features))254 255 all_probas = np.array(all_probas) # (n_estimators, n_samples, n_classes)256 257 # Mean prediction across ensemble258 mean_proba = np.mean(all_probas, axis=0) # (n_samples, n_classes)259 260 # H[y|x] - entropy of mean prediction261 entropy_mean = -np.sum(mean_proba * np.log(mean_proba + 1e-10), axis=1)262 263 # E_theta[H[y|x,theta]] - mean of individual entropies264 individual_entropies = -np.sum(all_probas * np.log(all_probas + 1e-10), axis=2)265 mean_entropy = np.mean(individual_entropies, axis=0)266 267 # Mutual information = H[y|x] - E[H[y|x,theta]]268 mutual_info = entropy_mean - mean_entropy269 270 ranked = sorted(enumerate(mutual_info), key=lambda x: x[1], reverse=True)271 return ranked272 except Exception as e:273 logger.warning(f"BaldStrategy ensemble ranking failed: {e}")274 return [(i, 0.5) for i in range(len(texts))]275 276 277class HybridStrategy(QueryStrategy):278 """Weighted combination of uncertainty and diversity scores.279 280 Combines strategies with configurable weights. Default: 0.7 uncertainty +281 0.3 diversity.282 """283 284 def __init__(self, weights: Optional[Dict[str, float]] = None):285 self.weights = weights or {"uncertainty": 0.7, "diversity": 0.3}286 287 def rank(self, texts, model, vectorizer, annotated_texts=None):288 try:289 strategies = {}290 if self.weights.get("uncertainty", 0) > 0:291 strategies["uncertainty"] = UncertaintySampling()292 if self.weights.get("diversity", 0) > 0:293 strategies["diversity"] = DiversitySampling()294 295 # Collect raw scores from each strategy296 all_scores = {}297 for name, strategy in strategies.items():298 rankings = strategy.rank(texts, model, vectorizer, annotated_texts)299 score_map = {idx: score for idx, score in rankings}300 all_scores[name] = score_map301 302 # Normalize each strategy's scores to [0, 1]303 for name in all_scores:304 vals = list(all_scores[name].values())305 min_val, max_val = min(vals), max(vals)306 rng = max_val - min_val if max_val > min_val else 1.0307 all_scores[name] = {308 idx: (s - min_val) / rng for idx, s in all_scores[name].items()309 }310 311 # Weighted combination312 combined = {}313 for i in range(len(texts)):314 combined[i] = sum(315 self.weights.get(name, 0) * all_scores.get(name, {}).get(i, 0)316 for name in self.weights317 )318 319 ranked = sorted(combined.items(), key=lambda x: x[1], reverse=True)320 return ranked321 except Exception as e:322 logger.warning(f"HybridStrategy failed: {e}")323 return UncertaintySampling().rank(texts, model, vectorizer, annotated_texts)324 325 326# Strategy registry327STRATEGY_REGISTRY = {328 "uncertainty": UncertaintySampling,329 "diversity": DiversitySampling,330 "badge": BadgeStrategy,331 "bald": BaldStrategy,332 "hybrid": HybridStrategy,333}334 335 336def create_query_strategy(config: 'ActiveLearningConfig') -> QueryStrategy:337 """Create a query strategy from config."""338 strategy_name = config.query_strategy339 if strategy_name == "hybrid":340 return HybridStrategy(weights=config.hybrid_weights)341 elif strategy_name == "bald":342 params = config.bald_params343 return BaldStrategy(344 n_estimators=params.get("n_estimators", 5),345 bootstrap_fraction=params.get("bootstrap_fraction", 0.8),346 )347 elif strategy_name in STRATEGY_REGISTRY:348 return STRATEGY_REGISTRY[strategy_name]()349 else:350 logger.warning(f"Unknown strategy '{strategy_name}', falling back to uncertainty")351 return UncertaintySampling()352 353 354# ---------------------------------------------------------------------------355# ICLClassifier wrapper (Phase 5A)356# ---------------------------------------------------------------------------357 358class ICLClassifier:359 """Wraps ICLLabeler as an sklearn-compatible classifier for ensemble use.360 361 Enables combining LLM-based ICL predictions with traditional classifier362 predictions in a hybrid ensemble for active learning scoring.363 """364 365 def __init__(self, icl_labeler, schema_name: str, label_names: List[str]):366 self.icl_labeler = icl_labeler367 self.schema_name = schema_name368 self.label_names = label_names369 self.classes_ = np.array(label_names)370 371 def predict_proba(self, texts: List[str]) -> np.ndarray:372 """Get label probabilities from LLM via ICL."""373 n_classes = len(self.label_names)374 probas = np.full((len(texts), n_classes), 1.0 / n_classes)375 376 for i, text in enumerate(texts):377 try:378 prediction = self.icl_labeler.label_instance(379 instance_id=f"_al_query_{i}",380 schema_name=self.schema_name,381 instance_text=text,382 )383 if prediction and prediction.predicted_label in self.label_names:384 idx = self.label_names.index(prediction.predicted_label)385 conf = prediction.confidence_score386 # Distribute: conf to predicted label, (1-conf)/(n-1) to others387 remaining = (1.0 - conf) / max(1, n_classes - 1)388 probas[i] = remaining389 probas[i, idx] = conf390 except Exception:391 pass # Keep uniform distribution392 393 return probas394 395 396# ---------------------------------------------------------------------------397# Configuration398# ---------------------------------------------------------------------------399 400@dataclass401class ActiveLearningConfig:402 """Enhanced configuration for active learning."""403 enabled: bool = False404 classifier_name: str = "sklearn.linear_model.LogisticRegression"405 classifier_kwargs: Dict[str, Any] = None406 vectorizer_name: str = "sklearn.feature_extraction.text.TfidfVectorizer"407 vectorizer_kwargs: Dict[str, Any] = None408 min_annotations_per_instance: int = 1409 min_instances_for_training: int = 10410 max_instances_to_reorder: Optional[int] = None411 resolution_strategy: ResolutionStrategy = ResolutionStrategy.MAJORITY_VOTE412 random_sample_percent: float = 0.2413 update_frequency: int = 5414 schema_names: List[str] = None415 416 # Classifier/vectorizer passthrough params (Phase 1C)417 classifier_params: Dict[str, Any] = field(default_factory=dict)418 vectorizer_params: Dict[str, Any] = field(default_factory=dict)419 420 # Probability calibration (Phase 1D)421 calibrate_probabilities: bool = True422 423 # Query strategy (Phase 2)424 query_strategy: str = "uncertainty"425 hybrid_weights: Dict[str, float] = field(426 default_factory=lambda: {"uncertainty": 0.7, "diversity": 0.3}427 )428 bald_params: Dict[str, Any] = field(429 default_factory=lambda: {"n_estimators": 5, "bootstrap_fraction": 0.8}430 )431 432 # Cold-start (Phase 3)433 cold_start_strategy: str = "random"434 cold_start_batch_size: int = 20435 436 # ICL ensemble (Phase 5)437 use_icl_ensemble: bool = False438 icl_ensemble_params: Dict[str, Any] = field(default_factory=lambda: {439 "initial_icl_weight": 0.7,440 "final_icl_weight": 0.2,441 "transition_instances": 100,442 })443 444 # Annotation routing (Phase 5D)445 annotation_routing: bool = False446 routing_thresholds: Dict[str, float] = field(default_factory=lambda: {447 "auto_label_min_confidence": 0.9,448 "show_suggestion_below": 0.5,449 })450 verification_sample_rate: float = 0.2451 452 # Database persistence453 database_enabled: bool = False454 database_config: Dict[str, Any] = None455 456 # Model persistence457 model_persistence_enabled: bool = False458 model_save_directory: Optional[str] = None459 model_retention_count: int = 2460 461 # LLM integration462 llm_enabled: bool = False463 llm_config: Dict[str, Any] = None464 465 def __post_init__(self):466 if self.classifier_kwargs is None:467 self.classifier_kwargs = {}468 if self.vectorizer_kwargs is None:469 self.vectorizer_kwargs = {}470 if self.schema_names is None:471 self.schema_names = []472 if self.database_config is None:473 self.database_config = {}474 if self.llm_config is None:475 self.llm_config = {}476 # Merge classifier_params into classifier_kwargs477 if self.classifier_params:478 self.classifier_kwargs.update(self.classifier_params)479 # Merge vectorizer_params into vectorizer_kwargs480 if self.vectorizer_params:481 self.vectorizer_kwargs.update(self.vectorizer_params)482 483 484@dataclass485class TrainingMetrics:486 """Metrics for a training run."""487 schema_name: str488 training_time: float489 accuracy: float490 instance_count: int491 timestamp: datetime492 model_file_path: Optional[str] = None493 confidence_distribution: Dict[str, float] = None494 error_message: Optional[str] = None495 496 497class ModelPersistence:498 """Handles model saving and loading with metadata."""499 500 def __init__(self, save_directory: str, retention_count: int = 2):501 self.save_directory = save_directory502 self.retention_count = retention_count503 self.logger = logging.getLogger(__name__)504 505 # Ensure directory exists506 os.makedirs(save_directory, exist_ok=True)507 508 def save_model(self, model: Pipeline, schema_name: str, instance_count: int) -> str:509 """Save a trained model with metadata."""510 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")511 filename = f"{schema_name}_{instance_count}_{timestamp}.pkl"512 filepath = os.path.join(self.save_directory, filename)513 514 try:515 # Save the complete model (including vectorizer)516 with open(filepath, 'wb') as f:517 pickle.dump(model, f)518 519 self.logger.info(f"Saved model to {filepath}")520 521 # Clean up old models522 self._cleanup_old_models(schema_name)523 524 return filepath525 except Exception as e:526 self.logger.error(f"Failed to save model: {e}")527 raise528 529 def load_model(self, filepath: str) -> Optional[Pipeline]:530 """Load a saved model."""531 try:532 with open(filepath, 'rb') as f:533 model = pickle.load(f)534 535 # TODO: Add schema validation here in the future536 # This is a placeholder for future schema validation enhancement537 538 self.logger.info(f"Loaded model from {filepath}")539 return model540 except Exception as e:541 self.logger.error(f"Failed to load model from {filepath}: {e}")542 return None543 544 def _cleanup_old_models(self, schema_name: str):545 """Clean up old models based on retention policy."""546 try:547 # Find all model files for this schema548 model_files = []549 550 for filename in os.listdir(self.save_directory):551 if filename.startswith(f"{schema_name}_") and filename.endswith(".pkl"):552 filepath = os.path.join(self.save_directory, filename)553 model_files.append((filepath, os.path.getmtime(filepath)))554 555 # Sort by modification time (newest first)556 model_files.sort(key=lambda x: x[1], reverse=True)557 558 # Remove old models beyond retention count559 for filepath, _ in model_files[self.retention_count:]:560 try:561 os.remove(filepath)562 self.logger.info(f"Removed old model: {filepath}")563 except Exception as e:564 self.logger.warning(f"Failed to remove old model {filepath}: {e}")565 566 except Exception as e:567 self.logger.error(f"Error during model cleanup: {e}")568 569 570class DatabaseStateManager:571 """Manages database persistence for active learning state."""572 573 def __init__(self, config: Dict[str, Any]):574 self.config = config575 self.logger = logging.getLogger(__name__)576 self.connection = None577 self._initialize_database()578 579 def _initialize_database(self):580 """Initialize database connection and create tables."""581 try:582 # Use the same database system as main Potato application583 if self.config.get('type') == 'mysql':584 self._init_mysql_connection()585 else:586 self._init_file_based_connection()587 588 self._create_tables()589 self.logger.info("Active learning database initialized successfully")590 except Exception as e:591 self.logger.error(f"Failed to initialize database: {e}")592 raise593 594 def _init_mysql_connection(self):595 """Initialize MySQL connection."""596 # TODO: Implement MySQL connection597 pass598 599 def _init_file_based_connection(self):600 """Initialize file-based database connection."""601 # TODO: Implement file-based database602 pass603 604 def _create_tables(self):605 """Create database tables for active learning."""606 # TODO: Implement table creation607 pass608 609 def save_training_metrics(self, metrics: TrainingMetrics):610 """Save training metrics to database."""611 # TODO: Implement metrics saving612 pass613 614 def get_training_history(self, schema_name: Optional[str] = None) -> List[TrainingMetrics]:615 """Get training history from database."""616 # TODO: Implement history retrieval617 return []618 619 def save_schema_cycling_state(self, current_schema: str, schema_order: List[str]):620 """Save current schema cycling state."""621 # TODO: Implement state saving622 pass623 624 def get_schema_cycling_state(self) -> Tuple[str, List[str]]:625 """Get current schema cycling state."""626 # TODO: Implement state retrieval627 return "", []628 629 630class SchemaCycler:631 """Manages cycling through multiple annotation schemes."""632 633 def __init__(self, schema_names: List[str], database_manager: Optional[DatabaseStateManager] = None):634 self.schema_names = self._validate_schemas(schema_names)635 self.database_manager = database_manager636 self.current_index = 0637 self.logger = logging.getLogger(__name__)638 self._lock = threading.Lock()639 640 # Load state from database if available641 if self.database_manager:642 self._load_state()643 644 def _validate_schemas(self, schema_names: List[str]) -> List[str]:645 """Validate and filter schema names."""646 valid_schemas = []647 648 for schema in schema_names:649 # Exclude text and span annotation schemes650 if schema in ['text', 'span']:651 raise ValueError(f"Text and span annotation schemes are not supported for active learning: {schema}")652 valid_schemas.append(schema)653 654 return valid_schemas655 656 def _load_state(self):657 """Load cycling state from database."""658 try:659 current_schema, schema_order = self.database_manager.get_schema_cycling_state()660 with self._lock:661 if current_schema in self.schema_names:662 self.current_index = self.schema_names.index(current_schema)663 except Exception as e:664 self.logger.warning(f"Failed to load schema cycling state: {e}")665 666 def get_current_schema(self) -> Optional[str]:667 """Get the current schema for training."""668 if not self.schema_names:669 return None670 with self._lock:671 return self.schema_names[self.current_index]672 673 def advance_schema(self):674 """Advance to the next schema in the cycle."""675 if not self.schema_names:676 return677 678 with self._lock:679 self.current_index = (self.current_index + 1) % len(self.schema_names)680 current_schema = self.schema_names[self.current_index]681 682 # Save state to database if available683 if self.database_manager:684 try:685 self.database_manager.save_schema_cycling_state(686 current_schema,687 self.schema_names688 )689 except Exception as e:690 self.logger.warning(f"Failed to save schema cycling state: {e}")691 692 def get_schema_order(self) -> List[str]:693 """Get the current schema cycling order."""694 return self.schema_names.copy()695 696 697class ActiveLearningManager:698 """699 Manages active learning operations including classifier training and instance reordering.700 701 This class provides thread-safe operations for:702 - Training classifiers on annotated data703 - Predicting confidence scores for unlabeled instances704 - Reordering instances based on configurable query strategies705 - Cold-start LLM-based instance selection706 - ICL/classifier ensemble for improved ranking707 - Noise-aware annotation routing708 - Managing training state and progress709 - Database persistence and model saving710 """711 712 def __init__(self, config: ActiveLearningConfig):713 self.config = config714 self.logger = logging.getLogger(__name__)715 716 # Thread safety717 self._lock = threading.RLock()718 self._training_queue = queue.Queue()719 self._training_thread = None720 self._stop_training = threading.Event()721 722 # State tracking723 self._last_training_time = 0724 self._training_count = 0725 self._models = {} # schema_name -> trained_model726 self._vectorizers = {} # schema_name -> fitted vectorizer727 self._bald_ensembles = {} # schema_name -> list of classifiers728 self._last_annotation_count = 0729 self._training_metrics = [] # List of TrainingMetrics730 self._annotated_texts = {} # schema_name -> list of annotated texts731 732 # Query strategy733 self._query_strategy = create_query_strategy(config)734 735 # Database and persistence736 self.database_manager = None737 self.model_persistence = None738 self.schema_cycler = None739 740 # Initialize components741 self._initialize_components()742 743 # Start training thread if enabled744 if self.config.enabled:745 self._start_training_thread()746 747 def _initialize_components(self):748 """Initialize database, model persistence, and schema cycler."""749 # Initialize database manager if enabled750 if self.config.database_enabled:751 try:752 self.database_manager = DatabaseStateManager(self.config.database_config)753 except Exception as e:754 self.logger.error(f"Failed to initialize database manager: {e}")755 # Continue without database persistence756 757 # Initialize model persistence if enabled758 if self.config.model_persistence_enabled and self.config.model_save_directory:759 try:760 self.model_persistence = ModelPersistence(761 self.config.model_save_directory,762 self.config.model_retention_count763 )764 except Exception as e:765 self.logger.error(f"Failed to initialize model persistence: {e}")766 # Continue without model persistence767 768 # Initialize schema cycler769 try:770 self.schema_cycler = SchemaCycler(self.config.schema_names, self.database_manager)771 except Exception as e:772 self.logger.error(f"Failed to initialize schema cycler: {e}")773 raise # Schema cycler is critical774 775 def _start_training_thread(self):776 """Start the background training thread."""777 if self._training_thread is None or not self._training_thread.is_alive():778 self._training_thread = threading.Thread(target=self._training_worker, daemon=True)779 self._training_thread.start()780 self.logger.info("Active learning training thread started")781 782 def _training_worker(self):783 """Background worker for training classifiers."""784 while not self._stop_training.is_set():785 try:786 # Wait for training request787 training_request = self._training_queue.get(timeout=1.0)788 if training_request is None: # Shutdown signal789 break790 791 self._perform_training()792 self._training_queue.task_done()793 794 except queue.Empty:795 continue796 except Exception as e:797 self.logger.error(f"Error in training worker: {e}")798 799 def _perform_training(self):800 """Perform the actual classifier training."""801 with self._lock:802 try:803 self.logger.info("Starting active learning classifier training")804 start_time = time.time()805 806 # Get current schema for training807 current_schema = self.schema_cycler.get_current_schema()808 if not current_schema:809 self.logger.warning("No schema available for training")810 return811 812 # Get current annotation state813 item_manager = get_item_state_manager()814 user_manager = get_user_state_manager()815 816 # Collect training data817 training_data = self._collect_training_data(item_manager, user_manager, current_schema)818 819 if not training_data:820 self.logger.warning(f"No training data available for schema {current_schema}")821 # If in cold-start phase, try LLM-based reordering822 if self.config.cold_start_strategy == "llm" and self.config.llm_enabled:823 self._cold_start_reorder(item_manager)824 return825 826 # Train classifier827 model, metrics = self._train_classifier(training_data, current_schema)828 829 if model:830 self._models[current_schema] = model831 self._annotated_texts[current_schema] = training_data["texts"]832 833 # Save model if persistence is enabled834 if self.model_persistence:835 try:836 model_path = self.model_persistence.save_model(837 model, current_schema, len(training_data["texts"])838 )839 metrics.model_file_path = model_path840 except Exception as e:841 self.logger.error(f"Failed to save model: {e}")842 843 # Save metrics to database if available844 if self.database_manager:845 try:846 self.database_manager.save_training_metrics(metrics)847 except Exception as e:848 self.logger.error(f"Failed to save metrics: {e}")849 850 # Reorder instances851 self._reorder_instances(item_manager, current_schema)852 853 # Advance to next schema854 self.schema_cycler.advance_schema()855 856 self._training_count += 1857 self._last_training_time = time.time()858 859 training_duration = time.time() - start_time860 self.logger.info(f"Active learning training completed for schema {current_schema} "861 f"(run #{self._training_count}, duration: {training_duration:.2f}s)")862 else:863 self.logger.warning(f"Failed to train model for schema {current_schema}")864 # Try cold-start if not enough data865 if (self.config.cold_start_strategy == "llm"866 and self.config.llm_enabled867 and len(training_data.get("texts", [])) < self.config.min_instances_for_training):868 self._cold_start_reorder(item_manager)869 870 except Exception as e:871 self.logger.error(f"Error during training: {e}")872 # Continue without failing the entire system873 874 def _collect_training_data(self, item_manager: ItemStateManager, user_manager, schema_name: str) -> Dict:875 """Collect training data for a specific schema."""876 training_data = {"texts": [], "labels": [], "instance_ids": []}877 878 # Get all user states879 user_states = user_manager.get_all_users()880 self.logger.debug(f"Found {len(user_states)} user states")881 882 # Collect annotations per instance883 instance_annotations = defaultdict(list)884 885 for user_state in user_states:886 user_annotations = user_state.get_all_annotations()887 self.logger.debug(f"User {user_state.user_id} has {len(user_annotations)} annotations")888 for instance_id, annotations in user_annotations.items():889 # Check if the schema exists in the labels section890 if 'labels' in annotations:891 labels_dict = annotations['labels']892 # Handle Label objects as keys893 for label_obj, value in labels_dict.items():894 if hasattr(label_obj, 'get_schema') and label_obj.get_schema() == schema_name:895 instance_annotations[instance_id].append({896 "label": label_obj.get_name(),897 "value": value,898 "user": user_state.user_id899 })900 901 self.logger.debug(f"Collected annotations for {len(instance_annotations)} instances")902 903 # Filter instances with sufficient annotations904 for instance_id, annotations in instance_annotations.items():905 if len(annotations) >= self.config.min_annotations_per_instance:906 # Resolve multiple annotations907 resolved_label = self._resolve_annotations(annotations)908 if resolved_label:909 item = item_manager.get_item(instance_id)910 if item:911 text = item.get_text()912 training_data["texts"].append(text)913 training_data["labels"].append(resolved_label)914 training_data["instance_ids"].append(instance_id)915 916 self.logger.debug(f"Training data collected: {len(training_data['texts'])} texts, {len(training_data['labels'])} labels")917 return training_data918 919 def _resolve_annotations(self, annotations: List[Dict]) -> Optional[str]:920 """Resolve multiple annotations using the configured strategy."""921 if not annotations:922 return None923 924 if self.config.resolution_strategy == ResolutionStrategy.MAJORITY_VOTE:925 return self._majority_vote(annotations)926 elif self.config.resolution_strategy == ResolutionStrategy.RANDOM:927 return self._random_selection(annotations)928 elif self.config.resolution_strategy == ResolutionStrategy.CONSENSUS:929 return self._consensus_resolution(annotations)930 else:931 return self._majority_vote(annotations) # Default fallback932 933 def _majority_vote(self, annotations: List[Dict]) -> str:934 """Resolve annotations using majority vote with random tie-breaking."""935 label_counts = Counter(ann["label"] for ann in annotations)936 max_count = max(label_counts.values())937 # Find all labels with the maximum count (handles ties)938 tied_labels = [label for label, count in label_counts.items() if count == max_count]939 # Break ties randomly940 return random.choice(tied_labels)941 942 def _random_selection(self, annotations: List[Dict]) -> str:943 """Resolve annotations by random selection."""944 return random.choice(annotations)["label"]945 946 def _consensus_resolution(self, annotations: List[Dict]) -> Optional[str]:947 """Resolve annotations by consensus (all must agree)."""948 labels = [ann["label"] for ann in annotations]949 if len(set(labels)) == 1:950 return labels[0]951 return None952 953 def _train_classifier(self, training_data: Dict, schema_name: str) -> Tuple[Optional[Pipeline], TrainingMetrics]:954 """Train a classifier for a specific schema."""955 start_time = time.time()956 957 if len(training_data["texts"]) < self.config.min_instances_for_training:958 error_msg = f"Insufficient training data for schema {schema_name}: {len(training_data['texts'])} < {self.config.min_instances_for_training}"959 self.logger.warning(error_msg)960 return None, TrainingMetrics(961 schema_name=schema_name,962 training_time=time.time() - start_time,963 accuracy=0.0,964 instance_count=len(training_data["texts"]),965 timestamp=datetime.now(),966 error_message=error_msg967 )968 969 # Check for sufficient label diversity970 unique_labels = set(training_data["labels"])971 if len(unique_labels) < 2:972 error_msg = f"Insufficient label diversity for schema {schema_name}: {len(unique_labels)} unique labels"973 self.logger.warning(error_msg)974 return None, TrainingMetrics(975 schema_name=schema_name,976 training_time=time.time() - start_time,977 accuracy=0.0,978 instance_count=len(training_data["texts"]),979 timestamp=datetime.now(),980 error_message=error_msg981 )982 983 try:984 # Create and train classifier985 classifier = self._create_classifier()986 vectorizer = self._create_vectorizer()987 988 pipeline = Pipeline([989 ("vectorizer", vectorizer),990 ("classifier", classifier)991 ])992 993 pipeline.fit(training_data["texts"], training_data["labels"])994 995 # Apply probability calibration if enabled996 if self.config.calibrate_probabilities and hasattr(classifier, 'predict_proba'):997 num_samples = len(training_data["texts"])998 if num_samples >= 5:999 try:1000 from sklearn.calibration import CalibratedClassifierCV1001 cv_folds = min(3, num_samples // 2)1002 if cv_folds >= 2:1003 calibrated = CalibratedClassifierCV(1004 pipeline, cv=cv_folds, method='isotonic'1005 )1006 calibrated.fit(training_data["texts"], training_data["labels"])1007 pipeline = calibrated1008 self.logger.debug(f"Applied probability calibration with {cv_folds}-fold CV")1009 except Exception as e:1010 self.logger.warning(f"Calibration failed, using uncalibrated model: {e}")1011 1012 # Store vectorizer separately for strategy use1013 self._vectorizers[schema_name] = pipeline.named_steps.get("vectorizer", vectorizer) if hasattr(pipeline, 'named_steps') else vectorizer1014 1015 # Train BALD ensemble if needed1016 if self.config.query_strategy == "bald":1017 self._train_bald_ensemble(training_data, schema_name)1018 1019 # Calculate accuracy1020 predictions = pipeline.predict(training_data["texts"])1021 accuracy = accuracy_score(training_data["labels"], predictions)1022 1023 # Calculate confidence distribution1024 confidence_distribution = self._calculate_confidence_distribution(pipeline, training_data["texts"])1025 1026 training_time = time.time() - start_time1027 1028 metrics = TrainingMetrics(1029 schema_name=schema_name,1030 training_time=training_time,1031 accuracy=accuracy,1032 instance_count=len(training_data["texts"]),1033 timestamp=datetime.now(),1034 confidence_distribution=confidence_distribution1035 )1036 1037 self.logger.info(f"Trained classifier for schema {schema_name} with {len(training_data['texts'])} instances, "1038 f"accuracy: {accuracy:.3f}, time: {training_time:.2f}s")1039 1040 return pipeline, metrics1041 1042 except Exception as e:1043 error_msg = f"Error training classifier for schema {schema_name}: {e}"1044 self.logger.error(error_msg)1045 return None, TrainingMetrics(1046 schema_name=schema_name,1047 training_time=time.time() - start_time,1048 accuracy=0.0,1049 instance_count=len(training_data["texts"]),1050 timestamp=datetime.now(),1051 error_message=error_msg1052 )1053 1054 def _train_bald_ensemble(self, training_data: Dict, schema_name: str):1055 """Train an ensemble of classifiers for BALD strategy."""1056 params = self.config.bald_params1057 n_estimators = params.get("n_estimators", 5)1058 bootstrap_fraction = params.get("bootstrap_fraction", 0.8)1059 1060 texts = training_data["texts"]1061 labels = training_data["labels"]1062 n_samples = len(texts)1063 bootstrap_size = max(2, int(n_samples * bootstrap_fraction))1064 1065 ensemble = []1066 for i in range(n_estimators):1067 indices = np.random.choice(n_samples, size=bootstrap_size, replace=True)1068 boot_texts = [texts[j] for j in indices]1069 boot_labels = [labels[j] for j in indices]1070 1071 # Need at least 2 classes1072 if len(set(boot_labels)) < 2:1073 continue1074 1075 clf = self._create_classifier()1076 vec = self._create_vectorizer()1077 pipe = Pipeline([("vectorizer", vec), ("classifier", clf)])1078 pipe.fit(boot_texts, boot_labels)1079 ensemble.append(pipe)1080 1081 if ensemble:1082 self._bald_ensembles[schema_name] = ensemble1083 self.logger.info(f"Trained BALD ensemble with {len(ensemble)} models for {schema_name}")1084 1085 def _calculate_confidence_distribution(self, pipeline, texts: List[str]) -> Dict[str, float]:1086 """Calculate confidence score distribution."""1087 try:1088 probas = pipeline.predict_proba(texts)1089 max_confidences = np.max(probas, axis=1)1090 1091 # Create histogram bins1092 bins = [0.0, 0.2, 0.4, 0.6, 0.8, 1.0]1093 hist, _ = np.histogram(max_confidences, bins=bins)1094 1095 # Convert to percentages1096 total = len(max_confidences)1097 distribution = {}1098 for i, count in enumerate(hist):1099 bin_label = f"{bins[i]:.1f}-{bins[i+1]:.1f}"1100 distribution[bin_label] = (count / total) * 100 if total > 0 else 01101 1102 return distribution1103 except Exception as e:1104 self.logger.warning(f"Failed to calculate confidence distribution: {e}")1105 return {}1106 1107 def _create_classifier(self):1108 """Create classifier instance based on configuration."""1109 kwargs = dict(self.config.classifier_kwargs)1110 1111 if self.config.classifier_name == "sklearn.linear_model.LogisticRegression":1112 return LogisticRegression(**kwargs)1113 elif self.config.classifier_name == "sklearn.ensemble.RandomForestClassifier":1114 return RandomForestClassifier(**kwargs)1115 elif self.config.classifier_name == "sklearn.svm.SVC":1116 kwargs.setdefault("probability", True)1117 return SVC(**kwargs)1118 else:1119 # Try to import dynamically1120 try:1121 module_name, class_name = self.config.classifier_name.rsplit('.', 1)1122 module = __import__(module_name, fromlist=[class_name])1123 classifier_class = getattr(module, class_name)1124 return classifier_class(**kwargs)1125 except Exception as e:1126 self.logger.error(f"Failed to create classifier {self.config.classifier_name}: {e}")1127 return LogisticRegression() # Fallback1128 1129 def _create_vectorizer(self):1130 """Create vectorizer instance based on configuration."""1131 kwargs = dict(self.config.vectorizer_kwargs)1132 1133 if self.config.vectorizer_name == "sklearn.feature_extraction.text.CountVectorizer":1134 return CountVectorizer(**kwargs)1135 elif self.config.vectorizer_name == "sklearn.feature_extraction.text.TfidfVectorizer":1136 return TfidfVectorizer(**kwargs)1137 elif self.config.vectorizer_name == "sentence-transformers":1138 model_name = kwargs.pop("model_name", "all-MiniLM-L6-v2")1139 return SentenceTransformerVectorizer(model_name=model_name)1140 else:1141 # Try to import dynamically1142 try:1143 module_name, class_name = self.config.vectorizer_name.rsplit('.', 1)1144 module = __import__(module_name, fromlist=[class_name])1145 vectorizer_class = getattr(module, class_name)1146 return vectorizer_class(**kwargs)1147 except Exception as e:1148 self.logger.error(f"Failed to create vectorizer {self.config.vectorizer_name}: {e}")1149 return TfidfVectorizer() # Fallback1150 1151 def _reorder_instances(self, item_manager: ItemStateManager, schema_name: str):1152 """Reorder instances based on the configured query strategy."""1153 if schema_name not in self._models:1154 self.logger.warning(f"No trained model available for schema {schema_name}")1155 return1156 1157 # Get unlabeled instances1158 unlabeled_instances = []1159 unlabeled_texts = []1160 for instance_id in item_manager.get_instance_ids():1161 if not item_manager.get_annotators_for_item(instance_id):1162 item = item_manager.get_item(instance_id)1163 if item:1164 unlabeled_instances.append(instance_id)1165 unlabeled_texts.append(item.get_text())1166 1167 if not unlabeled_texts:1168 self.logger.info("No unlabeled instances to reorder")1169 return1170 1171 # Limit number of instances to process1172 if self.config.max_instances_to_reorder:1173 limit = self.config.max_instances_to_reorder1174 unlabeled_instances = unlabeled_instances[:limit]1175 unlabeled_texts = unlabeled_texts[:limit]1176 1177 model = self._models[schema_name]1178 annotated = self._annotated_texts.get(schema_name, [])1179 1180 # Get rankings from strategy1181 if (self.config.query_strategy == "bald"1182 and schema_name in self._bald_ensembles1183 and isinstance(self._query_strategy, BaldStrategy)):1184 vectorizer = self._vectorizers.get(schema_name)1185 if vectorizer:1186 rankings = self._query_strategy.rank_with_ensemble(1187 unlabeled_texts, self._bald_ensembles[schema_name], vectorizer1188 )1189 else:1190 rankings = self._query_strategy.rank(unlabeled_texts, model, model, annotated)1191 else:1192 # Extract vectorizer and classifier from pipeline for strategy use1193 vectorizer = self._vectorizers.get(schema_name)1194 classifier = model1195 if vectorizer:1196 rankings = self._query_strategy.rank(1197 unlabeled_texts, classifier, vectorizer, annotated1198 )1199 else:1200 # Fallback: use confidence scores directly