Spaces:
Runtime error
Runtime error
File size: 6,914 Bytes
30cd86d | 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 | import asyncio
import time
from typing import Dict, List, Any, Optional
import logging
from dataclasses import dataclass
from enum import Enum
import heapq
from collections import OrderedDict
class MemoryLevel(Enum):
"""Niveaux de la mémoire hiérarchique"""
IMMEDIATE = "immediate" # Mémoire sensorielle (quelques secondes)
WORKING = "working" # Mémoire de travail (quelques minutes)
SHORT_TERM = "short_term" # Mémoire à court terme (heures)
LONG_TERM = "long_term" # Mémoire à long terme (jours/années)
PERMANENT = "permanent" # Mémoire permanente (cristallisée)
@dataclass
class MemoryItem:
"""Élément de mémoire avec métadonnées"""
id: str
content: Any
level: MemoryLevel
strength: float # 0.0 à 1.0
access_count: int
last_accessed: float
created_at: float
associations: List[str]
emotional_weight: float
class HierarchicalMemory:
"""
Système de mémoire hiérarchique avancé
avec mécanismes de consolidation et d'oubli
"""
def __init__(self):
self.logger = logging.getLogger("hierarchical_memory")
self.memory_levels = {
MemoryLevel.IMMEDIATE: OrderedDict(),
MemoryLevel.WORKING: OrderedDict(),
MemoryLevel.SHORT_TERM: OrderedDict(),
MemoryLevel.LONG_TERM: OrderedDict(),
MemoryLevel.PERMANENT: OrderedDict()
}
self.access_patterns = {}
self.consolidation_threshold = 0.7
self.forgetting_curve = {}
async def initialize(self):
"""Initialise le système de mémoire hiérarchique"""
self.logger.info("🏗️ Initialisation de la mémoire hiérarchique...")
try:
await self._setup_memory_structures()
await self._load_forgetting_curves()
# Démarre le processus de consolidation en arrière-plan
asyncio.create_task(self._consolidation_worker())
self.logger.info("✅ Mémoire hiérarchique initialisée")
return True
except Exception as e:
self.logger.error(f"❌ Erreur d'initialisation mémoire: {e}")
return False
async def store(self, content: Any, level: MemoryLevel, priority: int = 5) -> str:
"""Stocke un élément en mémoire avec niveau spécifié"""
try:
memory_id = self._generate_memory_id(content)
item = MemoryItem(
id=memory_id,
content=content,
level=level,
strength=0.1, # Force initiale faible
access_count=0,
last_accessed=time.time(),
created_at=time.time(),
associations=[],
emotional_weight=0.0
)
# Stockage selon le niveau
self.memory_levels[level][memory_id] = item
# Application des mécanismes d'attention basés sur la priorité
await self._apply_attention_mechanism(item, priority)
self.logger.info(f"💾 Mémoire stockée [{level.value}]: {memory_id}")
return memory_id
except Exception as e:
self.logger.error(f"Erreur stockage mémoire: {e}")
raise
async def retrieve(self, memory_id: str, strengthen: bool = True) -> Optional[Any]:
"""Récupère un élément de mémoire"""
try:
# Recherche dans tous les niveaux
for level_name, level_dict in self.memory_levels.items():
if memory_id in level_dict:
item = level_dict[memory_id]
# Mise à jour des métadonnées d'accès
item.access_count += 1
item.last_accessed = time.time()
if strengthen:
await self._strengthen_memory(item)
# Promotion potentielle vers un niveau supérieur
await self._consider_promotion(item)
self.logger.debug(f"📖 Mémoire récupérée: {memory_id} (niveau: {level_name.value})")
return item.content
self.logger.warning(f"⚠️ Mémoire introuvable: {memory_id}")
return None
except Exception as e:
self.logger.error(f"Erreur récupération mémoire: {e}")
return None
async def associative_recall(self, cue: Any, max_results: int = 5) -> List[MemoryItem]:
"""Rappel associatif basé sur des indices"""
try:
results = []
cue_str = str(cue).lower()
for level_name, level_dict in self.memory_levels.items():
for item in level_dict.values():
content_str = str(item.content).lower()
# Vérifie la similarité
similarity = await self._calculate_similarity(cue_str, content_str)
if similarity > 0.3: # Seuil de similarité
results.append((similarity, item))
# Trie par similarité
results.sort(key=lambda x: x[0], reverse=True)
# Retourne les meilleurs résultats
return [item for _, item in results[:max_results]]
except Exception as e:
self.logger.error(f"Erreur rappel associatif: {e}")
return []
async def promote(self, memory_id: str) -> bool:
"""Promouvoir un élément de mémoire vers un niveau supérieur"""
try:
# Trouve l'élément dans les niveaux
current_level = None
item = None
for level_name, level_dict in self.memory_levels.items():
if memory_id in level_dict:
current_level = level_name
item = level_dict[memory_id]
break
if not item or not current_level:
return False
# Détermine le niveau suivant
next_level = await self._get_next_level(current_level)
if not next_level:
return False # Déjà au niveau maximum
# Supprime de l'ancien niveau et ajoute au nouveau
del self.memory_levels[current_level][memory_id]
self.memory_levels[next_level][memory_id] = item
# Renforce la mémoire lors de la promotion
await self._strengthen_memory(item, boost=0.2)
self.logger.info(f"📈 Mémoire promue: {memory_id} ({current_level |