Spaces:
Running
Running
| from __future__ import annotations | |
| import uuid | |
| from typing import List | |
| from services.model_router import ModelRouter | |
| from services.ingestion import load_file | |
| from storage.local_db import DB | |
| from storage.mastery import MasteryStore | |
| from agents.document_agent import DocumentAgent, DocumentConcepts | |
| from agents.quest_agent import QuestAgent | |
| from agents.quiz_agent import QuizAgent | |
| from agents.tutor_agent import TutorAgent | |
| from agents.progress_agent import ProgressAgent | |
| from agents.speech_agent import SpeechAgent | |
| from agents.language_agent import LanguageAgent | |
| from quiz.models import Quest, QuizSession, Question | |
| from quiz.scoring import compute_grade | |
| from config.prompts import QUEST_AGENT_SYSTEM | |
| class LearningOrchestrator: | |
| """ | |
| Coordinates all specialist agents. | |
| app.py talks only to this class — all agent wiring is here. | |
| """ | |
| def __init__(self, router: ModelRouter, db: DB, mastery_store: MasteryStore, | |
| questions_per_quest: int = 3): | |
| self.document = DocumentAgent(router) | |
| self.quest = QuestAgent(router) | |
| self.quiz = QuizAgent(router) | |
| self.tutor = TutorAgent(router) | |
| self.progress = ProgressAgent(mastery_store) | |
| self.speech = SpeechAgent(router) | |
| self.language = LanguageAgent(router) | |
| self._router = router | |
| self._db = db | |
| self._mastery = mastery_store | |
| self._questions_per_quest = questions_per_quest | |
| def process_text(self, raw_text: str, language: str = "en") -> List[Quest]: | |
| concepts = self.document.extract(raw_text) | |
| return self._build_quests(concepts, language, source_text=raw_text) | |
| def process_image(self, image_b64: str, language: str = "en") -> List[Quest]: | |
| concepts = self.document.extract_from_image(image_b64) | |
| return self._build_quests(concepts, language, source_text=concepts.ocr_text) | |
| def process_voice(self, audio_bytes: bytes, language: str = "en") -> List[Quest]: | |
| text = self.speech.transcribe(audio_bytes) | |
| if not text: | |
| raise ValueError("Speech transcription returned empty text.") | |
| return self.process_text(text, language) | |
| def process_file(self, file_path: str, language: str = "en") -> List[Quest]: | |
| text, image_b64 = load_file(file_path) | |
| if image_b64 and not text: | |
| return self.process_image(image_b64, language) | |
| if image_b64 and text: | |
| return self.process_text(text, language) if text.strip() else self.process_image(image_b64, language) | |
| return self.process_text(text, language) | |
| def _build_quests(self, concepts: DocumentConcepts, language: str, source_text: str = "") -> List[Quest]: | |
| quests = self.quest.generate({"topics": concepts.topics, "definitions": concepts.definitions, | |
| "facts": concepts.facts, "formulae": concepts.formulae}) | |
| for quest in quests: | |
| quest.questions = self.quiz.generate_for_quest( | |
| quest, source_text=source_text, | |
| questions_per_quest=self._questions_per_quest, language=language) | |
| return quests | |
| def generate_revision_quest(self, weak_topics: List[str], source_text: str = "") -> Quest: | |
| """ | |
| Nemotron generates a targeted revision quest for weak topics. | |
| Adds to the user's quest queue — closes the learning loop. | |
| """ | |
| topics_str = ", ".join(weak_topics) | |
| prompt = (f"Create a revision quest for these weak topics: {topics_str}\n" | |
| f"Give it a dramatic RPG name like 'The Fallen Kingdom of {weak_topics[0]}'.\n" | |
| f"Set boss_topic to the most fundamental topic that needs reviewing.\n" | |
| f'Output: {{"quests":[{{"name":"string","topics":{weak_topics},"boss_topic":"string","difficulty":"medium"}}]}}') | |
| try: | |
| raw = self._router.reason(prompt, QUEST_AGENT_SYSTEM) | |
| from services.json_parser import extract_json | |
| data = extract_json(raw) | |
| q_data = data["quests"][0] | |
| quest = Quest(name=q_data["name"], topics=weak_topics, | |
| boss_topic=q_data.get("boss_topic", weak_topics[-1]), | |
| difficulty="medium") | |
| except Exception: | |
| quest = Quest(name=f"Revision: {weak_topics[0]}", topics=weak_topics, | |
| boss_topic=weak_topics[-1], difficulty="medium") | |
| quest.questions = self.quiz.generate_for_quest(quest, source_text=source_text) | |
| return quest | |
| def get_tutor_hint(self, question: Question, student_answer: str) -> str: | |
| return self.tutor.hint(question=question.text, student_answer=student_answer, | |
| correct_answer=question.correct_answer, | |
| explanation=question.explanation, | |
| source_excerpt=question.source_excerpt) | |
| def translate_hint(self, text: str, target_lang: str) -> str: | |
| return self.language.translate(text, target_lang) | |
| def complete_quest(self, session: QuizSession) -> dict: | |
| result = self.progress.update_from_session(session) | |
| self._db.save_session(str(uuid.uuid4()), session.quest_name, | |
| session.score, len(session.questions), | |
| session.xp_earned, | |
| compute_grade(session.score, len(session.questions))) | |
| return result | |