File size: 3,323 Bytes
6645e8f | 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 | # JARVIS Memory System — HuggingFace Edition
# SQLite3 persistent memory: facts, conversation history, preferences
# All paths relative / env-var driven — no hardcoded Windows paths
import sqlite3
import json
import datetime
import os
from config import DB_PATH, MEMORY_DIR
os.makedirs(MEMORY_DIR, exist_ok=True)
def _connect():
return sqlite3.connect(DB_PATH)
def init_db():
"""Create tables if they don't exist."""
with _connect() as conn:
conn.executescript("""
CREATE TABLE IF NOT EXISTS conversations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
role TEXT NOT NULL, -- 'user' or 'jarvis'
message TEXT NOT NULL,
timestamp TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS facts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
key TEXT NOT NULL UNIQUE, -- e.g. 'user_name'
value TEXT NOT NULL,
updated TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS preferences (
id INTEGER PRIMARY KEY AUTOINCREMENT,
category TEXT NOT NULL,
setting TEXT NOT NULL,
value TEXT NOT NULL
);
""")
print("[Memory] Database ready.")
def save_message(role: str, message: str):
"""Save a conversation turn."""
ts = datetime.datetime.now().isoformat()
with _connect() as conn:
conn.execute(
"INSERT INTO conversations (role, message, timestamp) VALUES (?,?,?)",
(role, message, ts)
)
def get_recent_history(n: int = 10) -> list[dict]:
"""Return the last n conversation turns."""
with _connect() as conn:
rows = conn.execute(
"SELECT role, message, timestamp FROM conversations ORDER BY id DESC LIMIT ?",
(n,)
).fetchall()
return [{"role": r[0], "message": r[1], "timestamp": r[2]} for r in reversed(rows)]
def save_fact(key: str, value: str):
"""Save or update a fact about the user."""
ts = datetime.datetime.now().isoformat()
with _connect() as conn:
conn.execute(
"INSERT INTO facts (key, value, updated) VALUES (?,?,?) "
"ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated=excluded.updated",
(key, value, ts)
)
def get_fact(key: str) -> str | None:
"""Retrieve a stored fact."""
with _connect() as conn:
row = conn.execute(
"SELECT value FROM facts WHERE key=?", (key,)
).fetchone()
return row[0] if row else None
def get_all_facts() -> dict:
"""Return all stored facts as a dict."""
with _connect() as conn:
rows = conn.execute("SELECT key, value FROM facts").fetchall()
return {r[0]: r[1] for r in rows}
def search_history(query: str, limit: int = 5) -> list[dict]:
"""Simple keyword search through conversation history."""
with _connect() as conn:
rows = conn.execute(
"SELECT role, message, timestamp FROM conversations "
"WHERE message LIKE ? ORDER BY id DESC LIMIT ?",
(f"%{query}%", limit)
).fetchall()
return [{"role": r[0], "message": r[1], "timestamp": r[2]} for r in rows]
|