Memory Compression¶
Overview¶
Memory grows over time. Memory compression is the process of reducing memory size while preserving essential information.
Key insight: Not all memories are equally important. Compress/discard low-value ones.
The Compression Challenge¶
Problem¶
Day 1: 10 interactions → 50K tokens
Day 7: 70 interactions → 350K tokens
Month 1: 300 interactions → 1.5M tokens
Year 1: 3650 interactions → 18M tokens
Issues:
- Storage costs grow
- Retrieval becomes slower
- No clear retention policy
- Old, stale information clutters memory
-
Compression Strategies¶
Strategy 1: Summarization¶
Idea: Combine similar memories into summaries
class SummarizationCompression:
def compress(self, memories, compression_ratio=0.1):
"""Compress memories via summarization"""
# Group similar memories
clusters = self.cluster_memories(memories)
# Summarize each cluster
summaries = []
for cluster in clusters:
summary = self.llm.summarize(
texts=[m['text'] for m in cluster],
max_tokens=len(cluster[0]['text']) // 10
)
summaries.append({
"summary": summary,
"count": len(cluster),
"avg_importance": np.mean([m['importance'] for m in cluster])
})
return summaries
# Usage
original_memories = retrieve_all_memories() # 1000 items
compressed = summarizer.compress(original_memories, ratio=0.1)
# Result
Strategy 2: Deletion by Age & Quality¶
Idea: Delete old, low-quality memories
class AgeQualityPruning:
def prune(self, memories, age_threshold_days=30, quality_threshold=0.3):
"""Delete old or low-quality memories"""
keep = []
delete = []
now = datetime.now()
for memory in memories:
age_days = (now - memory['timestamp']).days
quality = memory.get('quality', 0.5)
# Keep if recent OR high quality
if age_days < age_threshold_days or quality > quality_threshold:
keep.append(memory)
else:
delete.append(memory)
# Delete low-value memories
for memory in delete:
self.memory_db.delete(memory['id'])
return keep
# Usage
pruned = pruner.prune(all_memories)
# Automatically removes stale, low-quality items
Strategy 3: Hierarchical Compression¶
Idea: Different compression levels for different ages
Recent (< 1 day): No compression (keep everything)
│
Recent (1-7 days): Keep important items, discard noise
│
Historic (1-4 weeks): Summarize clusters
│
Old (> 4 weeks): Only keep high-quality or frequently accessed
Implementation:
class HierarchicalCompression:
def compress(self, memories):
"""Apply different compression by age"""
now = datetime.now()
compressed = []
for memory in sorted(memories, key=lambda x: x['timestamp']):
age_days = (now - memory['timestamp']).days
if age_days < 1:
# Keep everything recent
compressed.append(memory)
elif age_days < 7:
# Keep important items
if memory['importance'] > 0.7:
compressed.append(memory)
elif age_days < 30:
# Summarize or delete
if memory['access_count'] > 5:
compressed.append(memory) # Keep frequently used
else:
summary = self.create_summary(memory)
compressed.append(summary)
else:
# Old memories: keep only high-value
if memory['importance'] > 0.9 or memory['access_count'] > 10:
compressed.append(memory)
return compressed
Compression Techniques¶
Technique 1: Embedding-Based Deduplication¶
class DeduplicationCompressor:
def deduplicate(self, memories, similarity_threshold=0.95):
"""Remove near-duplicate memories"""
# Get embeddings
embeddings = [embed(m['text']) for m in memories]
# Find duplicates
duplicates = []
for i, emb1 in enumerate(embeddings):
for j, emb2 in enumerate(embeddings[i+1:], start=i+1):
similarity = cosine_similarity(emb1, emb2)
if similarity > similarity_threshold:
# Keep the more important one
if memories[i]['importance'] > memories[j]['importance']:
duplicates.append(j)
else:
duplicates.append(i)
# Remove duplicates
unique_memories = [
m for i, m in enumerate(memories)
if i not in duplicates
]
return unique_memories
Technique 2: Conceptual Compression¶
class ConceptualCompression:
def compress(self, memories):
"""Abstract specific memories into concepts"""
# Extract concepts from memories
concepts = self.extract_concepts(memories)
# Map memories to concepts
compressed = []
for concept in concepts:
compressed.append({
"type": "concept",
"concept": concept['name'],
"instances_count": concept['count'],
"evidence": concept['examples'][:3] # Keep a few examples
})
return compressed
# Usage
memories = [
"Analyzed Q1 sales",
"Analyzed Q2 sales",
"Analyzed Q3 sales",
]
compressed = conceptual_compressor.compress(memories)
# Result
-
Implementing Compression Pipeline¶
Complete Example¶
class MemoryCompressionPipeline:
def __init__(self):
self.logger = setup_logging()
def compress_memories(self, memories):
"""Apply full compression pipeline"""
self.logger.info(f"Starting compression on {len(memories)} items")
# Step 1: Remove exact duplicates
step1 = self.remove_exact_duplicates(memories)
self.logger.info(f"After dedupe: {len(step1)} items")
# Step 2: Remove near-duplicates by embedding
step2 = self.remove_near_duplicates(step1, threshold=0.95)
self.logger.info(f"After similarity: {len(step2)} items")
# Step 3: Apply hierarchical compression by age
step3 = self.hierarchical_compress(step2)
self.logger.info(f"After hierarchy: {len(step3)} items")
# Step 4: Summarize remaining
step4 = self.summarize_clusters(step3)
self.logger.info(f"After summarization: {len(step4)} items")
# Report compression
ratio = len(step4) / len(memories)
self.logger.info(f"Compression ratio: {ratio:.1%}")
return step4
def remove_exact_duplicates(self, memories):
"""Remove exact string matches"""
seen = set()
unique = []
for memory in memories:
text_hash = hash(memory['text'])
if text_hash not in seen:
seen.add(text_hash)
unique.append(memory)
return unique
def remove_near_duplicates(self, memories, threshold):
"""Remove similar items (by embedding distance)"""
#... (see Technique 1 above)
pass
def hierarchical_compress(self, memories):
"""Apply age-based compression"""
#... (see Hierarchical Compression above)
pass
def summarize_clusters(self, memories):
"""Group and summarize similar memories"""
#... (see Summarization above)
pass
# Usage
pipeline = MemoryCompressionPipeline()
compressed = pipeline.compress_memories(all_memories)
Scheduled Compression¶
import schedule
import time
def compress_and_archive():
"""Compress memories every night"""
# Get old memories
old_memories = memory_db.get_older_than(days=30)
# Compress
compressed = compressor.compress(old_memories)
# Archive originals
archive_db.store(old_memories)
# Replace with compressed
for memory in old_memories:
memory_db.delete(memory['id'])
for compressed_item in compressed:
memory_db.add(compressed_item)
print(f"Compression complete: {len(old_memories)} → {len(compressed)}")
# Schedule
schedule.every().day.at("02:00").do(compress_and_archive)
while True:
schedule.run_pending()
time.sleep(60)
-
Compression Quality Metrics¶
Metric 1: Compression Ratio¶
Original size: 1000 memories
Compressed size: 100 memories
Ratio: 10% (90% reduction)
Metric 2: Information Retention¶
Ask: "What reports have we analyzed?"
Before compression: 8/10 results relevant
After compression: 7/10 results relevant
Retention: 87.5%
Metric 3: Query Performance¶
Search time before: 500ms
Search time after: 50ms
Speedup: 10x
When to Compress¶
Automatic Triggers¶
def should_compress(memory_system):
# Compress if:
# 1. Memory size exceeds threshold
if memory_system.size_gb() > 5:
return True
# 2. Search latency degrades
if memory_system.avg_search_time_ms() > 1000:
return True
# 3. Scheduled time (e.g., nightly)
if datetime.now().hour == 2:
return True
return False
Best Practices¶
1. Preserve Important Memories¶
# Good
memory['importance'] = 0.9 # Mark as important
# Won't be deleted during compression
# Bad
compress_everything() # May lose critical info
2. Keep Access Patterns¶
# Good
memory['access_count'] = 15 # Frequently used
# Frequently accessed items kept even if old
# Bad
delete_if_old() # May delete useful memories
3. Version Control¶
# Good
archive_db.store(original_memories)
compressed = compress(original_memories)
# Can rollback if needed
# Bad
compress_and_discard() # No recovery possible
4. Test After Compression¶
# Good
original_results = retrieve(query)
memory = switch_to_compressed()
compressed_results = retrieve(query)
if similarity(original_results, compressed_results) < 0.9:
raise CompressionError("Too much quality loss")
# Bad
compress_and_deploy() # May have broken retrieval
-
Advanced: Learning-Based Compression¶
class LearningCompressor:
def __init__(self):
self.model = train_importance_model()
def predict_importance(self, memory):
"""Use ML to predict if memory is important"""
features = self.extract_features(memory)
# Features: age, access_count, quality, etc.
importance_score = self.model.predict(features)
return importance_score
def compress(self, memories):
"""Compress based on learned importance"""
scores = [self.predict_importance(m) for m in memories]
# Keep high-scoring memories
important = [m for m, s in zip(memories, scores) if s > 0.7]
# Compress/delete low-scoring
less_important = [m for m, s in zip(memories, scores) if s <= 0.7]
compressed = self.summarize(less_important)
return important + compressed
Key Takeaways¶
- Compression necessary - Memory grows indefinitely
- Multiple strategies - Summarization, deletion, deduplication
- Preserve important - Mark and protect high-value memories
- Monitor quality - Don't over-compress
- Automate - Schedule compression pipeline
- Test carefully - Verify retrieval quality after compression
-
Next Steps¶
- Read Episodic & Procedural - Different memory types
- Go To Planning & Reasoning - Next chapter
-
Last Updated: August 9, 2026