Spaces:
Sleeping
Sleeping
File size: 10,047 Bytes
582bf6b |
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 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 |
# modules/adaptive_engine.py
"""Adaptive difficulty engine for personalized learning"""
import json
import os
from typing import Dict, List, Optional
from datetime import datetime, timedelta
# Constants for adaptive logic
PERFORMANCE_HISTORY_LENGTH = 100
RECENT_HISTORY_WINDOW = 10
LEARNING_RATE_ALPHA = 0.2
# Thresholds for difficulty adjustment
DIFFICULTY_INCREASE_THRESHOLD = 0.8 # Recent success rate to increase difficulty
DIFFICULTY_DECREASE_THRESHOLD = 0.4 # Recent success rate to decrease difficulty
DIFFICULTY_SCORE_THRESHOLD_HARD = 0.7 # Medium score to recommend Hard
DIFFICULTY_SCORE_THRESHOLD_MEDIUM = 0.6 # Medium score to recommend Medium
class AdaptiveEngine:
def __init__(self, user_id: str):
self.user_id = user_id
self.performance_file = f"data/performance_{user_id}.json"
self.load_performance_data()
def load_performance_data(self):
"""Load user performance data"""
if os.path.exists(self.performance_file):
with open(self.performance_file, "r") as f:
self.data = json.load(f)
else:
self.data = {
"performance_history": [],
"difficulty_scores": {"Easy": 0.8, "Medium": 0.5, "Hard": 0.2},
"item_performance": {}, # Changed from topic_performance
"last_updated": datetime.now().isoformat(),
}
def save_performance_data(self):
"""Save performance data"""
os.makedirs("data", exist_ok=True)
self.data["last_updated"] = datetime.now().isoformat()
with open(self.performance_file, "w") as f:
json.dump(self.data, f, indent=2)
def update_performance(self, is_correct: bool, difficulty: str, item_id: str = None, topic: str = None):
"""Update performance based on quiz results or flashcard interaction"""
# Record overall performance history
performance_entry = {
"timestamp": datetime.now().isoformat(),
"correct": is_correct,
"difficulty": difficulty,
"item_id": item_id, # Store item_id if provided
"topic": topic,
}
self.data["performance_history"].append(performance_entry)
# Update difficulty scores using exponential moving average
alpha = LEARNING_RATE_ALPHA
score = 1.0 if is_correct else 0.0
old_score = self.data["difficulty_scores"].get(difficulty, 0.5)
new_score = alpha * score + (1 - alpha) * old_score
self.data["difficulty_scores"][difficulty] = new_score
# Update individual item performance (for spaced repetition)
if item_id:
if item_id not in self.data["item_performance"]:
self.data["item_performance"][item_id] = {
"attempts": 0,
"correct": 0,
"last_seen": None,
"topic": topic, # Store topic with item for context
}
self.data["item_performance"][item_id]["attempts"] += 1
if is_correct:
self.data["item_performance"][item_id]["correct"] += 1
self.data["item_performance"][item_id]["last_seen"] = datetime.now().isoformat()
# Keep only recent history (last 100 entries)
if len(self.data["performance_history"]) > PERFORMANCE_HISTORY_LENGTH:
self.data["performance_history"] = self.data["performance_history"][
-PERFORMANCE_HISTORY_LENGTH:
]
self.save_performance_data()
def get_recommended_difficulty(self) -> str:
"""Get recommended difficulty based on performance"""
scores = self.data["difficulty_scores"]
# Calculate recent performance (last 10 attempts)
recent_history = self.data["performance_history"][-RECENT_HISTORY_WINDOW:]
if recent_history:
recent_success_rate = sum(1 for h in recent_history if h["correct"]) / len(
recent_history
)
else:
return "Easy" # Default to Easy if no history
# Decision logic
if recent_success_rate > DIFFICULTY_INCREASE_THRESHOLD:
# Doing great, increase difficulty
if scores["Medium"] > DIFFICULTY_SCORE_THRESHOLD_HARD:
return "Hard"
else:
return "Medium"
elif recent_success_rate < DIFFICULTY_DECREASE_THRESHOLD:
# Struggling, decrease difficulty
return "Easy"
else:
# Balanced performance
if scores["Medium"] > DIFFICULTY_SCORE_THRESHOLD_MEDIUM:
return "Medium"
else:
return "Easy"
def get_weak_topics(self, limit: int = 5) -> List[str]:
"""Get topics where user needs more practice (based on item performance)"""
weak_topics = {} # Use dict to aggregate performance by topic
for item_id, performance in self.data["item_performance"].items():
topic = performance.get("topic")
if topic:
if topic not in weak_topics:
weak_topics[topic] = {"attempts": 0, "correct": 0}
weak_topics[topic]["attempts"] += performance["attempts"]
weak_topics[topic]["correct"] += performance["correct"]
sorted_weak_topics = []
for topic, agg_performance in weak_topics.items():
if agg_performance["attempts"] > 0:
success_rate = agg_performance["correct"] / agg_performance["attempts"]
if success_rate < 0.6:
sorted_weak_topics.append((topic, success_rate))
# Sort by success rate (ascending)
sorted_weak_topics.sort(key=lambda x: x[1])
return [topic for topic, _ in sorted_weak_topics[:limit]]
def get_strong_topics(self, limit: int = 5) -> List[str]:
"""Get topics where user excels (based on item performance)"""
strong_topics = {} # Use dict to aggregate performance by topic
for item_id, performance in self.data["item_performance"].items():
topic = performance.get("topic")
if topic:
if topic not in strong_topics:
strong_topics[topic] = {"attempts": 0, "correct": 0}
strong_topics[topic]["attempts"] += performance["attempts"]
strong_topics[topic]["correct"] += performance["correct"]
sorted_strong_topics = []
for topic, agg_performance in strong_topics.items():
if agg_performance["attempts"] >= 3: # Minimum attempts for strong topic
success_rate = agg_performance["correct"] / agg_performance["attempts"]
if success_rate > 0.8:
sorted_strong_topics.append((topic, success_rate))
# Sort by success rate (descending)
sorted_strong_topics.sort(key=lambda x: x[1], reverse=True)
return [topic for topic, _ in sorted_strong_topics[:limit]]
def should_review_item(self, item_id: str) -> bool:
"""Determine if an item (flashcard) needs review based on spaced repetition"""
if item_id not in self.data["item_performance"]:
return True # New item, should be reviewed
performance = self.data["item_performance"][item_id]
if performance["last_seen"]:
last_seen = datetime.fromisoformat(performance["last_seen"])
days_since = (datetime.now() - last_seen).days
# Spaced repetition intervals based on performance
success_rate = (
performance["correct"] / performance["attempts"]
if performance["attempts"] > 0
else 0
)
if success_rate < 0.5:
review_interval = 1 # Review daily
elif success_rate < 0.7:
review_interval = 3 # Review every 3 days
elif success_rate < 0.9:
review_interval = 7 # Review weekly
else:
review_interval = 14 # Review bi-weekly
return days_since >= review_interval
return True # Should be reviewed if no last_seen date
def get_items_due_for_review(self, topic: str = None, limit: int = 5) -> List[str]:
"""Get item_ids that are due for review for a given topic or all topics"""
review_items = []
for item_id, performance in self.data["item_performance"].items():
if (topic is None or performance.get("topic") == topic) and self.should_review_item(item_id):
review_items.append(item_id)
# Prioritize items with lower success rates
review_items.sort(key=lambda item_id: self.data["item_performance"][item_id]["correct"] / self.data["item_performance"][item_id]["attempts"] if self.data["item_performance"][item_id]["attempts"] > 0 else 0)
return review_items[:limit]
def get_performance_summary(self) -> Dict:
"""Get overall performance summary"""
total_attempts = len(self.data["performance_history"])
total_correct = sum(1 for h in self.data["performance_history"] if h["correct"])
summary = {
"total_attempts": total_attempts,
"total_correct": total_correct,
"overall_success_rate": total_correct / total_attempts
if total_attempts > 0
else 0,
"difficulty_mastery": self.data["difficulty_scores"],
"items_studied": len(self.data["item_performance"]), # Changed from topics_studied
"recommended_difficulty": self.get_recommended_difficulty(),
"weak_topics": self.get_weak_topics(3),
"strong_topics": self.get_strong_topics(3),
}
return summary
|