File size: 4,149 Bytes
99bb537 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """
Persistent Memory System (OpenClaw-style)
==========================================
Memory = markdown files on disk. Simple, readable, editable.
Skills = saved Python files agents create and reuse.
"""
import os, json, datetime
from pathlib import Path
HOME = Path(os.environ.get("HOME", "/home/user"))
MEM_DIR = HOME / ".praison_memory"
SKILLS_DIR= HOME / ".praison_skills"
for d in [MEM_DIR, SKILLS_DIR]:
d.mkdir(parents=True, exist_ok=True)
# ── Memory ────────────────────────────────────────────────────
def save_memory(key: str, content: str):
"""Save a memory note (key = filename without .md)."""
safe = "".join(c if c.isalnum() or c in "-_" else "_" for c in key)
path = MEM_DIR / f"{safe}.md"
ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")
path.write_text(f"# {key}\n_Updated: {ts}_\n\n{content}\n")
def load_memory(key: str) -> str:
safe = "".join(c if c.isalnum() or c in "-_" else "_" for c in key)
path = MEM_DIR / f"{safe}.md"
return path.read_text() if path.exists() else ""
def list_memories() -> list[str]:
return [p.stem for p in MEM_DIR.glob("*.md")]
def search_memories(query: str) -> str:
"""Search all memory files for query."""
query_lower = query.lower()
results = []
for p in MEM_DIR.glob("*.md"):
content = p.read_text()
if query_lower in content.lower():
results.append(f"### {p.stem}\n{content[:400]}")
return "\n\n".join(results) if results else "No memories found."
def get_memory_context() -> str:
"""Get all memories as context for the agent."""
memories = list(MEM_DIR.glob("*.md"))
if not memories:
return ""
parts = ["## Persistent Memory"]
for p in memories[:10]: # limit to 10 most recent
parts.append(f"### {p.stem}\n{p.read_text()[:300]}")
return "\n\n".join(parts)
# ── Skills ────────────────────────────────────────────────────
def save_skill(name: str, code: str, description: str = ""):
"""Save a Python skill file for reuse."""
safe = "".join(c if c.isalnum() or c in "-_" else "_" for c in name)
path = SKILLS_DIR / f"{safe}.py"
header = f'"""\nSkill: {name}\nDescription: {description}\nCreated: {datetime.datetime.now().strftime("%Y-%m-%d")}\n"""\n'
path.write_text(header + code)
return str(path)
def load_skill(name: str) -> str:
safe = "".join(c if c.isalnum() or c in "-_" else "_" for c in name)
path = SKILLS_DIR / f"{safe}.py"
return path.read_text() if path.exists() else ""
def list_skills() -> list[dict]:
skills = []
for p in SKILLS_DIR.glob("*.py"):
content = p.read_text()
# Extract description from docstring
desc = ""
if '"""' in content:
try:
desc = content.split('"""')[1].strip().split('\n')
desc = next((l for l in desc if l.startswith("Description:")), "")
desc = desc.replace("Description:", "").strip()
except Exception:
pass
skills.append({"name": p.stem, "description": desc, "path": str(p)})
return skills
def get_skills_context() -> str:
skills = list_skills()
if not skills:
return ""
parts = ["## Available Skills (reuse these)"]
for s in skills:
code = load_skill(s["name"])[:400]
parts.append(f"### {s['name']}\n{s['description']}\n```python\n{code}\n```")
return "\n\n".join(parts)
def delete_skill(name: str) -> bool:
safe = "".join(c if c.isalnum() or c in "-_" else "_" for c in name)
path = SKILLS_DIR / f"{safe}.py"
if path.exists():
path.unlink()
return True
return False
def get_all_for_api() -> dict:
return {
"memories": [{"key": p.stem, "preview": p.read_text()[:100]} for p in MEM_DIR.glob("*.md")],
"skills": list_skills(),
"mem_dir": str(MEM_DIR),
"skill_dir":str(SKILLS_DIR),
} |