Spaces:
Running
Running
File size: 5,340 Bytes
8b6568e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | """
MathPulse AI — Tutor Nudge Service
Generates proactive AI tutor nudges for at-risk students using DeepSeek,
then writes them to Firestore for the floating tutor to surface.
"""
import logging
from datetime import datetime, timezone
logger = logging.getLogger("mathpulse.tutor_nudge")
NUDGE_COOLDOWN_HOURS = 24
SYSTEM_PROMPT = (
"You are MathPulse's AI tutor. Write a single short, friendly message "
"to nudge the student to work on their weakest topic. "
"No long explanation, just a nudge plus a concrete action. "
"1-2 sentences max. No code, no LaTeX. Be warm and encouraging."
)
def _get_db():
try:
from firebase_admin import firestore as ff
return ff.client()
except Exception:
return None
def _has_recent_nudge(db, student_id: str, topic: str) -> bool:
"""Check if an unconsumed nudge for this topic exists within cooldown."""
from datetime import timedelta
cutoff = datetime.now(timezone.utc) - timedelta(hours=NUDGE_COOLDOWN_HOURS)
nudges_ref = db.collection("tutorNudges").document(student_id).collection("nudges")
existing = (
nudges_ref
.where("topic", "==", topic)
.where("createdAt", ">=", cutoff)
.limit(1)
.get()
)
return len(existing) > 0
async def generate_tutor_nudge_for_student(
student_id: str,
weak_topics: list[str],
grade_level: str = "Grade 11",
recent_score: float | None = None,
) -> dict | None:
"""Generate a nudge message via DeepSeek and write to Firestore."""
if not weak_topics:
return None
db = _get_db()
if not db:
logger.warning("Firestore unavailable, skipping nudge generation")
return None
# Pick the first weak topic that doesn't have a recent nudge
topic = None
for t in weak_topics[:3]:
if not _has_recent_nudge(db, student_id, t):
topic = t
break
if not topic:
return None # All topics have recent nudges
# Generate nudge via DeepSeek
try:
from services.ai_client import get_deepseek_client, CHAT_MODEL
client = get_deepseek_client()
user_content = (
f"Student grade: {grade_level}. "
f"Weak topic: {topic}. "
f"{'Recent score: ' + str(round(recent_score)) + '%.' if recent_score else ''} "
f"Write a short nudge to encourage them to practice this topic."
)
response = client.chat.completions.create(
model=CHAT_MODEL,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_content},
],
temperature=0.7,
max_tokens=100,
)
message = (response.choices[0].message.content or "").strip()
if not message:
return None
except Exception as e:
logger.error(f"DeepSeek nudge generation failed for {student_id}: {e}")
return None
# Write to Firestore
nudge_data = {
"message": message,
"topic": topic,
"createdAt": datetime.now(timezone.utc),
"consumed": False,
}
try:
db.collection("tutorNudges").document(student_id).collection("nudges").add(nudge_data)
logger.info(f"Nudge written for {student_id}: topic={topic}")
except Exception as e:
logger.error(f"Failed to write nudge for {student_id}: {e}")
return None
return {"message": message, "topic": topic, "created_at": nudge_data["createdAt"].isoformat()}
async def check_and_generate_nudge(student_id: str) -> dict | None:
"""
Check if a student already has risk data warranting a nudge,
and generate one if no recent unconsumed nudge exists.
Used for students who completed diagnostics but have no new pipeline events.
"""
db = _get_db()
if not db:
return None
# Read existing risk profile from managedStudents
managed_ref = db.collection("managedStudents").document(student_id)
managed_snap = managed_ref.get()
if not managed_snap.exists:
return None
data = managed_snap.to_dict() or {}
risk_status = data.get("riskStatus")
if risk_status not in ("watch", "intervene", "critical", "at_risk"):
return None
# Get weak topics from student_profiles
profile_ref = db.collection("student_profiles").document(student_id)
profile_snap = profile_ref.get()
weak_topics: list[str] = []
if profile_snap.exists:
profile = profile_snap.to_dict() or {}
weak_topics = (
profile.get("quiz_performance", {}).get("lowest_accuracy_topics", [])
or profile.get("diagnostic", {}).get("weak_topics", [])
)
if not weak_topics:
# Fallback: use atRiskSubjects from user doc
user_ref = db.collection("users").document(student_id)
user_snap = user_ref.get()
if user_snap.exists:
weak_topics = (user_snap.to_dict() or {}).get("atRiskSubjects", [])
if not weak_topics:
return None
return await generate_tutor_nudge_for_student(
student_id=student_id,
weak_topics=weak_topics[:3],
grade_level=data.get("grade", "Grade 11"),
recent_score=data.get("systemPerformanceAvg") or data.get("diagnosticScore"),
)
|