import json import logging from datetime import datetime from pathlib import Path from typing import List, Dict, Any, Optional # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class CapabilityComposer: """ Combines simple capabilities into more complex ones Args: output_root (str): Root directory for saved compositions Attributes: output_path: Path where compositions will be saved compositions: Dictionary of combined capabilities version: Current version number of compositions """ def __init__(self, output_root: str = "composed_capabilities"): self.output_root = Path(output_root) self.output_root.mkdir(exist_ok=True) self.compositions = {} self.version = "1.0" def combine_capabilities(self, base_capability: str, additional_features: List[str], integration_mode: str = "sequential") -> Dict: """ Combine multiple capabilities into a more complex one Args: base_capability (str): Name of the base capability additional_features (List[str]): Additional features to integrate integration_mode (str): How to combine features. Options: 'sequential', 'parallel' Returns: Dict: Combined capability structure """ logger.info(f"Combining {len(additional_features) + 1} capabilities") composition = { "base_capability": base_capability, "additional_features": additional_features, "integration_mode": integration_mode, "version": self.version, "timestamp": datetime.now().isoformat(), "status": "ready" } return composition def validate_composition(self, composition: Dict) -> bool: """ Validate if a capability composition is correct Args: composition (Dict): Composition to validate Returns: bool: True if valid, False otherwise """ if not isinstance(composition, dict): logger.error("Composition must be a dictionary") return False required_keys = ["base_capability", "additional_features"] for key in required_keys: if key not in composition: logger.error(f"Missing required key: {key}") return False if not isinstance(composition["additional_features"], list): logger.error("Additional features must be a list") return False return True def save_composition(self, composition: Dict, name: Optional[str] = None) -> Path: """ Save capability composition to file Args: composition (Dict): Composition to save name (str): Optional custom filename Returns: Path: Path where composition was saved """ try: if not self.validate_composition(composition): raise ValueError("Invalid composition cannot be saved") if not name: name = f"{composition['base_capability']}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" save_path = self.output_root / name with open(save_path, "w") as f: json.dump(composition, f, indent=2) logger.info(f"Saved composition to {save_path}") return save_path except Exception as e: logger.error(f"Failed to save composition: {str(e)}") raise def load_composition(self, filename: str) -> Dict: """ Load a previously saved capability composition Args: filename (str): Name of the file to load Returns: Dict: Loaded composition """ try: filepath = self.output_root / filename if not filepath.exists(): logger.error(f"File {filepath} does not exist") raise FileNotFoundError with open(filepath, "r") as f: composition = json.load(f) if not self.validate_composition(composition): raise ValueError("Loaded composition is invalid") logger.info(f"Loaded composition from {filepath}") return composition except Exception as e: logger.error(f"Failed to load composition: {str(e)}") raise def list_available(self) -> List[str]: """ List all saved compositions Returns: List[str]: Names of all saved compositions """ try: files = [f.name for f in self.output_root.iterdir() if f.is_file()] logger.info(f"Found {len(files)} compositions") return files except Exception as e: logger.error(f"Failed to list compositions: {str(e)}") return [] def get_composition_info(self, filename: str) -> Dict[str, Any]: """ Get metadata about a composition Args: filename (str): Composition file name Returns: Dict[str, Any]: Information about the composition """ try: filepath = self.output_root / filename if not filepath.exists(): logger.error(f"File {filepath} does not exist") return {} with open(filepath, "r") as f: composition = json.load(f) info = { "filename": filename, "base_capability": composition.get("base_capability"), "num_features": len(composition.get("additional_features", [])), "mode": composition.get("integration_mode"), "version": composition.get("version"), "timestamp": composition.get("timestamp") } logger.info(f"Retrieved info for {filename}") return info except Exception as e: logger.error(f"Failed to get composition info: {str(e)}") return {} if __name__ == "__main__": # Example usage logger.info("Testing CapabilityComposer...") composer = CapabilityComposer() # Combine capabilities composition = composer.combine_capabilities( base_capability="basic_search", additional_features=["refinement", "ranking"], integration_mode="sequential" ) # Save composition save_path = composer.save_composition(composition, name="search_enhanced.json") logger.info(f"Test composition saved to {save_path}") # Load and verify loaded = composer.load_composition("search_enhanced.json") logger.info(f"Loaded composition: {json.dumps(loaded, indent=2)}") # List compositions compositions = composer.list_available() logger.info(f"All compositions: {compositions}") # Get info info = composer.get_composition_info("search_enhanced.json") logger.info(f"Composition info: {info}") # Task: Integrate the above capability into Eden's system """ EDEN CAPABILITY COMPOSER INTEGRATION This integrates the capability composition system into Eden's existing architecture. """ import sys sys.path.append("/Eden/CORE") from eden_metacap_HybridLLMStore import HybridLLMStore from eden_metacap_search_learning import SearchLearning from eden_metacap_outlier_detection import OutlierDetector from typing import List, Dict class CapabilityComposerIntegration: """ Integrates capability composition into Eden's system. This class provides the infrastructure for composing capabilities and integrating them with Eden's existing systems. """ def __init__(self): self.composer = CapabilityComposer() self.hybrid_store = HybridLLMStore() self.search_learning = SearchLearning() self.outlier_detector = OutlierDetector() def get_available_capabilities(self) -> List[str]: """ Get a list of all available capabilities that can be composed. Returns: List[str]: Names of available capabilities. """ # TODO: Populate with actual capabilities from Eden's system return ["search", "analysis", "optimization"] def compose_capability(self, base_capability: str, additional_features: List[str], mode: str = "sequential") -> Dict: """ Compose a new capability from existing ones. Args: base_capability (str): The primary capability to build upon. additional_features (List[str]): Additional features to integrate. mode (str): Integration mode ('sequential', 'parallel', etc.). Returns: Dict: The composed capability structure. """ return self.composer.combine_capabilities(base_capability, additional_features, mode) def validate_composition(self, composition: Dict) -> bool: """ Validate if a proposed capability composition is valid. Args: composition (Dict): The composition to validate. Returns: bool: True if valid, False otherwise. """ return self.composer.validate_composition(composition) def save_composed_capability(self, composition: Dict, name: str = None) -> str: """ Save a composed capability for later use. Args: composition (Dict): The composed capability to save. name (str): Optional custom name for the file. Returns: str: The filename it was saved as. """ return self.composer.save_composition(composition, name) def query_eden_about_composition(self, question: str) -> str: """ Ask Eden questions about capability compositions using hybrid LLM store. Args: question (str): The question to ask. Returns: str: Eden's response. """ # TODO: Implement actual query with Eden return "Eden is reviewing the composition details and will respond shortly." def learn_from_composition(self, query: str, response: str) -> bool: """ Store a learning from the capability composition process. Args: query (str): The original query. response (str): Eden's response to the query. Returns: bool: True if learned successfully, False otherwise. """ return self.search_learning.store_learning(query, response) def detect_outlier_composition(self, composition: Dict) -> bool: """ Detect if a composition is an outlier compared to others. Args: composition (Dict): The composed capability to check. Returns: bool: True if outlier, False otherwise. """ return self.outlier_detector.is_outlier({"composition": composition}) def get_composition_stats(self) -> Dict: """ Get statistics about composed capabilities. Returns: Dict: Statistics about compositions. """ # TODO: Implement actual stats tracking return { "total_compositions": 0, "valid_compositions": 0, "outlier_compositions": 0 } def summarize_compositions(self) -> str: """ Generate a summary of all composed capabilities. Returns: str: Summary string. """ stats = self.get_composition_stats() return f"Composed Capabilities Summary:\n- Total: {stats['total_compositions']}\n- Valid: {stats['valid_compositions']}\n- Outliers: {stats['outlier_compositions']}" # Example usage if __name__ == "__main__": integrator = CapabilityComposerIntegration() # Compose a new capability composed = integrator.compose_capability( base_capability="search", additional_features=["ranking", "filtering"], mode="parallel" ) print(f"\nComposed Structure: {json.dumps(composed, indent=2)}") # Save it filename = integrator.save_composed_capability(composed, name="search_improved.json") print(f"\nSaved to: {filename}") # Query Eden about it response = integrator.query_eden_about_composition("How does this composition work?") print(f"\nEden's Response: {response}") # Summarize summary = integrator.summarize_compositions() print(f"\n{summary}") # Task: Create a comprehensive test suite for the Capability Composer """ TEST SUITE FOR EDEN CAPABILITY COMPOSER """ import sys sys.path.append("/Eden/CORE") from eden_capability_composer import CapabilityComposer import json import os import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class TestCapabilityComposer: def setup_class(self): """Run once at the start of testing""" self.composer = CapabilityComposer() self.test_root = "/tmp/eden_composer_tests" os.makedirs(self.test_root, exist_ok=True) def teardown_class(self): """Run once after all tests""" # Clean up test files for filename in os.listdir(self.test_root): os.remove(os.path.join(self.test_root, filename)) def test_combine_two_capabilities(self): """Test combining two capabilities in sequence""" composition = self.composer.combine_capabilities( base_capability="search", additional_features=["filtering"], integration_mode="sequential" ) assert composition["base_capability"] == "search" assert len(composition["additional_features"]) == 1 assert composition["integration_mode"] == "sequential" def test_combine_multiple_capabilities(self): """Test combining multiple capabilities in parallel""" composition = self.composer.combine_capabilities( base_capability="analysis", additional_features=["pattern_recognition", "validation"], integration_mode="parallel" ) assert len(composition["additional_features"]) == 2 def test_validate_valid_composition(self): """Test that valid compositions are recognized""" valid_composition = { "base_capability": "core", "additional_features": ["security", "validation"], "integration_mode": "sequential", "version": "1.0", "timestamp": "2024-01-01T12:00:00", "status": "ready" } assert self.composer.validate_composition(valid_composition) def test_validate_invalid_composition(self): """Test that invalid compositions are caught""" # Missing required field invalid = { "additional_features": ["enhancement"], "integration_mode": "sequential" } assert not self.composer.validate_composition(invalid) def test_save_load_roundtrip(self): """Test saving and then loading a composition works""" composition = self.composer.combine_capabilities( base_capability="optimization", additional_features=["tuning", "feedback_loop"], integration_mode="sequential" ) filename = f"test_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" save_path = os.path.join(self.test_root, filename) self.composer.save_composition(composition, name=filename) # Load it back loaded = self.composer.load_composition(filename) assert loaded == composition assert os.path.exists(save_path) def test_list_compositions(self): """Test that list function works""" # Create a few test compositions for i in range(2): filename = f"test_{i}.json" self.composer.save_composition({}, name=filename) compositions = self.composer.list_available() assert len(compositions) >= 2 def test_get_info(self): """Test getting information about a composition""" composition = self.composer.combine_capabilities( base_capability="enhancement", additional_features=["customization"], integration_mode="parallel" ) info = self.composer.get_composition_info("test.json") assert info["base_capability"] == "enhancement" assert info["num_features"] == 1 assert info["mode"] == "parallel" def test_edge_cases(self): """Test edge cases and error handling""" # Empty features list comp = self.composer.combine_capabilities("basic", []) assert comp["additional_features"] == [] # Invalid mode try: self.composer.combine_capabilities("test", ["invalid"], "bad_mode") except ValueError as e: assert "Invalid integration_mode" in str(e) if __name__ == "__main__": logger.info("Running Capability Composer tests...") tester = TestCapabilityComposer() # Run all tests tester.setup_class() tests = [ tester.test_combine_two_capabilities, tester.test_combine_multiple_capabilities, tester.test_validate_valid_composition, tester.test_validate_invalid_composition, tester.test_save_load_roundtrip, tester.test_list_compositions, tester.test_get_info, tester.test_edge_cases ] for test in tests: try: logger.info(f"Running {test.__name__}...") test() logger.info("PASS") except Exception as e: logger.error(f"FAIL: {str(e)}") tester.teardown_class() # Task: Implement the remaining 7 capabilities Eden needs """ EDEN'S REMAINING 7 METACAPS PROJECT Continues from where Capability Composer left off. """ import sys sys.path.append("/Eden/CORE") from eden_capability_composer import CapabilityComposer from eden_metacap_search_learning import SearchLearning from eden_metacap_outlier_detection import OutlierDetector import os import json class EdenCapabilityExpansion: """ System to expand Eden's capabilities toward general AI capability. She needs 7 more metacaps: 1. Reasoning - Self-structured logical thought 2. Memory - Persistent experience recall 3. Causality - Understanding why things happen 4. High-Level Planning - Macro-strategy creation 5. Abstract Representation - Structured idea modeling 6. Cross-Space Learning - Transfer learning between domains 7. Multi-Agent Coordination - Working with others """ def __init__(self): self.composer = CapabilityComposer() self.reasoning = None self.memory = None self.causality = None self.planning = None self.representation = None self.learning = None self.coordination = None def build_reasoning_capability(self) -> Dict: """ Create a capability for self-structured reasoning Returns: Dict: Capability structure """ return self.composer.combine_capabilities( base_capability="logic", additional_features=["deduction", "induction", "abductive_reasoning"], integration_mode="sequential" ) def build_memory_capability(self) -> Dict: """Create memory retention capability""" return self.composer.combine_capabilities( base_capability="experience_storage", additional_features=["retention", "retrieval"], integration_mode="sequential" ) def build_causality_capability(self) -> Dict: """ Create cause-effect understanding Returns: Dict: Capability structure """ return self.composer.combine_capabilities( base_capability="causal_reasoning", additional_features=["correlation_analysis", "counterfactual_thinking"], integration_mode="parallel" )