Memory Retrieval Strategies

Duration: 15 min

Memory Retrieval Strategies

Duration: 15 min

Overview

Storing memories is only half the problem - retrieving the right memories at the right time is what makes an agent effective. A poor retrieval strategy means the agent either misses critical context or floods its working memory with irrelevant information. This module covers the core retrieval strategies: recency-based, relevance-based, importance-based, and hybrid approaches that combine multiple signals.

Key Concepts & Foundations

  • What: Memory retrieval is the process of selecting which stored memories to inject into the agent's active context
  • Why: Retrieving wrong or irrelevant memories degrades response quality; retrieving too many wastes context space
  • How: Score memories by multiple signals (recency, relevance, importance, frequency) and select top candidates

Retrieval Signals

| Signal | Measures | Example | |--------|----------|---------| | Recency | How recently the memory was created/accessed | Prefer today's memories over last month's | | Relevance | Semantic similarity to current query | Prefer memories about the current topic | | Importance | How critical the memory is | User's name > random fact mentioned once | | Frequency | How often the memory has been accessed | Frequently recalled memories are likely useful |

The Retrieval Pipeline

1. Generate query - Convert current context into a retrieval query 2. Candidate retrieval - Get initial candidates (fast, broad) 3. Re-ranking - Score candidates by multiple signals (slower, precise) 4. Selection - Pick top-K memories within token budget 5. Formatting - Structure retrieved memories for the LLM prompt

Hands-On Implementation

import math
from dataclasses import dataclass, field
from datetime import datetime, timedelta

@dataclass
class Memory:
    content: str
    created_at: datetime = field(default_factory=datetime.now)
    last_accessed: datetime = field(default_factory=datetime.now)
    importance: float = 0.5
    access_count: int = 0
    tags: list[str] = field(default_factory=list)

def keyword_relevance(query: str, content: str) -> float:
    """Simple keyword overlap relevance score."""
    query_words = set(query.lower().split())
    content_words = set(content.lower().split())
    if not query_words:
        return 0.0
    overlap = len(query_words & content_words)
    return overlap / len(query_words)

def recency_score(memory: Memory, decay_hours: float = 24.0) -> float:
    """Exponential decay based on time since last access."""
    hours_ago = (datetime.now() - memory.last_accessed).total_seconds() / 3600
    return math.exp(-hours_ago / decay_hours)

class MemoryRetriever:
    """Multi-signal memory retrieval system."""
    def __init__(self, weights: dict = None):
        self.memories: list[Memory] = []
        self.weights = weights or {
            "relevance": 0.4,
            "recency": 0.3,
            "importance": 0.2,
            "frequency": 0.1
        }
    def add(self, content: str, importance: float = 0.5, tags: list[str] = None):
        self.memories.append(Memory(
            content=content, importance=importance, tags=tags or []
        ))
    def retrieve(self, query: str, top_k: int = 5, min_score: float = 0.1) -> list[tuple[Memory, float, dict]]:
        """Retrieve memories ranked by combined score."""
        max_access = max((m.access_count for m in self.memories), default=1) or 1
        scored = []
        for memory in self.memories:
            signals = {
                "relevance": keyword_relevance(query, memory.content),
                "recency": recency_score(memory),
                "importance": memory.importance,
                "frequency": memory.access_count / max_access
            }
            combined = sum(
                self.weights[k] * signals[k] for k in self.weights
            )
            if combined >= min_score:
                scored.append((memory, combined, signals))
        scored.sort(key=lambda x: x[1], reverse=True)
        # Update access metadata for retrieved memories
        for memory, _, _ in scored[:top_k]:
            memory.last_accessed = datetime.now()
            memory.access_count += 1
        return scored[:top_k]
    def retrieve_by_tag(self, tag: str) -> list[Memory]:
        return [m for m in self.memories if tag in m.tags]


Demo: Multi-signal retrieval

retriever = MemoryRetriever()
Add memories with varying importance and timing

retriever.add("User's name is Alex", importance=0.9, tags=["identity"])
retriever.add("User prefers Python for data science", importance=0.7, tags=["preference"])
retriever.add("User asked about decorators yesterday", importance=0.3, tags=["topic"])
retriever.add("User is building a RAG pipeline", importance=0.8, tags=["project"])
retriever.add("User mentioned they have a demo on Friday", importance=0.6, tags=["schedule"])
retriever.add("User likes concise code examples", importance=0.7, tags=["preference"])
retriever.add("User struggled with async Python", importance=0.5, tags=["struggle"])
Simulate some access patterns

for m in retriever.memories[:3]:
    m.access_count = 5  # frequently accessed
Retrieve for current query

query = "help with Python async code for my RAG pipeline"
results = retriever.retrieve(query, top_k=4)
print(f"Query: '{query}'\n")
print("Retrieved memories (ranked):")
for memory, score, signals in results:
    print(f"  [{score:.3f}] {memory.content}")
    print(f"         relevance={signals['relevance']:.2f} recency={signals['recency']:.2f} "
          f"importance={signals['importance']:.2f} frequency={signals['frequency']:.2f}")

Advanced Techniques

1. Learned Retrieval Weights - Tune signal weights based on user feedback (did the retrieved memory help?) 2. Query Expansion - Use the LLM to rephrase the retrieval query for better recall 3. Negative Retrieval - Explicitly exclude memories marked as outdated or corrected 4. Contextual Re-ranking - Re-rank candidates based on the full conversation, not just the last message 5. Adaptive Top-K - Dynamically adjust how many memories to retrieve based on available context space

Quiz

Q1: Why is pure relevance-based retrieval insufficient?

  • A) It is too fast
  • B) It ignores recency, importance, and usage patterns that affect memory utility ✓
  • C) It requires too much storage
  • D) It only works with short memories

Q2: What does exponential decay accomplish in recency scoring?

  • A) It makes all memories equally important
  • B) It gradually reduces the score of older memories, preferring recent information ✓
  • C) It deletes old memories permanently
  • D) It increases the score over time

Q3: When should retrieval weights be adjusted?

  • A) Never - default weights always work
  • B) When the agent consistently retrieves unhelpful memories for certain query types ✓
  • C) Only during initial setup
  • D) Only when adding new memory types

← Back ▶ Run in Colab