76 lines
3.0 KiB
Python
76 lines
3.0 KiB
Python
import sqlite3
|
|
import numpy as np
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from .emma_mdp import EmmaAction
|
|
from .emma_resonance import _cosine
|
|
|
|
DB_PATH = Path(__file__).parent / "data" / "morphic.db"
|
|
|
|
|
|
class PersistentMorphicMemory:
|
|
"""SQLite-persistent Morphic Memory. Overlever restart."""
|
|
|
|
def __init__(self, db_path: Path = DB_PATH, decay_rate: float = 0.95):
|
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self.db_path = db_path
|
|
self.decay = decay_rate
|
|
self._init_db()
|
|
|
|
def _init_db(self):
|
|
with sqlite3.connect(self.db_path) as con:
|
|
con.execute("""CREATE TABLE IF NOT EXISTS patterns (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
embedding BLOB NOT NULL,
|
|
action TEXT NOT NULL,
|
|
reward REAL NOT NULL,
|
|
count INTEGER NOT NULL DEFAULT 1,
|
|
lastseen TEXT NOT NULL
|
|
)""")
|
|
|
|
def store(self, state_embedding: np.ndarray, action: EmmaAction, reward_val: float):
|
|
rows = self._all_rows()
|
|
for row in rows:
|
|
emb = np.frombuffer(row["embedding"], dtype=np.float32)
|
|
if _cosine(state_embedding, emb) > 0.92:
|
|
new_reward = 0.7 * row["reward"] + 0.3 * reward_val
|
|
with sqlite3.connect(self.db_path) as con:
|
|
con.execute(
|
|
"UPDATE patterns SET count=count+1, reward=?, lastseen=? WHERE id=?",
|
|
(new_reward, datetime.now().isoformat(), row["id"])
|
|
)
|
|
return
|
|
with sqlite3.connect(self.db_path) as con:
|
|
con.execute(
|
|
"INSERT INTO patterns (embedding, action, reward, count, lastseen) VALUES (?,?,?,?,?)",
|
|
(state_embedding.astype(np.float32).tobytes(), action.name, reward_val, 1, datetime.now().isoformat())
|
|
)
|
|
|
|
def recall(self, state_embedding: np.ndarray, top_k: int = 3) -> list[dict]:
|
|
rows = self._all_rows()
|
|
scored = []
|
|
now = datetime.now()
|
|
for row in rows:
|
|
emb = np.frombuffer(row["embedding"], dtype=np.float32)
|
|
sim = _cosine(state_embedding, emb)
|
|
age_days = (now - datetime.fromisoformat(row["lastseen"])).days
|
|
recency = self.decay ** age_days
|
|
resonance = sim * np.log1p(row["count"]) * recency * row["reward"]
|
|
scored.append((resonance, row))
|
|
return [r for _, r in sorted(scored, reverse=True)[:top_k]]
|
|
|
|
def __len__(self) -> int:
|
|
with sqlite3.connect(self.db_path) as con:
|
|
return con.execute("SELECT COUNT(*) FROM patterns").fetchone()[0]
|
|
|
|
def stats(self):
|
|
with sqlite3.connect(self.db_path) as con:
|
|
return con.execute(
|
|
"SELECT action, reward, count FROM patterns ORDER BY count DESC LIMIT 5"
|
|
).fetchall()
|
|
|
|
def _all_rows(self) -> list[dict]:
|
|
with sqlite3.connect(self.db_path) as con:
|
|
con.row_factory = sqlite3.Row
|
|
return [dict(r) for r in con.execute("SELECT * FROM patterns").fetchall()]
|