MMo4/csit-ned-chatbot
0
1"""
2Query processor for CSIT RAG system.
3Handles query preprocessing and enhancement for hybrid architecture.
4"""
5
6import re
7import logging
8from typing import Dict, List, Any, Optional
9from dataclasses import dataclass
10
11logger = logging.getLogger(__name__)
12
13@dataclass
14class ProcessedQuery:
15 """Processed query with extracted metadata."""
16 original_query: str
17 enhanced_query: str
18 detected_intent: str
19 detected_departments: List[str]
20 detected_specializations: List[str] # NEW
21 detected_emotions: List[str]
22 extracted_concepts: List[str]
23 question_type: str
24 confidence: float
25
26class QueryProcessor:
27 """Processes and enhances queries for better retrieval."""
28
29 def __init__(self):
30 # Define intent patterns (CSIT department focused)
31 self.intent_patterns = {
32 'academic_info': [
33 r'(courses?|subjects?|curriculum|credit hours?|semesters?|prospectus)',
34 r'(teach|taught|syllabus|outline|structure|timetable)',
35 r'(year courses|1st year|2nd year|3rd year|4th year|final year|third year|fourth year)',
36 r'(electives?|specialisation|specialization|track courses|program courses)',
37 r'(ai courses|cyber courses|ds courses|gaming courses|data science courses)',
38 r'(compare.*courses|courses.*compare|versus courses|vs courses)'
39 ],
40 'student_activities': [
41 r'(student club|clubs?|societies|student organization)',
42 r'(koderz|cybersents|ai alliance|data insight|gameverse|ledger league|qubit)',
43 r'(what clubs|all clubs|list of clubs|clubs at|clubs in)',
44 r'(tell me about clubs|student activities|extracurricular)'
45 ],
46 'programs': [
47 r'(program|degree|specialization|track|major)',
48 r'(bs|ms|phd|bachelors|masters|doctorate)'
49 ],
50 'faculty_info': [
51 r'(professor|teacher|faculty|instructor|lecturer|chairman)',
52 r'(dr\.|prof\.|supervisor)',
53 r'(who are|list of|all faculty|faculty members|faculty list)',
54 r'(staff|department faculty|csit faculty|faculty at)'
55 ],
56 'events': [
57 r'(event|hackathon|competition|koderz kombat|webkode)',
58 r'(seminar|workshop|conference|sports fest|techfest|iconics)'
59 ],
60 'admissions': [
61 r'(admission|merit|requirement|eligibility|apply)',
62 r'(application|criteria|deadline|entry test)'
63 ],
64 'research': [
65 r'(research|publication|thesis|fyp|dissertation)',
66 r'(project|paper|journal|lab work)'
67 ],
68 'achievements': [
69 r'(alumni|achievements?|accomplishments?|success|notable)',
70 r'(shining stars?|graduates?|former students?)',
71 r'(career paths?|where.*working|successful|placements?)'
72 ],
73 'facilities': [
74 r'(lab|facility|resource|equipment|library)',
75 r'(infrastructure|computer lab|workspace)'
76 ],
77 'policies': [
78 r'(policy|rule|regulation|obe|grading|exam)',
79 r'(attendance|assessment|rubric)'
80 ],
81 'general': [
82 r'(what|how|why|when|where)',
83 r'(tell|explain|describe|about)'
84 ]
85 }
86
87 # Department detection (CSIT primary focus)
88 self.department_patterns = {
89 'CSIT': ['csit', 'computer science', 'cs', 'bcit', 'bs cs', 'bs computer science'],
90 'SE': ['se', 'software engineering', 'software engineer', r'\bse\b'],
91 'CIS': ['cis', 'computer engineering', 'ce', 'computer systems', 'computer information systems']
92 }
93
94 # 🔹 NEW: Specialization detection (improved patterns)
95 self.specialization_patterns = {
96 "Cybersecurity": [r'\bcyber\b', 'cybersecurity', 'cyber security', 'cyber track', 'information security'],
97 "AI": [r'\bai\b', 'artificial intelligence', 'ai track', 'ai specialization', 'machine learning track'],
98 "Data Science": ['data science', r'\bds\b', 'ds track', 'data analytics'],
99 "Gaming & Animation": ['gaming', 'game dev', 'game development', 'animation', 'gaming and animation', 'ga track'],
100 "General": ['general track', 'general path', 'standard path', 'no specialization', 'general cs', 'core cs', 'core computer science', 'standard cs', 'basic cs', 'common cs']
101 }
102
103 # Year-level patterns for smart defaulting
104 self.year_patterns = {
105 '1st': [r'\b1st\b', 'first year', 'year 1', 'year one'],
106 '2nd': [r'\b2nd\b', 'second year', 'year 2', 'year two'],
107 '3rd': [r'\b3rd\b', 'third year', 'year 3', 'year three'],
108 '4th': [r'\b4th\b', 'fourth year', 'final year', 'year 4', 'year four', 'last year']
109 }
110
111 # 🔹 External universities (out-of-scope detection)
112 self.external_universities = {
113 'FAST': ['fast university', 'nufast', 'nu-fast', 'nu fast', r'\bfast\b'],
114 'LUMS': ['lums', 'lahore university of management'],
115 'NUST': ['nust', 'national university of science'],
116 'IBA': ['iba karachi', 'iba sukkur', r'\biba\b'],
117 'GIKI': ['giki', 'ghulam ishaq khan'],
118 'COMSATS': ['comsats', 'ciit'],
119 'UET': ['uet lahore', 'uet peshawar', 'uet taxila', r'\buet\b'],
120 'SZABIST': ['szabist'],
121 'Bahria': ['bahria university'],
122 'Air University': ['air university'],
123 'PIEAS': ['pieas'],
124 'Karachi University': ['karachi university', 'ku'],
125 'Punjab University': ['punjab university', 'pu lahore']
126 }
127
128 # Emotion detection
129 self.emotion_patterns = {
130 'worried': ['worry', 'worried', 'concern', 'concerned', 'anxiety', 'anxious'],
131 'curious': ['curious', 'interest', 'interested', 'want to know'],
132 'confused': ['confused', 'unclear', 'don\'t understand', 'complex'],
133 'confident': ['confident', 'sure', 'certain', 'ready'],
134 'neutral': [] # Default
135 }
136
137 # Concept extraction patterns
138 self.concept_patterns = [
139 'theory', 'theoretical', 'practical', 'hands-on',
140 'curriculum', 'course', 'subject', 'lab',
141 'career', 'job', 'employment', 'salary', 'company',
142 'admission', 'merit', 'requirement', 'eligibility',
143 'graduate', 'student', 'alumni', 'placement',
144 'specialization'
145 ]
146
147 def process_query(self, query: str, conversation_history: Optional[List[Dict]] = None) -> ProcessedQuery:
148 """Process and enhance a user query."""
149 try:
150 logger.info(f"Processing query: {query[:100]}...")
151
152 # Clean the query
153 cleaned_query = self._clean_query(query)
154
155 # Extract metadata
156 intent = self._detect_intent(cleaned_query)
157 departments = self._detect_departments(cleaned_query)
158 specializations = self._detect_specializations(cleaned_query) # NEW
159 emotions = self._detect_emotions(cleaned_query)
160 concepts = self._extract_concepts(cleaned_query)
161 question_type = self._classify_question_type(cleaned_query)
162
163 # Enhance query with context
164 enhanced_query = self._enhance_query(
165 cleaned_query, intent, departments, concepts, conversation_history, specializations
166 )
167
168 # Calculate confidence
169 confidence = self._calculate_confidence(intent, departments, concepts)
170
171 processed = ProcessedQuery(
172 original_query=query,
173 enhanced_query=enhanced_query,
174 detected_intent=intent,
175 detected_departments=departments,
176 detected_specializations=specializations, # NEW
177 detected_emotions=emotions,
178 extracted_concepts=concepts,
179 question_type=question_type,
180 confidence=confidence
181 )
182
183 logger.info(f"Query processed - Intent: {intent}, Departments: {departments}, Specializations: {specializations}")
184 return processed
185
186 except Exception as e:
187 logger.error(f"Error processing query: {e}")
188 return ProcessedQuery(
189 original_query=query,
190 enhanced_query=query,
191 detected_intent='general',
192 detected_departments=[],
193 detected_specializations=[], # fallback
194 detected_emotions=['neutral'],
195 extracted_concepts=[],
196 question_type='general',
197 confidence=0.5
198 )
199
200 def _clean_query(self, query: str) -> str:
201 """Clean and normalize the query."""
202 cleaned = query.lower().strip()
203 cleaned = re.sub(r'\s+', ' ', cleaned)
204 return cleaned
205
206 def _detect_intent(self, query: str) -> str:
207 """
208 Detect the intent of the query with priority ordering.
209 Academic queries are prioritized over general queries.
210 """
211 # Priority order: specific intents first, general last
212 priority_intents = [
213 'academic_info', # HIGHEST PRIORITY
214 'student_activities', # Student clubs (before programs to catch "clubs" before "program")
215 'achievements', # Alumni and achievements
216 'programs',
217 'faculty_info',
218 'events',
219 'admissions',
220 'research',
221 'facilities',
222 'policies'
223 ]
224
225 # Check priority intents first
226 for intent in priority_intents:
227 if intent in self.intent_patterns:
228 for pattern in self.intent_patterns[intent]:
229 if re.search(pattern, query, re.IGNORECASE):
230 return intent
231
232 # Fallback to general if no specific intent matched
233 return 'general'
234
235 def _detect_departments(self, query: str) -> List[str]:
236 """Detect mentioned departments."""
237 detected = []
238 for dept, patterns in self.department_patterns.items():
239 for pattern in patterns:
240 if re.search(rf"\b{re.escape(pattern.lower())}\b", query):
241 detected.append(dept)
242 break
243 return list(set(detected))
244
245 def _detect_specializations(self, query: str) -> List[str]:
246 """
247 Detect mentioned specializations.
248 Smart logic: If asking about 3rd/4th year courses without mentioning
249 a specific specialization, default to showing ALL options (don't filter).
250 """
251 detected = []
252 query_lower = query.lower()
253
254 # First, check if any specific specialization is mentioned
255 for spec, patterns in self.specialization_patterns.items():
256 for pattern in patterns:
257 # Handle regex patterns (those starting with r'\b')
258 if pattern.startswith(r'\b'):
259 if re.search(pattern, query_lower):
260 detected.append(spec)
261 break
262 else:
263 # Simple substring match for non-regex patterns
264 if pattern in query_lower:
265 detected.append(spec)
266 break
267
268 # Smart defaulting logic:
269 # If query is about 3rd/4th year courses but NO specialization mentioned,
270 # DON'T add "General" - let it retrieve from all specializations
271 # This allows the user to see all options and choose
272
273 # Check if query mentions 3rd/4th year
274 is_upper_year_query = False
275 for year in ['3rd', '4th']:
276 if year in self.year_patterns:
277 for pattern in self.year_patterns[year]:
278 if pattern.startswith(r'\b'):
279 if re.search(pattern, query_lower):
280 is_upper_year_query = True
281 break
282 else:
283 if pattern in query_lower:
284 is_upper_year_query = True
285 break
286
287 # If it's an upper year query with NO detected specialization,
288 # don't filter by specialization (return empty list)
289 # This lets the retrieval return courses from ALL specializations
290 if is_upper_year_query and not detected:
291 logger.info(f"🎓 Upper year query without specialization - will show all options")
292 return [] # Don't filter - show all
293
294 return list(set(detected))
295
296 def detect_out_of_scope(self, query: str) -> Dict[str, Any]:
297 """Detect if query mentions external universities or out-of-scope topics."""
298 query_lower = query.lower()
299
300 # Check for external universities
301 for university, patterns in self.external_universities.items():
302 for pattern in patterns:
303 # Handle regex patterns (those with \b)
304 if pattern.startswith(r'\b'):
305 if re.search(pattern, query_lower):
306 return {
307 'is_out_of_scope': True,
308 'reason': 'external_university',
309 'entity': university
310 }
311 # Handle simple string patterns
312 elif pattern in query_lower:
313 return {
314 'is_out_of_scope': True,
315 'reason': 'external_university',
316 'entity': university
317 }
318
319 # Query is in scope
320 return {
321 'is_out_of_scope': False,
322 'reason': None,
323 'entity': None
324 }
325
326 def _detect_emotions(self, query: str) -> List[str]:
327 """Detect emotional context in the query."""
328 detected = []
329 for emotion, patterns in self.emotion_patterns.items():
330 if emotion == 'neutral': continue
331 if any(p.lower() in query for p in patterns):
332 detected.append(emotion)
333 return detected if detected else ['neutral']
334
335 def _extract_concepts(self, query: str) -> List[str]:
336 """Extract key concepts."""
337 return [c for c in self.concept_patterns if c.lower() in query.lower()]
338
339 def _classify_question_type(self, query: str) -> str:
340 if re.search(r'\?', query):
341 if re.search(r'^(what|how|why|when|where|which)', query, re.IGNORECASE):
342 return 'informational'
343 elif re.search(r'^(is|are|do|does|can|will)', query, re.IGNORECASE):
344 return 'yes_no'
345 else:
346 return 'open_ended'
347 return 'statement'
348
349 def _enhance_query(self, query: str, intent: str, departments: List[str],
350 concepts: List[str], conversation_history: Optional[List[Dict]] = None,
351 specializations: Optional[List[str]] = None) -> str:
352 """
353 Enhance the query with additional context for better retrieval.
354 This enhanced query is used for embedding generation to improve semantic matching.
355 """
356 # Start with original query
357 enhanced_parts = [query]
358
359 # Add department context for better matching
360 if departments:
361 dept_names = []
362 for dept in departments:
363 if dept == 'CSIT':
364 dept_names.append('Computer Science Information Technology BCIT')
365 elif dept == 'SE':
366 dept_names.append('Software Engineering')
367 elif dept == 'CIS':
368 dept_names.append('Computer Engineering Computer Information Systems')
369 if dept_names:
370 enhanced_parts.append(' '.join(dept_names))
371
372 # Add specialization context for better matching
373 if specializations:
374 spec_expansions = []
375 for spec in specializations:
376 if spec == 'AI':
377 spec_expansions.append('Artificial Intelligence Machine Learning')
378 elif spec == 'Cybersecurity':
379 spec_expansions.append('Cybersecurity Information Security Network Security')
380 elif spec == 'Data Science':
381 spec_expansions.append('Data Science Data Analytics Big Data')
382 elif spec == 'Gaming & Animation':
383 spec_expansions.append('Gaming Game Development Animation Graphics')
384 else:
385 spec_expansions.append(spec)
386 if spec_expansions:
387 enhanced_parts.append(' '.join(spec_expansions))
388
389 # Add intent-specific keywords for better matching
390 intent_keywords = {
391 'academic_info': 'courses curriculum syllabus subjects credit hours',
392 'faculty_info': 'professor teacher instructor lecturer faculty staff members specialization education background PhD',
393 'student_activities': 'student clubs societies organizations Koderz CyberSENTS AI Alliance Data Insight Gameverse Ledger League Qubits Qrew extracurricular activities',
394 'achievements': 'alumni graduates achievements accomplishments success stories shining stars career placements notable former students',
395 'events': 'event competition hackathon seminar workshop TechFest ICONICS SportsFest',
396 'admissions': 'admission merit eligibility requirements criteria',
397 'programs': 'program degree specialization track',
398 }
399 if intent in intent_keywords:
400 enhanced_parts.append(intent_keywords[intent])
401
402 # For comprehensive faculty queries, add specific boost keywords
403 if intent == 'faculty_info':
404 comprehensive_patterns = ['all faculty', 'faculty members', 'list of', 'who are', 'faculty list', 'tell me about faculty']
405 if any(pattern in query.lower() for pattern in comprehensive_patterns):
406 enhanced_parts.append('Chairman Professors Associate Professors Assistant Professors Lecturers CSIT Department NEDUET')
407
408 # Add conversation context for continuity
409 if conversation_history:
410 recent = self._extract_conversation_context(conversation_history)
411 if recent:
412 enhanced_parts.append(' '.join(recent[:3]))
413
414 # Combine all parts
415 enhanced = ' '.join(enhanced_parts)
416
417 # Add metadata tags for logging (not used in embedding)
418 metadata_tags = f" [Intent: {intent}]"
419 if departments:
420 metadata_tags += f" [Departments: {', '.join(departments)}]"
421 if specializations:
422 metadata_tags += f" [Specializations: {', '.join(specializations)}]"
423
424 logger.info(f"Enhanced query: {enhanced[:150]}... {metadata_tags}")
425
426 return enhanced
427
428 def _extract_conversation_context(self, history: List) -> List[str]:
429 """Extract relevant context from conversation history."""
430 context_topics = []
431 for interaction in history[-3:]:
432 if isinstance(interaction, dict):
433 query = interaction.get('query', '')
434 context_topics.extend(self._extract_concepts(query)[:2])
435 context_topics.extend(self._detect_departments(query))
436 return list(set(context_topics))[:5]
437
438 def _calculate_confidence(self, intent: str, departments: List[str], concepts: List[str]) -> float:
439 confidence = 0.5
440 if intent != 'general': confidence += 0.2
441 if departments: confidence += 0.15 * len(departments)
442 if concepts: confidence += 0.1 * min(len(concepts), 3)
443 return min(confidence, 1.0)
444
445 def get_query_suggestions(self, query: str) -> List[str]:
446 """Generate contextual query suggestions based on processed query."""
447 processed = self.process_query(query)
448 suggestions = []
449
450 # CSIT-focused suggestions
451 if processed.detected_intent == 'academic_info':
452 suggestions.extend([
453 "What are the specializations available in CSIT?",
454 "Show me the course structure for AI track",
455 "What electives can I choose in 3rd year?"
456 ])
457 elif processed.detected_intent == 'programs':
458 suggestions.extend([
459 "What's the difference between AI and Data Science specializations?",
460 "Tell me about MS programs in CSIT",
461 "What are the admission requirements for PhD?"
462 ])
463 elif processed.detected_intent == 'faculty_info':
464 suggestions.extend([
465 "Who are the professors in CSIT department?",
466 "What research areas does the faculty focus on?",
467 "Who is the department chairman?"
468 ])
469 elif processed.detected_intent == 'events':
470 suggestions.extend([
471 "What is Koderz Kombat competition?",
472 "Tell me about recent hackathons",
473 "When is the next CSIT event?"
474 ])
475 return suggestions[:3]
476
477 def is_se_cis_academic_query(self, processed: ProcessedQuery) -> bool:
478 """
479 Check if query is about SE/CIS academics (allowed).
480
481 Args:
482 processed: ProcessedQuery object
483
484 Returns:
485 True if SE/CIS academic query, False otherwise
486 """
487 has_se_cis = any(dept in processed.detected_departments for dept in ['SE', 'CIS'])
488 is_academic = processed.detected_intent == 'academic_info'
489
490 return has_se_cis and is_academic
491
492 def should_redirect_to_department(self, processed: ProcessedQuery) -> Optional[str]:
493 """
494 Determine if query should be redirected to another department.
495
496 Args:
497 processed: ProcessedQuery object
498
499 Returns:
500 Department name to redirect to, or None
501 """
502 # SE non-academic query
503 if 'SE' in processed.detected_departments and processed.detected_intent != 'academic_info':
504 return 'SE'
505
506 # CIS non-academic query
507 if 'CIS' in processed.detected_departments and processed.detected_intent != 'academic_info':
508 return 'CIS'
509
510 return None
511