Разработка AI-системы адаптивного обучения

AI-система адаптивного обучения

Адаптивное обучение — это не просто «предложи следующий урок». Это динамическая модель знаний студента, которая отслеживает паттерны ошибок, скорость усвоения, забывание (кривая Эббингауза) и перестраивает учебный план в реальном времени. Разница с линейными курсами: студент проходит только то, что ему нужно, в темпе, который ему подходит.

Модель знаний студента (Knowledge Tracing)

Основа — Deep Knowledge Tracing (DKT): LSTM/Transformer модель, предсказывающая вероятность правильного ответа на каждый навык с учётом истории ответов.

import torch
import torch.nn as nn
import numpy as np
import pandas as pd
from collections import defaultdict
from typing import Optional

class DeepKnowledgeTracer(nn.Module):
    """
    DKT: предсказывает P(correct | skill_id, history)
    Input: sequence of (skill_id, correct) pairs
    Output: probability of correct answer for each skill
    """

    def __init__(self, n_skills: int, hidden_dim: int = 128, n_layers: int = 2):
        super().__init__()
        self.n_skills = n_skills
        # Эмбеддинг: отдельно для (skill, correct=1) и (skill, correct=0)
        self.skill_embedding = nn.Embedding(n_skills * 2 + 1, hidden_dim)
        self.lstm = nn.LSTM(
            input_size=hidden_dim,
            hidden_size=hidden_dim,
            num_layers=n_layers,
            batch_first=True,
            dropout=0.2
        )
        self.output_layer = nn.Linear(hidden_dim, n_skills)
        self.sigmoid = nn.Sigmoid()

    def forward(self, skill_ids: torch.Tensor,
                correct: torch.Tensor) -> torch.Tensor:
        """
        skill_ids: (batch, seq_len)
        correct: (batch, seq_len) — 0 или 1
        Returns: (batch, seq_len, n_skills) — вероятности
        """
        # Encode: skill 0 правильно → index 1, skill 0 неправильно → index n_skills+1
        inputs = skill_ids + correct * self.n_skills
        emb = self.skill_embedding(inputs)
        hidden, _ = self.lstm(emb)
        logits = self.output_layer(hidden)
        return self.sigmoid(logits)

    def predict_mastery(self, history: list[dict]) -> dict:
        """
        history: [{'skill_id': 5, 'correct': 1}, ...]
        Returns: {skill_id: mastery_probability}
        """
        if not history:
            return {}

        skill_ids = torch.tensor([[h['skill_id'] for h in history]])
        correct = torch.tensor([[h['correct'] for h in history]])

        with torch.no_grad():
            probs = self.forward(skill_ids, correct)

        # Последний шаг = текущее состояние знаний
        last_probs = probs[0, -1, :].numpy()
        return {i: float(last_probs[i]) for i in range(self.n_skills)}


class ForgettingCurveModel:
    """Кривая Эббингауза: моделирование забывания"""

    def retention_probability(self, days_since_review: float,
                               review_count: int,
                               difficulty: float = 1.0) -> float:
        """
        Формула SuperMemo: R = e^(-t / S)
        S (stability) растёт с каждым повторением
        """
        base_stability = 1.0  # дней до 90% retention
        stability = base_stability * (2.0 ** review_count) / difficulty
        retention = np.exp(-days_since_review / stability)
        return float(np.clip(retention, 0, 1))

    def days_until_forgetting(self, review_count: int,
                               target_retention: float = 0.9,
                               difficulty: float = 1.0) -> float:
        """Через сколько дней retention упадёт ниже target"""
        base_stability = 1.0
        stability = base_stability * (2.0 ** review_count) / difficulty
        return -stability * np.log(target_retention)

    def get_due_skills(self, student_progress: pd.DataFrame,
                        target_retention: float = 0.85) -> list[dict]:
        """Навыки, требующие повторения сегодня"""
        due = []
        today = pd.Timestamp.now()

        for _, row in student_progress.iterrows():
            days_since = (today - row['last_reviewed']).days
            retention = self.retention_probability(
                days_since, row['review_count'], row['difficulty']
            )

            if retention < target_retention:
                due.append({
                    'skill_id': row['skill_id'],
                    'skill_name': row['skill_name'],
                    'current_retention': round(retention, 2),
                    'days_overdue': max(0, days_since - self.days_until_forgetting(
                        row['review_count'], target_retention, row['difficulty']
                    )),
                    'priority': (target_retention - retention) * row['importance']
                })

        return sorted(due, key=lambda x: -x['priority'])

Адаптивный планировщик учебного пути

class AdaptiveCurriculumPlanner:
    """Генерация персонального учебного плана"""

    def __init__(self, knowledge_graph: dict):
        """
        knowledge_graph: {skill_id: {'prerequisites': [...], 'difficulty': 1-5, 'name': '...'}}
        """
        self.knowledge_graph = knowledge_graph

    def get_next_items(self, mastery: dict,
                        student_profile: dict,
                        n_items: int = 5) -> list[dict]:
        """
        Выбор следующих заданий с учётом:
        - prerequisites (нельзя изучать тему без базы)
        - zone of proximal development (задания на 70-80% сложности)
        - forgetting (приоритет повторению)
        - learning_goal (к чему стремится студент)
        """
        candidates = []

        for skill_id, skill_info in self.knowledge_graph.items():
            current_mastery = mastery.get(skill_id, 0.0)

            # Пропускаем полностью освоенное (mastery > 0.9)
            if current_mastery > 0.9:
                continue

            # Проверка prerequisites
            prereqs_met = all(
                mastery.get(prereq, 0) > 0.7
                for prereq in skill_info.get('prerequisites', [])
            )
            if not prereqs_met:
                continue

            # Zone of Proximal Development: оптимальная сложность ≈ 70% угадаемость
            expected_success = current_mastery
            optimal_difficulty_fit = 1.0 - abs(expected_success - 0.70)

            # Приоритет навыкам, ведущим к цели
            goal_relevance = self._goal_alignment(skill_id, student_profile.get('goal'))

            candidates.append({
                'skill_id': skill_id,
                'skill_name': skill_info['name'],
                'difficulty': skill_info['difficulty'],
                'current_mastery': round(current_mastery, 2),
                'priority_score': optimal_difficulty_fit * 0.4 + goal_relevance * 0.6,
                'estimated_time_min': skill_info.get('avg_time_min', 15)
            })

        return sorted(candidates, key=lambda x: -x['priority_score'])[:n_items]

    def _goal_alignment(self, skill_id: int, goal: Optional[str]) -> float:
        """Релевантность навыка к учебной цели студента"""
        if not goal:
            return 0.5
        goal_skill_map = {
            'python_developer': {1, 2, 3, 5, 8, 13},
            'data_scientist': {1, 2, 4, 6, 9, 11, 14},
            'frontend_developer': {15, 16, 17, 20, 22},
        }
        relevant_skills = goal_skill_map.get(goal, set())
        return 1.0 if skill_id in relevant_skills else 0.3


class LearningSessionAdapter:
    """Адаптация в рамках одной учебной сессии"""

    def adapt_difficulty(self, recent_answers: list[bool],
                          current_difficulty: float) -> float:
        """Динамическая корректировка сложности внутри сессии"""
        if len(recent_answers) < 3:
            return current_difficulty

        recent_accuracy = sum(recent_answers[-5:]) / min(len(recent_answers), 5)

        # Target accuracy = 0.70 (Vygotsky's ZPD)
        if recent_accuracy > 0.85:
            # Слишком легко → увеличиваем сложность
            return min(1.0, current_difficulty + 0.1)
        elif recent_accuracy < 0.55:
            # Слишком трудно → снижаем
            return max(0.1, current_difficulty - 0.15)

        return current_difficulty

    def detect_frustration(self, session_events: list[dict]) -> bool:
        """Определение фрустрации: частые ошибки + долгое время ответа"""
        if len(session_events) < 5:
            return False

        recent = session_events[-5:]
        error_rate = sum(1 for e in recent if not e['correct']) / 5
        avg_time = np.mean([e['time_seconds'] for e in recent])

        return error_rate > 0.6 and avg_time > 45

Аналитика прогресса и отчётность

class LearningAnalyticsDashboard:
    def generate_student_report(self, student_id: str,
                                  mastery: dict,
                                  progress_history: pd.DataFrame) -> dict:
        """Еженедельный отчёт по студенту"""
        mastered_skills = sum(1 for m in mastery.values() if m > 0.8)
        total_skills = len(mastery)

        weekly_stats = progress_history[
            progress_history['date'] >= pd.Timestamp.now() - pd.Timedelta(days=7)
        ]

        return {
            'mastery_progress': f"{mastered_skills}/{total_skills} навыков",
            'completion_pct': round(mastered_skills / max(total_skills, 1) * 100, 1),
            'study_time_hours': weekly_stats['time_minutes'].sum() / 60,
            'avg_accuracy': weekly_stats['correct'].mean() if len(weekly_stats) > 0 else 0,
            'streak_days': self._calculate_streak(progress_history),
            'weakest_skills': [
                k for k, v in sorted(mastery.items(), key=lambda x: x[1])[:3]
            ],
            'strongest_skills': [
                k for k, v in sorted(mastery.items(), key=lambda x: -x[1])[:3]
            ]
        }

    def _calculate_streak(self, history: pd.DataFrame) -> int:
        if history.empty:
            return 0
        dates = sorted(history['date'].dt.date.unique(), reverse=True)
        streak = 0
        today = pd.Timestamp.now().date()
        for i, date in enumerate(dates):
            if (today - date).days == i:
                streak += 1
            else:
                break
        return streak

Сравнение подходов к адаптивному обучению

Корейское исследование 2023 на 12 000 студентов: DKT-адаптация сокращает время до освоения материала на 23% против линейного курса при той же итоговой оценке. Показатель завершения курсов (completion rate) растёт с 15-20% до 45-60%.