Memory · Deep Dive
Memory pollution happens when an agent's memory store fills with incorrect, outdated, or redundant facts. A single offhand remark gets treated as a confirmed preference. Old information sits in the store long after it has been superseded.
This is one of the most common failure modes in long-running agents. This article covers write policies, deduplication, and TTL eviction.
Treating a single mention as confirmed fact. Outdated info never updated. Duplicate storage of the same fact. Task-specific details saved to long-term memory. Each source degrades retrieval quality and agent reliability.
A write policy gates what enters the store. Only store facts from trusted sources (explicit user confirmation, verified tool output). Reject unverified assumptions and transient task state. Check for duplicates before writing.
Every fact should have a time-to-live. When the TTL is exceeded, the fact is automatically evicted. A periodic clean_expired() sweep removes zombie facts that would otherwise degrade retrieval quality.
Concrete Example
import time
class MemoryStore:
def __init__(self, default_ttl=3600):
self.facts = {}
self.default_ttl = default_ttl
def add_fact(self, key, value, source="user_confirmed", ttl=None):
trusted = {"user_confirmed", "tool_verified", "system_defined"}
if source not in trusted:
return {"status": "rejected", "reason": "untrusted source"}
nkey = key.strip().lower()
if nkey in self.facts and self.facts[nkey]["value"].strip().lower() == value.strip().lower():
self.facts[nkey]["ttl"] = time.time() + (ttl or self.default_ttl)
return {"status": "refreshed"}
self.facts[nkey] = {"value": value, "source": source, "created": time.time(), "ttl": time.time() + (ttl or self.default_ttl)}
return {"status": "stored"}
def get_fact(self, key):
entry = self.facts.get(key.strip().lower())
if not entry:
return None
if time.time() > entry["ttl"]:
del self.facts[key.strip().lower()]
return None
return entry["value"]
def clean_expired(self):
now = time.time()
expired = [k for k, v in self.facts.items() if now > v["ttl"]]
for k in expired:
del self.facts[k]
return len(expired)MemoryStore wraps every write in a policy that rejects untrusted sources and checks for duplicates. Each fact has a TTL. get_fact() returns None for expired entries. clean_expired() removes zombie facts periodically.
Only store facts from confirmed sources like explicit user confirmations.
Check whether an equivalent fact already exists before writing.
Assign every fact a TTL and auto-evict expired entries.
Keep task state in context window; save only confirmed facts to durable memory.