Spaces:
Runtime error
Runtime error
File size: 8,141 Bytes
0807024 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 | 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, |