my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""CLI commands for Mnemosyne memory provider.
#2
#3	Available via: hermes mnemosyne <subcommand>
#4	"""
#5
#6	from __future__ import annotations
#7
#8	import json
#9	import sys
#10	from pathlib import Path
#11
#12	_mnemosyne_root = Path(__file__).resolve().parent.parent
#13	if str(_mnemosyne_root) not in sys.path:
#14	sys.path.insert(0, str(_mnemosyne_root))
#15
#16
#17	def register_cli(subparser):
#18	"""Register CLI subcommands for ``hermes mnemosyne``."""
#19	mn_cmds = subparser.add_subparsers(dest="mnemosyne_cmd")
#20
#21	stats_cmd = mn_cmds.add_parser("stats", help="Show memory statistics")
#22	stats_cmd.add_argument("--global", "-g", action="store_true", help="Show global stats across all sessions")
#23
#24	sleep_cmd = mn_cmds.add_parser("sleep", help="Run consolidation cycle")
#25	sleep_cmd.add_argument("--all-sessions", action="store_true", help="Consolidate eligible old working memories across all sessions")
#26	sleep_cmd.add_argument("--dry-run", action="store_true", help="Report what would be consolidated without writing changes")
#27	mn_cmds.add_parser("version", help="Show Mnemosyne version")
#28
#29	inspect_cmd = mn_cmds.add_parser("inspect", help="Search memories")
#30	inspect_cmd.add_argument("query", nargs="?", default="", help="Search query")
#31	inspect_cmd.add_argument("--limit", type=int, default=10, help="Max results")
#32
#33	mn_cmds.add_parser("clear", help="Clear scratchpad")
#34
#35	export_cmd = mn_cmds.add_parser("export", help="Export all memories to a JSON file")
#36	export_cmd.add_argument("--output", "-o", type=str, required=True, help="Output JSON file path")
#37
#38	import_cmd = mn_cmds.add_parser("import", help="Import memories from a JSON file or another provider")
#39	import_cmd.add_argument("--input", "-i", type=str, help="Input JSON file path (for file imports)")
#40	import_cmd.add_argument("--file", type=str, help="Provider file input, e.g. Hindsight JSON export")
#41	import_cmd.add_argument("--force", action="store_true", help="Overwrite existing records (file import)")
#42	import_cmd.add_argument("--from", dest="from_provider", type=str, help="Provider to import from (e.g., 'mem0')")
#43	import_cmd.add_argument("--api-key", type=str, help="Provider API key (or set env var)")
#44	import_cmd.add_argument("--user-id", type=str, help="Filter by user ID (provider-specific)")
#45	import_cmd.add_argument("--agent-id", type=str, help="Filter by agent ID (provider-specific)")
#46	import_cmd.add_argument("--base-url", type=str, help="Provider base URL (for self-hosted)")
#47	import_cmd.add_argument("--bank", type=str, help="Provider memory bank, e.g. Hindsight bank")
#48	import_cmd.add_argument("--dry-run", action="store_true", help="Validate but don't import")
#49	import_cmd.add_argument("--session-id", type=str, help="Override session for imported memories")
#50	import_cmd.add_argument("--channel-id", type=str, help="Channel for imported memories")
#51	import_cmd.add_argument("--list-providers", action="store_true", help="List supported import providers")
#52	import_cmd.add_argument("--generate-script", action="store_true", help="Generate a migration script for the provider")
#53	import_cmd.add_argument("--agentic", action="store_true", help="Generate agent migration instructions (prompt to give your AI agent)")
#54	import_cmd.add_argument("--output-script", type=str, help="Save generated script to file")
#55
#56	subparser.set_defaults(func=mnemosyne_command)
#57
#58
#59	def mnemosyne_command(args):
#60	"""Dispatch ``hermes mnemosyne <subcommand>``."""
#61	cmd = getattr(args, "mnemosyne_cmd", None)
#62	if not cmd:
#63	print("Usage: hermes mnemosyne {stats\|sleep\|version\|inspect\|clear\|export\|import}")
#64	return 1
#65
#66	try:
#67	from mnemosyne.core.beam import BeamMemory
#68	beam = BeamMemory(session_id="hermes_default")
#69	except Exception as e:
#70	print(f"Error: Mnemosyne not available: {e}")
#71	return 1
#72
#73	if cmd == "stats":
#74	if getattr(args, "global", False):
#75	working = beam.get_global_working_stats()
#76	else:
#77	working = beam.get_working_stats()
#78	episodic = beam.get_episodic_stats()
#79	print(json.dumps({"working": working, "episodic": episodic}, indent=2))
#80
#81	elif cmd == "version":
#82	from mnemosyne import __version__, __author__
#83	print(f"Mnemosyne {__version__} by {__author__}")
#84
#85	elif cmd == "sleep":
#86	dry_run = bool(getattr(args, "dry_run", False))
#87	if getattr(args, "all_sessions", False):
#88	result = beam.sleep_all_sessions(dry_run=dry_run)
#89	else:
#90	result = beam.sleep(dry_run=dry_run)
#91	print(json.dumps(result, indent=2))
#92
#93	elif cmd == "inspect":
#94	query = getattr(args, "query", "") or ""
#95	limit = getattr(args, "limit", 10)
#96	if not query:
#97	query = input("Search query: ")
#98	results = beam.recall(query, top_k=limit)
#99	print(f"Results for '{query}': {len(results)}")
#100	for i, r in enumerate(results, 1):
#101	content = r.get("content", "")[:120]
#102	imp = r.get("importance", 0.0)
#103	print(f" {i}. [{imp:.2f}] {content}")
#104
#105	elif cmd == "clear":
#106	confirm = input("Clear scratchpad? This cannot be undone. [y/N]: ")
#107	if confirm.lower() in ("y", "yes"):
#108	beam.scratchpad_clear()
#109	print("Scratchpad cleared.")
#110	else:
#111	print("Cancelled.")
#112
#113	elif cmd == "export":
#114	output_path = getattr(args, "output", None)
#115	if not output_path:
#116	print("Usage: hermes mnemosyne export --output <path>")
#117	return 1
#118	try:
#119	from mnemosyne.core.memory import Mnemosyne
#120	mem = Mnemosyne(session_id="hermes_default")
#121	result = mem.export_to_file(output_path)
#122	print(f"Exported {result['working_memory_count']} working, {result['episodic_memory_count']} episodic, {result['legacy_memories_count']} legacy, {result['triples_count']} triples to {output_path}")
#123	except Exception as e:
#124	print(f"Export failed: {e}")
#125	return 1
#126
#127	elif cmd == "import":
#128	# --list-providers
#129	if getattr(args, "list_providers", False):
#130	from mnemosyne.core.importers import PROVIDERS
#131	print("Supported import providers:")
#132	for name, info in PROVIDERS.items():
#133	print(f" {name}: {info['description']}")
#134	print(f" docs: {info['docs']}")
#135	print(f" env key: {info['env_key']}")
#136	print(f" pip: {info['pypi_package']}")
#137	return 0
#138
#139	# --agentic: generate instructions for user's AI agent
#140	generate_script_flag = getattr(args, "generate_script", False)
#141	agentic_flag = getattr(args, "agentic", False)
#142	from_provider = getattr(args, "from_provider", None)
#143	output_script = getattr(args, "output_script", None)
#144
#145	if agentic_flag and from_provider:
#146	from mnemosyne.core.importers.agentic import generate_agent_instructions
#147	instructions = generate_agent_instructions(from_provider)
#148	if output_script:
#149	Path(output_script).write_text(instructions)
#150	print(f"Agent instructions saved to {output_script}")
#151	else:
#152	print(instructions)
#153	return 0
#154
#155	if generate_script_flag and from_provider:
#156	from mnemosyne.core.importers.agentic import generate_migration_script
#157	api_key = getattr(args, "api_key", None)
#158	user_id = getattr(args, "user_id", None)
#159	script = generate_migration_script(
#160	from_provider,
#161	api_key=api_key or "",
#162	user_id=user_id or "",
#163	)
#164	if output_script:
#165	Path(output_script).write_text(script)
#166	print(f"Migration script saved to {output_script}")
#167	else:
#168	print(script)
#169	return 0
#170
#171	cross_provider = from_provider
#172	input_path = getattr(args, "input", None)
#173	dry_run = getattr(args, "dry_run", False)
#174	session_id = getattr(args, "session_id", None)
#175	channel_id = getattr(args, "channel_id", None)
#176
#177	try:
#178	from mnemosyne.core.memory import Mnemosyne
#179	mem = Mnemosyne(session_id=session_id or "import_session",
#180	channel_id=channel_id)
#181	except Exception as e:
#182	print(f"Error: Mnemosyne not available: {e}")
#183	return 1
#184
#185	# Cross-provider import
#186	if cross_provider:
#187	api_key = getattr(args, "api_key", None)
#188	user_id = getattr(args, "user_id", None)
#189	agent_id = getattr(args, "agent_id", None)
#190	base_url = getattr(args, "base_url", None)
#191
#192	def _print_import_result(result):
#193	print(f"\nImport complete:")
#194	print(f" Total found: {result.total}")
#195	print(f" Imported: {result.imported}")
#196	print(f" Skipped: {result.skipped}")
#197	print(f" Failed: {result.failed}")
#198	if result.errors:
#199	print(f" Errors:")
#200	for err in result.errors[:10]:
#201	print(f" - {err}")
#202	if len(result.errors) > 10:
#203	print(f" ... and {len(result.errors) - 10} more")
#204
#205	if cross_provider == "hindsight":
#206	file_path = getattr(args, "file", None) or input_path
#207	bank = getattr(args, "bank", None) or "hermes"
#208	if not file_path and not base_url:
#209	print("Error: Hindsight import requires --file/--input or --base-url.")
#210	return 1
#211
#212	print("Importing from hindsight...")
#213	if dry_run:
#214	print(" (dry-run mode: no memories will be written)")
#215
#216	try:
#217	from mnemosyne.core.importers import import_from_provider
#218	import_kwargs = {
#219	"file_path": file_path,
#220	"base_url": base_url,
#221	"bank": bank,
#222	"dry_run": dry_run,
#223	"session_id": session_id,
#224	"channel_id": channel_id,
#225	}
#226	result = import_from_provider(
#227	"hindsight", mem,
#228	**import_kwargs,
#229	)
#230	_print_import_result(result)
#231	return 0 if result.failed == 0 else 1
#232	except ValueError as e:
#233	print(f"Error: {e}")
#234	return 1
#235	except Exception as e:
#236	print(f"Import failed: {e}")
#237	return 1
#238
#239	# Try env var fallback
#240	import os
#241	if not api_key:
#242	info = __import__("mnemosyne.core.importers", fromlist=["PROVIDERS"]).PROVIDERS
#243	pk = info.get(cross_provider, {}).get("env_key", "")
#244	if pk:
#245	api_key = os.environ.get(pk)
#246	if not api_key:
#247	print(f"Error: --api-key required for {cross_provider} import. "
#248	f"Or set the {cross_provider.upper()}_API_KEY env var.")
#249	return 1
#250
#251	print(f"Importing from {cross_provider}...")
#252	if dry_run:
#253	print(" (dry-run mode: no memories will be written)")
#254
#255	try:
#256	from mnemosyne.core.importers import import_from_provider
#257	result = import_from_provider(
#258	cross_provider, mem,
#259	api_key=api_key,
#260	user_id=user_id,
#261	agent_id=agent_id,
#262	base_url=base_url,
#263	dry_run=dry_run,
#264	session_id=session_id,
#265	channel_id=channel_id,
#266	)
#267	_print_import_result(result)
#268	return 0 if result.failed == 0 else 1
#269	except ValueError as e:
#270	print(f"Error: {e}")
#271	return 1
#272	except Exception as e:
#273	print(f"Import failed: {e}")
#274	return 1
#275
#276	# File import
#277	force = getattr(args, "force", False)
#278	if not input_path:
#279	print("Usage: hermes mnemosyne import --input <path> [--force]")
#280	print(" hermes mnemosyne import --from <provider> --api-key <key> [--dry-run]")
#281	print(" hermes mnemosyne import --list-providers")
#282	return 1
#283	try:
#284	stats = mem.import_from_file(input_path, force=force)
#285	beam_stats = stats.get("beam", {})
#286	legacy_stats = stats.get("legacy", {})
#287	triples_stats = stats.get("triples", {})
#288	print(f"Import complete:")
#289	print(f" Working: +{beam_stats.get('working_memory', {}).get('inserted', 0)}")
#290	print(f" Episodic: +{beam_stats.get('episodic_memory', {}).get('inserted', 0)}")
#291	print(f" Legacy: +{legacy_stats.get('inserted', 0)}")
#292	print(f" Triples: +{triples_stats.get('inserted', 0)}")
#293	if force:
#294	print(f" (force mode: overwrites applied)")
#295	except Exception as e:
#296	print(f"Import failed: {e}")
#297	return 1
#298
#299	return 0
#300

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public