Skip to content

Memory Deduplication System

Date: January 8, 2026 Status: Implemented Inspired By: Tolan Voice Agent Architecture

Overview

The Memory Deduplication System is a nightly background job that removes redundant memory entries to maintain high memory quality. This is inspired by Tolan's approach of running "a nightly compression job that removes low-value or redundant entries."

Problem Solved

Over time, similar memories accumulate: - "User mentioned wife Sarah likes hiking" - "User's wife Sarah enjoys hiking outdoors" - "Sarah (user's wife) is into hiking"

These duplicates waste storage, slow searches, and can cause inconsistent recall. The deduplication job consolidates them into a single, authoritative memory.

Architecture

Nightly Job (3am)
    │
    ▼
┌─────────────────────────────────────┐
│     Get Active Users (last 7 days)  │
│     SQL query on conversations      │
└─────────────────────────────────────┘
    │
    ▼
For each user:
┌─────────────────────────────────────┐
│     Fetch All Memories              │
│     Multiple broad queries          │
└─────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────┐
│     Find Duplicate Groups           │
│     Similarity score > 0.92         │
└─────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────┐
│     Merge Groups                    │
│     Keep most recent, delete rest   │
└─────────────────────────────────────┘

Components

1. Memory Maintenance Service (app/memory/memory_maintenance.py)

Core deduplication logic.

from app.memory.memory_maintenance import get_maintenance_service

maintenance = get_maintenance_service()
stats = await maintenance.run_deduplication(
    user_id="+1234567890",
    persona_id="sage",
    dry_run=False  # Set True to preview without deleting
)

# Returns:
# {
#     "deduplicated": 5,
#     "total_checked": 42,
#     "duplicate_groups_found": 2,
#     "kept_memories": [...],
#     "deleted_memories": [...]
# }

2. Scheduled Job (app/scheduler/background_service.py)

Runs nightly at 3am.

# In SchedulerService.start():
self.scheduler.add_job(
    self.run_memory_deduplication,
    CronTrigger(hour=3, minute=0),
    id='memory_deduplication',
    name='Nightly Memory Deduplication'
)

Key Features: - Distributed lock prevents duplicate execution across workers - Lock TTL: 20 hours (job runs once daily) - Processes users active in last 7 days only - Graceful error handling per user

Deduplication Algorithm

Step 1: Fetch All Memories

Since Supermemory doesn't have a "get all" endpoint, we use multiple broad queries: - Empty query (returns recent) - "user", "mentioned", "likes", "wants", "trip", "family", "work"

Results are deduplicated by memory ID during collection.

Step 2: Find Duplicate Groups

For each memory: 1. Search Supermemory using the memory's text as query 2. Find results with similarity score >= 0.92 3. Group together as duplicates

Step 3: Merge Groups

For each duplicate group: 1. Sort by timestamp (most recent first) 2. Keep the most recent memory 3. Delete all others

Configuration

Setting Value Location
Schedule 3:00 AM daily background_service.py:125
Similarity threshold 0.92 memory_maintenance.py:44
User activity window 7 days background_service.py:470
Lock TTL 20 hours background_service.py:412

Monitoring

Logs

[MemDedup] +1234567890: removed 3 duplicates
Memory deduplication complete: 15 duplicates removed from 8 users

Metrics (in stats dict)

  • deduplicated: Number of memories deleted
  • total_checked: Total memories analyzed
  • duplicate_groups_found: Number of duplicate clusters found
  • kept_memories: List of memories that were kept
  • deleted_memories: List of memories that were removed

Dry Run Mode

Test deduplication without actually deleting:

stats = await maintenance.run_deduplication(
    user_id="+1234567890",
    persona_id="sage",
    dry_run=True  # Preview only
)
print(f"Would delete: {stats['deduplicated']} memories")

Safety Features

  1. Distributed Lock - Only one worker runs the job
  2. Per-User Error Isolation - One user's failure doesn't stop others
  3. Keep Most Recent - Always preserves the newest memory
  4. Logging - Full audit trail of kept/deleted memories
  5. Dry Run - Test without side effects

Manual Execution

For debugging or immediate cleanup:

from app.memory.memory_maintenance import run_maintenance_for_user

# Run for a specific user
stats = await run_maintenance_for_user(
    user_id="+1234567890",
    persona_id="sage",
    dry_run=True
)

Future Enhancements

  1. Low-value cleanup - Remove trivial memories ("user said hi")
  2. Contradiction resolution - Handle conflicting facts ("lives in NYC" vs "lives in LA")
  3. Memory compression - Merge related facts into single memories
  4. Configurable thresholds - Per-user or per-memory-type settings
  5. Metrics dashboard - Track deduplication effectiveness over time