modules/metacognitive_engine.py

"""
Motor Metacognitivo de 11 Niveles
Implementa autoconciencia epistémica avanzada para sistemas de IA
"""

import numpy as np
import torch
import torch.nn as nn
from typing import Dict, List, Tuple, Optional, Any
from enum import Enum
from dataclasses import dataclass
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

class MetacognitiveLevel(Enum):
    """11 niveles de autoconciencia epistémica"""
    BASIC_AWARENESS = 1
    KNOWLEDGE_MONITORING = 2
    STRATEGY_SELECTION = 3
    UNCERTAINTY_DETECTION = 4
    CONFIDENCE_ASSESSMENT = 5
    ERROR_DETECTION = 6
    LEARNING_REGULATION = 7
    GOAL_MANAGEMENT = 8
    RESOURCE_ALLOCATION = 9
    TRANSFER_CONTROL = 10
    METAMEMORY_INTEGRATION = 11

@dataclass
class MetacognitiveState:
    """Estado metacognitivo en tiempo real"""
    level: MetacognitiveLevel
    confidence: float
    uncertainty: float
    knowledge_gaps: List[str]
    active_strategies: List[str]
    resource_usage: Dict[str, float]
    timestamp: datetime

class UncertaintyEstimator(nn.Module):
    """Red neuronal para estimación de incertidumbre epistémica"""
    
    def __init__(self, input_dim: int = 768, hidden_dim: int = 256):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, 1),
            nn.Sigmoid()
        )
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.encoder(x)

class StrategySelector:
    """Selector de estrategias cognitivas basado en contexto"""
    
    def __init__(self):
        self.strategies = {
            'analytical': {'weight': 0.8, 'context': ['scientific', 'mathematical']},
            'creative': {'weight': 0.6, 'context': ['artistic', 'innovative']},
            'collaborative': {'weight': 0.9, 'context': ['social', 'humanitarian']},
            'systematic': {'weight': 0.7, 'context': ['planning', 'organization']},
            'intuitive': {'weight': 0.5, 'context': ['uncertain', 'exploratory']}
        }
    
    def select_strategy(self, context: str, uncertainty: float) -> List[str]:
        """Selecciona estrategias óptimas según contexto e incertidumbre"""
        suitable_strategies = []
        
        for strategy, config in self.strategies.items():
            if any(ctx in context.lower() for ctx in config['context']):
                # Ajusta peso según incertidumbre
                adjusted_weight = config['weight'] * (1 - uncertainty * 0.3)
                if adjusted_weight > 0.5:
                    suitable_strategies.append(strategy)
        
        return suitable_strategies or ['analytical']  # fallback

class MetacognitiveEngine:
    """Motor principal de metacognición con 11 niveles de autoconciencia"""
    
    def __init__(self, device: str = 'cpu'):
        self.device = device
        self.current_level = MetacognitiveLevel.BASIC_AWARENESS
        self.uncertainty_estimator = UncertaintyEstimator().to(device)
        self.strategy_selector = StrategySelector()
        self.metacognitive_history = []
        self.knowledge_base = {}
        self.active_goals = []
        
        # Inicializar componentes especializados
        self._init_monitoring_systems()
        
    def _init_monitoring_systems(self):
        """Inicializa sistemas de monitoreo metacognitivo"""
        self.monitors = {
            'knowledge': self._monitor_knowledge_state,
            'strategy': self._monitor_strategy_effectiveness,
            'uncertainty': self._monitor_uncertainty_levels,
            'resources': self._monitor_resource_usage,
            'goals': self._monitor_goal_progress
        }
        
    def process_input(self, 
                     input_data: Any, 
                     context: str = "") -> Dict[str, Any]:
        """Procesamiento metacognitivo completo"""
        
        # Nivel 1: Conciencia básica
        awareness = self._basic_awareness(input_data)
        
        # Nivel 2: Monitoreo de conocimiento
        knowledge_state = self._monitor_knowledge_state(input_data, context)
        
        # Nivel 3: Selección de estrategia
        strategies = self.strategy_selector.select_strategy(context, knowledge_state['uncertainty'])
        
        # Nivel 4: Detección de incertidumbre
        uncertainty = self._detect_uncertainty(input_data)
        
        # Nivel 5: Evaluación de confianza
        confidence = self._assess_confidence(input_data, uncertainty)
        
        # Nivel 6: Detección de errores
        error_signals = self._detect_errors(input_data, confidence)
        
        # Nivel 7: Regulación del aprendizaje
        learning_adjustments = self._regulate_learning(error_signals)
        
        # Nivel 8: Gestión de objetivos
        goal_updates = self._manage_goals(input_data, context)
        
        # Nivel 9: Asignación de recursos
        resource_allocation = self._allocate_resources(strategies, uncertainty)
        
        # Nivel 10: Control de transferencia
        transfer_decisions = self._control_transfer(knowledge_state)
        
        # Nivel 11: Integración de metamemoria
        metamemory_integration = self._integrate_metamemory(input_data, context)
        
        # Crear estado metacognitivo
        state = MetacognitiveState(
            level=self.current_level,
            confidence=confidence,
            uncertainty=uncertainty,
            knowledge_gaps=knowledge_state.get('gaps', []),
            active_strategies=strategies,
            resource_usage=resource_allocation,
            timestamp=datetime.now()
        )
        
        self.metacognitive_history.append(state)
        
        return {
            'metacognitive_state': state,
            'processed_input': awareness,
            'strategies': strategies,
            'confidence': confidence,
            'uncertainty': uncertainty,
            'recommendations': self._generate_recommendations(state)
        }
    
    def _basic_awareness(self, input_data: Any) -> Dict[str, Any]:
        """Nivel 1: Conciencia básica del input"""
        return {
            'input_type': type(input_data).__name__,
            'input_size': len(str(input_data)) if hasattr(input_data, '__len__') else 1,
            'complexity_estimate': self._estimate_complexity(input_data)
        }
    
    def _monitor_knowledge_state(self, input_data: Any, context: str) -> Dict[str, Any]:
        """Nivel 2: Monitoreo del estado de conocimiento"""
        # Simula detección de gaps de conocimiento
        known_domains = list(self.knowledge_base.keys())
        input_domains = self._extract_domains(context)
        
        gaps = [domain for domain in input_domains if domain not in known_domains]
        
        return {
            'known_domains': known_domains,
            'required_domains': input_domains,
            'gaps': gaps,
            'coverage': len(known_domains) / max(len(input_domains), 1),
            'uncertainty': len(gaps) / max(len(input_domains), 1)
        }
    
    def _detect_uncertainty(self, input_data: Any) -> float:
        """Nivel 4: Detección de incertidumbre epistémica"""
        if isinstance(input_data, str):
            # Convierte texto a tensor para análisis
            input_tensor = torch.randn(1, 768)  # Simulación de embedding
        else:
            input_tensor = torch.tensor(input_data, dtype=torch.float32).unsqueeze(0)
        
        with torch.no_grad():
            uncertainty = self.uncertainty_estimator(input_tensor.to(self.device))
        
        return float(uncertainty.item())
    
    def _assess_confidence(self, input_data: Any, uncertainty: float) -> float:
        """Nivel 5: Evaluación de confianza"""
        base_confidence = 1.0 - uncertainty
        
        # Ajustes basados en experiencia previa
        if hasattr(self, 'metacognitive_history') and self.metacognitive_history:
            recent_performance = np.mean([
                state.confidence for state in self.metacognitive_history[-5:]
            ])
            base_confidence = 0.7 * base_confidence + 0.3 * recent_performance
        
        return np.clip(base_confidence, 0.0, 1.0)
    
    def _detect_errors(self, input_data: Any, confidence: float) -> List[str]:
        """Nivel 6: Detección de errores potenciales"""
        errors = []
        
        if confidence < 0.3:
            errors.append("low_confidence")
        
        # Detección de inconsistencias
        if self._detect_inconsistencies(input_data):
            errors.append("logical_inconsistency")
        
        # Detección de información faltante
        if self._detect_missing_info(input_data):
            errors.append("insufficient_information")
        
        return errors
    
    def _regulate_learning(self, error_signals: List[str]) -> Dict[str, Any]:
        """Nivel 7: Regulación del aprendizaje"""
        adjustments = {
            'learning_rate': 0.01,
            'exploration_factor': 0.1,
            'attention_focus': []
        }
        
        if 'low_confidence' in error_signals:
            adjustments['exploration_factor'] *= 2
            adjustments['attention_focus'].append('uncertainty_reduction')
        
        if 'logical_inconsistency' in error_signals:
            adjustments['learning_rate'] *= 0.5
            adjustments['attention_focus'].append('consistency_checking')
        
        return adjustments
    
    def _manage_goals(self, input_data: Any, context: str) -> Dict[str, Any]:
        """Nivel 8: Gestión de objetivos"""
        # Identifica objetivos implícitos en el contexto
        implicit_goals = self._extract_goals(context)
        
        # Actualiza objetivos activos
        for goal in implicit_goals:
            if goal not in self.active_goals:
                self.active_goals.append(goal)
        
        return {
            'active_goals': self.active_goals,
            'goal_priorities': self._prioritize_goals(),
            'goal_conflicts': self._detect_goal_conflicts()
        }
    
    def _allocate_resources(self, strategies: List[str], uncertainty: float) -> Dict[str, float]:
        """Nivel 9: Asignación de recursos computacionales"""
        base_allocation = {
            'processing_power': 0.7,
            'memory_usage': 0.5,
            'attention_bandwidth': 0.6,
            'exploration_budget': 0.3
        }
        
        # Ajusta según incertidumbre
        if uncertainty > 0.7:
            base_allocation['exploration_budget'] *= 2
            base_allocation['attention_bandwidth'] *= 1.5
        
        # Ajusta según estrategias activas
        if 'analytical' in strategies:
            base_allocation['processing_power'] *= 1.3
        
        if 'creative' in strategies:
            base_allocation['exploration_budget'] *= 1.5
        
        return base_allocation
    
    def _control_transfer(self, knowledge_state: Dict[str, Any]) -> Dict[str, Any]:
        """Nivel 10: Control de transferencia de conocimiento"""
        transfer_opportunities = []
        
        # Identifica oportunidades de transferencia
        for domain in knowledge_state.get('known_domains', []):
            similarity_score = self._calculate_domain_similarity(
                domain, knowledge_state.get('required_domains', [])
            )
            if similarity_score > 0.6:
                transfer_opportunities.append({
                    'source_domain': domain,
                    'similarity': similarity_score
                })
        
        return {
            'transfer_opportunities': transfer_opportunities,
            'transfer_strategy': 'analogical' if transfer_opportunities else 'direct'
        }
    
    def _integrate_metamemory(self, input_data: Any, context: str) -> Dict[str, Any]:
        """Nivel 11: Integración de metamemoria"""
        # Integra experiencias pasadas relevantes
        relevant_memories = self._retrieve_relevant_memories(context)
        
        # Actualiza modelo metacognitivo
        self._update_metamemory_model(input_data, context, relevant_memories)
        
        return {
            'relevant_memories': len(relevant_memories),
            'memory_confidence': self._assess_memory_confidence(relevant_memories),
            'learning_insights': self._extract_learning_insights()
        }
    
    def _generate_recommendations(self, state: MetacognitiveState) -> List[str]:
        """Genera recomendaciones basadas en estado metacognitivo"""
        recommendations = []
        
        if state.uncertainty > 0.7:
            recommendations.append("Buscar información adicional")
            recommendations.append("Considerar múltiples perspectivas")
        
        if state.confidence < 0.4:
            recommendations.append("Validar con fuentes externas")
            recommendations.append("Aplicar verificación cruzada")
        
        if not state.active_strategies:
            recommendations.append("Seleccionar estrategia cognitiva apropiada")
        
        return recommendations
    
    # Métodos auxiliares
    def _estimate_complexity(self, input_data: Any) -> float:
        """Estima complejidad del input"""
        if isinstance(input_data, str):
            return min(len(input_data.split()) / 100, 1.0)
        return 0.5  # valor por defecto
    
    def _extract_domains(self, context: str) -> List[str]:
        """Extrae dominios de conocimiento del contexto"""
        domain_keywords = {
            'science': ['research', 'experiment', 'hypothesis'],
            'technology': ['software', 'algorithm', 'programming'],
            'humanitarian': ['help', 'crisis', 'social', 'community'],
            'education': ['learn', 'teach', 'knowledge', 'skill']
        }
        
        domains = []
        context_lower = context.lower()
        
        for domain, keywords in domain_keywords.items():
            if any(keyword in context_lower for keyword in keywords):
                domains.append(domain)
        
        return domains or ['general']
    
    def _detect_inconsistencies(self, input_data: Any) -> bool:
        """Detecta inconsistencias lógicas"""
        # Implementación simplificada
        return False
    
    def _detect_missing_info(self, input_data: Any) -> bool:
        """Detecta información faltante"""
        if isinstance(input_data, str):
            return len(input_data.strip()) < 10
        return False
    
    def _extract_goals(self, context: str) -> List[str]:
        """Extrae objetivos implícitos del contexto"""
        goal_indicators = {
            'solve': 'problem_solving',
            'learn': 'knowledge_acquisition',
            'help': 'assistance',
            'create': 'creation',
            'analyze': 'analysis'
        }
        
        goals = []
        context_lower = context.lower()
        
        for indicator, goal in goal_indicators.items():
            if indicator in context_lower:
                goals.append(goal)
        
        return goals
    
    def _prioritize_goals(self) -> Dict[str, float]:
        """Prioriza objetivos activos"""
        if not self.active_goals:
            return {}
        
        # Priorización simple basada en importancia humanitaria
        humanitarian_priority = {
            'assistance': 1.0,
            'problem_solving': 0.9,
            'knowledge_acquisition': 0.7,
            'analysis': 0.6,
            'creation': 0.5
        }
        
        return {
            goal: humanitarian_priority.get(goal, 0.5) 
            for goal in self.active_goals
        }
    
    def _detect_goal_conflicts(self) -> List[Tuple[str, str]]:
        """Detecta conflictos entre objetivos"""
        conflicts = []
        
        # Conflictos conocidos
        conflict_pairs = [
            ('speed', 'accuracy'),
            ('exploration', 'exploitation'),
            ('creativity', 'consistency')
        ]
        
        for goal1, goal2 in conflict_pairs:
            if goal1 in self.active_goals and goal2 in self.active_goals:
                conflicts.append((goal1, goal2))
        
        return conflicts
    
    def _calculate_domain_similarity(self, domain1: str, domains: List[str]) -> float:
        """Calcula similitud entre dominios"""
        if not domains:
            return 0.0
        
        # Similitudes conocidas entre dominios
        similarity_matrix = {
            ('science', 'technology'): 0.8,
            ('humanitarian', 'education'): 0.7,
            ('technology', 'education'): 0.6
        }
        
        max_similarity = 0.0
        for domain2 in domains:
            key = tuple(sorted([domain1, domain2]))
            similarity = similarity_matrix.get(key, 0.0)
            max_similarity = max(max_similarity, similarity)
        
        return max_similarity
    
    def _retrieve_relevant_memories(self, context: str) -> List[Dict[str, Any]]:
        """Recupera memorias relevantes"""
        # Implementación simplificada
        return self.metacognitive_history[-5:]  # últimas 5 experiencias
    
    def _assess_memory_confidence(self, memories: List[Any]) -> float:
        """Evalúa confianza en las memorias recuperadas"""
        if not memories:
            return 0.0
        
        confidences = [
            getattr(memory, 'confidence', 0.5) 
            for memory in memories
        ]
        
        return np.mean(confidences)
    
    def _extract_learning_insights(self) -> List[str]:
        """Extrae insights de aprendizaje"""
        insights = []
        
        if len(self.metacognitive_history) >= 5:
            recent_confidence = np.mean([
                state.confidence for state in self.metacognitive_history[-5:]
            ])
            
            if recent_confidence > 0.8:
                insights.append("High confidence pattern detected")
            elif recent_confidence < 0.3:
                insights.append("Low confidence pattern - needs attention")
        
        return insights
    
    def _update_metamemory_model(self, input_data: Any, context: str, memories: List[Any]):
        """Actualiza modelo de metamemoria"""
        # Actualiza base de conocimiento
        domains = self._extract_domains(context)
        for domain in domains:
            if domain not in self.knowledge_base:
                self.knowledge_base[domain] = []
            
            self.knowledge_base[domain].append({
                'input': str(input_data)[:100],  # primeros 100 chars
                'timestamp': datetime.now(),
                'context': context
            })
    
    def get_metacognitive_summary(self) -> Dict[str, Any]:
        """Obtiene resumen del estado metacognitivo actual"""
        if not self.metacognitive_history:
            return {"status": "No metacognitive history available"}
        
        recent_states = self.metacognitive_history[-10:]
        
        return {
            'current_level': self.current_level.name,
            'average_confidence': np.mean([s.confidence for s in recent_states]),
            'average_uncertainty': np.mean([s.uncertainty for s in recent_states]),
            'active_strategies': list(set([
                strategy for state in recent_states 
                for strategy in state.active_strategies
            ])),
            'knowledge_domains': list(self.knowledge_base.keys()),
            'active_goals': self.active_goals,
            'total_experiences': len(self.metacognitive_history)
        }