Build OpenClaw Memory on DGX Spark - Local AI Tutorial
Build a 4-layer memory system for OpenClaw on NVIDIA DGX Spark. Local embeddings, vector search, zero API costs. Complete implementation guide.
Originally published:
Introduction
Running advanced AI agents with contextual memory traditionally requires expensive cloud API calls and external services. This tutorial demonstrates how to build a complete four-layer memory system for OpenClaw that runs entirely on an NVIDIA DGX Spark workstation, eliminating ongoing API costs while maintaining full data privacy and control.
OpenClaw is an open-source AI agent framework that benefits significantly from persistent memory capabilities. By implementing local memory on DGX Spark hardware, you gain sub-millisecond retrieval times, unlimited context storage, and complete independence from third-party services. This approach is particularly valuable for enterprise deployments, research environments, and privacy-sensitive applications.
Learning Objectives
By completing this tutorial, you will:
- Understand the four-layer memory architecture and when to use each layer
- Configure DGX Spark hardware for optimal memory operations
- Implement vector embeddings using local models instead of API services
- Build efficient retrieval systems with FAISS or similar vector databases
- Integrate the memory system with OpenClaw agents
- Optimize performance for real-time conversational AI
- Monitor memory usage and implement cleanup strategies
Prerequisites
Hardware Requirements
- NVIDIA DGX Spark workstation (or compatible system with NVIDIA A100/H100 GPUs)
- Minimum 80GB GPU memory recommended for embedding models
- 256GB+ system RAM for vector index caching
- 2TB+ NVMe storage for persistent memory databases
Software Requirements
- Ubuntu 22.04 LTS or later
- NVIDIA Driver 525+ and CUDA 12.0+
- Docker and NVIDIA Container Toolkit
- Python 3.10+ with pip and virtualenv
- Git for cloning repositories
Knowledge Requirements
- Intermediate Python programming skills
- Basic understanding of vector embeddings and similarity search
- Familiarity with OpenClaw Docker: Self-Hosted AI Assistant containerization
- Command-line proficiency in Linux environments
Step-by-Step Implementation Guide
Step 1: Environment Setup and Dependencies
Begin by creating an isolated Python environment and installing core dependencies. This ensures reproducibility and prevents conflicts with system packages.
# Create project directory
mkdir openclaw-memory && cd openclaw-memory
Set up Python virtual environment
python3 -m venv venv
source venv/bin/activate
Install core dependencies
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
pip install transformers sentence-transformers faiss-gpu
pip install chromadb sqlalchemy redis
pip install openclaw
Verify GPU availability and CUDA functionality:
python -c "import torch; print(f'CUDA available: {torch.cuda.is_available()}'); print(f'GPU count: {torch.cuda.device_count()}')"Step 2: Configure Local Embedding Model
The foundation of local memory is running embedding models directly on your hardware. We'll use sentence-transformers with a high-quality multilingual model optimized for semantic similarity.
from sentence_transformers import SentenceTransformer
import torch
class LocalEmbedder:
def init(self, model_name='all-mpnet-base-v2', device='cuda'):
self.device = device
self.model = SentenceTransformer(model_name)
self.model.to(device)
def embed(self, texts, batch_size=32):
"""Generate embeddings for text inputs"""
with torch.no_grad():
embeddings = self.model.encode(
texts,
batch_size=batch_size,
show_progress_bar=True,
convert_to_tensor=True,
device=self.device
)
return embeddings.cpu().numpy()
Initialize embedder
embedder = LocalEmbedder()
test_embedding = embedder.embed(["Hello, world!"])
print(f"Embedding dimension: {test_embedding.shape[1]}")
This approach eliminates dependency on OpenAI or other API-based embedding services, reducing latency from hundreds of milliseconds to under 10ms for batch operations.
Step 3: Implement Layer 1 - Working Memory
Working memory stores the current conversation context and immediate agent state. We'll use Redis for fast in-memory operations with optional persistence.
import redis
import json
from datetime import datetime, timedelta
class WorkingMemory:
def init(self, host='localhost', port=6379, ttl_seconds=3600):
self.redis = redis.Redis(host=host, port=port, decode_responses=True)
self.ttl = ttl_seconds
def store_context(self, session_id, context_data):
"""Store current conversation context"""
key = f"working:{session_id}"
self.redis.setex(
key,
self.ttl,
json.dumps({
'data': context_data,
'timestamp': datetime.now().isoformat()
})
)
def get_context(self, session_id):
"""Retrieve active context"""
key = f"working:{session_id}"
data = self.redis.get(key)
return json.loads(data) if data else None
def append_message(self, session_id, role, content):
"""Add message to conversation history"""
key = f"messages:{session_id}"
message = {
'role': role,
'content': content,
'timestamp': datetime.now().isoformat()
}
self.redis.rpush(key, json.dumps(message))
self.redis.expire(key, self.ttl)
def get_recent_messages(self, session_id, count=10):
"""Retrieve recent conversation messages"""
key = f"messages:{session_id}"
messages = self.redis.lrange(key, -count, -1)
return [json.loads(msg) for msg in messages]</code></pre><h3>Step 4: Build Layer 2 - Episodic Memory</h3><p>Episodic memory captures significant conversation events and decision points. This layer uses SQLite for structured storage with full-text search capabilities.</p><pre><code>from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
Base = declarative_base()
class Episode(Base):
tablename = 'episodes'
id = Column(Integer, primary_key=True)
session_id = Column(String(255), index=True)
timestamp = Column(DateTime, default=datetime.now, index=True)
event_type = Column(String(50), index=True)
summary = Column(Text)
importance = Column(Float, default=0.5)
metadata = Column(Text) # JSON-serialized additional data
class EpisodicMemory:
def init(self, db_path='episodic_memory.db'):
self.engine = create_engine(f'sqlite:///{db_path}')
Base.metadata.create_all(self.engine)
Session = sessionmaker(bind=self.engine)
self.session = Session()
def store_episode(self, session_id, event_type, summary, importance=0.5, metadata=None):
"""Record a significant event or decision"""
episode = Episode(
session_id=session_id,
event_type=event_type,
summary=summary,
importance=importance,
metadata=json.dumps(metadata or {})
)
self.session.add(episode)
self.session.commit()
return episode.id
def retrieve_episodes(self, session_id=None, event_type=None, min_importance=0.0, limit=50):
"""Query episodes with filters"""
query = self.session.query(Episode)
if session_id:
query = query.filter(Episode.session_id == session_id)
if event_type:
query = query.filter(Episode.event_type == event_type)
query = query.filter(Episode.importance >= min_importance)
query = query.order_by(Episode.timestamp.desc()).limit(limit)
return query.all()</code></pre><h3>Step 5: Create Layer 3 - Semantic Long-Term Memory</h3><p>This layer provides vector-based retrieval of relevant information from past conversations and knowledge. We'll use FAISS for GPU-accelerated similarity search.</p><pre><code>import faiss
import numpy as np
import pickle
from pathlib import Path
class SemanticMemory:
def init(self, dimension=768, index_path='semantic_index'):
self.dimension = dimension
self.index_path = Path(index_path)
self.index_path.mkdir(exist_ok=True)
# Initialize FAISS index with GPU support
self.cpu_index = faiss.IndexFlatL2(dimension)
self.gpu_index = faiss.index_cpu_to_all_gpus(self.cpu_index)
# Store metadata alongside vectors
self.metadata = []
self.id_counter = 0
self._load_index()
def add_memories(self, texts, embeddings, metadata_list):
"""Add new memories with embeddings"""
embeddings_np = np.array(embeddings).astype('float32')
# Normalize vectors for cosine similarity
faiss.normalize_L2(embeddings_np)
self.gpu_index.add(embeddings_np)
for text, meta in zip(texts, metadata_list):
self.metadata.append({
'id': self.id_counter,
'text': text,
'metadata': meta,
'timestamp': datetime.now().isoformat()
})
self.id_counter += 1
self._save_index()
def search(self, query_embedding, k=5, score_threshold=0.7):
"""Retrieve most similar memories"""
query_np = np.array([query_embedding]).astype('float32')
faiss.normalize_L2(query_np)
distances, indices = self.gpu_index.search(query_np, k)
results = []
for dist, idx in zip(distances[0], indices[0]):
if idx < len(self.metadata):
# Convert L2 distance to similarity score
similarity = 1 / (1 + dist)
if similarity >= score_threshold:
result = self.metadata[idx].copy()
result['similarity'] = float(similarity)
results.append(result)
return results
def _save_index(self):
"""Persist index and metadata to disk"""
faiss.write_index(self.cpu_index, str(self.index_path / 'faiss.index'))
with open(self.index_path / 'metadata.pkl', 'wb') as f:
pickle.dump({
'metadata': self.metadata,
'id_counter': self.id_counter
}, f)
def _load_index(self):
"""Load existing index from disk"""
index_file = self.index_path / 'faiss.index'
meta_file = self.index_path / 'metadata.pkl'
if index_file.exists() and meta_file.exists():
self.cpu_index = faiss.read_index(str(index_file))
self.gpu_index = faiss.index_cpu_to_all_gpus(self.cpu_index)
with open(meta_file, 'rb') as f:
data = pickle.load(f)
self.metadata = data['metadata']
self.id_counter = data['id_counter']</code></pre><h3>Step 6: Implement Layer 4 - Procedural Memory</h3><p>Procedural memory stores learned behaviors, successful interaction patterns, and reusable workflows. This layer tracks what strategies work best in different contexts.</p><pre><code>class ProceduralMemory:
def __init__(self, db_path='procedural_memory.db'):
self.engine = create_engine(f'sqlite:///{db_path}')
self.session = sessionmaker(bind=self.engine)()
self._init_schema()
def _init_schema(self):
"""Create tables for procedures and performance metrics"""
self.engine.execute('''
CREATE TABLE IF NOT EXISTS procedures (
id INTEGER PRIMARY KEY,
name TEXT UNIQUE,
description TEXT,
context_pattern TEXT,
action_sequence TEXT,
success_count INTEGER DEFAULT 0,
failure_count INTEGER DEFAULT 0,
avg_execution_time REAL DEFAULT 0.0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_used TIMESTAMP
)
''')
def record_procedure(self, name, context, actions, success=True, execution_time=0.0):
"""Log procedure execution and update performance metrics"""
# Check if procedure exists
result = self.session.execute(
"SELECT * FROM procedures WHERE name = ?", (name,)
).fetchone()
if result:
# Update existing procedure
success_col = 'success_count' if success else 'failure_count'
self.session.execute(f'''
UPDATE procedures
SET {success_col} = {success_col} + 1,
avg_execution_time = (avg_execution_time + ?) / 2,
last_used = CURRENT_TIMESTAMP
WHERE name = ?
''', (execution_time, name))
else:
# Create new procedure
self.session.execute('''
INSERT INTO procedures (name, context_pattern, action_sequence, success_count, failure_count, avg_execution_time, last_used)
VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
''', (name, json.dumps(context), json.dumps(actions), 1 if success else 0, 0 if success else 1, execution_time))
self.session.commit()
def get_best_procedure(self, context_embedding, embedder, k=3):
"""Find most successful procedure for similar context"""
procedures = self.session.execute(
"SELECT * FROM procedures WHERE success_count > 0 ORDER BY (success_count * 1.0 / (success_count + failure_count)) DESC LIMIT ?",
(k,)
).fetchall()
# Rank by context similarity and success rate
scored_procedures = []
for proc in procedures:
context_pattern = json.loads(proc['context_pattern'])
# Calculate context similarity using embeddings
success_rate = proc['success_count'] / (proc['success_count'] + proc['failure_count'])
scored_procedures.append({
'name': proc['name'],
'actions': json.loads(proc['action_sequence']),
'success_rate': success_rate,
'avg_time': proc['avg_execution_time']
})
return scored_procedures</code></pre><h3>Step 7: Integrate Memory System with OpenClaw</h3><p>Now we'll create a unified memory manager that coordinates all four layers and integrates with OpenClaw agents.</p><pre><code>class OpenClawMemorySystem:
def __init__(self, config=None):
self.embedder = LocalEmbedder()
self.working = WorkingMemory()
self.episodic = EpisodicMemory()
self.semantic = SemanticMemory()
self.procedural = ProceduralMemory()
def process_interaction(self, session_id, user_input, agent_response):
"""Process a complete interaction through all memory layers"""
# Layer 1: Update working memory
self.working.append_message(session_id, 'user', user_input)
self.working.append_message(session_id, 'assistant', agent_response)
# Generate embeddings for semantic storage
combined_text = f"User: {user_input}\nAssistant: {agent_response}"
embedding = self.embedder.embed([combined_text])[0]
# Layer 3: Add to semantic memory
self.semantic.add_memories(
[combined_text],
[embedding],
[{'session_id': session_id, 'type': 'interaction'}]
)
# Layer 2: Create episode if significant
importance = self._assess_importance(user_input, agent_response)
if importance > 0.6:
self.episodic.store_episode(
session_id=session_id,
event_type='significant_interaction',
summary=combined_text[:200],
importance=importance
)
def retrieve_context(self, session_id, query, max_items=5):
"""Retrieve relevant context from all memory layers"""
context = {
'working': self.working.get_recent_messages(session_id, count=10),
'episodic': [],
'semantic': [],
'procedural': []
}
# Get relevant episodic memories
episodes = self.episodic.retrieve_episodes(session_id=session_id, limit=max_items)
context['episodic'] = [{
'summary': ep.summary,
'importance': ep.importance,
'timestamp': ep.timestamp.isoformat()
} for ep in episodes]
# Search semantic memory
query_embedding = self.embedder.embed([query])[0]
semantic_results = self.semantic.search(query_embedding, k=max_items)
context['semantic'] = semantic_results
return context
def _assess_importance(self, user_input, agent_response):
"""Heuristic to determine interaction importance"""
# Simple keyword-based assessment (can be enhanced with ML)
important_keywords = ['important', 'remember', 'critical', 'always', 'never', 'error', 'problem']
text = (user_input + ' ' + agent_response).lower()
score = 0.3 # baseline
for keyword in important_keywords:
if keyword in text:
score += 0.2
return min(score, 1.0)</code></pre><h3>Step 8: Performance Optimization</h3><p>Optimize the memory system for production workloads with batch processing and caching strategies.</p><pre><code>class OptimizedMemorySystem(OpenClawMemorySystem):
def __init__(self, config=None):
super().__init__(config)
self.embedding_cache = {}
self.batch_buffer = []
self.batch_size = 32
def batch_process_interactions(self, interactions):
"""Process multiple interactions efficiently"""
# Collect all texts for batch embedding
texts = [f"User: {i['user']}\nAssistant: {i['agent']}" for i in interactions]
# Generate embeddings in single batch
embeddings = self.embedder.embed(texts, batch_size=self.batch_size)
# Process each interaction with pre-computed embeddings
for interaction, embedding in zip(interactions, embeddings):
self._process_with_embedding(
interaction['session_id'],
interaction['user'],
interaction['agent'],
embedding
)
def _process_with_embedding(self, session_id, user_input, agent_response, embedding):
"""Process interaction with pre-computed embedding"""
combined_text = f"User: {user_input}\nAssistant: {agent_response}"
self.working.append_message(session_id, 'user', user_input)
self.working.append_message(session_id, 'assistant', agent_response)
self.semantic.add_memories(
[combined_text],
[embedding],
[{'session_id': session_id}]
)
importance = self._assess_importance(user_input, agent_response)
if importance > 0.6:
self.episodic.store_episode(
session_id, 'interaction', combined_text[:200], importance
)</code></pre><h2>Troubleshooting Common Issues</h2><h3>GPU Memory Overflow</h3><p>If you encounter CUDA out-of-memory errors when processing large batches, reduce the batch size or use a smaller embedding model. Monitor GPU memory with <code>nvidia-smi</code> and adjust accordingly.</p><pre><code># Reduce batch size
embedder = LocalEmbedder()
embeddings = embedder.embed(texts, batch_size=16) # Instead of 32
Or use a smaller model
embedder = LocalEmbedder(model_name='all-MiniLM-L6-v2') # 384-dim instead of 768
Slow Vector Search Performance
FAISS indexing performance degrades with extremely large datasets. For production systems with millions of vectors, consider using IVF (Inverted File) indexes with product quantization.
# Create IVF index for faster search
quantizer = faiss.IndexFlatL2(dimension)
index = faiss.IndexIVFPQ(quantizer, dimension, 100, 8, 8)
index.train(training_vectors) # Train on representative sample
index.add(vectors)
index.nprobe = 10 # Search 10 cells for better recallRedis Connection Failures
Ensure Redis is running and configured for persistence. For production deployments, enable AOF (Append-Only File) or RDB snapshots.
# Start Redis with persistence
redis-server --appendonly yes --save 60 1000
Verify connection
redis-cli ping # Should return PONG
SQLite Database Locking
Under high concurrency, SQLite may experience locking issues. Enable WAL (Write-Ahead Logging) mode for better concurrent access.
engine = create_engine('sqlite:///memory.db', connect_args={'check_same_thread': False})
with engine.connect() as conn:
conn.execute('PRAGMA journal_mode=WAL')Best Practices
Memory Hygiene
Implement regular cleanup of stale working memory entries and low-importance episodic memories. This prevents unbounded growth and maintains system performance.
def cleanup_old_memories(memory_system, days_threshold=30):
"""Remove memories older than threshold"""
cutoff_date = datetime.now() - timedelta(days=days_threshold)
# Clean episodic memories with low importance
memory_system.episodic.session.execute(
"DELETE FROM episodes WHERE importance < 0.3 AND timestamp < ?",
(cutoff_date,)
)
# Archive old semantic memories to cold storage
# Implementation depends on archival strategy</code></pre><h3>Embedding Model Selection</h3><p>Choose embedding models based on your language requirements and performance constraints. Multilingual models (like paraphrase-multilingual) support 50+ languages but are larger. Domain-specific models may provide better accuracy for specialized applications.</p><h3>Monitoring and Observability</h3><p>Implement comprehensive logging and metrics collection to track memory system performance and identify bottlenecks.</p><pre><code>import logging
from prometheus_client import Counter, Histogram
Define metrics
embedding_time = Histogram('embedding_duration_seconds', 'Time to generate embeddings')
search_time = Histogram('semantic_search_duration_seconds', 'Time for semantic search')
class InstrumentedMemorySystem(OptimizedMemorySystem):
def retrieve_context(self, session_id, query, max_items=5):
with search_time.time():
return super().retrieve_context(session_id, query, max_items)
Backup and Disaster Recovery
Regularly backup FAISS indexes, SQLite databases, and Redis snapshots to prevent data loss. Implement versioned backups with rotation policies.
#!/bin/bash
Daily backup script
BACKUP_DIR="/backups/$(date +%Y%m%d)"
mkdir -p $BACKUP_DIR
Backup databases
cp episodic_memory.db $BACKUP_DIR/
cp procedural_memory.db $BACKUP_DIR/
Backup FAISS index
cp -r semantic_index $BACKUP_DIR/
Backup Redis
redis-cli BGSAVE
cp /var/lib/redis/dump.rdb $BACKUP_DIR/
Performance Benchmarks
On NVIDIA DGX Spark with A100 GPUs, expect these approximate performance characteristics:
- Embedding generation: ~5ms per text (batch of 32), ~1000 embeddings/second sustained
- Semantic search: Sub-millisecond for top-5 retrieval from 100K vectors
- Working memory operations: <1ms for Redis read/write
- Episodic queries: 5-10ms for complex SQLite queries with indexes
These numbers represent significant improvements over cloud API-based systems, which typically incur 100-300ms latency per request plus network overhead.
Next Steps and Advanced Topics
After completing this basic implementation, consider these enhancements:
Implement Memory Consolidation
Build a background process that periodically consolidates working memory into episodic and semantic storage, similar to human memory consolidation during sleep. This involves identifying patterns, merging similar memories, and strengthening important connections.
Add Cross-Session Learning
Extend procedural memory to learn from patterns across multiple users or sessions, enabling the agent to improve its strategies based on collective experience while respecting privacy boundaries.
Integrate with langchain
Connect your memory system to LangChain for enhanced agent capabilities, allowing sophisticated retrieval-augmented generation patterns and tool use with persistent context.
Implement Forgetting Mechanisms
Add intelligent forgetting to prevent information overload, using importance scoring and access patterns to determine what memories should fade over time, similar to human memory decay.
Multi-Agent Memory Sharing
For collaborative agent systems, implement secure memory sharing protocols that allow agents to access relevant memories from other agents while maintaining appropriate access controls.
Conclusion
Building local memory on DGX Spark provides OpenClaw agents with enterprise-grade contextual awareness without ongoing API costs or privacy concerns. This four-layer architecture—working, episodic, semantic, and procedural memory—mirrors cognitive science principles while leveraging modern GPU acceleration for real-time performance.
The system you've built is production-ready for most applications and can scale to millions of memories with appropriate indexing strategies. By running entirely on local hardware, you maintain complete control over your data while achieving superior performance compared to cloud-based alternatives.
As you deploy this system, monitor performance metrics closely and iterate on the importance scoring algorithms to match your specific use cases. The flexibility of local deployment allows rapid experimentation and optimization that would be costly or impossible with managed services.
Original Source
https://www.youtube.com/watch?v=XFtqEbfFyMc
Last updated: