Upload backend/routes/rag_routes.py with huggingface_hub

backend/routes/rag_routes.py

@@ -0,0 +1,427 @@
+from __future__ import annotations
+
+import json
+import logging
+import os
+import re
+from datetime import datetime, timezone
+from threading import Lock
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel, Field
+
+from services.inference_client import (
+    InferenceRequest,
+    create_default_client,
+    is_sequential_model,
+    get_model_for_task,
+)
+from rag.curriculum_rag import (
+    build_analysis_curriculum_context,
+    build_lesson_prompt,
+    build_lesson_query,
+    build_problem_generation_prompt,
+    format_retrieved_chunks,
+    retrieve_curriculum_context,
+    retrieve_lesson_pdf_context,
+    summarize_retrieval_confidence,
+)
+from rag.vectorstore_loader import get_vectorstore_health, reset_vectorstore_singleton
+
+try:
+    from firebase_admin import firestore as firebase_firestore
+except Exception:
+    firebase_firestore = None
+
+logger = logging.getLogger("mathpulse.rag")
+router = APIRouter(prefix="/api/rag", tags=["rag"])
+
+_inference_client = None
+_inference_lock = Lock()
+
+
+def _get_inference_client():
+    global _inference_client
+    if _inference_client is None:
+        with _inference_lock:
+            if _inference_client is None:
+                _inference_client = create_default_client()
+    return _inference_client
+
+
+async def _generate_text(
+    prompt: str,
+    task_type: str,
+    max_new_tokens: int = 900,
+    enable_thinking: bool = False,
+) -> str:
+    request = InferenceRequest(
+        messages=[
+            {"role": "system", "content": "You are a precise DepEd-aligned curriculum assistant."},
+            {"role": "user", "content": prompt},
+        ],
+        task_type=task_type,
+        max_new_tokens=max_new_tokens,
+        temperature=0.2,
+        top_p=0.9,
+        enable_thinking=enable_thinking,
+    )
+    return _get_inference_client().generate_from_messages(request)
+
+
+def _log_rag_usage(
+    request: Request,
+    *,
+    event_type: str,
+    topic: str,
+    subject: str,
+    quarter: Optional[int],
+    chunks: List[Dict[str, Any]],
+) -> None:
+    if firebase_firestore is None:
+        return
+    try:
+        user = getattr(request.state, "user", None)
+        uid = getattr(user, "uid", None)
+        domains = sorted({str(chunk.get("content_domain") or "").strip() for chunk in chunks if chunk.get("content_domain")})
+        top_score = max((float(chunk.get("score") or 0.0) for chunk in chunks), default=0.0)
+        payload = {
+            "userId": uid,
+            "type": event_type,
+            "topic": topic,
+            "subject": subject,
+            "quarter": quarter,
+            "retrievedChunks": len(chunks),
+            "topScore": top_score,
+            "curriculumDomainsHit": domains,
+            "timestamp": firebase_firestore.SERVER_TIMESTAMP,
+            "createdAtIso": datetime.now(timezone.utc).isoformat(),
+        }
+        firebase_firestore.client().collection("rag_usage").add(payload)
+    except Exception as exc:
+        logger.warning("rag_usage logging skipped: %s", exc)
+
+
+def _strip_thinking_and_parse(text: str) -> dict:
+    cleaned = text.strip()
+    cleaned = re.sub(r" </think>", "", cleaned, flags=re.DOTALL).strip()
+    if "{" in cleaned and "}" in cleaned:
+        try:
+            start = cleaned.find("{")
+            end = cleaned.rfind("}") + 1
+            parsed = json.loads(cleaned[start:end])
+            if isinstance(parsed, dict):
+                return parsed
+        except Exception:
+            pass
+    return {"explanation": text}
+
+
+class RagLessonRequest(BaseModel):
+    topic: str
+    subject: str
+    quarter: int
+    lessonTitle: Optional[str] = None
+    learningCompetency: Optional[str] = None
+    moduleUnit: Optional[str] = None
+    learnerLevel: Optional[str] = None
+    userId: Optional[str] = None
+    moduleId: Optional[str] = None
+    lessonId: Optional[str] = None
+    competencyCode: Optional[str] = None
+    storagePath: Optional[str] = None
+
+
+class RagProblemRequest(BaseModel):
+    topic: str
+    subject: str
+    quarter: int
+    difficulty: str = Field(default="medium")
+    userId: Optional[str] = None
+
+
+class RagAnalysisContextRequest(BaseModel):
+    weakTopics: List[str]
+    subject: str
+    userId: Optional[str] = None
+
+
+@router.get("/health")
+async def rag_health():
+    active_model = get_model_for_task("rag_lesson")
+    is_seq = is_sequential_model(active_model)
+    try:
+        health = get_vectorstore_health()
+        return {
+            "status": "ok",
+            "chunkCount": health["chunkCount"],
+            "subjects": health["subjects"],
+            "lastIngested": datetime.now(timezone.utc).isoformat(),
+            "activeModel": active_model,
+            "isSequentialModel": is_seq,
+        }
+    except Exception as exc:
+        return {
+            "status": "degraded",
+            "chunkCount": 0,
+            "subjects": {},
+            "lastIngested": None,
+            "activeModel": active_model,
+            "isSequentialModel": is_seq,
+            "warning": str(exc),
+        }
+
+
+def _fetch_youtube_video(lesson_title: str, subject: str, competency: str, quarter: int) -> dict:
+    try:
+        from backend.services.youtube_service import get_video_for_lesson
+    except ImportError:
+        return {}
+    try:
+        video = get_video_for_lesson(lesson_title, subject, competency, quarter)
+        return video or {}
+    except Exception as e:
+        logger.warning("YouTube search failed: %s", e)
+        return {}
+
+
+def _ensure_7_sections(lesson_data: dict, lesson_title: str) -> dict:
+    sections = lesson_data.get("sections", [])
+    section_types = {s.get("type") for s in sections}
+    required = ["introduction", "key_concepts", "video", "worked_examples", "important_notes", "try_it_yourself", "summary"]
+
+    default_content = {
+        "introduction": {"type": "introduction", "title": "Introduction", "content": f"Welcome to the lesson on {lesson_title}."},
+        "key_concepts": {"type": "key_concepts", "title": "Key Concepts", "content": "Below are the key concepts covered in this lesson.", "callouts": []},
+        "video": {"type": "video", "title": "Video Lesson", "content": "Watch this explanation to understand the concepts visually.", "videoId": "", "videoTitle": "", "videoChannel": "", "embedUrl": "", "thumbnailUrl": ""},
+        "worked_examples": {"type": "worked_examples", "title": "Worked Examples", "examples": []},
+        "important_notes": {"type": "important_notes", "title": "Important Notes", "bulletPoints": []},
+        "try_it_yourself": {"type": "try_it_yourself", "title": "Try It Yourself", "practiceProblems": []},
+        "summary": {"type": "summary", "title": "Summary", "content": f"Great job completing the lesson on {lesson_title}!"},
+    }
+
+    filled = {}
+    for req_type in required:
+        for existing in sections:
+            if existing.get("type") == req_type:
+                filled[req_type] = existing
+                break
+        else:
+            filled[req_type] = default_content[req_type]
+
+    ordered = [filled[t] for t in required]
+
+    for i, section in enumerate(ordered):
+        s_type = section.get("type")
+        if s_type == "key_concepts" and not section.get("callouts"):
+            section["callouts"] = []
+        if s_type == "worked_examples" and not section.get("examples"):
+            section["examples"] = []
+        if s_type == "important_notes" and not section.get("bulletPoints"):
+            section["bulletPoints"] = []
+        if s_type == "try_it_yourself" and not section.get("practiceProblems"):
+            section["practiceProblems"] = []
+        ordered[i] = section
+
+    return {**lesson_data, "sections": ordered}
+
+
+@router.post("/lesson")
+async def rag_lesson(request: Request, payload: RagLessonRequest):
+    try:
+        chunks, retrieval_mode = retrieve_lesson_pdf_context(
+        query=build_lesson_query(
+            payload.topic,
+            payload.subject,
+            payload.quarter,
+            lesson_title=payload.lessonTitle,
+            competency=payload.learningCompetency,
+            module_unit=payload.moduleUnit,
+            learner_level=payload.learnerLevel,
+        ),
+        subject=payload.subject,
+        quarter=payload.quarter,
+        lesson_title=payload.lessonTitle,
+        competency=payload.learningCompetency,
+        module_id=payload.moduleId,
+        lesson_id=payload.lessonId,
+        competency_code=payload.competencyCode,
+        storage_path=payload.storagePath,
+        top_k=8,
+    )
+
+    if not chunks:
+        raise HTTPException(
+            status_code=404,
+            detail={
+                "error": "no_curriculum_context",
+                "message": f"No curriculum content found for lesson '{payload.lessonTitle}' ({payload.subject} Q{payload.quarter}). Please ensure the PDF has been ingested.",
+                "retrievalBand": "low",
+                "sources": [],
+            },
+        )
+
+    prompt = build_lesson_prompt(
+        lesson_title=payload.lessonTitle or payload.topic,
+        competency=payload.learningCompetency or payload.topic,
+        grade_level="Grade 11-12",
+        subject=payload.subject,
+        quarter=payload.quarter,
+        learner_level=payload.learnerLevel,
+        module_unit=payload.moduleUnit,
+        curriculum_chunks=chunks,
+        competency_code=payload.competencyCode,
+    )
+
+    raw_explanation = await _generate_text(
+        prompt,
+        task_type="lesson_generation",
+        max_new_tokens=1800,
+        enable_thinking=True,
+    )
+
+    parsed_lesson = _strip_thinking_and_parse(raw_explanation)
+    parsed_lesson = _ensure_7_sections(parsed_lesson, payload.lessonTitle or payload.topic)
+
+    if parsed_lesson.get("sections"):
+        video_section = next((s for s in parsed_lesson["sections"] if s.get("type") == "video"), None)
+        if video_section:
+            video_data = _fetch_youtube_video(
+                payload.lessonTitle or payload.topic,
+                payload.subject,
+                payload.learningCompetency or "",
+                payload.quarter,
+            )
+            if video_data:
+                video_section["videoId"] = video_data.get("videoId", "")
+                video_section["videoTitle"] = video_data.get("videoTitle", "")
+                video_section["videoChannel"] = video_data.get("videoChannel", "")
+                video_section["embedUrl"] = video_data.get("embedUrl", "")
+                video_section["thumbnailUrl"] = video_data.get("thumbnailUrl", "")
+
+    retrieval_summary = summarize_retrieval_confidence(chunks)
+
+    _log_rag_usage(
+        request,
+        event_type="lesson",
+        topic=build_lesson_query(payload.topic, payload.subject, payload.quarter, lesson_title=payload.lessonTitle),
+        subject=payload.subject,
+        quarter=payload.quarter,
+        chunks=chunks,
+    )
+
+    needs_review = parsed_lesson.get("needsReview", False)
+    if retrieval_summary.get("band") == "low":
+        needs_review = True
+
+    return {
+        **parsed_lesson,
+        "retrievalConfidence": retrieval_summary.get("confidence", 0.0),
+        "retrievalBand": retrieval_summary.get("band", "low"),
+        "retrievalMode": retrieval_mode,
+        "needsReview": needs_review,
+        "sources": [
+            {
+                "subject": row.get("subject"),
+                "quarter": row.get("quarter"),
+                "source_file": row.get("source_file"),
+                "storage_path": row.get("storage_path"),
+                "page": row.get("page"),
+                "score": row.get("score"),
+                "content_domain": row.get("content_domain"),
+                "chunk_type": row.get("chunk_type"),
+                "content": row.get("content"),
+            }
+            for row in chunks
+        ],
+        "activeModel": get_model_for_task("rag_lesson"),
+    }
+    except Exception as exc:
+        import traceback
+        logger.error(f"RAG lesson error: {type(exc).__name__}: {exc}\n{traceback.format_exc()}")
+        raise HTTPException(
+            status_code=500,
+            detail={
+                "error": type(exc).__name__,
+                "message": str(exc),
+                "traceback": traceback.format_exc(),
+            },
+        )
+
+
+@router.post("/generate-problem")
+async def rag_generate_problem(request: Request, payload: RagProblemRequest):
+    chunks = retrieve_curriculum_context(
+        query=payload.topic,
+        subject=payload.subject,
+        quarter=payload.quarter,
+        top_k=5,
+    )
+    prompt = build_problem_generation_prompt(payload.topic, payload.difficulty, chunks)
+    raw = await _generate_text(
+        prompt,
+        task_type="quiz_generation",
+        max_new_tokens=600,
+        enable_thinking=False,
+    )
+
+    parsed = _strip_thinking_and_parse(raw)
+
+    problem = str(parsed.get("problem") or raw)
+    if not problem or problem.startswith("{"):
+        problem = str(parsed.get("content") or str(parsed))
+    if len(problem) < 3 or problem.startswith("{"):
+        problem = raw
+    solution = str(parsed.get("solution") or "")
+    competency_ref = str(parsed.get("competencyReference") or "DepEd competency-aligned")
+
+    _log_rag_usage(
+        request,
+        event_type="problem_generation",
+        topic=payload.topic,
+        subject=payload.subject,
+        quarter=payload.quarter,
+        chunks=chunks,
+    )
+
+    return {
+        "problem": problem,
+        "solution": solution,
+        "competencyReference": competency_ref,
+        "sources": [
+            {
+                "subject": row.get("subject"),
+                "quarter": row.get("quarter"),
+                "source_file": row.get("source_file"),
+                "page": row.get("page"),
+                "score": row.get("score"),
+            }
+            for row in chunks
+        ],
+    }
+
+
+@router.post("/analysis-context")
+async def rag_analysis_context(request: Request, payload: RagAnalysisContextRequest):
+    if not payload.weakTopics:
+        raise HTTPException(status_code=400, detail="weakTopics must be a non-empty list")
+
+    chunks = build_analysis_curriculum_context(payload.weakTopics, payload.subject)
+    lines = ["LEARNING COMPETENCIES:"]
+    for index, row in enumerate(chunks, start=1):
+        lines.append(
+            f"{index}. {row.get('content')} (Source: {row.get('source_file')} p.{row.get('page')}, "
+            f"Q{row.get('quarter')}, {row.get('content_domain')})"
+        )
+
+    _log_rag_usage(
+        request,
+        event_type="analysis_context",
+        topic=", ".join(payload.weakTopics),
+        subject=payload.subject,
+        quarter=None,
+        chunks=chunks,
+    )
+
+    return {"curriculumContext": "\n".join(lines)}