import json import logging from typing import Dict, List, Optional from dataclasses import dataclass from langchain_ollama import ChatOllama from langchain_core.prompts import PromptTemplate from langchain_core.messages import HumanMessage # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class OnlineCourseRecommendation: """Data class for online course recommendation""" course_name: str platform: str instructor: str duration: str skill_level: str description: str url: str price: str rating: str skills_covered: List[str] match_percentage: int recommendation_reason: str @dataclass class OnlineCourseResponse: """Data class for online course recommendation response""" domain: str field_prompt: str total_recommendations: int course_recommendations: List[OnlineCourseRecommendation] search_summary: str success: bool class OnlineCourseRecommendationSystem: """Online Course Recommendation System using LLM""" def __init__(self, model_name: str = "gemma3:12b"): """Initialize the online course recommendation system""" try: self.llm = ChatOllama( model=model_name, num_ctx=128000, # Use full 128K context temperature=0.2, # Slightly higher for more creativity in recommendations num_predict=4000 # Allow longer responses for detailed recommendations ) logger.info(f"Online Course Recommendation System initialized with model: {model_name}") except Exception as e: logger.error(f"Failed to initialize Online Course Recommendation System: {e}") raise def recommend_online_courses(self, domain: str, field_prompt: str, max_courses: int = 10) -> OnlineCourseResponse: """ Recommend online courses based on domain and field prompt Args: domain: The domain category (e.g., "Technology", "Business", "Healthcare") field_prompt: The specific field or skill area max_courses: Maximum number of courses to recommend Returns: OnlineCourseResponse with course recommendations """ try: logger.info(f"Generating online course recommendations for {domain} - {field_prompt}") # Generate course recommendations using LLM recommendations = self._generate_course_recommendations(domain, field_prompt, max_courses) # Create response response = OnlineCourseResponse( domain=domain, field_prompt=field_prompt, total_recommendations=len(recommendations), course_recommendations=recommendations, search_summary=f"Generated {len(recommendations)} online course recommendations for {field_prompt} in {domain} domain", success=len(recommendations) > 0 ) logger.info(f"Generated {len(recommendations)} online course recommendations") return response except Exception as e: logger.error(f"Error generating online course recommendations: {e}") return OnlineCourseResponse( domain=domain, field_prompt=field_prompt, total_recommendations=0, course_recommendations=[], search_summary=f"Error generating recommendations: {str(e)}", success=False ) def _generate_course_recommendations(self, domain: str, field_prompt: str, max_courses: int) -> List[OnlineCourseRecommendation]: """Generate course recommendations using LLM""" prompt_template = PromptTemplate( input_variables=["domain", "field_prompt", "max_courses"], template=""" You are an EXPERT online course curator and career advisor specializing in the {domain} domain. DOMAIN: {domain} TARGET FIELD/SKILL: {field_prompt} MAX COURSES: {max_courses} TASK: Recommend the BEST online courses available for someone who wants to learn "{field_prompt}" in the {domain} domain. COURSE SELECTION CRITERIA: 1. High-quality, reputable platforms (Coursera, Udemy, edX, Pluralsight, LinkedIn Learning, etc.) 2. Well-reviewed courses with good ratings 3. Comprehensive coverage of {field_prompt} 4. Mix of beginner to advanced levels 5. Practical, hands-on learning approach 6. Industry-relevant and up-to-date content PLATFORMS TO CONSIDER: - Coursera (University partnerships, certificates) - Udemy (Practical, affordable courses) - edX (Academic, high-quality content) - Pluralsight (Tech-focused, skill assessments) - LinkedIn Learning (Professional development) - Skillshare (Creative and business skills) - MasterClass (Expert-led courses) - Codecademy (Programming and tech) - Khan Academy (Fundamentals and academics) - FutureLearn (University partnerships) RESPONSE FORMAT: Return ONLY a JSON array with this exact structure: [ {{ "course_name": "Complete Course Title", "platform": "Platform Name", "instructor": "Instructor Name or Institution", "duration": "X hours/weeks", "skill_level": "Beginner/Intermediate/Advanced", "description": "Detailed course description covering what students will learn", "url": "https://platform.com/course-url", "price": "Free/Paid/$XX", "rating": "4.5/5 stars", "skills_covered": ["skill1", "skill2", "skill3"], "match_percentage": 95, "recommendation_reason": "Why this course is perfect for {field_prompt}" }} ] IMPORTANT GUIDELINES: - Provide realistic course information (avoid making up specific URLs) - Include a variety of skill levels and platforms - Ensure courses are directly relevant to {field_prompt} - Match percentage should reflect relevance (70-100%) - Include both free and paid options when possible - Focus on courses that are currently available and popular - Provide practical, actionable course recommendations """ ) try: formatted_prompt = prompt_template.format( domain=domain, field_prompt=field_prompt, max_courses=max_courses ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) # Parse LLM response recommendations = self._parse_course_response(response.content) logger.info(f"LLM generated {len(recommendations)} course recommendations") return recommendations except Exception as e: logger.error(f"Error in LLM course generation: {e}") return [] def _parse_course_response(self, response_content: str) -> List[OnlineCourseRecommendation]: """Parse LLM response to extract course recommendations""" try: # Try to extract JSON from response import re json_match = re.search(r'\[.*\]', response_content, re.DOTALL) if json_match: parsed_response = json.loads(json_match.group()) # Validate and convert to OnlineCourseRecommendation objects recommendations = [] for item in parsed_response: if self._validate_course_item(item): recommendation = OnlineCourseRecommendation( course_name=item["course_name"], platform=item["platform"], instructor=item["instructor"], duration=item["duration"], skill_level=item["skill_level"], description=item["description"], url=item.get("url", ""), price=item["price"], rating=item["rating"], skills_covered=item["skills_covered"], match_percentage=int(item["match_percentage"]), recommendation_reason=item["recommendation_reason"] ) recommendations.append(recommendation) # Sort by match percentage (highest first) recommendations.sort(key=lambda x: x.match_percentage, reverse=True) return recommendations # Fallback: create generic recommendations if JSON parsing fails return self._create_fallback_recommendations() except json.JSONDecodeError as e: logger.warning(f"JSON parsing failed: {e}") return self._create_fallback_recommendations() except Exception as e: logger.error(f"Error parsing course response: {e}") return [] def _validate_course_item(self, item: dict) -> bool: """Validate that a course item has all required fields""" required_fields = [ "course_name", "platform", "instructor", "duration", "skill_level", "description", "price", "rating", "skills_covered", "match_percentage", "recommendation_reason" ] for field in required_fields: if field not in item: logger.warning(f"Missing required field: {field}") return False # Validate percentage is in valid range try: percentage = int(item["match_percentage"]) if percentage < 70 or percentage > 100: logger.warning(f"Invalid match percentage: {percentage}") return False except (ValueError, TypeError): logger.warning(f"Invalid match percentage format: {item.get('match_percentage')}") return False # Validate skills_covered is a list if not isinstance(item["skills_covered"], list): logger.warning(f"skills_covered must be a list: {item.get('skills_covered')}") return False return True def _create_fallback_recommendations(self) -> List[OnlineCourseRecommendation]: """Create fallback recommendations when LLM parsing fails""" return [ OnlineCourseRecommendation( course_name="Please check platform directly for relevant courses", platform="Multiple Platforms", instructor="Various Instructors", duration="Varies", skill_level="All Levels", description="Course recommendations could not be generated. Please search directly on platforms like Coursera, Udemy, or edX.", url="", price="Varies", rating="N/A", skills_covered=["general skills"], match_percentage=75, recommendation_reason="Fallback recommendation due to system limitation" ) ] def process_json_request(self, json_data: str) -> Dict: """ Process JSON request and return JSON response Expected input format: { "domain": "Technology", "field_prompt": "Python Programming", "max_courses": 10 } """ try: request_data = json.loads(json_data) # Extract required fields domain = request_data.get('domain', '') field_prompt = request_data.get('field_prompt', '') max_courses = request_data.get('max_courses', 10) if not domain or not field_prompt: return { 'success': False, 'error': 'Both domain and field_prompt are required', 'course_recommendations': [] } # Generate recommendations response = self.recommend_online_courses(domain, field_prompt, max_courses) # Convert to JSON response course_recommendations_json = [] for rec in response.course_recommendations: course_recommendations_json.append({ 'course_name': rec.course_name, 'platform': rec.platform, 'instructor': rec.instructor, 'duration': rec.duration, 'skill_level': rec.skill_level, 'description': rec.description, 'url': rec.url, 'price': rec.price, 'rating': rec.rating, 'skills_covered': rec.skills_covered, 'match_percentage': rec.match_percentage, 'recommendation_reason': rec.recommendation_reason }) return { 'success': response.success, 'domain': response.domain, 'field_prompt': response.field_prompt, 'total_recommendations': response.total_recommendations, 'course_recommendations': course_recommendations_json, 'search_summary': response.search_summary } except json.JSONDecodeError as e: logger.error(f"JSON decode error: {e}") return { 'success': False, 'error': f'Invalid JSON format: {e}', 'course_recommendations': [] } except Exception as e: logger.error(f"Processing error: {e}") return { 'success': False, 'error': str(e), 'course_recommendations': [] } # Test function def test_online_course_recommendations(): """Test the online course recommendation system""" # Test cases test_cases = [ { "domain": "Technology", "field_prompt": "Python Programming", "max_courses": 5 }, { "domain": "Business", "field_prompt": "Digital Marketing", "max_courses": 7 }, { "domain": "Technology", "field_prompt": "Data Science and Machine Learning", "max_courses": 8 }, { "domain": "Creative", "field_prompt": "UI/UX Design", "max_courses": 6 } ] print("=== TESTING ONLINE COURSE RECOMMENDATION SYSTEM ===") course_system = OnlineCourseRecommendationSystem() for test_case in test_cases: print(f"\n{'='*80}") print(f"TESTING: {test_case['domain']} - {test_case['field_prompt']}") print(f"{'='*80}") # Process request result = course_system.process_json_request(json.dumps(test_case)) # Display results print(f"āœ… SUCCESS: {result['success']}") print(f"šŸ“Š STATS:") print(f" Total recommendations: {result['total_recommendations']}") print(f" Domain: {result['domain']}") print(f" Field: {result['field_prompt']}") print(f"\nšŸŽ“ COURSE RECOMMENDATIONS:") for i, course in enumerate(result['course_recommendations'], 1): print(f"\n {i}. šŸ“š {course['course_name']}") print(f" Platform: {course['platform']}") print(f" Instructor: {course['instructor']}") print(f" Duration: {course['duration']}") print(f" Level: {course['skill_level']}") print(f" Price: {course['price']}") print(f" Rating: {course['rating']}") print(f" Match: {course['match_percentage']}%") print(f" Skills: {', '.join(course['skills_covered'])}") print(f" Reason: {course['recommendation_reason']}") print(f" Description: {course['description'][:100]}...") if not result['course_recommendations']: print(f" No recommendations found for {test_case['field_prompt']}") print(f"\nšŸ“ Summary: {result['search_summary']}") if __name__ == "__main__": test_online_course_recommendations()