tahamajs's picture
download
raw
14.7 kB
#!/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
@dataclass
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
@dataclass
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
@dataclass
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
@dataclass
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"]
@dataclass
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
@dataclass
class LoggingConfig:
level: str = "INFO"
format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
file: Optional[str] = None
console: bool = True
@dataclass
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.