Spaces:
Sleeping
Sleeping
| import os | |
| import json | |
| import asyncio | |
| from datetime import datetime | |
| from typing import Dict, List, Optional, Tuple | |
| from dataclasses import dataclass, asdict | |
| import logging | |
| from dotenv import load_dotenv | |
| import gradio as gr | |
| from openai import AzureOpenAI | |
| import pandas as pd | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # Load environment variables | |
| # load_dotenv() | |
| # Azure OpenAI client configuration | |
| client = AzureOpenAI( | |
| azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT", "").replace('"', ''), | |
| api_key=os.getenv("AZURE_OPENAI_KEY", "").replace('"', ''), | |
| api_version=os.getenv("AZURE_OPENAI_API_VERSION", "2024-12-01-preview").replace('"', ''), | |
| ) | |
| # Model configurations | |
| LLM_DEPLOYMENT = os.getenv("AZURE_OPENAI_LLM_DEPLOYMENT", "gpt-4.1").replace('"', '') | |
| LLM_MODEL = os.getenv("AZURE_OPENAI_LLM_MODEL", "gpt-4.1").replace('"', '') | |
| class UserProgress: | |
| """Data class to track user learning progress""" | |
| user_id: str | |
| skill: str | |
| lessons_completed: int = 0 | |
| quiz_scores: List[float] = None | |
| current_difficulty: str = "beginner" | |
| streak_days: int = 0 | |
| total_time_spent: int = 0 # minutes | |
| last_activity: str = "" | |
| def __post_init__(self): | |
| if self.quiz_scores is None: | |
| self.quiz_scores = [] | |
| def get_average_score(self) -> float: | |
| """Calculate average quiz score""" | |
| return sum(self.quiz_scores) / len(self.quiz_scores) if self.quiz_scores else 0.0 | |
| def add_quiz_score(self, score: float): | |
| """Add a new quiz score and update difficulty if needed""" | |
| self.quiz_scores.append(score) | |
| # Adaptive difficulty adjustment | |
| avg_score = self.get_average_score() | |
| if avg_score >= 0.8 and len(self.quiz_scores) >= 3: | |
| if self.current_difficulty == "beginner": | |
| self.current_difficulty = "intermediate" | |
| elif self.current_difficulty == "intermediate": | |
| self.current_difficulty = "advanced" | |
| elif avg_score < 0.6 and len(self.quiz_scores) >= 3: | |
| if self.current_difficulty == "advanced": | |
| self.current_difficulty = "intermediate" | |
| elif self.current_difficulty == "intermediate": | |
| self.current_difficulty = "beginner" | |
| class Lesson: | |
| """Data class for lesson content""" | |
| title: str | |
| content: str | |
| skill: str | |
| difficulty: str | |
| duration_minutes: int | |
| key_concepts: List[str] | |
| class Quiz: | |
| """Data class for quiz content""" | |
| questions: List[Dict] | |
| skill: str | |
| difficulty: str | |
| lesson_title: str | |
| class LessonAgent: | |
| """Agent responsible for generating personalized micro-lessons""" | |
| def __init__(self, client: AzureOpenAI): | |
| self.client = client | |
| self.model = LLM_DEPLOYMENT | |
| async def generate_lesson(self, skill: str, difficulty: str = "beginner", | |
| previous_lessons: List[str] = None) -> Lesson: | |
| """Generate a personalized micro-lesson""" | |
| try: | |
| previous_context = "" | |
| if previous_lessons: | |
| previous_context = f"\nPrevious lessons covered: {', '.join(previous_lessons[-3:])}" | |
| prompt = f""" | |
| Create a concise, engaging micro-lesson for the skill: {skill} | |
| Difficulty level: {difficulty} | |
| {previous_context} | |
| Requirements: | |
| - Lesson should be 3-5 minutes to read | |
| - Include practical examples | |
| - Focus on one key concept | |
| - Make it actionable | |
| - Include 3-5 key takeaways | |
| Format your response as JSON with these fields: | |
| {{ | |
| "title": "Lesson title", | |
| "content": "Main lesson content (200-400 words)", | |
| "duration_minutes": 4, | |
| "key_concepts": ["concept1", "concept2", "concept3"] | |
| }} | |
| """ | |
| response = self.client.chat.completions.create( | |
| model=self.model, | |
| messages=[ | |
| {"role": "system", "content": "You are an expert educator creating micro-lessons. Always respond with valid JSON."}, | |
| {"role": "user", "content": prompt} | |
| ], | |
| temperature=0.7, | |
| max_tokens=1000 | |
| ) | |
| lesson_data = json.loads(response.choices[0].message.content) | |
| return Lesson( | |
| title=lesson_data["title"], | |
| content=lesson_data["content"], | |
| skill=skill, | |
| difficulty=difficulty, | |
| duration_minutes=lesson_data["duration_minutes"], | |
| key_concepts=lesson_data["key_concepts"] | |
| ) | |
| except Exception as e: | |
| logger.error(f"Error generating lesson: {e}") | |
| # Fallback lesson | |
| return Lesson( | |
| title=f"Introduction to {skill}", | |
| content=f"Let's start learning about {skill}. This is a fundamental skill that can help you grow professionally and personally.", | |
| skill=skill, | |
| difficulty=difficulty, | |
| duration_minutes=3, | |
| key_concepts=["basics", "fundamentals", "getting started"] | |
| ) | |
| class QuizAgent: | |
| """Agent responsible for generating adaptive quizzes""" | |
| def __init__(self, client: AzureOpenAI): | |
| self.client = client | |
| self.model = LLM_DEPLOYMENT | |
| async def generate_quiz(self, lesson: Lesson, user_progress: UserProgress) -> Quiz: | |
| """Generate an adaptive quiz based on the lesson content""" | |
| try: | |
| avg_score = user_progress.get_average_score() | |
| prompt = f""" | |
| Create a quiz for this lesson: | |
| Title: {lesson.title} | |
| Content: {lesson.content} | |
| Key concepts: {', '.join(lesson.key_concepts)} | |
| User's average score: {avg_score:.1f} | |
| Current difficulty: {lesson.difficulty} | |
| Create 3-5 questions that test understanding of the lesson. | |
| Mix question types: multiple choice, true/false, and short answer. | |
| Format as JSON: | |
| {{ | |
| "questions": [ | |
| {{ | |
| "type": "multiple_choice", | |
| "question": "Question text?", | |
| "options": ["A", "B", "C", "D"], | |
| "correct_answer": "A", | |
| "explanation": "Why this is correct" | |
| }}, | |
| {{ | |
| "type": "true_false", | |
| "question": "Statement to evaluate", | |
| "correct_answer": true, | |
| "explanation": "Explanation" | |
| }} | |
| ] | |
| }} | |
| """ | |
| response = self.client.chat.completions.create( | |
| model=self.model, | |
| messages=[ | |
| {"role": "system", "content": "You are a quiz expert. Always respond with valid JSON."}, | |
| {"role": "user", "content": prompt} | |
| ], | |
| temperature=0.7, | |
| max_tokens=1200 | |
| ) | |
| quiz_data = json.loads(response.choices[0].message.content) | |
| return Quiz( | |
| questions=quiz_data["questions"], | |
| skill=lesson.skill, | |
| difficulty=lesson.difficulty, | |
| lesson_title=lesson.title | |
| ) | |
| except Exception as e: | |
| logger.error(f"Error generating quiz: {e}") | |
| # Fallback quiz | |
| return Quiz( | |
| questions=[{ | |
| "type": "multiple_choice", | |
| "question": f"What is the main topic of this lesson about {lesson.skill}?", | |
| "options": [lesson.skill, "Something else", "Not sure", "All of the above"], | |
| "correct_answer": lesson.skill, | |
| "explanation": f"This lesson focuses on {lesson.skill}" | |
| }], | |
| skill=lesson.skill, | |
| difficulty=lesson.difficulty, | |
| lesson_title=lesson.title | |
| ) | |
| class ProgressAgent: | |
| """Agent responsible for tracking progress and making recommendations""" | |
| def __init__(self): | |
| self.user_data: Dict[str, UserProgress] = {} | |
| def get_user_progress(self, user_id: str, skill: str) -> UserProgress: | |
| """Get or create user progress tracking""" | |
| key = f"{user_id}_{skill}" | |
| if key not in self.user_data: | |
| self.user_data[key] = UserProgress(user_id=user_id, skill=skill) | |
| return self.user_data[key] | |
| def update_progress(self, user_id: str, skill: str, lesson_completed: bool = False, | |
| quiz_score: float = None) -> UserProgress: | |
| """Update user progress after lesson/quiz completion""" | |
| progress = self.get_user_progress(user_id, skill) | |
| if lesson_completed: | |
| progress.lessons_completed += 1 | |
| progress.last_activity = datetime.now().strftime("%Y-%m-%d %H:%M") | |
| if quiz_score is not None: | |
| progress.add_quiz_score(quiz_score) | |
| return progress | |
| def get_recommendation(self, progress: UserProgress) -> str: | |
| """Generate learning recommendations based on progress""" | |
| avg_score = progress.get_average_score() | |
| if progress.lessons_completed == 0: | |
| return "🎯 Ready to start your learning journey! Begin with your first lesson." | |
| elif avg_score >= 0.8: | |
| return f"🌟 Excellent work! You're mastering {progress.skill}. Ready for the next challenge?" | |
| elif avg_score >= 0.6: | |
| return f"📈 Good progress! Keep practicing {progress.skill} to build confidence." | |
| else: | |
| return f"💪 Don't give up! Review the concepts and try again. Practice makes perfect!" | |
| class AgenticSkillBuilder: | |
| """Main orchestrator for the agentic skill building platform""" | |
| def __init__(self): | |
| self.lesson_agent = LessonAgent(client) | |
| self.quiz_agent = QuizAgent(client) | |
| self.progress_agent = ProgressAgent() | |
| self.current_lesson: Optional[Lesson] = None | |
| self.current_quiz: Optional[Quiz] = None | |
| self.current_user = "demo_user" # In a real app, this would be from authentication | |
| # Predefined skills | |
| self.predefined_skills = [ | |
| "Python Programming", "Spanish Language", "Public Speaking", | |
| "Data Science", "Machine Learning", "JavaScript", "Project Management", | |
| "Digital Marketing", "Creative Writing", "Photography" | |
| ] | |
| async def start_lesson(self, skill: str) -> Tuple[str, str, str]: | |
| """Start a new lesson for the selected skill""" | |
| try: | |
| progress = self.progress_agent.get_user_progress(self.current_user, skill) | |
| # Get list of previous lesson titles for context | |
| previous_lessons = [] # In a real app, you'd store this | |
| self.current_lesson = await self.lesson_agent.generate_lesson( | |
| skill, progress.current_difficulty, previous_lessons | |
| ) | |
| lesson_content = f""" | |
| # 📚 {self.current_lesson.title} | |
| **Skill:** {self.current_lesson.skill} | **Level:** {self.current_lesson.difficulty.title()} | **Duration:** ~{self.current_lesson.duration_minutes} min | |
| {self.current_lesson.content} | |
| ### 🔑 Key Concepts: | |
| {chr(10).join([f"• {concept}" for concept in self.current_lesson.key_concepts])} | |
| """ | |
| return lesson_content, "✅ Complete Lesson", "" | |
| except Exception as e: | |
| logger.error(f"Error starting lesson: {e}") | |
| return f"❌ Error generating lesson: {str(e)}", "Try Again", "" | |
| async def complete_lesson_and_start_quiz(self) -> Tuple[str, str, str]: | |
| """Mark lesson as complete and start the quiz""" | |
| if not self.current_lesson: | |
| return "⚠️ No active lesson to complete.", "", "" | |
| try: | |
| # Update progress | |
| progress = self.progress_agent.update_progress( | |
| self.current_user, self.current_lesson.skill, lesson_completed=True | |
| ) | |
| # Generate quiz | |
| self.current_quiz = await self.quiz_agent.generate_quiz(self.current_lesson, progress) | |
| quiz_content = f""" | |
| # 🧠 Quiz: {self.current_lesson.title} | |
| Test your understanding of the lesson. Answer all questions to see your results! | |
| """ | |
| # Add questions to the content | |
| for i, q in enumerate(self.current_quiz.questions, 1): | |
| quiz_content += f"\n**Question {i}:** {q['question']}\n" | |
| if q['type'] == 'multiple_choice': | |
| quiz_content += f"Options: {', '.join(q['options'])}\n" | |
| elif q['type'] == 'true_false': | |
| quiz_content += "Answer: True or False\n" | |
| return quiz_content, "📝 Submit Quiz", "" | |
| except Exception as e: | |
| logger.error(f"Error generating quiz: {e}") | |
| return f"❌ Error generating quiz: {str(e)}", "", "" | |
| def submit_quiz(self, *answers) -> Tuple[str, str, str]: | |
| """Process quiz submission and show results""" | |
| if not self.current_quiz: | |
| return "⚠️ No active quiz to submit.", "", "" | |
| try: | |
| correct_answers = 0 | |
| total_questions = len(self.current_quiz.questions) | |
| results = [] | |
| for i, (question, answer) in enumerate(zip(self.current_quiz.questions, answers)): | |
| if answer is None or answer == "": | |
| continue | |
| is_correct = False | |
| correct_answer = question['correct_answer'] | |
| if question['type'] == 'multiple_choice': | |
| is_correct = answer.strip().upper() == str(correct_answer).upper() | |
| elif question['type'] == 'true_false': | |
| is_correct = answer.lower() == str(correct_answer).lower() | |
| if is_correct: | |
| correct_answers += 1 | |
| results.append({ | |
| 'question': question['question'], | |
| 'your_answer': answer, | |
| 'correct_answer': correct_answer, | |
| 'is_correct': is_correct, | |
| 'explanation': question.get('explanation', '') | |
| }) | |
| score = correct_answers / total_questions if total_questions > 0 else 0 | |
| # Update progress with quiz score | |
| progress = self.progress_agent.update_progress( | |
| self.current_user, self.current_lesson.skill, quiz_score=score | |
| ) | |
| # Generate results content | |
| results_content = f""" | |
| # 🎯 Quiz Results | |
| **Score:** {correct_answers}/{total_questions} ({score:.1%}) | |
| **Performance:** {'🌟 Excellent!' if score >= 0.8 else '👍 Good work!' if score >= 0.6 else '💪 Keep practicing!'} | |
| ### Detailed Results: | |
| """ | |
| for i, result in enumerate(results, 1): | |
| status = "✅" if result['is_correct'] else "❌" | |
| results_content += f""" | |
| **Q{i}:** {result['question']} | |
| {status} Your answer: {result['your_answer']} | |
| Correct answer: {result['correct_answer']} | |
| {result['explanation']} | |
| """ | |
| # Add progress and recommendations | |
| recommendation = self.progress_agent.get_recommendation(progress) | |
| results_content += f""" | |
| ### 📊 Your Progress | |
| - **Lessons completed:** {progress.lessons_completed} | |
| - **Average score:** {progress.get_average_score():.1%} | |
| - **Current level:** {progress.current_difficulty.title()} | |
| ### 🎯 Recommendation | |
| {recommendation} | |
| """ | |
| return results_content, "🔄 Start New Lesson", "" | |
| except Exception as e: | |
| logger.error(f"Error processing quiz: {e}") | |
| return f"❌ Error processing quiz: {str(e)}", "", "" | |
| # Initialize the main application | |
| app = AgenticSkillBuilder() | |
| def create_interface(): | |
| """Create the Gradio interface""" | |
| with gr.Blocks( | |
| title="SkillSprout", | |
| theme=gr.themes.Soft(), | |
| css=""" | |
| .gradio-container { | |
| max-width: 800px !important; | |
| margin: auto !important; | |
| } | |
| """ | |
| ) as demo: | |
| # Header | |
| gr.Markdown(""" | |
| # 🌱 SkillSprout | |
| ### AI-Powered Microlearning Platform | |
| Learn new skills through bite-sized lessons and adaptive quizzes powered by Azure OpenAI! | |
| """) | |
| # State variables | |
| current_skill = gr.State("") | |
| with gr.Tab("🎯 Start Learning"): | |
| gr.Markdown("### Choose a skill to begin your microlearning journey") | |
| with gr.Row(): | |
| with gr.Column(): | |
| skill_dropdown = gr.Dropdown( | |
| choices=app.predefined_skills, | |
| label="📚 Select a Skill", | |
| info="Choose from popular skills..." | |
| ) | |
| custom_skill = gr.Textbox( | |
| label="✍️ Or enter a custom skill", | |
| info="e.g., Cooking, Guitar, Time Management..." | |
| ) | |
| start_btn = gr.Button("🚀 Start Learning", variant="primary", size="lg") | |
| # Lesson content area | |
| lesson_output = gr.Markdown(visible=False) | |
| lesson_btn = gr.Button("Complete Lesson", visible=False) | |
| # Quiz area | |
| quiz_output = gr.Markdown(visible=False) | |
| # Dynamic quiz inputs (will be created based on quiz content) | |
| quiz_inputs = [] | |
| for i in range(5): # Max 5 questions | |
| quiz_inputs.append(gr.Textbox(label=f"Answer {i+1}", visible=False)) | |
| quiz_submit_btn = gr.Button("Submit Quiz", visible=False) | |
| # Results area | |
| results_output = gr.Markdown(visible=False) | |
| restart_btn = gr.Button("Start New Lesson", visible=False) | |
| with gr.Tab("📊 Progress Dashboard"): | |
| gr.Markdown("### Your Learning Analytics") | |
| progress_display = gr.Markdown(""" | |
| **Welcome to your progress dashboard!** | |
| Complete some lessons to see your learning analytics here. | |
| """) | |
| refresh_progress_btn = gr.Button("🔄 Refresh Progress") | |
| with gr.Tab("🔗 MCP Endpoints"): | |
| gr.Markdown(""" | |
| ### Model Context Protocol Integration | |
| This application exposes MCP endpoints for integration with external agents: | |
| - **GET /lesson/{skill}** - Fetch next lesson for a skill | |
| - **GET /progress/{user_id}** - Get user progress data | |
| - **POST /quiz/submit** - Submit quiz results | |
| *Coming soon: Full MCP server implementation* | |
| """) | |
| # Event handlers | |
| async def handle_start_learning(skill_choice, custom_skill_input): | |
| skill = custom_skill_input.strip() if custom_skill_input.strip() else skill_choice | |
| if not skill: | |
| return [ | |
| gr.update(value="⚠️ Please select or enter a skill to continue."), | |
| gr.update(visible=False), | |
| gr.update(visible=False), | |
| skill | |
| ] + [gr.update(visible=False, value="") for _ in range(5)] | |
| lesson_content, btn_text, _ = await app.start_lesson(skill) | |
| return [ | |
| gr.update(value=lesson_content), | |
| gr.update(value=btn_text, visible=True), | |
| gr.update(visible=False), | |
| skill | |
| ] + [gr.update(visible=False, value="") for _ in range(5)] | |
| async def handle_complete_lesson(): | |
| quiz_content, btn_text, _ = await app.complete_lesson_and_start_quiz() | |
| # Show quiz inputs based on number of questions | |
| quiz_updates = [] | |
| if app.current_quiz: | |
| for i, question in enumerate(app.current_quiz.questions): | |
| if i < len(quiz_inputs): | |
| label = f"Q{i+1}: {question['question'][:50]}..." | |
| quiz_updates.append(gr.update(label=label, visible=True)) | |
| else: | |
| quiz_updates.append(gr.update(visible=False)) | |
| # Hide remaining inputs | |
| for i in range(len(app.current_quiz.questions), len(quiz_inputs)): | |
| quiz_updates.append(gr.update(visible=False)) | |
| else: | |
| quiz_updates = [gr.update(visible=False) for _ in range(len(quiz_inputs))] | |
| return [ | |
| gr.update(visible=False), | |
| gr.update(value=quiz_content, visible=True), | |
| gr.update(value=btn_text, visible=True), | |
| gr.update(visible=False) | |
| ] + quiz_updates | |
| def handle_submit_quiz(*answers): | |
| # Filter out None values and empty strings | |
| valid_answers = [ans for ans in answers if ans is not None and ans != ""] | |
| results_content, btn_text, _ = app.submit_quiz(*valid_answers) | |
| return [ | |
| gr.update(visible=False), | |
| gr.update(value=results_content, visible=True), | |
| gr.update(value=btn_text, visible=True), | |
| gr.update(visible=False) | |
| ] + [gr.update(visible=False) for _ in range(len(quiz_inputs))] | |
| def handle_restart(): | |
| return [ | |
| gr.update(visible=False), | |
| gr.update(visible=False), | |
| gr.update(visible=False), | |
| gr.update(visible=False), | |
| gr.update(visible=False), | |
| "" | |
| ] + [gr.update(visible=False, value="") for _ in range(len(quiz_inputs))] | |
| def update_progress_display(): | |
| if not app.progress_agent.user_data: | |
| return "**No learning data yet.** Complete some lessons to see your progress!" | |
| progress_content = "# 📊 Your Learning Progress\n\n" | |
| for key, progress in app.progress_agent.user_data.items(): | |
| progress_content += f""" | |
| **Skill:** {progress.skill} | |
| - Lessons completed: {progress.lessons_completed} | |
| - Average quiz score: {progress.get_average_score():.1%} | |
| - Current difficulty: {progress.current_difficulty.title()} | |
| - Last activity: {progress.last_activity or 'Never'} | |
| """ | |
| return progress_content | |
| # Wire up the events | |
| start_btn.click( | |
| handle_start_learning, | |
| inputs=[skill_dropdown, custom_skill], | |
| outputs=[lesson_output, lesson_btn, quiz_output, current_skill] + quiz_inputs[:5] | |
| ) | |
| lesson_btn.click( | |
| handle_complete_lesson, | |
| outputs=[lesson_btn, quiz_output, quiz_submit_btn, results_output] + quiz_inputs | |
| ) | |
| quiz_submit_btn.click( | |
| handle_submit_quiz, | |
| inputs=quiz_inputs, | |
| outputs=[quiz_submit_btn, results_output, restart_btn, quiz_output] + quiz_inputs | |
| ) | |
| restart_btn.click( | |
| handle_restart, | |
| outputs=[lesson_output, quiz_output, results_output, lesson_btn, restart_btn, current_skill] + quiz_inputs | |
| ) | |
| refresh_progress_btn.click( | |
| update_progress_display, | |
| outputs=[progress_display] | |
| ) | |
| return demo | |
| def main(): | |
| """Main application entry point""" | |
| demo = create_interface() | |
| demo.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| share=False, | |
| show_error=True | |
| ) | |
| if __name__ == "__main__": | |
| main() | |