Update app.py

app.py

@@ -16,7 +16,7 @@ import secrets
 import string
 from huggingface_hub import HfApi, HfFolder
 import torch
-from transformers import AutoTokenizer, AutoModelForCausalLM
+from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
 import time
 import logging
 import asyncio
@@ -35,6 +35,7 @@ from dateutil.relativedelta import relativedelta
 import numpy as np
 import matplotlib.pyplot as plt
 from tqdm import tqdm
+import random
 
 # Enhanced Configuration
 PROFILES_DIR = "student_profiles"
@@ -61,7 +62,7 @@ logging.basicConfig(
 logger = logging.getLogger(__name__)
 
 # Model configuration
-MODEL_NAME = "deepseek-ai/deepseek-llm-7b"
+MODEL_NAME = "OpenAssistant/oasst-sft-4-pythia-12b-epoch-3.5"
 
 @lru_cache(maxsize=1)
 def get_model_and_tokenizer():
@@ -1472,374 +1473,188 @@ class EnhancedProfileManager:
 # Initialize profile manager
 profile_manager = EnhancedProfileManager()
 
-class EnhancedTeachingAssistant:
+class EducationalChatbot:
     def __init__(self):
-        self.context_history = []
-        self.max_context_length = MAX_CONTEXT_HISTORY
-        self.model, self.tokenizer = None, None
-        self.last_model_load_attempt = 0
-    
-    async def initialize_model(self):
-        if not self.model or not self.tokenizer:
-            if time.time() - self.last_model_load_attempt > 3600:
-                self.model, self.tokenizer = get_model_and_tokenizer()
-                self.last_model_load_attempt = time.time()
-    
-    async def generate_response(self, message: str, history: List[List[Union[str, None]]], session_token: str) -> str:
-        try:
-            await self.initialize_model()
-            
-            profile = profile_manager.load_profile(session_token=session_token)
-            if not profile:
-                return "Please complete and save your profile first to get personalized advice."
-            
-            self._update_context(message, history)
-            
-            student_name = profile.get('name', 'Student')
-            transcript = profile.get('transcript', {})
-            
-            if transcript.get('format') == 'progress_summary':
-                gpa = transcript.get('student_info', {}).get('weighted_gpa', None)
-                service_hours = transcript.get('student_info', {}).get('community_service_hours', 0)
-            else:
-                gpa = transcript.get('academic_summary', {}).get('gpa', {}).get('district', None)
-                service_hours = transcript.get('assessments', {}).get('community_service', {}).get('hours', 0)
-            
-            learning_style = re.search(r"Your primary learning style is\s*\*\*(.*?)\*\*", 
-                                      profile.get('learning_style', ''))
-            learning_style = learning_style.group(1) if learning_style else None
-            
-            context = f"You are an AI teaching assistant helping {student_name}. "
-            if gpa:
-                context += f"{student_name}'s current weighted GPA is {gpa}. "
-            if learning_style:
-                context += f"They are a {learning_style.lower()} learner. "
-            if service_hours:
-                context += f"They have completed {service_hours} community service hours. "
-            
-            if self.context_history:
-                context += "Recent conversation:\n"
-                for item in self.context_history[-self.max_context_length:]:
-                    role = "Student" if item['role'] == 'user' else "Assistant"
-                    context += f"{role}: {item['content']}\n"
-            
-            query_type = self._classify_query(message)
-            response = await self._generate_typed_response(query_type, message, context, profile)
-            
-            return response
-        
-        except Exception as e:
-            logger.error(f"Error generating response: {str(e)}")
-            return "I encountered an error processing your request. Please try again."
-    
-    def _classify_query(self, message: str) -> str:
-        message_lower = message.lower()
-        
-        if any(word in message_lower for word in ['gpa', 'grade', 'average']):
-            return 'gpa'
-        elif any(word in message_lower for word in ['study', 'learn', 'exam', 'test']):
-            return 'study'
-        elif any(word in message_lower for word in ['course', 'class', 'subject']):
-            return 'courses'
-        elif any(word in message_lower for word in ['college', 'university', 'apply']):
-            return 'college'
-        elif any(word in message_lower for word in ['plan', 'schedule', 'calendar']):
-            return 'planning'
-        elif any(word in message_lower for word in ['resource', 'book', 'website']):
-            return 'resources'
-        else:
-            return 'general'
-    
-    async def _generate_typed_response(self, query_type: str, message: str, context: str, profile: Dict) -> str:
-        if query_type == 'gpa':
-            return self._generate_gpa_response(profile)
-        elif query_type == 'study':
-            return self._generate_study_response(profile)
-        elif query_type == 'courses':
-            return self._generate_courses_response(profile)
-        elif query_type == 'college':
-            return self._generate_college_response(profile)
-        elif query_type == 'planning':
-            return self._generate_planning_response(profile)
-        elif query_type == 'resources':
-            return self._generate_resources_response(profile)
-        else:
-            return await self._generate_general_response(message, context)
-    
-    def _generate_gpa_response(self, profile: Dict) -> str:
-        transcript = profile.get('transcript', {})
-        analysis = academic_analyzer.analyze_gpa(transcript)
-        response = [
-            f"## 📊 GPA Analysis",
-            f"**Rating:** {analysis['rating']}",
-            f"{analysis['description']}",
-            f"{analysis['comparison']}",
-            "",
-            f"## 🎓 Graduation Status"
-        ]
-        
-        grad_status = academic_analyzer.analyze_graduation_status(transcript)
-        response.append(grad_status['status'])
-        response.append(f"**Completion:** {grad_status['completion_percentage']:.1f}%")
-        
-        if grad_status.get('missing_requirements'):
-            response.append("\n**Missing Requirements:**")
-            for req in grad_status['missing_requirements'][:3]:  # Show top 3 missing
-                if transcript.get('format') == 'progress_summary':
-                    response.append(f"- {req['code']}: {req['description']} ({req['remaining']} credits remaining)")
-                else:
-                    response.append(f"- {req['subject']}: {req['remaining']} credits remaining")
-        
-        response.append("\n## 🏫 College Recommendations")
-        college_recs = academic_analyzer.generate_college_recommendations(transcript)
-        
-        if college_recs['reach']:
-            response.append("\n**Reach Schools:**")
-            response.extend([f"- {school}" for school in college_recs['reach'][:3]])
-        
-        if college_recs['target']:
-            response.append("\n**Target Schools:**")
-            response.extend([f"- {school}" for school in college_recs['target'][:3]])
-        
-        if analysis.get('improvement_tips'):
-            response.append("\n**Improvement Tips:**")
-            response.extend([f"- {tip}" for tip in analysis['improvement_tips']])
-        
-        return "\n".join(response)
-    
-    def _generate_study_response(self, profile: Dict) -> str:
-        learning_style_match = re.search(r"Your primary learning style is\s*\*\*(.*?)\*\*", 
-                                       profile.get('learning_style', ''))
-        if not learning_style_match:
-            return "Please complete the learning style quiz first to get personalized study advice."
-        
-        learning_style = learning_style_match.group(1)
-        study_plan = profile.get('study_plan', {})
-        
-        response = [
-            f"As a **{learning_style}** learner, here are some study strategies for you:"
-        ]
-        
-        if study_plan.get('study_strategies'):
-            response.extend([f"- {strategy}" for strategy in study_plan['study_strategies']])
-        else:
-            if learning_style.lower() == 'visual':
-                response.extend([
-                    "- Use color coding in your notes",
-                    "- Create mind maps and diagrams",
-                    "- Watch educational videos to visualize concepts"
-                ])
-            elif learning_style.lower() == 'auditory':
-                response.extend([
-                    "- Record lectures and listen to them",
-                    "- Explain concepts out loud to yourself",
-                    "- Participate in study groups"
-                ])
-            elif learning_style.lower() == 'reading/writing':
-                response.extend([
-                    "- Write detailed summaries in your own words",
-                    "- Create question-answer sets for each topic",
-                    "- Rewrite your notes to reinforce learning"
-                ])
-            elif learning_style.lower() == 'kinesthetic':
-                response.extend([
-                    "- Use hands-on activities when possible",
-                    "- Study while walking or pacing",
-                    "- Create physical models to represent concepts"
-                ])
+        self.model_name = "OpenAssistant/oasst-sft-4-pythia-12b-epoch-3.5"
+        self.tokenizer = None
+        self.model = None
+        self.educational_topics = {
+            'math': ['algebra', 'calculus', 'geometry', 'trigonometry'],
+            'science': ['biology', 'chemistry', 'physics', 'astronomy'],
+            'humanities': ['history', 'literature', 'philosophy'],
+            'languages': ['english', 'spanish', 'french', 'grammar'],
+            'arts': ['music', 'art', 'drama'],
+            'technology': ['programming', 'computer science']
+        }
+        self.load_model()
+
+    def load_model(self):
+        """Load the HuggingFace model"""
+        self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
+        self.model = AutoModelForCausalLM.from_pretrained(
+            self.model_name,
+            torch_dtype=torch.float16,
+            device_map="auto"
+        )
+        logger.info("Educational chatbot model loaded")
+
+    def is_educational(self, question: str) -> bool:
+        """Check if question is educational"""
+        question_lower = question.lower()
+        for category, topics in self.educational_topics.items():
+            if any(topic in question_lower for topic in topics):
+                return True
+        return False
+
+    def generate_response(self, question: str, profile: Dict) -> Tuple[str, List[Dict]]:
+        """Generate a personalized educational response"""
+        if not self.is_educational(question):
+            return (
+                "I specialize in educational topics only. Please ask about subjects like math, "
+                "science, history, or literature. I can help with concepts, problem-solving methods, "
+                "and learning strategies.",
+                []
+            )
+
+        # Get learning style from profile
+        learning_style = self._get_learning_style(profile)
+        
+        # Generate base response using the model
+        prompt = self._build_prompt(question, profile)
+        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
+        
+        outputs = self.model.generate(
+            **inputs,
+            max_new_tokens=300,
+            temperature=0.7,
+            top_p=0.9,
+            repetition_penalty=1.2,
+            do_sample=True
+        )
         
-        if study_plan.get('time_management_tips'):
-            response.append("\n**Time Management Tips:**")
-            response.extend([f"- {tip}" for tip in study_plan['time_management_tips']])
+        raw_response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
         
-        return "\n".join(response)
-    
-    def _generate_courses_response(self, profile: Dict) -> str:
-        transcript = profile.get('transcript', {})
-        if not transcript.get('course_history'):
-            return "I couldn't find your course information. Please upload your transcript first."
+        # Process the response to be more pedagogical
+        processed_response = self._make_response_pedagogical(raw_response)
         
-        current_courses = [
-            course for course in transcript['course_history']
-            if (course.get('status', '').lower() == 'in progress' or 
-                (isinstance(course.get('credit_earned'), float) and course['credit_earned'] == 0))
-        ]
+        # Add multimedia based on learning style
+        multimedia = self._get_multimedia_suggestions(processed_response, learning_style)
         
-        completed_courses = [
-            course for course in transcript['course_history']
-            if (course.get('status', '').lower() == 'completed' or 
-                (isinstance(course.get('credit_earned'), float) and course['credit_earned'] > 0))
-        ]
-        
-        response = []
+        return processed_response, multimedia
+
+    def _get_learning_style(self, profile: Dict) -> str:
+        """Extract learning style from profile"""
+        if not profile or 'learning_style' not in profile:
+            return 'balanced'
+        
+        style_match = re.search(r"Your primary learning style is\s*\*\*(.*?)\*\*", 
+                              profile['learning_style'])
+        return style_match.group(1).lower() if style_match else 'balanced'
+
+    def _build_prompt(self, question: str, profile: Dict) -> str:
+        """Build a personalized prompt for the model"""
+        base_prompt = (
+            "You are an expert teaching assistant helping a student. Your role is to guide them "
+            "to discover answers themselves, not provide direct solutions. Use the Socratic method "
+            "by asking guiding questions and explaining concepts step-by-step.\n\n"
+        )
         
-        if current_courses:
-            response.append("**Your Current Courses:**")
-            for course in current_courses[:5]:
-                course_name = course.get('description') or course.get('course_title', 'Unknown')
-                response.append(
-                    f"- {course_name} "
-                    f"({course.get('course_code', '')})"
+        if profile:
+            # Add academic context if available
+            if 'transcript' in profile:
+                courses = [c['course_title'] for c in profile['transcript'].get('course_history', [])]
+                base_prompt += (
+                    f"The student has taken these courses: {', '.join(courses[:5])}. "
+                    "Consider their academic background when responding.\n\n"
+                )
+            
+            # Add learning style
+            learning_style = self._get_learning_style(profile)
+            if learning_style != 'balanced':
+                base_prompt += (
+                    f"The student is a {learning_style} learner. Adapt your teaching approach accordingly.\n\n"
                 )
-        else:
-            response.append("I couldn't find any current courses in your transcript.")
-        
-        if completed_courses:
-            response.append("\n**Recently Completed Courses:**")
-            for course in completed_courses[:5]:
-                course_name = course.get('description') or course.get('course_title', 'Unknown')
-                grade = course.get('grade_earned', '') or course.get('grade', '')
-                if grade:
-                    response.append(
-                        f"- {course_name} "
-                        f"(Grade: {grade})"
-                    )
-                else:
-                    response.append(f"- {course_name}")
-        
-        rigor = academic_analyzer.analyze_course_rigor(transcript)
-        if rigor['rating']:
-            response.append(f"\n**Course Rigor Analysis:** {rigor['rating']}")
-            if rigor['recommendations']:
-                response.append("\n**Recommendations:**")
-                response.extend([f"- {rec}" for rec in rigor['recommendations']])
-        
-        return "\n".join(response)
-    
-    def _generate_college_response(self, profile: Dict) -> str:
-        recommendations = academic_analyzer.generate_college_recommendations(profile.get('transcript', {}))
-        
-        response = ["**College Recommendations Based on Your Profile:**"]
-        
-        if recommendations['reach']:
-            response.append("\n**Reach Schools (Competitive):**")
-            response.extend([f"- {school}" for school in recommendations['reach'][:3]])
-        
-        if recommendations['target']:
-            response.append("\n**Target Schools (Good Match):**")
-            response.extend([f"- {school}" for school in recommendations['target'][:3]])
-        
-        if recommendations['safety']:
-            response.append("\n**Safety Schools (Likely Admission):**")
-            response.extend([f"- {school}" for school in recommendations['safety'][:3]])
-        
-        if recommendations['scholarships']:
-            response.append("\n**Scholarship Opportunities:**")
-            response.extend([f"- {scholarship}" for scholarship in recommendations['scholarships'][:3]])
-        
-        if recommendations['improvement_areas']:
-            response.append("\n**Areas to Improve for College Admissions:**")
-            response.extend([f"- {area}" for area in recommendations['improvement_areas']])
-        
-        return "\n".join(response)
-    
-    def _generate_planning_response(self, profile: Dict) -> str:
-        study_plan = profile.get('study_plan', {})
-        
-        response = ["**Study Planning Advice:**"]
-        
-        if study_plan.get('weekly_schedule'):
-            response.append("\nHere's a suggested weekly study schedule:")
-            for day, activities in study_plan['weekly_schedule'].items():
-                if activities:
-                    response.append(f"\n**{day}:**")
-                    for activity in activities[:2]:
-                        response.append(
-                            f"- {activity.get('course', 'Course')}: "
-                            f"{activity.get('duration', '45-60 minutes')}"
-                        )
-        else:
-            response.append("\nA good study schedule should include:")
-            response.append("- 45-60 minute study blocks with short breaks")
-            response.append("- Focus on 1-2 subjects per day")
-            response.append("- Regular review sessions")
-        
-        if study_plan.get('time_management_tips'):
-            response.append("\n**Time Management Tips:**")
-            response.extend([f"- {tip}" for tip in study_plan['time_management_tips'][:3]])
-        
-        return "\n".join(response)
-    
-    def _generate_resources_response(self, profile: Dict) -> str:
-        study_plan = profile.get('study_plan', {})
-        transcript = profile.get('transcript', {})
-        
-        response = ["**Recommended Learning Resources:**"]
         
-        if study_plan.get('resource_recommendations'):
-            response.extend([f"- {resource}" for resource in study_plan['resource_recommendations'][:3]])
-        else:
-            response.extend([
-                "- Khan Academy for math and science",
-                "- Quizlet for flashcards",
-                "- Wolfram Alpha for math help"
-            ])
+        base_prompt += (
+            f"Student Question: {question}\n\n"
+            "Teaching Assistant Response:\n"
+            "1. First, let's understand the key concepts involved...\n"
+            "2. What do you think would be the first step in solving this?\n"
+            "3. Consider this approach...\n"
+            "4. Here's how we might break this down...\n"
+            "Remember, the goal is understanding, not just the answer."
+        )
         
-        current_courses = [
-            course for course in transcript.get('course_history', [])
-            if (course.get('status', '').lower() == 'in progress' or 
-                (isinstance(course.get('credit_earned'), float) and course['credit_earned'] == 0))
+        return base_prompt
+
+    def _make_response_pedagogical(self, response: str) -> str:
+        """Process the raw response to be more teaching-oriented"""
+        # Remove direct answers if present
+        response = re.sub(r"(the answer is|it is|direct solution:) .*?(\n|$)", "", response, flags=re.I)
+        
+        # Add more guiding language
+        guiding_phrases = [
+            "What do you think about...",
+            "Have you considered...",
+            "Let's break this down...",
+            "One approach might be...",
+            "Think about how you would...",
+            "What steps would you take to..."
         ]
         
-        if current_courses:
-            response.append("\n**Course-Specific Resources:**")
-            for course in current_courses[:2]:
-                course_name = course.get('description') or course.get('course_title', 'your course')
-                if 'MATH' in course_name.upper():
-                    response.append(f"- For {course_name}: Desmos Graphing Calculator, Art of Problem Solving")
-                elif 'SCIENCE' in course_name.upper():
-                    response.append(f"- For {course_name}: PhET Simulations, Crash Course Science videos")
-                elif 'HISTORY' in course_name.upper():
-                    response.append(f"- For {course_name}: Crash Course History videos, Library of Congress resources")
+        # Ensure response has at least 2 guiding questions
+        if sum(1 for phrase in guiding_phrases if phrase.lower() in response.lower()) < 2:
+            response += "\n\n" + "\n".join(guiding_phrases[:2])
         
-        return "\n".join(response)
-    
-    async def _generate_general_response(self, message: str, context: str) -> str:
-        if not self.model or not self.tokenizer:
-            return "I'm still loading my knowledge base. Please try again in a moment."
+        return response
+
+    def _get_multimedia_suggestions(self, response: str, learning_style: str) -> List[Dict]:
+        """Generate multimedia suggestions based on learning style and content"""
+        suggestions = []
+        
+        # Common educational platforms
+        resources = {
+            'visual': [
+                {"type": "video", "source": "Khan Academy", "url": "https://www.khanacademy.org"},
+                {"type": "diagram", "source": "Math is Fun", "url": "https://www.mathsisfun.com"},
+                {"type": "infographic", "source": "InfoGram", "url": "https://infogram.com"}
+            ],
+            'auditory': [
+                {"type": "podcast", "source": "Stuff You Should Know", "url": "https://www.iheart.com/podcast/stuff-you-should-know-26940277"},
+                {"type": "audio_lecture", "source": "The Great Courses", "url": "https://www.thegreatcourses.com"}
+            ],
+            'reading/writing': [
+                {"type": "article", "source": "Britannica", "url": "https://www.britannica.com"},
+                {"type": "textbook", "source": "OpenStax", "url": "https://openstax.org"}
+            ],
+            'kinesthetic': [
+                {"type": "interactive", "source": "PhET Simulations", "url": "https://phet.colorado.edu"},
+                {"type": "hands-on", "source": "Science Buddies", "url": "https://www.sciencebuddies.org"}
+            ]
+        }
         
-        try:
-            prompt = f"{context}\nStudent: {message}\nAssistant:"
-            
-            inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
-            
-            outputs = self.model.generate(
-                **inputs,
-                max_new_tokens=200,
-                temperature=0.7,
-                top_p=0.9,
-                repetition_penalty=1.1,
-                do_sample=True
-            )
-            
-            response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
-            
-            response = response[len(prompt):].strip()
-            
-            if response and response[-1] not in {'.', '!', '?'}:
-                last_period = response.rfind('.')
-                if last_period > 0:
-                    response = response[:last_period + 1]
-            
-            return response if response else "I'm not sure how to respond to that. Could you rephrase your question?"
-        except Exception as e:
-            logger.error(f"Model generation error: {str(e)}")
-            return "I encountered an error generating a response. Please try again."
-    
-    def _update_context(self, message: str, history: List[List[Union[str, None]]]) -> None:
-        self.context_history.append({"role": "user", "content": message})
+        # Add general suggestions based on learning style
+        if learning_style in resources:
+            suggestions.extend(resources[learning_style][:2])
         
-        if history:
-            for h in history[-self.max_context_length:]:
-                if h[0]:
-                    self.context_history.append({"role": "user", "content": h[0]})
-                if h[1]:
-                    self.context_history.append({"role": "assistant", "content": h[1]})
+        # Add specific content based on response
+        if "math" in response.lower():
+            suggestions.append({
+                "type": "practice_problems",
+                "source": "Art of Problem Solving",
+                "url": "https://artofproblemsolving.com"
+            })
+        elif "science" in response.lower():
+            suggestions.append({
+                "type": "experiment",
+                "source": "Science Journal",
+                "url": "https://sciencejournal.withgoogle.com"
+            })
         
-        self.context_history = self.context_history[-(self.max_context_length * 2):]
+        return suggestions
 
-# Initialize teaching assistant
-teaching_assistant = EnhancedTeachingAssistant()
+# Initialize the chatbot
+educational_chatbot = EducationalChatbot()
 
 class StudyCalendar:
     def __init__(self):
@@ -2574,20 +2389,32 @@ def create_enhanced_interface():
                     ]
                 )
 
-            with gr.Tab("AI Assistant", id=4):
-                gr.Markdown("## 💬 Your Personalized Learning Assistant")
-                gr.Markdown("Ask me anything about studying, your courses, grades, or learning strategies.")
+            with gr.Tab("AI Teaching Assistant", id=4):
+                gr.Markdown("## 💬 Your Personalized Teaching Assistant")
+                gr.Markdown("Ask educational questions about any subject. I'll guide you to discover the answers yourself.")
                 
                 chatbot = gr.Chatbot(height=500)
-                msg = gr.Textbox(label="Your Message")
-                clear = gr.Button("Clear")
+                msg = gr.Textbox(label="Your Educational Question")
+                clear = gr.Button("Clear Chat")
 
-                def respond(message, chat_history):
-                    bot_message = teaching_assistant.generate_response(message, chat_history, session_token.value)
-                    chat_history.append((message, bot_message))
+                def respond(message: str, chat_history: List, profile: Dict) -> Tuple[str, List]:
+                    """Handle chat responses with multimedia"""
+                    response, multimedia = educational_chatbot.generate_response(message, profile)
+                    
+                    # Format multimedia suggestions
+                    if multimedia:
+                        response += "\n\n**Suggested Resources:**\n"
+                        for item in multimedia:
+                            response += f"- [{item['type'].title()}] {item['source']}: {item['url']}\n"
+                    
+                    chat_history.append((message, response))
                     return "", chat_history
 
-                msg.submit(respond, [msg, chatbot], [msg, chatbot])
+                msg.submit(
+                    respond,
+                    inputs=[msg, chatbot, gr.State(profile_manager.load_profile(session_token.value))],
+                    outputs=[msg, chatbot]
+                )
                 clear.click(lambda: None, None, chatbot, queue=False)
 
             with gr.Tab("Goals & Planning", id=5):