monoai.conf

  8class Conf:
  9    """
 10    A singleton class for managing configuration in a programmatic way.
 11    Configuration defined here overrides the configuration in ai.yaml file.
 12
 13    Examples
 14    --------
 15    ```
 16    from monoai.conf import Conf
 17    
 18    # override the base model
 19    Conf()["base_model"] = {
 20        "provider": "openai",
 21        "model": "gpt-4o-mini"
 22    }
 23    
 24    # set prompts path programmatically
 25    Conf()["prompts_path"] = "prompts"
 26    ```
 27    """
 28    
 29    _instance = None
 30    _DEFAULT_CONFIG = {
 31        'keysfile_path': "providers.keys",
 32        'supported_files': {
 33            "text": ["txt", "py", "md"],
 34            "image": ["png", "jpg", "jpeg", "gif", "webp"]
 35        },
 36        'prompts_path': "",
 37        'default_prompt': {
 38            "rag": "Use also the following information to answer the question: ",
 39            "summary":"Here is the summary of the conversation so far:\n\n",
 40            "file":"Here is the content of the file:\n\n"
 41        },
 42        'observability': []
 43    }
 44    
 45    def __new__(cls):
 46        """
 47        Create or return the singleton instance.
 48
 49        Returns
 50        -------
 51        Config
 52            The singleton instance
 53        """
 54        if cls._instance is None:
 55            cls._instance = super(Conf, cls).__new__(cls)
 56            cls._instance._initialize()
 57        return cls._instance
 58    
 59    def _initialize(self):
 60        """
 61        Initialize the Config instance.
 62        
 63        Loads and validates the configuration file, setting up defaults
 64        and performing environment variable interpolation.
 65        """
 66        self._config_path = Path('ai.yaml')
 67        self._config = self._DEFAULT_CONFIG.copy()
 68        self._load_config()
 69        self._load_observability_keys()
 70    
 71    def _load_config(self) -> None:
 72        """
 73        Load configuration from the YAML file.
 74        
 75        Handles file reading, YAML parsing, environment variable interpolation,
 76        and configuration validation.
 77
 78        Raises
 79        ------
 80        FileNotFoundError
 81            If the configuration file doesn't exist
 82        yaml.ParserError
 83            If the YAML syntax is invalid
 84        ValueError
 85            If the configuration structure is invalid
 86        """
 87        try:
 88            if self._config_path.exists():
 89                with open(self._config_path, 'r') as f:
 90                    file_config = yaml.safe_load(f)
 91                if file_config:
 92                    self._merge_config(self._config, file_config)            
 93            
 94        except ParserError as e:
 95            raise ValueError(f"Invalid YAML syntax in {self._config_path}: {e}")
 96        except Exception as e:
 97            raise ValueError(f"Error loading configuration: {e}")
 98    
 99    def _load_observability_keys(self) -> None:
100        """
101        Load observability environment variables from observability.keys file.
102        
103        Only loads keys for services that are present in the observability list
104        in the configuration. Matches keys by checking if the first part of the
105        key name matches any service in the observability list.
106        """
107        try:
108            # Check if observability list exists and is not empty
109            observability_services = self._config.get('observability', [])
110            if not observability_services or not isinstance(observability_services, list):
111                return
112            
113            # Load observability.keys file
114            observability_keys_path = Path('observability.keys')
115            if not observability_keys_path.exists():
116                return
117            
118            with open(observability_keys_path, 'r') as f:
119                lines = f.readlines()
120            
121            # Process each line in the keys file
122            for line in lines:
123                line = line.strip()
124                if not line or line.startswith('#'):
125                    continue
126                
127                # Parse key=value format
128                if '=' in line:
129                    key, value = line.split('=', 1)
130                    key = key.strip()
131                    value = value.strip()
132                    
133                    # Check if this key should be loaded based on observability services
134                    if self._should_load_key(key, observability_services):
135                        os.environ[key] = value
136                        
137        except Exception as e:
138            # Don't raise error, just log it as observability is optional
139            print(f"Warning: Failed to load observability keys: {e}")
140    
141    def _should_load_key(self, key: str, observability_services: List[str]) -> bool:
142        """
143        Check if a key should be loaded based on the observability services list.
144        
145        Parameters
146        ----------
147        key : str
148            The environment variable key to check
149        observability_services : List[str]
150            List of observability services to load keys for
151            
152        Returns
153        -------
154        bool
155            True if the key should be loaded, False otherwise
156        """
157        # Extract the first part of the key (before first underscore or dot)
158        key_prefix = key.split('_')[0].split('.')[0].upper()
159        
160        # Check if any service in the list matches the key prefix
161        for service in observability_services:
162            if isinstance(service, str):
163                service_upper = service.upper()
164                if key_prefix == service_upper or key_prefix.startswith(service_upper):
165                    return True
166        
167        return False
168    
169    def _merge_config(self, base: Dict, override: Dict) -> None:
170        """
171        Recursively merge override configuration into base configuration.
172
173        Parameters
174        ----------
175        base : Dict
176            The base configuration dictionary
177        override : Dict
178            The override configuration dictionary
179        """
180        for key, value in override.items():
181            if (
182                key in base and 
183                isinstance(base[key], dict) and 
184                isinstance(value, dict)
185            ):
186                self._merge_config(base[key], value)
187            else:
188                base[key] = value
189        
190        
191    def __getitem__(self, key: str) -> Any:
192        return self._config[key]
193    
194    def __setitem__(self, key: str, value: Any) -> None:
195        self._config[key] = value