ZerdaZerda

MemorySelector

On this page

Purpose

Scores every candidate MemoryEntry against the current query on three axes — relevance (cosine similarity), recency (exponential decay), and importance (the entry's own effective_importance) — combines them into a weighted composite score, and returns the top entries that fit within a token budget. This is the ranking step between raw candidate gathering (STM + LTM + semantic search results) and WorkingMemory.load().

SelectionConfig (dataclass)

Python · 9 lines
@dataclass
class SelectionConfig:
    relevance_weight: float = 0.50
    recency_weight: float = 0.20
    importance_weight: float = 0.30
    recency_half_life_hours: float = 24.0
    min_composite_score: float = 0.10
    top_k: int = 10
    token_budget: int = 3000

__post_init__ validates that the three weights sum to 1.0 (within rel_tol=1e-3 via math.isclose), raising ValueError(f"Weights must sum to 1.0, got {total:.3f}") otherwise. No other field is validated — recency_half_life_hours <= 0, min_composite_score outside [0, 1], top_k <= 0, or token_budget <= 0 are all accepted silently and produce degenerate-but-non-crashing behavior (e.g. top_k=0 simply selects nothing via _apply_token_budget's len(selected) >= 0 immediately breaking).

ScoredMemory (dataclass)

Python · 9 lines
@dataclass
class ScoredMemory:
    entry: MemoryEntry
    relevance_score: float = 0.0
    recency_score: float = 0.0
    importance_score: float = 0.0
    composite_score: float = 0.0

    def to_dict(self) -> Dict[str, Any]: ...

to_dict() truncates content to 120 characters (via str(entry.content)[:120]) and rounds all scores to 4 decimal places — intended for logging/debugging (see explain() below), not for reconstructing a full entry.

MemorySelector constructor

Python
def __init__(self, config: Optional[SelectionConfig] = None) -> None

Defaults to SelectionConfig() if none given.

select()

Python · 4 lines
def select(
    self, query: str, candidates: List[MemoryEntry],
    similarity_scores: Optional[Dict[str, float]] = None,
) -> List[ScoredMemory]
  1. Returns [] immediately if candidates is empty.
  2. For each candidate: rel = similarity_scores.get(entry.id, 0.5) (defaults to 0.5 — a neutral middle score — for any entry with no precomputed similarity, not 0.0), rec = _recency_score(entry, now), imp = entry.effective_importance.
  3. Computes composite = relevance_weight*rel + recency_weight*rec + importance_weight*imp.
  4. Filters out any candidate whose composite score is below config.min_composite_score — these are dropped entirely, not merely deprioritized.
  5. Sorts survivors by composite_score descending.
  6. Passes to _apply_token_budget() (see below) for final trimming.

Note that query itself is not actually used inside select() when similarity_scores is provided — it's accepted as a parameter (and documented as being used "when similarity_scores not provided") but there is no code path in this method that computes relevance from query directly; if you call select() without a similarity_scores dict, every candidate simply gets the same flat rel = 0.5, regardless of query's content. Actual query-based relevance only happens upstream, wherever you compute similarity_scores (e.g. via SemanticMemory.search()).

_recency_score(entry, now) -> float

Python · 3 lines
hours_old = (now - entry.timestamp) / 3600.0
lam = math.log(2) / half_life
return math.exp(-lam * hours_old)

Exponential decay based on entry.timestamp (creation time), not entry.last_access — an entry accessed recently but created long ago still scores low on recency here. This differs from ForgettingMechanism, which bases its decay on last_access.

_apply_token_budget(scored) -> List[ScoredMemory] (internal)

Python · 11 lines
def _apply_token_budget(self, scored):
    selected, used = [], 0
    for sm in scored:
        if len(selected) >= self.config.top_k:
            break
        tokens = self._token_estimate(sm.entry)
        if used + tokens > self.config.token_budget:
            continue
        selected.append(sm)
        used += tokens
    return selected

[!DANGER] Docstring says 'fit the budget' but the implementation skips, not stops The select() docstring states it returns "the top_k that fit the token budget," implying a straightforward top-k-by-score cutoff limited by budget. The actual loop uses continue when an entry doesn't fit — meaning it keeps checking smaller, lower-ranked entries further down the sorted list rather than stopping. Concretely: if the #1-ranked entry is too large to fit the remaining budget, but the #2-ranked entry is smaller and fits, _apply_token_budget will select #2 (skipping #1) rather than stopping there. This is a reasonable greedy packing strategy in isolation, but it means the final selection isn't strictly "top_k by composite score" when budget constraints bind — it's "the highest-scoring subset that happens to fit, evaluated greedily in score order." Contrast with WorkingMemory.load(), which uses break instead, stopping entirely at the first entry that doesn't fit. See Known Issues.

select_entries()

Python
def select_entries(self, query, candidates, similarity_scores=None) -> List[MemoryEntry]

[sm.entry for sm in self.select(...)] — the entry-only convenience form used by AIMemoryManager.build_context().

explain()

Python
def explain(self, query, candidates, similarity_scores=None) -> List[Dict[str, Any]]

[sm.to_dict() for sm in self.select(...)] — a diagnostic, human-readable breakdown of the same selection, useful for debugging why a particular memory was or wasn't chosen.

Raises

ExceptionWhen
ValueErrorSelectionConfig() constructed with weights that don't sum to 1.0 (within 1e-3 tolerance).

Thread safety

Read-only with respect to instance state (self.config is set once and never mutated) — select/select_entries/explain can be called concurrently from multiple threads on the same MemorySelector instance.

Complexity

O(n log n) for the sort in select(), where n = len(candidates). _apply_token_budget is O(min(n, top_k)) in the best case but can scan all n candidates in the worst case (if many are skipped for exceeding remaining budget).

Examples

Python · 36 lines
from zerda.memory import MemorySelector, SelectionConfig, MemoryEntry, MemoryType
import time

selector = MemorySelector(SelectionConfig(top_k=5, token_budget=1000))

candidates = [
    MemoryEntry(content={"input": f"fact {i}", "output": "ok"},
                timestamp=time.time() - i * 3600, memory_type=MemoryType.SEMANTIC,
                importance=0.5)
    for i in range(20)
]

# 1. Basic — no precomputed similarity scores (all get rel=0.5, query has no effect here)
selected = selector.select_entries("irrelevant query text", candidates)

# 2. Intermediate — with real similarity scores from SemanticMemory.search()
sim_scores = {candidates[0].id: 0.9, candidates[5].id: 0.7}
selected2 = selector.select_entries("What facts do we know?", candidates, sim_scores)

# 3. Advanced — diagnostic breakdown
breakdown = selector.explain("query", candidates, sim_scores)
for row in breakdown[:3]:
    print(row)

# 4. Error handling — invalid weights
try:
    SelectionConfig(relevance_weight=0.5, recency_weight=0.5, importance_weight=0.5)
except ValueError as e:
    print(e)  # "Weights must sum to 1.0, got 1.500"

# 5. Production — tune weights toward recency for a live-chat use case
live_chat_config = SelectionConfig(
    relevance_weight=0.4, recency_weight=0.4, importance_weight=0.2,
    recency_half_life_hours=2.0,  # much shorter half-life
)
live_selector = MemorySelector(live_chat_config)

Best practices

  • Always compute similarity_scores upstream (e.g. via SemanticMemory.search()) before calling select() — passing query alone with no similarity_scores produces a flat, query-independent relevance score for every candidate.
  • If you need strict top-k-by-score behavior when budget-constrained, don't rely on _apply_token_budget's current continue-based logic — pre-filter or post-verify against your own top-k requirement, or subclass/reimplement if this greedy substitution behavior isn't acceptable for your use case.

Common mistakes

  • Assuming select(query, candidates) (without similarity_scores) actually ranks by semantic relevance to query — it doesn't; it flat-scores every candidate at 0.5 relevance.
  • Assuming the returned list is strictly "top_k by composite score" when the token budget binds — a lower-ranked, smaller entry can appear ahead of a higher-ranked, larger one that was skipped.

Related pages