Upload cognitive_kernel.py

cognitive_kernel.py

@@ -0,0 +1,311 @@
+"""
+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()
+    }