""" Memory System — conversations + progress tracking + spaced repetition. Schema: conversations (id, timestamp, user_input, response) progress (id, timestamp, subject, topic, correct, time_sec, question) sr_cards (id, topic, easiness, interval, repetitions, next_review, last_review) """ import sqlite3 import os from datetime import datetime, date, timedelta from typing import List, Tuple, Optional DB_PATH = os.path.join(os.path.dirname(__file__), "anveshai_memory.db") HISTORY_LIMIT = 10 def _get_connection() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def initialize_db() -> None: with _get_connection() as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS conversations ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, user_input TEXT NOT NULL, response TEXT NOT NULL ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS progress ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, subject TEXT NOT NULL, topic TEXT NOT NULL, correct INTEGER NOT NULL, time_sec REAL NOT NULL DEFAULT 0, question TEXT NOT NULL DEFAULT '' ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS sr_cards ( id INTEGER PRIMARY KEY AUTOINCREMENT, topic TEXT NOT NULL UNIQUE, easiness REAL NOT NULL DEFAULT 2.5, interval INTEGER NOT NULL DEFAULT 1, repetitions INTEGER NOT NULL DEFAULT 0, next_review TEXT NOT NULL, last_review TEXT NOT NULL ) """) conn.commit() # ───────────────────────────────────────────────────────────────────────────── # Conversation history # ───────────────────────────────────────────────────────────────────────────── def save_interaction(user_input: str, response: str) -> None: timestamp = datetime.now().isoformat(sep=" ", timespec="seconds") with _get_connection() as conn: conn.execute( "INSERT INTO conversations (timestamp, user_input, response) VALUES (?, ?, ?)", (timestamp, user_input, response), ) conn.commit() def get_recent_history(limit: int = HISTORY_LIMIT) -> List[sqlite3.Row]: with _get_connection() as conn: cursor = conn.execute( "SELECT id, timestamp, user_input, response FROM conversations " "ORDER BY id DESC LIMIT ?", (limit,) ) rows = cursor.fetchall() return list(reversed(rows)) def format_history() -> str: rows = get_recent_history() if not rows: return "No conversation history yet." lines = [f" Last {len(rows)} interaction(s):\n"] for row in rows: lines.append(f" [{row['timestamp']}]") lines.append(f" You : {row['user_input']}") lines.append(f" AnveshAI : {row['response']}") lines.append("") return "\n".join(lines).rstrip() def clear_history() -> None: with _get_connection() as conn: conn.execute("DELETE FROM conversations") conn.commit() # ───────────────────────────────────────────────────────────────────────────── # Progress tracking # ───────────────────────────────────────────────────────────────────────────── def save_progress(subject: str, topic: str, correct: bool, time_sec: float = 0.0, question: str = "") -> None: timestamp = datetime.now().isoformat(sep=" ", timespec="seconds") with _get_connection() as conn: conn.execute( "INSERT INTO progress (timestamp, subject, topic, correct, time_sec, question) " "VALUES (?, ?, ?, ?, ?, ?)", (timestamp, subject, topic, int(correct), time_sec, question), ) conn.commit() def get_progress_summary() -> dict: """Return dict: subject → {attempted, correct, accuracy, topics: {topic: {attempted, correct}}}""" with _get_connection() as conn: rows = conn.execute( "SELECT subject, topic, correct, time_sec FROM progress" ).fetchall() summary: dict = {} for row in rows: subj = row['subject'] topic = row['topic'] if subj not in summary: summary[subj] = {'attempted': 0, 'correct': 0, 'time_sec': 0.0, 'topics': {}} summary[subj]['attempted'] += 1 summary[subj]['correct'] += row['correct'] summary[subj]['time_sec'] += row['time_sec'] if topic not in summary[subj]['topics']: summary[subj]['topics'][topic] = {'attempted': 0, 'correct': 0} summary[subj]['topics'][topic]['attempted'] += 1 summary[subj]['topics'][topic]['correct'] += row['correct'] for subj in summary: a = summary[subj]['attempted'] c = summary[subj]['correct'] summary[subj]['accuracy'] = c / a * 100 if a > 0 else 0.0 return summary def get_weak_topics(threshold: float = 60.0, min_attempts: int = 1) -> List[Tuple[str, str, float]]: """Return [(subject, topic, accuracy%)] for topics below threshold.""" summary = get_progress_summary() weak = [] for subj, data in summary.items(): for topic, tdata in data['topics'].items(): if tdata['attempted'] >= min_attempts: acc = tdata['correct'] / tdata['attempted'] * 100 if acc < threshold: weak.append((subj, topic, acc)) weak.sort(key=lambda x: x[2]) return weak # ───────────────────────────────────────────────────────────────────────────── # Spaced repetition (SM-2 algorithm) # ───────────────────────────────────────────────────────────────────────────── def _today() -> str: return date.today().isoformat() def update_sr_card(topic: str, quality: int) -> None: """ Update SM-2 card for a topic. quality: 0-5 (0-1 = blackout, 2 = incorrect but close, 3 = correct+hard, 4 = correct, 5 = easy) """ today = _today() with _get_connection() as conn: row = conn.execute( "SELECT * FROM sr_cards WHERE topic = ?", (topic,) ).fetchone() if row is None: # New card easiness = 2.5 repetitions = 0 interval = 1 else: easiness = row['easiness'] repetitions = row['repetitions'] interval = row['interval'] # SM-2 update if quality < 3: repetitions = 0 interval = 1 else: if repetitions == 0: interval = 1 elif repetitions == 1: interval = 6 else: interval = round(interval * easiness) repetitions += 1 easiness = max(1.3, easiness + 0.1 - (5 - quality) * (0.08 + (5 - quality) * 0.02)) next_review = (date.today() + timedelta(days=interval)).isoformat() if row is None: conn.execute( "INSERT INTO sr_cards (topic, easiness, interval, repetitions, next_review, last_review) " "VALUES (?, ?, ?, ?, ?, ?)", (topic, easiness, interval, repetitions, next_review, today), ) else: conn.execute( "UPDATE sr_cards SET easiness=?, interval=?, repetitions=?, " "next_review=?, last_review=? WHERE topic=?", (easiness, interval, repetitions, next_review, today, topic), ) conn.commit() def get_due_topics() -> List[dict]: """Return list of topics due for review today.""" today = _today() with _get_connection() as conn: rows = conn.execute( "SELECT topic, easiness, interval, repetitions, next_review, last_review " "FROM sr_cards WHERE next_review <= ? ORDER BY next_review", (today,) ).fetchall() return [dict(r) for r in rows] def get_all_sr_cards() -> List[dict]: with _get_connection() as conn: rows = conn.execute( "SELECT topic, easiness, interval, repetitions, next_review, last_review " "FROM sr_cards ORDER BY next_review" ).fetchall() return [dict(r) for r in rows]