207 lines
6.8 KiB
Python
Executable File
207 lines
6.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
def load_env():
|
|
env_paths = [
|
|
Path(__file__).parent / ".env",
|
|
Path.home() / ".config/opencode/.env",
|
|
]
|
|
|
|
for env_path in env_paths:
|
|
if env_path.exists():
|
|
for line in env_path.read_text().splitlines():
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
k, v = line.split("=", 1)
|
|
os.environ.setdefault(k.strip(), v.strip().strip("\"'"))
|
|
|
|
load_env()
|
|
|
|
try:
|
|
from shodh_memory import MemorySystem
|
|
memory = MemorySystem()
|
|
except Exception as e:
|
|
print(f"Error: Failed to import shodh_memory: {e}", file=sys.stderr)
|
|
print("Install with: pip install shodh-memory", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_remember(content, memory_type="Context", tags=None, user_id=None):
|
|
try:
|
|
tag_list = [t.strip() for t in tags.split(",")] if tags else None
|
|
result = memory.remember(content, memory_type=memory_type, tags=tag_list)
|
|
print(f"Result: memory stored [{result}]")
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_recall(query, limit=5, user_id=None):
|
|
try:
|
|
results = memory.recall(query, limit=limit)
|
|
|
|
if not results:
|
|
print("No memories found")
|
|
return
|
|
|
|
for i, mem in enumerate(results, 1):
|
|
content = mem.get("content", "") if isinstance(mem, dict) else str(mem)[:100]
|
|
print(f"\n{i}. {content[:100]}{'...' if len(str(content)) > 100 else ''}")
|
|
|
|
print(f"\nResult: {len(results)} memories found")
|
|
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_proactive(context, limit=5, user_id=None):
|
|
try:
|
|
results = memory.proactive_context(context, limit=limit)
|
|
|
|
if not results:
|
|
print("No relevant memories found")
|
|
return
|
|
|
|
print("Relevant memories:")
|
|
for i, mem in enumerate(results, 1):
|
|
content = mem.get("content", "") if isinstance(mem, dict) else str(mem)
|
|
print(f"\n{i}. {content[:150]}{'...' if len(str(content)) > 150 else ''}")
|
|
|
|
print(f"\nResult: {len(results)} relevant memories")
|
|
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_stats(user_id=None):
|
|
try:
|
|
stats = memory.get_stats()
|
|
|
|
print("Memory Statistics:")
|
|
if isinstance(stats, dict):
|
|
for key, value in stats.items():
|
|
print(f" {key}: {value}")
|
|
else:
|
|
print(f" {stats}")
|
|
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_forget(memory_id):
|
|
try:
|
|
memory.forget(memory_id)
|
|
print(f"Result: memory deleted [{memory_id}]")
|
|
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_context(limit=10, user_id=None):
|
|
try:
|
|
summary = memory.context_summary(limit=limit)
|
|
|
|
print("Context Summary:")
|
|
if isinstance(summary, dict):
|
|
for key, value in summary.items():
|
|
print(f" {key}: {value}")
|
|
else:
|
|
print(f" {summary}")
|
|
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_list(user_id=None, limit=20):
|
|
try:
|
|
results = memory.list_memories(limit=limit)
|
|
|
|
if not results:
|
|
print("No memories stored")
|
|
return
|
|
|
|
print(f"Stored memories ({len(results)}):")
|
|
for i, mem in enumerate(results, 1):
|
|
if isinstance(mem, dict):
|
|
content = mem.get("content", "")[:60]
|
|
mem_id = mem.get("id", "")
|
|
else:
|
|
content = str(mem)[:60]
|
|
mem_id = "unknown"
|
|
print(f"{i}. {content}... [{mem_id}]")
|
|
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Shodh Memory - Persistent cognitive memory for AI agents"
|
|
)
|
|
subparsers = parser.add_subparsers(dest="command", help="Commands")
|
|
|
|
remember_parser = subparsers.add_parser("remember", help="Store a memory")
|
|
remember_parser.add_argument("content", help="Memory content")
|
|
remember_parser.add_argument("--type", "-t", default="Context",
|
|
help="Memory type (Decision, Learning, Error, etc.)")
|
|
remember_parser.add_argument("--tags", help="Comma-separated tags")
|
|
remember_parser.add_argument("--user", help="User ID")
|
|
|
|
recall_parser = subparsers.add_parser("recall", help="Search memories by meaning")
|
|
recall_parser.add_argument("query", help="Search query")
|
|
recall_parser.add_argument("--limit", "-l", type=int, default=5, help="Number of results")
|
|
recall_parser.add_argument("--user", help="User ID")
|
|
|
|
proactive_parser = subparsers.add_parser("proactive", help="Get relevant memories for context")
|
|
proactive_parser.add_argument("context", help="Current context")
|
|
proactive_parser.add_argument("--limit", "-l", type=int, default=5, help="Number of results")
|
|
proactive_parser.add_argument("--user", help="User ID")
|
|
|
|
stats_parser = subparsers.add_parser("stats", help="Get memory statistics")
|
|
stats_parser.add_argument("--user", help="User ID")
|
|
|
|
forget_parser = subparsers.add_parser("forget", help="Delete a memory")
|
|
forget_parser.add_argument("memory_id", help="Memory ID to delete")
|
|
|
|
context_parser = subparsers.add_parser("context", help="Get context summary")
|
|
context_parser.add_argument("--limit", "-l", type=int, default=10, help="Number of results")
|
|
context_parser.add_argument("--user", help="User ID")
|
|
|
|
list_parser = subparsers.add_parser("list", help="List all memories")
|
|
list_parser.add_argument("--limit", "-l", type=int, default=20, help="Number of results")
|
|
list_parser.add_argument("--user", help="User ID")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not args.command:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
if args.command == "remember":
|
|
cmd_remember(args.content, args.type, args.tags, args.user)
|
|
elif args.command == "recall":
|
|
cmd_recall(args.query, args.limit, args.user)
|
|
elif args.command == "proactive":
|
|
cmd_proactive(args.context, args.limit, args.user)
|
|
elif args.command == "stats":
|
|
cmd_stats(args.user)
|
|
elif args.command == "forget":
|
|
cmd_forget(args.memory_id)
|
|
elif args.command == "context":
|
|
cmd_context(args.limit, args.user)
|
|
elif args.command == "list":
|
|
cmd_list(args.user, args.limit)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|