Spaces:
Sleeping
Sleeping
| """ | |
| Dynamic Learning Component for Codette | |
| Handles real-time learning and adaptation of the AI system | |
| """ | |
| import logging | |
| from typing import Dict, List, Any, Optional | |
| from datetime import datetime | |
| logger = logging.getLogger(__name__) | |
| try: | |
| import numpy as np | |
| except Exception: | |
| np = None | |
| class DynamicLearner: | |
| """Handles dynamic learning and adaptation for Codette""" | |
| def __init__(self, | |
| learning_rate: float = 0.01, | |
| memory_size: int = 1000, | |
| adaptation_threshold: float = 0.75): | |
| """Initialize the dynamic learner""" | |
| self.learning_rate = learning_rate | |
| self.memory_size = memory_size | |
| self.adaptation_threshold = adaptation_threshold | |
| self.memory_buffer = [] | |
| self.learning_history = [] | |
| self.current_state = { | |
| "performance": 1.0, | |
| "adaptability": 1.0, | |
| "complexity": 0.0 | |
| } | |
| logger.info("Dynamic learner initialized") | |
| def update(self, new_data: Dict[str, Any]) -> float: | |
| """Update the learner with new data""" | |
| try: | |
| # Add to memory buffer | |
| self.memory_buffer.append({ | |
| "data": new_data, | |
| "timestamp": datetime.now().isoformat() | |
| }) | |
| # Trim memory if needed | |
| if len(self.memory_buffer) > self.memory_size: | |
| self.memory_buffer = self.memory_buffer[-self.memory_size:] | |
| # Calculate adaptation score | |
| adaptation_score = self._calculate_adaptation_score(new_data) | |
| # Update internal state | |
| self._update_state(adaptation_score) | |
| # Record learning progress | |
| self.learning_history.append({ | |
| "score": adaptation_score, | |
| "state": self.current_state.copy(), | |
| "timestamp": datetime.now().isoformat() | |
| }) | |
| logger.debug(f"Learning update complete. Adaptation score: {adaptation_score}") | |
| return adaptation_score | |
| except Exception as e: | |
| logger.error(f"Error in dynamic learning update: {e}") | |
| return 0.0 | |
| def _calculate_adaptation_score(self, data: Dict[str, Any]) -> float: | |
| """Calculate how well the system is adapting to new data""" | |
| try: | |
| # Extract features for adaptation calculation | |
| complexity = self._estimate_complexity(data) | |
| novelty = self._calculate_novelty(data) | |
| consistency = self._check_consistency(data) | |
| # Weighted combination of factors | |
| score = (0.4 * novelty + | |
| 0.3 * consistency + | |
| 0.3 * (1.0 - complexity)) | |
| return max(0.0, min(1.0, score)) | |
| except Exception as e: | |
| logger.error(f"Error calculating adaptation score: {e}") | |
| return 0.0 | |
| def _estimate_complexity(self, data: Dict[str, Any]) -> float: | |
| """Estimate the complexity of new data""" | |
| try: | |
| # Basic complexity estimation | |
| if isinstance(data, dict): | |
| depth = self._get_dict_depth(data) | |
| size = len(str(data)) | |
| return min(1.0, (depth * 0.1 + size * 0.0001)) | |
| return 0.0 | |
| except Exception as e: | |
| logger.error(f"Error estimating complexity: {e}") | |
| return 0.0 | |
| def _calculate_novelty(self, data: Dict[str, Any]) -> float: | |
| """Calculate how novel the new data is compared to memory""" | |
| try: | |
| if not self.memory_buffer: | |
| return 1.0 | |
| similarities = [] | |
| for memory in self.memory_buffer[-10:]: # Compare with last 10 memories | |
| similarity = self._calculate_similarity(memory['data'], data) | |
| similarities.append(similarity) | |
| if np is not None: | |
| avg_similarity = float(np.mean(similarities)) if similarities else 0.0 | |
| else: | |
| avg_similarity = float(sum(similarities)/len(similarities)) if similarities else 0.0 | |
| return 1.0 - avg_similarity | |
| except Exception as e: | |
| logger.error(f"Error calculating novelty: {e}") | |
| return 0.0 | |
| def _check_consistency(self, data: Dict[str, Any]) -> float: | |
| """Check if new data is consistent with existing knowledge""" | |
| try: | |
| if not self.memory_buffer: | |
| return 1.0 | |
| # Compare with recent memories for consistency | |
| recent_memories = self.memory_buffer[-5:] | |
| consistency_scores = [] | |
| for memory in recent_memories: | |
| score = self._check_data_consistency(memory['data'], data) | |
| consistency_scores.append(score) | |
| if np is not None: | |
| return float(np.mean(consistency_scores)) if consistency_scores else 1.0 | |
| return float(sum(consistency_scores)/len(consistency_scores)) if consistency_scores else 1.0 | |
| except Exception as e: | |
| logger.error(f"Error checking consistency: {e}") | |
| return 0.0 | |
| def _get_dict_depth(self, d: Dict) -> int: | |
| """Calculate the maximum depth of a dictionary""" | |
| if not isinstance(d, dict) or not d: | |
| return 0 | |
| return 1 + max(self._get_dict_depth(v) if isinstance(v, dict) else 0 | |
| for v in d.values()) | |
| def _calculate_similarity(self, data1: Dict[str, Any], | |
| data2: Dict[str, Any]) -> float: | |
| """Calculate similarity between two data points""" | |
| try: | |
| # Convert to sets of items for comparison | |
| items1 = set(str(item) for item in self._flatten_dict(data1).items()) | |
| items2 = set(str(item) for item in self._flatten_dict(data2).items()) | |
| # Calculate Jaccard similarity | |
| intersection = len(items1.intersection(items2)) | |
| union = len(items1.union(items2)) | |
| return intersection / union if union > 0 else 0.0 | |
| except Exception as e: | |
| logger.error(f"Error calculating similarity: {e}") | |
| return 0.0 | |
| def _check_data_consistency(self, data1: Dict[str, Any], | |
| data2: Dict[str, Any]) -> float: | |
| """Check structural and value consistency between data points""" | |
| try: | |
| # Check structure consistency | |
| keys1 = set(self._flatten_dict(data1).keys()) | |
| keys2 = set(self._flatten_dict(data2).keys()) | |
| # Calculate structural similarity | |
| key_similarity = len(keys1.intersection(keys2)) / len(keys1.union(keys2)) | |
| return key_similarity | |
| except Exception as e: | |
| logger.error(f"Error checking data consistency: {e}") | |
| return 0.0 | |
| def _flatten_dict(self, d: Dict[str, Any], | |
| parent_key: str = '', | |
| sep: str = '.') -> Dict[str, Any]: | |
| """Flatten a nested dictionary""" | |
| items = [] | |
| for k, v in d.items(): | |
| new_key = f"{parent_key}{sep}{k}" if parent_key else k | |
| if isinstance(v, dict): | |
| items.extend(self._flatten_dict(v, new_key, sep).items()) | |
| else: | |
| items.append((new_key, v)) | |
| return dict(items) | |
| def _update_state(self, adaptation_score: float): | |
| """Update internal state based on adaptation score""" | |
| try: | |
| # Update performance metric | |
| if np is not None: | |
| self.current_state["performance"] = float(np.mean([ | |
| self.current_state["performance"], | |
| adaptation_score | |
| ])) | |
| else: | |
| self.current_state["performance"] = float((self.current_state["performance"] + adaptation_score)/2) | |
| # Update adaptability | |
| if adaptation_score > self.adaptation_threshold: | |
| self.current_state["adaptability"] = min( | |
| 1.0, | |
| self.current_state["adaptability"] + self.learning_rate | |
| ) | |
| else: | |
| self.current_state["adaptability"] = max( | |
| 0.0, | |
| self.current_state["adaptability"] - self.learning_rate | |
| ) | |
| # Update complexity | |
| if np is not None: | |
| self.current_state["complexity"] = float(np.mean([ | |
| self.current_state["complexity"], | |
| 1.0 - adaptation_score | |
| ])) | |
| else: | |
| self.current_state["complexity"] = float((self.current_state["complexity"] + (1.0 - adaptation_score))/2) | |
| except Exception as e: | |
| logger.error(f"Error updating state: {e}") | |
| def get_state(self) -> Dict[str, float]: | |
| """Get current learning state""" | |
| return self.current_state.copy() | |
| def get_learning_history(self) -> List[Dict[str, Any]]: | |
| """Get learning history""" | |
| return self.learning_history.copy() |