Spaces:
Running
Running
File size: 5,064 Bytes
ba68fc1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 |
#!/usr/bin/env python3
"""
Centralized configuration management for GAIA agent.
"""
import os
from typing import Dict, Optional, Any
from dataclasses import dataclass, field
from enum import Enum
from dotenv import load_dotenv
class ModelType(Enum):
"""Available model types."""
KLUSTER = "kluster"
GEMINI = "gemini"
QWEN = "qwen"
class AgentType(Enum):
"""Available agent types."""
MULTIMEDIA = "multimedia"
RESEARCH = "research"
LOGIC_MATH = "logic_math"
FILE_PROCESSING = "file_processing"
CHESS = "chess"
GENERAL = "general"
@dataclass
class ModelConfig:
"""Configuration for AI models."""
# Model names
GEMINI_MODEL: str = "gemini/gemini-2.0-flash"
QWEN_MODEL: str = "Qwen/Qwen2.5-72B-Instruct"
CLASSIFICATION_MODEL: str = "Qwen/Qwen2.5-7B-Instruct"
# Kluster.ai models
KLUSTER_MODELS: Dict[str, str] = field(default_factory=lambda: {
"gemma3-27b": "openai/google/gemma-3-27b-it",
"qwen3-235b": "openai/Qwen/Qwen3-235B-A22B-FP8",
"qwen2.5-72b": "openai/Qwen/Qwen2.5-72B-Instruct",
"llama3.1-405b": "openai/meta-llama/Meta-Llama-3.1-405B-Instruct"
})
# API endpoints
KLUSTER_API_BASE: str = "https://api.kluster.ai/v1"
# Model parameters
MAX_STEPS: int = 12
VERBOSITY_LEVEL: int = 2
TEMPERATURE: float = 0.7
MAX_TOKENS: int = 4000
# Retry settings
MAX_RETRIES: int = 3
BASE_DELAY: float = 2.0
# Memory management
ENABLE_FRESH_AGENTS: bool = True
ENABLE_TOKEN_MANAGEMENT: bool = True
@dataclass
class ToolConfig:
"""Configuration for tools."""
# File processing limits
MAX_FILE_SIZE: int = 100 * 1024 * 1024 # 100MB
MAX_FRAMES: int = 10
MAX_PROCESSING_TIME: int = 1800 # 30 minutes
# Cache settings
CACHE_TTL: int = 900 # 15 minutes
ENABLE_CACHING: bool = True
# Search settings
MAX_SEARCH_RESULTS: int = 10
SEARCH_TIMEOUT: int = 30
# YouTube settings
YOUTUBE_QUALITY: str = "medium"
MAX_VIDEO_DURATION: int = 3600 # 1 hour
@dataclass
class UIConfig:
"""Configuration for user interfaces."""
# Gradio settings
SERVER_NAME: str = "0.0.0.0"
SERVER_PORT: int = 7860
SHARE: bool = False
# Interface limits
MAX_QUESTION_LENGTH: int = 5000
MAX_QUESTIONS_BATCH: int = 20
DEMO_MODE: bool = False
class Config:
"""Centralized configuration management."""
def __init__(self):
# Load environment variables
load_dotenv()
# Initialize configurations
self.model = ModelConfig()
self.tools = ToolConfig()
self.ui = UIConfig()
# API keys
self._api_keys = self._load_api_keys()
# Validation
self._validate_config()
def _load_api_keys(self) -> Dict[str, Optional[str]]:
"""Load API keys from environment."""
return {
"gemini": os.getenv("GEMINI_API_KEY"),
"huggingface": os.getenv("HUGGINGFACE_TOKEN"),
"kluster": os.getenv("KLUSTER_API_KEY"),
"serpapi": os.getenv("SERPAPI_API_KEY")
}
def _validate_config(self) -> None:
"""Validate configuration and API keys."""
if not any(self._api_keys.values()):
raise ValueError(
"At least one API key must be provided: "
"GEMINI_API_KEY, HUGGINGFACE_TOKEN, or KLUSTER_API_KEY"
)
def get_api_key(self, provider: str) -> Optional[str]:
"""Get API key for specific provider."""
return self._api_keys.get(provider.lower())
def has_api_key(self, provider: str) -> bool:
"""Check if API key exists for provider."""
key = self.get_api_key(provider)
return key is not None and len(key.strip()) > 0
def get_available_models(self) -> list[ModelType]:
"""Get list of available models based on API keys."""
available = []
if self.has_api_key("kluster"):
available.append(ModelType.KLUSTER)
if self.has_api_key("gemini"):
available.append(ModelType.GEMINI)
if self.has_api_key("huggingface"):
available.append(ModelType.QWEN)
return available
def get_fallback_chain(self) -> list[ModelType]:
"""Get model fallback chain based on availability."""
available = self.get_available_models()
# Prefer Kluster -> Gemini -> Qwen
priority_order = [ModelType.KLUSTER, ModelType.GEMINI, ModelType.QWEN]
return [model for model in priority_order if model in available]
@property
def debug_mode(self) -> bool:
"""Check if debug mode is enabled."""
return os.getenv("DEBUG", "false").lower() == "true"
@property
def log_level(self) -> str:
"""Get logging level."""
return os.getenv("LOG_LEVEL", "INFO").upper()
# Global configuration instance
config = Config() |