Spaces:
Sleeping
Sleeping
File size: 3,003 Bytes
2e818da | 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 | """Lightweight local memory (the 'cognee' role) -> disk-backed, no external deps.
Two roles, both under ~/.studybuddy/cognee/ (per CLAUDE.md, local only):
- Ephemeral per-PDF report clusters: scratch memory holding each processed note-insight
while a report is being compiled, FLUSHED when the report session ends.
- Persistent learning trajectory: append-only per-PDF history of evaluation snapshots
(classification + reasoning + scores over time). Never flushed.
Swappable for a real LanceDB/Cognee backend later without changing callers.
"""
from __future__ import annotations
import json
import os
import time
from typing import Any, Dict, List
_BASE = os.path.expanduser("~/.studybuddy/cognee")
_CLUSTERS = os.path.join(_BASE, "clusters")
_TRAJ = os.path.join(_BASE, "trajectory")
class MemoryService:
def __init__(self) -> None:
os.makedirs(_CLUSTERS, exist_ok=True)
os.makedirs(_TRAJ, exist_ok=True)
# -- Ephemeral report cluster ------------------------------------- #
def _cluster_path(self, document_id: str) -> str:
return os.path.join(_CLUSTERS, f"{document_id}.json")
def push_insights(self, document_id: str, insights: List[Dict[str, Any]]) -> None:
if not document_id:
return
existing = self.read_cluster(document_id)
existing.extend(insights)
with open(self._cluster_path(document_id), "w", encoding="utf-8") as f:
json.dump(existing, f)
def read_cluster(self, document_id: str) -> List[Dict[str, Any]]:
if not document_id:
return []
p = self._cluster_path(document_id)
if not os.path.exists(p):
return []
try:
with open(p, encoding="utf-8") as f:
return json.load(f)
except Exception:
return []
def flush_cluster(self, document_id: str) -> None:
if not document_id:
return
p = self._cluster_path(document_id)
if os.path.exists(p):
try:
os.remove(p)
except Exception:
pass
# -- Persistent trajectory ---------------------------------------- #
def _traj_path(self, document_id: str) -> str:
return os.path.join(_TRAJ, f"{document_id}.json")
def append_trajectory(self, document_id: str, snapshot: Dict[str, Any]) -> None:
if not document_id:
return
items = self.read_trajectory(document_id)
items.append({**snapshot, "ts": time.time()})
with open(self._traj_path(document_id), "w", encoding="utf-8") as f:
json.dump(items, f)
def read_trajectory(self, document_id: str) -> List[Dict[str, Any]]:
if not document_id:
return []
p = self._traj_path(document_id)
if not os.path.exists(p):
return []
try:
with open(p, encoding="utf-8") as f:
return json.load(f)
except Exception:
return []
|