SolusOps/Study-with-ChampAI

1from __future__ import annotations2import uuid3from typing import List4from services.model_router import ModelRouter5from services.ingestion import load_file6from storage.local_db import DB7from storage.mastery import MasteryStore8from agents.document_agent import DocumentAgent, DocumentConcepts9from agents.quest_agent import QuestAgent10from agents.quiz_agent import QuizAgent11from agents.tutor_agent import TutorAgent12from agents.progress_agent import ProgressAgent13from agents.speech_agent import SpeechAgent14from agents.language_agent import LanguageAgent15from quiz.models import Quest, QuizSession, Question16from quiz.scoring import compute_grade17from config.prompts import QUEST_AGENT_SYSTEM18 19class LearningOrchestrator:20    """21    Coordinates all specialist agents.22    app.py talks only to this class — all agent wiring is here.23    """24    def __init__(self, router: ModelRouter, db: DB, mastery_store: MasteryStore,25                 questions_per_quest: int = 3):26        self.document = DocumentAgent(router)27        self.quest = QuestAgent(router)28        self.quiz = QuizAgent(router)29        self.tutor = TutorAgent(router)30        self.progress = ProgressAgent(mastery_store)31        self.speech = SpeechAgent(router)32        self.language = LanguageAgent(router)33        self._router = router34        self._db = db35        self._mastery = mastery_store36        self._questions_per_quest = questions_per_quest37 38    def process_text(self, raw_text: str, language: str = "en") -> List[Quest]:39        concepts = self.document.extract(raw_text)40        return self._build_quests(concepts, language, source_text=raw_text)41 42    def process_image(self, image_b64: str, language: str = "en") -> List[Quest]:43        concepts = self.document.extract_from_image(image_b64)44        return self._build_quests(concepts, language, source_text=concepts.ocr_text)45 46    def process_voice(self, audio_bytes: bytes, language: str = "en") -> List[Quest]:47        text = self.speech.transcribe(audio_bytes)48        if not text:49            raise ValueError("Speech transcription returned empty text.")50        return self.process_text(text, language)51 52    def process_file(self, file_path: str, language: str = "en") -> List[Quest]:53        text, image_b64 = load_file(file_path)54        if image_b64 and not text:55            return self.process_image(image_b64, language)56        if image_b64 and text:57            return self.process_text(text, language) if text.strip() else self.process_image(image_b64, language)58        return self.process_text(text, language)59 60    def _build_quests(self, concepts: DocumentConcepts, language: str, source_text: str = "") -> List[Quest]:61        quests = self.quest.generate({"topics": concepts.topics, "definitions": concepts.definitions,62                                       "facts": concepts.facts, "formulae": concepts.formulae})63        for quest in quests:64            quest.questions = self.quiz.generate_for_quest(65                quest, source_text=source_text,66                questions_per_quest=self._questions_per_quest, language=language)67        return quests68 69    def generate_revision_quest(self, weak_topics: List[str], source_text: str = "") -> Quest:70        """71        Nemotron generates a targeted revision quest for weak topics.72        Adds to the user's quest queue — closes the learning loop.73        """74        topics_str = ", ".join(weak_topics)75        prompt = (f"Create a revision quest for these weak topics: {topics_str}\n"76                  f"Give it a dramatic RPG name like 'The Fallen Kingdom of {weak_topics[0]}'.\n"77                  f"Set boss_topic to the most fundamental topic that needs reviewing.\n"78                  f'Output: {{"quests":[{{"name":"string","topics":{weak_topics},"boss_topic":"string","difficulty":"medium"}}]}}')79        try:80            raw = self._router.reason(prompt, QUEST_AGENT_SYSTEM)81            from services.json_parser import extract_json82            data = extract_json(raw)83            q_data = data["quests"][0]84            quest = Quest(name=q_data["name"], topics=weak_topics,85                          boss_topic=q_data.get("boss_topic", weak_topics[-1]),86                          difficulty="medium")87        except Exception:88            quest = Quest(name=f"Revision: {weak_topics[0]}", topics=weak_topics,89                          boss_topic=weak_topics[-1], difficulty="medium")90        quest.questions = self.quiz.generate_for_quest(quest, source_text=source_text)91        return quest92 93    def get_tutor_hint(self, question: Question, student_answer: str) -> str:94        return self.tutor.hint(question=question.text, student_answer=student_answer,95                               correct_answer=question.correct_answer,96                               explanation=question.explanation,97                               source_excerpt=question.source_excerpt)98 99    def translate_hint(self, text: str, target_lang: str) -> str:100        return self.language.translate(text, target_lang)101 102    def complete_quest(self, session: QuizSession) -> dict:103        result = self.progress.update_from_session(session)104        self._db.save_session(str(uuid.uuid4()), session.quest_name,105                              session.score, len(session.questions),106                              session.xp_earned,107                              compute_grade(session.score, len(session.questions)))108        return result109