Buckets:
| #!/usr/bin/env python3 | |
| import os | |
| import json | |
| from typing import Dict, Any, List, Optional | |
| from dataclasses import dataclass, asdict | |
| from pathlib import Path | |
| import logging | |
| class MemorySystemConfig: | |
| initial_capacity: int = 16 | |
| max_capacity: int = 10000 | |
| load_factor_threshold: float = 0.75 | |
| similarity_threshold: float = 0.8 | |
| compression_level: int = 6 | |
| cache_size: int = 50 | |
| pattern_dimension: int = 50 | |
| max_iterations: int = 100 | |
| threshold: float = 0.5 | |
| class LLMConfig: | |
| gemini_api_key: str = "" | |
| gemini_model: str = "gemini-1.5-pro" | |
| temperature: float = 0.7 | |
| max_tokens: int = 2048 | |
| max_memory_size: int = 10000 | |
| memory_compression_threshold: float = 0.8 | |
| similarity_threshold: float = 0.75 | |
| redis_host: str = "localhost" | |
| redis_port: int = 6379 | |
| redis_db: int = 0 | |
| enable_streaming: bool = True | |
| enable_memory_summarization: bool = True | |
| summary_threshold: int = 50 | |
| enable_workflow_optimization: bool = True | |
| max_concurrent_requests: int = 10 | |
| workflow_timeout: int = 30 | |
| class AIBenchmarkConfig: | |
| semantic_accuracy_iterations: int = 5 | |
| workflow_efficiency_iterations: int = 3 | |
| concurrent_requests: int = 10 | |
| embedding_quality_tests: int = 5 | |
| min_semantic_accuracy: float = 0.7 | |
| max_llm_response_time: float = 5.0 | |
| max_error_rate: float = 0.1 | |
| test_dataset_size: int = 1000 | |
| enable_load_testing: bool = True | |
| enable_scalability_testing: bool = True | |
| class BenchmarkConfig: | |
| dataset_sizes: List[int] = None | |
| data_types: List[str] = None | |
| operations: List[str] = None | |
| iterations: int = 3 | |
| max_size_scalability: int = 5000 | |
| step_size_scalability: int = 500 | |
| workload_pattern_size: int = 1000 | |
| def __post_init__(self): | |
| if self.dataset_sizes is None: | |
| self.dataset_sizes = [100, 500, 1000, 2000] | |
| if self.data_types is None: | |
| self.data_types = ["mixed", "simple", "numeric", "text"] | |
| if self.operations is None: | |
| self.operations = ["store", "retrieve", "mixed"] | |
| class VisualizationConfig: | |
| output_dir: str = "visualizations" | |
| figure_size: tuple = (15, 12) | |
| dpi: int = 300 | |
| style: str = "seaborn-v0_8" | |
| color_palette: str = "husl" | |
| save_format: str = "png" | |
| show_plots: bool = False | |
| class LoggingConfig: | |
| level: str = "INFO" | |
| format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" | |
| file: Optional[str] = None | |
| console: bool = True | |
| class ProjectConfig: | |
| project_name: str = "Memory Systems in AI" | |
| version: str = "1.0.0" | |
| author: str = "AI Memory Systems Research" | |
| description: str = ( | |
| "Comprehensive implementation and analysis of memory systems for AI applications" | |
| ) | |
| base_dir: str = "." | |
| results_dir: str = "benchmark_results" | |
| visualizations_dir: str = "visualizations" | |
| logs_dir: str = "logs" | |
| data_dir: str = "data" | |
| memory_systems: MemorySystemConfig = None | |
| llm_config: LLMConfig = None | |
| ai_benchmark: AIBenchmarkConfig = None | |
| benchmark: BenchmarkConfig = None | |
| visualization: VisualizationConfig = None | |
| logging: LoggingConfig = None | |
| def __post_init__(self): | |
| if self.memory_systems is None: | |
| self.memory_systems = MemorySystemConfig() | |
| if self.llm_config is None: | |
| self.llm_config = LLMConfig() | |
| if self.ai_benchmark is None: | |
| self.ai_benchmark = AIBenchmarkConfig() | |
| if self.benchmark is None: | |
| self.benchmark = BenchmarkConfig() | |
| if self.visualization is None: | |
| self.visualization = VisualizationConfig() | |
| if self.logging is None: | |
| self.logging = LoggingConfig() | |
| class ConfigManager: | |
| def __init__(self, config_file: Optional[str] = None): | |
| self.config_file = config_file or "config.json" | |
| self.config = self._load_config() | |
| self._setup_logging() | |
| self._ensure_directories() | |
| def _load_config(self) -> ProjectConfig: | |
| if os.path.exists(self.config_file): | |
| try: | |
| with open(self.config_file, "r") as f: | |
| config_data = json.load(f) | |
| return self._dict_to_config(config_data) | |
| except Exception as e: | |
| print(f"โ ๏ธ Error loading config file: {e}") | |
| print("๐ Using default configuration") | |
| return ProjectConfig() | |
| else: | |
| print("๐ No config file found, using default configuration") | |
| return ProjectConfig() | |
| def _dict_to_config(self, data: Dict[str, Any]) -> ProjectConfig: | |
| memory_systems_data = data.get("memory_systems", {}) | |
| benchmark_data = data.get("benchmark", {}) | |
| visualization_data = data.get("visualization", {}) | |
| logging_data = data.get("logging", {}) | |
| config = ProjectConfig( | |
| project_name=data.get("project_name", "Memory Systems in AI"), | |
| version=data.get("version", "1.0.0"), | |
| author=data.get("author", "AI Memory Systems Research"), | |
| description=data.get( | |
| "description", | |
| "Comprehensive implementation and analysis of memory systems for AI applications", | |
| ), | |
| base_dir=data.get("base_dir", "."), | |
| results_dir=data.get("results_dir", "benchmark_results"), | |
| visualizations_dir=data.get("visualizations_dir", "visualizations"), | |
| logs_dir=data.get("logs_dir", "logs"), | |
| data_dir=data.get("data_dir", "data"), | |
| memory_systems=MemorySystemConfig(**memory_systems_data), | |
| benchmark=BenchmarkConfig(**benchmark_data), | |
| visualization=VisualizationConfig(**visualization_data), | |
| logging=LoggingConfig(**logging_data), | |
| ) | |
| return config | |
| def _setup_logging(self): | |
| log_level = getattr(logging, self.config.logging.level.upper(), logging.INFO) | |
| formatter = logging.Formatter(self.config.logging.format) | |
| root_logger = logging.getLogger() | |
| root_logger.setLevel(log_level) | |
| root_logger.handlers.clear() | |
| if self.config.logging.console: | |
| console_handler = logging.StreamHandler() | |
| console_handler.setFormatter(formatter) | |
| root_logger.addHandler(console_handler) | |
| if self.config.logging.file: | |
| file_path = os.path.join(self.config.logs_dir, self.config.logging.file) | |
| file_handler = logging.FileHandler(file_path) | |
| file_handler.setFormatter(formatter) | |
| root_logger.addHandler(file_handler) | |
| def _ensure_directories(self): | |
| directories = [ | |
| self.config.results_dir, | |
| self.config.visualizations_dir, | |
| self.config.logs_dir, | |
| self.config.data_dir, | |
| ] | |
| for directory in directories: | |
| Path(directory).mkdir(parents=True, exist_ok=True) | |
| def save_config(self, filename: Optional[str] = None): | |
| filepath = filename or self.config_file | |
| config_dict = self._config_to_dict(self.config) | |
| with open(filepath, "w") as f: | |
| json.dump(config_dict, f, indent=2) | |
| print(f"๐พ Configuration saved to {filepath}") | |
| def _config_to_dict(self, config: ProjectConfig) -> Dict[str, Any]: | |
| config_dict = asdict(config) | |
| return config_dict | |
| def get_memory_system_config(self, system_type: str) -> Dict[str, Any]: | |
| base_config = asdict(self.config.memory_systems) | |
| system_configs = { | |
| "sequential": {"initial_capacity": 10, "load_factor": 0.75}, | |
| "associative": {"initial_capacity": 16, "load_factor_threshold": 0.75}, | |
| "content_addressable": {"similarity_threshold": 0.8}, | |
| "adaptive_lru": {"max_size": 100, "initial_size": 50}, | |
| "neural_associative": { | |
| "memory_size": 100, | |
| "pattern_dimension": 50, | |
| "max_iterations": 100, | |
| "threshold": 0.5, | |
| }, | |
| "compressed": {"compression_level": 6, "cache_size": 50}, | |
| "hierarchical": { | |
| }, | |
| } | |
| system_specific = system_configs.get(system_type, {}) | |
| base_config.update(system_specific) | |
| return base_config | |
| def update_config(self, **kwargs): | |
| for key, value in kwargs.items(): | |
| if hasattr(self.config, key): | |
| setattr(self.config, key, value) | |
| else: | |
| print(f"โ ๏ธ Unknown configuration key: {key}") | |
| def get_environment_config(self) -> Dict[str, Any]: | |
| env_config = {} | |
| if os.getenv("MEMORY_INITIAL_CAPACITY"): | |
| env_config["memory_systems"] = { | |
| "initial_capacity": int(os.getenv("MEMORY_INITIAL_CAPACITY")) | |
| } | |
| if os.getenv("MEMORY_MAX_CAPACITY"): | |
| env_config["memory_systems"] = env_config.get("memory_systems", {}) | |
| env_config["memory_systems"]["max_capacity"] = int( | |
| os.getenv("MEMORY_MAX_CAPACITY") | |
| ) | |
| if os.getenv("BENCHMARK_ITERATIONS"): | |
| env_config["benchmark"] = { | |
| "iterations": int(os.getenv("BENCHMARK_ITERATIONS")) | |
| } | |
| if os.getenv("BENCHMARK_MAX_SIZE"): | |
| env_config["benchmark"] = env_config.get("benchmark", {}) | |
| env_config["benchmark"]["max_size_scalability"] = int( | |
| os.getenv("BENCHMARK_MAX_SIZE") | |
| ) | |
| if os.getenv("LOG_LEVEL"): | |
| env_config["logging"] = {"level": os.getenv("LOG_LEVEL")} | |
| return env_config | |
| def apply_environment_config(self): | |
| env_config = self.get_environment_config() | |
| for section, values in env_config.items(): | |
| if hasattr(self.config, section): | |
| section_obj = getattr(self.config, section) | |
| for key, value in values.items(): | |
| if hasattr(section_obj, key): | |
| setattr(section_obj, key, value) | |
| def validate_config(self) -> List[str]: | |
| issues = [] | |
| if self.config.memory_systems.initial_capacity <= 0: | |
| issues.append("Memory system initial capacity must be positive") | |
| if ( | |
| self.config.memory_systems.max_capacity | |
| <= self.config.memory_systems.initial_capacity | |
| ): | |
| issues.append( | |
| "Memory system max capacity must be greater than initial capacity" | |
| ) | |
| if not 0 < self.config.memory_systems.load_factor_threshold < 1: | |
| issues.append("Load factor threshold must be between 0 and 1") | |
| if not 0 < self.config.memory_systems.similarity_threshold <= 1: | |
| issues.append("Similarity threshold must be between 0 and 1") | |
| if not self.config.benchmark.dataset_sizes: | |
| issues.append("Benchmark dataset sizes cannot be empty") | |
| if any(size <= 0 for size in self.config.benchmark.dataset_sizes): | |
| issues.append("All benchmark dataset sizes must be positive") | |
| if self.config.benchmark.iterations <= 0: | |
| issues.append("Benchmark iterations must be positive") | |
| if self.config.visualization.dpi <= 0: | |
| issues.append("Visualization DPI must be positive") | |
| if ( | |
| self.config.visualization.figure_size[0] <= 0 | |
| or self.config.visualization.figure_size[1] <= 0 | |
| ): | |
| issues.append("Visualization figure size must be positive") | |
| return issues | |
| def print_config_summary(self): | |
| print("๐ Configuration Summary") | |
| print("=" * 40) | |
| print(f"Project: {self.config.project_name} v{self.config.version}") | |
| print(f"Author: {self.config.author}") | |
| print(f"Description: {self.config.description}") | |
| print() | |
| print("๐ Directories:") | |
| print(f" Base: {self.config.base_dir}") | |
| print(f" Results: {self.config.results_dir}") | |
| print(f" Visualizations: {self.config.visualizations_dir}") | |
| print(f" Logs: {self.config.logs_dir}") | |
| print(f" Data: {self.config.data_dir}") | |
| print() | |
| print("๐ง Memory Systems:") | |
| print(f" Initial Capacity: {self.config.memory_systems.initial_capacity}") | |
| print(f" Max Capacity: {self.config.memory_systems.max_capacity}") | |
| print( | |
| f" Load Factor Threshold: {self.config.memory_systems.load_factor_threshold}" | |
| ) | |
| print( | |
| f" Similarity Threshold: {self.config.memory_systems.similarity_threshold}" | |
| ) | |
| print() | |
| print("๐ Benchmark:") | |
| print(f" Dataset Sizes: {self.config.benchmark.dataset_sizes}") | |
| print(f" Data Types: {self.config.benchmark.data_types}") | |
| print(f" Operations: {self.config.benchmark.operations}") | |
| print(f" Iterations: {self.config.benchmark.iterations}") | |
| print() | |
| print("๐ Visualization:") | |
| print(f" Output Directory: {self.config.visualization.output_dir}") | |
| print(f" Figure Size: {self.config.visualization.figure_size}") | |
| print(f" DPI: {self.config.visualization.dpi}") | |
| print(f" Style: {self.config.visualization.style}") | |
| print() | |
| print("๐ Logging:") | |
| print(f" Level: {self.config.logging.level}") | |
| print(f" Console: {self.config.logging.console}") | |
| print(f" File: {self.config.logging.file}") | |
| def create_default_config_file(filename: str = "config.json"): | |
| config_manager = ConfigManager() | |
| config_manager.save_config(filename) | |
| print(f"๐ Default configuration file created: {filename}") | |
| def load_config_from_file(filename: str) -> ConfigManager: | |
| config_manager = ConfigManager(filename) | |
| return config_manager | |
| if __name__ == "__main__": | |
| print("๐ง Memory Systems Configuration Manager") | |
| print("=" * 50) | |
| config_manager = ConfigManager() | |
| config_manager.print_config_summary() | |
| issues = config_manager.validate_config() | |
| if issues: | |
| print("\nโ ๏ธ Configuration Issues:") | |
| for issue in issues: | |
| print(f" - {issue}") | |
| else: | |
| print("\nโ Configuration is valid") | |
| config_manager.save_config() | |
| print("\n๐ก Usage Examples:") | |
| print(" - Load config: config_manager = ConfigManager('my_config.json')") | |
| print( | |
| " - Get system config: config_manager.get_memory_system_config('associative')" | |
| ) | |
| print(" - Update config: config_manager.update_config(results_dir='new_results')") | |
| print(" - Save config: config_manager.save_config('updated_config.json')") | |
Xet Storage Details
- Size:
- 14.7 kB
- Xet hash:
- d9e1fa154bd6482ad6d36d48f456588c6813f5e249df2baa6339f54809ac7f6d
ยท
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.