How to Implement Custom Memory Retention and Cleanup Policies in Mem0

Mem0 stores every memory in both a vector store and a SQLite history database, exposing delete(), delete_all(), and reset() methods in mem0/memory/main.py that you can orchestrate into time-based or metadata-driven retention policies.

Mem0 is an open-source memory layer for AI applications that persists data across two distinct storage layers. To implement custom memory retention and cleanup policies in Mem0, you leverage the timestamp metadata automatically recorded in the SQLite history layer and the explicit deletion APIs exposed in the core memory client.

Understanding Mem0's Dual Storage Architecture

Mem0 persists every memory operation across two synchronized layers:

1. Vector store – The searchable embedding collection (Faiss, Qdrant, Pinecone, etc.) that powers semantic retrieval.
2. SQLite history database – A lightweight local database defined in mem0/memory/storage.py that records every add, update, and delete action with created_at and updated_at timestamps.

This architecture means retention policies can query the SQLite layer for temporal metadata, then target specific records for deletion in the vector store using the core client APIs.

Core Deletion APIs Available

The Memory class in mem0/memory/main.py exposes three primary methods for removal:

Method	Purpose	Location
Memory.delete(memory_id)	Remove a single memory by its unique identifier	mem0/memory/main.py (lines 1450-1460)
Memory.delete_all(user_id=..., agent_id=..., run_id=...)	Batch-delete memories matching session identifiers	mem0/memory/main.py (lines 1025-1055)
Memory.reset()	Destroy all data—drops the vector collection and truncates the history table	mem0/memory/main.py (lines 1212-1235)

When you call delete(), the operation also writes a DELETE entry to the SQLite history table via self.db.add_history(..., "DELETE", ...) (lines 1000-1009), creating a full audit trail for compliance or debugging.

Building a Custom Retention Policy

Because every memory record carries ISO-8601 timestamps in its metadata, you can implement TTL (time-to-live) or age-based cleanup in three steps:

Step 1: Fetch Candidate Memories

Use Memory.get_all() with optional session filters to retrieve memories subject to your policy. The method internally calls _build_filters_and_metadata (lines 81-106) to inject user_id, agent_id, or run_id scoping.

Step 2: Filter by Age or Metadata

Compare the stored created_at timestamp against your retention threshold. The timestamps are generated in _create_memory (lines 1080-1086) using datetime.now(pytz.timezone("US/Pacific")).isoformat() and stored as strings in the memory metadata.

Step 3: Execute Targeted Deletion

Iterate through expired memory IDs and call Memory.delete(memory_id) for each stale entry. For batch operations where an entire session should be purged, use delete_all() with the appropriate session identifiers instead.

Complete Implementation Example

The following example implements a 90-day retention policy for a specific user, filtering memories by age and removing expired entries:

from datetime import datetime, timedelta
from mem0 import Memory, MemoryConfig

# ----------------------------------------------------------------------

# 1️⃣  Initialize Mem0 client

# ----------------------------------------------------------------------

mem = Memory(config=MemoryConfig())   # Uses defaults; customize DB path if needed

# ----------------------------------------------------------------------

# 2️⃣  Define retention helper

# ----------------------------------------------------------------------

def purge_expired_memories(
    *, 
    user_id: str | None = None,
    agent_id: str | None = None,
    run_id: str | None = None,
    max_age_days: int = 30,
) -> list[str]:
    """Delete memories older than `max_age_days` for the supplied session IDs."""
    
    # 2a – Fetch all memories for the given session(s)

    results = mem.get_all(
        user_id=user_id,
        agent_id=agent_id,
        run_id=run_id,
        limit=10_000,                     # Raise as needed for your corpus size

    )["results"]

    # 2b – Compute cutoff datetime (UTC)

    cutoff = datetime.utcnow() - timedelta(days=max_age_days)

    # 2c – Collect IDs that exceed the cutoff

    stale_ids = [
        m["id"]
        for m in results
        if datetime.fromisoformat(m["created_at"]) < cutoff
    ]

    # 2d – Delete each stale memory (parallelize with asyncio if scaling)

    for mem_id in stale_ids:
        mem.delete(mem_id)

    return stale_ids

# ----------------------------------------------------------------------

# 3️⃣  Run the policy (e.g., as a daily background task)

# ----------------------------------------------------------------------

deleted = purge_expired_memories(user_id="user-123", max_age_days=90)
print(f"Purged {len(deleted)} old memories.")

Advanced Retention Scenarios

Scenario	Implementation Strategy
Per-agent retention – Different agents keep memories for different durations	Pass the specific agent_id to purge_expired_memories() and vary max_age_days per invocation based on agent configuration.
Periodic cleanup – Run nightly without blocking the main application	Wrap the helper in an async background task using asyncio.create_task() or schedule it with APScheduler to run independently of your request handlers.
Soft-delete and archival – Maintain records after vector removal	The SQLite history table already marks deletions with is_deleted = 1. To archive vector payloads externally, retrieve them first via mem.get(memory_id) before calling delete().
Metadata-driven filtering – Delete only memories tagged as "temporary"	Add custom metadata (e.g., metadata={"tags": ["temporary"]}) when calling mem.add(). Then use mem.get_all(filters={"tags": {"contains": "temporary"}})—the filter syntax is processed by _process_metadata_filters (lines 858-889) in the main memory module.

Key Implementation Details

Timestamp Handling: According to the Mem0 source code, timestamps are generated using datetime.now(pytz.timezone("US/Pacific")).isoformat() during the _create_memory workflow and stored as ISO-8601 strings in the memory metadata dictionary.

History Auditing: Every deletion operation persists to the SQLite history table via the add_history method, ensuring you maintain a complete log of when memories were removed and by which session identifiers.

Filter Construction: The _build_filters_and_metadata helper (lines 81-106) automatically injects session-scoping parameters into vector store queries, while _process_metadata_filters (lines 858-889) handles complex metadata queries for custom retention logic beyond simple age checks.

Summary

• Mem0 maintains synchronized storage in a vector store and a SQLite history database (mem0/memory/storage.py), with every memory carrying created_at timestamps.
• Use Memory.get_all() to fetch candidates, filter by datetime comparisons, and remove stale entries with Memory.delete() or Memory.delete_all().
• Deletion operations are automatically audited in the SQLite history table, providing compliance trails without additional code.
• Implement periodic cleanup by wrapping retention logic in background tasks or schedulers, and extend policies using metadata filters processed by _process_metadata_filters.

Frequently Asked Questions

How does Mem0 track when a memory was created?

Mem0 automatically records created_at and updated_at timestamps in the memory metadata during the _create_memory workflow in mem0/memory/main.py (lines 1080-1086). These timestamps are ISO-8601 formatted strings generated using datetime.now(pytz.timezone("US/Pacific")).isoformat(), stored in both the vector store metadata and the SQLite history table for audit purposes.

Can I delete all memories for a specific user without iterating through individual IDs?

Yes. Use the Memory.delete_all() method in mem0/memory/main.py (lines 1025-1055) and pass the specific user_id, agent_id, or run_id session identifiers. This performs a batch deletion across both the vector store and the history table without requiring you to fetch and iterate over individual memory IDs.

Does Mem0 support soft-deletion for compliance requirements?

Yes. The SQLite history layer in mem0/memory/storage.py automatically records a DELETE action entry via add_history when you call delete(), marking records with is_deleted = 1. This creates a soft-delete trail while removing the vector payload. If you need to archive vector data before deletion, retrieve it using mem.get(memory_id) and store it externally before calling delete().

How do I implement retention policies based on custom metadata tags rather than age?

When adding memories, include custom metadata such as metadata={"retention_tier": "short_lived"}. Then use Memory.get_all() with the filters parameter (e.g., filters={"retention_tier": {"contains": "short_lived"}}). The _process_metadata_filters helper in mem0/memory/main.py (lines 858-889) translates these filters into vector store queries, allowing you to target specific metadata categories for cleanup.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}