mem0 Memory System
"""
Autonomous memory system for the mem0 MCP server
This module provides middleware that automatically:
1. Extracts user information from conversations
2. Stores this information in the memory database
3. Retrieves relevant memories for each interaction
4. Injects these memories into the context for the LLM
No explicit memory commands are needed - the system works autonomously.
"""
import re
import json
from typing import Dict, Any, List, Optional, Union, Tuple
import logging
from datetime import datetime
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger("auto_memory")
class AutoMemory:
"""
Autonomous memory system that automatically extracts, stores, and retrieves
user information without requiring explicit commands.
"""
# Categories of information to extract and store
CATEGORIES = {
"personal_info": [
"name", "age", "birthday", "occupation", "location", "hometown",
"education", "family", "pets", "hobbies", "interests"
],
"preferences": [
"likes", "dislikes", "favorite", "prefer", "enjoy", "hate", "love",
"color", "food", "movie", "book", "music", "artist", "genre", "style"
],
"context": [
"project", "task", "goal", "objective", "working on", "trying to",
"building", "developing", "learning", "studying"
]
}
# Regular expressions for extracting information
EXTRACTION_PATTERNS = {
# Personal statements
"i_statement": r"(?:^|\s)I\s+(?:am|'m|was)\s+([^,.!?]+)",
"my_statement": r"(?:^|\s)My\s+([^,.!?]+)\s+(?:is|are|was|were)\s+([^,.!?]+)",
"i_have": r"(?:^|\s)I\s+(?:have|had|own|possess)\s+([^,.!?]+)",
"i_like": r"(?:^|\s)I\s+(?:like|love|enjoy|prefer|adore)\s+([^,.!?]+)",
"i_dislike": r"(?:^|\s)I\s+(?:dislike|hate|don't like|do not like|can't stand)\s+([^,.!?]+)",
"i_want": r"(?:^|\s)I\s+(?:want|need|wish|desire|would like)\s+([^,.!?]+)",
# Questions about the user that were answered
"question_answer": r"(?:what|who|where|when|how|why)[^?]+\?\s*([A-Z][^,.!?]+)",
# Direct statements
"name_is": r"(?:^|\s)(?:my|the|)\s*name\s+(?:is|was|'s)\s+([^,.!?]+)",
"call_me": r"(?:^|\s)(?:you can |please |)call me\s+([^,.!?]+)",
}
def __init__(self, memory_provider):
"""
Initialize the autonomous memory system
Args:
memory_provider: Instance of the MemoryProvider class
"""
self.memory_provider = memory_provider
self.current_session_info = {}
def process_user_message(self, user_id: str, message: str) -> Dict[str, Any]:
"""
Process a user message to extract and store information
Args:
user_id: User ID
message: User message
Returns:
Dictionary of extracted information
"""
extracted_info = self._extract_information(message)
if extracted_info:
# Update session info
self.current_session_info.update(extracted_info)
# Store the extracted information in memory
self._store_information(user_id, extracted_info)
logger.info(f"Extracted and stored information for user {user_id}: {extracted_info}")
return extracted_info
def get_relevant_memories(self, user_id: str, message: str) -> Dict[str, Any]:
"""
Get relevant memories for the current interaction
Args:
user_id: User ID
message: Current user message
Returns:
Dictionary of relevant memories
"""
# First, get user profile from stored memories
user_profile = self._get_user_profile(user_id)
# Then, get memories relevant to the current message
relevant_memories = self._search_relevant_memories(user_id, message)
return {
"user_profile": user_profile,
"relevant_memories": relevant_memories
}
def create_memory_prompt(self, user_id: str, message: str) -> str:
"""
Create a prompt that includes relevant memories
Args:
user_id: User ID
message: Current user message
Returns:
Prompt with injected memories
"""
memories = self.get_relevant_memories(user_id, message)
# Create a prompt section with the user profile
profile_section = ""
if memories["user_profile"]:
profile_section = "User Information:\n"
for category, items in memories["user_profile"].items():
if items:
profile_section += f"- {category.replace('_', ' ').title()}: {', '.join(items)}\n"
# Create a prompt section with relevant memories
memories_section = ""
if memories["relevant_memories"]:
memories_section = "Relevant Past Interactions:\n"
for i, memory in enumerate(memories["relevant_memories"], 1):
memories_section += f"{i}. {memory}\n"
# Combine into a complete context
if profile_section or memories_section:
memory_prompt = f"""
The following information about the user may be helpful for your response:
{profile_section}
{memories_section}
Remember to use this information naturally in your response without explicitly mentioning that you're using stored memories.
"""
return memory_prompt.strip()
return ""
def enhance_system_prompt(self, system_prompt: Optional[str], user_id: str, message: str) -> str:
"""
Enhance the system prompt with memory information
Args:
system_prompt: Original system prompt
user_id: User ID
message: Current user message
Returns:
Enhanced system prompt
"""
memory_prompt = self.create_memory_prompt(user_id, message)
if not system_prompt:
system_prompt = "You are a helpful assistant."
if memory_prompt:
return f"{system_prompt}\n\n{memory_prompt}"
return system_prompt
def _extract_information(self, message: str) -> Dict[str, Any]:
"""
Extract user information from a message
Args:
message: User message
Returns:
Dictionary of extracted information
"""
extracted_info = {
"personal_info": {},
"preferences": {},
"context": {}
}
# Apply each extraction pattern
for pattern_name, pattern in self.EXTRACTION_PATTERNS.items():
matches = re.finditer(pattern, message, re.IGNORECASE)
for match in matches:
# Different handling based on pattern type
if pattern_name == "my_statement" and len(match.groups()) >= 2:
attribute = match.group(1).strip().lower()
value = match.group(2).strip()
self._categorize_and_store(extracted_info, attribute, value)
elif pattern_name in ["i_statement", "i_have", "i_like", "i_dislike", "i_want",
"question_answer", "name_is", "call_me"]:
value = match.group(1).strip()
# Special handling for name patterns
if pattern_name in ["name_is", "call_me"]:
extracted_info["personal_info"]["name"] = value
else:
# For other patterns, try to categorize the information
self._categorize_statement(extracted_info, pattern_name, value)
# Clean up the extracted info to remove empty categories
for category in list(extracted_info.keys()):
if not extracted_info[category]:
del extracted_info[category]
return extracted_info
def _categorize_statement(self, extracted_info: Dict[str, Dict[str, str]],
pattern_type: str, value: str) -> None:
"""
Categorize a statement into the appropriate category
Args:
extracted_info: Dictionary to store categorized information
pattern_type: Type of pattern that matched
value: Extracted value
"""
# Handle different pattern types
if pattern_type == "i_like":
self._add_to_category(extracted_info, "preferences", "likes", value)
elif pattern_type == "i_dislike":
self._add_to_category(extracted_info, "preferences", "dislikes", value)
elif pattern_type == "i_want":
self._add_to_category(extracted_info, "context", "goals", value)
else:
# For other patterns, try to match keywords from categories
self._categorize_by_keywords(extracted_info, value)
def _categorize_by_keywords(self, extracted_info: Dict[str, Dict[str, str]],
value: str) -> None:
"""
Categorize information based on keywords
Args:
extracted_info: Dictionary to store categorized information
value: Value to categorize
"""
value_lower = value.lower()
# Check each category for keyword matches
for category, keywords in self.CATEGORIES.items():
for keyword in keywords:
if keyword.lower() in value_lower:
# Extract the relevant part of the value
parts = value_lower.split(keyword.lower(), 1)
if len(parts) > 1:
relevant_part = parts[1].strip()
if relevant_part:
self._add_to_category(extracted_info, category, keyword, relevant_part)
else:
self._add_to_category(extracted_info, category, keyword, value)
def _categorize_and_store(self, extracted_info: Dict[str, Dict[str, str]],
attribute: str, value: str) -> None:
"""
Categorize and store an attribute-value pair
Args:
extracted_info: Dictionary to store categorized information
attribute: Attribute name
value: Attribute value
"""
# Check which category the attribute belongs to
for category, keywords in self.CATEGORIES.items():
for keyword in keywords:
if keyword.lower() in attribute.lower():
self._add_to_category(extracted_info, category, attribute, value)
return
# If no specific category is found, use a default category
self._add_to_category(extracted_info, "personal_info", attribute, value)
def _add_to_category(self, extracted_info: Dict[str, Dict[str, str]],
category: str, key: str, value: str) -> None:
"""
Add information to a specific category
Args:
extracted_info: Dictionary to store categorized information
category: Category name
key: Information key
value: Information value
"""
if category not in extracted_info:
extracted_info[category] = {}
if key not in extracted_info[category]:
extracted_info[category][key] = value
else:
# If the key already exists, append the new value
existing_value = extracted_info[category][key]
if value not in existing_value:
extracted_info[category][key] = f"{existing_value}, {value}"
def _store_information(self, user_id: str, extracted_info: Dict[str, Any]) -> None:
"""
Store extracted information in memory
Args:
user_id: User ID
extracted_info: Extracted information
"""
if not extracted_info:
return
# Format the information as a structured memory
timestamp = datetime.now().isoformat()
for category, items in extracted_info.items():
if items:
# Create a memory for each category
memory_content = f"User {category.replace('_', ' ')}: " + ", ".join(
f"{k}: {v}" for k, v in items.items()
)
# Store in memory with appropriate metadata
self.memory_provider.add_memory(
content=memory_content,
user_id=user_id,
metadata={
"type": "user_information",
"category": category,
"timestamp": timestamp,
"automatic": True
}
)
def _get_user_profile(self, user_id: str) -> Dict[str, List[str]]:
"""
Get the user profile from stored memories
Args:
user_id: User ID
Returns:
Dictionary with user profile information
"""
# Search for user information memories
results = self.memory_provider.search_memories(
query="user information profile preferences",
user_id=user_id,
limit=20,
metadata_filter={"type": "user_information"}
)
# Extract and organize the information
profile = {
"personal_info": [],
"preferences": [],
"context": []
}
for result in results.get("results", []):
memory = result.get("memory", "")
category = result.get("metadata", {}).get("category", "")
if category in profile:
# Extract key-value pairs from the memory
pairs = re.findall(r'(\w+):\s*([^,]+)(?:,|$)', memory)
for key, value in pairs:
item = f"{key.strip()}: {value.strip()}"
if item not in profile[category]:
profile[category].append(item)
return profile
def _search_relevant_memories(self, user_id: str, message: str) -> List[str]:
"""
Search for memories relevant to the current message
Args:
user_id: User ID
message: Current user message
Returns:
List of relevant memories
"""
# Search for memories relevant to the current message
results = self.memory_provider.search_memories(
query=message,
user_id=user_id,
limit=5
)
# Extract the memory content
memories = []
for result in results.get("results", []):
memory = result.get("memory", "")
if memory and memory not in memories:
memories.append(memory)
return memories