K12 AI 智能教育系统的工程化实践：知识追踪、自适应推荐与未成年人数据合规

接入层   学生端、教师端、家长端（权限分离）
服务层   学情分析、推荐、作业、报告
模型层   知识追踪、难度估计、内容标签
数据层   学习记录、知识点图谱、题库
合规层   同意管理、脱敏、留存、审核

from dataclasses import dataclass, field
from enum import Enum

class Mastery(str, Enum):
    unknown = "unknown"
    weak = "weak"
    medium = "medium"
    strong = "strong"

@dataclass
class KnowledgeNode:
    node_id: str
    name: str
    prerequisites: list[str] = field(default_factory=list)

@dataclass
class BKTParams:
    learn: float = 0.15      # 未掌握到掌握的概率
    forget: float = 0.05     # 掌握到遗忘的概率
    guess: float = 0.20      # 未掌握但答对的概率
    slip: float = 0.10       # 掌握但答错的概率

def update_mastery(p_mastery: float, correct: bool,
                   p: BKTParams) -> float:
    if correct:
        num = p_mastery * (1 - p.slip)
        den = num + (1 - p_mastery) * p.guess
    else:
        num = p_mastery * p.slip
        den = num + (1 - p_mastery) * (1 - p.guess)
    posterior = num / den if den else p_mastery
    return posterior + (1 - posterior) * p.learn

def to_level(p: float) -> Mastery:
    if p >= 0.85:
        return Mastery.strong
    if p >= 0.6:
        return Mastery.medium
    if p >= 0.3:
        return Mastery.weak
    return Mastery.unknown

from dataclasses import dataclass

@dataclass
class Exercise:
    ex_id: str
    node_id: str
    difficulty: float        # 0-1
    est_minutes: float

def recommend(nodes: dict[str, KnowledgeNode],
              mastery: dict[str, float],
              exercises: list[Exercise],
              target: float = 0.65,
              limit: int = 5) -> list[Exercise]:
    candidates = []
    for ex in exercises:
        p = mastery.get(ex.node_id, 0.0)
        # 只推荐前置知识已掌握的知识点
        node = nodes.get(ex.node_id)
        if node and any(mastery.get(pre, 0) < 0.6
                        for pre in node.prerequisites):
            continue
        # 难度贴近当前掌握度 + 适度挑战
        gap = abs(ex.difficulty - min(p + 0.15, 1.0))
        if gap > 0.35:
            continue
        candidates.append((gap, ex))

    candidates.sort(key=lambda x: x[0])
    return [ex for _, ex in candidates[:limit]]

import re
import time
import hashlib
from dataclasses import dataclass
from enum import Enum

class AgeGroup(str, Enum):
    under_14 = "under_14"     # 需监护人同意
    teen = "teen"             # 14-18
    adult = "adult"

@dataclass
class ConsentRecord:
    user_id_hash: str
    age_group: AgeGroup
    guardian_consent: bool
    consented_at: float
    purpose: str

PII = re.compile(r"(\d{11}|\d{17}[\dXx]|[\w.+-]+@[\w-]+\.[\w.]+)")

def pseudonymize(user_id: str, salt: str) -> str:
    raw = f"{salt}:{user_id}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

def mask_pii(text: str) -> str:
    return PII.sub("[REDACTED]", text)

def can_collect(consent: ConsentRecord, field: str) -> bool:
    """最小必要原则：字段必须在授权目的范围内"""
    allowed = {
        "answer_record": {"教学分析"},
        "study_time": {"教学分析"},
        "error_type": {"教学分析"},
    }
    if field not in allowed:
        return False
    if consent.age_group == AgeGroup.under_14 \
            and not consent.guardian_consent:
        return False
    return consent.purpose in allowed[field]

def retention_ok(created_at: float, days: int = 365) -> bool:
    return (time.time() - created_at) < days * 86400

import re

# 面向未成年人的额外禁止项
MINOR_BANNED = {
    "排名倒数", "差生", "笨", "没救了",      # 羞辱性表达
    "必须", "否则", "再不学就晚了",           # 焦虑营销
    "暴力", "色情", "赌博", "自杀",
}

PII = re.compile(r"(\d{11}|\d{17}[\dXx]|[\w.+-]+@[\w-]+\.[\w.]+)")

def audit_for_minor(text: str) -> list[str]:
    issues = []
    for w in MINOR_BANNED:
        if w in text:
            issues.append(f"含不适宜用语：{w}")
    if PII.search(text):
        issues.append("含个人信息，需脱敏")
    return issues

def safe_feedback(score: float, text: str) -> str:
    """反馈必须正向、具体、不比较"""
    issues = audit_for_minor(text)
    if issues:
        return "本次练习已完成，我们来看看哪些地方可以再巩固。"
    return text