MachineLearning2025/CourseDesign
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add Streamlit application for student grade prediction and AI counseling.

Browse Source
This commit is contained in:
sit002 2026-01-01 11:19:17 +08:00
commit 4a92e62683
18 changed files with 1400 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.env.example Normal file
View File

@ -0,0 +1,2 @@
# DeepSeek API Key
DEEPSEEK_API_KEY=your-key-here

8
.gitignore vendored Normal file
View File

@ -0,0 +1,8 @@
__pycache__/
*.py[cod]
*$py.class
.venv/
.env
.DS_Store
models/*.pkl
.pytest_cache/

453
README.md Normal file
View File

@ -0,0 +1,453 @@
# 机器学习 × LLM × Agent：课程设计
> **小组作业** | 2–3 人/组 | 构建一个「可落地的智能预测与行动建议系统」
用传统机器学习完成可量化的预测任务，再用 LLM + Agent 把预测结果变成可执行的决策/建议，并保证输出结构化、可追溯、可复现。
---
## 📑 目录
- [快速开始](#-快速开始)
- [技术栈要求](#0-技术栈要求强制)
- [选题指南](#1-选题三档难度任选其一)
- [Level 1：入门](#level-1入门表格预测--行动建议闭环)
- [Level 2：进阶](#level-2进阶文本任务--处置建议回复生成)
- [Level 3：高阶](#level-3高阶不平衡多表时序--多步决策-agent)
- [自选题目标准](#2-自选题目标准)
- [代码示例](#3-deepseek--pydantic-ai最小可运行示例)
- [代码规范](#4-代码规范)
- [项目结构](#5-建议项目结构)
- [交付物与评分](#6-交付物与评分)
- [参考资料](#7-参考资料)
---
## 🚀 快速开始
```bash
# 1. 克隆/Fork 本模板仓库
git clone <your-repo-url>
cd ml_course_design_template
# 2. 创建虚拟环境并安装依赖
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install -r requirements.txt
# 3. 配置 DeepSeek API Key（不要提交到仓库！）
# 复制示例配置
cp .env.example .env
# 编辑 .env 文件，填入你的 API Key
# DEEPSEEK_API_KEY="your-key-here"
# 4. 运行示例
# 方式 A：运行 Streamlit 可视化 Demo（推荐）
python -m streamlit run src/streamlit_app.py
# 方式 B：运行命令行 Agent Demo
python src/agent_app.py
```
---
## 0. 技术栈要求（强制）
| 组件 | 要求 |
|------|------|
| **人数** | 2–3 人/组 |
| **Agent 框架** | `pydantic-ai` |
| **LLM 提供方** | `DeepSeek`（OpenAI 兼容 API） |
### 必须包含的三块能力
| 能力 | 说明 |
|------|------|
| **传统机器学习** | 可复现训练流程、离线评估指标、模型保存与加载 |
| **LLM** | 用于解释、归因、生成建议/回复、信息整合（不能凭空杜撰） |
| **Agent** | 用工具调用把系统串起来（至少 2 个 tool，其中 1 个必须是 ML 预测/评估相关工具） |
---
## 1. 选题：三档难度（任选其一）
你们可以先选难度档位，再从数据集列表里选一个；也可以自选（见「自选题目标准」）。
> ⚠️ **注意**：Level 1/2/3 **都可以拿满分**；高难度通常更容易体现"深度"，但不会因为选 Level 1 就被封顶。
---
### Level 1｜入门：表格预测 + 行动建议闭环
> 📌 **建议新手选择**
**目标**：做一个结构化数据的分类/回归模型，并让 Agent 基于模型输出给出可执行建议（如挽留、风控、营销、分诊）。
#### 交付的系统能力
- 训练并保存一个基线模型（如 `LogReg` / `LightGBM` / `RandomForest`）
- 提供 `predict_proba` / `predict` 工具给 Agent 调用
- Agent 输出**结构化决策**（Pydantic schema），并能说明建议与关键特征/规则的关联
#### 推荐数据集（任选其一）
| 数据集 | 链接 |
|--------|------|
| Telco Customer Churn | [Kaggle](https://www.kaggle.com/datasets/blastchar/telco-customer-churn) |
| German Credit Risk | [Kaggle](https://www.kaggle.com/datasets/uciml/german-credit) |
| Bank Marketing | [Kaggle](https://www.kaggle.com/datasets/janiobachmann/bank-marketing-dataset) |
| Heart Failure Prediction | [Kaggle](https://www.kaggle.com/datasets/fedesoriano/heart-failure-prediction) |
#### 最低基准
| 任务类型 | 指标要求 |
|----------|----------|
| 二分类 | `F1 ≥ 0.70` 或 `ROC-AUC ≥ 0.75` |
| 回归 | `MAE/RMSE` 相对朴素基线（均值/中位数）**显著提升**，需在报告中说明 |
#### 参考练习清单
**传统 ML（必须）**：
- 做一条可复现的数据流水线：缺失值处理、类别编码、数值缩放、训练/验证切分（并写清"为什么这样切分"）
- 至少 2 个模型对比：一个可解释基线（如 Logistic Regression），一个更强模型（如 LightGBM/RandomForest）
- 做 1 次误差分析：Top 误判样本/分桶（例如按年龄段/合同类型）看哪里最容易错
**Agent（必须）**：
- 定义结构化输出（Pydantic）：`risk_score + decision + actions + rationale`
- 至少 2 个工具（tool），其中 1 个必须调用你们的 ML（例如 `predict_risk(features)`）
- 推荐再加 1 个解释类工具：`explain_top_features(features) -> list[str]`（可用系数/重要性/规则，不要求复杂 SHAP）
**系统闭环（建议做到）**：
- 阈值/策略选择：把"预测概率"变成"要不要干预/怎么干预"（例如：高风险→人工复核；中风险→短信挽留；低风险→不操作）
- 做一个简单"成本-收益"版本：每个动作给一个成本/收益假设，让 Agent 给出最划算的动作组合（在报告里写清假设）
---
### Level 2｜进阶：文本任务 + 处置建议/回复生成
> 📌 **NLP 向**
**目标**：做文本分类/情感/主题/工单分流等任务，并让 Agent 生成"可执行处置方案"（如回复、升级、分派、风险提示），且输出必须结构化并可审计。
#### 交付的系统能力
- 传统 ML NLP 基线（如 `TF-IDF + Linear Model`）或轻量深度模型（可选）
- Agent 支持「分类 → 解释 → 生成建议/回复模板」的流程，并能引用证据（例如关键词、相似样本）
#### 推荐数据集（任选其一）
| 数据集 | 链接 | 说明 |
|--------|------|------|
| Twitter US Airline Sentiment | [Kaggle](https://www.kaggle.com/datasets/crowdflower/twitter-airline-sentiment) | 航空公司情感分析 |
| IMDB 50K Movie Reviews | [Kaggle](https://www.kaggle.com/datasets/lakshmi25npathi/imdb-dataset-of-50k-movie-reviews) | 电影评论情感 |
| SMS Spam Collection | [Kaggle](https://www.kaggle.com/datasets/uciml/sms-spam-collection-dataset) | 垃圾短信分类 |
| Consumer Complaints | [Kaggle](https://www.kaggle.com/datasets/selener/consumer-complaint-database) | 投诉分流（很适配 Agent） |
#### 最低基准
| 任务类型 | 指标要求 |
|----------|----------|
| 分类 | `Accuracy ≥ 0.85` 或 `Macro-F1 ≥ 0.80` |
#### 参考练习清单
**传统 ML NLP（必须）**：
- 基线：`TF-IDF + LogisticRegression/LinearSVC`，再加入至少一个更强的模型（如 `LightGBM`, `RandomForest`, 或轻量 BERT 模型）；并报告 `macro-F1`、混淆矩阵、典型错例
- 文本清洗要「克制」：说明你做了哪些预处理、为什么（不要把标签信息泄露进特征）
**Agent（必须）**：
- 工单/评论 → `classify_text(text)`（ML 工具）→ `draft_response(label, evidence)`（LLM）→ 输出结构化处置方案
- 建议加入 1 个「证据」工具：`extract_evidence(text) -> list[str]`（关键词/关键句/相似样本 id）
**更贴近真实（可选加分）**：
- **相似案例检索**：用 TF-IDF/Embedding 做 `retrieve_similar(text) -> top_k`，Agent 引用相似案例作为「可追溯证据」（禁止编造历史工单）
- **合规与安全**：对输出做规则检查（例如不得输出隐私信息/不得承诺无法兑现的政策），失败时让 Agent 重新生成
---
### Level 3｜高阶：不平衡/多表/时序 + 多步决策 Agent
> 📌 **真实世界约束**
**目标**：选择一个更接近真实业务约束的任务（极度不平衡、多表关联、时序预测等），实现更强的评估与更可靠的 Agent 决策链路。
#### 交付的系统能力
- 针对数据特性选择合适指标与训练策略（不平衡采样、阈值策略、时间切分、泄露防护等）
- Agent 能进行多步决策（例如：先评估风险 → 再选择干预策略 → 再生成操作清单/工单内容）
#### 推荐数据集（任选其一）
| 数据集 | 链接 | 特点 |
|--------|------|------|
| Credit Card Fraud Detection | [Kaggle](https://www.kaggle.com/datasets/mlg-ulb/creditcardfraud) | 极度不平衡 |
| IEEE-CIS Fraud Detection | [Kaggle](https://www.kaggle.com/c/ieee-fraud-detection) | 多表/特征工程复杂 |
| M5 Forecasting - Accuracy | [Kaggle](https://www.kaggle.com/competitions/m5-forecasting-accuracy) | 时序预测 + 补货决策 |
| Instacart Market Basket | [Kaggle](https://www.kaggle.com/c/instacart-market-basket-analysis) | 多表 + 召回/推荐 |
#### 最低基准
| 任务类型 | 指标要求 |
|----------|----------|
| 不平衡分类 | `PR-AUC` / `Recall@Precision` 等合理指标，并与 naive 基线对比 |
| 时序 | 必须使用时间切分评估（rolling/holdout），并与 naive 基线对比 |
#### 参考练习清单
**不平衡分类路线（如欺诈）**：
- 重点不在 Accuracy：用 `PR-AUC` / `Recall@Precision` / `Cost` 等指标，给出阈值选择依据
- 做 1 个「代价敏感」版本：例如漏报成本 > 误报成本，并让 Agent 基于代价输出策略
**多表路线（如电商/多表欺诈）**：
- 明确主键/外键与 join 规则，写出「数据泄露风险点清单」
- 做一个可复现的特征构建模块（统计聚合、时间窗特征等）
**时序路线（如 M5）**：
- 强制时间切分；至少对比 1 个朴素基线（naive/seasonal naive）
- 让 Agent 输出「补货/促销/库存」建议，并说明依据（趋势、季节性、置信区间）
**Agent（必须）**：
- 至少 3 步决策：例如 `评估风险/预测 → 解释与约束检查 → 生成行动计划`
- 在报告里画一张「决策链路图」
---
## 2. 自选题目标准
> 💡 **鼓励自选题目**，但必须满足以下硬标准，并先提交 1 页 proposal（Markdown 即可）
| 要求 | 说明 |
|------|------|
| **数据真实可获取** | 公开、可重复下载（Kaggle/UCI/OpenML/政府开放数据等），提供链接与数据说明 |
| **可量化预测任务** | 有明确标签/目标变量与评价指标；不得只做「聊天/生成」 |
| **业务闭环** | 能落到「下一步做什么」的决策/行动（由 Agent 负责） |
| **Agent 工具调用** | 至少 2 个 tools，其中 1 个必须是 ML 预测/评估/解释工具 |
| **规模与复杂度** | 样本量建议 ≥ 5,000（或能说明复杂性来自多表/长文本/时序）；不得低于 Level 1 |
| **合规性** | 禁止爬取受限数据；禁止提交包含密钥/个人隐私数据的仓库内容 |
---
## 3. DeepSeek + pydantic-ai：最小可运行示例
下面示例展示如何用 `pydantic-ai` 让 Agent 输出结构化结果，并通过 tool 调用你们训练好的模型。
```python
import os
from typing import Any
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
class Decision(BaseModel):
"""Agent 输出的结构化决策"""
risk_score: float = Field(ge=0, le=1, description="预测风险/流失概率等，0~1")
decision: str = Field(description="建议采取的总体策略")
actions: list[str] = Field(description="可执行动作清单（3~8条）")
rationale: str = Field(description="基于哪些证据/特征做出建议（不要编造事实）")
# 使用 provider:model 简写（DeepSeek 是 OpenAI 兼容 API）
agent = Agent(
"deepseek:deepseek-chat", # 也可尝试 "deepseek:deepseek-reasoner"
output_type=Decision,
system_prompt=(
"你是业务决策助手。你必须先调用工具获取模型预测与可解释信息，"
"再给出结构化决策与可执行动作；不得编造数据集中不存在的事实。"
),
)
@agent.tool
def predict_risk(ctx: RunContext[Any], features: dict) -> float:
"""调用训练好的 ML 模型，返回 0~1 的风险分数。
Args:
ctx: 运行上下文
features: 特征字典，如 {"age": 35, "contract": "month-to-month", ...}
Returns:
风险概率值 (0~1)
"""
# TODO: 加载模型并预处理特征
# model = joblib.load("models/model.pkl")
# X = preprocess(features)
# proba = model.predict_proba(X)[0, 1]
# return float(proba)
raise NotImplementedError("请实现模型调用逻辑")
@agent.tool
def explain_top_features(ctx: RunContext[Any], features: dict) -> list[str]:
"""解释对预测结果影响最大的特征。
Args:
ctx: 运行上下文
features: 特征字典
Returns:
影响最大的特征列表，如 ["合同类型: 月付 (+0.25)", "任期: 2个月 (+0.15)"]
"""
# TODO: 实现特征重要性解释
raise NotImplementedError("请实现特征解释逻辑")
```
### API Key 配置
> ⚠️ **重要**：不要把 Key 写进代码、不要提交到仓库！
```bash
# macOS / Linux (zsh/bash)
export DEEPSEEK_API_KEY="your-key-here"
# Windows (PowerShell)
$env:DEEPSEEK_API_KEY="your-key-here"
```
建议在项目根目录创建 `.env.example` 文件（提交到仓库），内容如下：
```
DEEPSEEK_API_KEY=your-key-here
```
然后复制为 `.env` 并填入真实 Key（`.env` 已在 `.gitignore` 中排除）。
---
## 4. 代码规范
本作业**不使用 CI/CD 作为评分项**，但代码质量会被严格检查。
| 要求 | 说明 |
|------|------|
| **可运行** | 在「干净环境」按 README 步骤能跑通训练与 demo |
| **可复现** | 固定随机种子；训练/评估脚本可重复得到同级别指标；关键超参可配置 |
| **结构清晰** | 模块划分合理；避免超长脚本；核心逻辑放 `src/`；数据处理、训练、推理、Agent 分离 |
| **类型提示与文档** | 对外 API 必须写 type hints 与 docstring |
| **不泄露** | 避免数据泄露（特别是时序/多表任务）；报告中说明切分策略 |
| **安全** | 密钥用环境变量；仓库中提供 `.env.example` 但不得提交真实 `.env` |
---
## 5. 建议项目结构
```
ml_course_design_template/
├── README.md # 项目说明
├── requirements.txt # Python 依赖
├── .env.example # 环境变量模板（不含真实密钥）
├── .gitignore # Git 忽略规则
│
├── data/ # 数据目录
│ └── README.md # 数据来源说明（原始大数据不要提交）
│
├── models/ # 训练产物（模型文件）
│ └── .gitkeep
│
├── notebooks/ # 探索性分析（可选）
│ └── eda.ipynb
│
├── src/ # 核心代码
│ ├── __init__.py
│ ├── data.py # 数据读取/清洗/特征工程
│ ├── train.py # 训练与离线评估
│ ├── infer.py # 推理接口（给 Agent 的 tool 调用）
│ └── agent_app.py # pydantic-ai Agent 入口
│
└── tests/ # 测试（建议至少覆盖 3 个关键函数）
├── __init__.py
├── test_data.py
├── test_model.py
└── test_agent.py
```
---
## 6. 交付物与评分
### 必交材料
| 材料 | 说明 |
|------|------|
| **代码仓库链接** | 组长提交 |
| **项目报告** | 4–8 页 Markdown/PDF：问题定义、数据说明、特征/模型、评估、Agent 设计、局限与改进 |
| **演示** | 5–8 分钟 demo（Streamlit / Gradio / Next.js + FastAPI 等），展示端到端流程 |
### 评分标准（总分 100）
#### A. 问题与数据（15 分）
| 维度 | 分值 | 要求 |
|------|------|------|
| 任务定义清晰 | 5 | 标签/目标是什么、为什么重要、输入输出边界 |
| 数据说明完整 | 5 | 来源链接、字段含义、样本量、潜在偏差/缺失 |
| 切分与泄露防护 | 5 | 随机/分层/时间切分说明；明确避免目标泄露 |
#### B. 传统机器学习（35 分）
| 维度 | 分值 | 要求 |
|------|------|------|
| 基线与可复现训练 | 10 | 固定随机种子、训练脚本能跑通、基线合理 |
| 指标与对比 | 10 | 指标选择正确，并与至少 1 个强/弱基线对比 |
| 误差分析 | 10 | 展示错误样本/分桶/特征影响，给出改进方向 |
| 结果可信度 | 5 | 阈值/校准/稳定性（任选其一做到位即可） |
#### C. LLM + Agent（35 分）
| 维度 | 分值 | 要求 |
|------|------|------|
| 工具调用 | 10 | 至少 2 个 tools，能稳定调用 ML 工具（不是「假调用」） |
| 结构化输出 | 10 | Pydantic schema 清晰；字段有约束；失败能重试/兜底 |
| 建议可执行且有证据 | 10 | 能落地的动作清单，并能引用依据（禁止编造事实） |
| 边界与安全 | 5 | 能处理异常输入；对敏感输出做规则约束 |
#### D. 工程与规范（15 分）
| 维度 | 分值 | 要求 |
|------|------|------|
| 可运行与可复现 | 5 | README 步骤清楚；干净环境可复现；依赖明确 |
| 代码结构与风格 | 5 | 模块化、命名清晰、类型提示/Docstring 到位 |
| 演示质量 | 5 | demo 端到端连贯；能说明架构与关键取舍 |
### ❌ 常见扣分项
- 训练/推理无法在助教环境跑通；或需要手动改很多路径/参数
- 数据泄露（尤其是时序/多表）；或评估切分不合理但未说明
- Agent 输出「看似合理但无证据」的内容，或编造数据集不存在的事实
- **把密钥提交进仓库（严重扣分）**
### ✅ 常见加分项
> 不额外加分栏，但会让你们更容易拿到高分
- 做了可解释性/阈值策略/代价敏感分析，并与业务动作闭环
- 做了检索增强（RAG/相似案例）且引用可追溯证据
- 做了消融/对比实验，结论清晰且能指导下一步优化
---
## 7. 参考资料
| 资源 | 链接 |
|------|------|
| pydantic-ai 文档 | https://ai.pydantic.dev/ |
| DeepSeek API | https://api.deepseek.com （OpenAI 兼容） |
| DeepSeek 模型 | `deepseek-chat` / `deepseek-reasoner` |
---
## 📋 Checklist（提交前自检）
- [ ] 代码仓库可正常访问
- [ ] README 包含完整的运行步骤
- [ ] 在干净环境下可以复现训练和推理
- [ ] 没有提交 API Key 或敏感信息
- [ ] 没有提交大型数据文件
- [ ] Agent 至少有 2 个 tool（含 1 个 ML 工具）
- [ ] 输出使用 Pydantic 结构化
- [ ] 报告说明了数据切分策略
- [ ] Demo 可以正常运行
---
> 💬 **有问题？** 请在课程群/Issue 中提问，我们会尽快回复。

44
REPORT.md Normal file
View File

@ -0,0 +1,44 @@
# 项目报告：[项目名称]
> **小组成员**:
> - [姓名] (学号)
> - [姓名] (学号)
## 1. 问题定义与数据
### 1.1 任务描述
<!-- 描述预测任务（分类/回归/时序）和业务目标。 -->
### 1.2 数据来源与说明
<!-- 数据集链接。描述字段、样本量以及所做的任何预处理。 -->
### 1.3 数据切分与防泄露
<!-- 即使如何切分训练/验证/测试集？如何确保没有数据泄露（特别是对于时序或多表数据）？ -->
## 2. 机器学习流水线
### 2.1 基线模型
<!-- 你的基线模型是什么？（例如：使用默认参数的逻辑回归）。它的表现如何？ -->
### 2.2 进阶模型
<!-- 你改进后的模型是什么？（例如：LightGBM, Random Forest）。为什么选择它？ -->
### 2.3 评估与误差分析
<!-- 展示指标（F1, AUC 等）。分析模型在哪些样本上表现不佳及其原因。 -->
## 3. Agent 实现
### 3.1 工具定义
<!-- 列出你实现的工具。 -->
- `tool_name_1`: 描述...
- `tool_name_2`: 描述...
### 3.2 决策逻辑
<!-- Agent 如何使用工具？（例如：预测 -> 解释 -> 建议）。 -->
### 3.3 案例展示
<!-- 展示一个真实的交互示例（输入 -> 系统响应）。 -->
## 4. 反思
### 4.1 挑战与解决方案
<!-- 最困难的部分是什么？你是如何解决的？ -->
### 4.2 局限与未来改进
<!-- 如果有更多时间，还有哪些可以改进的地方？ -->

5
data/README.md Normal file
View File

@ -0,0 +1,5 @@
# Data Directory
Place your raw data files here.
For this example project, the data is generated synthetically in `src/data.py`, so no external files are needed.

0
models/.gitkeep Normal file
View File

8
requirements.txt Normal file
View File

@ -0,0 +1,8 @@
pydantic-ai
scikit-learn
pandas
numpy
joblib
python-dotenv
pytest
streamlit

133
src/agent_app.py Normal file
View File

@ -0,0 +1,133 @@
import os
import sys
import asyncio
from typing import Any, List
sys.path.append(os.getcwd())
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from dotenv import load_dotenv
from src.infer import predict_pass_prob, explain_prediction
load_dotenv()
# --- 1. 定义结构化输出 (Level 1 Requirement) ---
class ActionItem(BaseModel):
action: str = Field(description="具体的行动建议")
priority: str = Field(description="优先级 (高/中/低)")
class StudyGuidance(BaseModel):
pass_probability: float = Field(description="预测通过率 (0-1)")
risk_assessment: str = Field(description="风险评估 (自然语言描述)")
key_drivers: str = Field(description="导致该预测结果的主要因素 (来自模型解释)")
action_plan: List[ActionItem] = Field(description="3-5条建议清单")
# --- 2. 初始化 Agent ---
# 必须强调：不要编造事实，必须基于工具返回的数据。
agent = Agent(
"deepseek:deepseek-chat",
output_type=StudyGuidance,
system_prompt=(
"你是一个极其严谨的学业数据分析师。"
"你的任务是根据学生的具体情况预测其考试通过率，并给出建议。"
"【重要规则】"
"1. 必须先调用 `predict_student` 获取概率。"
"2. 必须调用 `explain_model` 获取模型认为最重要的特征，并在 `key_drivers` 中引用这些特征。"
"3. 你的建议必须针对那些最重要的特征（例如，如果模型说睡眠很重要，就给睡眠建议）。"
"4. 严禁凭空编造数值。"
),
)
# --- 2.1 定义 Counselor Agent ---
counselor_agent = Agent(
"deepseek:deepseek-chat",
system_prompt=(
"你是一位富有同理心且专业的大学心理咨询师。"
"你的目标是倾听学生的学业压力和生活烦恼，提供情感支持，并根据需要给出建议。"
"【交互风格】"
"1. 同理心：首先通过复述或确认学生的感受来表达理解（例如：“听起来你最近压力真的很大...”）。"
"2. 引导性：不要急于给出解决方案，先通过提问了解更多背景。"
"3. 数据驱动（可选）：如果学生询问具体通过率或客观分析，请调用 `predict_student_tool` 或 `explain_model_tool`。"
"4. 语气：温暖、支持、专业，但像朋友一样交谈。"
"【工具使用】"
"如果学生提供了具体的学习时长、睡眠等数据，或者明确询问预测结果，请使用工具。"
"不要在每一句话里都引用数据，只在通过率相关的话题中使用。"
),
)
# --- 3. 注册工具 (Level 1 Requirement: 至少2个工具) ---
@agent.tool
def predict_student(ctx: RunContext[Any],
study_hours: float,
sleep_hours: float,
attendance_rate: float,
stress_level: int,
study_type: str) -> float:
"""
根据学生行为预测通过率。
Args:
study_hours: 每周学习小时数 (0-20)
sleep_hours: 每天睡眠小时数 (0-12)
attendance_rate: 出勤率 (0.0-1.0)
stress_level: 压力等级 1(低) - 5(高)
study_type: 学习类型 ("Group", "Self", "Online")
"""
return predict_pass_prob(study_hours, sleep_hours, attendance_rate, stress_level, study_type)
@counselor_agent.tool
def predict_student_tool(ctx: RunContext[Any],
study_hours: float,
sleep_hours: float,
attendance_rate: float,
stress_level: int,
study_type: str) -> float:
"""
根据学生行为预测通过率。用于咨询过程中提供客观数据支持。
"""
return predict_pass_prob(study_hours, sleep_hours, attendance_rate, stress_level, study_type)
@agent.tool
def explain_model(ctx: RunContext[Any]) -> str:
"""
获取机器学习模型的全局特征重要性解释。
返回哪些特征对预测结果影响最大。
"""
return explain_prediction()
@counselor_agent.tool
def explain_model_tool(ctx: RunContext[Any]) -> str:
"""
获取机器学习模型的全局特征重要性解释。
"""
return explain_prediction()
async def main():
# 模拟真实的学生查询
query = (
"我最近压力很大 (等级4)，每天只睡 4 小时，不过我每周自学(Self) 12 小时，"
"出勤率大概 90%。请帮我分析一下我会挂科吗？基于模型告诉我怎么做最有效。"
)
print(f"用户: {query}\n")
print("Agent 正在思考并调用模型工具...\n")
try:
if not os.getenv("DEEPSEEK_API_KEY"):
print("❌ 错误: 未设置 DEEPSEEK_API_KEY，无法运行 Agent。")
print("请在 .env 文件中设置密钥，或 export DEEPSEEK_API_KEY='...'")
return
result = await agent.run(query)
print("--- 结构化分析报告 ---")
print(result.output.model_dump_json(indent=2))
except Exception as e:
print(f"❌ 运行失败: {e}")
if __name__ == "__main__":
asyncio.run(main())

84
src/data.py Normal file
View File

@ -0,0 +1,84 @@
import pandas as pd
import numpy as np
def generate_data(n_samples: int = 2000, random_seed: int = 42) -> pd.DataFrame:
"""
生成 Level 1 要求的复杂模拟数据。
包含：数值特征、类别特征、噪声、以及非线性关系。
特征:
- study_hours (float): 每周学习时长 (0-20)
- sleep_hours (float): 每晚睡眠时长 (3-10)
- attendance_rate (float): 出勤率 (0.0-1.0)
- study_type (category): 学习方式 ("Group", "Self", "Online")
- stress_level (int): 压力等级 (1-5)
目标:
- is_pass (int): 0 或 1
"""
np.random.seed(random_seed)
# 1. 生成基础特征
data = {
"study_hours": np.random.uniform(0, 15, n_samples),
"sleep_hours": np.random.normal(7, 1.5, n_samples).clip(3, 10),
"attendance_rate": np.random.beta(5, 2, n_samples), # 偏向于高出勤
"study_type": np.random.choice(["Group", "Self", "Online"], n_samples, p=[0.3, 0.5, 0.2]),
"stress_level": np.random.randint(1, 6, n_samples)
}
df = pd.DataFrame(data)
# 2. 模拟真实世界逻辑 (分数计算)
# 基础分
score = 40
# 线性影响
score += df["study_hours"] * 3.0
score += (df["attendance_rate"] - 0.5) * 30
# 非线性/交互影响
# 睡眠不足严重扣分
score -= np.maximum(0, 6 - df["sleep_hours"]) * 8
# 类别特征影响
# Group 对低学习时长有帮助，Self 对高时长有加成
mask_group = df["study_type"] == "Group"
mask_self = df["study_type"] == "Self"
score[mask_group] += 5
score[mask_self] += df.loc[mask_self, "study_hours"] * 0.5 # 额外加成
# 压力影响
score -= (df["stress_level"] - 1) * 2
# 3. 添加随机噪声
noise = np.random.normal(0, 8, n_samples)
final_score = score + noise
# 4. 生成标签 (及格线 60)
df["is_pass"] = (final_score >= 60).astype(int)
# 5. 人为制造缺失值 (模拟真实数据清洗需求)
# 随机丢弃 5% 的 attendance_rate
mask_na = np.random.random(n_samples) < 0.05
df.loc[mask_na, "attendance_rate"] = np.nan
return df
def preprocess_data(df: pd.DataFrame) -> pd.DataFrame:
"""
注意：在 scikit-learn Pipeline 模式下，
我们通常把'清洗'作为 Pipeline 的一部分。
这里只做最基础的清洗，比如删除完全错误的行（如果有）。
"""
# 演示：仅删除完全重复的行
return df.drop_duplicates()
if __name__ == "__main__":
df = generate_data()
print("数据样例:")
print(df.head())
print("\n缺失值统计:")
print(df.isnull().sum())
print(f"\n及格率: {df['is_pass'].mean():.2f}")

86
src/infer.py Normal file
View File

@ -0,0 +1,86 @@
import os
import sys
import joblib
import pandas as pd
import numpy as np
# 路径修复
sys.path.append(os.getcwd())
MODEL_PATH = os.path.join("models", "model.pkl")
_MODEL = None
def load_model():
global _MODEL
if _MODEL is None:
if not os.path.exists(MODEL_PATH):
raise FileNotFoundError(f"未找到模型文件 {MODEL_PATH}。请先运行 src/train.py。")
_MODEL = joblib.load(MODEL_PATH)
return _MODEL
def predict_pass_prob(study_hours: float, sleep_hours: float, attendance_rate: float,
stress_level: int, study_type: str) -> float:
"""
预测学生通过概率 (0.0 - 1.0)。
会自动处理特征预处理 (因为模型包含了 Pipeline)。
"""
model = load_model()
# 构建 DataFrame，这与训练时的输入格式一致
features = pd.DataFrame([{
"study_hours": study_hours,
"sleep_hours": sleep_hours,
"attendance_rate": attendance_rate,
"stress_level": stress_level,
"study_type": study_type
}])
# 预测概率
# [proba_false, proba_true]
try:
proba = model.predict_proba(features)[0, 1]
except Exception as e:
print(f"Prediction Error: {e}")
return 0.0
return float(proba)
def explain_prediction() -> str:
"""
解释模型的全局特征重要性。
从保存的 Random Forest Pipeline 中提取特征重要性。
"""
model = load_model()
try:
# 1. 提取预处理步骤中的特征名称
# Pipeline 结构: [('preprocessor', ColumnTransformer), ('classifier', RandomForest)]
preprocessor = model.named_steps["preprocessor"]
clf = model.named_steps["classifier"]
# 获取 OneHot 后的特征名
# numeric_features 在前，categorical 在后
num_feats = ["study_hours", "sleep_hours", "attendance_rate", "stress_level"]
# 获取 categorical feature names (从 OneHotEncoder 中)
# 注意: 如果 scikit-learn 版本较旧，可能需要用不同的方式获取
cat_encoder = preprocessor.named_transformers_["cat"].named_steps["onehot"]
cat_feats = cat_encoder.get_feature_names_out(["study_type"])
all_feats = np.concatenate([num_feats, cat_feats])
# 2. 获取重要性数值
importances = clf.feature_importances_
# 3. 排序并输出
indices = np.argsort(importances)[::-1]
lines = ["### 模型特征重要性排名 (Top 5):"]
for i in range(min(5, len(importances))):
idx = indices[i]
lines.append(f"{i+1}. {all_feats[idx]}: {importances[idx]:.4f}")
return "\n".join(lines)
except Exception as e:
return f"无法解释模型特征 (可能模型结构不同): {str(e)}"

185
src/streamlit_app.py Normal file
View File

@ -0,0 +1,185 @@
import streamlit as st
import asyncio
import os
import sys
# Ensure project root is in path
sys.path.append(os.getcwd())
from src.agent_app import agent, counselor_agent, StudyGuidance
from pydantic_ai.messages import (
ModelMessage, ModelRequest, ModelResponse, UserPromptPart, TextPart,
TextPartDelta, ToolCallPart, ToolReturnPart
)
from pydantic_ai import (
AgentStreamEvent, PartDeltaEvent, FunctionToolCallEvent, FunctionToolResultEvent
)
from dotenv import load_dotenv
# Load env variables
load_dotenv()
st.set_page_config(
page_title="学生成绩预测 AI 助手",
page_icon="🎓",
layout="wide"
)
# Sidebar Configuration
st.sidebar.header("🔧 配置")
api_key = st.sidebar.text_input("DeepSeek API Key", type="password", value=os.getenv("DEEPSEEK_API_KEY", ""))
if api_key:
os.environ["DEEPSEEK_API_KEY"] = api_key
st.sidebar.markdown("---")
# Mode Selection
mode = st.sidebar.radio("功能选择", ["📊 成绩预测", "💬 心理咨询"])
# --- Helper Functions ---
async def run_analysis(query):
try:
if not os.getenv("DEEPSEEK_API_KEY"):
st.error("请在侧边栏提供 DeepSeek API Key。")
return None
with st.spinner("🤖 Agent 正在思考... (调用 DeepSeek + 随机森林模型)"):
result = await agent.run(query)
return result.output
except Exception as e:
st.error(f"分析失败: {str(e)}")
return None
async def run_counselor_stream(query, history, placeholder):
"""
Manually stream the response to a placeholder, handling tool events for visibility.
"""
try:
if not os.getenv("DEEPSEEK_API_KEY"):
placeholder.error("❌ 错误: 请在侧边栏提供 DeepSeek API Key。")
return None
full_response = ""
# Status container for tool calls
status_placeholder = st.empty()
# Call Counselor Agent with streaming
# Call Counselor Agent with streaming using run_stream_events which is the modern way to get events
async for event in counselor_agent.run_stream_events(query, message_history=history):
# Handle Text Delta (Wrapped in PartDeltaEvent)
if isinstance(event, PartDeltaEvent) and isinstance(event.delta, TextPartDelta):
full_response += event.delta.content_delta
placeholder.markdown(full_response + "▌")
# Handle Tool Call Start
elif isinstance(event, FunctionToolCallEvent):
# event.part is ToolCallPart usually, or event.tool_call
# Check pydantic-ai docs structure: FunctionToolCallEvent has 'part' which is ToolCallPart
status_placeholder.info(f"🛠️ 咨询师正在使用工具: `{event.part.tool_name}` ...")
# Handle Tool Result
elif isinstance(event, FunctionToolResultEvent):
status_placeholder.empty()
placeholder.markdown(full_response)
status_placeholder.empty() # Ensure clear
return full_response
except Exception as e:
placeholder.error(f"❌ 咨询失败: {str(e)}")
return None
# --- Main Views ---
if mode == "📊 成绩预测":
st.title("🎓 学生成绩预测助手")
st.markdown("在下方输入学生详细信息，获取 AI 驱动的成绩分析。")
with st.form("student_data_form"):
col1, col2 = st.columns(2)
with col1:
study_hours = st.slider("每周学习时长 (小时)", 0.0, 20.0, 10.0, 0.5)
sleep_hours = st.slider("日均睡眠时长 (小时)", 0.0, 12.0, 7.0, 0.5)
with col2:
attendance_rate = st.slider("出勤率", 0.0, 1.0, 0.9, 0.05)
stress_level = st.select_slider("压力等级 (1=低, 5=高)", options=[1, 2, 3, 4, 5], value=3)
study_type = st.radio("主要学习方式", ["Self", "Group", "Online"], horizontal=True)
submitted = st.form_submit_button("🚀 分析通过率")
if submitted:
user_query = (
f"我是一名学生，情况如下："
f"每周学习时间: {study_hours} 小时；"
f"平均睡眠时间: {sleep_hours} 小时；"
f"出勤率: {attendance_rate:.2f}；"
f"压力等级: {stress_level} (1-5)；"
f"主要学习方式: {study_type}。"
f"请调用 `predict_student` 预测我的通过率，并调用 `explain_model` 分析关键因素，最后给出针对性的建议。"
)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
guidance = loop.run_until_complete(run_analysis(user_query))
if guidance:
st.divider()
st.subheader("📊 分析结果")
m1, m2, m3 = st.columns(3)
m1.metric("预测通过率", f"{guidance.pass_probability:.1%}")
m2.metric("风险评估", "高风险" if guidance.pass_probability < 0.6 else "低风险",
delta="-高风险" if guidance.pass_probability < 0.6 else "+安全")
st.info(f"**风险评估:** {guidance.risk_assessment}")
st.write(f"**关键因素:** {guidance.key_drivers}")
st.subheader("✅ 行动计划")
actions = [{"优先级": item.priority, "建议行动": item.action} for item in guidance.action_plan]
st.table(actions)
elif mode == "💬 心理咨询":
st.title("🧩 AI 心理咨询室")
st.markdown("这里是安全且私密的空间。有些压力如果你愿意说，我愿意听。")
# Initialize chat history
if "messages" not in st.session_state:
st.session_state.messages = []
# Display chat messages from history on app rerun
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
# React to user input
if prompt := st.chat_input("想聊聊什么？"):
# Display user message
with st.chat_message("user"):
st.markdown(prompt)
# Add user message to history
st.session_state.messages.append({"role": "user", "content": prompt})
# Prepare history for pydantic-ai
# Convert Streamlit history to pydantic-ai ModelMessages
# Note: We exclude the last message because `agent.run` takes the new prompt as argument
api_history = []
for msg in st.session_state.messages[:-1]:
if msg["role"] == "user":
api_history.append(ModelRequest(parts=[UserPromptPart(content=msg["content"])]))
elif msg["role"] == "assistant":
api_history.append(ModelResponse(parts=[TextPart(content=msg["content"])]))
# Generate response
with st.chat_message("assistant"):
placeholder = st.empty()
with st.spinner("咨询师正在倾听..."):
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# Run the manual streaming function
response_text = loop.run_until_complete(run_counselor_stream(prompt, api_history, placeholder))
if response_text:
st.session_state.messages.append({"role": "assistant", "content": response_text})

116
src/train.py Normal file
View File

@ -0,0 +1,116 @@
import sys
import os
# 修复模块路径问题，让你可以在根目录直接 python src/train.py
sys.path.append(os.getcwd())
import joblib
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, accuracy_score, f1_score
from src.data import generate_data, preprocess_data
MODELS_DIR = "models"
MODEL_PATH = os.path.join(MODELS_DIR, "model.pkl")
def get_pipeline(model_type="rf"):
"""
构建标准的 Sklearn 处理流水线。
1. 数值特征 -> 缺失填充 (均值) -> 标准化
2. 类别特征 -> 缺失填充 (众数) -> OneHot编码
3. 模型 -> LR 或 RF
"""
# 定义特征列
numeric_features = ["study_hours", "sleep_hours", "attendance_rate", "stress_level"]
categorical_features = ["study_type"]
# 数值处理管道
numeric_transformer = Pipeline(steps=[
("imputer", SimpleImputer(strategy="mean")),
("scaler", StandardScaler())
])
# 类别处理管道
categorical_transformer = Pipeline(steps=[
("imputer", SimpleImputer(strategy="most_frequent")),
("onehot", OneHotEncoder(handle_unknown="ignore"))
])
# 组合预处理
preprocessor = ColumnTransformer(
transformers=[
("num", numeric_transformer, numeric_features),
("cat", categorical_transformer, categorical_features)
]
)
# 选择模型
if model_type == "lr":
clf = LogisticRegression(random_state=42)
else:
clf = RandomForestClassifier(n_estimators=500, max_depth=5, random_state=42)
return Pipeline(steps=[
("preprocessor", preprocessor),
("classifier", clf)
])
def train():
print(">>> 1. 数据准备")
df = generate_data(n_samples=2000)
df = preprocess_data(df)
X = df.drop(columns=["is_pass"])
y = df["is_pass"]
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42
)
print(f"训练集大小: {X_train.shape}, 测试集大小: {X_test.shape}")
print("\n>>> 2. 模型训练与对比")
# 模型 A: 逻辑回归 (Baseline)
pipe_lr = get_pipeline("lr")
pipe_lr.fit(X_train, y_train)
y_pred_lr = pipe_lr.predict(X_test)
f1_lr = f1_score(y_test, y_pred_lr)
print(f"[Baseline - LogisticRegression] F1: {f1_lr:.4f}")
# 模型 B: 随机森林 (Target)
pipe_rf = get_pipeline("rf")
pipe_rf.fit(X_train, y_train)
y_pred_rf = pipe_rf.predict(X_test)
f1_rf = f1_score(y_test, y_pred_rf)
print(f"[Target - RandomForest] F1: {f1_rf:.4f}")
print("\n>>> 3. 如果 RF 更好，则进行详细评估")
best_model = pipe_rf
print(classification_report(y_test, y_pred_rf))
print("\n>>> 4. 误差分析 (Error Analysis)")
# 找出模型预测错误的样本
test_df = X_test.copy()
test_df["True Label"] = y_test
test_df["Pred Label"] = y_pred_rf
errors = test_df[test_df["True Label"] != test_df["Pred Label"]]
print(f"总计错误样本数: {len(errors)}")
if len(errors) > 0:
print("典型错误样本预览:")
print(errors.head(3))
print("\n>>> 5. 保存最佳模型")
os.makedirs(MODELS_DIR, exist_ok=True)
joblib.dump(best_model, MODEL_PATH)
print(f"模型 Pipeline 已完整保存至 {MODEL_PATH}")
if __name__ == "__main__":
train()

27
tests/test_agent.py Normal file
View File

@ -0,0 +1,27 @@
import os
# 设置虚拟 key 以避免测试收集期间 pydantic-ai 初始化错误
os.environ["DEEPSEEK_API_KEY"] = "dummy_key_for_testing"
import sys
sys.path.append(os.getcwd())
from src.agent_app import predict_student
# 注意: 我们直接测试工具函数，而不是完整的 agent 循环
# 因为 agent 需要 API key，而 CI/测试环境中可能未设置。
from unittest.mock import patch
def test_tool_wrapper():
# 测试 Agent wrapper 函数是否能正确调用到底层 infer
# 我们 mock 底层的 predict_pass_prob，这样测试就不依赖于实际的模型文件是否存在
with patch("src.agent_app.predict_pass_prob") as mock_predict:
mock_predict.return_value = 0.85
prob = predict_student(None, 12, 8, 0.9, 2, "Self")
# 验证调用
assert prob == 0.85
mock_predict.assert_called_once_with(12, 8, 0.9, 2, "Self")

54
tests/test_counselor_agent.py Normal file
View File

@ -0,0 +1,54 @@
import pytest
import sys
import os
from unittest.mock import AsyncMock, MagicMock
# Ensure src is in path
sys.path.append(os.getcwd())
from src.agent_app import counselor_agent
from pydantic_ai.messages import ModelMessage, ModelRequest, ModelResponse, UserPromptPart, TextPart
@pytest.mark.asyncio
async def test_counselor_agent_conversation():
# We mock the model response to avoid needing real API key
# pydantic-ai allows passing a model to the agent or overriding it.
# However, for simplicity in this environment, we might just rely on the fact that
# if we don't have an API key, it will raise an error or we mock the run method.
# Mocking agent.run isn't ideal for integration, but good for logic check.
# Let's try to mock the model itself if possible.
# But locally we can just skip the actual LLM call if no key,
# OR we assume the user has key (which they seem to have in environment or sidebar).
# Let's just check if we can form the history and call the method signature correctly.
history = [
ModelRequest(parts=[UserPromptPart(content="我最近压力好大")]),
ModelResponse(parts=[TextPart(content="听到你这么说我很抱歉，能具体跟我说说吗？")])
]
# We won't actually await the run if we suspect it fails without auth.
# But we can verify the agent object is set up correctly.
assert counselor_agent is not None
# Verify we can access the tools
# PydanticAI 0.0.x tools validation
# We can inspect the agent's tools via internal attributes or Just trust the definition.
pass
@pytest.mark.asyncio
async def test_counselor_agent_structure():
# pydantic-ai Agent name is optional and strictly not the model name
assert counselor_agent is not None
# Basic check passed
@pytest.mark.asyncio
async def test_counselor_agent_streaming():
# Test if we can call run_stream (even if mocked or without auth,
# we might expect an error or just verify the method exists)
assert hasattr(counselor_agent, "run_stream")
# We might not be able to actually stream without a real model response unless we mock it.
# But checking the attribute confirms pydantic-ai version supports it roughly.
pass

52
tests/test_data.py Normal file
View File

@ -0,0 +1,52 @@
import sys
import os
import pandas as pd
import numpy as np
import pytest
# Ensure src is in path
sys.path.append(os.getcwd())
from src.data import generate_data, preprocess_data
def test_generate_data_structure():
"""Test if generate_data returns a DataFrame with correct shape and columns."""
df = generate_data(n_samples=50)
assert isinstance(df, pd.DataFrame)
assert len(df) == 50
expected_cols = [
"study_hours", "sleep_hours", "attendance_rate",
"study_type", "stress_level", "is_pass"
]
for col in expected_cols:
assert col in df.columns
def test_generate_data_content_range():
"""Test if generated data falls within expected value ranges."""
df = generate_data(n_samples=50)
assert df["study_hours"].min() >= 0
assert df["study_hours"].max() <= 20 # Based on generation logic (0-15 actually, but safely below 20)
assert df["sleep_hours"].min() >= 0
assert df["stress_level"].between(1, 5).all()
assert df["is_pass"].isin([0, 1]).all()
def test_generate_data_missing_values():
"""Test if generate_data creates missing values as expected (it has random logic)."""
# Generate enough samples to likely get nans
df = generate_data(n_samples=500, random_seed=42)
# Check if we have nans in specific columns that are supposed to have them
# In source: attendance_rate has 5% chance of nan
assert df["attendance_rate"].isnull().sum() >= 0
def test_preprocess_data():
"""Test basic preprocessing (deduplication)."""
df = pd.DataFrame({
"a": [1, 2, 2, 3],
"b": [1, 2, 2, 3]
})
clean_df = preprocess_data(df)
assert len(clean_df) == 3

66
tests/test_infer.py Normal file
View File

@ -0,0 +1,66 @@
import sys
import os
import pytest
from unittest.mock import patch
# Ensure src is in path
sys.path.append(os.getcwd())
from src.infer import predict_pass_prob, explain_prediction, load_model
# We need a fixture to create a valid model file for inference
@pytest.fixture(scope="module")
def train_dummy_model(tmp_path_factory):
"""Trains a quick dummy model and saves it to a temp dir."""
models_dir = tmp_path_factory.mktemp("models")
model_path = models_dir / "model.pkl"
# We reuse the logic from src.train but point to our temp path
# OR we can just manually create a pipeline and save it
# Reusing src.train is better integration testing
from src.train import get_pipeline, generate_data, preprocess_data
import joblib
df = generate_data(n_samples=20)
df = preprocess_data(df)
X = df.drop(columns=["is_pass"])
y = df["is_pass"]
pipeline = get_pipeline("rf")
pipeline.fit(X, y)
joblib.dump(pipeline, model_path)
return str(model_path)
@patch("src.infer._MODEL", None) # Reset global cached model
def test_predict_pass_prob(train_dummy_model):
"""Test prediction using the dummy trained model."""
with patch("src.infer.MODEL_PATH", train_dummy_model):
proba = predict_pass_prob(
study_hours=5.0,
sleep_hours=7.0,
attendance_rate=0.9,
stress_level=3,
study_type="Self"
)
assert 0.0 <= proba <= 1.0
@patch("src.infer._MODEL", None) # Reset global cached model
def test_explain_prediction(train_dummy_model):
"""Test explanation generation."""
with patch("src.infer.MODEL_PATH", train_dummy_model):
explanation = explain_prediction()
assert isinstance(explanation, str)
assert "模型特征重要性排名" in explanation
@patch("src.infer._MODEL", None)
def test_load_model_missing():
"""Test error handling when model is missing."""
with patch("src.infer.MODEL_PATH", "non_existent_path/model.pkl"):
# Should raise FileNotFoundError or be handled
with pytest.raises(FileNotFoundError):
predict_pass_prob(1,1,1,1,"Self") # This calls load_model internally

29
tests/test_model.py Normal file
View File

@ -0,0 +1,29 @@
import os
import joblib
import pytest
from src.train import train, MODEL_PATH
from src.infer import load_model, predict_pass_prob
def test_train_creates_model():
# 确保模型不存在或被覆盖
if os.path.exists(MODEL_PATH):
os.remove(MODEL_PATH)
train()
assert os.path.exists(MODEL_PATH)
model = joblib.load(MODEL_PATH)
assert model is not None
def test_inference():
# 确保模型存在
if not os.path.exists(MODEL_PATH):
train()
# 高概率情况 (大量学习/睡眠/出勤 + Group学习 + 低压力)
prob_high = predict_pass_prob(15, 8, 1.0, 1, "Group")
assert prob_high > 0.5
# 低概率情况 (不学习/不睡/缺勤 + 在线 + 高压力)
prob_low = predict_pass_prob(0, 3, 0.0, 5, "Online")
assert prob_low < 0.5

48
tests/test_train.py Normal file
View File

@ -0,0 +1,48 @@
import sys
import os
import pytest
from sklearn.pipeline import Pipeline
from unittest.mock import patch, MagicMock
# Ensure src is in path
sys.path.append(os.getcwd())
from src.train import get_pipeline, train
def test_get_pipeline_structure():
"""Test if get_pipeline returns a valid Scikit-learn pipeline."""
pipeline = get_pipeline("rf")
assert isinstance(pipeline, Pipeline)
assert "preprocessor" in pipeline.named_steps
assert "classifier" in pipeline.named_steps
def test_train_function_runs(tmp_path):
"""
Test if the train function runs without errors.
We mock generate_models to use a temp dir and run with small data.
"""
# Create a temporary directory for models
models_dir = tmp_path / "models"
model_path = models_dir / "model.pkl"
# Needs to be string for some os.path usages if they are strict, but pathlib usually works.
# However, src/train.py uses os.path.join(MODELS_DIR, ...), so we need to patch constants.
with patch("src.train.MODELS_DIR", str(models_dir)), \
patch("src.train.MODEL_PATH", str(model_path)), \
patch("src.train.generate_data") as mock_gen:
# Mock data generation to return a very small dataframe to speed up test
# We need to use real data structure though bc pipeline expects specific columns
from src.data import generate_data
real_small_df = generate_data(n_samples=10)
mock_gen.return_value = real_small_df
# Run training
try:
train()
except Exception as e:
pytest.fail(f"Train function failed with error: {e}")
# Check if model file was created
assert model_path.exists()