Mem0: Universal Memory Layer for AI Agents

Mem0 Architecture:

┌─────────────────────────────────────────┐
│  User Interaction / Agent Input         │
└──────────────┬──────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────┐
│  LLM Extraction (GPT-4-mini default)    │
│  • Extract atomic facts                 │
│  • Deduplicate against existing memory  │
└──────────────┬──────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────┐
│  Embedding (text-embedding-3-small)     │
│  • Convert facts to 1536-dim vectors    │
└──────────────┬──────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────┐
│  Vector Store (Qdrant/Pinecone/etc)     │
│  • Index by user_id, session_id, agent  │
│  • Store embeddings + metadata          │
└──────────────┬──────────────────────────┘
               │
        [Retrieval Phase]
               │
               ▼
┌─────────────────────────────────────────┐
│  Multi-Signal Retrieval                 │
│  • Semantic similarity search           │
│  • BM25 keyword matching                │
│  • Entity matching                      │
│  • Temporal filtering                   │
└──────────────┬──────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────┐
│  Inject into LLM Context Window         │
│  • Top-k relevant memories              │
│  • Minimal token overhead               │
└─────────────────────────────────────────┘

# Verify Python version (must be 3.10+)
python --version

# Set OpenAI API key (required for default configuration)
export OPENAI_API_KEY="your-openai-api-key"

# Optional: Verify Docker (if using self-hosted vector stores)
docker --version

# Optional: Check available memory and disk space
free -h
df -h

# Basic installation
pip install mem0ai

# Install with vector store support (includes Chroma, Pinecone, etc.)
pip install "mem0ai[vector_stores]"

# Install with NLP enhancements for better extraction
pip install "mem0ai[nlp]"

# Install all optional dependencies
pip install "mem0ai[all]"

# Verify installation
python -c "from mem0 import Memory; print('Mem0 installed successfully')"

from mem0 import Memory

# Initialize with default configuration
m = Memory()

# Add memories from a conversation
# Mem0 automatically extracts and stores atomic facts
messages = [
    {"role": "user", "content": "Hi, I'm Alice. I'm a software engineer."},
    {"role": "assistant", "content": "Hello Alice! Nice to meet you."},
    {"role": "user", "content": "I love Python and prefer dark mode in my IDE."},
]

# Store memories with user_id for personalization
result = m.add(messages, user_id="alice")
print(f"Stored {len(result)} memories for alice")

# Retrieve relevant memories
query = "What programming languages does Alice like?"
memories = m.search(query, user_id="alice")

# Display retrieved memories
for mem in memories:
    print(f"Memory: {mem['memory']}")
    print(f"Score: {mem['score']}")
    print(f"Created: {mem['created_at']}")
    print("---")

from mem0 import Memory

# Example 1: Custom OpenAI + Qdrant (cloud)
config = {
    "llm": {
        "provider": "openai",
        "config": {
            "model": "gpt-4o-mini",
            "temperature": 0.1,
            "max_tokens": 2000,
        }
    },
    "embedder": {
        "provider": "openai",
        "config": {
            "model": "text-embedding-3-small",
            "embedding_dims": 1536
        }
    },
    "vector_store": {
        "provider": "qdrant",
        "config": {
            "host": "localhost",
            "port": 6333,
            "collection_name": "memories",
            "embedding_model_dims": 1536,
            "distance": "cosine"
        }
    }
}

m = Memory.from_config(config)

# Example 2: Anthropic Claude + Pinecone
config_claude = {
    "llm": {
        "provider": "anthropic",
        "config": {
            "model": "claude-sonnet-4-6",
            "temperature": 0.0,
            "max_tokens": 4000,
        }
    },
    "vector_store": {
        "provider": "pinecone",
        "config": {
            "api_key": "your-pinecone-api-key",
            "index_name": "mem0-memories",
            "environment": "us-west1-gcp"
        }
    }
}

# Example 3: Fully local with Ollama + ChromaDB
config_local = {
    "llm": {
        "provider": "ollama",
        "config": {
            "model": "llama3.2",
            "ollama_base_url": "http://localhost:11434"
        }
    },
    "embedder": {
        "provider": "ollama",
        "config": {
            "model": "nomic-embed-text"
        }
    },
    "vector_store": {
        "provider": "chroma",
        "config": {
            "collection_name": "memories",
            "path": "./chroma_db"
        }
    }
}

# Qdrant - Docker setup
docker run -d -p 6333:6333 -p 6334:6334 \
  -v $(pwd)/qdrant_storage:/qdrant/storage \
  qdrant/qdrant

# Verify Qdrant is running
curl http://localhost:6333/

# pgvector - Enable in PostgreSQL
# First, install the extension (Ubuntu/Debian):
sudo apt-get install postgresql-16-pgvector

# Then in PostgreSQL:
# CREATE EXTENSION vector;

# ChromaDB - No setup needed, runs in-process
# Just install: pip install chromadb

# Pinecone - Cloud service, no local setup
# Sign up at https://www.pinecone.io/
# Get API key and environment name from dashboard

from mem0 import Memory

m = Memory()

# 1. ADD - Store memories from messages
messages = [
    {"role": "user", "content": "My name is Bob and I work at Acme Corp."},
    {"role": "user", "content": "I prefer email updates over phone calls."}
]

result = m.add(messages, user_id="bob")
print(f"Added memories: {result}")

# ADD - Store from raw text
m.add("Bob's favorite color is blue.", user_id="bob")

# 2. SEARCH - Retrieve relevant memories
memories = m.search(
    query="What is Bob's communication preference?",
    user_id="bob",
    limit=5  # Top 5 results
)

for mem in memories:
    print(f"{mem['memory']} (score: {mem['score']})")

# 3. UPDATE - Modify existing memory
memory_id = memories[0]['id']
m.update(
    memory_id=memory_id,
    data="Bob now prefers Slack messages over email."
)

# 4. DELETE - Remove specific memory
m.delete(memory_id=memory_id)

# DELETE - Clear all memories for a user
m.delete_all(user_id="bob")

# GET - Retrieve all memories for a user
all_memories = m.get_all(user_id="bob")
print(f"Total memories for Bob: {len(all_memories)}")

# MULTI-LEVEL - Combine user and session context
m.add(
    "Currently discussing project budget",
    user_id="bob",
    session_id="meeting-2026-05-27"
)

# Search with multiple filters
memories = m.search(
    query="project discussion",
    user_id="bob",
    session_id="meeting-2026-05-27"
)

# Install CLI
pip install mem0-cli

# Initialize configuration (creates ~/.mem0/config.yaml)
mem0 init

# Add a memory
mem0 add "Alice prefers Python over JavaScript" --user-id alice

# Search memories
mem0 search "programming preferences" --user-id alice

# List all memories for a user
mem0 list --user-id alice

# Delete a specific memory
mem0 delete <memory-id>

# Clear all memories for a user
mem0 clear --user-id alice

# View CLI help
mem0 --help
mem0 add --help

# LangGraph Integration Example
from langgraph.graph import StateGraph
from mem0 import Memory

memory = Memory()

def chatbot_node(state):
    user_id = state["user_id"]
    message = state["message"]
    
    # Retrieve relevant memories
    memories = memory.search(message, user_id=user_id, limit=3)
    context = "\n".join([m['memory'] for m in memories])
    
    # Add context to LLM prompt
    response = llm.invoke(f"Context: {context}\n\nUser: {message}")
    
    # Store new interaction
    memory.add([
        {"role": "user", "content": message},
        {"role": "assistant", "content": response}
    ], user_id=user_id)
    
    return {"response": response}

# CrewAI Integration Example
from crewai import Agent, Task, Crew
from mem0 import Memory

memory = Memory()

researcher = Agent(
    role="Researcher",
    goal="Research topics thoroughly",
    backstory="Expert researcher with access to memory",
    memory=memory  # Attach Mem0 to agent
)

task = Task(
    description="Research AI agent memory systems",
    agent=researcher,
    user_id="research-session-001"  # Memory scoping
)

crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()

# Hierarchical memory scoping
memory = Memory()

# Long-term user facts
memory.add(
    "Alice is a senior engineer at Acme Corp",
    user_id="alice",
    metadata={"category": "profile", "retention_days": 365}
)

# Session-specific context
memory.add(
    "Currently debugging authentication issue",
    user_id="alice",
    session_id="support-ticket-1234",
    metadata={"category": "session", "retention_days": 30}
)

# Agent-level patterns
memory.add(
    "Users from this domain prefer technical explanations",
    agent_id="support-bot-v2",
    metadata={"category": "pattern", "learned_from": "alice"}
)

# Scoped retrieval with metadata filtering
memories = memory.search(
    query="authentication",
    user_id="alice",
    session_id="support-ticket-1234",
    filters={"category": "session"}  # Only session memories
)

# Memory cleanup based on metadata
import datetime

def cleanup_old_sessions():
    all_memories = memory.get_all()
    for mem in all_memories:
        metadata = mem.get('metadata', {})
        if metadata.get('category') == 'session':
            created = datetime.datetime.fromisoformat(mem['created_at'])
            age_days = (datetime.datetime.now() - created).days
            if age_days > metadata.get('retention_days', 30):
                memory.delete(mem['id'])
                print(f"Deleted expired session memory: {mem['id']}")

cleanup_old_sessions()

# docker-compose.yml for Mem0 self-hosted deployment
version: '3.8'

services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_storage:/qdrant/storage
    restart: unless-stopped

  mem0-server:
    image: mem0ai/mem0:latest
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - QDRANT_HOST=qdrant
      - QDRANT_PORT=6333
      - AUTH_ENABLED=true
      - SECRET_KEY=${SECRET_KEY}  # Generate with: openssl rand -base64 32
    depends_on:
      - qdrant
    restart: unless-stopped

  postgres:
    image: postgres:16-alpine
    environment:
      - POSTGRES_USER=mem0
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - POSTGRES_DB=mem0_history
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: unless-stopped

volumes:
  qdrant_storage:
  postgres_data:

# Performance monitoring example
import time
import logging
from mem0 import Memory

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

memory = Memory()

def monitor_operation(operation_name, func, *args, **kwargs):
    """Wrapper to monitor Mem0 operations"""
    start = time.time()
    try:
        result = func(*args, **kwargs)
        duration = time.time() - start
        logger.info(f"{operation_name} completed in {duration:.2f}s")
        
        # Log to metrics system (Prometheus, DataDog, etc.)
        # metrics.record_latency(operation_name, duration)
        
        return result
    except Exception as e:
        duration = time.time() - start
        logger.error(f"{operation_name} failed after {duration:.2f}s: {e}")
        # metrics.record_error(operation_name, str(e))
        raise

# Monitor add operation
messages = [{"role": "user", "content": "I prefer dark mode"}]
result = monitor_operation(
    "memory_add",
    memory.add,
    messages,
    user_id="alice"
)

# Monitor search operation
memories = monitor_operation(
    "memory_search",
    memory.search,
    "preferences",
    user_id="alice"
)

# Analyze retrieval quality
def analyze_retrieval_quality(memories, threshold=0.7):
    """Check if retrieved memories meet quality threshold"""
    if not memories:
        logger.warning("No memories retrieved")
        return False
    
    avg_score = sum(m['score'] for m in memories) / len(memories)
    logger.info(f"Average relevance score: {avg_score:.3f}")
    
    if avg_score < threshold:
        logger.warning(f"Low relevance score: {avg_score:.3f}")
        return False
    
    return True

analyze_retrieval_quality(memories)

# Troubleshooting commands

# Check Qdrant is running and accessible
curl http://localhost:6333/
curl http://localhost:6333/collections

# View Qdrant logs
docker logs <qdrant-container-id>

# Check Mem0 local storage
ls -lh /tmp/qdrant/
ls -lh ~/.mem0/

# Test Mem0 installation
python -c "from mem0 import Memory; m = Memory(); print('OK')"

# Debug Python dependencies
pip show mem0ai
pip list | grep -E '(mem0|qdrant|openai|chromadb)'

# Clear local Qdrant data (reset)
rm -rf /tmp/qdrant/

# Clear Mem0 history database
rm ~/.mem0/history.db

# Enable debug logging in Python
import logging
logging.basicConfig(level=logging.DEBUG)

# Check OpenAI API key
echo $OPENAI_API_KEY | head -c 20
curl https://api.openai.com/v1/models -H "Authorization: Bearer $OPENAI_API_KEY"

# Migration example: Bulk import from existing system
from mem0 import Memory
import json

memory = Memory()

# Example: Import from JSON export
with open('existing_memories.json', 'r') as f:
    data = json.load(f)

# Bulk import with progress tracking
for user_data in data['users']:
    user_id = user_data['id']
    conversations = user_data['conversations']
    
    for conv in conversations:
        # Convert to Mem0 message format
        messages = [
            {"role": msg['role'], "content": msg['content']}
            for msg in conv['messages']
        ]
        
        try:
            result = memory.add(
                messages,
                user_id=user_id,
                metadata={
                    "imported_from": "openai-memory",
                    "original_timestamp": conv['timestamp']
                }
            )
            print(f"Imported {len(result)} memories for user {user_id}")
        except Exception as e:
            print(f"Error importing for {user_id}: {e}")

# Verify migration
total_memories = len(memory.get_all())
print(f"Total memories after migration: {total_memories}")

# Quality check: Test retrieval
test_query = "What are user preferences?"
results = memory.search(test_query, limit=10)
print(f"Retrieval test returned {len(results)} results")

# Advanced use case: Multi-agent customer support system
from mem0 import Memory

memory = Memory()

class CustomerSupportSystem:
    def __init__(self):
        self.memory = Memory()
    
    def handle_customer_query(self, customer_id, query, agent_id):
        """
        Multi-level memory retrieval:
        - Customer history (user_id)
        - Current session context (session_id)
        - Agent best practices (agent_id)
        """
        # Retrieve customer history
        customer_memories = self.memory.search(
            query,
            user_id=customer_id,
            limit=5
        )
        
        # Retrieve agent-level best practices
        agent_memories = self.memory.search(
            query,
            agent_id=agent_id,
            limit=3
        )
        
        # Combine context
        context = {
            "customer_history": [m['memory'] for m in customer_memories],
            "agent_knowledge": [m['memory'] for m in agent_memories]
        }
        
        # Generate response with full context
        response = self.generate_response(query, context)
        
        # Store new interaction
        self.memory.add(
            [
                {"role": "user", "content": query},
                {"role": "assistant", "content": response}
            ],
            user_id=customer_id,
            metadata={
                "agent_id": agent_id,
                "resolved": False,
                "category": "support"
            }
        )
        
        return response
    
    def learn_from_interaction(self, customer_id, agent_id, outcome):
        """Store learned patterns for future use"""
        if outcome == "successful":
            self.memory.add(
                f"This approach worked well for customer issues",
                agent_id=agent_id,
                metadata={"learned_from": customer_id, "outcome": outcome}
            )
    
    def generate_response(self, query, context):
        # Implement LLM call with context
        pass

# Usage
support = CustomerSupportSystem()
response = support.handle_customer_query(
    customer_id="customer-123",
    query="My API key isn't working",
    agent_id="support-bot-v2"
)

support.learn_from_interaction(
    customer_id="customer-123",
    agent_id="support-bot-v2",
    outcome="successful"
)