diff --git a/services/teacher_ai_advisor.py b/services/teacher_ai_advisor.py index 7160a7b..ad14810 100644 --- a/services/teacher_ai_advisor.py +++ b/services/teacher_ai_advisor.py @@ -10,6 +10,11 @@ from utils.sse import sse_event, stream_text_chunks from utils.timezone import format_display_datetime from utils.access import class_student_filter +from utils.teacher_advice_prompt import ( + RISK_TAG_CALIBER_INSTRUCTION, + format_attention_student_line, + format_risk_reason, +) logger = logging.getLogger(__name__) @@ -145,7 +150,8 @@ def generate_class_suggestions(class_id, teacher_id, demo_run_id=None): 'student_id': row['student'].student_id, 'name': row['student'].full_name or row['student'].username, 'risk_tags': row['risk_tags'], - 'latest_score': row['latest_score'] + 'latest_score': row['latest_score'], + 'historical_score': row['student'].user_ascore }) # 2. 获取弱势知识点 (班级平均分最低的前3个) @@ -211,7 +217,7 @@ def generate_class_suggestions(class_id, teacher_id, demo_run_id=None): { 'student_id': s['student_id'], 'name': s['name'], - 'risk_reason': f"存在{'/'.join(s['risk_tags'])}风险，最近得分 {s['latest_score'] if s['latest_score'] is not None else '无'}" + 'risk_reason': format_risk_reason(s) } for s in attention_students ], 'weak_knowledge_points': [ @@ -236,7 +242,7 @@ def generate_class_suggestions(class_id, teacher_id, demo_run_id=None): if llm.is_available(): try: attention_details_str = "\n".join([ - f"- {s['name']} ({s['student_id']}): 风险标签 {s['risk_tags']}, 最近得分 {s['latest_score']}" + format_attention_student_line(s) for s in attention_students ]) if attention_students else "暂无高风险学生" @@ -255,6 +261,7 @@ def generate_class_suggestions(class_id, teacher_id, demo_run_id=None): "role": "system", "content": ( "你是一个专业的编程教育专家，擅长从班级学生成绩、提交活跃度、弱点概念等多维度给出点对点的教学建议。\n" + + RISK_TAG_CALIBER_INSTRUCTION + "\n" "请用简洁专业的中文直接输出分析内容，并在正文结束后，输出一行由 `===JSON===` 分隔的 JSON 字符串，包含系统结构。" ) }, @@ -500,7 +507,8 @@ def _generate_class_suggestions_stream(class_id, teacher_id, demo_run_id=None, * 'student_id': row['student'].student_id, 'name': row['student'].full_name or row['student'].username, 'risk_tags': row['risk_tags'], - 'latest_score': row['latest_score'] + 'latest_score': row['latest_score'], + 'historical_score': row['student'].user_ascore }) yield sse_event({'type': 'status', 'message': '正在聚合知识点雷达掌握度...'}) @@ -564,7 +572,7 @@ def _generate_class_suggestions_stream(class_id, teacher_id, demo_run_id=None, * { 'student_id': s['student_id'], 'name': s['name'], - 'risk_reason': f"存在{'/'.join(s['risk_tags'])}风险，最近得分 {s['latest_score'] if s['latest_score'] is not None else '无'}" + 'risk_reason': format_risk_reason(s) } for s in attention_students ], 'weak_knowledge_points': [ @@ -593,7 +601,7 @@ def _generate_class_suggestions_stream(class_id, teacher_id, demo_run_id=None, * if llm.is_available(): try: attention_details_str = "\n".join([ - f"- {s['name']} ({s['student_id']}): 风险标签 {s['risk_tags']}, 最近得分 {s['latest_score']}" + format_attention_student_line(s) for s in attention_students ]) if attention_students else "暂无高风险学生" @@ -612,6 +620,7 @@ def _generate_class_suggestions_stream(class_id, teacher_id, demo_run_id=None, * "role": "system", "content": ( "你是一个专业的编程教育专家，擅长从班级 student 分数、提交活跃度、弱点概念等多维度给出建议。\n" + + RISK_TAG_CALIBER_INSTRUCTION + "\n" "请用学术严谨且易懂的中文直接输出 Markdown 格式的分析正文。只输出正文，不要输出 JSON、分隔符或额外说明。左侧结构化卡片由系统根据班级数据生成。" ) }, diff --git a/tests/test_teacher_advice_prompt.py b/tests/test_teacher_advice_prompt.py new file mode 100644 index 0000000..b18f1e1 --- /dev/null +++ b/tests/test_teacher_advice_prompt.py @@ -0,0 +1,88 @@ +"""教师 AI 建议 prompt 构造的纯函数测试。 + +仅标准库，不 import Flask、不连数据库/Redis、不访问网络。 + +背景：服务端把"风险标签 ['低分']"与"最近得分 100"两个未解释口径的数据 +并列喂给 LLM，LLM 据此推断"数据记录异常或系统误判"，误导教师排查不存 +在的系统故障。实际上"低分"标签在历史综合分 <60 或最近得分 <60 时都会 +触发（services/teacher_analytics.py）。 +""" +from utils.teacher_advice_prompt import ( + RISK_TAG_CALIBER_INSTRUCTION, + format_attention_student_line, + format_risk_reason, +) + + +def test_line_explains_low_tag_comes_from_history_when_latest_is_high(): + # 孙三场景：历史综合分 44（<60）触发低分标签，最近一次 100 + line = format_attention_student_line({ + 'name': '孙三', + 'student_id': 'demo_s_003', + 'risk_tags': ['低分'], + 'latest_score': 100, + 'historical_score': 44, + }) + + assert '孙三' in line and 'demo_s_003' in line + assert '100' in line and '44' in line + # 必须显式给出低分标签的来源口径，防止 LLM 臆断为数据矛盾 + assert '历史综合分' in line + assert '来自' in line + + +def test_line_normal_low_score_keeps_both_scores(): + line = format_attention_student_line({ + 'name': '周四', + 'student_id': 'demo_s_004', + 'risk_tags': ['低分'], + 'latest_score': 40, + 'historical_score': 35, + }) + + assert '40' in line and '35' in line + assert '低分' in line + + +def test_line_no_submission_shows_none_for_scores(): + line = format_attention_student_line({ + 'name': '新手', + 'student_id': 'demo_s_099', + 'risk_tags': ['未提交'], + 'latest_score': None, + 'historical_score': None, + }) + + assert '未提交' in line + assert '无' in line # 最近得分无 + + +def test_instruction_explains_tag_caliber_and_forbids_false_anomaly_claim(): + text = RISK_TAG_CALIBER_INSTRUCTION + + # 口径：低分标签在历史综合分或最近得分低于阈值时触发 + assert '历史综合分' in text + assert '60' in text + # 明确禁止：标签与最近得分并存不是数据矛盾，不得推断为系统异常 + assert '系统异常' in text or '系统误判' in text + + +def test_risk_reason_explains_low_tag_from_history(): + reason = format_risk_reason({ + 'risk_tags': ['低分'], + 'latest_score': 100, + 'historical_score': 44, + }) + + assert '低分' in reason and '100' in reason and '44' in reason + assert '历史综合分' in reason + + +def test_risk_reason_plain_low_score_stays_concise(): + reason = format_risk_reason({ + 'risk_tags': ['低分'], + 'latest_score': 20, + 'historical_score': 30, + }) + + assert reason == '存在低分风险，最近得分 20' diff --git a/utils/teacher_advice_prompt.py b/utils/teacher_advice_prompt.py new file mode 100644 index 0000000..14e2fe2 --- /dev/null +++ b/utils/teacher_advice_prompt.py @@ -0,0 +1,88 @@ +"""教师 AI 个性化建议的 prompt 片段构造（纯函数，无 Flask/DB 依赖）。 + +口径事实与 ``services/teacher_analytics.py`` 的 +``_risk_tags_for_student`` 保持一致：“低分”标签在历史综合分 +（``User.user_ascore``）低于阈值 **或** 最近一次得分低于阈值时都会触发。 +因此“低分标签 + 最近一次高分”不是数据矛盾，不能让 LLM 推断为系统异常。 +""" + +LOW_SCORE_THRESHOLD = 60 + +RISK_TAG_CALIBER_INSTRUCTION = ( + "数据口径说明：\n" + "- 学生的历史综合分与最近一次得分是两个不同指标。\n" + "- “低分”风险标签在历史综合分低于 60 或最近一次得分低于 60 时触发。\n" + "- 因此“风险标签含低分”与“最近一次得分较高”同时出现并不矛盾，" + "通常表示该生历史基础薄弱但近期有进步。禁止在没有证据时把这种情况" + "描述为数据记录异常或系统误判，应结合历史趋势给出鼓励或巩固建议。" +) + + +def _score_text(value): + return "无" if value is None else str(value) + + +def format_attention_student_line(info): + """构造单个需关注学生的 prompt 行。 + + Args: + info: ``{name, student_id, risk_tags, latest_score, historical_score}``， + 分数可为 None（未提交 / 无历史评分）。 + + 行内同时给出最近一次得分与历史综合分；当“低分”标签实际来自历史 + 综合分而最近一次已达阈值时，追加口径注释，避免 LLM 把两个分数误读 + 为互相矛盾。 + """ + name = info["name"] + student_id = info["student_id"] + tags = "、".join(info.get("risk_tags") or []) or "无" + latest = info.get("latest_score") + historical = info.get("historical_score") + + line = ( + f"- {name} ({student_id}): 风险标签 {tags}，" + f"最近一次得分 {_score_text(latest)}，" + f"历史综合分 {_score_text(historical)}" + ) + + low_tag_from_history = ( + "低分" in tags + and historical is not None + and historical < LOW_SCORE_THRESHOLD + and (latest is None or latest >= LOW_SCORE_THRESHOLD) + ) + if low_tag_from_history: + line += ( + f"（注：低分标签来自历史综合分 {historical} 低于 60，" + f"最近一次已达 {_score_text(latest)}，不属数据矛盾，" + "请按历史薄弱、近期进步给出建议）" + ) + + return line + + +def format_risk_reason(info): + """构造结构化卡片中的 ``risk_reason`` 短文本。 + + 与 :func:`format_attention_student_line` 同一口径：低分标签来自历史 + 综合分而最近一次已达阈值时，补充来源说明，避免教师误读为数据矛盾。 + """ + tags = "/".join(info.get("risk_tags") or []) + latest = info.get("latest_score") + historical = info.get("historical_score") + + reason = f"存在{tags}风险，最近得分 {_score_text(latest)}" + + low_tag_from_history = ( + "低分" in tags + and historical is not None + and historical < LOW_SCORE_THRESHOLD + and (latest is None or latest >= LOW_SCORE_THRESHOLD) + ) + if low_tag_from_history: + reason += ( + f"；低分标签来自历史综合分 {historical} 低于 60，" + f"最近一次已达 {_score_text(latest)}，属历史薄弱、近期提升" + ) + + return reason