diff --git a/src/powermem/intelligence/ebbinghaus_algorithm.py b/src/powermem/intelligence/ebbinghaus_algorithm.py index c12aa1f28..a2388803a 100644 --- a/src/powermem/intelligence/ebbinghaus_algorithm.py +++ b/src/powermem/intelligence/ebbinghaus_algorithm.py @@ -17,21 +17,22 @@ class EbbinghausAlgorithm: """ Implements Ebbinghaus forgetting curve algorithm for memory management. """ + DEFAULT_DECAY_RATE_MULTIPLIERS = { "working": 1, "short_term": 7, "long_term": 60, } - + def __init__(self, config: Dict[str, Any]): """ Initialize Ebbinghaus algorithm. - + Args: config: Algorithm configuration """ self.config = config - + # Ebbinghaus curve parameters self.initial_retention = config.get("initial_retention", 1.0) self.decay_rate = config.get("decay_rate", 1.5) @@ -39,52 +40,54 @@ def __init__(self, config: Dict[str, Any]): config.get("decay_rate_multipliers") ) self.reinforcement_factor = config.get("reinforcement_factor", 0.3) - + # Memory type thresholds self.working_threshold = config.get("working_threshold", 0.3) self.short_term_threshold = config.get("short_term_threshold", 0.6) self.long_term_threshold = config.get("long_term_threshold", 0.8) - + # Time intervals (in hours) self.review_intervals = config.get("review_intervals", [1, 6, 24, 72, 168]) # Review schedule: higher importance -> shorter intervals (shared by persist + query) self.review_adjustment_factor = config.get("review_adjustment_factor", 0.3) self.review_interval_min_hours = config.get("review_interval_min_hours", 0.5) - + logger.info("EbbinghausAlgorithm initialized") - + def process_memory_metadata( - self, - content: str, - importance_score: float, - memory_type: str + self, content: str, importance_score: float, memory_type: str ) -> Dict[str, Any]: """ Process memory using Ebbinghaus algorithm and return metadata. - + Args: content: Memory content importance_score: Importance score memory_type: Type of memory - + Returns: Dictionary containing intelligence metadata """ try: current_time = get_current_datetime() - - # Calculate initial retention based on importance - initial_retention = self.initial_retention * importance_score - + + initial_retention = self._calculate_initial_retention(importance_score) + # Calculate decay rate based on memory type decay_rate = self._get_decay_rate_for_type(memory_type) - + # Generate review schedule - review_schedule = self._generate_review_schedule(importance_score, current_time) - + review_schedule = self._generate_review_schedule( + importance_score, current_time + ) + # Calculate next review time - next_review = review_schedule[0] if review_schedule else current_time + timedelta(hours=1) - + next_review = ( + review_schedule[0] + if review_schedule + else current_time + timedelta(hours=1) + ) + intelligence_metadata = { # Ebbinghaus algorithm data "intelligence": { @@ -111,21 +114,23 @@ def process_memory_metadata( "created_at": current_time.isoformat(), "updated_at": current_time.isoformat(), } - - logger.debug(f"Generated intelligence metadata for type: {memory_type}, importance: {importance_score}") - + + logger.debug( + f"Generated intelligence metadata for type: {memory_type}, importance: {importance_score}" + ) + return intelligence_metadata - + except Exception as e: logger.error(f"Failed to process memory metadata: {e}") return { "intelligence": { "importance_score": importance_score, "memory_type": memory_type, - "error": str(e) + "error": str(e), } } - + def calculate_decay( self, created_at, @@ -133,11 +138,11 @@ def calculate_decay( ) -> float: """ Calculate decay factor based on time elapsed. - + Args: created_at: When the memory was created (datetime object or ISO string) decay_rate: Per-memory strength parameter. Larger values decay slower. - + Returns: Decay factor between 0 and 1 """ @@ -145,75 +150,73 @@ def calculate_decay( # Handle both datetime objects and ISO string formats if isinstance(created_at, str): if created_at: - created_at = datetime.fromisoformat(created_at.replace('Z', '+00:00')) + created_at = datetime.fromisoformat( + created_at.replace("Z", "+00:00") + ) else: # If empty string, use current time created_at = get_current_datetime() elif created_at is None: # If None, use current time created_at = get_current_datetime() - + time_elapsed = get_current_datetime() - created_at hours_elapsed = time_elapsed.total_seconds() / 3600 - + rate = self.decay_rate if decay_rate is None else decay_rate if rate <= 0: logger.warning("Invalid decay_rate %s, falling back to default", rate) rate = self.decay_rate - + # Ebbinghaus forgetting curve: R = e^(-t/S) # where R is retention, t is time, S is strength. decay_factor = math.exp(-hours_elapsed / (24 * rate)) - + return max(decay_factor, 0.0) - + except Exception as e: logger.error(f"Failed to calculate decay: {e}") return 0.5 - + def calculate_relevance(self, memory: Dict[str, Any], query: str) -> float: """ Calculate relevance score for a memory given a query. - + Args: memory: Memory data query: Search query - + Returns: Relevance score between 0 and 1 """ try: - content = ( - memory.get("content") - or memory.get("memory") - or "" - ).lower() + content = (memory.get("content") or memory.get("memory") or "").lower() query_lower = query.lower() - + # Simple keyword matching query_words = query_lower.split() content_words = content.split() - + matches = 0 for word in query_words: if word in content_words: matches += 1 - + relevance_score = matches / len(query_words) if query_words else 0.0 - + return min(relevance_score, 1.0) - + except Exception as e: logger.error(f"Failed to calculate relevance: {e}") return 0.0 - + def should_promote(self, memory: Dict[str, Any]) -> bool: """ Determine if a memory should be promoted to a higher tier. - + Args: memory: Memory data - + Returns: True if memory should be promoted """ @@ -222,35 +225,37 @@ def should_promote(self, memory: Dict[str, Any]) -> bool: access_count = memory.get("access_count", 0) if access_count >= 3: return True - + # Check recency — only promote if the memory has been accessed at # least once, so that old never-accessed memories can still be forgotten. created_at = memory.get("created_at") if created_at and access_count > 0: if isinstance(created_at, str): - created_at = datetime.fromisoformat(created_at.replace('Z', '+00:00')) + created_at = datetime.fromisoformat( + created_at.replace("Z", "+00:00") + ) time_elapsed = get_current_datetime() - created_at if time_elapsed > timedelta(hours=24): return True - + # Check importance importance = memory.get("importance_score", 0.5) if importance >= self.short_term_threshold: return True - + return False - + except Exception as e: logger.error(f"Failed to check promotion: {e}") return False - + def should_forget(self, memory: Dict[str, Any]) -> bool: """ Determine if a memory should be forgotten. - + Args: memory: Memory data - + Returns: True if memory should be forgotten """ @@ -258,18 +263,18 @@ def should_forget(self, memory: Dict[str, Any]) -> bool: if self.calculate_current_retention(memory) < self.working_threshold: return True return False - + except Exception as e: logger.error(f"Failed to check forgetting: {e}") return False - + def should_archive(self, memory: Dict[str, Any]) -> bool: """ Determine if a memory should be archived. - + Args: memory: Memory data - + Returns: True if memory should be archived """ @@ -279,22 +284,24 @@ def should_archive(self, memory: Dict[str, Any]) -> bool: if created_at: # Parse string to datetime if needed if isinstance(created_at, str): - created_at = datetime.fromisoformat(created_at.replace('Z', '+00:00')) + created_at = datetime.fromisoformat( + created_at.replace("Z", "+00:00") + ) time_elapsed = get_current_datetime() - created_at if time_elapsed > timedelta(days=30): return True - + # Check importance importance = memory.get("importance_score", 0.5) if importance < self.working_threshold: return True - + return False - + except Exception as e: logger.error(f"Failed to check archiving: {e}") return False - + def reinforce(self, memory: Dict[str, Any]) -> Dict[str, Any]: """Boost current_retention on review and advance the review schedule. @@ -383,7 +390,7 @@ def get_review_schedule( except Exception as e: logger.error(f"Failed to get review schedule: {e}") return [] - + def _load_decay_rate_multipliers( self, raw: Optional[Dict[str, Any]] ) -> Dict[str, float]: @@ -394,16 +401,16 @@ def _load_decay_rate_multipliers( try: multiplier = float(value) except (TypeError, ValueError): - logger.warning("Invalid decay multiplier for %s: %s", memory_type, value) + logger.warning( + "Invalid decay multiplier for %s: %s", memory_type, value + ) continue if multiplier > 0: multipliers[memory_type] = multiplier self._validate_decay_rate_multipliers(multipliers) return multipliers - def _validate_decay_rate_multipliers( - self, multipliers: Dict[str, float] - ) -> None: + def _validate_decay_rate_multipliers(self, multipliers: Dict[str, float]) -> None: """Warn when default tier ordering would not make higher tiers last longer.""" working = multipliers.get("working", self.decay_rate) short_term = multipliers.get("short_term", self.decay_rate) @@ -413,7 +420,7 @@ def _validate_decay_rate_multipliers( "decay_rate_multipliers should satisfy " "working < short_term < long_term" ) - + def _get_decay_rate_for_type(self, memory_type: str) -> float: """Get decay strength S based on memory type; larger S decays slower.""" multiplier = self.decay_rate_multipliers.get(memory_type) @@ -454,11 +461,7 @@ def _resolve_metadata_sections( ) -> Tuple[Dict[str, Any], Dict[str, Any]]: """Return metadata and intelligence sections from supported layouts.""" meta = memory.get("metadata") or {} - intelligence = ( - meta.get("intelligence") - or memory.get("intelligence") - or {} - ) + intelligence = meta.get("intelligence") or memory.get("intelligence") or {} return meta, intelligence @staticmethod @@ -469,9 +472,7 @@ def _first_present(*values: Any) -> Any: return value return None - def _apply_reinforcement( - self, memory: Dict[str, Any], base_rate: float - ) -> float: + def _apply_reinforcement(self, memory: Dict[str, Any], base_rate: float) -> float: """Increase decay strength with diminishing returns on access frequency.""" access_count = self._resolve_access_count(memory) reinforcement_factor = self._resolve_reinforcement_factor(memory) @@ -526,9 +527,22 @@ def _resolve_initial_retention(self, memory: Dict[str, Any]) -> float: return val return self.initial_retention - def _resolve_current_retention( - self, memory: Dict[str, Any] - ) -> Optional[float]: + def _calculate_initial_retention(self, importance_score: float) -> float: + """Calculate bounded initial retention for newly persisted memories.""" + try: + score = float(importance_score) + except (TypeError, ValueError): + logger.warning( + "Invalid importance_score for retention: %s", importance_score + ) + score = 0.0 + + score = max(0.0, min(1.0, score)) + max_retention = max(0.0, min(1.0, float(self.initial_retention))) + retention_floor = min(max_retention, max(0.0, min(1.0, self.working_threshold))) + return max(max_retention * score, retention_floor) + + def _resolve_current_retention(self, memory: Dict[str, Any]) -> Optional[float]: """Resolve stored current_retention as a bounded snapshot value.""" meta, intelligence = self._resolve_metadata_sections(memory) raw = self._first_present( @@ -624,9 +638,7 @@ def _build_review_schedule( ) -> List[datetime]: """Build review datetimes from importance and anchor time.""" intervals = self._adjust_review_intervals(importance_score) - return [ - created_at + timedelta(hours=hours) for hours in intervals - ] + return [created_at + timedelta(hours=hours) for hours in intervals] def _generate_review_schedule( self, importance_score: float, created_at: datetime diff --git a/tests/unit/intelligence/test_retention_runtime.py b/tests/unit/intelligence/test_retention_runtime.py index 66bc0c036..e30c86ece 100644 --- a/tests/unit/intelligence/test_retention_runtime.py +++ b/tests/unit/intelligence/test_retention_runtime.py @@ -33,6 +33,7 @@ def algo_low_retention(): # ---- Test 1: should_forget considers initial_retention ---- + def test_should_forget_considers_initial_retention(algo): """High-importance memory should survive longer than low-importance at same age.""" created_at = get_current_datetime() - timedelta(hours=30) @@ -66,6 +67,7 @@ def test_should_forget_considers_initial_retention(algo): # ---- Test 2: reinforce boosts current_retention ---- + def test_reinforce_boosts_current_retention(algo): """reinforce() should increase current_retention with diminishing returns.""" now = get_current_datetime() @@ -114,6 +116,7 @@ def test_reinforce_never_exceeds_one(algo): # ---- Test 3: on_get triggers reinforcement when review is due ---- + def test_on_get_triggers_reinforcement_when_review_due(): """When now >= next_review, on_get should boost current_retention.""" config = { @@ -146,7 +149,7 @@ def test_on_get_triggers_reinforcement_when_review_due(): "review_count": 0, "next_review": past_review, "review_schedule": [past_review, future_review], - } + }, }, } @@ -162,6 +165,7 @@ def test_on_get_triggers_reinforcement_when_review_due(): # ---- Test 4: on_get does NOT reinforce before review time ---- + def test_on_get_does_not_reinforce_before_review_due(): """When now < next_review, current_retention should not change via reinforce.""" config = { @@ -193,7 +197,7 @@ def test_on_get_does_not_reinforce_before_review_due(): "review_count": 0, "next_review": future_review, "review_schedule": [future_review], - } + }, }, } @@ -208,6 +212,7 @@ def test_on_get_does_not_reinforce_before_review_due(): # ---- Test 5: reprocessing preserves current_retention ---- + def test_reprocess_preserves_current_retention(): """When access_count%5 triggers reprocessing, current_retention from reinforcement should not be reset to initial_retention.""" @@ -243,7 +248,7 @@ def test_reprocess_preserves_current_retention(): "last_reviewed": (now - timedelta(hours=1)).isoformat(), "next_review": past_review, "review_schedule": [past_review, future_review], - } + }, }, } @@ -260,6 +265,7 @@ def test_reprocess_preserves_current_retention(): # ---- Test 6: calculate_current_retention combines initial and decay ---- + def test_calculate_current_retention_combines_initial_and_decay(algo): """Without stored current_retention, should return initial_retention * decay_factor.""" created_at = get_current_datetime() - timedelta(hours=24) @@ -408,9 +414,7 @@ def test_reinforce_uses_decayed_current_retention_before_boost(algo): result = algo.reinforce(memory) - assert result["current_retention"] == pytest.approx( - decayed + 0.3 * (1.0 - decayed) - ) + assert result["current_retention"] == pytest.approx(decayed + 0.3 * (1.0 - decayed)) assert result["current_retention"] < 0.8 @@ -429,9 +433,7 @@ def test_search_ranking_uses_effective_retention(): "created_at": created_at, "memory_type": "working", "access_count": 0, - "metadata": { - "intelligence": {"initial_retention": 0.3} - }, + "metadata": {"intelligence": {"initial_retention": 0.3}}, }, { "id": "high-init", @@ -440,9 +442,7 @@ def test_search_ranking_uses_effective_retention(): "created_at": created_at, "memory_type": "working", "access_count": 0, - "metadata": { - "intelligence": {"initial_retention": 0.95} - }, + "metadata": {"intelligence": {"initial_retention": 0.95}}, }, ] @@ -453,8 +453,62 @@ def test_search_ranking_uses_effective_retention(): assert "effective_retention" in by_id["high-init"] +def test_fresh_zero_importance_memory_keeps_search_relevance(): + """A fresh relevant hit should not be zeroed out by importance_score=0.""" + manager = IntelligentMemoryManager( + {"intelligent_memory": {"decay_rate": 1.5, "initial_retention": 1.0}} + ) + algo = manager.ebbinghaus_algorithm + now = get_current_datetime() + relevant_meta = algo.process_memory_metadata( + "Zhang San is a software engineer", + importance_score=0.0, + memory_type="working", + ) + + results = [ + { + "id": "relevant", + "content": "Zhang San is a software engineer", + "score": 0.62, + "created_at": now, + "memory_type": "working", + "access_count": 0, + "metadata": { + "memory_type": "working", + "intelligence": relevant_meta["intelligence"], + }, + }, + { + "id": "irrelevant", + "content": "Wang Wu likes running", + "score": 0.01, + "created_at": now, + "memory_type": "working", + "access_count": 0, + "metadata": { + "memory_type": "working", + "intelligence": { + "initial_retention": 1.0, + "current_retention": 1.0, + "last_reviewed": now.isoformat(), + }, + }, + }, + ] + + processed = manager.process_search_results(results, "Zhang San occupation") + + assert relevant_meta["intelligence"]["initial_retention"] == pytest.approx( + algo.working_threshold + ) + assert processed[0]["id"] == "relevant" + assert processed[0]["final_score"] > 0 + + # ---- Tests for review-reinforcement protecting against forgetting ---- + def test_should_forget_respects_stored_current_retention(algo): """Recent reinforced retention can protect a memory from forgetting.""" created_at = get_current_datetime() - timedelta(hours=50) @@ -512,7 +566,7 @@ def test_on_get_reinforced_memory_not_forgotten_same_call(): "review_count": 0, "next_review": past_review, "review_schedule": [past_review, future_review], - } + }, }, } @@ -570,9 +624,7 @@ def test_search_ranking_reflects_reinforced_current_retention(): "created_at": created_at, "memory_type": "working", "access_count": 0, - "metadata": { - "intelligence": {"initial_retention": 0.3} - }, + "metadata": {"intelligence": {"initial_retention": 0.3}}, }, { "id": "reinforced", @@ -594,5 +646,8 @@ def test_search_ranking_reflects_reinforced_current_retention(): processed = manager.process_search_results(results, "keyword") by_id = {item["id"]: item for item in processed} - assert by_id["reinforced"]["effective_retention"] > by_id["unreinforced"]["effective_retention"] + assert ( + by_id["reinforced"]["effective_retention"] + > by_id["unreinforced"]["effective_retention"] + ) assert processed[0]["id"] == "reinforced"