Implement Semantic Memory Eviction with Relevance Scoring

Hard
Agents

Semantic Memory Eviction with Relevance Scoring

Naive LRU eviction loses important memories. Smart eviction uses multiple signals.

Task

Implement SemanticMemoryManager with intelligent eviction:

  1. Add entries with semantic embeddings.
  2. Retrieve by semantic similarity (cosine), updating access stats.
  3. Evict lowest-scoring entries considering: importance, recency, access frequency.
  4. Temporal decay reduces score of stale, unaccessed entries.
  5. Consolidate: merge semantically similar entries to free space.

Scoring Formula

score = importance × temporal_decay × log(1 + access_count)

Where:

temporal_decay = decay_factor ^ hours_since_last_access

Constraints

  • Capacity enforced after every add.
  • consolidate() merges entries with cosine similarity > 0.9.
  • Eviction log must record what was evicted and why.

Examples

Example 1:
Input: mgr = SemanticMemoryManager(capacity=3, embed_fn=my_embed) mgr.add('Python basics', importance=0.5) mgr.add('Math formulas', importance=0.9) mgr.add('Weather today', importance=0.2) mgr.add('New entry', importance=0.6) # Should evict 'Weather today'
Output: Evicted: 'Weather today' (lowest composite score)
Explanation: Low importance (0.2) + never accessed → lowest score.

Starter Code

from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
import math

@dataclass
class MemoryEntry:
    entry_id: str
    content: str
    embedding: List[float]
    created_at: float
    last_accessed: float
    access_count: int = 0
    importance: float = 0.5  # 0-1, manually or automatically assigned
    decay_factor: float = 0.95  # per-entry decay rate
    tags: List[str] = field(default_factory=list)

class SemanticMemoryManager:
    def __init__(self, capacity: int, embed_fn: callable):
        self.capacity = capacity
        self.embed_fn = embed_fn
        self.entries: Dict[str, MemoryEntry] = {}
        self.eviction_log: List[Dict] = []

    def add(self, content: str, importance: float = 0.5, tags: List[str] = None) -> str:
        # TODO: Add entry, evict if over capacity
        pass

    def retrieve(self, query: str, k: int = 5) -> List[MemoryEntry]:
        # TODO: Semantic search + update access stats
        pass

    def _score_entry(self, entry: MemoryEntry, current_time: float) -> float:
        # TODO: Combined relevance score for eviction decision
        # Score = importance * recency * access_frequency
        pass

    def _evict(self) -> Optional[MemoryEntry]:
        # TODO: Remove lowest-scoring entry
        pass

    def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
        pass

    def _temporal_decay(self, entry: MemoryEntry, current_time: float) -> float:
        # TODO: Exponential decay based on time since last access
        pass

    def consolidate(self, llm_fn: callable) -> List[str]:
        # TODO: Merge similar entries to free capacity
        pass
Lines: 1Characters: 0
Ready
The AI Interview - Master AI/ML Interviews