cognitive_kernel.py

"""
KERNEL COGNITIVO MEJORADO
- Memoria de contexto extendida
- Selección de modelos inteligente
- Análisis de intención avanzado
"""
import time
import hashlib
import re
from collections import defaultdict, deque
from typing import Dict, List, Tuple, Any
from dataclasses import dataclass
from enum import Enum

# ===================== ENUMERACIONES =====================
class IntentType(Enum):
    IMAGE = "IMAGE"
    CODE = "CODE"
    REASONING = "REASONING"
    ARCHITECTURE = "ARCHITECTURE"
    DEVOPS = "DEVOPS"
    QA = "QA"
    VISUAL = "VISUAL"
    GENERAL = "GENERAL"

# ===================== MODELOS DE DATOS =====================
@dataclass
class MemoryNode:
    prompt: str
    intent: IntentType
    model: str
    timestamp: float
    success_score: float = 1.0
    context_hash: str = ""

# ===================== REGISTRO DE LATENCIA MEJORADO =====================
class AdaptiveLatencyRegistry:
    def __init__(self, window_size: int = 50):
        self.data = defaultdict(lambda: deque(maxlen=window_size))
        self.success_rates = defaultdict(lambda: deque(maxlen=window_size))
        self.response_times = defaultdict(lambda: deque(maxlen=window_size))
    
    def record(self, model: str, latency: float, success: bool = True):
        self.data[model].append(latency)
        self.success_rates[model].append(1.0 if success else 0.0)
        self.response_times[model].append(time.time())
    
    def get_composite_score(self, model: str) -> float:
        """Puntaje compuesto: latencia, tasa de éxito y frescura"""
        if not self.data[model]:
            return float("inf")
        
        # Puntaje de latencia (menor es mejor)
        latency_score = sum(self.data[model]) / len(self.data[model])
        
        # Puntaje de éxito
        success_rate = sum(self.success_rates[model]) / len(self.success_rates[model]) if self.success_rates[model] else 0.5
        
        # Puntaje de frescura (preferir modelos usados recientemente)
        freshness = 0.0
        if self.response_times[model]:
            latest = max(self.response_times[model])
            freshness = min(1.0, (time.time() - latest) / 3600)  # Normalizado por hora
        
        return latency_score * (1.1 - success_rate) * (1.0 + freshness * 0.1)

# ===================== GRAFO DE MEMORIA AVANZADO =====================
class ContextMemoryGraph:
    def __init__(self, max_nodes: int = 5000, embedding_dim: int = 384):
        self.nodes: Dict[str, MemoryNode] = {}
        self.context_order = deque(maxlen=max_nodes)
        self.intent_clusters = defaultdict(list)
        self.embedding_cache = {}
        
    def _generate_hash(self, text: str, context: str = "") -> str:
        """Hash contextual para mejor recuperación"""
        combined = f"{text}::{context}"
        return hashlib.sha256(combined.encode()).hexdigest()[:24]
    
    def _extract_keywords(self, text: str) -> List[str]:
        """Extrae palabras clave para clustering"""
        words = re.findall(r'\b[a-z]{4,}\b', text.lower())
        return [w for w in words if len(w) > 3][:10]
    
    def store(self, prompt: str, intent: IntentType, model: str, 
              context: str = "", success_score: float = 1.0):
        """Almacena memoria con contexto"""
        h = self._generate_hash(prompt, context)
        
        if h not in self.nodes:
            node = MemoryNode(
                prompt=prompt,
                intent=intent,
                model=model,
                timestamp=time.time(),
                success_score=success_score,
                context_hash=hashlib.md5(context.encode()).hexdigest()[:12]
            )
            self.nodes[h] = node
            self.context_order.append(h)
            
            # Clustering por intención
            self.intent_clusters[intent.value].append(h)
    
    def recall(self, intent: IntentType, current_context: str = "", 
               limit: int = 5) -> List[MemoryNode]:
        """Recupera memorias relevantes con matching contextual"""
        relevant = []
        
        for h in reversed(self.context_order):
            node = self.nodes[h]
            if node.intent == intent:
                # Puntaje de relevancia basado en éxito y contexto
                context_match = 1.0 if node.context_hash == hashlib.md5(
                    current_context.encode()).hexdigest()[:12] else 0.8
                relevance = node.success_score * context_match
                relevant.append((relevance, node))
        
        # Ordenar por relevancia
        relevant.sort(key=lambda x: x[0], reverse=True)
        return [node for _, node in relevant[:limit]]
    
    def find_similar(self, prompt: str, intent: IntentType = None) -> List[MemoryNode]:
        """Encuentra prompts similares usando keywords"""
        keywords = set(self._extract_keywords(prompt))
        similar = []
        
        for h, node in self.nodes.items():
            if intent and node.intent != intent:
                continue
            
            node_keywords = set(self._extract_keywords(node.prompt))
            overlap = len(keywords.intersection(node_keywords))
            
            if overlap >= 2:  # Al menos 2 palabras clave coincidentes
                similarity = overlap / max(len(keywords), len(node_keywords))
                similar.append((similarity, node))
        
        similar.sort(key=lambda x: x[0], reverse=True)
        return [node for _, node in similar[:3]]

# ===================== SISTEMA DE ORQUESTACIÓN =====================
class CognitiveOrchestrator:
    def __init__(self):
        self.latency_registry = AdaptiveLatencyRegistry()
        self.memory_graph = ContextMemoryGraph()
        self.intent_history = deque(maxlen=100)
        
    def analyze_intent(self, text: str, context: List[str] = None) -> IntentType:
        """Análisis de intención con contexto histórico"""
        text_lower = text.lower()
        
        # Patrones específicos
        patterns = {
            IntentType.IMAGE: [
                r'(imagen|ilustración|arte|render|diseño|visual|foto|retrato|cinematográfico)',
                r'genera.*imagen|crea.*visual|diseña.*arte'
            ],
            IntentType.CODE: [
                r'(código|programa|bug|error|python|api|endpoint|clase|función)',
                r'implementa.*código|escribe.*programa|resuelve.*bug'
            ],
            IntentType.REASONING: [
                r'(analiza|razona|piensa|estratégia|lógica|proceso|explica)',
                r'por qué|cómo funciona|qué significa|analiza.*situación'
            ],
            IntentType.ARCHITECTURE: [
                r'(arquitectura|microservicios|sistema|escala|diseño|patrón)',
                r'diseña.*sistema|arquitectura.*para|esquema.*tecnológico'
            ],
            IntentType.DEVOPS: [
                r'(docker|kubernetes|ci/cd|terraform|aws|gcp|azure|infraestructura)',
                r'deploy|implementa.*infraestructura|configura.*servidor'
            ],
            IntentType.QA: [
                r'(test|prueba|pytest|unitario|integración|cobertura|qa)',
                r'escribe.*test|prueba.*código|cobertura.*tests'
            ],
            IntentType.VISUAL: [
                r'(prompt.*visual|análisis.*imagen|describe.*foto|interpreta.*visual)',
                r'qué hay.*imagen|describe.*escena'
            ]
        }
        
        # Ponderación por historial
        intent_scores = defaultdict(float)
        
        # Análisis por patrones
        for intent, pattern_list in patterns.items():
            for pattern in pattern_list:
                if re.search(pattern, text_lower, re.IGNORECASE):
                    intent_scores[intent] += 2.0
        
        # Contexto histórico
        if context:
            recent_context = " ".join(context[-3:]).lower()
            for intent, pattern_list in patterns.items():
                for pattern in pattern_list:
                    if re.search(pattern, recent_context, re.IGNORECASE):
                        intent_scores[intent] += 1.0
        
        if intent_scores:
            # Seleccionar intención con mayor puntaje
            selected = max(intent_scores.items(), key=lambda x: x[1])[0]
            self.intent_history.append(selected)
            return selected
        
        # Fallback: detectar por contenido
        if any(word in text_lower for word in ['?', 'cómo', 'por qué', 'qué']):
            return IntentType.REASONING
        
        return IntentType.GENERAL
    
    def select_optimal_model(self, intent: IntentType, candidates: Dict[str, str], 
                            context: str = "") -> Tuple[str, float]:
        """Selección de modelo con memoria contextual"""
        # Revisar memoria para decisiones anteriores exitosas
        memories = self.memory_graph.recall(intent, context)
        
        for memory in memories:
            if memory.model in candidates.values():
                # Modelo previamente exitoso para este contexto
                confidence = memory.success_score * 0.7
                return memory.model, confidence
        
        # Selección basada en latencia adaptativa
        scored_models = []
        for name, model_id in candidates.items():
            score = self.latency_registry.get_composite_score(model_id)
            scored_models.append((score, model_id, name))
        
        scored_models.sort(key=lambda x: x[0])
        
        if scored_models:
            best_model = scored_models[0][1]
            confidence = 1.0 / (1.0 + scored_models[0][0])
            return best_model, min(confidence, 0.95)
        
        # Fallback al primer modelo
        fallback = next(iter(candidates.values()))
        return fallback, 0.5
    
    def record_interaction(self, prompt: str, intent: IntentType, 
                          model: str, latency: float, success: bool = True,
                          context: str = ""):
        """Registra interacción completa"""
        success_score = 1.0 if success else 0.3
        self.latency_registry.record(model, latency, success)
        self.memory_graph.store(prompt, intent, model, context, success_score)

# ===================== INSTANCIAS GLOBALES =====================
LATENCY = AdaptiveLatencyRegistry()
MEMORY = ContextMemoryGraph()
ORCHESTRATOR = CognitiveOrchestrator()

# ===================== FUNCIONES PÚBLICAS =====================
def analyze_intent_detailed(text: str, history: List[Tuple] = None) -> Tuple[IntentType, Dict]:
    """Analiza intención con metadatos detallados"""
    context = []
    if history:
        context = [msg for pair in history[-3:] for msg in pair if msg]
    
    intent = ORCHESTRATOR.analyze_intent(text, context)
    
    # Metadatos adicionales
    metadata = {
        "confidence": 0.85,
        "keywords": MEMORY._extract_keywords(text),
        "context_used": len(context) > 0,
        "similar_prompts": [node.prompt[:50] + "..." 
                          for node in MEMORY.find_similar(text, intent)],
        "timestamp": time.time()
    }
    
    return intent, metadata

def select_model_with_context(intent: IntentType, candidates: Dict[str, str],
                             context: str = "") -> Tuple[str, float, Dict]:
    """Selección de modelo con contexto y explicación"""
    model, confidence = ORCHESTRATOR.select_optimal_model(intent, candidates, context)
    
    explanation = {
        "selection_method": "composite_scoring",
        "candidates_evaluated": len(candidates),
        "confidence_score": confidence,
        "historical_matches": len(MEMORY.recall(intent, context)),
        "recommendation_reason": "Optimal balance of latency, success rate, and contextual relevance"
    }
    
    return model, confidence, explanation

def record_complete_interaction(prompt: str, intent: IntentType, model: str,
                               latency: float, success: bool = True,
                               response_quality: float = 1.0,
                               user_context: str = ""):
    """Registro completo de interacción con calidad de respuesta"""
    adjusted_success = success and (response_quality > 0.6)
    ORCHESTRATOR.record_interaction(
        prompt, intent, model, latency, adjusted_success, user_context
    )
    
    # Registrar también en memoria global
    MEMORY.store(prompt, intent, model, user_context, 
                response_quality if success else 0.1)
    
    return {
        "recorded": True,
        "success": adjusted_success,
        "quality_score": response_quality,
        "timestamp": time.time()
    }