Cortex/memory/associative.py

import asyncio
import numpy as np
from typing import Dict, List, Any, Optional, Set
import logging
from dataclasses import dataclass
import networkx as nx
from collections import defaultdict
import re

@dataclass
class Association:
    """Représente une association entre concepts"""
    source: str
    target: str
    strength: float  # 0.0 à 1.0
    type: str        # 'semantic', 'temporal', 'emotional', 'causal'
    created_at: float
    last_accessed: float

class AssociativeMemory:
    """
    Mémoire associative avec réseaux sémantiques
    et mécanismes de rappel par similarité
    """
    
    def __init__(self):
        self.logger = logging.getLogger("associative_memory")
        self.associations: Dict[str, List[Association]] = defaultdict(list)
        self.concept_vectors: Dict[str, np.ndarray] = {}
        self.semantic_network = nx.Graph()
        self.activation_threshold = 0.3
        self.spreading_activation_depth = 3
        
    async def initialize(self):
        """Initialise la mémoire associative"""
        self.logger.info("🧠 Initialisation de la mémoire associative...")
        
        try:
            await self._load_semantic_primitives()
            await self._build_initial_network()
            
            self.logger.info("✅ Mémoire associative initialisée")
            return True
            
        except Exception as e:
            self.logger.error(f"❌ Erreur d'initialisation associative: {e}")
            return False
    
    async def create_association(self, concept_a: str, concept_b: str, 
                               association_type: str = "semantic", 
                               strength: float = 0.5) -> bool:
        """Crée une association entre deux concepts"""
        try:
            # Vérifie si l'association existe déjà
            existing = await self._find_association(concept_a, concept_b)
            if existing:
                # Renforce l'association existante
                existing.strength = min(1.0, existing.strength + 0.1)
                existing.last_accessed = asyncio.get_event_loop().time()
                return True
            
            # Crée une nouvelle association
            association = Association(
                source=concept_a,
                target=concept_b,
                strength=strength,
                type=association_type,
                created_at=asyncio.get_event_loop().time(),
                last_accessed=asyncio.get_event_loop().time()
            )
            
            # Ajoute dans les deux directions
            self.associations[concept_a].append(association)
            self.associations[concept_b].append(Association(
                source=concept_b,
                target=concept_a,
                strength=strength,
                type=association_type,
                created_at=association.created_at,
                last_accessed=association.last_accessed
            ))
            
            # Met à jour le réseau sémantique
            self.semantic_network.add_edge(concept_a, concept_b, 
                                         weight=strength, type=association_type)
            
            # Met à jour les vecteurs conceptuels
            await self._update_concept_vectors(concept_a, concept_b)
            
            self.logger.info(f"🔗 Association créée: {concept_a} ↔ {concept_b} ({association_type})")
            return True
            
        except Exception as e:
            self.logger.error(f"Erreur création association: {e}")
            return False
    
    async def get_associations(self, concept: str, max_results: int = 10) -> List[Dict[str, Any]]:
        """Récupère les associations d'un concept"""
        try:
            if concept not in self.associations:
                return []
            
            associations = self.associations[concept]
            # Trie par force et récence
            associations.sort(key=lambda a: (a.strength, a.last_accessed), reverse=True)
            
            results = []
            for assoc in associations[:max_results]:
                results.append({
                    'target': assoc.target,
                    'strength': assoc.strength,
                    'type': assoc.type,
                    'last_accessed': assoc.last_accessed
                })
            
            return results
            
        except Exception as e:
            self.logger.error(f"Erreur récupération associations: {e}")
            return []
    
    async def pattern_completion(self, partial_pattern: List[str], context: Dict[str, Any] = None) -> List[str]:
        """Complète un pattern partiel basé sur les associations"""
        try:
            completions = []
            
            for concept in partial_pattern:
                associations = await self.get_associations(concept, max_results=20)
                
                for assoc in associations:
                    if assoc['target'] not in partial_pattern and assoc['target'] not in completions:
                        # Calcule un score de pertinence
                        relevance_score = await self._calculate_relevance_score(
                            assoc['target'], partial_pattern, context
                        )
                        
                        if relevance_score > self.activation_threshold:
                            completions.append(assoc['target'])
            
            # Trie par pertinence
            scored_completions = []
            for completion in completions:
                score = await self._calculate_relevance_score(completion, partial_pattern, context)
                scored_completions.append((score, completion))
            
            scored_completions.sort(reverse=True)
            return [comp for score, comp in scored_completions[:10]]
            
        except Exception as e:
            self.logger.error(f"Erreur complétion pattern: {e}")
            return []
    
    async def spreading_activation(self, start_concepts: List[str], depth: int = 3) -> Dict[str, float]:
        """Simule l'activation propagée dans le réseau sémantique"""
        try:
            activation_levels = defaultdict(float)
            
            # Activation initiale
            for concept in start_concepts:
                activation_levels[concept] = 1.0
            
            # Propagation sur plusieurs niveaux
            for current_depth in range(depth):
                next_activation = activation_levels.copy()
                
                for concept, activation in activation_levels.items():
                    if activation < 0.1:  # Seuil d'activation minimal
                        continue
                    
                    associations = await self.get_associations(concept, max_results=20)
                    
                    for assoc in associations:
                        # Propagation avec décroissance
                        propagation_strength = activation * assoc['strength'] * 0.7
                        
                        # Mise à jour de l'activation
                        if assoc['target'] not in start_concepts:
                            next_activation[assoc['target']] = max(
                                next_activation[assoc['target']],
                                propagation_strength
                            )
                
                activation_levels = next_activation
            
            # Filtre les concepts avec activation significative
            significant_activations = {
                concept: activation 
                for concept, activation in activation_levels.items() 
                if activation > self.activation_threshold
            }
            
            return dict(sorted(significant_activations.items(), 
                            key=lambda x: x[1], reverse=True))
            
        except Exception as e:
            self.logger.error(f"Erreur activation propagée: {e}")
            return {}
    
    async def semantic_similarity(self,