DEV Community

Mukesh
Mukesh

Posted on

Building a Self-Pruning Memory Layer for AI Agents: A Hands-On Mem0 Tutorial

Most "give your agent memory" tutorials stop at memory.add() and memory.search(). That's the easy 20%. The part nobody writes about is what happens six weeks later, when your agent has accumulated 40,000 memories, half of them stale, and every search() call is now pulling in noise instead of signal.

This tutorial builds the missing piece: a self-pruning memory layer on top of Mem0 that decides what to keep, what to decay, and what to delete — using a combination of TTL, importance scoring, and access-frequency decay. By the end you'll have a working multi-agent support-desk example where two agents share a memory store but retain information on different schedules.

Setup

pip install mem0ai python-dotenv
Enter fullscreen mode Exit fullscreen mode
# .env
MEM0_API_KEY=your_key_here
OPENAI_API_KEY=your_key_here
Enter fullscreen mode Exit fullscreen mode
# memory_layer.py
import os
import time
from datetime import datetime, timedelta
from dotenv import load_dotenv
from mem0 import Memory

load_dotenv()

m = Memory()
Enter fullscreen mode Exit fullscreen mode

Step 1: Store memories with eviction metadata

Mem0 lets you attach arbitrary metadata to every memory. That metadata is where your eviction policy lives — Mem0 itself won't decide relevance for you, so you encode the signals yourself at write time.

def remember(user_id, text, category="general", importance=0.5, ttl_days=None):
    expires_at = None
    if ttl_days is not None:
        expires_at = (datetime.utcnow() + timedelta(days=ttl_days)).isoformat()

    m.add(
        text,
        user_id=user_id,
        metadata={
            "category": category,
            "importance": importance,
            "created_at": datetime.utcnow().isoformat(),
            "last_accessed": datetime.utcnow().isoformat(),
            "access_count": 0,
            "expires_at": expires_at,
        },
    )
Enter fullscreen mode Exit fullscreen mode

Three category examples for a support-desk agent:

# Ephemeral: this ticket's context, gone in 3 days
remember("agent_alice", "Customer reports checkout button greyed out on Safari",
         category="session", importance=0.3, ttl_days=3)

# Durable: account-level facts that should survive indefinitely
remember("agent_alice", "Customer is on the Enterprise plan, SSO enabled",
         category="account", importance=0.9)

# Medium-lived: a resolved bug pattern worth remembering for a while
remember("agent_alice", "Safari 17.4 checkout bug traced to a CSP header issue, fixed in v2.3.1",
         category="knowledge", importance=0.7, ttl_days=60)
Enter fullscreen mode Exit fullscreen mode

Step 2: Track access on every read

An eviction policy that only looks at created_at will delete memories the agent still actively relies on. You need to know what's actually being used, not just what's old.

def recall(user_id, query, limit=5):
    results = m.search(query, user_id=user_id, limit=limit)

    for r in results.get("results", results):
        mem_id = r["id"]
        meta = r.get("metadata", {})
        meta["last_accessed"] = datetime.utcnow().isoformat()
        meta["access_count"] = meta.get("access_count", 0) + 1
        m.update(mem_id, metadata=meta)

    return results
Enter fullscreen mode Exit fullscreen mode

This is the detail that's easy to skip and expensive to skip: without it, your "importance" score is frozen at write time forever, and a memory that turns out to be load-bearing gets pruned right alongside genuine noise.

Step 3: The pruning pass

Run this on a schedule (cron, a Celery beat task, whatever your agent's runtime already uses). It computes a decay score per memory and deletes anything below threshold — but never touches anything still inside its TTL.

def prune(user_id, min_score=0.25, dry_run=False):
    all_memories = m.get_all(user_id=user_id)
    now = datetime.utcnow()
    deleted = []

    for entry in all_memories.get("results", all_memories):
        meta = entry.get("metadata", {}) or {}

        expires_at = meta.get("expires_at")
        if expires_at and datetime.fromisoformat(expires_at) < now:
            score = -1  # force delete, TTL is authoritative
        else:
            score = decay_score(meta, now)

        if score < min_score:
            deleted.append((entry["id"], entry.get("memory", "")[:60], score))
            if not dry_run:
                m.delete(entry["id"])

    return deleted


def decay_score(meta, now):
    importance = meta.get("importance", 0.5)
    access_count = meta.get("access_count", 0)

    last_accessed_str = meta.get("last_accessed") or meta.get("created_at")
    last_accessed = datetime.fromisoformat(last_accessed_str)
    days_idle = (now - last_accessed).total_seconds() / 86400

    # Exponential decay on idle time, boosted by real usage and set importance.
    idle_decay = 0.5 ** (days_idle / 14)  # half-life of 14 idle days
    usage_boost = min(access_count * 0.05, 0.3)

    return min(importance * idle_decay + usage_boost, 1.0)
Enter fullscreen mode Exit fullscreen mode

Run it:

results = prune("agent_alice", min_score=0.25, dry_run=True)
for mem_id, snippet, score in results:
    print(f"[{score:.2f}] would delete: {snippet}")
Enter fullscreen mode Exit fullscreen mode

The dry_run flag matters more than it looks — the first time you run a pruning pass against a real agent's memory store, look at what it would delete before you let it delete anything. Tune min_score and the half-life constant against that output, not against a guess.

Step 4: Sharing memory across agents with different retention needs

The multi-agent case is where naive TTL falls apart: two agents reading the same memory store often need to keep different things. A billing agent needs account-level facts (category="account") to persist far longer than a triage agent's session notes.

def prune_by_category_policy(user_id, policies, dry_run=False):
    """policies: {"session": 0.4, "knowledge": 0.2, "account": 0.05}"""
    all_memories = m.get_all(user_id=user_id)
    now = datetime.utcnow()
    deleted = []

    for entry in all_memories.get("results", all_memories):
        meta = entry.get("metadata", {}) or {}
        category = meta.get("category", "general")
        threshold = policies.get(category, 0.25)
        score = decay_score(meta, now)

        if score < threshold:
            deleted.append(entry["id"])
            if not dry_run:
                m.delete(entry["id"])

    return deleted


# Session notes get pruned aggressively, account facts almost never
prune_by_category_policy(
    "agent_alice",
    policies={"session": 0.4, "knowledge": 0.2, "account": 0.05},
)
Enter fullscreen mode Exit fullscreen mode

Both agent_alice (triage) and a second agent, agent_bob (billing), read from the same user_id namespace but apply this policy independently. The billing agent never has to worry that a triage-related pruning pass wiped an SSO configuration note — the category threshold protects it structurally, not by convention.

What this buys you

Without this layer, a long-running Mem0-backed agent accumulates memories monotonically — search() gets slower and noisier, and your embedding/storage costs grow forever. With it, memory volume plateaus: new memories keep entering, but idle, low-importance, low-access ones exit at roughly the same rate. That's the actual shape a production memory layer needs — not "remember everything," but "remember what earns its keep, on a schedule you control instead of one Mem0 (or any provider) chooses for you."

The full pattern generalizes past Mem0: TTL for known-ephemeral data, importance for known-durable data, and a decay function for the huge middle category you can't classify in advance. If you're building an agent that runs for months instead of a single session, that middle category is where most of your memory budget actually goes.

Top comments (1)

Collapse
 
ahmetozel profile image
Ahmet Özel

Good to see eviction treated as the design problem rather than an afterthought. One distinction worth building into the metadata early: whether a memory is an episode or a claim. An episode is true about a moment and can keep its timestamp forever - the user asked about billing on 3 March. A claim asserts something is true now - the user prefers email over Slack - and the failure mode is that a throwaway remark from testing outranks the current reality for months because nothing ever supersedes it. Access-frequency decay does not help there, since a wrong memory that keeps getting retrieved looks popular. What has worked better for me is letting a newer claim on the same subject invalidate the older one rather than competing with it, and keeping a supersedes pointer so you can still explain why the agent believed something last month. Reproducing a bad answer means reproducing the memory state, and that is much easier when the history is a chain rather than a soup.