diff options
| author | Dhravya Shah <[email protected]> | 2026-01-18 16:55:32 -0800 |
|---|---|---|
| committer | GitHub <[email protected]> | 2026-01-18 16:55:32 -0800 |
| commit | 87b361c26bf5fc16049cd2727825891aa14b8e8b (patch) | |
| tree | c2f9f4f6223a7c1734578b772a16490ba2eb8b96 /apps/docs/migration/mem0-migration-script.py | |
| parent | Add Claude Code GitHub Workflow (#681) (diff) | |
| download | supermemory-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.py | 193 |
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() |