File size: 3,924 Bytes
29cdc9d fc2988c 29cdc9d fc2988c 8663f54 fc2988c 8663f54 fc2988c 8663f54 fc2988c 29cdc9d 8663f54 fc2988c 8663f54 fc2988c 8663f54 fc2988c 29cdc9d 8663f54 fc2988c 29cdc9d 8663f54 fc2988c | 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 | import numpy as np
import os
import json
import time
class Hippocampus:
"""
Biologically-inspired vector memory.
Memories strengthen with use. Memories decay without use.
Based on Ebbinghaus forgetting curve.
"""
def __init__(self, path=None):
self.path = path or os.path.expanduser("~/.vitalis_workspace/hippocampus.npy")
self.meta_path = os.path.expanduser("~/.vitalis_workspace/hippocampus_meta.json")
os.makedirs(os.path.dirname(self.path), exist_ok=True)
self.memory = np.load(self.path, allow_pickle=True).item() if os.path.exists(self.path) else {}
self._load_meta()
def _load_meta(self):
if os.path.exists(self.meta_path):
with open(self.meta_path) as f:
self.meta = json.load(f)
else:
self.meta = {}
def _save_meta(self):
with open(self.meta_path, 'w') as f:
json.dump(self.meta, f, indent=2)
def _strength(self, slot) -> float:
"""Ebbinghaus forgetting curve: R = e^(-t/S)"""
if slot not in self.meta:
return 1.0
m = self.meta[slot]
t = (time.time() - m.get("last_access", time.time())) / 3600
S = m.get("stability", 24.0)
return float(np.exp(-t / S))
def store(self, slot, vector):
slot = str(slot)
self.memory[slot] = vector
now = time.time()
if slot not in self.meta:
self.meta[slot] = {"created": now, "access_count": 0, "stability": 24.0}
self.meta[slot]["last_access"] = now
np.save(self.path, self.memory)
self._save_meta()
def recall(self, slot):
slot = str(slot)
vec = self.memory.get(slot)
if vec is not None:
# Strengthen on recall — spaced repetition
if slot in self.meta:
self.meta[slot]["access_count"] = self.meta[slot].get("access_count", 0) + 1
self.meta[slot]["stability"] = min(
self.meta[slot].get("stability", 24.0) * 1.2, 720.0
)
self.meta[slot]["last_access"] = time.time()
self._save_meta()
return vec
def forget_weak(self, threshold=0.05):
"""Prune memories below strength threshold. Called during dream mode."""
pruned = []
for slot in list(self.memory.keys()):
if self._strength(slot) < threshold:
del self.memory[slot]
self.meta.pop(slot, None)
pruned.append(slot)
if pruned:
np.save(self.path, self.memory)
self._save_meta()
print(f"[HIPPOCAMPUS] Pruned {len(pruned)} weak memories.")
return pruned
def all_slots(self):
return list(self.memory.keys())
def memory_report(self) -> dict:
report = {}
for slot in self.memory:
report[slot] = {
"strength": round(self._strength(slot), 3),
"access_count": self.meta.get(slot, {}).get("access_count", 0),
"stability_hours": round(self.meta.get(slot, {}).get("stability", 24.0), 1)
}
return report
def similarity_search(self, query_vec, top_k=5):
qf = query_vec.astype(np.float32)
qn = np.linalg.norm(qf)
if qn == 0:
return []
results = []
for slot, vec in self.memory.items():
if vec is None:
continue
strength = self._strength(slot)
if strength < 0.01:
continue
vf = vec.astype(np.float32)
vn = np.linalg.norm(vf)
if vn == 0:
continue
# Similarity weighted by memory strength
sim = float(np.dot(qf, vf) / (qn * vn)) * strength
results.append((sim, slot))
results.sort(reverse=True)
return results[:top_k]
|