FlyBrain-Lab / src /memory /working.py
timfromhcs's picture
FlyBrain v4.1.0 Space build (REAL_SUBGRAPH, CPU-only, honest backend)
3d46076 verified
Raw
History Blame Contribute Delete
2.14 kB
import time
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
@dataclass
class WorkingMemoryItem:
key: str
content: Any
salience: float = 1.0
created_at: float = field(default_factory=time.time)
access_count: int = 0
class WorkingMemory:
"""
Capacity-limited working memory (Miller's Law: 7 +/- 2 slots).
Decays over time unless refreshed by attention or access.
"""
def __init__(self, capacity: int = 7, decay_rate: float = 0.05):
self.capacity = capacity
self.decay_rate = decay_rate
self.items: Dict[str, WorkingMemoryItem] = {}
def put(self, key: str, content: Any, salience: float = 1.0):
if key in self.items:
item = self.items[key]
item.content = content
item.salience = min(2.0, item.salience + salience)
item.access_count += 1
return
# If full, evict lowest salience item
if len(self.items) >= self.capacity:
lowest_key = min(self.items.keys(), key=lambda k: self.items[k].salience)
del self.items[lowest_key]
self.items[key] = WorkingMemoryItem(key=key, content=content, salience=salience)
def get(self, key: str) -> Optional[Any]:
if key in self.items:
item = self.items[key]
item.salience = min(2.0, item.salience + 0.2)
item.access_count += 1
return item.content
return None
def decay_step(self):
to_delete = []
for key, item in self.items.items():
item.salience -= self.decay_rate
if item.salience <= 0.0:
to_delete.append(key)
for k in to_delete:
del self.items[k]
def get_all_active(self) -> List[Dict[str, Any]]:
return [
{
"key": item.key,
"content": str(item.content),
"salience": round(item.salience, 3),
"access_count": item.access_count
}
for item in sorted(self.items.values(), key=lambda x: x.salience, reverse=True)
]