Spaces:
Running
Running
File size: 7,053 Bytes
dd87944 | 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 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 | """Gestion mémoire long terme — port Mark-XLVIII."""
from __future__ import annotations
import json
import sys
from datetime import datetime
from pathlib import Path
from threading import Lock
_DESKTOP_DIR = Path(__file__).resolve().parent.parent
MEMORY_PATH = _DESKTOP_DIR / "memory" / "long_term.json"
_lock = Lock()
MAX_VALUE_LENGTH = 380
MEMORY_MAX_CHARS = 2200
def _empty_memory() -> dict:
return {
"identity": {},
"preferences": {},
"projects": {},
"relationships": {},
"wishes": {},
"notes": {},
}
def load_memory() -> dict:
if not MEMORY_PATH.exists():
return _empty_memory()
with _lock:
try:
data = json.loads(MEMORY_PATH.read_text(encoding="utf-8"))
if isinstance(data, dict):
base = _empty_memory()
for key in base:
if key not in data:
data[key] = {}
return data
return _empty_memory()
except Exception as e:
print(f"[Memory] Load error: {e}")
return _empty_memory()
def _all_entries(memory: dict) -> list[tuple]:
entries = []
for cat, items in memory.items():
if not isinstance(items, dict):
continue
for key, entry in items.items():
if isinstance(entry, dict) and "value" in entry:
entries.append((cat, key, entry))
return entries
def _trim_to_limit(memory: dict) -> dict:
if len(json.dumps(memory, ensure_ascii=False)) <= MEMORY_MAX_CHARS:
return memory
entries = _all_entries(memory)
entries.sort(key=lambda t: t[2].get("updated", "0000-00-00"))
for cat, key, _ in entries:
if len(json.dumps(memory, ensure_ascii=False)) <= MEMORY_MAX_CHARS:
break
del memory[cat][key]
print(f"[Memory] Trimmed {cat}/{key}")
return memory
def save_memory(memory: dict) -> None:
if not isinstance(memory, dict):
return
memory = _trim_to_limit(memory)
MEMORY_PATH.parent.mkdir(parents=True, exist_ok=True)
with _lock:
MEMORY_PATH.write_text(
json.dumps(memory, indent=2, ensure_ascii=False),
encoding="utf-8",
)
def _truncate_value(val: str) -> str:
if isinstance(val, str) and len(val) > MAX_VALUE_LENGTH:
return val[:MAX_VALUE_LENGTH].rstrip() + "…"
return val
def _recursive_update(target: dict, updates: dict) -> bool:
changed = False
for key, value in updates.items():
if value is None:
continue
if isinstance(value, str) and not value.strip():
continue
if isinstance(value, dict) and "value" not in value:
if key not in target or not isinstance(target[key], dict):
target[key] = {}
changed = True
if _recursive_update(target[key], value):
changed = True
else:
new_val = _truncate_value(str(value["value"] if isinstance(value, dict) else value))
entry = {"value": new_val, "updated": datetime.now().strftime("%Y-%m-%d")}
existing = target.get(key, {})
if not isinstance(existing, dict) or existing.get("value") != new_val:
target[key] = entry
changed = True
return changed
def update_memory(memory_update: dict) -> dict:
if not isinstance(memory_update, dict) or not memory_update:
return load_memory()
memory = load_memory()
if _recursive_update(memory, memory_update):
save_memory(memory)
print(f"[Memory] Saved: {list(memory_update.keys())}")
return memory
def format_memory_for_prompt(memory: dict | None) -> str:
if not memory:
return ""
lines = []
identity = memory.get("identity", {})
id_fields = ["name", "age", "birthday", "city", "job", "language", "school", "nationality"]
for field in id_fields:
entry = identity.get(field)
if entry:
val = entry.get("value") if isinstance(entry, dict) else entry
if val:
lines.append(f"{field.title()}: {val}")
for key, entry in identity.items():
if key in id_fields:
continue
val = entry.get("value") if isinstance(entry, dict) else entry
if val:
lines.append(f"{key.replace('_', ' ').title()}: {val}")
prefs = memory.get("preferences", {})
if prefs:
lines.append("")
lines.append("Preferences:")
for key, entry in list(prefs.items())[:15]:
val = entry.get("value") if isinstance(entry, dict) else entry
if val:
lines.append(f" - {key.replace('_', ' ').title()}: {val}")
projects = memory.get("projects", {})
if projects:
lines.append("")
lines.append("Active Projects / Goals:")
for key, entry in list(projects.items())[:8]:
val = entry.get("value") if isinstance(entry, dict) else entry
if val:
lines.append(f" - {key.replace('_', ' ').title()}: {val}")
rels = memory.get("relationships", {})
if rels:
lines.append("")
lines.append("People in their life:")
for key, entry in list(rels.items())[:10]:
val = entry.get("value") if isinstance(entry, dict) else entry
if val:
lines.append(f" - {key.replace('_', ' ').title()}: {val}")
wishes = memory.get("wishes", {})
if wishes:
lines.append("")
lines.append("Wishes / Plans / Wants:")
for key, entry in list(wishes.items())[:8]:
val = entry.get("value") if isinstance(entry, dict) else entry
if val:
lines.append(f" - {key.replace('_', ' ').title()}: {val}")
notes = memory.get("notes", {})
if notes:
lines.append("")
lines.append("Other notes:")
for key, entry in list(notes.items())[:8]:
val = entry.get("value") if isinstance(entry, dict) else entry
if val:
lines.append(f" - {key}: {val}")
if not lines:
return ""
header = "[WHAT YOU KNOW ABOUT THIS PERSON — use naturally, never recite like a list]\n"
result = header + "\n".join(lines)
if len(result) > 2000:
result = result[:1997] + "…"
return result + "\n"
def remember(key: str, value: str, category: str = "notes") -> str:
valid = {"identity", "preferences", "projects", "relationships", "wishes", "notes"}
if category not in valid:
category = "notes"
update_memory({category: {key: {"value": value}}})
return f"Remembered: {category}/{key} = {value}"
def forget(key: str, category: str = "notes") -> str:
memory = load_memory()
cat = memory.get(category, {})
if key in cat:
del cat[key]
memory[category] = cat
save_memory(memory)
return f"Forgotten: {category}/{key}"
return f"Not found: {category}/{key}"
forget_memory = forget
|