07. Memory Consolidation

Chapter 7 of 24 · 15 min

KEY INSIGHT

Memory consolidation transfers important information between storage tiers while pruning irrelevant data, maintaining a memory system that stays useful without unbounded growth. Consolidation is the process that keeps the memory system healthy over time. Without consolidation, short-term memory grows unbounded while long-term memory accumulates noise. The consolidation strategy defines when and how memories move between tiers. Consolidation Triggers Consolidation runs on multiple schedules. Immediate consolidation happens when the agent explicitly learns something important. Periodic consolidation runs on a schedule to handle routine memory movement. Emergency consolidation triggers when storage approaches capacity limits. ```python import asyncio from datetime import datetime, timedelta from dataclasses import dataclass from enum import Enum class MemoryTier(Enum): SHORT_TERM = "short_term" MEDIUM_TERM = "medium_term" LONG_TERM = "long_term" @dataclass class ConsolidationTask: memory_id: str source_tier: MemoryTier target_tier: MemoryTier reason: str scheduled_at: datetime async def consolidation_loop(agent: "OpenCLawAgent"): """Background task that manages memory consolidation.""" while True: try: # Check for immediate consolidation requests await process_immediate_tasks(agent) # Run periodic consolidation await run_periodic_consolidation(agent) # Check capacity and consolidate if needed await check_and_emergency_consolidate(agent) except Exception as e: agent.logger.error(f"Consolidation error: {e}") await asyncio.sleep(300) # Check every 5 minutes async def run_periodic_consolidation(agent: "OpenCLawAgent"): """Consolidate memories on a scheduled basis.""" cutoff = datetime.now() - timedelta(days=7) # Get short-term memories older than cutoff candidates = await agent.memory.get_short_term_older_than(cutoff) for memory in candidates: importance = await assess_importance(agent, memory) if importance > 0.7: await promote_memory(agent, memory, MemoryTier.MEDIUM_TERM) elif importance > 0.3: await promote_memory(agent, memory, MemoryTier.LONG_TERM) else: await prune_memory(agent, memory) ``` Importance Assessment Determining which memories to keep requires assessing their importance. The assessment considers multiple factors: frequency of reference, recency of access, explicit user feedback, and relationship to other important memories. User feedback provides the strongest signal. When users indicate a memory was useful, its importance increases. When they correct the agent's understanding, the old incorrect memory may need revision or removal.

EXERCISE

Implement a memory importance scoring function that combines access frequency, recency, user feedback ratings, and semantic novelty compared to existing long-term memories. Explain how each factor should be weighted.