保姆级教程！基于大语言模型的学生用户画像与个性化推荐全流程（DeepSeek + 学生画像 + 协同过滤 + 向量检索）
·
标签：#助睿数智 #商业数据分析 #大语言模型 #学生画像 #个性化推荐 #DeepSeek #Python #数据可视化
摘要
本实验基于老师提供的《基于大语言模型的学生用户画像构建和个性化推荐方法》教程，使用 Python 在 PyCharm 中完整复现学生画像与推荐系统分析流程。实验围绕数智教育数据集，完成了数据清洗、学生多维特征构建、传统规则画像、DeepSeek 大模型画像、协同过滤推荐、向量检索推荐、冷启动实验、相似学生匹配以及多类可视化图表生成。本文适合小白跟着一步一步完成实验，并预留了关键截图位置，方便整理实验报告或博客发布。
第一部分：实验背景
1. 实验目的
本次实验的核心目标是：用真实校园数据构建学生用户画像，并基于画像生成个性化学习推荐。
通过本实验，我主要掌握了以下能力：
- 使用 Python 读取并清洗多张教育业务数据表。
- 将学生基本信息、成绩、考勤、消费等数据融合成学生多维特征表。
- 使用传统规则方法构建学生画像标签。
- 使用 DeepSeek 大语言模型生成可解释的自然语言学生画像。
- 使用协同过滤算法构建传统推荐系统基线。
- 使用画像文本与向量检索实现个性化推荐、冷启动推荐和相似学生匹配。
- 生成实验图表，用可视化方式验证分析结果。
2. 实验环境
本实验基于 Python 完成，同时参考助睿数智一站式数据科学实验平台的课程体系。
| 项目 | 说明 |
|---|---|
| 平台定位 | 覆盖数据接入、ETL处理、机器学习建模到可视化分析的全链路Agentic零代码数据智能 |
| 本次实现方式 | 基于 Python + PyCharm 完成实验复现 |
| 使用模型 | DeepSeek Chat |
| 主要库 | pandas、numpy、matplotlib、seaborn、scikit-learn、chromadb、openai |
3. 数据说明
本实验使用“数智教育数据集”，共包含 7 张表：
| 数据表 | 文件名 | 主要内容 |
|---|---|---|
| 教师信息 | 1_teacher.csv | 教师、班级、学科、学期等信息 |
| 学生信息 | 2_student_info.csv | 学生性别、班级、生源地、住校等信息 |
| 考勤记录 | 3_kaoqin.csv | 学生考勤时间、考勤类型、班级等信息 |
| 考勤类型 | 4_kaoqintype.csv | 考勤设备和考勤任务类型 |
| 成绩数据 | 5_chengji.csv | 学生考试成绩、学科、考试类型等信息 |
| 考试类型 | 6_exam_type.csv | 期中、期末等考试类型 |
| 消费数据 | 7_consumption.csv | 学生校园消费时间、金额、性别等信息 |
4. 整体实验流程
本实验整体流程如下：
原始数据
↓
数据加载与清洗
↓
学生多维特征表
↓
传统规则画像
↓
DeepSeek 大模型画像
↓
协同过滤推荐基线
↓
画像文本向量检索推荐
↓
冷启动推荐 + 相似学生匹配 + 可视化图表第二部分：实验步骤
步骤 0：项目文件结构准备
主要脚本如下：
| 脚本 | 作用 |
|---|---|
| 公共工具.py | 统一路径、读取 CSV、保存结果、配置字体、DeepSeek 调用 |
| 第一步_数据准备.py | 读取并清洗 7 张原始表 |
| 第二步_构建学生多维特征.py | 构建学生多维特征表 |
| 第三步_传统规则画像.py | 构建传统标签画像并分析缺陷 |
| 第四步_大模型学生画像.py | 调用 DeepSeek 生成学生画像 |
| 第五步_协同过滤推荐.py | 构建 User-CF 协同过滤推荐 |
| 第六步_向量检索个性化推荐.py | 构建资源库、向量检索推荐、冷启动和相似学生 |
| 第七步_综合可视化与博客素材.py | 生成博客和报告所需图表 |
| 运行全部流程.py | 一键运行所有步骤 |
配置要点：
.\.venv\Scripts\python.exe .\运行全部流程.py如果要调用真实 DeepSeek，需要在 PyCharm 运行配置或 PowerShell 中设置环境变量：
$env:DEEPSEEK_API_KEY="你的DeepSeek密钥"
$env:DEEPSEEK_API_NAME="aaaaa"注意：不要把 API Key 直接写进代码，避免泄露。
步骤 1：数据加载与预处理
操作说明：
from __future__ import annotations
import pandas as pd
from 公共工具 import (
TABLE_FILES,
classify_meal,
load_raw_tables,
numeric,
print_title,
save_table,
save_text,
)
def clean_teacher(df: pd.DataFrame) -> pd.DataFrame:
return df.drop_duplicates().copy()
def clean_student(df: pd.DataFrame) -> pd.DataFrame:
cleaned = df.drop_duplicates().copy()
cleaned["bf_StudentID"] = numeric(cleaned["bf_StudentID"]).astype("Int64")
for col in ["bf_zhusu", "bf_leaveSchool"]:
if col in cleaned.columns:
cleaned[col] = numeric(cleaned[col]).fillna(0).astype(int)
return cleaned
def clean_attendance(df: pd.DataFrame) -> pd.DataFrame:
cleaned = df.drop_duplicates().copy()
cleaned["bf_studentID"] = numeric(cleaned["bf_studentID"]).astype("Int64")
cleaned["DataDateTime"] = pd.to_datetime(cleaned["DataDateTime"], errors="coerce")
cleaned["考勤日期"] = cleaned["DataDateTime"].dt.date
cleaned["考勤小时"] = cleaned["DataDateTime"].dt.hour
return cleaned
def clean_score(df: pd.DataFrame) -> pd.DataFrame:
cleaned = df.drop_duplicates().copy()
cleaned["mes_StudentID"] = numeric(cleaned["mes_StudentID"]).astype("Int64")
cleaned["mes_Score"] = numeric(cleaned["mes_Score"])
cleaned["成绩异常标记"] = cleaned["mes_Score"].lt(0)
cleaned.loc[cleaned["mes_Score"] < 0, "mes_Score"] = pd.NA
for col in ["mes_Z_Score", "mes_T_Score"]:
if col in cleaned.columns:
cleaned[col] = numeric(cleaned[col])
cleaned["exam_sdate"] = pd.to_datetime(cleaned["exam_sdate"], errors="coerce")
return cleaned
def clean_consumption(df: pd.DataFrame) -> pd.DataFrame:
cleaned = df.drop_duplicates().copy()
cleaned["bf_StudentID"] = numeric(cleaned["bf_StudentID"]).astype("Int64")
cleaned["MonDeal"] = numeric(cleaned["MonDeal"])
cleaned["DealTime"] = pd.to_datetime(cleaned["DealTime"], errors="coerce")
cleaned["消费日期"] = cleaned["DealTime"].dt.date
cleaned["消费小时"] = cleaned["DealTime"].dt.hour
cleaned["消费时段"] = cleaned["消费小时"].apply(classify_meal)
cleaned["消费金额"] = (-cleaned["MonDeal"]).where(cleaned["MonDeal"] < 0, 0)
return cleaned
def main() -> None:
print_title("第一步：数据加载与预处理")
raw_tables = load_raw_tables()
cleaned_tables = {
"教师信息": clean_teacher(raw_tables["教师信息"]),
"学生信息": clean_student(raw_tables["学生信息"]),
"考勤记录": clean_attendance(raw_tables["考勤记录"]),
"考勤类型": raw_tables["考勤类型"].drop_duplicates().copy(),
"成绩数据": clean_score(raw_tables["成绩数据"]),
"考试类型": raw_tables["考试类型"].drop_duplicates().copy(),
"消费数据": clean_consumption(raw_tables["消费数据"]),
}
report_lines = ["# 数据清洗概览", ""]
for name, cleaned in cleaned_tables.items():
raw = raw_tables[name]
filename = TABLE_FILES[name].replace(".csv", "_clean.csv")
save_path = save_table(cleaned, filename)
report_lines.append(
f"- {name}: 原始 {raw.shape[0]} 行 x {raw.shape[1]} 列，"
f"清洗后 {cleaned.shape[0]} 行 x {cleaned.shape[1]} 列，"
f"缺失值 {int(cleaned.isna().sum().sum())} 个，保存至 {save_path.name}"
)
print(report_lines[-1])
score = cleaned_tables["成绩数据"]
consume = cleaned_tables["消费数据"]
attend = cleaned_tables["考勤记录"]
report_lines.extend(
[
"",
"## 关键清洗规则",
f"- 成绩表中负分代表缺考/作弊/免考等异常，已转为空值；异常记录数：{int(score['成绩异常标记'].sum())}",
f"- 消费表中 MonDeal 为负数时代表支出，已转为正向字段“消费金额”；消费总额：{consume['消费金额'].sum():.2f}",
f"- 考勤时间已解析为日期和小时；有效时间记录数：{attend['DataDateTime'].notna().sum()}",
]
)
save_text("\n".join(report_lines), "第一步_数据清洗概览.md")
print("\n第一步完成：清洗后的表已保存到 output/中间数据")
if __name__ == "__main__":
main()
运行：
.\.venv\Scripts\python.exe .\第一步_数据准备.py该脚本完成以下任务：
- 自动读取 7 张 CSV 数据表。
- 自动兼容 UTF-8、GBK 等编码。
- 对成绩表中的负分进行异常标记，并转为空值。
- 将消费表中的负数消费金额转成正向“消费金额”。
- 将考勤时间和消费时间解析为日期、小时等字段。
- 将清洗后的数据保存到
output/中间数据。
关键结果：
教师信息：3088 行
学生信息：1765 行
考勤记录：23630 行
成绩数据：471686 行
消费数据：463520 行
配置要点：
4_kaoqintype.csv实际是制表符分隔，需要按\t读取。- 成绩中的负数不是正常分数，而是缺考、作弊、免考等异常情况。
- 消费表中
MonDeal为负数时代表支出，因此需要转为正数。
步骤 2：构建学生多维特征表
操作说明：
from __future__ import annotations
import pandas as pd
from 公共工具 import (
ARTS_SUBJECTS,
MAIN_SUBJECTS,
SCIENCE_SUBJECTS,
load_middle,
numeric,
print_title,
save_table,
save_text,
)
def build_score_features(df_score: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
score = df_score.copy()
score["mes_Score"] = numeric(score["mes_Score"])
student_score = (
score.groupby("mes_StudentID")
.agg(
avg_score=("mes_Score", "mean"),
std_score=("mes_Score", "std"),
avg_zscore=("mes_Z_Score", "mean"),
exam_count=("mes_Score", "count"),
max_score=("mes_Score", "max"),
min_score=("mes_Score", "min"),
)
.reset_index()
.rename(columns={"mes_StudentID": "bf_StudentID"})
)
subject_score = (
score[score["mes_sub_name"].isin(MAIN_SUBJECTS)]
.pivot_table(
index="mes_StudentID",
columns="mes_sub_name",
values="mes_Score",
aggfunc="mean",
)
.reset_index()
.rename(columns={"mes_StudentID": "bf_StudentID"})
)
subject_score.columns.name = None
return student_score, subject_score
def build_attendance_features(df_attend: pd.DataFrame) -> pd.DataFrame:
attend = df_attend.copy()
attend["bf_studentID"] = numeric(attend["bf_studentID"])
features = (
attend.groupby("bf_studentID")
.agg(
kaoqin_total=("kaoqing_id", "count"),
kaoqin_type_count=("controler_name", "nunique"),
first_kaoqin=("DataDateTime", "min"),
last_kaoqin=("DataDateTime", "max"),
)
.reset_index()
.rename(columns={"bf_studentID": "bf_StudentID"})
)
top_types = attend["controler_name"].value_counts().head(6).index.tolist()
for attend_type in top_types:
type_count = (
attend[attend["controler_name"] == attend_type]
.groupby("bf_studentID")
.size()
.rename(f"考勤_{attend_type}")
.reset_index()
.rename(columns={"bf_studentID": "bf_StudentID"})
)
features = features.merge(type_count, on="bf_StudentID", how="left")
return features
def build_consumption_features(df_consume: pd.DataFrame) -> pd.DataFrame:
consume = df_consume.copy()
consume["bf_StudentID"] = numeric(consume["bf_StudentID"])
consume["消费金额"] = numeric(consume["消费金额"])
features = (
consume.groupby("bf_StudentID")
.agg(
total_consume=("消费金额", "sum"),
consume_count=("消费金额", "count"),
consume_days=("消费日期", "nunique"),
max_single_consume=("消费金额", "max"),
)
.reset_index()
)
features["daily_avg"] = features["total_consume"] / features["consume_days"].replace(0, pd.NA)
features["per_txn_avg"] = features["total_consume"] / features["consume_count"].replace(0, pd.NA)
meal_pivot = (
consume.pivot_table(
index="bf_StudentID",
columns="消费时段",
values="消费金额",
aggfunc="count",
fill_value=0,
)
.reset_index()
)
meal_pivot.columns.name = None
meal_cols = [c for c in meal_pivot.columns if c != "bf_StudentID"]
total_count = meal_pivot[meal_cols].sum(axis=1).replace(0, pd.NA)
for col in meal_cols:
meal_pivot[f"{col}_占比"] = meal_pivot[col] / total_count
keep_cols = ["bf_StudentID"] + [c for c in meal_pivot.columns if c.endswith("_占比")]
features = features.merge(meal_pivot[keep_cols], on="bf_StudentID", how="left")
return features
def add_subject_summary(df_profile: pd.DataFrame) -> pd.DataFrame:
df = df_profile.copy()
for subject in MAIN_SUBJECTS:
if subject not in df.columns:
df[subject] = pd.NA
df["理科均分"] = df[SCIENCE_SUBJECTS].mean(axis=1)
df["文科均分"] = df[ARTS_SUBJECTS].mean(axis=1)
def list_weak_subjects(row) -> str:
weak = [s for s in MAIN_SUBJECTS if pd.notna(row[s]) and row[s] < 60]
return "、".join(weak)
def list_strong_subjects(row) -> str:
strong = [s for s in MAIN_SUBJECTS if pd.notna(row[s]) and row[s] >= 80]
return "、".join(strong)
df["薄弱学科"] = df.apply(list_weak_subjects, axis=1)
df["优势学科"] = df.apply(list_strong_subjects, axis=1)
return df
def main() -> None:
print_title("第二步：构建学生多维特征表")
df_student = load_middle("2_student_info_clean.csv")
df_score = load_middle("5_chengji_clean.csv")
df_attend = load_middle("3_kaoqin_clean.csv")
df_consume = load_middle("7_consumption_clean.csv")
student_score, subject_score = build_score_features(df_score)
attendance_features = build_attendance_features(df_attend)
consumption_features = build_consumption_features(df_consume)
profile = df_student[
[
"bf_StudentID",
"bf_Name",
"bf_sex",
"bf_nation",
"bf_BornDate",
"cla_Name",
"cla_id",
"cla_term",
"bf_NativePlace",
"Bf_ResidenceType",
"bf_zhusu",
]
].copy()
profile = profile.merge(student_score, on="bf_StudentID", how="left")
profile = profile.merge(subject_score, on="bf_StudentID", how="left")
profile = profile.merge(attendance_features, on="bf_StudentID", how="left")
profile = profile.merge(consumption_features, on="bf_StudentID", how="left")
profile = add_subject_summary(profile)
numeric_fill_zero = [
"kaoqin_total",
"kaoqin_type_count",
"total_consume",
"consume_count",
"consume_days",
"daily_avg",
"per_txn_avg",
"max_single_consume",
]
for col in numeric_fill_zero:
if col in profile.columns:
profile[col] = profile[col].fillna(0)
for col in [c for c in profile.columns if c.startswith("考勤_") or c.endswith("_占比")]:
profile[col] = profile[col].fillna(0)
save_path = save_table(profile, "学生多维特征表.csv")
report = f"""# 学生多维特征表构建结果
- 学生总数：{len(profile)}
- 特征字段数：{profile.shape[1]}
- 有成绩数据学生数：{int(profile['avg_score'].notna().sum())}
- 有考勤记录学生数：{int((profile['kaoqin_total'] > 0).sum())}
- 有消费记录学生数：{int((profile['consume_count'] > 0).sum())}
- 保存文件：{save_path}
"""
save_text(report, "第二步_学生多维特征表说明.md")
print(report)
if __name__ == "__main__":
main()
运行：
.\.venv\Scripts\python.exe .\第二步_构建学生多维特征.py该步骤将多张业务表融合成一张学生画像基础表。
构建的特征包括：
| 维度 | 特征示例 |
|---|---|
| 基本信息 | 性别、班级、民族、生源地、是否住校 |
| 学业表现 | 平均分、最高分、最低分、成绩标准差、考试次数 |
| 学科能力 | 语文、数学、英语、物理、化学、生物、政治、历史、地理均分 |
| 考勤行为 | 考勤异常次数、考勤类型数 |
| 消费行为 | 消费总额、消费次数、消费天数、日均消费、单笔均额 |
运行结果：
学生总数：1765
特征字段数：51
有成绩数据学生数：1571
有考勤记录学生数：1007
有消费记录学生数：1730
配套图表：
结果解读：
学业数据覆盖率约 89%，消费数据覆盖率约 98%，说明消费数据相对完整；考勤数据覆盖率较低，后续画像分析中不能只依赖考勤特征。
步骤 3：传统规则学生画像构建
操作说明：
from __future__ import annotations
import matplotlib.pyplot as plt
import pandas as pd
from 公共工具 import (
ARTS_SUBJECTS,
FONT_PROP,
MAIN_SUBJECTS,
SCIENCE_SUBJECTS,
load_middle,
print_title,
save_figure,
save_table,
save_text,
)
def academic_label(avg_score) -> str:
if pd.isna(avg_score):
return "无成绩"
if avg_score >= 90:
return "学霸"
if avg_score >= 80:
return "优良"
if avg_score >= 60:
return "中等"
return "学困"
def detailed_academic_label(avg_score) -> str:
if pd.isna(avg_score):
return "无成绩"
if avg_score >= 90:
return "学霸"
if avg_score >= 80:
return "优良"
if avg_score >= 70:
return "中等偏上"
if avg_score >= 60:
return "中等"
if avg_score >= 50:
return "中等偏下"
return "学困"
def subject_label(row: pd.Series) -> str:
science = row.get("理科均分")
arts = row.get("文科均分")
if pd.isna(science) and pd.isna(arts):
return "无学科数据"
if pd.notna(science) and pd.notna(arts):
diff = science - arts
if diff >= 8:
return "偏理科"
if diff <= -8:
return "偏文科"
weak = [s for s in MAIN_SUBJECTS if pd.notna(row.get(s)) and row.get(s) < 60]
strong = [s for s in MAIN_SUBJECTS if pd.notna(row.get(s)) and row.get(s) >= 80]
if len(weak) >= 4:
return "多科薄弱"
if len(strong) >= 4:
return "多科优势"
return "均衡型"
def attendance_label(total) -> str:
total = 0 if pd.isna(total) else total
if total == 0:
return "全勤标兵"
if total <= 5:
return "正常"
if total <= 20:
return "偶有异常"
return "考勤预警"
def consumption_label(daily_avg) -> str:
daily_avg = 0 if pd.isna(daily_avg) else daily_avg
if daily_avg == 0:
return "无消费记录"
if daily_avg < 15:
return "节俭"
if daily_avg <= 35:
return "适中"
return "高消费"
def build_rule_profile(df_profile: pd.DataFrame) -> pd.DataFrame:
df = df_profile.copy()
df["academic_label_v1"] = df["avg_score"].apply(academic_label)
df["academic_label_v2"] = df["avg_score"].apply(detailed_academic_label)
df["subject_label"] = df.apply(subject_label, axis=1)
df["attendance_label"] = df["kaoqin_total"].apply(attendance_label)
df["consumption_label"] = df["daily_avg"].apply(consumption_label)
df["traditional_profile"] = df.apply(
lambda row: (
f"学业:{row['academic_label_v1']} | "
f"学科:{row['subject_label']} | "
f"考勤:{row['attendance_label']} | "
f"消费:{row['consumption_label']}"
),
axis=1,
)
return df
def plot_label_distribution(df: pd.DataFrame) -> None:
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
plot_items = [
("academic_label_v1", "学业表现标签分布"),
("subject_label", "学科特征标签分布"),
("attendance_label", "考勤行为标签分布"),
("consumption_label", "消费行为标签分布"),
]
for ax, (col, title) in zip(axes.ravel(), plot_items):
counts = df[col].value_counts()
counts.plot.bar(ax=ax, color="#4C78A8", edgecolor="white")
ax.set_title(title, fontproperties=FONT_PROP)
ax.set_ylabel("人数", fontproperties=FONT_PROP)
ax.tick_params(axis="x", labelrotation=20)
for tick in ax.get_xticklabels() + ax.get_yticklabels():
tick.set_fontproperties(FONT_PROP)
save_figure("第三步_传统画像标签分布.png")
def analyze_rule_limitations(df: pd.DataFrame) -> str:
academic_top = df["academic_label_v1"].value_counts(normalize=True).head(1)
subject_top = df["subject_label"].value_counts(normalize=True).head(1)
lines = [
"# 传统规则画像缺陷分析",
"",
"## D1：维度孤立",
"传统规则把学业、学科、考勤、消费分别打标签，但不会自动理解“成绩波动大且考勤异常”这类交叉关系。",
"",
"## D2：语义缺失",
"标签只能给出“学困/均衡/适中”等短词，无法解释具体薄弱学科、行为模式和改进路径。",
"",
"## D3：输出不可读",
"传统画像更像数据编码，不适合直接给老师、学生或家长阅读。",
"",
"## 标签集中度",
f"- 占比最高的学业标签：{academic_top.index[0]}，占比 {academic_top.iloc[0]:.1%}",
f"- 占比最高的学科标签：{subject_top.index[0]}，占比 {subject_top.iloc[0]:.1%}",
"",
"标签高度集中时，说明规则画像区分度不足，难以支持精细化推荐。",
]
return "\n".join(lines)
def main() -> None:
print_title("第三步：传统规则画像构建")
df_profile = load_middle("学生多维特征表.csv")
df_rule = build_rule_profile(df_profile)
save_path = save_table(df_rule, "学生画像_传统规则.csv")
plot_label_distribution(df_rule)
report = analyze_rule_limitations(df_rule)
save_text(report, "第三步_传统规则画像缺陷分析.md")
print("传统规则画像示例：")
cols = ["bf_Name", "avg_score", "kaoqin_total", "daily_avg", "traditional_profile"]
print(df_rule[cols].head(10).to_string(index=False))
print(f"\n第三步完成：{save_path}")
if __name__ == "__main__":
main()
运行：
.\.venv\Scripts\python.exe .\第三步_传统规则画像.py传统画像通过阈值规则给学生打标签：
| 标签类型 | 示例 |
|---|---|
| 学业标签 | 学霸、优良、中等、学困 |
| 学科标签 | 偏理科、偏文科、多科薄弱、均衡型 |
| 考勤标签 | 全勤标兵、正常、偶有异常、考勤预警 |
| 消费标签 | 节俭、适中、高消费、无消费记录 |
传统画像示例：
学业:学困 | 学科:多科薄弱 | 考勤:正常 | 消费:适中
学业:中等 | 学科:偏理科 | 考勤:考勤预警 | 消费:适中
配套图表：
结果分析：
传统规则画像的优点是简单、透明、容易解释；缺点是标签比较粗糙。例如两个学生都被标记为“学困”，但一个可能是数学薄弱，另一个可能是考勤异常导致学习不稳定，传统标签很难表达这些细节。
步骤 4：接入 DeepSeek 构建大模型学生画像
操作说明：
from __future__ import annotations
import json
import os
import pandas as pd
from 公共工具 import (
ARTS_SUBJECTS,
MAIN_SUBJECTS,
SCIENCE_SUBJECTS,
llm_chat,
load_middle,
parse_json_from_text,
pause_for_api_limit,
print_title,
save_table,
save_text,
short_text,
)
SYSTEM_PROMPT = """你是一位资深教育数据分析师，擅长根据学生的多维度数据构建精准、可解释的学生画像。
请只输出严格 JSON，不要输出解释文字。JSON 结构如下：
{
"academic_level": "学业等级，从[学霸, 优良, 中等偏上, 中等, 中等偏下, 学困]中选择",
"subject_profile": "学科特征，说明优势和薄弱学科",
"learning_attitude": "学习态度评价，综合成绩稳定性和考勤情况",
"behavior_pattern": "行为模式，综合消费、住校、考勤等信息",
"strengths": ["优势1", "优势2"],
"weaknesses": ["待改进1", "待改进2"],
"suggestions": ["建议1", "建议2"],
"overall_description": "100-180字自然语言画像，可直接给老师阅读"
}
"""
def build_student_summary(row: pd.Series) -> str:
summary = []
summary.append("【基本信息】")
summary.append(
f"姓名: {short_text(row.get('bf_Name'))}, 性别: {short_text(row.get('bf_sex'))}, "
f"班级: {short_text(row.get('cla_Name'))}"
)
summary.append(
f"民族: {short_text(row.get('bf_nation'))}, 生源地: {short_text(row.get('bf_NativePlace'))}, "
f"家庭/居住类型: {short_text(row.get('Bf_ResidenceType'))}, "
f"是否住校: {'是' if row.get('bf_zhusu', 0) == 1 else '否'}"
)
summary.append("\n【学业表现】")
if pd.notna(row.get("avg_score")):
summary.append(
f"考试均分: {row['avg_score']:.1f}分, 成绩标准差: {row.get('std_score', 0):.1f}, "
f"考试次数: {int(row.get('exam_count', 0))}, 最高分: {row.get('max_score', 0):.1f}, "
f"最低分: {row.get('min_score', 0):.1f}"
)
subject_parts = []
for subject in MAIN_SUBJECTS:
value = row.get(subject)
if pd.notna(value):
subject_parts.append(f"{subject}:{value:.0f}")
summary.append("各科均分: " + "，".join(subject_parts))
summary.append(f"优势学科: {short_text(row.get('优势学科'), '无明显优势')}")
summary.append(f"薄弱学科: {short_text(row.get('薄弱学科'), '无明显薄弱')}")
else:
summary.append("暂无有效成绩数据")
summary.append("\n【考勤情况】")
summary.append(f"考勤异常总次数: {int(row.get('kaoqin_total', 0))}次")
summary.append("\n【消费情况】")
summary.append(
f"日均消费: {row.get('daily_avg', 0):.1f}元, 消费总额: {row.get('total_consume', 0):.1f}元, "
f"消费天数: {int(row.get('consume_days', 0))}天, 消费笔数: {int(row.get('consume_count', 0))}笔"
)
return "\n".join(summary)
def local_mock_profile(row: pd.Series) -> dict:
avg = row.get("avg_score")
if pd.isna(avg):
level = "无成绩"
elif avg >= 90:
level = "学霸"
elif avg >= 80:
level = "优良"
elif avg >= 70:
level = "中等偏上"
elif avg >= 60:
level = "中等"
elif avg >= 50:
level = "中等偏下"
else:
level = "学困"
science = row.get("理科均分")
arts = row.get("文科均分")
if pd.notna(science) and pd.notna(arts) and science - arts >= 8:
subject_profile = "理科表现相对突出，适合继续强化数理思维，同时保持文科基础。"
elif pd.notna(science) and pd.notna(arts) and arts - science >= 8:
subject_profile = "文科表现相对突出，理科基础需要重点补强。"
else:
subject_profile = "各科差异不算极端，需要结合薄弱学科做针对性提升。"
weak = [s for s in MAIN_SUBJECTS if pd.notna(row.get(s)) and row.get(s) < 60]
strong = [s for s in MAIN_SUBJECTS if pd.notna(row.get(s)) and row.get(s) >= 80]
kaoqin = int(row.get("kaoqin_total", 0))
daily = float(row.get("daily_avg", 0))
attitude = "学习状态较稳定"
if pd.notna(row.get("std_score")) and row.get("std_score") >= 25:
attitude = "成绩波动较大，需要提升学习稳定性"
if kaoqin > 20:
attitude += "；考勤异常较多，自律习惯需要重点关注"
elif kaoqin > 5:
attitude += "；偶有考勤异常，需要及时提醒"
behavior = "消费记录较少，生活行为信息有限"
if daily > 35:
behavior = "日均消费偏高，建议关注消费结构和预算管理"
elif daily > 0:
behavior = "消费水平较为正常，生活规律性整体可观察"
strengths = []
if strong:
strengths.append("优势学科：" + "、".join(strong[:3]))
if daily > 0 and daily <= 35:
strengths.append("消费水平相对合理")
if kaoqin <= 5:
strengths.append("考勤表现较好")
if not strengths:
strengths.append("具备进一步提升空间")
weaknesses = []
if weak:
weaknesses.append("薄弱学科：" + "、".join(weak[:4]))
if kaoqin > 5:
weaknesses.append("考勤自律需要加强")
if pd.notna(row.get("std_score")) and row.get("std_score") >= 25:
weaknesses.append("成绩波动较大")
if not weaknesses:
weaknesses.append("需继续保持并寻找新的提升点")
suggestions = [
"围绕薄弱学科建立错题本和周复盘机制",
"结合优势学科形成同伴互助或讲题输出",
]
if kaoqin > 5:
suggestions.append("建立到校、离校和晚自习考勤提醒机制")
if daily > 35:
suggestions.append("制定月度消费预算，区分必要消费和非必要消费")
name = str(row.get("bf_Name", "该学生"))[0]
overall = (
f"{name}同学当前学业等级为{level}。{subject_profile}"
f"{attitude}。{behavior}。建议后续以薄弱学科补强和学习节奏稳定为主线，"
f"通过阶段性目标、错题复盘和同伴互助逐步提升综合表现。"
)
return {
"academic_level": level,
"subject_profile": subject_profile,
"learning_attitude": attitude,
"behavior_pattern": behavior,
"strengths": strengths,
"weaknesses": weaknesses,
"suggestions": suggestions,
"overall_description": overall,
}
def choose_sample_students(df: pd.DataFrame) -> list[int]:
sample_size = int(os.getenv("LLM_SAMPLE_SIZE", "8"))
if os.getenv("LLM_RUN_ALL", "0") == "1":
return df.index.tolist()
sample_ids: list[int] = []
for label in ["学霸", "优良", "中等", "学困"]:
group = df[df["academic_label_v1"] == label]
if len(group) > 0:
sample_ids.extend(group.head(2).index.tolist())
warning_group = df[(df["kaoqin_total"] > 20) | (df["daily_avg"] > 35)]
sample_ids.extend(warning_group.head(2).index.tolist())
unique_ids = []
for idx in sample_ids:
if idx not in unique_ids:
unique_ids.append(idx)
if len(unique_ids) < sample_size:
unique_ids.extend([idx for idx in df.head(sample_size * 2).index if idx not in unique_ids])
return unique_ids[:sample_size]
def generate_profile(row: pd.Series) -> dict:
student_summary = build_student_summary(row)
user_prompt = f"请根据下面学生数据生成画像：\n\n{student_summary}"
text = llm_chat(SYSTEM_PROMPT, user_prompt)
if not text:
return local_mock_profile(row)
try:
return parse_json_from_text(text)
except Exception:
fallback = local_mock_profile(row)
fallback["overall_description"] += "（注：API返回格式未能解析，本条使用本地规则兜底生成。）"
return fallback
def main() -> None:
print_title("第四步：大模型学生画像构建")
df = load_middle("学生画像_传统规则.csv")
df["data_summary"] = df.apply(build_student_summary, axis=1)
for col in [
"llm_academic_level",
"llm_subject_profile",
"llm_learning_attitude",
"llm_behavior_pattern",
"llm_strengths",
"llm_weaknesses",
"llm_suggestions",
"llm_overall_description",
]:
df[col] = ""
sample_indices = choose_sample_students(df)
profile_records = []
print(f"本次生成 {len(sample_indices)} 名学生的 LLM/模拟画像")
for order, idx in enumerate(sample_indices, start=1):
row = df.loc[idx]
profile = generate_profile(row)
df.loc[idx, "llm_academic_level"] = profile.get("academic_level", "")
df.loc[idx, "llm_subject_profile"] = profile.get("subject_profile", "")
df.loc[idx, "llm_learning_attitude"] = profile.get("learning_attitude", "")
df.loc[idx, "llm_behavior_pattern"] = profile.get("behavior_pattern", "")
df.loc[idx, "llm_strengths"] = "；".join(profile.get("strengths", []))
df.loc[idx, "llm_weaknesses"] = "；".join(profile.get("weaknesses", []))
df.loc[idx, "llm_suggestions"] = "；".join(profile.get("suggestions", []))
df.loc[idx, "llm_overall_description"] = profile.get("overall_description", "")
profile_records.append(
{
"bf_StudentID": row["bf_StudentID"],
"bf_Name": row["bf_Name"],
"profile_json": json.dumps(profile, ensure_ascii=False),
}
)
print(f"[{order}/{len(sample_indices)}] {row['bf_Name']} -> {profile.get('academic_level', '')}")
pause_for_api_limit(0.2)
profile_sample = pd.DataFrame(profile_records)
save_table(df, "学生画像_含大模型画像.csv")
save_table(profile_sample, "大模型画像样本明细.csv")
preview_cols = [
"bf_Name",
"avg_score",
"kaoqin_total",
"daily_avg",
"traditional_profile",
"llm_academic_level",
"llm_overall_description",
]
preview = df.loc[sample_indices, preview_cols].to_string(index=False)
save_text("# 大模型画像样本预览\n\n```text\n" + preview + "\n```", "第四步_大模型画像样本预览.md")
print("\n第四步完成：学生画像_含大模型画像.csv 已生成")
if __name__ == "__main__":
main()
运行：
$env:DEEPSEEK_API_KEY="你的DeepSeek密钥"
$env:DEEPSEEK_API_NAME="aaaaa"
.\.venv\Scripts\python.exe .\第四步_大模型学生画像.py该步骤的核心是把学生数据整理成自然语言摘要，再交给 DeepSeek 生成 JSON 格式画像。
Prompt 要求 DeepSeek 输出以下字段：
{
"academic_level": "学业等级",
"subject_profile": "学科特征",
"learning_attitude": "学习态度",
"behavior_pattern": "行为模式",
"strengths": ["优势1", "优势2"],
"weaknesses": ["待改进1", "待改进2"],
"suggestions": ["建议1", "建议2"],
"overall_description": "自然语言画像"
}
配套图表：
结果分析：
与传统规则画像相比，DeepSeek 输出的画像更像一段“班主任能读懂的分析报告”，不仅能判断学生学业等级，还能结合成绩波动、考勤、消费等信息给出具体建议。这正是大语言模型相比传统标签规则的优势。
步骤 5：协同过滤推荐算法
操作说明：
from __future__ import annotations
import matplotlib.pyplot as plt
import pandas as pd
import seaborn as sns
from sklearn.metrics.pairwise import cosine_similarity
from 公共工具 import FONT_PROP, MAIN_SUBJECTS, load_middle, print_title, save_figure, save_table, save_text
def build_rating_matrix(df_score: pd.DataFrame) -> pd.DataFrame:
score = df_score[df_score["mes_sub_name"].isin(MAIN_SUBJECTS)].copy()
matrix = score.pivot_table(
index="mes_StudentID",
columns="mes_sub_name",
values="mes_Score",
aggfunc="mean",
)
subject_max = matrix.max().replace(0, pd.NA)
normalized = matrix / subject_max
return normalized.clip(0, 1)
def build_student_similarity(rating_matrix: pd.DataFrame) -> pd.DataFrame:
filled = rating_matrix.fillna(0)
sim = cosine_similarity(filled.values)
return pd.DataFrame(sim, index=rating_matrix.index, columns=rating_matrix.index)
def user_based_cf_recommend(
target_id,
rating_matrix: pd.DataFrame,
similarity_df: pd.DataFrame,
k: int = 8,
n_recommend: int = 3,
) -> list[dict]:
if target_id not in rating_matrix.index:
return []
target_scores = rating_matrix.loc[target_id]
sims = similarity_df.loc[target_id].drop(target_id).sort_values(ascending=False).head(k)
candidate_scores = {}
for subject in rating_matrix.columns:
current = target_scores[subject]
if pd.notna(current) and current >= 0.75:
continue
weighted_sum = 0.0
weight_total = 0.0
for other_id, sim_value in sims.items():
other_score = rating_matrix.loc[other_id, subject]
if pd.notna(other_score):
weighted_sum += sim_value * other_score
weight_total += abs(sim_value)
if weight_total > 0:
predicted = weighted_sum / weight_total
current_value = 0 if pd.isna(current) else current
candidate_scores[subject] = {
"current": current_value,
"predicted": predicted,
"improvement": predicted - current_value,
}
ordered = sorted(candidate_scores.items(), key=lambda x: x[1]["improvement"], reverse=True)
return [
{
"subject": subject,
"current": info["current"],
"predicted": info["predicted"],
"improvement": info["improvement"],
}
for subject, info in ordered[:n_recommend]
if info["improvement"] > 0
]
def item_based_subject_similarity(rating_matrix: pd.DataFrame) -> pd.DataFrame:
item_matrix = rating_matrix.T.fillna(0)
sim = cosine_similarity(item_matrix.values)
return pd.DataFrame(sim, index=rating_matrix.columns, columns=rating_matrix.columns)
def build_top_similar_students(similarity_df: pd.DataFrame, top_n: int = 5) -> pd.DataFrame:
records = []
for sid in similarity_df.index:
sims = similarity_df.loc[sid].drop(sid).sort_values(ascending=False).head(top_n)
for other_id, score in sims.items():
records.append(
{
"bf_StudentID": sid,
"similar_student_id": other_id,
"similarity": round(float(score), 4),
}
)
return pd.DataFrame(records)
def plot_cf_figures(rating_matrix: pd.DataFrame, similarity_df: pd.DataFrame) -> None:
fig, axes = plt.subplots(1, 2, figsize=(16, 6))
show_n = min(25, len(similarity_df))
sns.heatmap(
similarity_df.iloc[:show_n, :show_n],
cmap="YlOrRd",
ax=axes[0],
xticklabels=False,
yticklabels=False,
)
axes[0].set_title("学生相似度热力图", fontproperties=FONT_PROP)
mask = rating_matrix.iloc[: min(60, len(rating_matrix))].notna().astype(int)
sns.heatmap(mask, cmap=["#FEE0D2", "#DE2D26"], ax=axes[1], cbar=False, yticklabels=False)
axes[1].set_title("学生-学科评分矩阵非空情况", fontproperties=FONT_PROP)
for tick in axes[1].get_xticklabels():
tick.set_fontproperties(FONT_PROP)
tick.set_rotation(30)
save_figure("第五步_协同过滤矩阵可视化.png")
def main() -> None:
print_title("第五步：协同过滤推荐算法")
df_score = load_middle("5_chengji_clean.csv")
df_profile = load_middle("学生画像_含大模型画像.csv")
rating_matrix = build_rating_matrix(df_score)
similarity_df = build_student_similarity(rating_matrix)
item_sim = item_based_subject_similarity(rating_matrix)
top_similar = build_top_similar_students(similarity_df, top_n=5)
save_table(rating_matrix.reset_index().rename(columns={"mes_StudentID": "bf_StudentID"}), "协同过滤_学生学科评分矩阵.csv")
save_table(top_similar, "协同过滤_学生相似度Top5.csv")
save_table(item_sim.reset_index().rename(columns={"mes_sub_name": "subject"}), "协同过滤_学科相似度矩阵.csv")
plot_cf_figures(rating_matrix, similarity_df)
sample_ids = [sid for sid in df_profile["bf_StudentID"].head(30).tolist() if sid in rating_matrix.index][:8]
records = []
for sid in sample_ids:
row = df_profile[df_profile["bf_StudentID"] == sid].iloc[0]
recs = user_based_cf_recommend(sid, rating_matrix, similarity_df)
if not recs:
records.append(
{
"bf_StudentID": sid,
"bf_Name": row["bf_Name"],
"推荐学科": "无推荐",
"当前得分": "",
"预测得分": "",
"提升空间": "",
}
)
continue
for rec in recs:
records.append(
{
"bf_StudentID": sid,
"bf_Name": row["bf_Name"],
"推荐学科": rec["subject"],
"当前得分": round(rec["current"], 3),
"预测得分": round(rec["predicted"], 3),
"提升空间": round(rec["improvement"], 3),
}
)
rec_df = pd.DataFrame(records)
save_table(rec_df, "协同过滤_UserCF推荐结果.csv")
sparsity = rating_matrix.isna().sum().sum() / (rating_matrix.shape[0] * rating_matrix.shape[1])
report = f"""# 协同过滤推荐结果与缺陷
- 评分矩阵规模：{rating_matrix.shape[0]} 名学生 x {rating_matrix.shape[1]} 个学科
- 评分矩阵稀疏度：{sparsity:.1%}
- User-CF 推荐样本数：{len(sample_ids)}
## 传统协同过滤的主要问题
1. 冷启动：没有成绩的新学生无法进入相似度计算。
2. 数据稀疏：学生并非每个学科都有完整成绩，矩阵缺失会影响相似度。
3. 解释不足：只能输出推荐学科和预测分，不能解释为什么推荐、该怎么学。
4. 画像利用不足：无法自然融入考勤、消费、住校等多维画像信息。
"""
save_text(report, "第五步_协同过滤推荐分析.md")
print(report)
print(rec_df.head(12).to_string(index=False))
if __name__ == "__main__":
main()
运行：
.\.venv\Scripts\python.exe .\第五步_协同过滤推荐.py该步骤构建传统推荐系统基线：
- 构建学生-学科评分矩阵。
- 使用余弦相似度计算学生相似度。
- 基于相似学生进行 User-CF 推荐。
- 输出推荐学科、当前得分、预测得分和提升空间。
运行结果：
评分矩阵规模：3860 名学生 x 9 个学科
评分矩阵稀疏度：7.0%示例推荐：
陈某某 推荐：地理、化学、英语
曹某某 推荐：物理、英语、地理
配套图表：
结果分析：
协同过滤能根据相似学生推荐学科，但它有明显问题：
- 没有成绩的新生无法推荐，存在冷启动问题。
- 推荐结果只有学科和分数，不会解释为什么推荐。
- 无法自然利用考勤、消费、住校等画像信息。
步骤 6：画像向量检索与个性化推荐
操作说明：
from __future__ import annotations
import pandas as pd
from sklearn.feature_extraction.text import HashingVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from 公共工具 import (
MAIN_SUBJECTS,
llm_chat,
load_middle,
print_title,
save_table,
save_text,
short_text,
)
class HashingEmbeddingFunction:
def __init__(self, n_features: int = 1024):
self.vectorizer = HashingVectorizer(
analyzer="char_wb",
ngram_range=(2, 4),
n_features=n_features,
alternate_sign=False,
norm="l2",
)
def __call__(self, input):
if isinstance(input, str):
input = [input]
return self.vectorizer.transform(input).astype("float32").toarray().tolist()
def build_resource_library() -> pd.DataFrame:
records = [
("r01", "补弱策略", "数学", "数学基础薄弱的学生应先回到课本例题，按函数、几何、概率模块建立错题本，每周复盘高频错误。"),
("r02", "补弱策略", "物理", "物理薄弱学生建议先掌握受力分析和运动学基本模型，通过画图和公式来源理解题目。"),
("r03", "补弱策略", "化学", "化学基础薄弱可从元素周期表、化学方程式和实验现象入手，建立知识网络。"),
("r04", "补弱策略", "生物", "生物遗传题需要先掌握孟德尔遗传定律，再逐步练习基因频率和系谱分析。"),
("r05", "补弱策略", "语文", "语文薄弱学生应坚持阅读积累和作文素材整理，同时训练现代文阅读答题模板。"),
("r06", "补弱策略", "英语", "英语薄弱学生应每天背诵高频词汇，结合阅读理解和完形填空训练语感。"),
("r07", "补弱策略", "政治", "政治学习要建立知识框架，关注材料题关键词和规范化答题语言。"),
("r08", "补弱策略", "历史", "历史薄弱学生应按时间轴梳理事件，建立阶段特征和因果关系。"),
("r09", "补弱策略", "地理", "地理学习要结合地图、气候图和区域特征，训练图表信息提取能力。"),
("r10", "提优建议", "数学", "数学成绩优秀的学生可以挑战综合题和竞赛题，重点训练多方法解题和抽象思维。"),
("r11", "提优建议", "英语", "英语优势学生可增加原版阅读和续写训练，提高表达丰富度和语篇理解能力。"),
("r12", "提优建议", "理科", "理科优势学生适合开展跨学科综合题训练，形成模型迁移能力。"),
("r13", "提优建议", "文科", "文科优势学生可加强材料分析和观点表达，提升论证深度。"),
("r14", "习惯养成", "通用", "考勤异常学生需要建立固定作息、到校提醒和阶段性自我监督表。"),
("r15", "习惯养成", "通用", "成绩波动大的学生应制定周计划，记录每日完成情况并做周末复盘。"),
("r16", "习惯养成", "住校", "住校生建议合理安排晚自习时间，避免熬夜，保证第二天课堂效率。"),
("r17", "消费管理", "通用", "消费偏高的学生建议记录每日开支，制定月度预算，区分必要消费和非必要消费。"),
("r18", "消费管理", "通用", "消费过低或无消费记录的学生需要关注是否存在生活保障不足或校内消费数据缺失。"),
("r19", "同伴学习", "理科", "理科薄弱但文科较好的学生可与理科优势同学结对，互补讲题。"),
("r20", "同伴学习", "文科", "文科薄弱学生可与表达能力较强的同学组成阅读和背诵互助小组。"),
("r21", "心理支持", "通用", "长期低分或波动大的学生需要增强自我效能感，设置可达成的小目标。"),
("r22", "家校沟通", "通用", "考勤异常和成绩下滑同时出现时，建议班主任与家长共同跟踪作息和学习状态。"),
("r23", "学习计划", "通用", "中等学生建议采用基础题巩固、错题复盘、限时训练三段式提升路径。"),
("r24", "学习计划", "通用", "学困学生应先保证课堂听懂和作业完成，再逐步增加拓展训练。"),
("r25", "学习计划", "通用", "优良学生应关注薄弱学科短板，避免因偏科影响整体排名。"),
]
return pd.DataFrame(records, columns=["resource_id", "type", "subject", "text"])
def build_recommendation_query(row: pd.Series) -> str:
llm_desc = short_text(row.get("llm_overall_description"), "")
if llm_desc:
return llm_desc
subject_scores = []
for subject in MAIN_SUBJECTS:
value = row.get(subject)
if pd.notna(value):
subject_scores.append(f"{subject}{value:.0f}分")
return (
f"学生{short_text(row.get('bf_Name'))}，{short_text(row.get('bf_sex'))}，"
f"班级{short_text(row.get('cla_Name'))}。"
f"传统画像：{short_text(row.get('traditional_profile'))}。"
f"均分{row.get('avg_score', 0):.1f}，考勤异常{int(row.get('kaoqin_total', 0))}次，"
f"日均消费{row.get('daily_avg', 0):.1f}元。"
f"优势学科：{short_text(row.get('优势学科'), '无')}；薄弱学科：{short_text(row.get('薄弱学科'), '无')}。"
f"各科：{'，'.join(subject_scores)}"
)
def try_chroma_recommend(query: str, resources: pd.DataFrame, n_results: int = 5) -> list[dict] | None:
try:
import chromadb
embedding_fn = HashingEmbeddingFunction()
client = chromadb.EphemeralClient()
collection = client.create_collection(
name="education_resources",
embedding_function=embedding_fn,
metadata={"description": "教育资源向量库"},
)
collection.add(
ids=resources["resource_id"].tolist(),
documents=resources["text"].tolist(),
metadatas=resources[["type", "subject"]].to_dict("records"),
)
result = collection.query(query_texts=[query], n_results=n_results)
recs = []
for rid, doc, meta, distance in zip(
result["ids"][0],
result["documents"][0],
result["metadatas"][0],
result["distances"][0],
):
recs.append(
{
"resource_id": rid,
"type": meta["type"],
"subject": meta["subject"],
"text": doc,
"relevance": round(1 / (1 + float(distance)), 4),
"method": "Chroma",
}
)
return recs
except Exception:
return None
def sklearn_recommend(query: str, resources: pd.DataFrame, n_results: int = 5) -> list[dict]:
vectorizer = HashingVectorizer(
analyzer="char_wb",
ngram_range=(2, 4),
n_features=1024,
alternate_sign=False,
norm="l2",
)
docs = resources["text"].tolist()
matrix = vectorizer.transform(docs + [query])
scores = cosine_similarity(matrix[-1], matrix[:-1]).ravel()
top_idx = scores.argsort()[::-1][:n_results]
recs = []
for idx in top_idx:
row = resources.iloc[idx]
recs.append(
{
"resource_id": row["resource_id"],
"type": row["type"],
"subject": row["subject"],
"text": row["text"],
"relevance": round(float(scores[idx]), 4),
"method": "sklearn_hashing",
}
)
return recs
def recommend_resources(query: str, resources: pd.DataFrame, n_results: int = 5) -> list[dict]:
recs = try_chroma_recommend(query, resources, n_results=n_results)
if recs is not None:
return recs
return sklearn_recommend(query, resources, n_results=n_results)
def generate_recommendation_report(row: pd.Series, recs: list[dict]) -> str:
resource_text = "\n".join(
[f"{i+1}. [{r['type']}-{r['subject']}] {r['text']}" for i, r in enumerate(recs)]
)
system_prompt = "你是一位教育顾问，请根据学生画像和候选资源生成简洁、可执行的个性化推荐报告。"
user_prompt = f"""学生画像：
{build_recommendation_query(row)}
候选资源：
{resource_text}
请输出三部分：
1. 推荐理由
2. 学习计划
3. 给学生或家长的100字建议
"""
text = llm_chat(system_prompt, user_prompt)
if text:
return text
weak = short_text(row.get("薄弱学科"), "薄弱学科")
name = short_text(row.get("bf_Name"), "该学生")
top_resources = "、".join([f"{r['type']}-{r['subject']}" for r in recs[:3]])
return (
f"{name}当前画像显示需要重点关注{weak}，同时结合考勤、消费和学习稳定性进行综合提升。"
f"推荐优先使用 {top_resources} 等资源。建议每天安排固定时间进行薄弱学科基础巩固，"
f"每周整理错题并复盘一次；若存在考勤异常，应同步建立作息提醒和家校沟通机制。"
)
def find_similar_students(df: pd.DataFrame, target_index: int, n_results: int = 5) -> pd.DataFrame:
docs = df.apply(build_recommendation_query, axis=1).tolist()
vectorizer = HashingVectorizer(
analyzer="char_wb",
ngram_range=(2, 4),
n_features=1024,
alternate_sign=False,
norm="l2",
)
matrix = vectorizer.transform(docs)
scores = cosine_similarity(matrix[target_index], matrix).ravel()
order = scores.argsort()[::-1][:n_results]
rows = []
for idx in order:
row = df.iloc[idx]
rows.append(
{
"目标学生": df.iloc[target_index]["bf_Name"],
"相似学生": row["bf_Name"],
"bf_StudentID": row["bf_StudentID"],
"画像相似度": round(float(scores[idx]), 4),
"学业等级": row.get("llm_academic_level") or row.get("academic_label_v2"),
"考勤次数": row.get("kaoqin_total", 0),
}
)
return pd.DataFrame(rows)
def main() -> None:
print_title("第六步：画像向量检索与个性化推荐")
df_profile = load_middle("学生画像_含大模型画像.csv")
resources = build_resource_library()
save_table(resources, "教育资源库.csv")
candidates = df_profile[df_profile["llm_overall_description"].fillna("") != ""]
if candidates.empty:
candidates = df_profile.head(5)
candidates = candidates.head(5)
all_records = []
report_blocks = ["# 个性化推荐结果", ""]
for _, row in candidates.iterrows():
query = build_recommendation_query(row)
recs = recommend_resources(query, resources, n_results=5)
report = generate_recommendation_report(row, recs)
report_blocks.append(f"## 学生：{row['bf_Name']}（ID: {row['bf_StudentID']}）")
report_blocks.append("")
report_blocks.append("### 检索到的资源")
for rec in recs:
all_records.append(
{
"bf_StudentID": row["bf_StudentID"],
"bf_Name": row["bf_Name"],
**rec,
}
)
report_blocks.append(
f"- {rec['resource_id']} [{rec['type']}-{rec['subject']}] "
f"相关度 {rec['relevance']}: {rec['text']}"
)
report_blocks.append("")
report_blocks.append("### 推荐报告")
report_blocks.append(report)
report_blocks.append("")
rec_df = pd.DataFrame(all_records)
save_table(rec_df, "画像向量检索推荐结果.csv")
cold_query = "高一新生，男生，来自宁波，住校生，暂无成绩数据，需要适应高中学习节奏"
cold_recs = recommend_resources(cold_query, resources, n_results=5)
cold_df = pd.DataFrame(cold_recs)
cold_df.insert(0, "查询场景", "冷启动新生")
save_table(cold_df, "冷启动推荐结果.csv")
target_index = candidates.index[0]
similar_df = find_similar_students(df_profile, target_index=target_index, n_results=6)
save_table(similar_df, "相似学生匹配结果.csv")
report_blocks.extend(
[
"## 冷启动实验",
"",
f"查询：{cold_query}",
"",
]
)
for rec in cold_recs:
report_blocks.append(
f"- {rec['resource_id']} [{rec['type']}-{rec['subject']}] "
f"相关度 {rec['relevance']}: {rec['text']}"
)
report_blocks.extend(["", "## 相似学生匹配", "", "```text", similar_df.to_string(index=False), "```"])
save_text("\n".join(report_blocks), "第六步_画像向量检索个性化推荐报告.md")
print("推荐结果示例：")
print(rec_df.head(10).to_string(index=False))
print("\n第六步完成：画像向量检索推荐结果、冷启动结果、相似学生结果已保存")
if __name__ == "__main__":
main()
运行：
$env:DEEPSEEK_API_KEY="你的DeepSeek密钥"
.\.venv\Scripts\python.exe .\第六步_向量检索个性化推荐.py该步骤完成老师教程中最核心的“画像 + 推荐”部分。
主要流程：
- 构建教育资源库。
- 将学生画像转成推荐查询文本。
- 使用向量相似度检索 Top-K 学习资源。
- 调用 DeepSeek 生成个性化推荐报告。
- 模拟冷启动新生场景。
- 基于画像文本匹配相似学生。
教育资源库示例：
| 资源类型 | 资源内容 |
|---|---|
| 补弱策略 | 数学、物理、英语等学科补弱 |
| 提优建议 | 优势学科拔高训练 |
| 习惯养成 | 考勤异常、成绩波动等行为改进 |
| 消费管理 | 消费偏高或消费异常提醒 |
| 同伴学习 | 推荐互助学习搭子 |
配套图表：
结果分析：
画像向量检索推荐相比传统协同过滤更灵活。即使一个新生没有成绩数据，只要有“高一新生、住校、需要适应高中学习节奏”等文本描述，也能检索出习惯养成、学习计划等相关资源。这说明“画像文本 + 向量检索 + LLM 解释”能够有效缓解传统推荐系统的冷启动问题。
步骤 7：综合可视化生成
操作说明：
from __future__ import annotations
import re
from collections import Counter
import matplotlib.pyplot as plt
import pandas as pd
import seaborn as sns
from 公共工具 import FIGURE_DIR, FONT_PROP, load_middle, print_title, save_figure, save_text
def set_font(ax):
if FONT_PROP:
ax.title.set_fontproperties(FONT_PROP)
ax.xaxis.label.set_fontproperties(FONT_PROP)
ax.yaxis.label.set_fontproperties(FONT_PROP)
for tick in ax.get_xticklabels() + ax.get_yticklabels():
tick.set_fontproperties(FONT_PROP)
def plot_data_coverage(df: pd.DataFrame) -> None:
coverage = pd.Series(
{
"学业数据": (df["avg_score"].notna()).mean(),
"考勤数据": (df["kaoqin_total"] > 0).mean(),
"消费数据": (df["consume_count"] > 0).mean(),
"大模型画像": (df["llm_overall_description"].fillna("") != "").mean(),
}
).sort_values()
fig, ax = plt.subplots(figsize=(9, 5))
bars = ax.barh(coverage.index, coverage.values, color=["#8ECAE6", "#219EBC", "#FFB703", "#FB8500"])
ax.set_xlim(0, 1)
ax.set_xlabel("覆盖率")
ax.set_title("学生多维数据覆盖率")
for bar in bars:
width = bar.get_width()
ax.text(width + 0.01, bar.get_y() + bar.get_height() / 2, f"{width:.1%}", va="center")
set_font(ax)
save_figure("第七步_学生多维数据覆盖率.png")
def plot_score_distribution(df: pd.DataFrame) -> None:
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
sns.histplot(df["avg_score"].dropna(), bins=30, kde=True, color="#4C78A8", ax=axes[0])
axes[0].set_title("学生平均成绩分布")
axes[0].set_xlabel("平均成绩")
axes[0].set_ylabel("人数")
subject_cols = [c for c in ["语文", "数学", "英语", "物理", "化学", "生物", "政治", "历史", "地理"] if c in df.columns]
subject_mean = df[subject_cols].mean().sort_values(ascending=False)
subject_mean.plot.bar(ax=axes[1], color="#59A14F")
axes[1].set_title("各学科平均成绩对比")
axes[1].set_xlabel("学科")
axes[1].set_ylabel("平均分")
axes[1].tick_params(axis="x", labelrotation=30)
for ax in axes:
set_font(ax)
save_figure("第七步_成绩分布与学科均分.png")
def plot_profile_compare(df: pd.DataFrame) -> None:
sample = df[df["llm_academic_level"].fillna("") != ""].copy()
if sample.empty:
return
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
df["academic_label_v2"].value_counts().plot.bar(ax=axes[0], color="#4C78A8")
axes[0].set_title("传统规则学业等级分布")
axes[0].set_xlabel("传统规则标签")
axes[0].set_ylabel("人数")
axes[0].tick_params(axis="x", labelrotation=30)
sample["llm_academic_level"].value_counts().plot.bar(ax=axes[1], color="#E15759")
axes[1].set_title("DeepSeek样本画像学业等级分布")
axes[1].set_xlabel("大模型画像标签")
axes[1].set_ylabel("人数")
axes[1].tick_params(axis="x", labelrotation=30)
for ax in axes:
set_font(ax)
save_figure("第七步_传统画像与大模型画像对比.png")
def plot_radar(df: pd.DataFrame) -> None:
sample = df[df["llm_overall_description"].fillna("") != ""].head(4).copy()
if sample.empty:
sample = df.head(4).copy()
metrics = ["学业表现", "成绩稳定", "考勤表现", "消费规律", "理科能力", "文科能力"]
def normalize(series, reverse=False):
series = pd.to_numeric(series, errors="coerce").fillna(0)
mn, mx = series.min(), series.max()
if mx == mn:
value = pd.Series([0.5] * len(series), index=series.index)
else:
value = (series - mn) / (mx - mn)
return 1 - value if reverse else value
all_df = df.copy()
all_scores = pd.DataFrame(index=all_df.index)
all_scores["学业表现"] = normalize(all_df["avg_score"])
all_scores["成绩稳定"] = normalize(all_df["std_score"], reverse=True)
all_scores["考勤表现"] = normalize(all_df["kaoqin_total"], reverse=True)
all_scores["消费规律"] = 1 - (normalize((all_df["daily_avg"] - 25).abs()))
all_scores["理科能力"] = normalize(all_df["理科均分"])
all_scores["文科能力"] = normalize(all_df["文科均分"])
all_scores = all_scores.clip(0, 1)
angles = list(range(len(metrics)))
angles += angles[:1]
fig = plt.figure(figsize=(10, 8))
ax = plt.subplot(111, polar=True)
for _, row in sample.iterrows():
values = all_scores.loc[row.name, metrics].tolist()
values += values[:1]
ax.plot(angles, values, linewidth=2, label=f"{row['bf_Name']}-{row.get('llm_academic_level') or row.get('academic_label_v2')}")
ax.fill(angles, values, alpha=0.08)
ax.set_xticks(range(len(metrics)))
ax.set_xticklabels(metrics)
ax.set_yticks([0.2, 0.4, 0.6, 0.8, 1.0])
ax.set_ylim(0, 1)
ax.set_title("样本学生画像雷达图")
ax.legend(loc="upper right", bbox_to_anchor=(1.25, 1.1), prop=FONT_PROP)
set_font(ax)
save_figure("第七步_样本学生画像雷达图.png")
def plot_word_frequency(df: pd.DataFrame) -> None:
text = " ".join(df["llm_overall_description"].fillna("").tolist())
if not text.strip():
return
words = re.findall(r"[\u4e00-\u9fa5]{2,}", text)
stop = {"学生", "建议", "需要", "当前", "可以", "通过", "进行", "整体", "学习", "成绩"}
counts = Counter([w for w in words if w not in stop])
top = pd.Series(dict(counts.most_common(20)))
if top.empty:
return
fig, ax = plt.subplots(figsize=(11, 6))
top.sort_values().plot.barh(ax=ax, color="#B07AA1")
ax.set_title("DeepSeek画像描述高频词")
ax.set_xlabel("出现次数")
set_font(ax)
save_figure("第七步_大模型画像高频词.png")
def plot_recommendation_results() -> None:
rec = load_middle("画像向量检索推荐结果.csv")
cold = load_middle("冷启动推荐结果.csv")
similar = load_middle("相似学生匹配结果.csv")
fig, axes = plt.subplots(1, 3, figsize=(18, 5))
first_student = rec["bf_Name"].iloc[0]
rec_first = rec[rec["bf_Name"] == first_student].head(5)
axes[0].barh(rec_first["resource_id"], rec_first["relevance"], color="#F28E2B")
axes[0].set_title(f"{first_student}推荐资源相关度")
axes[0].set_xlabel("相关度")
axes[1].barh(cold["resource_id"], cold["relevance"], color="#76B7B2")
axes[1].set_title("冷启动新生推荐相关度")
axes[1].set_xlabel("相关度")
similar_show = similar.head(6)
axes[2].barh(similar_show["相似学生"], similar_show["画像相似度"], color="#EDC948")
axes[2].set_title("相似学生匹配结果")
axes[2].set_xlabel("画像相似度")
for ax in axes:
set_font(ax)
save_figure("第七步_推荐结果综合可视化.png")
def main() -> None:
print_title("第七步：综合可视化与博客素材生成")
df = load_middle("学生画像_含大模型画像.csv")
plot_data_coverage(df)
plot_score_distribution(df)
plot_profile_compare(df)
plot_radar(df)
plot_word_frequency(df)
plot_recommendation_results()
figure_list = sorted([p.name for p in FIGURE_DIR.glob("*.png")])
lines = ["# 博客可用图表清单", ""]
for name in figure_list:
lines.append(f"- {name}")
save_text("\n".join(lines), "第七步_博客图表清单.md")
print("已生成博客图表：")
for name in figure_list:
print(f"- {name}")
if __name__ == "__main__":
main()
运行：
.\.venv\Scripts\python.exe .\第七步_综合可视化.py该步骤统一生成博客和实验报告中需要展示的图表。
生成图表包括：
- 学生多维数据覆盖率。
- 成绩分布与学科均分。
- 传统画像标签分布。
- 传统画像与大模型画像对比。
- 样本学生画像雷达图。
- 大模型画像高频词。
- 协同过滤矩阵可视化。
- 推荐结果综合可视化。
第三部分：实验结果
1. 输出文件结果
实验运行后，结果主要保存在 output 文件夹中。
核心输出包括：
| 文件 | 说明 |
|---|---|
| 学生多维特征表.csv | 学生基础画像特征表 |
| 学生画像_传统规则.csv | 传统规则画像结果 |
| 学生画像_含大模型画像.csv | DeepSeek 画像结果 |
| 协同过滤_UserCF推荐结果.csv | 协同过滤推荐结果 |
| 画像向量检索推荐结果.csv | 向量检索推荐资源 |
| 冷启动推荐结果.csv | 新生冷启动推荐结果 |
| 相似学生匹配结果.csv | 基于画像文本的相似学生匹配 |
| 第六步_画像向量检索个性化推荐报告.md | DeepSeek 个性化推荐报告 |
【截图位置 24】插入 output/中间数据 文件夹截图。
【截图位置 25】插入 output/报告 文件夹截图。
2. 关键结果验证
学生多维特征表结果：
学生总数：1765
特征字段数：51
有成绩数据学生数：1571
有考勤记录学生数：1007
有消费记录学生数：1730大模型画像结果：
本次生成 8 名学生的 DeepSeek 画像
输出字段包括：学业等级、学科特征、学习态度、行为模式、优势、短板、建议、综合描述协同过滤结果：
评分矩阵规模：3860 x 9
可输出推荐学科、当前得分、预测得分和提升空间向量推荐结果：
可输出 Top-K 推荐资源
可生成 DeepSeek 个性化推荐报告
可完成冷启动新生推荐
可匹配画像相似学生3. 实验结果分析
从实验结果看，传统规则画像可以快速完成标签化，但表达能力有限；协同过滤可以作为推荐基线，但冷启动和解释能力不足。DeepSeek 大模型画像可以将多维数据转化为自然语言画像，再结合向量检索生成可解释的推荐报告，更适合教育场景下给老师、学生和家长阅读。
第四部分：问题与解决
问题 1：Windows 控制台中文或特殊符号乱码
问题现象：
运行脚本时，控制台打印中文、对勾符号或 emoji 时出现乱码，甚至报错：
UnicodeEncodeError: 'gbk' codec can't encode character问题原因：
Windows PowerShell 默认编码可能是 GBK，而 Python 输出中包含 UTF-8 字符。
解决方法：
在代码中加入：
import sys
sys.stdout.reconfigure(encoding="utf-8")运行时也可以设置：
$env:PYTHONIOENCODING="utf-8"问题 2：考勤类型文件读取后只有一列
问题现象：
4_kaoqintype.csv 读取后只有 1 列，字段全部挤在一起。
问题原因：
该文件虽然是 .csv 后缀，但实际分隔符是制表符 \t。
解决方法：
针对该文件单独设置分隔符：
sep = "\t" if filename == "4_kaoqintype.csv" else ","
pd.read_csv(path, sep=sep)问题 3：成绩中存在负数，影响均分计算
问题现象：
成绩最低分出现 -1、-2、-3 等异常值。
问题原因：
负分不是正常成绩，而是缺考、作弊、免考等业务编码。
解决方法：
将负分标记为异常，并转为空值：
df_chengji["成绩异常标记"] = df_chengji["mes_Score"] < 0
df_chengji.loc[df_chengji["mes_Score"] < 0, "mes_Score"] = pd.NA问题 4：DeepSeek 没有接入时画像无法生成
问题现象：
未配置 API Key 时，无法调用大模型。
问题原因：
代码从环境变量 DEEPSEEK_API_KEY 读取密钥，如果没有配置，就不能调用 DeepSeek。
解决方法：
在 PowerShell 或 PyCharm 运行配置中加入环境变量：
$env:DEEPSEEK_API_KEY="你的DeepSeek密钥"同时，为保证实验流程不断掉，代码中加入了本地模拟画像兜底逻辑。没有 API Key 时也能跑通流程；配置 API Key 后则使用真实 DeepSeek。
问题 5：完整学生相似度矩阵文件太大
问题现象：
最初保存完整学生相似度矩阵时，文件超过 200MB。
问题原因：
学生数量较多，完整相似度矩阵是“学生数 x 学生数”，文件体积会迅速变大。
解决方法：
只保存每个学生 Top5 相似学生：
协同过滤_学生相似度Top5.csv这样既保留了推荐分析需要的信息，又避免输出目录过大。
第五部分：实验总结
1. 实验收获
通过本次实验，我完成了从原始校园数据到学生画像推荐系统的完整流程。
我学会了：
- 如何清洗多表教育数据。
- 如何融合成绩、考勤、消费和基础信息构建学生特征。
- 如何用规则方法构建传统学生画像。
- 如何设计 Prompt 调用 DeepSeek 生成自然语言画像。
- 如何构建协同过滤推荐系统基线。
- 如何用画像文本进行向量检索推荐。
- 如何做冷启动实验和相似学生匹配。
- 如何用图表展示实验结果。
2. 方法对比
| 方法 | 优点 | 缺点 |
|---|---|---|
| 传统规则画像 | 简单、透明、容易解释 | 标签粗糙，语义不足 |
| 协同过滤推荐 | 有成熟算法基础，可做推荐基线 | 冷启动弱，解释能力差 |
| DeepSeek 画像 | 自然语言表达强，可解释性好 | 依赖 API，需控制成本和隐私 |
| 画像向量检索推荐 | 能结合画像语义，适合冷启动 | 资源库质量会影响推荐效果 |
3. 对平台的整体评价
一站式数据科学实验平台覆盖数据接入、ETL处理、机器学习建模到可视化分析的全链路 Agentic 零代码数据智能。虽然本次实验主要在 Python 和 PyCharm 中完成，但整体实验思路与平台课程目标一致：从数据接入、数据处理、建模分析到结果展示形成完整闭环。
对于商业数据分析初学者来说，这类平台和实验能帮助我们快速理解数据分析项目的真实流程。先通过平台理解流程，再用 Python 复现代码，可以同时提升业务理解能力和编程实现能力。
4. 最终结论
本实验成功完成了“基于大语言模型的学生用户画像构建和个性化推荐”全流程。最终结果证明：
- 多维校园数据可以有效构建学生画像。
- 传统规则画像适合做基础标签，但个性化不足。
- DeepSeek 能生成更细腻、可读、可解释的学生画像。
- 画像文本结合向量检索，可以为学生推荐更有针对性的学习资源。
- 该方法能缓解传统协同过滤的冷启动问题，更适合教育场景中的个性化服务。
更多推荐
8
0- 0
已为社区贡献1条内容
所有评论(0)