LLM Memory Systems: Building AI That Remembers

from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime

@dataclass
class Memory:
    content: str
    memory_type: str  # "fact", "episode", "preference"
    importance: float  # 0-1
    created_at: datetime
    last_accessed: datetime
    access_count: int

class MemorySystem:
    def __init__(self, redis, vector_db, postgres):
        self.working = WorkingMemory(max_tokens=4000)
        self.short_term = ShortTermMemory(redis, ttl_hours=24)
        self.long_term = LongTermMemory(vector_db, postgres)
    
    async def remember(self, user_id: str, content: str, memory_type: str):
        """Store a new memory with appropriate importance scoring."""
        
        # Calculate importance based on content analysis
        importance = await self._calculate_importance(content, memory_type)
        
        memory = Memory(
            content=content,
            memory_type=memory_type,
            importance=importance,
            created_at=datetime.now(),
            last_accessed=datetime.now(),
            access_count=0
        )
        
        # Store in appropriate layer based on importance
        if importance > 0.8:
            await self.long_term.store(user_id, memory)
        else:
            await self.short_term.store(user_id, memory)
    
    async def recall(self, user_id: str, query: str, k: int = 5) -> List[Memory]:
        """Retrieve relevant memories for the current context."""
        
        # Get from all layers
        working_context = self.working.get_context()
        short_term = await self.short_term.search(user_id, query, k=k)
        long_term = await self.long_term.search(user_id, query, k=k)
        
        # Combine and rank by relevance + recency + importance
        all_memories = short_term + long_term
        ranked = self._rank_memories(all_memories, query)
        
        # Update access patterns
        for memory in ranked[:k]:
            memory.last_accessed = datetime.now()
            memory.access_count += 1
        
        return ranked[:k]
    
    async def consolidate(self, user_id: str):
        """Background job: promote important short-term to long-term."""
        
        # Find frequently accessed short-term memories
        candidates = await self.short_term.get_high_access(user_id, min_count=3)
        
        for memory in candidates:
            # Extract durable facts
            facts = await self._extract_facts(memory.content)
            for fact in facts:
                await self.long_term.store(user_id, Memory(
                    content=fact,
                    memory_type="fact",
                    importance=memory.importance,
                    created_at=datetime.now(),
                    last_accessed=datetime.now(),
                    access_count=0
                ))
            
            # Remove from short-term
            await self.short_term.delete(user_id, memory.id)
    
    def _rank_memories(self, memories: List[Memory], query: str) -> List[Memory]:
        """Rank memories by composite score."""
        
        def score(m: Memory) -> float:
            # Recency decay
            age_hours = (datetime.now() - m.last_accessed).total_seconds() / 3600
            recency = 1 / (1 + age_hours / 24)
            
            # Combine factors
            return (
                0.4 * m.importance +
                0.3 * recency +
                0.2 * min(m.access_count / 10, 1) +
                0.1 * self._semantic_similarity(m.content, query)
            )
        
        return sorted(memories, key=score, reverse=True)