my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Agentic (LLM-guided) memory importer.
#3
#4	For providers without programmatic export APIs, this generates:
#5	1. A Python migration script the user can run in their environment
#6	2. Step-by-step instructions for manual extraction
#7	3. Instructions the user can give to their AI agent to perform the migration
#8
#9	This covers the "agent extraction" use case:
#10	- User tells their AI agent (Claude, ChatGPT, etc.) to export from Provider X
#11	- Agent follows Mnemosyne-provided instructions to produce a JSON file
#12	- User runs `hermes mnemosyne import --file export.json` to ingest
#13	"""
#14
#15	import json
#16	from datetime import datetime
#17	from typing import List, Dict, Optional, Any
#18	from pathlib import Path
#19
#20	from mnemosyne.core.importers.base import BaseImporter, ImporterResult
#21
#22
#23	# Known provider metadata for script generation
#24	PROVIDER_SCRIPTS = {
#25	"mem0": {
#26	"install": "pip install mem0ai",
#27	"api_import": "from mem0 import MemoryClient",
#28	"extract_code": """# Connect to Mem0
#29	client = MemoryClient(api_key="YOUR_API_KEY", host="https://api.mem0.ai")
#30
#31	# Extract all memories (paginated)
#32	all_memories = []
#33	page = 1
#34	while True:
#35	resp = client.get_all(filters={"user_id": "YOUR_USER_ID"}, page=page, page_size=200)
#36	results = resp.get("results", [])
#37	all_memories.extend(results)
#38	if resp.get("next") is None:
#39	break
#40	page += 1
#41
#42	# Save to JSON
#43	import json
#44	with open("mem0_export.json", "w") as f:
#45	json.dump(all_memories, f, indent=2)
#46
#47	print(f"Exported {len(all_memories)} memories to mem0_export.json")""",
#48	"env_hint": "MEM0_API_KEY",
#49	},
#50	"letta": {
#51	"install": "pip install letta-client",
#52	"api_import": "from letta_client import Letta",
#53	"extract_code": """client = Letta(api_key="YOUR_API_KEY")
#54
#55	# Export specific agent or list all
#56	agent_id = "YOUR_AGENT_ID" # or list agents: agents = client.agents.list()
#57	agent_file = client.agents.export_file(agent_id)
#58
#59	with open("letta_export.json", "w") as f:
#60	json.dump(agent_file, f, indent=2)
#61
#62	print(f"Exported agent to letta_export.json")""",
#63	"env_hint": "LETTA_API_KEY",
#64	},
#65	"zep": {
#66	"install": "pip install zep-cloud",
#67	"api_import": "from zep_cloud.client import Zep",
#68	"extract_code": """client = Zep(api_key="YOUR_API_KEY")
#69
#70	# List users
#71	users = client.user.list_ordered()
#72	all_data = []
#73
#74	for user in users.get("users", []):
#75	uid = user["user_id"]
#76	sessions = client.user.get_sessions(uid)
#77	for session in sessions:
#78	sid = session["session_id"]
#79	mem = client.memory.get(sid)
#80	all_data.append({
#81	"user_id": uid,
#82	"session_id": sid,
#83	"messages": mem.get("messages", []),
#84	"summary": mem.get("summary", ""),
#85	"facts": mem.get("relevant_facts", []),
#86	})
#87
#88	with open("zep_export.json", "w") as f:
#89	json.dump(all_data, f, indent=2)
#90
#91	print(f"Exported {len(all_data)} sessions to zep_export.json")""",
#92	"env_hint": "ZEP_API_KEY",
#93	},
#94	"cognee": {
#95	"install": "pip install cognee",
#96	"api_import": "import cognee\nimport asyncio",
#97	"extract_code": """async def extract():
#98	graph_data = await cognee.graph_db.get_graph_data()
#99	# graph_data is (nodes, edges) tuple
#100	nodes, edges = graph_data
#101
#102	export = {
#103	"nodes": [{"id": str(n[0]), "properties": n[1]} for n in nodes],
#104	"edges": [{"source": str(e[0]), "target": str(e[1]), "label": str(e[2]), "properties": e[3] if len(e) > 3 else {}} for e in edges],
#105	}
#106
#107	with open("cognee_export.json", "w") as f:
#108	json.dump(export, f, indent=2)
#109
#110	print(f"Exported {len(nodes)} nodes and {len(edges)} edges")
#111
#112	asyncio.run(extract())""",
#113	"env_hint": None,
#114	},
#115	"honcho": {
#116	"install": "pip install honcho-ai",
#117	"api_import": "from honcho import Honcho",
#118	"extract_code": """honcho = Honcho(workspace_id="YOUR_WORKSPACE_ID")
#119
#120	# List peers
#121	peers = honcho.list_peers()
#122	all_messages = []
#123
#124	for peer in peers.get("peers", []):
#125	peer_id = peer["peer_id"]
#126	sessions = honcho.list_sessions(peer_id=peer_id)
#127	for session in sessions.get("sessions", []):
#128	sid = session["session_id"]
#129	messages = honcho.session(sid).list_messages()
#130	for msg in messages.get("messages", []):
#131	all_messages.append({
#132	"peer_id": peer_id,
#133	"session_id": sid,
#134	"content": msg["content"],
#135	"role": "user",
#136	"timestamp": msg.get("created_at"),
#137	})
#138
#139	with open("honcho_export.json", "w") as f:
#140	json.dump(all_messages, f, indent=2)
#141
#142	print(f"Exported {len(all_messages)} messages")""",
#143	"env_hint": None,
#144	},
#145	"supermemory": {
#146	"install": "pip install supermemory",
#147	"api_import": "from supermemory import SuperMemory",
#148	"extract_code": """client = SuperMemory(api_key="YOUR_API_KEY")
#149
#150	# List documents
#151	docs = client.documents.list()
#152	all_memories = []
#153
#154	for doc in docs:
#155	all_memories.append({
#156	"content": doc.get("content", ""),
#157	"container_tag": doc.get("containerTag", ""),
#158	"is_static": doc.get("isStatic", False),
#159	"created_at": doc.get("createdAt"),
#160	})
#161
#162	# Also search for memories
#163	results = client.search.execute(q="*", containerTags=["YOUR_CONTAINER_TAG"])
#164	for mem in results.get("results", []):
#165	all_memories.append({
#166	"content": mem.get("content", mem.get("memory", "")),
#167	"container_tag": mem.get("containerTag", ""),
#168	"created_at": mem.get("createdAt"),
#169	})
#170
#171	with open("supermemory_export.json", "w") as f:
#172	json.dump(all_memories, f, indent=2)
#173
#174	print(f"Exported {len(all_memories)} items")""",
#175	"env_hint": "SUPERMEMORY_API_KEY",
#176	},
#177	}
#178
#179
#180	class AgenticImporter:
#181	"""Generate migration scripts and instructions for any provider.
#182
#183	This doesn't extract data itself — it produces scripts and prompts
#184	that the user or their AI agent can use to perform the extraction.
#185
#186	Usage:
#187	gen = AgenticImporter()
#188	script = gen.generate_script("mem0", api_key="sk-xxx", user_id="alice")
#189	print(script)
#190
#191	# Or generate agent instructions
#192	instructions = gen.generate_agent_instructions("zep")
#193	print(instructions)
#194	"""
#195
#196	def generate_script(self, provider: str, **kwargs) -> str:
#197	"""Generate a ready-to-run Python migration script.
#198
#199	Args:
#200	provider: Provider name (mem0, letta, zep, etc.)
#201	**kwargs: api_key, user_id, agent_id, workspace_id, container_tag
#202
#203	Returns:
#204	Python script as a string.
#205	"""
#206	meta = PROVIDER_SCRIPTS.get(provider)
#207	if not meta:
#208	return self._generate_generic_script(provider, **kwargs)
#209
#210	api_key = kwargs.get("api_key", "YOUR_API_KEY")
#211	install = meta["install"]
#212	api_import = meta["api_import"]
#213	extract = meta["extract_code"]
#214	env_hint = meta.get("env_hint", "")
#215
#216	header = f'''#!/usr/bin/env python3
#217	"""
#218	Mnemosyne Migration Script: {provider.title()} → Mnemosyne
#219	Generated by Mnemosyne Agentic Importer
#220
#221	Prerequisites:
#222	{install}
#223
#224	Environment:
#225	{f" Set {env_hint} env var or pass api_key directly." if env_hint else " No special env vars required."}
#226
#227	Usage:
#228	python3 migrate_{provider}.py
#229	# Then: hermes mnemosyne import --file {provider}_export.json
#230	"""
#231
#232	import json
#233	import os
#234
#235	# Configuration
#236	API_KEY = os.environ.get("{env_hint}", "{api_key}")
#237	'''
#238
#239	script = f'''{header}
#240	{api_import}
#241
#242	{extract}
#243
#244	print("\\nNext step:")
#245	print(" hermes mnemosyne import --file {provider}_export.json")
#246	'''
#247	return script
#248
#249	def generate_agent_instructions(self, provider: str) -> str:
#250	"""Generate instructions the user can paste to their AI agent.
#251
#252	The user gives these instructions to their AI agent (Claude, ChatGPT, etc.)
#253	and the agent follows them to extract memories and produce a JSON file.
#254	"""
#255	meta = PROVIDER_SCRIPTS.get(provider)
#256	install = meta["install"] if meta else f"pip install <{provider}-sdk>"
#257
#258	return f"""I need to migrate all my memories from {provider.title()} to Mnemosyne.
#259
#260	Please help me extract ALL memories from {provider.title()} and save them as a JSON file.
#261
#262	Requirements:
#263	1. Install the SDK: `{install}`
#264	2. Extract every available memory/message/fact/document
#265	3. Save to a JSON file called `{provider}_export.json`
#266	4. The JSON should be an array of objects, each with at minimum a "content" key
#267	5. Preserve any metadata, timestamps, and user/agent IDs
#268
#269	After extraction, I'll run:
#270	hermes mnemosyne import --file {provider}_export.json
#271
#272	Please give me the exact Python script to run. Test it for syntax errors before giving it to me."""
#273
#274	def generate_docs_instructions(self, provider: str) -> str:
#275	"""Generate step-by-step migration instructions for documentation."""
#276	meta = PROVIDER_SCRIPTS.get(provider)
#277	if not meta:
#278	return self._generate_generic_docs(provider)
#279
#280	install = meta["install"]
#281	env_hint = meta.get("env_hint", "")
#282
#283	return f"""## Migrating from {provider.title()} to Mnemosyne
#284
#285	### Step 1: Install dependencies
#286	```bash
#287	pip install mnemosyne-memory
#288	{install}
#289	```
#290
#291	{ "### Step 2: Set API key" + chr(10) + "```bash" + chr(10) + f"export {env_hint}=sk-xxx" + chr(10) + "```" if env_hint else ""}
#292
#293	### Step 3: Run the migration script
#294	Save the script below as `migrate_{provider}.py` and run it:
#295
#296	```bash
#297	python3 migrate_{provider}.py
#298	```
#299
#300	This extracts all memories to `{provider}_export.json`.
#301
#302	### Step 4: Import into Mnemosyne
#303	```bash
#304	hermes mnemosyne import --file {provider}_export.json
#305	```
#306
#307	Or via CLI provider import:
#308	```bash
#309	hermes mnemosyne import --from {provider} --api-key sk-xxx
#310	```
#311
#312	### Step 5: Verify
#313	```bash
#314	hermes mnemosyne stats
#315	```
#316	"""
#317
#318	def _generate_generic_script(self, provider: str, **kwargs) -> str:
#319	"""Generate a generic extraction script for unknown providers."""
#320	return f'''#!/usr/bin/env python3
#321	"""
#322	Generic migration script for {provider.title()}
#323
#324	Replace the extraction logic below with your provider's API calls.
#325	"""
#326
#327	import json
#328
#329	all_memories = []
#330
#331	# TODO: Replace with your provider's extraction logic
#332	# Example: client = YourProviderSDK(api_key="...")
#333	# Example: memories = client.list_all()
#334
#335	# Each memory should be a dict with at minimum "content":
#336	# {{"content": "the memory text", "metadata": {{}}, "timestamp": "2026-01-01T00:00:00Z"}}
#337
#338	with open("{provider}_export.json", "w") as f:
#339	json.dump(all_memories, f, indent=2)
#340
#341	print(f"Exported {{len(all_memories)}} items to {provider}_export.json")
#342	print("Then run: hermes mnemosyne import --file {provider}_export.json")
#343	'''
#344
#345	def _generate_generic_docs(self, provider: str) -> str:
#346	"""Generate generic docs for unknown providers."""
#347	return f"""## Migrating from {provider.title()} to Mnemosyne
#348
#349	{provider.title()} doesn't have a built-in importer yet. Use the generic extraction approach:
#350
#351	### Option A: Export manually
#352	Export your memories from {provider.title()} to a JSON file, then import:
#353
#354	```bash
#355	hermes mnemosyne import --file export.json
#356	```
#357
#358	### Option B: Use your AI agent
#359	Paste these instructions to your AI agent:
#360
#361	```
#362	I need to extract ALL memories from {provider.title()} and save them as a JSON file.
#363	The JSON should be an array of objects with a "content" key for each memory.
#364	Save as "{provider}_export.json" so I can import into Mnemosyne with:
#365	hermes mnemosyne import --file {provider}_export.json
#366	```
#367
#368	### Option C: Request an importer
#369	Open an issue on GitHub requesting a {provider.title()} importer:
#370	https://github.com/AxDSan/mnemosyne/issues
#371	"""
#372
#373
#374	# Convenience functions
#375
#376	def generate_migration_script(provider: str, **kwargs) -> str:
#377	"""Generate a ready-to-run migration script for the given provider."""
#378	gen = AgenticImporter()
#379	return gen.generate_script(provider, **kwargs)
#380
#381
#382	def generate_agent_instructions(provider: str) -> str:
#383	"""Generate instructions for an AI agent to perform the migration."""
#384	gen = AgenticImporter()
#385	return gen.generate_agent_instructions(provider)
#386
#387
#388	def generate_docs_instructions(provider: str) -> str:
#389	"""Generate step-by-step documentation for migration."""
#390	gen = AgenticImporter()
#391	return gen.generate_docs_instructions(provider)
#392

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public