aboutsummaryrefslogtreecommitdiff
path: root/apps/docs/migration/mem0-migration-script.py
diff options
context:
space:
mode:
authorDhravya Shah <[email protected]>2026-01-18 16:55:32 -0800
committerGitHub <[email protected]>2026-01-18 16:55:32 -0800
commit87b361c26bf5fc16049cd2727825891aa14b8e8b (patch)
treec2f9f4f6223a7c1734578b772a16490ba2eb8b96 /apps/docs/migration/mem0-migration-script.py
parentAdd Claude Code GitHub Workflow (#681) (diff)
downloadsupermemory-87b361c26bf5fc16049cd2727825891aa14b8e8b.tar.xz
supermemory-87b361c26bf5fc16049cd2727825891aa14b8e8b.zip
docs changes (#678)
Co-authored-by: Claude Opus 4.5 <[email protected]>
Diffstat (limited to 'apps/docs/migration/mem0-migration-script.py')
-rw-r--r--apps/docs/migration/mem0-migration-script.py193
1 files changed, 96 insertions, 97 deletions
diff --git a/apps/docs/migration/mem0-migration-script.py b/apps/docs/migration/mem0-migration-script.py
index 7b05edf6..ff33f10a 100644
--- a/apps/docs/migration/mem0-migration-script.py
+++ b/apps/docs/migration/mem0-migration-script.py
@@ -1,8 +1,8 @@
#!/usr/bin/env python3
"""
-Mem0.ai to Supermemory Migration Script
+Mem0 to Supermemory Migration Script
========================================
-Simple script to migrate memories from Mem0.ai to Supermemory.
+Simple script to migrate memories from Mem0 to Supermemory.
Prerequisites:
1. Install required packages:
@@ -18,36 +18,34 @@ Usage:
python mem0-migration-script.py
"""
-import os
import json
+import os
import time
from datetime import datetime
-from typing import Dict, Any, Optional
+from typing import Any, Dict, Optional
+
+from dotenv import load_dotenv
from mem0 import MemoryClient
from supermemory import Supermemory
-from dotenv import load_dotenv
# Load environment variables
load_dotenv()
+
def export_from_mem0(
api_key: str,
org_id: Optional[str] = None,
project_id: Optional[str] = None,
- filters: Optional[Dict] = None
+ filters: Optional[Dict] = None,
) -> Dict[str, Any]:
"""
- Export memories from Mem0.ai using their export API
+ Export memories from mem0 using their export API
"""
- print("šŸ”„ Starting Mem0.ai export...")
-
+ print("šŸ”„ Starting mem0 export...")
+
# Initialize Mem0 client
- client = MemoryClient(
- api_key=api_key,
- org_id=org_id,
- project_id=project_id
- )
-
+ client = MemoryClient(api_key=api_key, org_id=org_id, project_id=project_id)
+
# Define export schema - this matches what Mem0 actually returns
export_schema = {
"type": "object",
@@ -65,41 +63,40 @@ def export_from_mem0(
"run_id": {"type": "string"},
"metadata": {"type": "object"},
"created_at": {"type": "string"},
- "updated_at": {"type": "string"}
- }
- }
+ "updated_at": {"type": "string"},
+ },
+ },
}
- }
+ },
}
-
+
try:
# Step 1: Create export job
print("šŸ“¤ Creating export job...")
export_response = client.create_memory_export(
- schema=export_schema,
- filters=filters if filters else {}
+ schema=export_schema, filters=filters if filters else {}
)
-
+
export_id = export_response.get("id")
print(f"āœ… Export job created with ID: {export_id}")
-
+
# Step 2: Wait for export to complete
print("ā³ Waiting for export to complete...")
time.sleep(5) # Usually takes a few seconds
-
+
# Step 3: Retrieve the exported data using the correct method
print("šŸ“„ Retrieving exported data...")
export_data = client.get_memory_export(memory_export_id=export_id)
-
+
# Step 4: Save backup
backup_filename = f"mem0_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(backup_filename, "w") as f:
json.dump(export_data, f, indent=2)
print(f"šŸ’¾ Backup saved to: {backup_filename}")
-
+
memory_count = len(export_data.get("memories", []))
- print(f"āœ… Successfully exported {memory_count} memories from Mem0.ai")
-
+ print(f"āœ… Successfully exported {memory_count} memories from mem0")
+
# Show sample of exported data
if memory_count > 0:
print("\nšŸ“‹ Sample exported memory:")
@@ -107,36 +104,33 @@ def export_from_mem0(
print(f" Content: {sample.get('content', 'N/A')[:50]}...")
print(f" ID: {sample.get('id', 'None')}")
print(f" User ID: {sample.get('user_id', 'None')}")
-
+
return export_data
-
+
except Exception as e:
print(f"āŒ Error exporting from Mem0: {str(e)}")
raise
+
def import_to_supermemory(mem0_data: Dict[str, Any], api_key: str) -> Dict[str, int]:
"""
Import Mem0 memories into Supermemory
"""
print("\nšŸš€ Starting import to Supermemory...")
-
+
# Initialize Supermemory client
client = Supermemory(api_key=api_key)
-
+
memories = mem0_data.get("memories", [])
if not memories:
print("āš ļø No memories found to import")
return {"imported": 0, "failed": 0, "skipped": 0}
-
+
# Statistics
- stats = {
- "imported": 0,
- "failed": 0,
- "skipped": 0
- }
-
+ stats = {"imported": 0, "failed": 0, "skipped": 0}
+
print(f"šŸ“¦ Processing {len(memories)} memories...")
-
+
for i, memory in enumerate(memories, 1):
try:
# Check if content exists
@@ -145,174 +139,178 @@ def import_to_supermemory(mem0_data: Dict[str, Any], api_key: str) -> Dict[str,
print(f"āš ļø [{i}/{len(memories)}] Skipping: No content")
stats["skipped"] += 1
continue
-
+
# Build container tags
container_tags = ["imported_from_mem0"]
-
+
# Add user tag if present (handle None values)
user_id = memory.get("user_id")
if user_id and user_id != "None":
container_tags.append(f"user_{user_id}")
-
+
# Add agent tag if present
agent_id = memory.get("agent_id")
if agent_id and agent_id != "None":
container_tags.append(f"agent_{agent_id}")
-
+
# Add app tag if present
app_id = memory.get("app_id")
if app_id and app_id != "None":
container_tags.append(f"app_{app_id}")
-
+
# Add session tag if present
session_id = memory.get("session_id")
if session_id and session_id != "None":
container_tags.append(f"session_{session_id}")
-
+
# Generate a unique ID if Mem0 didn't provide one
memory_id = memory.get("id")
if not memory_id or memory_id == "None":
# Use content hash for uniqueness
import hashlib
+
memory_id = hashlib.md5(content.encode()).hexdigest()[:8]
-
+
# Prepare metadata
metadata = {
"source": "mem0_migration",
- "migration_date": datetime.now().isoformat()
+ "migration_date": datetime.now().isoformat(),
}
-
+
# Add original ID if it existed
if memory.get("id") and memory["id"] != "None":
metadata["original_id"] = memory["id"]
-
+
# Add timestamps if available and not None
created_at = memory.get("created_at")
if created_at and created_at != "None":
metadata["original_created_at"] = created_at
-
+
updated_at = memory.get("updated_at")
if updated_at and updated_at != "None":
metadata["original_updated_at"] = updated_at
-
+
# Add hash information if available
hash_val = memory.get("hash")
if hash_val and hash_val != "None":
metadata["original_hash"] = hash_val
-
+
prev_hash = memory.get("prev_hash")
if prev_hash and prev_hash != "None":
metadata["original_prev_hash"] = prev_hash
-
+
# Merge with existing metadata if it's a valid dict
if memory.get("metadata") and isinstance(memory["metadata"], dict):
metadata.update(memory["metadata"])
-
+
# Import to Supermemory
- result = client.memories.add(
+ result = client.add(
content=content,
container_tags=container_tags,
custom_id=f"mem0_{memory_id}",
- metadata=metadata
+ metadata=metadata,
)
-
+
stats["imported"] += 1
print(f"āœ… [{i}/{len(memories)}] Imported: {content[:50]}...")
-
+
# Small delay to avoid rate limiting
if i % 10 == 0:
time.sleep(0.5)
-
+
except Exception as e:
stats["failed"] += 1
print(f"āŒ [{i}/{len(memories)}] Failed: {str(e)}")
-
+
return stats
+
def verify_migration(api_key: str, expected_count: int):
"""
Verify that memories were imported correctly
"""
print("\nšŸ” Verifying migration...")
-
+
client = Supermemory(api_key=api_key)
-
+
try:
# Check imported memories
- result = client.memories.list(
- container_tags=["imported_from_mem0"],
- limit=100
- )
-
- total_imported = result['pagination']['totalItems']
+ result = client.documents.list(container_tags=["imported_from_mem0"], limit=100)
+
+ total_imported = result["pagination"]["totalItems"]
print(f"āœ… Found {total_imported} imported memories in Supermemory")
-
+
# Show sample memories
- if result['memories']:
+ if result["memories"]:
print("\nšŸ“‹ Sample imported memories:")
- for memory in result['memories'][:3]:
- print(f" - {memory['id']}: {memory.get('summary', 'No summary')[:50]}...")
-
+ for memory in result["memories"][:3]:
+ print(
+ f" - {memory['id']}: {memory.get('summary', 'No summary')[:50]}..."
+ )
+
# Check success rate
- success_rate = (total_imported / expected_count * 100) if expected_count > 0 else 0
+ success_rate = (
+ (total_imported / expected_count * 100) if expected_count > 0 else 0
+ )
print(f"\nšŸ“Š Migration success rate: {success_rate:.1f}%")
-
+
return total_imported
-
+
except Exception as e:
print(f"āŒ Error during verification: {str(e)}")
return 0
+
def main():
"""Main migration function"""
print("=" * 60)
- print("šŸŽÆ Mem0.ai to Supermemory Migration Tool")
+ print("šŸŽÆ mem0 to Supermemory Migration Tool")
print("=" * 60)
-
+
# Get credentials from environment
mem0_api_key = os.getenv("MEM0_API_KEY")
mem0_org_id = os.getenv("MEM0_ORG_ID")
mem0_project_id = os.getenv("MEM0_PROJECT_ID")
supermemory_api_key = os.getenv("SUPERMEMORY_API_KEY")
-
+
# Validate credentials
if not mem0_api_key:
print("āŒ Error: MEM0_API_KEY environment variable not set")
return
-
+
if not supermemory_api_key:
print("āŒ Error: SUPERMEMORY_API_KEY environment variable not set")
return
-
+
try:
# Step 1: Export from Mem0
- print("\nšŸ“¤ STEP 1: Export from Mem0.ai")
+ print("\nšŸ“¤ STEP 1: Export from mem0")
print("-" * 40)
-
+
# You can add filters here if needed
# Example: filters = {"AND": [{"user_id": "specific_user"}]}
filters = None
-
+
mem0_data = export_from_mem0(
api_key=mem0_api_key,
org_id=mem0_org_id,
project_id=mem0_project_id,
- filters=filters
+ filters=filters,
)
-
+
# Step 2: Import to Supermemory
print("\nšŸ“„ STEP 2: Import to Supermemory")
print("-" * 40)
-
+
stats = import_to_supermemory(mem0_data, supermemory_api_key)
-
+
# Step 3: Verify migration
print("\nāœ”ļø STEP 3: Verification")
print("-" * 40)
-
+
expected_count = len(mem0_data.get("memories", []))
verify_migration(supermemory_api_key, expected_count)
-
+
# Final summary
print("\n" + "=" * 60)
print("šŸ“Š MIGRATION SUMMARY")
@@ -321,17 +319,18 @@ def main():
print(f"āœ… Successfully imported: {stats['imported']}")
print(f"āš ļø Skipped (no content): {stats['skipped']}")
print(f"āŒ Failed: {stats['failed']}")
-
- if stats['imported'] == expected_count - stats['skipped']:
+
+ if stats["imported"] == expected_count - stats["skipped"]:
print("\nšŸŽ‰ Migration completed successfully!")
- elif stats['imported'] > 0:
+ elif stats["imported"] > 0:
print("\nāš ļø Migration completed with some issues. Check the logs above.")
else:
print("\nāŒ Migration failed. Please check your credentials and try again.")
-
+
except Exception as e:
print(f"\nāŒ Migration error: {str(e)}")
print("Please check your credentials and network connection.")
+
if __name__ == "__main__":
- main() \ No newline at end of file
+ main()