Retrieval-Augmented Memory

This is RAG whose corpus happens to be the user's own history. The mechanics are familiar; what differs is that relevance is not the only thing that matters, because a memory has an age and a weight as well as a meaning.

Two kinds of memory, two behaviours

Semantic memories are statements believed currently true — "deploys on Fridays are forbidden" — and should not decay with age; they should be superseded when contradicted. Episodic memories are records of events — "on 12 March we debugged a connection pool leak" — and are inherently historical; they decay in relevance but remain true. Storing them in one undifferentiated pile is why memory systems return stale preferences and forget useful incidents.

class Memory(BaseModel):
    id: str
    kind: Literal["semantic", "episodic"]
    text: str
    created_at: datetime
    importance: float = Field(ge=0, le=1)   # assigned at write time
    superseded_by: str | None = None        # semantic only
    last_accessed: datetime | None = None


HALF_LIFE_DAYS = {"episodic": 45.0, "semantic": math.inf}  # facts do not fade

Score on relevance, recency, and importance

Retrieve a candidate set by similarity, then re-score. Recency uses exponential decay with a per-kind half-life. Importance is set at write time — a stated constraint outranks an offhand remark. Superseded memories are excluded entirely rather than down-weighted, because a contradicted fact is not a weak signal, it is a wrong one.

W_RELEVANCE, W_RECENCY, W_IMPORTANCE = 0.55, 0.25, 0.20


def score(m: Memory, similarity: float, now: datetime) -> float:
    half_life = HALF_LIFE_DAYS[m.kind]
    age_days = (now - m.created_at).total_seconds() / 86_400
    recency = 1.0 if half_life == math.inf else 0.5 ** (age_days / half_life)
    return (W_RELEVANCE * similarity) + (W_RECENCY * recency) + (W_IMPORTANCE * m.importance)


def recall(user_id: str, query: str, k: int = 6) -> list[Memory]:
    now = datetime.now(UTC)
    hits = memory_index.search(embed(query), user_id=user_id, limit=40)

    live = [(m, s) for m, s in hits if m.superseded_by is None]   # never recall a corrected fact
    ranked = sorted(live, key=lambda p: score(p[0], p[1], now), reverse=True)[:k]

    for m, _ in ranked:
        memory_index.touch(m.id, now)   # access reinforces; see forgetting below
    return [m for m, _ in ranked]

Forgetting is a feature

Unbounded memory degrades: retrieval quality falls as the index fills with material never recalled, and storage cost grows forever. Implement decay explicitly — prune episodic memories that are old, unimportant, and never accessed, and consolidate clusters of related episodes into one semantic fact. That mirrors how memory is supposed to work: specifics fade, the lesson remains.

def prune(user_id: str, now: datetime) -> int:
    stale = memory_index.query(
        user_id=user_id, kind="episodic",
        created_before=now - timedelta(days=180),
        importance_below=0.3,
        never_accessed=True,
    )
    for m in stale:
        memory_index.delete(m.id)
    return len(stale)


def consolidate(user_id: str) -> None:
    """Five episodes about the same thing become one semantic fact."""
    for cluster in memory_index.cluster(user_id, kind="episodic", min_size=5):
        fact = summarize_to_fact(cluster)
        memory_index.add(Memory(kind="semantic", text=fact, importance=0.6, created_at=datetime.now(UTC)))
        memory_index.archive([m.id for m in cluster])

• Separate semantic (durable, supersedable) from episodic (historical, decaying)
• Rank on relevance plus recency decay plus write-time importance, not similarity alone
• Exclude superseded memories outright — they are wrong, not merely weak
• Render timestamps with recalled memories so age is visible to the model and the user
• Prune and consolidate on a schedule; unbounded memory degrades retrieval quality