memory-optimization

What this skill does

# Memory Optimization

Performance optimization patterns and tools for Mem0 memory systems. This skill provides comprehensive optimization techniques for query performance, cost reduction, caching strategies, and infrastructure tuning for both Platform and OSS deployments.

## Instructions

### Phase 1: Performance Assessment

Start by analyzing your current memory system performance:

```bash
bash scripts/analyze-performance.sh [project_name]
```

This generates a comprehensive performance report including:
- Query latency metrics (average, P95, P99)
- Operation throughput (searches, adds, updates, deletes)
- Cache performance statistics
- Resource utilization (memory, storage, CPU)
- Slow query identification
- Cost analysis

**Review the output to identify optimization priorities:**
- Query latency > 200ms → Focus on query optimization
- High costs → Focus on cost optimization
- Low cache hit rate < 60% → Focus on caching
- High resource usage → Focus on infrastructure tuning

### Phase 2: Query Optimization

Optimize memory search operations for speed and efficiency.

#### 2.1 Limit Search Results

**Problem**: Retrieving too many results increases latency and costs.

**Solution**: Use appropriate limit values based on use case.

```python
# ❌ BAD: Using default or excessive limits
memories = memory.search(query, user_id=user_id)  # Default: 10

# ✅ GOOD: Optimized limits
memories = memory.search(query, user_id=user_id, limit=5)  # Chat apps
memories = memory.search(query, user_id=user_id, limit=3)  # Quick context
memories = memory.search(query, user_id=user_id, limit=10) # RAG systems
```

**Impact**: 30-40% reduction in query time

**Guidelines**:
- Chat applications: 3-5 results
- RAG context retrieval: 8-12 results
- Recommendation systems: 10-20 results
- Semantic search: 20-50 results

#### 2.2 Use Filters to Reduce Search Space

**Problem**: Searching entire index is slow and expensive.

**Solution**: Apply filters to narrow search scope.

```python
# ❌ BAD: Full index scan
memories = memory.search(query)

# ✅ GOOD: Filtered search
memories = memory.search(
    query
    filters={
        "user_id": user_id
        "categories": ["preferences", "profile"]
    }
    limit=5
)

# ✅ BETTER: Multiple filter conditions
memories = memory.search(
    query
    filters={
        "AND": [
            {"user_id": user_id}
            {"agent_id": "support_v2"}
            {"created_after": "2025-01-01"}
        ]
    }
    limit=5
)
```

**Impact**: 40-60% reduction in query time

**Available Filters**:
- `user_id`: Scope to specific user
- `agent_id`: Scope to specific agent
- `run_id`: Scope to session/run
- `categories`: Filter by memory categories
- `metadata`: Custom metadata filters
- Date ranges: `created_after`, `created_before`

#### 2.3 Optimize Reranking

**Problem**: Default reranking may be overkill for simple queries.

**Solution**: Configure reranker based on accuracy requirements.

```python
# Platform Mode (Mem0 Cloud)
from mem0 import MemoryClient

# Disable reranking for fast, simple queries
memory = MemoryClient(api_key=api_key)
memories = memory.search(
    query
    user_id=user_id
    rerank=False  # 2x faster, slightly lower accuracy
)

# OSS Mode
from mem0 import Memory
from mem0.configs.base import MemoryConfig

# Use lightweight reranker
config = MemoryConfig(
    reranker={
        "provider": "cohere"
        "config": {
            "model": "rerank-english-v3.0",  # Fast model
            "top_n": 5  # Rerank only top results
        }
    }
)
memory = Memory(config)
```

**Reranker Options**:
- **No reranking**: Fastest, 90-95% accuracy
- **Lightweight (Cohere rerank-english-v3.0)**: 2x faster than full
- **Full reranking (Cohere rerank-english-v3.5)**: Highest accuracy

**Decision Guide**:
- Simple preference retrieval → No reranking
- Chat context → Lightweight reranking
- Critical RAG applications → Full reranking

#### 2.4 Use Async Operations

**Problem**: Blocking operations limit throughput.

**Solution**: Use async for high-concurrency scenarios.

```python
import asyncio
from mem0 import AsyncMemory

async def get_user_context(user_id: str, queries: list[str]):
    memory = AsyncMemory()

    # Run multiple searches concurrently
    results = await asyncio.gather(*[
        memory.search(q, user_id=user_id, limit=3)
        for q in queries
    ])

    return results

# Usage
contexts = await get_user_context(
    "user_123"
    ["preferences", "recent activity", "goals"]
)
```

**Impact**: 3-5x throughput improvement under load

### Phase 3: Caching Strategies

Implement multi-layer caching to reduce API calls and improve response times.

#### 3.1 In-Memory Caching (Python)

**Use for**: Frequently accessed, rarely changing data (user preferences).

```python
from functools import lru_cache
import hashlib

@lru_cache(maxsize=1000)
def get_user_preferences(user_id: str) -> list:
    """Cache user preferences for 5 minutes"""
    return memory.search(
        "user preferences"
        user_id=user_id
        limit=5
    )

# Clear cache when preferences update
get_user_preferences.cache_clear()
```

**Impact**: Near-instant response for cached queries

**Configuration**:
- `maxsize=1000`: Cache 1000 users' preferences
- Clear cache on memory updates
- TTL: Implement with time-based wrapper

#### 3.2 Redis Caching (Production)

**Use for**: Shared caching across services, TTL control.

```python
import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def get_user_context_cached(user_id: str, query: str) -> list:
    # Generate cache key
    cache_key = f"mem0:search:{user_id}:{hashlib.md5(query.encode()).hexdigest()}"

    # Check cache
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    # Cache miss - query Mem0
    result = memory.search(query, user_id=user_id, limit=5)

    # Cache result (5 minute TTL)
    redis_client.setex(
        cache_key
        300,  # 5 minutes
        json.dumps(result)
    )

    return result

# Invalidate cache on update
def update_memory(user_id: str, message: str):
    memory.add(message, user_id=user_id)

    # Clear user's cache
    pattern = f"mem0:search:{user_id}:*"
    for key in redis_client.scan_iter(match=pattern):
        redis_client.delete(key)
```

**Impact**: 50-70% reduction in API calls

**TTL Guidelines**:
- User preferences: 5-15 minutes
- Agent knowledge: 30-60 minutes
- Session context: 1-2 minutes
- Static content: 1 hour

Use the caching template generator:
```bash
bash scripts/generate-cache-config.sh redis [ttl_seconds]
```

#### 3.3 Edge Caching (Advanced)

**Use for**: Global applications, very high traffic.

See template: `templates/edge-cache-config.yaml`

### Phase 4: Embedding Optimization

Optimize embedding generation and storage costs.

#### 4.1 Choose Appropriate Embedding Model

**Problem**: Oversized embeddings increase cost and latency.

**Solution**: Match model to use case.

```python
from mem0 import Memory
from mem0.configs.base import MemoryConfig

# ❌ EXPENSIVE: Large model for simple data
config = MemoryConfig(
    embedder={
        "provider": "openai"
        "config": {
            "model": "text-embedding-3-large",  # 3072 dims, $0.13/1M tokens
        }
    }
)

# ✅ OPTIMIZED: Appropriate model
config = MemoryConfig(
    embedder={
        "provider": "openai"
        "config": {
            "model": "text-embedding-3-small",  # 1536 dims, $0.02/1M tokens
        }
    }
)
```

**Model Selection Guide**:

| Use Case | Recommended Model | Dimensions | Cost |
|----------|------------------|------------|------|
| User preferences | text-embedding-3-small | 1536 | $0.02/1M |
| Simple chat context | text-embedding-3-small | 1536 | $0.02/1M |
| Advanced RAG | text-embedding-3-large | 3072 | $0.13/1M |
| Multilingual | text-embedding-3-large | 3072 | $0.13/1M |
| Budget-conscious | text-embedding-ada-002 | 1536 | $0.0001/1M |

**Impact**: 70-85% cost reduc