""" 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() }