教育行业数据分析：AI 学情诊断与个性化学习路径推荐实战

教育行业数据分析：AI 学情诊断与个性化学习路径推荐实战

一、背景与需求

最近接了个教育行业的数据分析项目，一家在线教育平台想要通过 AI 技术实现学情诊断和个性化学习路径推荐。他们的痛点很明确：学生数量上来了，但完课率和续费率一直在掉。运营团队想知道到底哪个环节出了问题，教学团队想知道怎么能给每个学生推最合适的学习内容。

拆解下来，核心需求有这么几点：

这个场景其实非常典型，本质上就是一个"数据采集 → 特征工程 → 模型诊断 → 推荐排序 → 效果反馈"的完整链路。下面咱们一步步来拆解。

二、数据采集与特征工程

教育场景的数据维度其实挺丰富的，关键是怎么把这些原始行为转化为可用的特征。来看一下核心数据表的字段：

import pandas as pd
import numpy as np
from datetime import datetime

# —- 1. 学生做题记录表 —-
# 每条记录代表一次答题行为
exercise_log = pd.DataFrame({
'student_id': ['S001', 'S001', 'S002', 'S002', 'S003'],
'question_id': ['Q101', 'Q102', 'Q101', 'Q103', 'Q101'],
'is_correct': [1, 0, 1, 0, 1], # 1=答对, 0=答错
'knowledge_point': ['一元二次方程', '一元二次方程', '一次函数', '二次函数', '一元二次方程'],
'response_time_sec': [45, 120, 30, 90, 50], # 答题耗时（秒）
'attempt_count': [1, 1, 1, 2, 1], # 第几次尝试
'timestamp': pd.to_datetime(['2026-07-01 10:00', '2026-07-01 10:05',
'2026-07-01 11:00', '2026-07-02 09:00',
'2026-07-02 10:00'])
})

# —- 2. 视频观看记录表 —-
video_log = pd.DataFrame({
'student_id': ['S001', 'S002', 'S003', 'S001'],
'video_id': ['V01', 'V01', 'V02', 'V03'],
'knowledge_point': ['一元二次方程', '一元二次方程', '一次函数', '二次函数'],
'watch_duration_sec': [3600, 1800, 2400, 1200],
'total_duration_sec': [3600, 3600, 2400, 3600], # 视频总时长
'timestamp': pd.to_datetime(['2026-07-01 09:00', '2026-07-01 10:00',
'2026-07-01 14:00', '2026-07-02 08:00'])
})

接下来做特征工程。教育场景最有用的特征之一就是"知识点掌握度"，这里我们借鉴 IRT（项目反应理论）的思路，用一个简化的公式来计算：

# —- 知识点掌握度特征计算 —-
# 思路：综合考虑 正确率、答题速度、尝试次数、视频完成度 四个维度

def calculate_knowledge_mastery(exercise_df, video_df):
"""
计算每个学生对每个知识点的掌握度得分

参数:
exercise_df: 做题记录 DataFrame
video_df: 视频观看记录 DataFrame

返回:
mastery_score: DataFrame，包含 student_id, knowledge_point, mastery_score
"""

# — 维度1：答题正确率（exercises） —
exercise_stats = exercise_df.groupby(['student_id', 'knowledge_point']).agg(
total_questions=('question_id', 'count'),
correct_questions=('is_correct', 'sum'),
avg_response_time=('response_time_sec', 'mean'),
max_attempts=('attempt_count', 'max')
).reset_index()

# 正确率 = 答对数 / 总题数
exercise_stats['accuracy'] = exercise_stats['correct_questions'] / exercise_stats['total_questions']

# — 维度2：视频完成度 —
video_stats = video_df.groupby(['student_id', 'knowledge_point']).agg(
total_watch=('watch_duration_sec', 'sum'),
total_available=('total_duration_sec', 'sum')
).reset_index()

# 完成度 = 实际观看时长 / 视频总时长
video_stats['completion_rate'] = video_stats['total_watch'] / video_stats['total_available']
video_stats['completion_rate'] = video_stats['completion_rate'].clip(0, 1) # 控制到 [0, 1]

# — 合并数据 —
mastery_df = pd.merge(exercise_stats, video_stats,
on=['student_id', 'knowledge_point'],
how='outer').fillna(0)

# — 答题速度归一化：用时越短说明越熟练 —
# 将答题时间映射到 [0, 1] 区间，用时越短得分越高
max_time = mastery_df['avg_response_time'].max()
if max_time > 0:
mastery_df['speed_score'] = 1 – (mastery_df['avg_response_time'] / max_time)
else:
mastery_df['speed_score'] = 0

# — 综合掌握度得分 —
# 权重分配：正确率 0.5 + 视频完成度 0.2 + 速度得分 0.2 + 尝试次数惩罚 0.1
mastery_df['attempt_penalty'] = 1 / mastery_df['max_attempts'].clip(lower=1)

mastery_df['mastery_score'] = (
0.5 * mastery_df['accuracy'] + # 正确率权重最高
0.2 * mastery_df['completion_rate'] + # 视频学习完成度
0.2 * mastery_df['speed_score'] + # 答题速度
0.1 * mastery_df['attempt_penalty'] # 尝试次数惩罚
)

return mastery_df[['student_id', 'knowledge_point', 'mastery_score']]

# 调用函数
mastery_result = calculate_knowledge_mastery(exercise_log, video_log)
print("=== 知识点掌握度得分 ===")
print(mastery_result)

三、学情诊断：知识追踪与薄弱点定位

有了掌握度得分，还需要能动态追踪学生的知识状态变化。这里引入贝叶斯知识追踪（BKT）的思想：

from sklearn.preprocessing import LabelEncoder
from collections import defaultdict

# —- 简化的贝叶斯知识追踪模型 —-
class SimplifiedBKT:
"""
简化的贝叶斯知识追踪模型

BKT 核心思想：学生的知识状态是隐变量，我们通过观测行为（答题对错）
来推断其真实掌握概率。这里用滑动窗口方法来近似。
"""

def __init__(self, window_size=10):
self.window_size = window_size # 滑动窗口大小，看最近N次的表现
self.student_kp_history = defaultdict(list) # 按学生+知识点存储历史记录

def add_observation(self, student_id, knowledge_point, is_correct, timestamp):
"""添加一次答题观测"""
self.student_kp_history[(student_id, knowledge_point)].append({
'correct': is_correct,
'time': timestamp
})

def get_current_mastery(self, student_id, knowledge_point):
"""
获取当前掌握概率

使用最近N次答题的加权正确率，越近的观测权重越大
"""
key = (student_id, knowledge_point)
history = self.student_kp_history[key]

if not history:
return 0.0 # 没有历史记录，认为未掌握

# 取最近 window_size 条记录
recent = history[-self.window_size:]

# 时间衰减加权：越近的答题权重越大
# 按照在窗口中的位置分配线性增长的权重
total_weight = 0
weighted_correct = 0

for i, record in enumerate(recent):
weight = (i + 1) / len(recent) # 位置越靠后（越近），权重越大
weighted_correct += record['correct'] * weight
total_weight += weight

return weighted_correct / total_weight if total_weight > 0 else 0.0

def get_weak_points(self, student_id, threshold=0.6):
"""
获取薄弱知识点列表

参数:
threshold: 掌握度阈值，低于此值视为薄弱点
"""
weak_points = []
for (sid, kp), history in self.student_kp_history.items():
if sid == student_id:
mastery = self.get_current_mastery(sid, kp)
if mastery < threshold:
weak_points.append({
'knowledge_point': kp,
'mastery': round(mastery, 3),
'sample_count': len(history)
})

# 按掌握度升序排列（最薄弱的排前面）
weak_points.sort(key=lambda x: x['mastery'])
return weak_points

# —- 使用示例 —-
bkt = SimplifiedBKT(window_size=5)

# 模拟学生 S001 在一元二次方程上的连续答题记录
bkt.add_observation('S001', '一元二次方程', 0, '2026-07-01')
bkt.add_observation('S001', '一元二次方程', 0, '2026-07-02')
bkt.add_observation('S001', '一元二次方程', 1, '2026-07-03')
bkt.add_observation('S001', '一元二次方程', 1, '2026-07-04')
bkt.add_observation('S001', '一元二次方程', 1, '2026-07-05')

mastery = bkt.get_current_mastery('S001', '一元二次方程')
print(f"S001 在一元二次方程的当前掌握度: {mastery:.2%}")

# 获取所有薄弱点
weak = bkt.get_weak_points('S001', threshold=0.6)
print(f"\\n=== S001 的薄弱知识点 ===")
for item in weak:
print(f" {item['knowledge_point']}: 掌握度 {item['mastery']}")

这个简化的 BKT 模型虽然不如完整版的 HMM（隐马尔可夫模型）那么严谨，但在实际生产中已经够用了。它的优势在于：

• 可解释性强：每个学生的掌握概率可以直接用"最近 N 次答题的加权正确率"来解释。
• 计算高效：不需要 EM 算法迭代训练，实时更新即可。
• 部署简单：完全可以放在业务代码里跑，不用上模型服务。这对教育机构的技术团队特别友好，不需要额外搭建模型服务基础设施。

四、个性化学习路径推荐

有了每个学生的知识状态，下一步就是做推荐。这里我们用知识图谱 + 协同过滤的混合方案。

# —- 知识图谱 + 协同过滤推荐 —-

# 1. 构建知识点依赖关系图（知识图谱）
# 用字典表示各知识点的前置依赖关系
knowledge_graph = {
'一元二次方程': [],
'二次函数': ['一元二次方程'], # 学二次函数需要先会一元二次方程
'一次函数': [],
'函数综合': ['一次函数', '二次函数'],
'不等式': ['一次函数']
}

def recommend_learning_path(student_id, weak_points, knowledge_graph,
mastery_df, top_n=3):
"""
基于知识图谱和薄弱点的学习路径推荐

参数:
student_id: 学生ID
weak_points: 薄弱知识点列表（来自BKT模型）
knowledge_graph: 知识点依赖关系图
mastery_df: 所有学生的掌握度 DataFrame
top_n: 推荐知识点数量

返回:
推荐的学习路径列表
"""

# —- 第1步：确定候选推荐知识点 —-
candidates = set()

for wp in weak_points:
kp = wp['knowledge_point']
# 如果薄弱点的前置知识还没掌握，优先推荐前置知识
prerequisites = knowledge_graph.get(kp, [])
for pre in prerequisites:
# 检查前置知识是否也已经薄弱
pre_weak = any(w['knowledge_point'] == pre for w in weak_points)
if pre_weak:
candidates.add(pre) # 前置知识也薄弱，优先学

# 当前薄弱点也加入候选
candidates.add(kp)

# —- 第2步：过滤已掌握的知识点 —-
student_mastery = mastery_df[mastery_df['student_id'] == student_id]
mastered_kps = set(student_mastery[
student_mastery['mastery_score'] >= 0.8
]['knowledge_point'])

# 如果所有前置知识都掌握了，则跳过
final_candidates = []
for kp in candidates:
if kp in mastered_kps:
continue
# 检查前置依赖是否满足
prerequisites = knowledge_graph.get(kp, [])
all_prereq_mastered = all(p in mastered_kps for p in prerequisites)
if all_prereq_mastered:
# 取该知识点对应的掌握得分（越低越优先推荐）
kp_score = student_mastery[
student_mastery['knowledge_point'] == kp
]['mastery_score'].values
priority = kp_score[0] if len(kp_score) > 0 else 0.0
final_candidates.append((kp, priority))

# —- 第3步：按优先级排序（掌握度越低越靠前） —-
final_candidates.sort(key=lambda x: x[1])

# 返回 top_n 推荐
return final_candidates[:top_n]

# —- 测试推荐 —-
recommendations = recommend_learning_path(
student_id='S001',
weak_points=weak,
knowledge_graph=knowledge_graph,
mastery_df=mastery_result,
top_n=3
)

print("\\n=== 为 S001 推荐的学习路径 ===")
for i, (kp, score) in enumerate(recommendations, 1):
print(f" 第{i}步: 学习【{kp}】（掌握度: {score:.2%}）")

推荐系统的核心逻辑就三步：找薄弱 → 查依赖 → 排优先。实际生产环境中，还会加入学习资源的匹配（视频、习题、文章）和难度分层（入门/进阶/挑战），让推荐更加精准。

五、总结

教育行业的数据分析是一个"慢工出细活"的领域。和电商推荐不同，教育推荐不能只盯着点击率，更关键的是学习效果——学生真学会了没。

这个项目做完，几点心得和大家分享：

数据驱动教育这件事，路还长，但方向绝对是对的。