Spaces:
Running
Running
File size: 5,595 Bytes
28a08e7 | 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 | import logging
from .working import WorkingMemory
from .episodic import EpisodicMemory
from .semantic import SemanticMemory
from .reflection import ReflectionMemory
_logger = logging.getLogger("memory.manager")
class MemoryManager:
"""
S569: Unified Memory Manager (ARCH-K2.3).
Coordina i 4 layer di memoria dell'agente.
"""
def __init__(self, sb_client=None, chroma_client=None):
self.working = WorkingMemory()
self.episodic = EpisodicMemory()
self.semantic = SemanticMemory(sb_client, chroma_client)
self.reflection = ReflectionMemory()
async def init(self):
"""Inizializzazione asincrona (es. caricamento snapshot)."""
await self.semantic.init()
# S569: Auto-restore semantica se vuota
await self._auto_restore_semantic()
_logger.info("[MemoryManager] Layer inizializzati: working, episodic, semantic (pgvector=%s), reflection",
getattr(self.semantic, '_pgvector', False))
async def save_working(self, goal: str, plan: list, facts: list):
self.working.update(goal, plan, facts)
# S569: backup periodico della working memory su episodic
await self.save_episode("checkpoint", goal, f"Plan: {len(plan)} steps, Facts: {len(facts)}", True)
async def save_episode(self, type_: str, task: str, output: str, success: bool, tags: list | None = None):
self.episodic.add(type_, task, output, success, tags)
if self.semantic.available and task:
self.semantic.add(task, {"type": type_, "success": success})
async def search(self, query: str, n: int = 5, layer: str | None = None) -> list[dict]:
results = []
if layer in (None, "semantic") and self.semantic.available:
for h in self.semantic.search(query, n_results=n):
results.append({**h, "layer": "semantic"})
if layer in (None, "episodic"):
for ep in self.episodic.search_text(query, n=n):
results.append({
"content": f"{ep.task} → {ep.output[:300]}",
"layer": "episodic",
"type": ep.type,
"success": ep.success,
})
if layer == "reflection":
lessons = self.reflection.get_relevant_lessons(query, n=n)
results.extend([{**l, "layer": "reflection"} for l in lessons])
return results[:n]
async def reflect(self, task: str, output: str, success: bool, error: str | None = None) -> dict:
if success:
self.reflection.record_success(task, output[:500])
await self.save_episode("fix", task, output, True)
else:
self.reflection.record_failure(task, error or output[:500])
await self.save_episode("error", task, error or output[:500], False)
return {
"recorded": True,
"top_patterns": self.reflection.get_top_patterns(5),
"lessons": self.reflection.get_relevant_lessons(task, 4),
}
async def _auto_restore_semantic(self) -> None:
"""Auto-restore: se la semantic memory è vuota, carica l'ultimo snapshot da GitHub."""
import asyncio as _asyncio, os
if not self.semantic.available:
return
count = await _asyncio.to_thread(self.semantic.count)
if count > 0:
return
token = os.environ.get("GITHUB_TOKEN", "")
if not token:
return
try:
import urllib.request as _urq, json as _json, base64 as _b64
req = _urq.Request(
"https://api.github.com/repos/Baida98/AI/contents/data/semantic_snapshot.json",
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github.v3+json",
"User-Agent": "agente-ai-backend",
},
)
with _urq.urlopen(req, timeout=10) as resp:
meta = _json.loads(resp.read())
records = _json.loads(_b64.b64decode(meta["content"]).decode("utf-8"))
if not records:
return
result = await _asyncio.to_thread(self.semantic.import_all, records)
_logger.info("[MemoryManager] ✓ Auto-restore semantica: %d record da GitHub snapshot", result["imported"])
except Exception as exc:
_logger.debug("[MemoryManager] Auto-restore semantica: snapshot non disponibile (%s)", exc.__class__.__name__)
def stats(self) -> dict:
return {
"working": self.working.stats(),
"episodic": self.episodic.stats(),
"semantic": self.semantic.stats(),
"reflection": self.reflection.stats(),
}
async def clear(self, layer: str | None = None):
if layer in (None, "working"):
self.working.clear()
if layer in (None, "episodic"):
import sqlite3
if self.episodic._db:
try:
self.episodic._db.execute("DELETE FROM episodes")
self.episodic._db.commit()
except Exception as _e:
try:
self.episodic._db.rollback()
except Exception:
pass
raise RuntimeError(f"clear episodic fallito: {_e}") from _e
# ── Singleton globale — inizializzato in main.py _on_startup (GAP-5-fix) ─────
_global_manager: 'MemoryManager | None' = None
|