教育 Agent 个性化学习与辅导完整实现
import time
import json
import math
import random
from typing import Dict, List, Any, Optional, Tuple, Set
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import numpy as np
from collections import deque, defaultdict
import threading
import uuid
from abc import ABC, abstractmethod
class KnowledgeLevel(Enum):
"""知识掌握水平"""
NOVICE = "novice"
BEGINNER = "beginner"
INTERMEDIATE = "intermediate"
ADVANCED = "advanced"
EXPERT = "expert"
class LearningStyle(Enum):
"""学习风格"""
VISUAL = "visual"
AUDITORY = "auditory"
READ_WRITE = "read_write"
KINESTHETIC = "kinesthetic"
class EngagementLevel(Enum):
"""参与度"""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
VERY_HIGH = "very_high"
class DifficultyLevel(Enum):
"""难度等级"""
VERY_EASY = "very_easy"
EASY = "easy"
MEDIUM = "medium"
HARD = "hard"
VERY_HARD = "very_hard"
@dataclass
class Student:
"""学生信息"""
student_id: str
name: str
age: int
grade: str
learning_style: LearningStyle
knowledge_levels: Dict[str, KnowledgeLevel]
learning_goals: List[str]
preferences: Dict[str, Any]
@dataclass
class KnowledgeComponent:
"""知识组件"""
kc_id: str
name: str
subject: str
grade_level: str
prerequisites: List[str]
learning_objectives: List[str]
difficulty: DifficultyLevel
estimated_time: int # minutes
@dataclass
class LearningActivity:
"""学习活动"""
activity_id: str
kc_id: str
activity_type: str # video, quiz, exercise, project
content_url: str
difficulty: DifficultyLevel
estimated_time: int
points: int
prerequisites: List[str]
@dataclass
class StudentResponse:
"""学生响应"""
response_id: str
student_id: str
activity_id: str
timestamp: datetime
answer: Any
is_correct: bool
time_spent: int # seconds
hints_used: int
confidence: float # 0-1
@dataclass
class KnowledgeState:
"""知识状态"""
student_id: str
kc_id: str
mastery_probability: float # 0-1
last_practiced: datetime
practice_count: int
correct_count: int
avg_response_time: float
decay_rate: float
@dataclass
class LearningPath:
"""学习路径"""
path_id: str
student_id: str
goal: str
activities: List[str]
estimated_duration: int
progress: float # 0-1
created_at: datetime
updated_at: datetime
@dataclass
class Feedback:
"""学习反馈"""
feedback_id: str
student_id: str
activity_id: str
feedback_type: str # corrective, explanatory, motivational
content: str
timestamp: datetime
effectiveness_score: Optional[float]
class KnowledgeTracer:
"""
知识追踪器
支持:
1. BKT 模型 (Bayesian Knowledge Tracing)
2. DKT 模型 (Deep Knowledge Tracing)
3. 知识状态更新
4. 遗忘曲线
"""
def __init__(self):
self.knowledge_states: Dict[str, Dict[str, KnowledgeState]] = defaultdict(dict)
# BKT 参数
self.p_init = 0.01 # 初始掌握概率
self.p_transit = 0.3 # 学习概率
self.p_slip = 0.1 # 失误概率
self.p_guess = 0.25 # 猜测概率
# 遗忘曲线参数 (Ebbinghaus)
self.forgetting_curve = lambda t: math.exp(-t / 1440) # t in minutes
def update_knowledge_state(self,
student_id: str,
kc_id: str,
response: StudentResponse) -> KnowledgeState:
"""更新知识状态"""
if kc_id not in self.knowledge_states[student_id]:
# 初始化知识状态
state = KnowledgeState(
student_id=student_id,
kc_id=kc_id,
mastery_probability=self.p_init,
last_practiced=response.timestamp,
practice_count=1,
correct_count=1 if response.is_correct else 0,
avg_response_time=response.time_spent,
decay_rate=0.1
)
else:
state = self.knowledge_states[student_id][kc_id]
# BKT 更新
p_correct = state.mastery_probability * (1 - self.p_slip) + \
(1 - state.mastery_probability) * self.p_guess
if response.is_correct:
# 答对：增加掌握概率
p_master_given_correct = (state.mastery_probability * (1 - self.p_slip)) / p_correct if p_correct > 0 else 0
state.mastery_probability = p_master_given_correct + \
(1 - p_master_given_correct) * self.p_transit
state.correct_count += 1
else:
# 答错：降低掌握概率
p_master_given_incorrect = (state.mastery_probability * self.p_slip) / \
(1 - p_correct) if (1 - p_correct) > 0 else 0
state.mastery_probability = p_master_given_incorrect
state.practice_count += 1
state.avg_response_time = (state.avg_response_time * (state.practice_count - 1) +
response.time_spent) / state.practice_count
state.last_practiced = response.timestamp
# 应用遗忘曲线
time_since_last = (datetime.now() - state.last_practiced).total_seconds() / 60
forgetting_factor = self.forgetting_curve(time_since_last)
state.mastery_probability *= forgetting_factor
state.mastery_probability = max(0, min(1, state.mastery_probability))
self.knowledge_states[student_id][kc_id] = state
return state
def get_knowledge_level(self, student_id: str, kc_id: str) -> KnowledgeLevel:
"""获取知识水平"""
if student_id not in self.knowledge_states or kc_id not in self.knowledge_states[student_id]:
return KnowledgeLevel.NOVICE
mastery = self.knowledge_states[student_id][kc_id].mastery_probability
if mastery >= 0.9:
return KnowledgeLevel.EXPERT
elif mastery >= 0.7:
return KnowledgeLevel.ADVANCED
elif mastery >= 0.5:
return KnowledgeLevel.INTERMEDIATE
elif mastery >= 0.3:
return KnowledgeLevel.BEGINNER
else:
return KnowledgeLevel.NOVICE
def predict_forgetting(self, student_id: str, kc_id: str,
hours_ahead: int) -> float:
"""预测遗忘"""
if student_id not in self.knowledge_states or kc_id not in self.knowledge_states[student_id]:
return 0.0
current_mastery = self.knowledge_states[student_id][kc_id].mastery_probability
minutes_ahead = hours_ahead * 60
forgetting_factor = self.forgetting_curve(minutes_ahead)
return current_mastery * forgetting_factor
class IntelligentTutor:
"""
智能导师系统
支持:
1. 个性化教学
2. 即时反馈
3. 提示生成
4. 难度调整
"""
def __init__(self, knowledge_tracer: KnowledgeTracer):
self.knowledge_tracer = knowledge_tracer
self.feedback_history = deque(maxlen=1000)
# 提示模板
self.hint_templates = {
'conceptual': [
"回想一下这个概念的定义...",
"考虑一下这个原理的核心思想...",
"试着从另一个角度思考这个问题..."
],
'procedural': [
"第一步应该是什么？",
"还记得我们学过的类似方法吗？",
"试着分解这个问题..."
],
'strategic': [
"有没有更简单的方法？",
"这个模式和之前学的有什么联系？",
"如果换个思路会怎样？"
]
}
def generate_feedback(self,
student: Student,
activity: LearningActivity,
response: StudentResponse) -> Feedback:
"""生成反馈"""
# 获取知识状态
knowledge_level = self.knowledge_tracer.get_knowledge_level(
student.student_id, activity.kc_id
)
# 生成反馈内容
if response.is_correct:
if response.confidence > 0.8:
feedback_content = f"太棒了！你完全理解了这个概念。{self._generate_encouragement(student)}"
feedback_type = "motivational"
else:
feedback_content = "回答正确！但看起来你还有些不确定。让我再解释一下..."
feedback_type = "explanatory"
else:
if response.hints_used > 0:
feedback_content = "没关系，让我们一起看看哪里出了问题。"
feedback_type = "corrective"
else:
feedback_content = "这个答案不太对。别担心，错误是学习的一部分。让我给你一些提示..."
feedback_type = "corrective"
feedback = Feedback(
feedback_id=f"fb_{uuid.uuid4().hex[:8]}",
student_id=student.student_id,
activity_id=activity.activity_id,
feedback_type=feedback_type,
content=feedback_content,
timestamp=datetime.now(),
effectiveness_score=None
)
self.feedback_history.append(feedback)
return feedback
def generate_hint(self,
student: Student,
activity: LearningActivity,
attempt_number: int) -> str:
"""生成提示"""
# 根据尝试次数选择提示类型
if attempt_number == 1:
hint_type = 'conceptual'
elif attempt_number == 2:
hint_type = 'procedural'
else:
hint_type = 'strategic'
# 随机选择提示
hints = self.hint_templates.get(hint_type, self.hint_templates['conceptual'])
hint = random.choice(hints)
# 个性化提示
if student.learning_style == LearningStyle.VISUAL:
hint += " 试着画个图来理解。"
elif student.learning_style == LearningStyle.AUDITORY:
hint += " 试着大声解释给自己听。"
return hint
def adjust_difficulty(self,
student: Student,
current_difficulty: DifficultyLevel,
recent_performance: List[StudentResponse]) -> DifficultyLevel:
"""调整难度"""
if not recent_performance:
return current_difficulty
# 计算正确率
correct_rate = sum(1 for r in recent_performance if r.is_correct) / len(recent_performance)
# 计算平均响应时间
avg_time = sum(r.time_spent for r in recent_performance) / len(recent_performance)
# 调整难度
difficulty_levels = [DifficultyLevel.VERY_EASY, DifficultyLevel.EASY,
DifficultyLevel.MEDIUM, DifficultyLevel.HARD, DifficultyLevel.VERY_HARD]
current_idx = difficulty_levels.index(current_difficulty)
if correct_rate > 0.8 and avg_time < 60: # 高正确率且快速
new_idx = min(current_idx + 1, len(difficulty_levels) - 1)
elif correct_rate < 0.5 or avg_time > 180: # 低正确率或很慢
new_idx = max(current_idx - 1, 0)
else:
new_idx = current_idx
return difficulty_levels[new_idx]
def _generate_encouragement(self, student: Student) -> str:
"""生成鼓励语"""
encouragements = [
"继续保持！",
"你的进步很明显！",
"做得很好！",
"你正在掌握这个技能！"
]
return random.choice(encouragements)
class LearningPathPlanner:
"""
学习路径规划器
支持:
1. 目标分解
2. 先决条件检查
3. 个性化推荐
4. 进度跟踪
"""
def __init__(self):
self.learning_paths: Dict[str, LearningPath] = {}
self.activity_database: Dict[str, LearningActivity] = {}
self.knowledge_graph: Dict[str, List[str]] = defaultdict(list)
def create_learning_path(self,
student: Student,
goal: str,
available_activities: List[LearningActivity]) -> LearningPath:
"""创建学习路径"""
# 基于学生知识水平推荐活动
recommended_activities = []
for activity in available_activities:
# 检查先决条件
if self._check_prerequisites(student, activity):
# 计算匹配度
match_score = self._calculate_match_score(student, activity)
if match_score > 0.6:
recommended_activities.append((activity, match_score))
# 按匹配度排序
recommended_activities.sort(key=lambda x: x[1], reverse=True)
# 选择前 N 个活动
selected_activities = [a[0].activity_id for a in recommended_activities[:20]]
# 估算总时间
total_time = sum(
next(a.estimated_time for a in available_activities if a.activity_id == aid)
for aid in selected_activities
)
path = LearningPath(
path_id=f"path_{uuid.uuid4().hex[:8]}",
student_id=student.student_id,
goal=goal,
activities=selected_activities,
estimated_duration=total_time,
progress=0.0,
created_at=datetime.now(),
updated_at=datetime.now()
)
self.learning_paths[path.path_id] = path
return path
def update_progress(self, path_id: str, completed_activity_id: str) -> float:
"""更新进度"""
if path_id not in self.learning_paths:
return 0.0
path = self.learning_paths[path_id]
# 计算进度
completed_count = sum(1 for aid in path.activities[:path.activities.index(completed_activity_id)+1])
path.progress = completed_count / len(path.activities)
path.updated_at = datetime.now()
return path.progress
def recommend_next_activity(self,
student: Student,
path: LearningPath,
completed_activities: Set[str]) -> Optional[str]:
"""推荐下一个活动"""
for activity_id in path.activities:
if activity_id not in completed_activities:
# 检查先决条件
activity = self.activity_database.get(activity_id)
if activity and self._check_prerequisites(student, activity):
return activity_id
return None
def _check_prerequisites(self,
student: Student,
activity: LearningActivity) -> bool:
"""检查先决条件"""
for prereq_kc in activity.prerequisites:
level = self.knowledge_tracer.get_knowledge_level(student.student_id, prereq_kc)
if level in [KnowledgeLevel.NOVICE, KnowledgeLevel.BEGINNER]:
return False
return True
def _calculate_match_score(self,
student: Student,
activity: LearningActivity) -> float:
"""计算匹配度"""
score = 0.5 # 基础分
# 学习风格匹配
if activity.activity_type == 'video' and student.learning_style == LearningStyle.VISUAL:
score += 0.2
elif activity.activity_type == 'exercise' and student.learning_style == LearningStyle.KINESTHETIC:
score += 0.2
# 难度匹配
knowledge_level = self.knowledge_tracer.get_knowledge_level(student.student_id, activity.kc_id)
difficulty_map = {
KnowledgeLevel.NOVICE: DifficultyLevel.VERY_EASY,
KnowledgeLevel.BEGINNER: DifficultyLevel.EASY,
KnowledgeLevel.INTERMEDIATE: DifficultyLevel.MEDIUM,
KnowledgeLevel.ADVANCED: DifficultyLevel.HARD,
KnowledgeLevel.EXPERT: DifficultyLevel.VERY_HARD
}
if difficulty_map.get(knowledge_level) == activity.difficulty:
score += 0.3
return min(1.0, score)
class AdaptiveLearningEngine:
"""
自适应学习引擎
支持:
1. 实时调整
2. 个性化推荐
3. 学习分析
4. 效果评估
"""
def __init__(self):
self.knowledge_tracer = KnowledgeTracer()
self.tutor = IntelligentTutor(self.knowledge_tracer)
self.path_planner = LearningPathPlanner()
self.learning_sessions: Dict[str, List[StudentResponse]] = defaultdict(list)
def start_learning_session(self, student: Student) -> Dict[str, Any]:
"""开始学习会话"""
# 获取推荐的学习路径
path = self._get_or_create_path(student)
# 推荐第一个活动
completed = set()
next_activity_id = self.path_planner.recommend_next_activity(student, path, completed)
return {
'session_id': f"session_{uuid.uuid4().hex[:8]}",
'student_id': student.student_id,
'path_id': path.path_id,
'next_activity_id': next_activity_id,
'timestamp': datetime.now()
}
def process_response(self,
student: Student,
activity: LearningActivity,
response: StudentResponse) -> Dict[str, Any]:
"""处理学生响应"""
# 更新知识状态
knowledge_state = self.knowledge_tracer.update_knowledge_state(
student.student_id, activity.kc_id, response
)
# 生成反馈
feedback = self.tutor.generate_feedback(student, activity, response)
# 记录响应
self.learning_sessions[student.student_id].append(response)
# 获取下一个活动
path = self.path_planner.learning_paths.get(response.response_id.split('_')[0])
if path:
completed = {r.activity_id for r in self.learning_sessions[student.student_id]}
next_activity_id = self.path_planner.recommend_next_activity(student, path, completed)
else:
next_activity_id = None
# 调整难度
recent_responses = self.learning_sessions[student.student_id][-5:]
new_difficulty = self.tutor.adjust_difficulty(student, activity.difficulty, recent_responses)
return {
'knowledge_state': {
'kc_id': knowledge_state.kc_id,
'mastery': knowledge_state.mastery_probability,
'level': knowledge_state.mastery_probability
},
'feedback': feedback.content,
'next_activity_id': next_activity_id,
'new_difficulty': new_difficulty.value,
'progress': self._calculate_overall_progress(student)
}
def get_learning_analytics(self, student: Student) -> Dict[str, Any]:
"""获取学习分析"""
responses = self.learning_sessions.get(student.student_id, [])
if not responses:
return {'error': 'No learning data available'}
# 计算统计
total_activities = len(responses)
correct_rate = sum(1 for r in responses if r.is_correct) / total_activities
avg_time = sum(r.time_spent for r in responses) / total_activities
avg_hints = sum(r.hints_used for r in responses) / total_activities
# 知识掌握分布
knowledge_distribution = defaultdict(int)
for kc_id in set(r.activity_id for r in responses):
level = self.knowledge_tracer.get_knowledge_level(student.student_id, kc_id)
knowledge_distribution[level.value] += 1
# 参与度评估
if correct_rate > 0.8 and avg_time < 90:
engagement = EngagementLevel.VERY_HIGH
elif correct_rate > 0.6:
engagement = EngagementLevel.HIGH
elif correct_rate > 0.4:
engagement = EngagementLevel.MEDIUM
else:
engagement = EngagementLevel.LOW
return {
'total_activities': total_activities,
'correct_rate': correct_rate,
'avg_time_seconds': avg_time,
'avg_hints': avg_hints,
'knowledge_distribution': dict(knowledge_distribution),
'engagement_level': engagement.value,
'recommendations': self._generate_recommendations(student, correct_rate, engagement)
}
def _get_or_create_path(self, student: Student) -> LearningPath:
"""获取或创建学习路径"""
# 简化实现：创建新路径
mock_activities = [
LearningActivity(
activity_id=f"act_{i}",
kc_id=f"kc_{i}",
activity_type='quiz',
content_url=f'url_{i}',
difficulty=DifficultyLevel.MEDIUM,
estimated_time=10,
points=10,
prerequisites=[]
)
for i in range(10)
]
return self.path_planner.create_learning_path(
student,
"Master the subject",
mock_activities
)
def _calculate_overall_progress(self, student: Student) -> float:
"""计算总体进度"""
responses = self.learning_sessions.get(student.student_id, [])
if not responses:
return 0.0
# 简化：基于正确率
correct_rate = sum(1 for r in responses if r.is_correct) / len(responses)
return correct_rate
def _generate_recommendations(self,
student: Student,
correct_rate: float,
engagement: EngagementLevel) -> List[str]:
"""生成建议"""
recommendations = []
if correct_rate < 0.5:
recommendations.append("建议复习基础概念，多做练习。")
if engagement == EngagementLevel.LOW:
recommendations.append("尝试更有趣的学习活动，保持学习动机。")
if correct_rate > 0.8:
recommendations.append("表现优秀！可以尝试更有挑战性的内容。")
return recommendations
# 使用示例
if __name__ == "__main__":
print("=== 教育 Agent 个性化学习与辅导 ===\n")
print("=== 创建自适应学习引擎 ===")
engine = AdaptiveLearningEngine()
print(f"\n=== 创建学生信息 ===")
student = Student(
student_id="STU001",
name="李明",
age=15,
grade="10",
learning_style=LearningStyle.VISUAL,
knowledge_levels={'math': KnowledgeLevel.INTERMEDIATE},
learning_goals=['master_algebra', 'prepare_exam'],
preferences={'session_duration': 30, 'difficulty_preference': 'medium'}
)
print(f"学生：{student.name} ({student.age}岁，{student.grade}年级)")
print(f"学习风格：{student.learning_style.value}")
print(f"学习目标：{', '.join(student.learning_goals)}")
print(f"\n=== 开始学习会话 ===")
session = engine.start_learning_session(student)
print(f"会话 ID: {session['session_id']}")
print(f"学习路径：{session['path_id']}")
print(f"推荐活动：{session['next_activity_id']}")
print(f"\n=== 模拟学习活动 ===")
# 创建模拟活动
activity = LearningActivity(
activity_id="act_math_001",
kc_id="kc_algebra_001",
activity_type='quiz',
content_url='url_quiz_001',
difficulty=DifficultyLevel.MEDIUM,
estimated_time=10,
points=10,
prerequisites=[]
)
# 模拟学生响应
response = StudentResponse(
response_id=f"resp_{uuid.uuid4().hex[:8]}",
student_id=student.student_id,
activity_id=activity.activity_id,
timestamp=datetime.now(),
answer="x = 5",
is_correct=True,
time_spent=45,
hints_used=0,
confidence=0.85
)
print(f"活动：{activity.activity_id}")
print(f"类型：{activity.activity_type}")
print(f"难度：{activity.difficulty.value}")
print(f"学生答案：{response.answer}")
print(f"是否正确：{response.is_correct}")
print(f"用时：{response.time_spent}秒")
print(f"置信度：{response.confidence}")
print(f"\n=== 处理响应 ===")
result = engine.process_response(student, activity, response)
print(f"知识状态:")
print(f" 知识点：{result['knowledge_state']['kc_id']}")
print(f" 掌握度：{result['knowledge_state']['mastery']:.2%}")
print(f"反馈：{result['feedback']}")
print(f"下一个活动：{result['next_activity_id']}")
print(f"新难度：{result['new_difficulty']}")
print(f"总体进度：{result['progress']:.2%}")
print(f"\n=== 获取学习分析 ===")
# 模拟更多响应
for i in range(4):
mock_response = StudentResponse(
response_id=f"resp_{i}",
student_id=student.student_id,
activity_id=f"act_{i}",
timestamp=datetime.now(),
answer=f"answer_{i}",
is_correct=random.random() > 0.3,
time_spent=random.randint(30, 120),
hints_used=random.randint(0, 2),
confidence=random.uniform(0.5, 1.0)
)
engine.learning_sessions[student.student_id].append(mock_response)
analytics = engine.get_learning_analytics(student)
print(f"学习分析:")
print(f" 总活动数：{analytics['total_activities']}")
print(f" 正确率：{analytics['correct_rate']:.2%}")
print(f" 平均用时：{analytics['avg_time_seconds']:.1f}秒")
print(f" 平均提示：{analytics['avg_hints']:.1f}次")
print(f" 参与度：{analytics['engagement_level']}")
print(f" 知识分布：{analytics['knowledge_distribution']}")
print(f" 建议：{', '.join(analytics['recommendations'])}")
print(f"\n关键观察:")
print("1. 学习感知：知识追踪 + 情感识别 + 学习分析")
print("2. 智能辅导：即时反馈 + 个性化提示 + 难度调整")
print("3. 个性化学习：学习路径 + 自适应推荐 + 成长追踪")
print("4. 自适应引擎：实时调整 + 效果评估 + 持续优化")
print("5. 智能教育：感知 + 辅导 + 个性 = 可信赖")
print("\n智能教育的使命：让学习更高效、更个性、更有趣")