feat(memory): complete M.2-M.5 memory upgrade phases with tests

- M.2: ForgettingCurve, MemoryDecay, MemoryReinforcement (selective forgetting)
- M.3: DailyDigestGenerator, ReminderScheduler, ProactiveInformer (proactive reminders)
- M.4: MemoryExtractor with LLM-based memory extraction from conversations
- M.5: MemoryRecallInjector with token budget control for prompt injection
- All phases include comprehensive unit tests (109 tests passing)
- Updated checklist.md to mark all tasks complete
This commit is contained in:
2026-04-05 14:09:51 +08:00
parent 9bfa0dcc11
commit 11160ec4d2
22 changed files with 4117 additions and 186 deletions

View File

@@ -0,0 +1,70 @@
"""
ForgettingCurve
Calculates memory decay based on Ebbinghaus forgetting curve.
decay_score = exp(-days_since_access / half_life)
Importance level affects half-life:
- high: half_life = 30 * 3 = 90 days (slowest decay)
- medium: half_life = 30 * 1 = 30 days
- low: half_life = 30 * 0.5 = 15 days (fastest decay)
"""
import math
from datetime import UTC, datetime
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from app.models.memory import UserMemory
class ForgettingCurve:
"""Calculate memory decay based on time and importance."""
BASE_HALF_LIFE_DAYS = 30
# Half-life multipliers by importance level
HALF_LIFE_MULTIPLIERS = {
"high": 3.0,
"medium": 1.0,
"low": 0.5,
}
def calculate_decay(self, memory: "UserMemory") -> float:
"""Calculate decay score (0.0-1.0). Higher = more remembered.
Uses exponential decay: exp(-days_since_access / half_life)
"""
last_accessed = getattr(memory, "last_accessed_at", None) or getattr(
memory, "last_recalled_at", None
)
if last_accessed is None:
return 1.0 # Never accessed = full retention
now = datetime.now(UTC)
if isinstance(last_accessed, datetime):
if last_accessed.tzinfo is None:
last_accessed = last_accessed.replace(tzinfo=UTC)
days_since = (now - last_accessed).total_seconds() / 86400
else:
days_since = 0
half_life = self.get_half_life(memory)
decay = math.exp(-days_since / half_life)
return min(1.0, max(0.0, decay))
def get_half_life(self, memory: "UserMemory") -> float:
"""Get half-life in days based on importance level."""
importance_level = getattr(memory, "importance_level", "medium") or "medium"
multiplier = self.HALF_LIFE_MULTIPLIERS.get(importance_level, 1.0)
return self.BASE_HALF_LIFE_DAYS * multiplier
def should_archive(self, memory: "UserMemory") -> bool:
"""decay < 0.2 → memory should be archived (cold storage)."""
decay = self.calculate_decay(memory)
return decay < 0.2
def should_deprioritize(self, memory: "UserMemory") -> bool:
"""decay < 0.5 → memory should be deprioritized (not in active reminders)."""
decay = self.calculate_decay(memory)
return decay < 0.5