fix: 教师 AI 周报补充低分标签口径，避免误报数据异常

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、分隔符或额外说明。左侧结构化卡片由系统根据班级数据生成。"
                     )
                 },

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'

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