Spaces:
Configuration error
Configuration error
File size: 3,583 Bytes
e6fbc88 4cbd6c4 e6fbc88 4cbd6c4 e6fbc88 7cc12d8 e6fbc88 7cc12d8 e6fbc88 7cc12d8 e6fbc88 | 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 | """
reflection.py — Reflection Memory
Impara dagli errori, pattern di fallimento, strategie vincenti.
"""
import json
from pathlib import Path
from datetime import datetime
import logging
_logger = logging.getLogger("memory.reflection")
# D1: usa /data/ se disponibile (HF Spaces volume persistente) — evita perdita a restart
import os as _os_ref
_REFLECT_DATA_DIR = _os_ref.getenv('CHROMA_DATA_DIR') or ('/data' if Path('/data').exists() else '.')
REFLECTION_PATH = Path(_REFLECT_DATA_DIR) / 'reflection_memory.json'
class ReflectionMemory:
def __init__(self):
self._data: dict = {"failures": [], "successes": [], "patterns": {}}
self._load()
def _load(self):
if REFLECTION_PATH.exists():
try:
self._data = json.loads(REFLECTION_PATH.read_text())
except Exception as _exc:
_logger.debug("[reflection] silenced %s", type(_exc).__name__) # noqa: BLE001
def _save(self):
REFLECTION_PATH.write_text(json.dumps(self._data, indent=2, ensure_ascii=False))
def record_failure(self, task: str, error: str, attempted_strategy: str = ""):
entry = {
"ts": datetime.now().isoformat(),
# S600: task 300→500 — task description può essere più lunga
"task": task[:500],
"error": error[:500],
"strategy": attempted_strategy,
}
self._data["failures"].append(entry)
# Track pattern frequency
key = error[:120].lower() # S582: 80→120 (chiave pattern freq — conservativo)
self._data["patterns"][key] = self._data["patterns"].get(key, 0) + 1
# Cap at 200 entries
self._data["failures"] = self._data["failures"][-200:]
self._save()
def record_success(self, task: str, strategy: str):
entry = {
"ts": datetime.now().isoformat(),
# S600: task 300→500, strategy 300→500 — più contesto per replay futuro
"task": task[:500],
"strategy": strategy[:500],
}
self._data["successes"].append(entry)
self._data["successes"] = self._data["successes"][-100:]
self._save()
def get_relevant_lessons(self, task: str, n: int = 3) -> list[dict]:
"""Find past failures/successes relevant to current task."""
task_lower = task.lower()
words = set(task_lower.split())
def score(entry):
text = (entry.get("task", "") + " " + entry.get("strategy", "") + entry.get("error", "")).lower()
return sum(1 for w in words if w in text and len(w) > 3)
failures = sorted(self._data["failures"], key=score, reverse=True)[:n]
successes = sorted(self._data["successes"], key=score, reverse=True)[:n]
lessons = []
for f in failures[:2]:
if score(f) > 0:
lessons.append({"type": "failure", "task": f["task"], "avoid": f["error"]})
for s in successes[:1]:
if score(s) > 0:
lessons.append({"type": "success", "task": s["task"], "strategy": s["strategy"]})
return lessons
def get_top_patterns(self, n: int = 5) -> list[dict]:
return sorted(
[{"pattern": k, "count": v} for k, v in self._data["patterns"].items()],
key=lambda x: x["count"], reverse=True
)[:n]
def stats(self) -> dict:
return {
"failures": len(self._data["failures"]),
"successes": len(self._data["successes"]),
"patterns": len(self._data["patterns"]),
}
|