import json import logging from typing import Dict, List, Optional from dataclasses import dataclass from langchain_ollama import ChatOllama from langchain.prompts import PromptTemplate from langchain.schema import HumanMessage # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class Course: """Data class for course information""" course_id: str name: str short_description: str description: str skill: str @dataclass class CourseFilterRequest: """Data class for course filtering request""" identified_field: str courses: List[Course] @dataclass class CourseFilterResponse: """Data class for course filtering response""" identified_field: str total_courses_input: int matching_courses: List[Course] total_matching_courses: int filtering_reasoning: str class CourseFieldIdentifier: """Handles course filtering based on identified field using LLM""" def __init__(self, model_name: str = "gemma3:12b"): """Initialize the course field identifier""" try: self.llm = ChatOllama(model=model_name) logger.info(f"Course Field Identifier initialized with model: {model_name}") except Exception as e: logger.error(f"Failed to initialize Course Field Identifier: {e}") raise def filter_courses_by_field(self, identified_field: str, courses: List[Course]) -> CourseFilterResponse: """ Filter courses that match the identified field using LLM analysis Args: identified_field: The field identified from skill analysis (e.g., "Frontend Development") courses: List of all available courses Returns: CourseFilterResponse with matching courses """ try: # Prepare courses data for LLM courses_text = self._format_courses_for_llm(courses) # Get LLM analysis matching_course_ids, reasoning = self._get_llm_course_analysis( identified_field, courses_text ) # Filter matching courses matching_courses = [ course for course in courses if course.course_id in matching_course_ids ] response = CourseFilterResponse( identified_field=identified_field, total_courses_input=len(courses), matching_courses=matching_courses, total_matching_courses=len(matching_courses), filtering_reasoning=reasoning ) logger.info(f"Filtered {len(matching_courses)} courses for {identified_field} field") return response except Exception as e: logger.error(f"Error in course filtering: {e}") return CourseFilterResponse( identified_field=identified_field, total_courses_input=len(courses), matching_courses=[], total_matching_courses=0, filtering_reasoning=f"Error in filtering: {str(e)}" ) def _format_courses_for_llm(self, courses: List[Course]) -> str: """Format courses data for LLM input""" courses_formatted = [] for course in courses: # Combine all course information full_description = f"{course.name}" if course.short_description: full_description += f" - {course.short_description}" if course.description: full_description += f" | {course.description}" if course.skill: full_description += f" | Skills: {course.skill}" courses_formatted.append(f"ID: {course.course_id} | {full_description}") return "\n".join(courses_formatted) def _get_llm_course_analysis(self, identified_field: str, courses_text: str) -> tuple: """Get LLM analysis of which courses match the identified field""" prompt_template = PromptTemplate( input_variables=["identified_field", "courses_text"], template=""" You are an EXPERT course curator. Be VERY STRICT in matching courses to the field. TARGET FIELD: {identified_field} COURSES TO EVALUATE: {courses_text} STRICT MATCHING CRITERIA: For {identified_field} field, ONLY approve courses that: 1. Directly teach {identified_field} skills and knowledge 2. Use tools/technologies commonly used in {identified_field} 3. Solve real problems that {identified_field} professionals face 4. Build competencies essential for {identified_field} roles DEFINITELY REJECT: - Courses from completely different fields (e.g., Programming courses for Sales) - General knowledge or academic subjects not job-relevant - Courses too basic/generic to help {identified_field} professionals - Administrative, testing, or demo content EXAMPLES FOR DIFFERENT FIELDS: Sales Field - APPROVE: "Sales Techniques", "CRM Training", "Negotiation Skills", "Customer Relations" Sales Field - REJECT: "Python Programming", "Angular Development", "Database Administration" Frontend Development - APPROVE: "Angular", "React", "JavaScript", "HTML/CSS", "UI/UX" Frontend Development - REJECT: "Sales Training", "HR Management", "Accounting Principles" BE EXTREMELY SELECTIVE. When in doubt, REJECT the course. Return JSON: {{ "matching_course_ids": ["id1", "id2"], "reasoning": "Detailed explanation of why specific courses were selected/rejected for {identified_field}" }} Analysis for {identified_field}: """ ) try: formatted_prompt = prompt_template.format( identified_field=identified_field, courses_text=courses_text ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) # Parse LLM response analysis_result = self._parse_llm_response(response.content) matching_course_ids = analysis_result.get("matching_course_ids", []) reasoning = analysis_result.get("reasoning", "LLM analysis completed") logger.info(f"LLM identified {len(matching_course_ids)} matching courses") return matching_course_ids, reasoning except Exception as e: logger.error(f"Error in LLM course analysis: {e}") return [], f"Error in LLM analysis: {str(e)}" def _parse_llm_response(self, response_content: str) -> Dict: """Parse LLM response to extract course IDs and reasoning""" 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 structure if "matching_course_ids" in parsed_response: return parsed_response # Fallback: try to extract course IDs from text return self._fallback_parse_response(response_content) except json.JSONDecodeError as e: logger.warning(f"JSON parsing failed: {e}") return self._fallback_parse_response(response_content) except Exception as e: logger.error(f"Error parsing LLM response: {e}") return {"matching_course_ids": [], "reasoning": "Failed to parse LLM response"} def _fallback_parse_response(self, response_content: str) -> Dict: """Fallback parsing when JSON extraction fails""" import re # Look for course ID patterns in text course_id_patterns = [ r'ID:\s*(\w+)', # "ID: 123" r'Course\s*(\w+)', # "Course 123" r'^\s*(\w+)\s*-', # "123 - Course Name" ] matching_ids = [] for pattern in course_id_patterns: matches = re.findall(pattern, response_content, re.MULTILINE) matching_ids.extend(matches) # Remove duplicates matching_ids = list(set(matching_ids)) return { "matching_course_ids": matching_ids, "reasoning": "Extracted course IDs from text response (fallback parsing)" } def process_json_request(self, json_data: str) -> Dict: """ Process JSON request and return JSON response Args: json_data: JSON string with identified_field and courses data Returns: Dictionary with filtered courses """ try: request_data = json.loads(json_data) # Extract identified field identified_field = request_data.get('identified_field', '') if not identified_field: return { 'success': False, 'error': 'No identified_field provided', 'matching_courses': [] } # Parse courses courses_data = request_data.get('courses', []) courses = [] for course_data in courses_data: course = Course( course_id=str(course_data.get('course_id', '')), name=course_data.get('course_name', ''), short_description=course_data.get('course_short_description', ''), description=course_data.get('course_description', ''), skill=course_data.get('skill', '') ) courses.append(course) # Filter courses filter_response = self.filter_courses_by_field(identified_field, courses) # Convert to JSON response matching_courses_json = [] for course in filter_response.matching_courses: matching_courses_json.append({ 'course_id': course.course_id, 'course_name': course.name, 'course_short_description': course.short_description, 'course_description': course.description, 'skill': course.skill }) return { 'success': True, 'identified_field': filter_response.identified_field, 'total_courses_input': filter_response.total_courses_input, 'matching_courses': matching_courses_json, 'total_matching_courses': filter_response.total_matching_courses, 'filtering_reasoning': filter_response.filtering_reasoning } except json.JSONDecodeError as e: logger.error(f"JSON decode error: {e}") return { 'success': False, 'error': f'Invalid JSON format: {e}', 'matching_courses': [] } except Exception as e: logger.error(f"Processing error: {e}") return { 'success': False, 'error': str(e), 'matching_courses': [] } # Example usage and testing def main(): """Example usage of the Course Field Identifier System""" # Initialize system course_identifier = CourseFieldIdentifier() # Sample request data sample_request = { "identified_field": "Sales ", "courses": [ { "course_id": "32", "course_name": "General Knowledge", "course_short_description": "General Knowledge", "course_description": "", "skills": [] }, { "course_id": "33", "course_name": "Current Affairs", "course_short_description": "Current Affairs", "course_description": "", "skills": [] }, { "course_id": "43", "course_name": "Python Programming for Complete Beginners ", "course_short_description": "Python Programming for Complete Beginners ", "course_description": "", "skills": [] }, { "course_id": "58", "course_name": "Laravel Beginner", "course_short_description": "Laravel Beginner description ", "course_description": "", "skills": [] }, { "course_id": "59", "course_name": "Codeigntier Beginner", "course_short_description": "Codeigntier Beginner short desc", "course_description": "", "skills": [] }, { "course_id": "60", "course_name": "New CCourse ", "course_short_description": "New COurse short desc", "course_description": "", "skills": [] }, { "course_id": "63", "course_name": "Angular Basic", "course_short_description": "Angular is a platform and framework for building single-page client applications using HTML and TypeScript. Angular is written in TypeScript. It implements core and optional functionality as a set of TypeScript libraries that you import into your applicat", "course_description": "

Angular is a platform and framework for building single-page client applications using HTML and TypeScript. Angular is written in TypeScript. It implements core and optional functionality as a set of TypeScript libraries that you import into your applications.<\/p>

The architecture of an Angular application relies on certain fundamental concepts. The basic building blocks of the Angular framework are Angular components that are organized into NgModules<\/i>. NgModules collect related code into functional sets; an Angular application is defined by a set of NgModules. An application always has at least a root module<\/i> that enables bootstrapping, and typically has many more feature modules<\/i>.<\/p>

Components define views<\/i>, which are sets of screen elements that Angular can choose among and modify according to your program logic and data.<\/p>

Components use services<\/i>, which provide specific functionality not directly related to views. Service providers can be injected<\/i> into components as dependencies<\/i>, making your code modular, reusable, and efficient.<\/p>", "skills": [] }, { "course_id": "64", "course_name": "Leadership Chapter 1", "course_short_description": "Leadership Chapter 1", "course_description": "", "skills": [] }, { "course_id": "65", "course_name": "Sorting", "course_short_description": "Sorting", "course_description": "", "skills": [] }, { "course_id": "66", "course_name": "CSS Selector", "course_short_description": "CSS Selector", "course_description": "