One of the most requested features in AI chat is “remember me.” Users want their AI to know their preferences, their projects, their context – without re-explaining everything each conversation. But building memory that’s useful without being creepy, fast without being shallow, and private when it needs to be? That’s the hard part.
Here’s how we did it.
Starting Point: The Adaptive AI Memory Filter
We started with the great work of @alexgrama7 and Adaptive Memory an open webui filter that hooks into the request/response lifecycle. The filter implements two core methods:
async def inlet(
self,
body: Dict[str, Any],
__event_emitter__: Optional[Callable[[Any], Awaitable[None]]] = None,
__user__: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Pre-process: retrieve and inject relevant memories before LLM sees the message"""
async def outlet(
self,
body: dict,
__event_emitter__: Optional[Callable[[Any], Awaitable[None]]] = None,
__user__: Optional[dict] = None,
) -> dict:
"""Post-process: extract new memories from the conversation"""The flow:
User Message → inlet() → Retrieve Memories → Inject Context → LLM → outlet() → Extract & StoreFrom there we wen and updated various parts to work within our collaborative context.
Memory Extraction
The extraction happens via a structured LLM call (GPT-4o-mini via OpenRouter). We send the conversation context and get back structured JSON:
memory_schema = {
"type": "object",
"properties": {
"operations": {
"type": "array",
"items": {
"type": "object",
"properties": {
"operation": {"type": "string", "enum": ["NEW", "UPDATE", "DELETE"]},
"content": {"type": "string"},
"tags": {"type": "array", "items": {"type": "string"}},
"memory_bank": {"type": "string", "enum": ["Personal", "Interaction Patterns", "Project Memory"]},
"id": {"type": "string", "nullable": True}
}
}
}
}
}Sample Categories we extract:
- identity: Name, role, background
- preference: Likes coffee, prefers bullet points
- goal: Learning Python, building a SaaS
- relationship: Has a sister named Jane, works with Bob
- behavior: Works late, likes detailed explanations
- possession: Has a MacBook Pro, uses VS Code
Storage
Memories are stored in two places:
- PostgreSQL – The source of truth with metadata
class Memory(Base):
__tablename__ = "memory"
id = Column(String, primary_key=True) # UUID
user_id = Column(String) # Owner
folder_id = Column(String, nullable=True) # Project scope
content = Column(Text) # "[Tags: preference] User likes coffee"
updated_at = Column(BigInteger) # Epoch timestamp
created_at = Column(BigInteger)- Vector DB (pgvector) – For similarity search
# Collection naming enforces isolation
user_collection = f"user-memory-{user_id}"
folder_collection = f"folder-memory-{folder_id}"
# Embedding model: all-MiniLM-L6-v2 (384 dimensions)
VECTOR_DB_CLIENT.insert(
collection_name=user_collection,
items=[{"id": memory_id, "vector": embedding, "metadata": {...}}]
)Problem 1: LLM Call Latency
Vector search is fast (~10ms). LLM relevance scoring is not (~300-500ms).
Our v1 hit the LLM for every memory retrieval to score relevance. That’s unacceptable for a chat interface.
Solution: Tiered Relevance with Smart Skipping
async def get_relevant_memories(self, message: str, user_id: str) -> List[Dict]:
# Step 1: Fast vector search
query_embedding = await self._embed(message)
candidates = VECTOR_DB_CLIENT.search(
collection_name=f"user-memory-{user_id}",
vectors=[query_embedding],
limit=self.valves.reranking_top_k # default: 50
)
# Step 2: Convert distances to similarity scores
scored = []
for candidate in candidates:
similarity = 1.0 - (candidate.distance / 2.0) # Cosine distance → similarity
# HIGH CONFIDENCE: Skip LLM entirely
if similarity > self.valves.llm_skip_relevance_threshold: # default: 0.93
scored.append((candidate, similarity))
continue
# MEDIUM CONFIDENCE: Optionally use LLM
if self.valves.use_llm_for_relevance and similarity > self.valves.vector_similarity_threshold:
llm_score = await self._llm_relevance_score(message, candidate.content)
scored.append((candidate, llm_score))
elif similarity > self.valves.vector_similarity_threshold: # default: 0.7
scored.append((candidate, similarity))
# Step 3: Apply time decay
return self._apply_time_decay(scored)Result: 70% latency reduction for most messages while preserving accuracy where it matters.
Time Decay Scoring
Recent memories get boosted, old ones dampened:
def _apply_time_decay_score(self, semantic_score: float, memory: Dict) -> float:
age_days = (time.time() - memory["updated_at"]) / 86400
decay_factor = math.exp(-age_days / self.valves.time_decay_half_life_days) # default: 60
# Blend: 80% semantic, 20% recency
weight = self.valves.recency_weight # default: 0.2
return (1 - weight) * semantic_score + weight * decay_factorProblem 2: AI Memory Bloat
Active users generate 500+ memories within a month. Issues:
- Retrieval noise dilutes signal
- Can’t inject 500 memories into context
Solution: Background Summarization
We took the inspiration from Claude where memory is always just a summary that gets injected into the system prompt. We created a task that runs every 24 hours:
async def _summarize_old_memories_loop(self):
while True:
await asyncio.sleep(self.valves.summarization_interval) # 86400 sec
for user_id in self._get_users_above_threshold():
# 1. Find clusters of related memories
clusters = await self._find_memory_clusters(
memories=self._get_old_memories(user_id, min_age_days=7),
strategy=self.valves.summarization_strategy # "embeddings", "tags", or "hybrid"
)
# 2. Summarize each cluster
for cluster in clusters:
if len(cluster) >= self.valves.summarization_min_cluster_size: # default: 3
summary = await self._summarize_cluster(cluster)
# 3. Replace originals with summary
await self._store_memory(summary)
for mem in cluster:
await self._delete_memory(mem.id)Clustering strategies:
- embeddings: Greedy clustering by cosine similarity (most accurate)
- tags: Group by extracted tags (fast)
- hybrid: Combine both
Limit-Triggered Summarization
When approaching the limit (default: 200 memories), auto-trigger:
current_count = await self._get_memory_count(user_id)
threshold = self.valves.max_total_memories * self.valves.summarization_trigger_threshold # 0.8
if current_count >= threshold:
await self._trigger_emergency_summarization(
user_id,
target_reduction=self.valves.summarization_target_reduction # 0.2 = free 40 slots
)Problem 3: Group Chat Privacy
The hard question: What happens when Alice, Bob, and Carol share a group chat?
If Alice mentions “I prefer bullet points,” should it be stored? To whom? Should Bob see Alice’s memories?
Solution: Privacy-First + Memory Banks
Rule 1: Personal memories are NEVER extracted from collaborative chats.
async def outlet(self, body: dict, __user__: dict) -> dict:
# Check collaborative status
is_collaborative = await self._is_collaborative_chat(body)
if is_collaborative:
logger.info("🤝 Collaborative chat detected - skipping memory extraction for privacy")
# Still process but filter allowed banks
async def _is_collaborative_chat(self, body: Dict) -> bool:
chat_id = body.get("metadata", {}).get("chat_id")
if not chat_id or chat_id.startswith("local:"):
return False
chat = Chats.get_chat_by_id(chat_id)
return chat.is_collaborative if chat else FalseRule 2: Memory Banks define access levels.
| Bank | Scope | Collaborative Access | Use Case |
|---|---|---|---|
| Personal | user_id only | NEVER injected | “User is named Alice” |
| Interaction Patterns | user_id | Allowed | “Prefers concise responses” |
| Project Memory | folder_id | All folder members | “Project uses React + TypeScript” |
def get_collaborative_allowed_banks(self) -> Set[str]:
"""Banks that can be injected in collaborative contexts"""
return {"Interaction Patterns", "Project Memory"}
# In memory extraction:
if is_collaborative:
allowed_banks = self.get_collaborative_allowed_banks()
filtered_memories = [
m for m in extracted_memories
if m.get("memory_bank") in allowed_banks
]Vector Store Isolation
We use the documents_chunk table to manage embeddings and give each memory a collection_name that we can look for. Collections enforce isolation at the storage level:
async def get_relevant_memories(self, message: str, user_id: str, folder_id: str = None, is_collaborative: bool = False):
results = []
# Always query user's Interaction Patterns (safe for collaborative)
if not is_collaborative:
# Full personal memory access
user_results = VECTOR_DB_CLIENT.search(
collection_name=f"user-memory-{user_id}",
vectors=[query_embedding],
limit=search_limit
)
results.extend(user_results)
else:
# Filter to allowed banks only
user_results = VECTOR_DB_CLIENT.search(
collection_name=f"user-memory-{user_id}",
vectors=[query_embedding],
limit=search_limit,
filter={"memory_bank": {"$in": ["Interaction Patterns"]}}
)
results.extend(user_results)
# Query project memories if in a folder
if folder_id:
folder_results = VECTOR_DB_CLIENT.search(
collection_name=f"folder-memory-{folder_id}",
vectors=[query_embedding],
limit=search_limit
)
for r in folder_results:
r["_source"] = "project" # Label for UI
results.extend(folder_results)
return self._deduplicate_and_rank(results)Optimization: Core Memories (Fast Path)
Even optimized vector search adds latency. For users with 100+ AI memories, we wanted a near-instant path.
Core Memories: A pre-computed 200-word summary generated on login.
# core_memories.py
SUMMARY_PROMPT = """Create a brief user profile based ONLY on the memories below.
Write 1-2 concise paragraphs covering:
- Who they are (use their account name; only include role/location if EXPLICITLY stated)
- How they communicate and work (preferences, style)
- What they're focused on (current projects, goals, interests)
Guidelines:
- Write in third person ("The user...", "They...")
- ONLY include information EXPLICITLY stated in the memories
- Keep under 200 words
- NEVER invent details not present in the memories
"""
async def generate_core_memories_summary(user_id: str) -> Optional[str]:
memories = Memories.get_memories_by_user_id(user_id)
# Guard against hallucination with sparse data
if len(memories) < MIN_MEMORIES_FOR_SUMMARY: # 5
return None
# Take most recent N memories
sorted_memories = sorted(memories, key=lambda m: m.updated_at, reverse=True)
selected = sorted_memories[:MAX_MEMORIES_FOR_SUMMARY] # 50
# Generate via LLM
summary = await _call_llm_for_summary(
SUMMARY_PROMPT.format(
user_name=user.name,
formatted_memories=format_memories(selected)
)
)
return summary
async def schedule_core_memories_update(user_id: str):
"""Called on login - fire-and-forget background task"""
summary = await generate_core_memories_summary(user_id)
if summary:
user = Users.get_user_by_id(user_id)
info = user.info or {}
info["core_memories"] = summary
info["core_memories_updated_at"] = int(time.time())
Users.update_user_by_id(user_id, {"info": info})Injection: Core memories go into every prompt without any vector search:
def _inject_core_memories(self, body: Dict, core_memories: str):
system_msg = self._get_or_create_system_message(body)
system_msg["content"] = f"""<userMemories>
{core_memories}
</userMemories>
{system_msg["content"]}"""Example output:
Alex is a senior software engineer working on developer tools. They prefer concise, technical responses with code examples over lengthy explanations. Currently focused on building an AI coding assistant, they often work late and appreciate responses that account for their time zone (PST). They’re learning Rust and have strong opinions about TypeScript.
Deduplication Pipeline
New memories go through multi-stage deduplication:
async def _deduplicate_memory(self, new_memory: str, user_id: str) -> bool:
"""Returns True if memory is a duplicate and should be skipped"""
existing = await self._get_user_memories(user_id)
for existing_mem in existing:
# Stage 1: Fast text similarity
text_sim = SequenceMatcher(None, new_memory, existing_mem.content).ratio()
if text_sim > 0.95:
return True # Near-exact duplicate
# Stage 2: Semantic similarity (if enabled)
if self.valves.use_embeddings_for_deduplication:
new_emb = await self._embed(new_memory)
existing_emb = self._get_cached_embedding(existing_mem.id)
cosine_sim = np.dot(new_emb, existing_emb)
if cosine_sim > self.valves.embedding_similarity_threshold: # 0.97
return True # Semantic duplicate
return FalseCaching Strategy
Three LRU caches prevent unbounded memory growth:
class AdaptiveMemoryFilter:
def __init__(self):
self._memory_embeddings = LRUCache(maxsize=5000) # memory_id → vector
self._relevance_cache = LRUCache(maxsize=10000) # (query_emb, mem_emb) → score
self._memory_to_relevance_keys = LRUCache(maxsize=5000) # Reverse index for invalidationWhen a memory is updated or deleted, we invalidate its cache entries:
def _invalidate_memory_caches(self, memory_id: str):
# Remove embedding
self._memory_embeddings.pop(memory_id, None)
# Remove relevance scores that used this memory
keys_to_remove = self._memory_to_relevance_keys.get(memory_id, set())
for key in keys_to_remove:
self._relevance_cache.pop(key, None)The Architecture Today
┌─────────────────────────────────────────────────────────────────┐
│ Chat Request │
│ body.metadata.chat_id, __user__ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ inlet() - Pre-processing │
│ 1. Check if collaborative chat │
│ 2. Determine allowed memory banks │
└─────────────────────────────────────────────────────────────────┘
│
┌─────────────────┴─────────────────┐
▼ ▼
┌─────────────────────────┐ ┌─────────────────────────────┐
│ Core Memories (Fast) │ │ Vector Search (Detailed) │
│ user.info["core_mem"] │ │ ChromaDB collections: │
│ ~200 words, 0ms │ │ • user-memory-{user_id} │
│ │ │ • folder-memory-{folder_id}│
└─────────────────────────┘ │ ~10-50ms │
│ └─────────────────────────────┘
│ │
│ ┌─────────────┴─────────────┐
│ ▼ ▼
│ ┌─────────────────────┐ ┌─────────────────────┐
│ │ Time Decay Scoring │ │ Semantic Reranking │
│ │ exp(-age/half_life) │ │ (optional) │
│ └─────────────────────┘ └─────────────────────┘
│ │ │
└─────────────────────┼───────────────────────────┘
▼
┌─────────────────────────────────────────────────────────────────┐
│ Inject into System Prompt │
│ <userMemories>Core summary</userMemories> │
│ + "I recall the following about you: [relevant memories]" │
└─────────────────────────────────────────────────────────────────┘
│
▼
LLM Response
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ outlet() - Post-processing │
│ 1. Skip if collaborative && Personal bank │
│ 2. Extract memories via LLM (structured JSON) │
│ 3. Deduplicate (text + semantic) │
│ 4. Assign to memory bank │
│ 5. Store in PostgreSQL + ChromaDB │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Background Tasks (async) │
│ • Summarization loop (24h) - cluster & consolidate │
│ • Cache cleanup - evict stale entries │
│ • Core memories refresh - on login │
└─────────────────────────────────────────────────────────────────┘Configuration
We have a lot of valves for every aspect of the memory management. The important ones:
# Performance
use_llm_for_relevance: bool = False # Expensive but accurate
llm_skip_relevance_threshold: float = 0.93 # Skip LLM above this
vector_similarity_threshold: float = 0.7 # Initial filter
# Limits
max_total_memories: int = 200 # Per-user cap
pruning_strategy: str = "fifo" # "fifo" or "least_relevant"
# Time decay
time_decay_half_life_days: int = 60
recency_weight: float = 0.2 # 20% recency boost
# Summarization
summarization_interval: int = 86400 # 24 hours
summarization_trigger_threshold: float = 0.8 # Trigger at 80% capacity
summarization_strategy: str = "hybrid" # "embeddings", "tags", "hybrid"Key Takeaways
- Tiered retrieval is essential. Fast vector search for most cases, expensive LLM scoring only when confidence is low.
- Privacy boundaries must be structural, not policy. We enforce isolation through collection naming and bank filtering, not just “please don’t leak this.”
- Pre-computation beats real-time. Core memories eliminate vector search latency for the common case. Detailed memories are available when needed.
- Background tasks keep the system healthy. Summarization, deduplication, and cache cleanup run async. The hot path stays fast.
- Give users control.
/memory list,/memory forget [id],/memory assign_bank [id] [bank]– users can inspect and manage their memories directly.
Memory is one of those features that seems simple until you build it. The naive version (store everything, retrieve by similarity) works for demos but breaks down at scale and in multi-user contexts.
The version that actually works? Layers of pragmatic trade-offs: fast paths for common cases, privacy boundaries that fail safe, and background processes that keep the system healthy over time.
And we’re still iterating.
Questions? Find us on GitHub or reach out at hello@cochat.ai
