🤖 AI-Powered Personal Learning Assistant
Version 2.0 - Clean Production Build
Multi-Agent Reasoning • Voice AI • Real-Time Data • ZeroGPU Optimized
#!/usr/bin/env python3 """ AI-Powered Personal Learning Assistant Version 2.0 - Clean Production Build for HuggingFace Spaces Gradio Agents & MCP Hackathon 2025 Features: - Multi-agent AI reasoning with smolagents - Voice AI processing with SambaNova Cloud - Real-time data integration via MCP - ZeroGPU optimized for HuggingFace Spaces """ import os import gc import sys import logging import sqlite3 import tempfile import traceback from pathlib import Path from typing import Dict, List, Tuple, Optional, Any from datetime import datetime, timedelta # Core dependencies import gradio as gr import requests import plotly.graph_objects as go import plotly.express as px from dotenv import load_dotenv # Audio processing try: import speech_recognition as sr import pydub import soundfile as sf AUDIO_AVAILABLE = True except ImportError as e: AUDIO_AVAILABLE = False print(f"⚠️ Audio libraries not available: {e}") # AI and ML dependencies with graceful fallbacks try: import transformers TRANSFORMERS_AVAILABLE = True except ImportError: TRANSFORMERS_AVAILABLE = False print("⚠️ Transformers not available") try: import smolagents from smolagents import CodeAgent, ReactCodeAgent, tool, HfApiModel SMOLAGENTS_AVAILABLE = True except ImportError: SMOLAGENTS_AVAILABLE = False print("⚠️ Smolagents not available - using fallback mode") # Define tool decorator fallback BEFORE using it if not SMOLAGENTS_AVAILABLE: def tool(func): """Fallback tool decorator when smolagents is not available""" func._is_tool = True return func else: # Import tool from smolagents if available pass # tool is already imported above # HuggingFace Spaces support try: import spaces SPACES_AVAILABLE = True except ImportError: SPACES_AVAILABLE = False # Mock spaces decorator for local development class spaces: @staticmethod def GPU(func): return func # Environment setup load_dotenv() logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Global configuration SAMBANOVA_API_KEY = os.getenv("SAMBANOVA_API_KEY") HF_TOKEN = os.getenv("HF_TOKEN") SAMBANOVA_AVAILABLE = bool(SAMBANOVA_API_KEY) # ============================================================================ # Database Layer - Learning Progress Tracking # ============================================================================ class LearningDatabase: """SQLite database for tracking learning progress and user profiles""" def __init__(self, db_path: str = "learning_assistant.db"): self.db_path = db_path self.init_database() def init_database(self): """Initialize database tables""" try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() # User profiles table cursor.execute(""" CREATE TABLE IF NOT EXISTS user_profiles ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, learning_style TEXT, goals TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Learning sessions table cursor.execute(""" CREATE TABLE IF NOT EXISTS learning_sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, subject TEXT NOT NULL, level TEXT, session_data TEXT, progress_score REAL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES user_profiles (id) ) """) # Progress tracking table cursor.execute(""" CREATE TABLE IF NOT EXISTS progress_tracking ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, subject TEXT, skill TEXT, mastery_level REAL, last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES user_profiles (id) ) """) conn.commit() logger.info("✅ Database initialized successfully") except Exception as e: logger.error(f"❌ Database initialization failed: {e}") # ============================================================================ # Multi-Agent AI System with Smolagents # ============================================================================ class LearningAgents: """Multi-agent system using smolagents for advanced reasoning""" def __init__(self): self.agents_available = SMOLAGENTS_AVAILABLE self.setup_agents() def setup_agents(self): """Initialize smolagents-based multi-agent system""" if not self.agents_available: logger.warning("⚠️ Smolagents not available, using fallback agents") self.setup_fallback_agents() return try: # Initialize smolagents with proper configuration from smolagents import HfApiModel # Use HuggingFace models for reasoning model = HfApiModel("microsoft/DialoGPT-medium") # Create specialized agents self.curriculum_agent = self.create_curriculum_agent(model) self.content_agent = self.create_content_agent(model) self.assessment_agent = self.create_assessment_agent(model) logger.info("✅ Smolagents multi-agent system initialized") except Exception as e: logger.error(f"❌ Smolagents setup failed: {e}") self.agents_available = False self.setup_fallback_agents() def create_curriculum_agent(self, model): """Create curriculum planning agent with smolagents""" if not self.agents_available: return self.create_fallback_agent("curriculum") try: agent = CodeAgent( tools=[self.curriculum_planning_tool], model=model, max_iterations=3 ) return agent except Exception as e: logger.error(f"Curriculum agent creation failed: {e}") return self.create_fallback_agent("curriculum") def create_content_agent(self, model): """Create content generation agent with smolagents""" if not self.agents_available: return self.create_fallback_agent("content") try: agent = ReactCodeAgent( tools=[self.content_generation_tool], model=model, max_iterations=2 ) return agent except Exception as e: logger.error(f"Content agent creation failed: {e}") return self.create_fallback_agent("content") def create_assessment_agent(self, model): """Create assessment agent with smolagents""" if not self.agents_available: return self.create_fallback_agent("assessment") try: agent = CodeAgent( tools=[self.assessment_generation_tool], model=model, max_iterations=2 ) return agent except Exception as e: logger.error(f"Assessment agent creation failed: {e}") return self.create_fallback_agent("assessment") def setup_fallback_agents(self): """Setup fallback agents when smolagents is not available""" self.curriculum_agent = self.create_fallback_agent("curriculum") self.content_agent = self.create_fallback_agent("content") self.assessment_agent = self.create_fallback_agent("assessment") def create_fallback_agent(self, agent_type: str): """Create fallback agent for when smolagents is unavailable""" class FallbackAgent: def __init__(self, agent_type): self.agent_type = agent_type def run(self, prompt): return f"🔄 **Fallback {self.agent_type.title()} Agent**\n\n{self.generate_fallback_response(prompt)}" def generate_fallback_response(self, prompt): if self.agent_type == "curriculum": return self.generate_curriculum_fallback(prompt) elif self.agent_type == "content": return self.generate_content_fallback(prompt) elif self.agent_type == "assessment": return self.generate_assessment_fallback(prompt) else: return "This feature requires smolagents. Please install with: pip install smolagents" return FallbackAgent(agent_type) # Tool methods - decorated only when smolagents is available def curriculum_planning_tool(self, subject: str, level: str, goals: str) -> str: """Curriculum planning tool for smolagents""" return f""" # 📚 AI-Generated Curriculum: {subject} ## 🎯 Learning Path for {level.title()} Level ### Phase 1: Foundation Building - Core concepts and terminology - Essential prerequisites review - Hands-on introduction exercises ### Phase 2: Skill Development - Practical application projects - Guided practice sessions - Real-world case studies ### Phase 3: Advanced Application - Complex problem solving - Integration with other topics - Portfolio development ### 📈 Progress Milestones 1. **Week 1-2**: Foundation mastery 2. **Week 3-4**: Practical application 3. **Week 5-6**: Advanced projects **Personalized for:** {goals} """ def content_generation_tool(self, topic: str, difficulty: str) -> str: """Content generation tool for smolagents""" return f""" # 📖 Learning Content: {topic} ## 🔍 Overview This {difficulty}-level content covers essential concepts in {topic}. ## 🎯 Key Learning Objectives - Understand fundamental principles - Apply concepts to real scenarios - Develop practical skills ## 📚 Content Structure 1. **Introduction & Context** 2. **Core Concepts** 3. **Practical Examples** 4. **Hands-on Exercises** 5. **Assessment & Review** ## 🚀 Next Steps Continue with advanced topics or apply skills in projects. """ def assessment_generation_tool(self, topic: str, num_questions: int = 5) -> Dict: """Assessment generation tool for smolagents""" return { "quiz_title": f"{topic} Assessment", "questions": [ { "question": f"What is the main concept of {topic}?", "options": ["Option A", "Option B", "Option C", "Option D"], "correct": 0 } for i in range(num_questions) ], "difficulty": "intermediate", "estimated_time": f"{num_questions * 2} minutes" } # Apply @tool decorator if smolagents is available if SMOLAGENTS_AVAILABLE: LearningAgents.curriculum_planning_tool = tool(LearningAgents.curriculum_planning_tool) LearningAgents.content_generation_tool = tool(LearningAgents.content_generation_tool) LearningAgents.assessment_generation_tool = tool(LearningAgents.assessment_generation_tool) # ============================================================================ # SambaNova Audio AI Integration # ============================================================================ class SambaNovaAudioAI: """SambaNova Cloud integration for Qwen2-Audio-7B-Instruct processing""" def __init__(self): self.api_key = SAMBANOVA_API_KEY self.available = bool(self.api_key) and AUDIO_AVAILABLE self.base_url = "https://api.sambanova.ai/v1" if not self.available: logger.warning("⚠️ SambaNova Audio AI not available") def process_audio_with_qwen(self, audio_path: str, prompt: str = None) -> Dict: """Process audio with Qwen2-Audio-7B-Instruct model""" if not self.available: return { "error": "SambaNova Audio AI not available", "message": "Please set SAMBANOVA_API_KEY environment variable" } try: # Convert audio to required format audio_data = self.prepare_audio(audio_path) # Prepare request for SambaNova API headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } payload = { "model": "Qwen2-Audio-7B-Instruct", "messages": [ { "role": "user", "content": [ {"type": "text", "text": prompt or "Analyze this audio and provide educational insights"}, {"type": "audio", "audio": audio_data} ] } ], "max_tokens": 1000, "temperature": 0.7 } response = requests.post( f"{self.base_url}/chat/completions", headers=headers, json=payload, timeout=30 ) if response.status_code == 200: return response.json() else: return { "error": f"API request failed with status {response.status_code}", "details": response.text } except Exception as e: logger.error(f"SambaNova API error: {e}") return { "error": "Audio processing failed", "details": str(e) } def prepare_audio(self, audio_path: str) -> str: """Prepare audio for SambaNova API""" try: # Convert to WAV format if needed if not audio_path.endswith('.wav'): audio = pydub.AudioSegment.from_file(audio_path) wav_path = audio_path.replace(Path(audio_path).suffix, '.wav') audio.export(wav_path, format="wav") audio_path = wav_path # Read and encode audio import base64 with open(audio_path, 'rb') as f: audio_data = base64.b64encode(f.read()).decode('utf-8') return audio_data except Exception as e: logger.error(f"Audio preparation failed: {e}") raise def generate_learning_plan_from_audio(self, audio_path: str) -> str: """Generate learning plan from audio input""" prompt = """ Listen to this audio and create a comprehensive learning plan. Include: 1. Identified learning goals from the audio 2. Recommended curriculum structure 3. Timeline and milestones 4. Resources and next steps """ result = self.process_audio_with_qwen(audio_path, prompt) if "error" in result: return f"❌ **Audio Processing Error**: {result['error']}" try: content = result['choices'][0]['message']['content'] return f""" # 🎤 Audio-Generated Learning Plan {content} --- *Generated by Qwen2-Audio-7B-Instruct via SambaNova Cloud* """ except (KeyError, IndexError) as e: return f"❌ **Response Processing Error**: {e}" def answer_audio_question(self, audio_path: str) -> str: """Answer questions from audio input""" prompt = """ Listen to this audio question and provide a comprehensive educational answer. Structure your response with: 1. Clear explanation of the concept 2. Practical examples 3. Additional learning resources 4. Related topics to explore """ result = self.process_audio_with_qwen(audio_path, prompt) if "error" in result: return f"❌ **Audio Processing Error**: {result['error']}" try: content = result['choices'][0]['message']['content'] return f""" # 🎤 Audio Q&A Response {content} --- *Powered by Qwen2-Audio-7B-Instruct* """ except (KeyError, IndexError) as e: return f"❌ **Response Processing Error**: {e}" def convert_speech_to_text(self, audio_path: str) -> str: """Convert speech to text using local speech recognition""" if not AUDIO_AVAILABLE: return "❌ Speech recognition not available" try: recognizer = sr.Recognizer() with sr.AudioFile(audio_path) as source: audio_data = recognizer.record(source) text = recognizer.recognize_google(audio_data) return f"**Transcribed Text**: {text}" except Exception as e: return f"❌ **Speech Recognition Error**: {e}" # ============================================================================ # Main Learning Assistant Class # ============================================================================ class LearningAssistant: """Main learning assistant orchestrating all components""" def __init__(self): self.db = LearningDatabase() self.agents = LearningAgents() self.audio_ai = SambaNovaAudioAI() logger.info("✅ Learning Assistant initialized") def generate_curriculum_with_multistep_reasoning(self, subject: str, level: str, goals: str) -> str: """Generate curriculum using multi-step AI reasoning""" try: if self.agents.agents_available: # Use smolagents for advanced reasoning prompt = f""" Create a comprehensive curriculum for: Subject: {subject} Level: {level} Goals: {goals} Use multi-step reasoning to analyze prerequisites, create learning phases, and establish milestones. """ result = self.agents.curriculum_agent.run(prompt) return result else: # Fallback curriculum generation return self.generate_fallback_curriculum(subject, level, goals) except Exception as e: logger.error(f"Curriculum generation error: {e}") return f"❌ **Error**: {e}\n\nPlease try again or use the fallback interface." def generate_fallback_curriculum(self, subject: str, level: str, goals: str) -> str: """Fallback curriculum generation when agents are unavailable""" return f""" # 📚 Learning Curriculum: {subject} ## 🎯 Customized for {level.title()} Level ### 📋 Learning Goals {goals} ### 🗓️ Structured Learning Path #### Phase 1: Foundation (Weeks 1-2) - **Core Concepts**: Introduction to {subject} fundamentals - **Prerequisites**: Review essential background knowledge - **Initial Projects**: Hands-on practice exercises #### Phase 2: Development (Weeks 3-4) - **Skill Building**: Intermediate concepts and techniques - **Practical Applications**: Real-world project work - **Problem Solving**: Guided challenges and exercises #### Phase 3: Mastery (Weeks 5-6) - **Advanced Topics**: Complex applications and integrations - **Portfolio Development**: Showcase projects - **Knowledge Integration**: Connecting concepts across domains ### 📈 Progress Tracking - **Weekly Assessments**: Track understanding and skill development - **Milestone Projects**: Demonstrate cumulative learning - **Peer Reviews**: Collaborative learning opportunities ### 🔗 Recommended Resources - Online courses and tutorials - Practice platforms and tools - Community forums and support groups ### 🎯 Next Steps Continue to advanced topics or apply skills in specialized areas. --- *Generated by AI Learning Assistant - Fallback Mode* """ def process_audio_learning_request(self, audio_input) -> str: """Process audio input for learning plan generation""" if not audio_input: return "❌ **Error**: No audio provided" try: # Save audio from Gradio input audio_path = self.save_gradio_audio(audio_input) # Process with SambaNova result = self.audio_ai.generate_learning_plan_from_audio(audio_path) # Cleanup temp file self.cleanup_temp_file(audio_path) return result except Exception as e: logger.error(f"Audio processing error: {e}") return f"❌ **Audio Processing Failed**: {e}" def answer_audio_question(self, audio_input) -> str: """Answer questions from audio input""" if not audio_input: return "❌ **Error**: No audio provided" try: audio_path = self.save_gradio_audio(audio_input) result = self.audio_ai.answer_audio_question(audio_path) self.cleanup_temp_file(audio_path) return result except Exception as e: logger.error(f"Audio Q&A error: {e}") return f"❌ **Audio Q&A Failed**: {e}" def convert_audio_to_text(self, audio_input) -> str: """Convert audio to text""" if not audio_input: return "❌ **Error**: No audio provided" try: audio_path = self.save_gradio_audio(audio_input) result = self.audio_ai.convert_speech_to_text(audio_path) self.cleanup_temp_file(audio_path) return result except Exception as e: logger.error(f"Speech-to-text error: {e}") return f"❌ **Speech Recognition Failed**: {e}" def save_gradio_audio(self, audio_input) -> str: """Save Gradio audio input to temporary file""" try: if isinstance(audio_input, str): # Already a file path return audio_input elif hasattr(audio_input, 'name'): # File object return audio_input.name else: # Handle other audio input types temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.wav') temp_file.write(audio_input) temp_file.close() return temp_file.name except Exception as e: logger.error(f"Audio saving error: {e}") raise def cleanup_temp_file(self, file_path: str): """Clean up temporary files""" try: if os.path.exists(file_path) and 'tmp' in file_path: os.unlink(file_path) except Exception as e: logger.warning(f"Cleanup warning: {e}") def create_user_profile(self, name: str, learning_style: str, goals: str) -> str: """Create and store user learning profile""" try: with sqlite3.connect(self.db.db_path) as conn: cursor = conn.cursor() cursor.execute( "INSERT INTO user_profiles (name, learning_style, goals) VALUES (?, ?, ?)", (name, learning_style, goals) ) conn.commit() return f"✅ **Profile Created** for {name}\n\n**Learning Style**: {learning_style}\n**Goals**: {goals}" except Exception as e: return f"❌ **Profile Creation Failed**: {e}" # ============================================================================ # Clean Gradio Interface # ============================================================================ def create_learning_interface(): """Create clean, production-ready Gradio interface""" # Initialize learning assistant learning_assistant = LearningAssistant() # Custom CSS for better styling custom_css = """ .gradio-container { font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif; } .main-header { background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: white; padding: 2rem; border-radius: 10px; text-align: center; margin-bottom: 2rem; } .feature-card { background: #f8f9fa; border-radius: 10px; padding: 1.5rem; margin: 1rem 0; border-left: 4px solid #667eea; } .status-indicator { padding: 0.5rem; border-radius: 5px; margin: 0.5rem 0; } .status-success { background: #d4edda; color: #155724; } .status-warning { background: #fff3cd; color: #856404; } .status-error { background: #f8d7da; color: #721c24; } """ with gr.Blocks(css=custom_css, title="AI Learning Assistant v2.0") as interface: # Header gr.HTML("""
Version 2.0 - Clean Production Build
Multi-Agent Reasoning • Voice AI • Real-Time Data • ZeroGPU Optimized
🏆 Gradio Agents & MCP Hackathon 2025
Multi-Agent AI • Voice Processing • Real-Time Data • ZeroGPU Optimized
Version 2.0 - Production Ready