Update agents.py

agents.py

@@ -1,77 +1,233 @@
 import os
 import json
 import requests
+import re
+from collections import Counter
 
-# GLM-4.5 API configuration
 GLM_API_URL = "https://api.your-glm-provider.com/v1/chat/completions"
 GLM_API_KEY = os.getenv("ZHIPUAI_API_KEY")  # Hugging Face Secret
 
-def call_glm(system_prompt, user_prompt, temperature=0.3):
-    headers = {"Authorization": f"Bearer {GLM_API_KEY}"}
+def _safe_json_loads(s):
+    """
+    Try to extract JSON substring and load. Handles cases where model returns extraneous text.
+    """
+    if not s:
+        return None
+    try:
+        return json.loads(s)
+    except Exception:
+        # try to find first { ... } block
+        m = re.search(r"(\{[\s\S]*\})", s)
+        if m:
+            try:
+                return json.loads(m.group(1))
+            except Exception:
+                return None
+    return None
+
+def call_glm(system_prompt, user_prompt, temperature=0.2, max_tokens=800):
+    if not GLM_API_KEY:
+        raise RuntimeError("ZHIPUAI_API_KEY not set in environment")
+    headers = {"Authorization": f"Bearer {GLM_API_KEY}", "Content-Type": "application/json"}
     payload = {
         "model": "glm-4.5",
         "messages": [
             {"role": "system", "content": system_prompt},
-            {"role": "user", "content": user_prompt},
+            {"role": "user", "content": user_prompt}
         ],
         "temperature": temperature,
+        "max_tokens": max_tokens
     }
-    response = requests.post(GLM_API_URL, headers=headers, json=payload)
-    result = response.json()
-    return result["choices"][0]["message"]["content"]
+    resp = requests.post(GLM_API_URL, headers=headers, json=payload, timeout=60)
+    resp.raise_for_status()
+    data = resp.json()
+    # get content robustly
+    content = None
+    try:
+        # different APIs may return different shapes
+        content = data["choices"][0]["message"]["content"]
+    except Exception:
+        # fallback try common fields
+        content = data["choices"][0]["text"] if "choices" in data and data["choices"] else None
+    return content
 
 class AnalyzerAgent:
-    def analyze(self, answers, exam_data):
-        system_prompt = "You are an exam analysis assistant."
-        user_prompt = f"""
-        Analyze student answers against correct answers.
-        Answers: {json.dumps(answers)}
-        Exam questions: {json.dumps(exam_data)}
-        Return JSON: {{
-          "topic_accuracy": {{"topic": 0-1}},
-          "weak_topics": ["..."]
-        }}
-        """
-        try:
-            response = call_glm(system_prompt, user_prompt)
-            return json.loads(response)
-        except Exception as e:
-            return {"error": f"Analyzer failed: {str(e)}"}
-
-class ForecastAgent:
-    def forecast(self, level, subject):
-        system_prompt = "You are an exam forecast assistant."
-        user_prompt = f"""
-        Predict 3 high-probability exam topics for {level} {subject}.
-        Return JSON: {{
-          "predicted_topics": [
-            {{"topic": "...", "confidence": 0-1}}
-          ]
-        }}
-        """
+    def analyze(self, per_question):
+        # Build stats
+        topic_stats = {}
+        for qid, info in per_question.items():
+            topics = info.get("topics", [])
+            correct = 1 if (info.get("user") is not None and info.get("correct") is not None and str(info["user"]).strip() == str(info["correct"]).strip()) else 0
+            for t in topics:
+                if t not in topic_stats:
+                    topic_stats[t] = {"correct": 0, "total": 0}
+                topic_stats[t]["total"] += 1
+                topic_stats[t]["correct"] += correct
+
+        stats_json = {t: {"correct": v["correct"], "total": v["total"], "accuracy": round(v["correct"]/v["total"], 3) if v["total"] else 0.0} for t, v in topic_stats.items()}
+
+        system_prompt = "You are an exam analysis assistant for SPM-style multiple choice exams. Return only valid JSON."
+        user_prompt = (
+            f"Input: topic_stats = {json.dumps(stats_json)}\n\n"
+            "Compute: topic accuracy and list weak_topics (accuracy < 0.65 and at least 3 questions). "
+            "Return JSON like: {\"topic_accuracy\": {\"topic\": 0.0}, \"weak_topics\": [\"topic1\", ...], \"recommendation_summary\": \"short text\"}."
+        )
         try:
-            response = call_glm(system_prompt, user_prompt)
-            return json.loads(response)
-        except Exception as e:
-            return {"error": f"Forecast failed: {str(e)}"}
+            resp = call_glm(system_prompt, user_prompt, temperature=0.0, max_tokens=300)
+            parsed = _safe_json_loads(resp)
+            if parsed:
+                return parsed
+        except Exception:
+            pass
+
+        # deterministic fallback
+        weak = [t for t, v in stats_json.items() if v["total"] >= 3 and v["accuracy"] < 0.65]
+        rec = "Focus on: " + ", ".join(weak) if weak else "No major weak topics detected."
+        return {"topic_accuracy": {t: v["accuracy"] for t, v in stats_json.items()}, "weak_topics": weak, "recommendation_summary": rec}
 
 class CoachAgent:
     def coach(self, analysis, level, subject):
-        system_prompt = "You are a study coach for school exams."
-        user_prompt = f"""
-        Based on this analysis: {json.dumps(analysis)},
-        suggest a study plan and 3 practice questions for {level} {subject}.
-        Return JSON: {{
-          "tips": ["..."],
-          "study_plan": "...",
-          "practice_questions": [
-            {{"text": "...", "answer": "..."}}
-          ]
-        }}
+        system_prompt = "You are a concise study coach helping a Form5 (SPM) student. Return only JSON."
+        user_prompt = (
+            f"Student analysis: {json.dumps(analysis)}\n"
+            f"Level: {level}, Subject: {subject}\n\n"
+            "Return JSON with keys: 'tips' (list of 3 short tips), 'study_plan' (1-line daily plan), "
+            "'practice_questions' (array of 3 objects {'text','choices','answer','explanation','topic'})."
+        )
+        try:
+            resp = call_glm(system_prompt, user_prompt, temperature=0.25, max_tokens=700)
+            parsed = _safe_json_loads(resp)
+            if parsed:
+                return parsed
+        except Exception:
+            pass
+        return {"tips": ["Practice regularly", "Focus on weak topics", "Review solutions"], "study_plan": "20 mins/day for 2 weeks", "practice_questions": []}
+
+class PredictiveAgent:
+    """
+    PredictiveAgent generates predicted questions for a subject (SPM Form5),
+    caches predictions to disk, and provides helper methods to inject them into the question pool.
+    """
+    def __init__(self, cache_path="predictions_cache.json"):
+        self.cache_path = cache_path
+        if not os.path.exists(self.cache_path):
+            with open(self.cache_path, "w", encoding="utf-8") as f:
+                json.dump({}, f)
+
+    def _compute_stats(self, level, subject, question_bank):
+        topic_counter = Counter()
+        difficulty_counts = Counter()
+        total = 0
+        for q in question_bank:
+            if q.get("subject") != f"{level}_{subject}":
+                continue
+            total += 1
+            for t in q.get("topics", []):
+                topic_counter[t] += 1
+            d = q.get("difficulty")
+            if isinstance(d, (int, float)):
+                difficulty_counts[int(d)] += 1
+        top_topics = topic_counter.most_common(30)
+        topic_freqs = [{"topic": t, "count": c, "pct": round(c/total, 3) if total else 0.0} for t, c in top_topics]
+        difficulty_dist = {str(k): v for k, v in difficulty_counts.items()}
+        return {"total_questions": total, "topic_freqs": topic_freqs, "difficulty_dist": difficulty_dist}
+
+    def _load_cache(self):
+        with open(self.cache_path, "r", encoding="utf-8") as f:
+            return json.load(f)
+
+    def _save_cache(self, cache):
+        with open(self.cache_path, "w", encoding="utf-8") as f:
+            json.dump(cache, f, indent=2, ensure_ascii=False)
+
+    def get_or_generate_predictions(self, level, subject, question_bank, n=5):
+        """
+        Return cached predictions if present; otherwise call GLM to generate n predicted questions.
+        Each predicted question: {text, choices, predicted_answer, confidence, topic, difficulty}
         """
+        key = f"{level}_{subject}"
+        cache = self._load_cache()
+        if key in cache and cache[key].get("predictions"):
+            return cache[key]["predictions"]
+
+        # compute stats and send to GLM
+        stats = self._compute_stats(level, subject, question_bank)
+        system_prompt = "You are an expert SPM forecaster and question writer. Return only JSON."
+        user_prompt = (
+            f"Context: aggregated SPM past-paper stats for {level} {subject}.\n"
+            f"Stats: {json.dumps(stats, ensure_ascii=False)}\n\n"
+            f"Task: Produce {n} *predicted* exam-style MCQ questions that are likely to appear in SPM 2025-2026. "
+            "For each question return: text, choices (array), predicted_answer (exact choice text), confidence (0-1), topic (short), difficulty (1-5). "
+            "Return JSON: {\"predicted_questions\": [{...}] , \"predicted_topics\": [{\"topic\":\"\",\"confidence\":0.0}], \"rationale\":\"short\"}.\n"
+            "Be conservative with confidence and do NOT claim certainty. Mark source as 'predicted' in each question object."
+        )
         try:
-            response = call_glm(system_prompt, user_prompt)
-            return json.loads(response)
-        except Exception as e:
-            return {"error": f"Coach failed: {str(e)}"}
+            resp = call_glm(system_prompt, user_prompt, temperature=0.25, max_tokens=1200)
+            parsed = _safe_json_loads(resp)
+            if parsed and "predicted_questions" in parsed:
+                preds = parsed["predicted_questions"]
+            else:
+                # Try to parse direct list returned
+                parsed_list = _safe_json_loads(resp)
+                if isinstance(parsed_list, list):
+                    preds = parsed_list[:n]
+                else:
+                    preds = []
+        except Exception:
+            preds = []
+
+        # fallback heuristic: empty predictions
+        if not preds:
+            preds = []
+            # create n simple placeholders using top topics
+            top_topics = [t["topic"] for t in stats["topic_freqs"][:min(3, len(stats["topic_freqs"]))]]
+            for i in range(n):
+                t = top_topics[i % (len(top_topics) if top_topics else 1)] if top_topics else "general"
+                preds.append({
+                    "text": f"Practice predicted question on {t} (placeholder) #{i+1}",
+                    "choices": ["A","B","C","D"],
+                    "predicted_answer": "A",
+                    "confidence": 0.3,
+                    "topic": t,
+                    "difficulty": 3
+                })
+
+        # store in cache
+        cache[key] = {"predictions": preds}
+        self._save_cache(cache)
+        return preds
+
+    def predict(self, level, subject, question_bank):
+        """
+        Return a prediction summary for UI: predicted_topics, rationale, sample_questions.
+        """
+        key = f"{level}_{subject}"
+        cache = self._load_cache()
+        if key in cache and cache[key].get("predictions"):
+            preds = cache[key]["predictions"]
+            # Build a simple summary
+            sample_questions = []
+            for p in preds[:5]:
+                sample_questions.append({
+                    "text": p.get("text"),
+                    "choices": p.get("choices", []),
+                    "predicted_answer": p.get("predicted_answer", ""),
+                    "confidence": p.get("confidence", 0.0),
+                    "topic": p.get("topic", "")
+                })
+            return {"predicted_topics": [p.get("topic") for p in preds[:6]], "rationale": "Cached predictions", "sample_questions": sample_questions}
+        else:
+            # generate on the fly and return the structured full JSON from GLM
+            preds = self.get_or_generate_predictions(level, subject, question_bank, n=6)
+            sample_questions = []
+            for p in preds[:5]:
+                sample_questions.append({
+                    "text": p.get("text"),
+                    "choices": p.get("choices", []),
+                    "predicted_answer": p.get("predicted_answer", ""),
+                    "confidence": p.get("confidence", 0.0),
+                    "topic": p.get("topic", "")
+                })
+            return {"predicted_topics": [p.get("topic") for p in preds[:6]], "rationale": "Generated predictions", "sample_questions": sample_questions}
+