| """ |
| STRATA-MEMORY: 3-layer memory with continuous scores, decay, and promotion. |
| |
| All pattern values are continuous scores (0..1), never boolean. |
| Short → Mid → Long promotion based on frequency and consistency. |
| """ |
|
|
| from typing import Dict, List, Optional |
|
|
|
|
| DECAY_RATE = 0.90 |
| PROMOTE_MIN_N = 2 |
| PROMOTE_THRESHOLD = 0.50 |
| LONG_THRESHOLD = 0.65 |
| FORGET_THRESHOLD = 0.05 |
|
|
|
|
| class MemoryLayer: |
| """Single memory layer: key → {score, count}.""" |
|
|
| def __init__(self, name: str): |
| self.name = name |
| self._store: Dict[str, Dict] = {} |
|
|
| def upsert(self, key: str, score: float) -> None: |
| """Insert or update a pattern with a new score observation.""" |
| if key in self._store: |
| old = self._store[key] |
| |
| old["score"] = old["score"] * 0.6 + score * 0.4 |
| old["count"] += 1 |
| else: |
| self._store[key] = {"score": score, "count": 1} |
|
|
| def decay(self) -> None: |
| """Apply per-step importance decay and prune forgotten entries.""" |
| to_delete = [] |
| for key, entry in self._store.items(): |
| entry["score"] *= DECAY_RATE |
| if entry["score"] < FORGET_THRESHOLD: |
| to_delete.append(key) |
| for key in to_delete: |
| del self._store[key] |
|
|
| def get(self, key: str, default: float = 0.0) -> float: |
| return self._store.get(key, {}).get("score", default) |
|
|
| def get_count(self, key: str) -> int: |
| return self._store.get(key, {}).get("count", 0) |
|
|
| def all_scores(self) -> Dict[str, float]: |
| return {k: v["score"] for k, v in self._store.items()} |
|
|
| def pop(self, key: str) -> Optional[Dict]: |
| return self._store.pop(key, None) |
|
|
| def keys(self): |
| return list(self._store.keys()) |
|
|
| def __repr__(self): |
| return f"MemoryLayer({self.name}, {self._store})" |
|
|
|
|
| class StrataMEMORY: |
| """ |
| 3-layer memory system: |
| Short → recent per-step observations |
| Mid → session-level patterns (promoted from Short) |
| Long → stable pattern bank (promoted from Mid) |
| """ |
|
|
| def __init__(self): |
| self.short = MemoryLayer("short") |
| self.mid = MemoryLayer("mid") |
| self.long = MemoryLayer("long") |
|
|
| def observe(self, pattern: str, score: float) -> None: |
| """Record a pattern observation into Short memory.""" |
| self.short.upsert(pattern, score) |
|
|
| def step(self) -> None: |
| """ |
| Advance one time step: |
| 1. Try to promote Short → Mid |
| 2. Try to promote Mid → Long |
| 3. Decay all layers |
| """ |
| self._promote_short_to_mid() |
| self._promote_mid_to_long() |
| self.short.decay() |
| self.mid.decay() |
| |
| for key, entry in self.long._store.items(): |
| entry["score"] *= 0.995 |
|
|
| def _promote_short_to_mid(self) -> None: |
| for key in self.short.keys(): |
| score = self.short.get(key) |
| count = self.short.get_count(key) |
| if count >= PROMOTE_MIN_N and score >= PROMOTE_THRESHOLD: |
| self.mid.upsert(key, score) |
|
|
| def _promote_mid_to_long(self) -> None: |
| for key in self.mid.keys(): |
| score = self.mid.get(key) |
| count = self.mid.get_count(key) |
| if count >= PROMOTE_MIN_N and score >= LONG_THRESHOLD: |
| self.long.upsert(key, score) |
|
|
| def snapshot(self) -> Dict[str, float]: |
| """ |
| Return merged view of all layers for STRATA-CORE consumption. |
| Long > Mid > Short priority when keys overlap. |
| """ |
| merged = {} |
| merged.update(self.short.all_scores()) |
| merged.update(self.mid.all_scores()) |
| merged.update(self.long.all_scores()) |
| return merged |
|
|
| def __repr__(self): |
| return ( |
| f"STRATA-MEMORY\n" |
| f" Short: {self.short.all_scores()}\n" |
| f" Mid: {self.mid.all_scores()}\n" |
| f" Long: {self.long.all_scores()}" |
| ) |
|
|
