File size: 3,727 Bytes
031e289 | 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 | """Bounded episodic record plus context-specific conditioning and extinction."""
from __future__ import annotations
import math
from dataclasses import asdict, dataclass, field
from .config import AffectConfig, finite
@dataclass
class CueEstimate:
expected_harm: float = 0.0
variance: float = 0.0
observations: int = 0
updated_at: float = 0.0
@dataclass
class EpisodicMemory:
capacity: int = 256
episodes: list[dict] = field(default_factory=list)
cues: dict[str, CueEstimate] = field(default_factory=dict)
sequence: int = 0
def __post_init__(self):
if not isinstance(self.capacity, int) or self.capacity < 1:
raise ValueError("capacity must be a positive integer")
def predict(self, cue: str, now: float, cfg: AffectConfig) -> float:
estimate = self.cues.get(cue)
if estimate is None:
return 0.0
age = max(0.0, now - estimate.updated_at)
return estimate.expected_harm * 0.5 ** (age / cfg.memory_half_life)
def observe(
self,
cue: str,
harm: float,
now: float,
cfg: AffectConfig,
event: str,
levels: dict,
) -> dict:
finite(harm, "harm", 0, 1)
expected = self.predict(cue, now, cfg)
item = self.cues.setdefault(cue, CueEstimate())
error = harm - expected
rate = cfg.conditioning_rate if error >= 0 else cfg.extinction_rate
item.expected_harm = min(1.0, max(0.0, expected + rate * error))
item.variance = (1.0 - rate) * (item.variance + rate * error * error)
item.observations += 1
item.updated_at = now
self.sequence += 1
episode = {
"id": self.sequence,
"t": now,
"cue": cue,
"event": event,
"harm": harm,
"prediction_before": expected,
"prediction_error": error,
"levels": dict(levels),
"salience": max(harm, abs(error), max(levels.values())),
}
self.episodes.append(episode)
self.episodes = self.episodes[-self.capacity :]
if len(self.cues) > self.capacity:
oldest = min(self.cues, key=lambda k: self.cues[k].updated_at)
del self.cues[oldest]
return episode
def retrieve(self, cue: str, now: float, limit: int = 3) -> list[dict]:
if limit < 0:
raise ValueError("limit must be nonnegative")
matches = [e for e in self.episodes if e["cue"] == cue]
score = lambda e: 0.7 * e["salience"] + 0.3 * math.exp(
-max(0.0, now - e["t"]) / 600.0
)
return sorted(matches, key=score, reverse=True)[:limit]
def to_dict(self):
return {
"capacity": self.capacity,
"episodes": self.episodes,
"sequence": self.sequence,
"cues": {key: asdict(value) for key, value in self.cues.items()},
}
@classmethod
def from_dict(cls, data):
data = dict(data)
data["cues"] = {
key: CueEstimate(**value) for key, value in data.get("cues", {}).items()
}
for value in data["cues"].values():
finite(value.expected_harm, "expected_harm", 0, 1)
finite(value.variance, "variance", 0, 1)
finite(value.updated_at, "updated_at")
if value.observations < 0:
raise ValueError("Invalid observation count")
memory = cls(**data)
if (
memory.capacity < 1
or len(memory.episodes) > memory.capacity
or len(memory.cues) > memory.capacity
):
raise ValueError("Invalid persisted memory bounds")
return memory
|