File size: 4,604 Bytes
0e2e9ee
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""tutor/adaptive.py — LearnerState with Bayesian Knowledge Tracing."""
from __future__ import annotations
import random
from collections import defaultdict
from typing import List, Dict, Any

BKT_DEFAULTS = {"p_init": 0.30, "p_learn": 0.10, "p_guess": 0.20, "p_slip": 0.10}

AGE_CONFIG = {
    5: {"diff_min": 1, "diff_max": 3,  "target_acc": 0.70},
    6: {"diff_min": 1, "diff_max": 4,  "target_acc": 0.72},
    7: {"diff_min": 2, "diff_max": 6,  "target_acc": 0.75},
    8: {"diff_min": 3, "diff_max": 7,  "target_acc": 0.75},
    9: {"diff_min": 4, "diff_max": 9,  "target_acc": 0.78},
}

DYSC_STREAK_THRESHOLD = 5
DYSC_MIN_ATTEMPTS     = 8


class SkillBKT:
    def __init__(self, skill: str, params: Dict | None = None):
        self.skill  = skill
        p           = params or BKT_DEFAULTS
        self.p_init  = p["p_init"]
        self.p_learn = p["p_learn"]
        self.p_guess = p["p_guess"]
        self.p_slip  = p["p_slip"]
        self.p_know  = self.p_init
        self.attempts     = 0
        self.correct      = 0
        self.streak_wrong = 0

    def update(self, is_correct: bool):
        self.attempts += 1
        if is_correct:
            self.correct += 1
            self.streak_wrong = 0
        else:
            self.streak_wrong += 1
        p_ok  = (1 - self.p_slip) if is_correct else self.p_slip
        p_nok = self.p_guess      if is_correct else (1 - self.p_guess)
        num   = p_ok * self.p_know
        den   = num + p_nok * (1 - self.p_know)
        p_kno = num / (den + 1e-9)
        self.p_know = p_kno + (1 - p_kno) * self.p_learn

    def mastery(self) -> float:
        return self.p_know

    def to_dict(self):
        return {k: getattr(self, k) for k in
                ("skill","p_know","p_init","p_learn","p_guess","p_slip",
                 "attempts","correct","streak_wrong")}

    @classmethod
    def from_dict(cls, d):
        obj = cls(d["skill"], {k: d[k] for k in ("p_init","p_learn","p_guess","p_slip")})
        obj.p_know       = d["p_know"]
        obj.attempts     = d["attempts"]
        obj.correct      = d["correct"]
        obj.streak_wrong = d.get("streak_wrong", 0)
        return obj


class LearnerState:
    def __init__(self, learner_id: str, lang: str = "en", age: int = 7):
        self.learner_id = learner_id
        self.lang       = lang
        self.age        = max(5, min(9, age))
        self.bkt: Dict[str, SkillBKT] = {}
        self._history: List[Dict] = []

    @property
    def age_config(self):
        return AGE_CONFIG.get(self.age, AGE_CONFIG[7])

    def _bkt(self, skill: str) -> SkillBKT:
        if skill not in self.bkt:
            self.bkt[skill] = SkillBKT(skill)
        return self.bkt[skill]

    def record_response(self, item: Dict, is_correct: bool):
        self._bkt(item.get("skill", "unknown")).update(is_correct)
        self._history.append({"item_id": item.get("id"), "correct": is_correct})

    def mastery(self, skill: str) -> float:
        return self._bkt(skill).mastery()

    def select_next_item(self, all_items: List[Dict], use_bkt: bool = True) -> Dict:
        cfg = self.age_config
        by_skill: Dict[str, List[Dict]] = defaultdict(list)
        for item in all_items:
            d = item.get("difficulty", 5)
            if cfg["diff_min"] <= d <= cfg["diff_max"]:
                by_skill[item.get("skill", "unknown")].append(item)
        if not by_skill:
            return random.choice(all_items)
        skills = sorted(by_skill.keys(), key=lambda s: self.mastery(s)) if use_bkt \
                 else list(by_skill.keys())
        for skill in skills:
            cands = by_skill[skill]
            if cands:
                m = self.mastery(skill)
                t = cfg["diff_min"] + m * (cfg["diff_max"] - cfg["diff_min"])
                cands.sort(key=lambda i: abs(i.get("difficulty", 5) - t))
                return cands[0]
        return random.choice(all_items)

    def dyscalculia_warning(self) -> List[str]:
        return [s for s, b in self.bkt.items()
                if b.attempts >= DYSC_MIN_ATTEMPTS and b.streak_wrong >= DYSC_STREAK_THRESHOLD]

    def to_dict(self):
        return {
            "learner_id": self.learner_id, "lang": self.lang, "age": self.age,
            "bkt": {k: v.to_dict() for k, v in self.bkt.items()},
            "history_len": len(self._history),
        }

    @classmethod
    def from_dict(cls, d):
        obj = cls(d["learner_id"], d.get("lang","en"), d.get("age",7))
        for skill, bd in d.get("bkt", {}).items():
            obj.bkt[skill] = SkillBKT.from_dict(bd)
        return obj