Spaces:
Running
Running
File size: 3,538 Bytes
3e46637 | 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 | """
backend/memory/evolutionary.py — Evolutionary Memory (S960)
Distilla preferenze utente, stili di codice e regole operative dalle sessioni.
Alimenta il layer 'Reflection' con conoscenze di alto livello (Long-term Evolution).
"""
import json
import logging
import os
from pathlib import Path
from datetime import datetime
from typing import List, Dict, Any
_logger = logging.getLogger("memory.evolutionary")
# Directory per i dati evolutivi (Sync con reflection.py)
_DATA_DIR = os.getenv('CHROMA_DATA_DIR') or ('/data' if Path('/data').exists() else '.')
EVO_PATH = Path(_DATA_DIR) / 'evolutionary_rules.json'
class EvolutionaryMemory:
def __init__(self, ai_client=None):
self.ai_client = ai_client
self.rules: Dict[str, Any] = {
"user_preferences": {}, # es. "language": "python", "style": "functional"
"operational_rules": [], # es. "Usa sempre pnpm invece di npm"
"domain_knowledge": {}, # es. "path/to/project": "description"
"last_updated": None
}
self._load()
def _load(self):
if EVO_PATH.exists():
try:
self.rules = json.loads(EVO_PATH.read_text())
except Exception as e:
_logger.error(f"[S960] Load error: {e}")
def _save(self):
try:
EVO_PATH.write_text(json.dumps(self.rules, indent=2, ensure_ascii=False))
except Exception as e:
_logger.error(f"[S960] Save error: {e}")
async def distill_and_evolve(self, session_summary: Dict[str, Any]):
"""
Prende un sommario distillato (dal MemoryDistiller) e aggiorna le regole evolutive.
"""
# 1. Estrazione euristica (in attesa di LLM integration)
# Se il sommario contiene fatti chiave, li integriamo
facts = session_summary.get("facts", [])
for fact in facts:
if ":" in fact:
k, v = fact.split(":", 1)
self.rules["domain_knowledge"][k.strip()] = v.strip()
# 2. Rilevamento preferenze (es. linguaggi usati con successo)
lessons = session_summary.get("lessons", [])
for lesson in lessons:
if lesson.get("type") == "success":
# Esempio: "Usato FastAPI con successo" -> preferenza per FastAPI
pass
self.rules["last_updated"] = datetime.now().isoformat()
self._save()
_logger.info("[S960] Memoria evolutiva aggiornata.")
def get_evolutionary_context(self) -> str:
"""
Ritorna una stringa formattata da iniettare nel System Prompt.
"""
if not self.rules["user_preferences"] and not self.rules["operational_rules"] and not self.rules["domain_knowledge"]:
return ""
context = "\n[MEMORIA EVOLUTIVA - REGOLE APPRESE]\n"
if self.rules["user_preferences"]:
context += "Preferenze Utente:\n"
for k, v in self.rules["user_preferences"].items():
context += f"- {k}: {v}\n"
if self.rules["operational_rules"]:
context += "Regole Operative:\n"
for rule in self.rules["operational_rules"]:
context += f"- {rule}\n"
if self.rules["domain_knowledge"]:
context += "Conoscenza Dominio:\n"
for k, v in self.rules["domain_knowledge"].items():
context += f"- {k}: {v}\n"
return context
# Singleton
evo_memory = EvolutionaryMemory()
|