""" core/memory.py — Mémoire hiérarchique 6 tiers (SQLite) """ import sqlite3, hashlib, time from typing import List from dataclasses import dataclass from enum import Enum class MemoryTier(Enum): SENSORY = "sensory" WORKING = "working" EPISODIC = "episodic" SEMANTIC = "semantic" PROCEDURAL = "procedural" META = "meta" @dataclass class MemoryEntry: id: str; content: str; tier: MemoryTier; source: str importance: float = 0.5; confidence: float = 0.5; ts: float = 0.0 class HierarchicalMemory: def __init__(self, db_path: str = "/tmp/vortex_data/vortex_memory.db"): self.conn = sqlite3.connect(db_path, check_same_thread=False) self.conn.execute(""" CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, content TEXT, tier TEXT, source TEXT, importance REAL, confidence REAL, ts REAL ) """) self.conn.commit() def ingest(self, content: str, tier: MemoryTier, source: str, importance: float = 0.5, confidence: float = 0.5, **kwargs): mid = hashlib.sha256(content.encode()).hexdigest()[:16] self.conn.execute( "INSERT OR REPLACE INTO memories VALUES (?,?,?,?,?,?,?)", (mid, content[:2000], tier.value, source, importance, confidence, time.time()) ) self.conn.commit() def retrieve(self, query: str, tier: MemoryTier = MemoryTier.SEMANTIC, top_k: int = 5) -> List[MemoryEntry]: rows = self.conn.execute( "SELECT id, content, tier, source, importance, confidence, ts " "FROM memories WHERE tier=? ORDER BY importance DESC, ts DESC LIMIT ?", (tier.value, top_k) ).fetchall() return [MemoryEntry(r[0], r[1], MemoryTier(r[2]), r[3], r[4], r[5], r[6]) for r in rows] def stats(self) -> dict: total = self.conn.execute("SELECT COUNT(*) FROM memories").fetchone()[0] by_tier = { row[0]: row[1] for row in self.conn.execute( "SELECT tier, COUNT(*) FROM memories GROUP BY tier" ).fetchall() } return {"total": total, "by_tier": by_tier}