Milestone 3.5.4 — Memory Composite Scorer

Score = 0.40 × relevance_score
      + 0.25 × recency_score
      + 0.20 × usefulness_score
      + 0.10 × confidence_score
      + 0.05 × access_frequency_score

from __future__ import annotations
 
import math
from dataclasses import dataclass
from datetime import datetime, timezone
 
WEIGHT_RELEVANCE  = 0.40
WEIGHT_RECENCY    = 0.25
WEIGHT_USEFULNESS = 0.20
WEIGHT_CONFIDENCE = 0.10
WEIGHT_FREQUENCY  = 0.05
 
RECENCY_DECAY_LAMBDA = 0.05      # 14-day half-life: ln(2)/14 ≈ 0.0495
ACCESS_FREQUENCY_CAP = 50.0      # retrieval_count above this → score = 1.0
 
assert abs(WEIGHT_RELEVANCE + WEIGHT_RECENCY + WEIGHT_USEFULNESS +
           WEIGHT_CONFIDENCE + WEIGHT_FREQUENCY - 1.0) < 1e-9, \
    "Scoring weights must sum to 1.0"
 
@dataclass(frozen=True)
class ScoredMemory:
    memory:           dict
    composite_score:  float
    relevance_score:  float
    recency_score:    float
    usefulness_score: float
    confidence_score: float
    frequency_score:  float
 
class MemoryScorer:
    """
    Applies the composite scoring formula from ARCHITECTURE.md to a list of
    retrieved memories. The formula weights are constants — changing them
    requires a new ADR and re-evaluation against a test corpus.
    """
 
    def score(
        self,
        memories: list[dict],
        query_embedding: list[float],
    ) -> list[ScoredMemory]:
        """
        Score all memories and return sorted list (highest score first).
        `query_embedding` is used for relevance score (cosine similarity
        is pre-computed by the vector search; stored in memory["similarity"]).
        """
        scored = [self._score_one(mem) for mem in memories]
        scored.sort(key=lambda m: m.composite_score, reverse=True)
        return scored
 
    def _score_one(self, memory: dict) -> ScoredMemory:
        relevance  = self._relevance(memory)
        recency    = self._recency(memory)
        usefulness = self._usefulness(memory)
        confidence = self._confidence(memory)
        frequency  = self._frequency(memory)
 
        composite = (
            WEIGHT_RELEVANCE  * relevance
          + WEIGHT_RECENCY    * recency
          + WEIGHT_USEFULNESS * usefulness
          + WEIGHT_CONFIDENCE * confidence
          + WEIGHT_FREQUENCY  * frequency
        )
 
        return ScoredMemory(
            memory=memory,
            composite_score=round(composite, 6),
            relevance_score=round(relevance, 6),
            recency_score=round(recency, 6),
            usefulness_score=round(usefulness, 6),
            confidence_score=round(confidence, 6),
            frequency_score=round(frequency, 6),
        )
 
    @staticmethod
    def _relevance(memory: dict) -> float:
        """Cosine similarity from vector search; pre-computed and stored in the record."""
        return float(memory.get("similarity", 0.0))
 
    @staticmethod
    def _recency(memory: dict) -> float:
        """Exponential decay: score = exp(-λ × age_days). 14-day half-life."""
        created_at_str = memory.get("created_at")
        if not created_at_str:
            return 0.5  # unknown age: neutral score
        try:
            created_at = datetime.fromisoformat(created_at_str).replace(tzinfo=timezone.utc)
            age_days = (datetime.now(timezone.utc) - created_at).total_seconds() / 86_400
            return math.exp(-RECENCY_DECAY_LAMBDA * age_days)
        except (ValueError, TypeError):
            return 0.5
 
    @staticmethod
    def _usefulness(memory: dict) -> float:
        return float(memory.get("usefulness_score", 0.5))
 
    @staticmethod
    def _confidence(memory: dict) -> float:
        return float(memory.get("confidence", 0.8))
 
    @staticmethod
    def _frequency(memory: dict) -> float:
        count = float(memory.get("retrieval_count", 0))
        return min(count / ACCESS_FREQUENCY_CAP, 1.0)