Performance Tuning

This guide covers optimization strategies and best practices for maximizing SmartMemory performance in production environments.

Performance Overview

SmartMemory performance depends on several key factors:

Graph database backend (FalkorDB, SQLite)
Vector store configuration (FalkorDB native HNSW, optional usearch)
Background processing settings
Caching strategies
Query optimization

Backend Optimization

Graph Database Tuning

FalkorDB Configuration

falkordb_config = {
    "graph_backend": {
        "type": "falkordb",
        "host": "localhost",
        "port": 6379,
        "connection_pool": {
            "max_connections": 20,
            "retry_on_timeout": True,
            "socket_keepalive": True,
            "socket_keepalive_options": {
                "TCP_KEEPIDLE": 1,
                "TCP_KEEPINTVL": 3,
                "TCP_KEEPCNT": 5
            }
        },
        "query_optimization": {
            "enable_query_cache": True,
            "cache_size": "256MB",
            "query_timeout": 30
        }
    }
}

SQLite Configuration (Lite Mode)

For zero-infrastructure local use, the SQLite backend stores the graph in a single file. It needs no connection pool or external service — point it at a file path:

sqlite_config = {
    "graph_db": {
        "backend_class": "SQLiteBackend",
        "db_path": "./data/memory.db"
    }
}

SQLite is single-tenant and intended for SmartMemory Lite; for multi-tenant production workloads use FalkorDB.

SmartMemory ships two vector backends: falkordb (the default — embeddings are stored natively inside FalkorDB via an HNSW index, so there is no separate vector service) and usearch (optional, embedded). The default is the recommended production choice.

FalkorDB HNSW Tuning

The HNSW index parameters are the primary recall/latency knobs. All of these are real config keys read by the FalkorDB vector backend:

falkordb_vector_config = {
    "vector": {
        "backend": "falkordb",
        "host": "localhost",
        "port": 9010,
        "dimension": 768,            # derived from the embedding model; falls back to this
        "metric": "cosine",          # 'cosine' | 'euclidean'
        "hnsw_m": 16,                # graph connectivity — higher = better recall, more memory
        "hnsw_ef_construction": 200, # build-time candidate breadth — higher = better index, slower build
        "hnsw_ef_runtime": 64        # query-time candidate breadth — higher = better recall, slower queries
    }
}

Tuning guidance:

Recall too low — raise hnsw_ef_runtime first (query-time only), then hnsw_m / hnsw_ef_construction (require a rebuild).
Queries too slow — lower hnsw_ef_runtime.
Index build too slow / memory-heavy — lower hnsw_m and hnsw_ef_construction.
dimension is auto-detected from the active embedding model; only set it explicitly if the probe cannot run.

usearch (Optional)

The optional embedded backend honors the same dimension and metric keys:

usearch_vector_config = {
    "vector": {
        "backend": "usearch",
        "dimension": 768,
        "metric": "cosine"
    }
}

Memory Configuration Optimization

Background Processing Tuning

background_config = {
    "background_processing": {
        "enabled": True,
        "max_workers": 4,  # Adjust based on CPU cores
        "queue_size": 10000,
        "batch_processing": {
            "enabled": True,
            "batch_size": 100,
            "batch_timeout": 5.0
        },
        "worker_config": {
            "enrichment_workers": 2,
            "evolution_workers": 1,
            "cleanup_workers": 1
        }
    }
}

Caching Configuration

caching_config = {
    "caching": {
        "l1_cache": {
            "enabled": True,
            "max_size": 1000,
            "ttl": 300  # 5 minutes
        },
        "l2_cache": {
            "enabled": True,
            "type": "redis",
            "host": "localhost",
            "port": 6379,
            "max_memory": "1GB",
            "ttl": 3600  # 1 hour
        },
        "query_cache": {
            "enabled": True,
            "max_size": 500,
            "ttl": 600  # 10 minutes
        }
    }
}

# Use specific memory types to reduce search space
results = memory.search(
    query="machine learning",
    memory_type="semantic",  # Limit to specific type
    top_k=10,  # Reasonable limit
    filters={
        "user_id": "user123",  # Pre-filter by user
        "date_range": {
            "start": "2024-01-01",
            "end": "2024-12-31"
        }
    }
)
# Batch multiple searches
queries = ["AI", "ML", "Deep Learning"]
results = {q: memory.search(q, top_k=10, memory_type="semantic") for q in queries}

# Get direct neighbors of a node
related = memory.get_neighbors("memory_123")
# Apply any filtering/weighting client-side as needed

Bulk Operations

Batch Ingestion

# Efficient batch processing
items = [
    {"content": f"Document {i}", "metadata": {"batch": "1"}}
    for i in range(1000)
]

# Process in optimized batches (enqueue for background)
def chunk(iterable, size):
    for i in range(0, len(iterable), size):
        yield iterable[i:i+size]

for chunk_items in chunk(items, 100):
    for it in chunk_items:
        memory.ingest(it, sync=False)

Bulk Updates

# Batch update operations
updates = [
    {"item_id": f"memory_{i}", "properties": {"importance": 0.8}}
    for i in range(100)
]

for upd in updates:
    memory.update_properties(upd["item_id"], upd["properties"])  # merge semantics

Monitoring and Profiling

Performance Metrics Collection

Illustrative only. SmartMemory does not ship a smartmemory.monitoring module or a PerformanceMonitor class. The pattern below sketches a user-authored timing wrapper around the real memory.search(...) call — adapt it to your own metrics backend. The decorator and stats dict are not part of the public API.

import time
from functools import wraps

# Example: a thin timing wrapper you write yourself.
def profile_operation(name):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            result = fn(*args, **kwargs)
            elapsed_ms = (time.perf_counter() - start) * 1000
            print(f"{name}: {elapsed_ms:.2f}ms")  # send to your metrics sink
            return result
        return wrapper
    return decorator

@profile_operation("custom_search")
def complex_search(query):
    return memory.search(query, top_k=50)

Resource Monitoring

import smartmemory.utils
import psutil
import asyncio


class ResourceMonitor:
    def __init__(self, memory_instance):
        self.memory = memory_instance
        self.metrics = []

    async def monitor_resources(self):
        while True:
            # System metrics
            cpu_percent = psutil.cpu_percent(interval=1)
            memory_info = psutil.virtual_memory()

            # Optional SmartMemory summary (counts by type)
            summary = self.memory.summary()

            metrics = {
                "timestamp": smartmemory.utils.now(),
                "cpu_percent": cpu_percent,
                "memory_percent": memory_info.percent,
                # add SmartMemory-derived metrics as needed from `summary`
            }

            self.metrics.append(metrics)

            # Alert on high resources usage
            if cpu_percent > 80 or memory_info.percent > 85:
                await self.send_alert(metrics)

            await asyncio.sleep(60)  # Check every minute

Optimization Strategies

Memory Lifecycle Optimization

Illustrative only. There is no memory.configure_lifecycle(...) method on SmartMemory. The config dict below is a design sketch of retention and optimization knobs — it is not a callable API. Real lifecycle behaviour is driven by the sleep-cycle daemon, snapshot retention sweeps, and tier compaction (see the environment variables in the project README and the Tuning & Surfacing guide), not by a single config call.

# Sketch of a retention/optimization policy (not a callable API).
lifecycle_config = {
    "retention_policies": {
        "working_memory": {
            "max_items": 1000,
            "cleanup_interval": 300,  # 5 minutes
            "cleanup_strategy": "lru"
        },
        "episodic_memory": {
            "max_age_days": 90,
            "importance_threshold": 0.3,
            "cleanup_interval": 3600  # 1 hour
        }
    },
    "optimization": {
        "enable_compression": True,
        "compression_threshold": 1000,  # Compress after 1000 characters
        "enable_indexing": True,
        "index_rebuild_interval": 86400  # 24 hours
    }
}

Query Pattern Optimization

class QueryOptimizer:
    def __init__(self, memory):
        self.memory = memory
        self.query_patterns = {}
    
    def optimize_query(self, query, user_context=None):
        # Analyze query pattern
        pattern = self.analyze_query_pattern(query)
        
        # Apply optimizations based on pattern
        if pattern == "semantic_search":
            return self.optimize_semantic_search(query)
        elif pattern == "temporal_search":
            return self.optimize_temporal_search(query, user_context)
        elif pattern == "relationship_search":
            return self.optimize_relationship_search(query)
        
        return query
    
    def optimize_semantic_search(self, query):
        # Use semantic-specific optimizations
        return {
            "query": query,
            "memory_type": "semantic",
            "use_vector_index": True,
            "embedding_cache": True
        }
    
    def optimize_temporal_search(self, query, user_context):
        # Add temporal constraints
        time_range = self.infer_time_range(query, user_context)
        return {
            "query": query,
            "memory_type": "episodic",
            "time_range": time_range,
            "use_temporal_index": True
        }

Scaling Strategies

Horizontal Scaling

# Distributed SmartMemory setup
class DistributedSmartMemory:
    def __init__(self, shard_config):
        self.shards = {}
        self.load_balancer = LoadBalancer()
        
        for shard_id, config in shard_config.items():
            self.shards[shard_id] = SmartMemory(
                config=config,
                shard_id=shard_id
            )
    
    def route_query(self, query, user_id=None):
        # Route based on user_id or content hash
        if user_id:
            shard_id = self.hash_user_to_shard(user_id)
        else:
            shard_id = self.hash_content_to_shard(query)
        
        return self.shards[shard_id]
    
    async def distributed_search(self, query, user_id=None):
        if user_id:
            # Search specific user shard
            shard = self.route_query(query, user_id)
            return await shard.search_async(query)
        else:
            # Search all shards and merge results
            tasks = [
                shard.search_async(query)
                for shard in self.shards.values()
            ]
            
            all_results = await asyncio.gather(*tasks)
            return self.merge_distributed_results(all_results)

Vertical Scaling

# High-performance single-instance configuration
high_performance_config = {
    "graph_backend": {
        "type": "falkordb",
        "connection_pool_size": 50,
        "query_cache_size": "1GB",
        "memory_mapping": True
    },
    "vector": {
        "backend": "falkordb",
        "metric": "cosine",
        "hnsw_m": 32,                # higher connectivity for max recall
        "hnsw_ef_construction": 400, # broader build-time search
        "hnsw_ef_runtime": 128       # broader query-time search
    },
    "background_processing": {
        "max_workers": 16,
        "queue_size": 50000,
        "batch_size": 500
    },
    "caching": {
        "memory_limit": "8GB",
        "enable_compression": True,
        "compression_algorithm": "lz4"
    }
}

Performance Testing

Benchmarking Framework

import smartmemory.utils
import time
import statistics
from concurrent.futures import ThreadPoolExecutor


class PerformanceBenchmark:
    def __init__(self, memory):
        self.memory = memory
        self.results = {}

    def benchmark_search_performance(self, queries, iterations=100):
        """Benchmark search performance with multiple queries."""
        times = []

        for _ in range(iterations):
            start_time = time.time()

            for query in queries:
                results = self.memory.search(query, top_k=10)

            end_time = time.time()
            times.append(end_time - start_time)

        return {
            "avg_time": statistics.mean(times),
            "median_time": statistics.median(times),
            "min_time": min(times),
            "max_time": max(times),
            "std_dev": statistics.stdev(times)
        }

    def benchmark_concurrent_access(self, query, concurrent_users=10):
        """Benchmark concurrent access performance."""

        def search_worker():
            start_time = time.time()
            results = self.memory.search(query)
            end_time = time.time()
            return end_time - start_time

        with ThreadPoolExecutor(max_workers=concurrent_users) as executor:
            futures = [
                executor.submit(search_worker)
                for _ in range(concurrent_users)
            ]

            times = [future.result() for future in futures]

        return {
            "concurrent_users": concurrent_users,
            "avg_response_time": statistics.mean(times),
            "max_response_time": max(times),
            "throughput": concurrent_users / max(times)
        }

    def run_full_benchmark(self):
        """Run comprehensive performance benchmark."""
        test_queries = [