File size: 2,170 Bytes
2b3fd6a
 
8e3a425
2b3fd6a
8e3a425
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2b3fd6a
 
 
 
 
 
8e3a425
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2b3fd6a
 
 
 
 
 
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
"""
core/memory.py - Mémoire hiérarchique 6 tiers (SQLite)
"""
import sqlite3, hashlib, time, os
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 = None):
        if db_path is None:
            base_dir = os.getcwd()
            db_dir = os.path.join(base_dir, "data")
            os.makedirs(db_dir, exist_ok=True)
            db_path = os.path.join(db_dir, "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(
            id=r[0], content=r[1], tier=MemoryTier(r[2]),
            source=r[3], importance=r[4], confidence=r[5], ts=r[6]
        ) for r in rows]
    def close(self):
        self.conn.close()