158 lines
5.0 KiB
Python
Executable File
158 lines
5.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""odysseus-memory — shell wrapper for the AI memory store.
|
|
|
|
The chat assistant accumulates durable facts about the user (name,
|
|
projects, preferences, contacts) into `data/memory.json`. This CLI
|
|
lets you inspect, search, and prune them from the shell.
|
|
|
|
odysseus-memory list # all entries (paged)
|
|
odysseus-memory list --category preference
|
|
odysseus-memory search "tokyo" # substring match
|
|
odysseus-memory show MEMORY_ID
|
|
odysseus-memory add "User lives in Tokyo" --category fact
|
|
odysseus-memory delete MEMORY_ID
|
|
odysseus-memory categories # counts per category
|
|
|
|
Reads/writes the same `data/memory.json` the web UI uses via the
|
|
`MemoryManager` helper. Atomic — the manager handles the temp-file
|
|
swap so partial writes can't corrupt the file.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import sys
|
|
import os, sys
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "_lib"))
|
|
from cli import quiet_logs, emit, fail, common_parser, run, REPO_ROOT as _REPO_ROOT
|
|
quiet_logs()
|
|
|
|
import argparse, json, logging, os, sys
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from services.memory.memory import MemoryManager
|
|
quiet_logs()
|
|
except ModuleNotFoundError as e:
|
|
sys.stderr.write(f"error: {e}\nhint: run from repo root with venv active.\n")
|
|
sys.exit(2)
|
|
|
|
|
|
_DATA_DIR = str(_REPO_ROOT / "data")
|
|
_mgr: MemoryManager | None = None
|
|
|
|
|
|
def _manager() -> MemoryManager:
|
|
global _mgr
|
|
if _mgr is None:
|
|
_mgr = MemoryManager(_DATA_DIR)
|
|
return _mgr
|
|
|
|
|
|
def _memory_entries(entries):
|
|
return [e for e in entries or [] if isinstance(e, dict)]
|
|
|
|
|
|
def cmd_list(args):
|
|
entries = _memory_entries(_manager().load_all())
|
|
if args.category:
|
|
entries = [e for e in entries if (e.get("category") or "fact") == args.category]
|
|
if args.source:
|
|
entries = [e for e in entries if (e.get("source") or "user") == args.source]
|
|
if args.owner:
|
|
entries = [e for e in entries if (e.get("owner") or "") == args.owner]
|
|
# Newest first
|
|
entries = sorted(entries, key=lambda e: e.get("timestamp", 0), reverse=True)
|
|
emit(entries[: args.limit], args)
|
|
|
|
|
|
def cmd_search(args):
|
|
q = args.query.lower()
|
|
entries = _memory_entries(_manager().load_all())
|
|
matches = [e for e in entries if q in (e.get("text") or "").lower()]
|
|
matches = sorted(matches, key=lambda e: e.get("timestamp", 0), reverse=True)
|
|
emit(matches[: args.limit], args)
|
|
|
|
|
|
def cmd_show(args):
|
|
for e in _memory_entries(_manager().load_all()):
|
|
if e.get("id") == args.id:
|
|
emit(e, args)
|
|
return
|
|
fail(f"no memory with id {args.id!r}")
|
|
|
|
|
|
def cmd_add(args):
|
|
entry = _manager().add_entry(
|
|
args.text,
|
|
source="cli",
|
|
category=args.category,
|
|
owner=args.owner,
|
|
)
|
|
# add_entry doesn't save by default — the call in chat does it
|
|
# after dedup checks. Persist here so a one-shot CLI add sticks.
|
|
all_entries = _manager().load_all()
|
|
if not any(e.get("id") == entry.get("id") for e in all_entries):
|
|
all_entries.append(entry)
|
|
_manager().save(all_entries)
|
|
emit(entry, args)
|
|
|
|
|
|
def cmd_delete(args):
|
|
entries = _memory_entries(_manager().load_all())
|
|
target = next((e for e in entries if e.get("id") == args.id), None)
|
|
if not target:
|
|
fail(f"no memory with id {args.id!r}")
|
|
remaining = [e for e in entries if e.get("id") != args.id]
|
|
_manager().save(remaining)
|
|
emit({"ok": True, "deleted": target}, args)
|
|
|
|
|
|
def cmd_categories(args):
|
|
counts: dict[str, int] = {}
|
|
for e in _memory_entries(_manager().load_all()):
|
|
cat = e.get("category") or "fact"
|
|
counts[cat] = counts.get(cat, 0) + 1
|
|
rows = sorted(counts.items(), key=lambda kv: -kv[1])
|
|
emit([{"category": c, "count": n} for c, n in rows], args)
|
|
|
|
|
|
def _build_parser():
|
|
common = argparse.ArgumentParser(add_help=False)
|
|
common.add_argument("--pretty", action="store_true")
|
|
p = argparse.ArgumentParser(prog="odysseus-memory", parents=[common])
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
|
|
pl = sub.add_parser("list", parents=[common])
|
|
pl.add_argument("--category")
|
|
pl.add_argument("--source")
|
|
pl.add_argument("--owner")
|
|
pl.add_argument("--limit", type=int, default=50)
|
|
pl.set_defaults(func=cmd_list)
|
|
|
|
ps = sub.add_parser("search", parents=[common])
|
|
ps.add_argument("query")
|
|
ps.add_argument("--limit", type=int, default=50)
|
|
ps.set_defaults(func=cmd_search)
|
|
|
|
psh = sub.add_parser("show", parents=[common])
|
|
psh.add_argument("id")
|
|
psh.set_defaults(func=cmd_show)
|
|
|
|
pa = sub.add_parser("add", parents=[common])
|
|
pa.add_argument("text")
|
|
pa.add_argument("--category", default="fact")
|
|
pa.add_argument("--owner")
|
|
pa.set_defaults(func=cmd_add)
|
|
|
|
pd = sub.add_parser("delete", parents=[common])
|
|
pd.add_argument("id")
|
|
pd.set_defaults(func=cmd_delete)
|
|
|
|
pc = sub.add_parser("categories", parents=[common])
|
|
pc.set_defaults(func=cmd_categories)
|
|
|
|
return p
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(run(_build_parser()))
|