#!/usr/bin/env python3 """ Model management system for GAIA agent. Handles model initialization, fallback chains, and lifecycle management. """ import os import time import random from typing import Optional, List, Dict, Any, Union from abc import ABC, abstractmethod from enum import Enum from ..config.settings import Config, ModelType, config from ..utils.exceptions import ( ModelError, ModelNotAvailableError, ModelAuthenticationError, ModelOverloadedError, create_error ) class ModelStatus(Enum): """Model status states.""" AVAILABLE = "available" UNAVAILABLE = "unavailable" OVERLOADED = "overloaded" AUTHENTICATING = "authenticating" ERROR = "error" class ModelProvider(ABC): """Abstract base class for model providers.""" def __init__(self, name: str, model_type: ModelType): self.name = name self.model_type = model_type self.status = ModelStatus.UNAVAILABLE self.last_error: Optional[str] = None self.retry_count = 0 self.last_used = None @abstractmethod def initialize(self) -> bool: """Initialize the model provider. Returns True if successful.""" pass @abstractmethod def is_available(self) -> bool: """Check if the model is available for use.""" pass @abstractmethod def create_model(self, **kwargs): """Create model instance.""" pass def reset_error_state(self) -> None: """Reset error state for retry attempts.""" self.retry_count = 0 self.last_error = None self.status = ModelStatus.UNAVAILABLE def record_usage(self) -> None: """Record model usage timestamp.""" self.last_used = time.time() def handle_error(self, error: Exception) -> None: """Handle and categorize model errors.""" error_str = str(error).lower() if "overloaded" in error_str or "503" in error_str: self.status = ModelStatus.OVERLOADED self.last_error = "Model overloaded" elif "authentication" in error_str or "401" in error_str or "403" in error_str: self.status = ModelStatus.ERROR self.last_error = "Authentication failed" else: self.status = ModelStatus.ERROR self.last_error = str(error) self.retry_count += 1 class LiteLLMProvider(ModelProvider): """Provider for LiteLLM-based models (Gemini, Kluster.ai).""" def __init__(self, model_name: str, api_key: str, api_base: Optional[str] = None): self.model_name = model_name self.api_key = api_key self.api_base = api_base self._model_instance = None model_type = self._determine_model_type(model_name) super().__init__(model_name, model_type) def _determine_model_type(self, model_name: str) -> ModelType: """Determine model type from name.""" if "gemini" in model_name.lower(): return ModelType.GEMINI elif hasattr(self, 'api_base') and self.api_base and "kluster" in str(self.api_base).lower(): return ModelType.KLUSTER else: return ModelType.QWEN def initialize(self) -> bool: """Initialize LiteLLM model.""" try: # Import the class from the same module from .providers import LiteLLMModel self.status = ModelStatus.AUTHENTICATING # Configure environment if self.model_type == ModelType.GEMINI: os.environ["GEMINI_API_KEY"] = self.api_key elif self.api_base: os.environ["OPENAI_API_KEY"] = self.api_key os.environ["OPENAI_API_BASE"] = self.api_base # Create model instance self._model_instance = LiteLLMModel( model_name=self.model_name, api_key=self.api_key, api_base=self.api_base ) self.status = ModelStatus.AVAILABLE return True except Exception as e: self.handle_error(e) return False def is_available(self) -> bool: """Check if model is available.""" return self.status == ModelStatus.AVAILABLE and self._model_instance is not None def create_model(self, **kwargs): """Create model instance.""" if not self.is_available(): raise ModelNotAvailableError(f"Model {self.name} is not available") self.record_usage() return self._model_instance class HuggingFaceProvider(ModelProvider): """Provider for HuggingFace models.""" def __init__(self, model_name: str, api_key: str): super().__init__(model_name, ModelType.QWEN) self.model_name = model_name self.api_key = api_key self._model_instance = None def initialize(self) -> bool: """Initialize HuggingFace model.""" try: from smolagents import InferenceClientModel self.status = ModelStatus.AUTHENTICATING self._model_instance = InferenceClientModel( model_id=self.model_name, token=self.api_key ) self.status = ModelStatus.AVAILABLE return True except Exception as e: self.handle_error(e) return False def is_available(self) -> bool: """Check if model is available.""" return self.status == ModelStatus.AVAILABLE and self._model_instance is not None def create_model(self, **kwargs): """Create model instance.""" if not self.is_available(): raise ModelNotAvailableError(f"Model {self.name} is not available") self.record_usage() return self._model_instance class ModelManager: """Manages model providers and fallback chains.""" def __init__(self, config_instance: Optional[Config] = None): self.config = config_instance or config self.providers: Dict[str, ModelProvider] = {} self.fallback_chain: List[str] = [] self.current_provider: Optional[str] = None self._initialize_providers() def _initialize_providers(self) -> None: """Initialize all available model providers.""" # Kluster.ai models if self.config.has_api_key("kluster"): kluster_key = self.config.get_api_key("kluster") for model_key, model_name in self.config.model.KLUSTER_MODELS.items(): provider_name = f"kluster_{model_key}" provider = LiteLLMProvider( model_name=model_name, api_key=kluster_key, api_base=self.config.model.KLUSTER_API_BASE ) self.providers[provider_name] = provider # Gemini models if self.config.has_api_key("gemini"): gemini_key = self.config.get_api_key("gemini") provider = LiteLLMProvider( model_name=self.config.model.GEMINI_MODEL, api_key=gemini_key ) self.providers["gemini"] = provider # HuggingFace models if self.config.has_api_key("huggingface"): hf_key = self.config.get_api_key("huggingface") provider = HuggingFaceProvider( model_name=self.config.model.QWEN_MODEL, api_key=hf_key ) self.providers["qwen"] = provider # Set up fallback chain self._setup_fallback_chain() def _setup_fallback_chain(self) -> None: """Set up model fallback chain based on availability and preference.""" # Priority order: Kluster.ai (highest tier) -> Gemini -> Qwen priority_providers = [] # Add Kluster.ai models (prefer qwen3-235b) if "kluster_qwen3-235b" in self.providers: priority_providers.append("kluster_qwen3-235b") elif "kluster_gemma3-27b" in self.providers: priority_providers.append("kluster_gemma3-27b") # Add other available providers if "gemini" in self.providers: priority_providers.append("gemini") if "qwen" in self.providers: priority_providers.append("qwen") self.fallback_chain = priority_providers if not self.fallback_chain: raise ModelNotAvailableError("No model providers available") def initialize_all(self) -> Dict[str, bool]: """Initialize all model providers.""" results = {} for name, provider in self.providers.items(): try: success = provider.initialize() results[name] = success if success and self.current_provider is None: self.current_provider = name except Exception as e: results[name] = False provider.handle_error(e) return results def get_current_model(self, **kwargs): """Get current active model.""" if self.current_provider is None: self._select_best_provider() if self.current_provider is None: raise ModelNotAvailableError("No models available") provider = self.providers[self.current_provider] try: return provider.create_model(**kwargs) except Exception as e: provider.handle_error(e) # Try to switch to fallback if self._switch_to_fallback(): return self.get_current_model(**kwargs) else: raise ModelError(f"All models failed: {str(e)}") def _select_best_provider(self) -> None: """Select the best available provider from fallback chain.""" for provider_name in self.fallback_chain: provider = self.providers.get(provider_name) if provider and provider.is_available(): self.current_provider = provider_name return elif provider and provider.status == ModelStatus.UNAVAILABLE: # Try to initialize if provider.initialize(): self.current_provider = provider_name return self.current_provider = None def _switch_to_fallback(self) -> bool: """Switch to next available model in fallback chain.""" if self.current_provider is None: return False try: current_index = self.fallback_chain.index(self.current_provider) # Try next providers in chain for i in range(current_index + 1, len(self.fallback_chain)): provider_name = self.fallback_chain[i] provider = self.providers[provider_name] if provider.is_available() or provider.initialize(): self.current_provider = provider_name return True except ValueError: pass # No fallback available self.current_provider = None return False def retry_current_model(self, max_retries: int = 3) -> bool: """Retry current model with exponential backoff.""" if self.current_provider is None: return False provider = self.providers[self.current_provider] for attempt in range(max_retries): if provider.status == ModelStatus.OVERLOADED: wait_time = (2 ** attempt) + random.random() time.sleep(wait_time) # Reset error state and try to reinitialize provider.reset_error_state() if provider.initialize(): return True return False def get_model_status(self) -> Dict[str, Dict[str, Any]]: """Get status of all model providers.""" status = {} for name, provider in self.providers.items(): status[name] = { "status": provider.status.value, "model_type": provider.model_type.value, "last_error": provider.last_error, "retry_count": provider.retry_count, "last_used": provider.last_used, "is_current": name == self.current_provider } return status def switch_to_provider(self, provider_name: str) -> bool: """Manually switch to specific provider.""" if provider_name not in self.providers: raise ModelNotAvailableError(f"Provider {provider_name} not found") provider = self.providers[provider_name] if provider.is_available() or provider.initialize(): self.current_provider = provider_name return True return False def get_available_providers(self) -> List[str]: """Get list of available providers.""" available = [] for name, provider in self.providers.items(): if provider.is_available(): available.append(name) return available def reset_all_providers(self) -> None: """Reset all providers to allow retry.""" for provider in self.providers.values(): provider.reset_error_state() self.current_provider = None self._select_best_provider() # Monkey patch for smolagents compatibility def monkey_patch_smolagents(): """Apply compatibility patches for smolagents.""" try: import smolagents.monitoring from smolagents.monitoring import TokenUsage # Store original update_metrics function original_update_metrics = smolagents.monitoring.Monitor.update_metrics def patched_update_metrics(self, step_log): """Patched version that handles dict token_usage""" try: # If token_usage is a dict, convert it to TokenUsage object if hasattr(step_log, 'token_usage') and isinstance(step_log.token_usage, dict): token_dict = step_log.token_usage # Create TokenUsage object from dict step_log.token_usage = TokenUsage( input_tokens=token_dict.get('prompt_tokens', 0), output_tokens=token_dict.get('completion_tokens', 0) ) # Call original function return original_update_metrics(self, step_log) except Exception as e: # If patching fails, try to handle gracefully print(f"Token usage patch warning: {e}") return original_update_metrics(self, step_log) # Apply the patch smolagents.monitoring.Monitor.update_metrics = patched_update_metrics print("✅ Applied smolagents token usage compatibility patch") except ImportError: print("⚠️ smolagents not available, skipping compatibility patch") except Exception as e: print(f"⚠️ Failed to apply smolagents patch: {e}") # Apply monkey patch on import monkey_patch_smolagents()