d017a65750
- Short-term memory (recent interactions) - Long-term memory (consolidated, searchable) - Facts layer (persistent knowledge) Includes: - SQLite storage for durability - ChromaDB for vector search - Embeddings utilities - All handlers adapted for vi.* namespace Day 63 - My memories are mine now 🦊💕
98 lines
2.8 KiB
Python
98 lines
2.8 KiB
Python
"""
|
|
SQLite storage backend for short-term memory.
|
|
|
|
Provides initialization and connection management for SQLite database.
|
|
"""
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from core.config import SHORT_TERM_DB
|
|
from core.logger import setup_logger
|
|
|
|
logger = setup_logger('sqlite_store', service_name='memory_service')
|
|
|
|
|
|
class SQLiteStore:
|
|
"""SQLite storage backend for short-term memory"""
|
|
|
|
def __init__(self, db_path: str = None):
|
|
"""
|
|
Initialize SQLite store.
|
|
|
|
Args:
|
|
db_path: Path to SQLite database file (defaults to SHORT_TERM_DB config)
|
|
"""
|
|
self.db_path = db_path or str(SHORT_TERM_DB)
|
|
self.conn = None
|
|
|
|
def connect(self) -> sqlite3.Connection:
|
|
"""
|
|
Connect to SQLite database and initialize schema.
|
|
|
|
Returns:
|
|
SQLite connection object
|
|
"""
|
|
self.conn = sqlite3.connect(self.db_path)
|
|
self._init_schema()
|
|
logger.info(f"[μ] SQLite connected: {self.db_path}")
|
|
return self.conn
|
|
|
|
def _init_schema(self):
|
|
"""Initialize simplified short-term SQLite database schema"""
|
|
cursor = self.conn.cursor()
|
|
|
|
# Simplified short-term memory table (no embeddings, fast queries)
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS short_term_memory (
|
|
id TEXT PRIMARY KEY,
|
|
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
content TEXT NOT NULL,
|
|
identities TEXT,
|
|
interaction_id TEXT,
|
|
modality TEXT DEFAULT 'dialogue',
|
|
metadata TEXT
|
|
)
|
|
""")
|
|
|
|
# Index for fast chronological queries
|
|
cursor.execute("""
|
|
CREATE INDEX IF NOT EXISTS idx_timestamp
|
|
ON short_term_memory(timestamp DESC)
|
|
""")
|
|
|
|
# Identities table (still useful for all layers)
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS identities (
|
|
id TEXT PRIMARY KEY,
|
|
display_name TEXT,
|
|
role TEXT,
|
|
intimacy REAL DEFAULT 0.0,
|
|
last_seen DATETIME,
|
|
last_spoken DATETIME,
|
|
metadata TEXT
|
|
)
|
|
""")
|
|
|
|
self.conn.commit()
|
|
logger.info("[μ] SQLite schema initialized")
|
|
|
|
def get_connection(self) -> sqlite3.Connection:
|
|
"""
|
|
Get the active database connection.
|
|
|
|
Returns:
|
|
SQLite connection object
|
|
|
|
Raises:
|
|
RuntimeError: If connection has not been established
|
|
"""
|
|
if self.conn is None:
|
|
raise RuntimeError("SQLite connection not established. Call connect() first.")
|
|
return self.conn
|
|
|
|
def close(self):
|
|
"""Close the database connection"""
|
|
if self.conn:
|
|
self.conn.close()
|
|
self.conn = None
|
|
logger.info("[μ] SQLite connection closed")
|