41cd78207a
- Configure mem0 to use self-hosted Supabase instead of Qdrant for vector storage - Update docker-compose to connect containers to localai network - Install vecs library for Supabase pgvector integration - Create comprehensive test suite for Supabase + mem0 integration - Update documentation to reflect Supabase configuration - All containers now connected to shared localai network - Successful vector storage and retrieval tests completed 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
134 lines
4.0 KiB
Python
134 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test mem0 integration with self-hosted Supabase
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from dotenv import load_dotenv
|
|
from mem0 import Memory
|
|
from config import load_config, get_mem0_config
|
|
import tempfile
|
|
|
|
def test_supabase_mem0_integration():
|
|
"""Test mem0 with Supabase vector store"""
|
|
print("🧪 Testing mem0 with Supabase integration...")
|
|
|
|
# Load configuration
|
|
config = load_config()
|
|
|
|
if not config.database.supabase_url or not config.database.supabase_key:
|
|
print("❌ Supabase configuration not found")
|
|
return False
|
|
|
|
try:
|
|
# Get mem0 configuration for Supabase
|
|
mem0_config = get_mem0_config(config, "openai")
|
|
print(f"📋 Configuration: {mem0_config}")
|
|
|
|
# Create memory instance
|
|
m = Memory.from_config(mem0_config)
|
|
|
|
# Test basic operations
|
|
print("💾 Testing memory addition...")
|
|
messages = [
|
|
{"role": "user", "content": "I love programming in Python"},
|
|
{"role": "assistant", "content": "That's great! Python is an excellent language for development."}
|
|
]
|
|
|
|
result = m.add(messages, user_id="test_user_supabase")
|
|
print(f"✅ Memory added: {result}")
|
|
|
|
print("🔍 Testing memory search...")
|
|
search_results = m.search(query="Python programming", user_id="test_user_supabase")
|
|
print(f"✅ Search results: {search_results}")
|
|
|
|
print("📜 Testing memory retrieval...")
|
|
all_memories = m.get_all(user_id="test_user_supabase")
|
|
print(f"✅ Retrieved {len(all_memories)} memories")
|
|
|
|
# Cleanup
|
|
print("🧹 Cleaning up test data...")
|
|
for memory in all_memories:
|
|
if 'id' in memory:
|
|
m.delete(memory_id=memory['id'])
|
|
|
|
print("✅ Supabase integration test successful!")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Supabase integration test failed: {e}")
|
|
return False
|
|
|
|
def test_supabase_direct_connection():
|
|
"""Test direct Supabase connection"""
|
|
print("🔗 Testing direct Supabase connection...")
|
|
|
|
try:
|
|
import requests
|
|
|
|
config = load_config()
|
|
supabase_url = config.database.supabase_url
|
|
supabase_key = config.database.supabase_key
|
|
|
|
# Test REST API connection
|
|
headers = {
|
|
'apikey': supabase_key,
|
|
'Authorization': f'Bearer {supabase_key}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
# Test health endpoint
|
|
response = requests.get(f"{supabase_url}/rest/v1/", headers=headers, timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
print("✅ Supabase REST API is accessible")
|
|
return True
|
|
else:
|
|
print(f"❌ Supabase REST API returned status {response.status_code}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"❌ Direct Supabase connection failed: {e}")
|
|
return False
|
|
|
|
def main():
|
|
"""Main test function"""
|
|
print("=" * 60)
|
|
print("MEM0 + SUPABASE INTEGRATION TESTS")
|
|
print("=" * 60)
|
|
|
|
# Load environment
|
|
load_dotenv()
|
|
|
|
results = []
|
|
|
|
# Test 1: Direct Supabase connection
|
|
results.append(("Supabase Connection", test_supabase_direct_connection()))
|
|
|
|
# Test 2: mem0 + Supabase integration
|
|
results.append(("mem0 + Supabase Integration", test_supabase_mem0_integration()))
|
|
|
|
# Summary
|
|
print("\n" + "=" * 60)
|
|
print("TEST SUMMARY")
|
|
print("=" * 60)
|
|
|
|
passed = 0
|
|
for test_name, result in results:
|
|
status = "✅ PASS" if result else "❌ FAIL"
|
|
print(f"{status} {test_name}")
|
|
if result:
|
|
passed += 1
|
|
|
|
print(f"\nOverall: {passed}/{len(results)} tests passed")
|
|
|
|
if passed == len(results):
|
|
print("🎉 All Supabase integration tests passed!")
|
|
sys.exit(0)
|
|
else:
|
|
print("💥 Some tests failed - check configuration")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main() |