Pattern Memory

Memory-Augmented Generation

🎯 Trigger Conditions

Use when asked about memory-augmented generation, retrieval-augmented generation, or persistent context for DSPy.

📚 Prerequisites

• dspy package installed
• Memory store available
• Retrieval mechanism configured

🛠️ Memory Patterns

1. Short-Term Memory

class ShortTermMemory(dspy.Module):
    def __init__(self, max_entries=10):
        self.max_entries = max_entries
        self.memory = []
    
    def add(self, entry):
        self.memory.append(entry)
        if len(self.memory) > self.max_entries:
            self.memory.pop(0)
    
    def retrieve(self, query):
        # Retrieve relevant memories
        relevant = [m for m in self.memory if similar(m, query)]
        return relevant[:3]
    
    def forward(self, query):
        memories = self.retrieve(query)
        return self.generator(query=query, memories=memories)

2. Long-Term Memory

import chromadb
import dspy

class LongTermMemory(dspy.Module):
    def __init__(self, collection_name="memories"):
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection(collection_name)
        self.retriever = dspy.Retrieve(k=5)
    
    def store(self, content, metadata):
        self.collection.add(
            documents=[content],
            metadatas=[metadata],
            ids=[str(uuid4())]
        )
    
    def retrieve(self, query):
        results = self.collection.query(
            query_texts=[query],
            n_results=5
        )
        return results["documents"][0]
    
    def forward(self, query):
        memories = self.retrieve(query)
        return self.generator(query=query, memories=memories)

3. Episodic Memory

class EpisodicMemory(dspy.Module):
    def __init__(self):
        self.episodes = []
        self.summarizer = dspy.Predict("episode -> summary")
    
    def add_episode(self, episode):
        self.episodes.append(episode)
        summary = self.summarizer(episode=episode).summary
        self.store_summary(summary, episode)
    
    def retrieve_episodes(self, query):
        # Retrieve relevant episodes
        relevant = [e for e in self.episodes if similar(e, query)]
        return relevant
    
    def forward(self, query):
        episodes = self.retrieve_episodes(query)
        return self.generator(query=query, episodes=episodes)

4. Working Memory

class WorkingMemory(dspy.Module):
    def __init__(self):
        self.context = {}
        self.update_predictor = dspy.Predict("context, update -> new_context")
    
    def update(self, key, value):
        self.context[key] = value
    
    def retrieve(self, key):
        return self.context.get(key)
    
    def forward(self, query):
        # Retrieve working memory
        relevant_context = self.get_relevant_context(query)
        
        # Generate with context
        return self.generator(query=query, context=relevant_context)
    
    def get_relevant_context(self, query):
        # Filter context based on query
        return {k: v for k, v in self.context.items() if relevant(k, query)}

⚠️ Pitfalls

• Memory bloat: Unbounded memory consumption
• Relevance: Retrieving irrelevant memories
• Consistency: Maintaining consistent memory state
• Privacy: Sensitive data in memory

📖 References

• Memory-Augmented NLP
• DSPy Memory