MMo4/csit-ned-chatbot
0
1"""2Gemini Client for CSIT RAG System3Updated for Google GenAI SDK (November 2025)4Model: gemini-2.5-flash5SDK: google.genai (new API)6"""7 8from google import genai9from typing import List, Dict, Any, Optional10import logging11import os12import time13from dataclasses import dataclass14 15logger = logging.getLogger(__name__)16 17@dataclass18class QueryAnalysis:19 """Structure for query analysis results"""20 intent: str # academic_info, programs, faculty_info, events, admissions, research, etc.21 departments: List[str] # CSIT, SE, CIS22 keywords: List[str]23 question_type: str # factual, comparative, list, procedural24 confidence: float25 26@dataclass27class ResponseGeneration:28 """Structure for generated response"""29 content: str30 confidence: float31 sources_used: List[str]32 follow_up_questions: List[str]33 34class GeminiClient:35 """36 Client for Google Gemini API with key rotation support.37 Uses new google.genai SDK (2025) with gemini-2.5-flash model.38 """39 40 def __init__(self, api_keys: Optional[List[str]] = None):41 """42 Initialize Gemini client with API key rotation.43 44 Args:45 api_keys: Optional list of API keys. If not provided, loads from environment.46 """47 # Load API keys from environment if not provided48 if api_keys:49 self.api_keys = api_keys50 else:51 self.api_keys = self._load_api_keys_from_env()52 53 if not self.api_keys:54 raise ValueError("No valid Gemini API keys found. Set GEMINI_API_KEY_1, GEMINI_API_KEY_2, etc.")55 56 self.current_key_index = 057 self.model_name = "gemini-2.5-flash"58 self.client = None59 60 logger.info(f"Loaded {len(self.api_keys)} API keys for rotation")61 self._initialize_client()62 63 def _load_api_keys_from_env(self) -> List[str]:64 """Load API keys from environment variables (GEMINI_API_KEY_1, etc.)"""65 keys = []66 for i in range(1, 20): # Support up to 20 keys67 key = os.getenv(f'GEMINI_API_KEY_{i}')68 if key:69 keys.append(key)70 71 # Fallback to single key72 if not keys:73 single_key = os.getenv('GEMINI_API_KEY')74 if single_key:75 keys.append(single_key)76 77 return keys78 79 def _initialize_client(self):80 """Initialize Gemini client with current API key"""81 try:82 current_key = self.api_keys[self.current_key_index]83 84 # Set environment variable for the SDK85 os.environ['GEMINI_API_KEY'] = current_key86 87 # Initialize new SDK client88 self.client = genai.Client()89 90 logger.info(f"Gemini client initialized with key index {self.current_key_index} (model: {self.model_name})")91 except Exception as e:92 logger.error(f"Failed to initialize Gemini client: {e}")93 raise94 95 def _rotate_api_key(self):96 """Rotate to next API key"""97 if len(self.api_keys) > 1:98 self.current_key_index = (self.current_key_index + 1) % len(self.api_keys)99 self._initialize_client()100 logger.info(f"Rotated to API key index {self.current_key_index}")101 else:102 logger.warning("Only one API key available, cannot rotate")103 104 def _make_request(self, prompt: str, max_retries: int = 3) -> str:105 """106 Make request to Gemini API with retry logic and key rotation.107 108 Args:109 prompt: The prompt to send110 max_retries: Maximum number of retry attempts111 112 Returns:113 Generated text response114 115 Raises:116 RuntimeError: If all retries fail117 """118 for attempt in range(max_retries):119 try:120 # Use new SDK method121 response = self.client.models.generate_content(122 model=self.model_name,123 contents=prompt124 )125 126 return response.text127 128 except Exception as e:129 logger.warning(f"Request attempt {attempt + 1} failed: {e}")130 131 if attempt < max_retries - 1:132 # Rotate key and retry133 self._rotate_api_key()134 time.sleep(2 ** attempt) # Exponential backoff135 else:136 logger.error(f"All {max_retries} request attempts failed")137 raise138 139 raise RuntimeError("Failed to make request after all retries")140 141 def analyze_query(self, query: str) -> QueryAnalysis:142 """143 Analyze user query to extract intent, departments, and metadata.144 145 Updated for CSIT department chatbot with SE/CIS academic support.146 147 Returns:148 QueryAnalysis with:149 - intent: academic_info, programs, faculty_info, events, etc.150 - departments: List of detected departments (CSIT, SE, CIS)151 - keywords: Extracted key terms152 - question_type: factual, comparative, list, procedural153 - confidence: Analysis confidence score154 """155 analysis_prompt = f"""Analyze this query about NED University departments:156 157Query: "{query}"158 159Identify:1601. Intent - Choose ONE from:161 - academic_info (courses/curriculum/syllabus)162 - programs (degree details/specializations)163 - faculty_info (professors/teachers)164 - events (hackathons/competitions)165 - admissions (requirements/application)166 - research (projects/publications/thesis)167 - facilities (labs/equipment)168 - policies (exam rules/grading/OBE)169 - general (department overview/contact)170 1712. Departments mentioned - Can be multiple from:172 - CSIT (Computer Science & IT)173 - SE (Software Engineering)174 - CIS (Computer Information Systems)175 1763. Key terms/topics from the query177 1784. Question type:179 - factual (specific facts)180 - comparative (comparing options)181 - list (requesting multiple items)182 - procedural (how-to/process)183 184Respond in this EXACT format (one line each):185INTENT: <intent>186DEPARTMENTS: <comma-separated list or "CSIT" as default>187KEYWORDS: <comma-separated keywords>188QUESTION_TYPE: <type>189"""190 191 try:192 response = self._make_request(analysis_prompt)193 return self._parse_query_analysis(response, query)194 195 except Exception as e:196 logger.error(f"Query analysis failed: {e}")197 # Return default analysis198 return QueryAnalysis(199 intent='general',200 departments=['CSIT'],201 keywords=query.split()[:5],202 question_type='factual',203 confidence=0.5204 )205 206 def _parse_query_analysis(self, response: str, original_query: str) -> QueryAnalysis:207 """Parse LLM response into QueryAnalysis structure."""208 lines = response.strip().split('\n')209 analysis = {210 'intent': 'general',211 'departments': [],212 'keywords': [],213 'question_type': 'factual'214 }215 216 for line in lines:217 if ':' not in line:218 continue219 220 key, value = line.split(':', 1)221 key = key.strip().upper()222 value = value.strip()223 224 if key == 'INTENT':225 analysis['intent'] = value.lower()226 elif key == 'DEPARTMENTS':227 if value.lower() != 'none':228 analysis['departments'] = [d.strip() for d in value.split(',')]229 elif key == 'KEYWORDS':230 analysis['keywords'] = [k.strip() for k in value.split(',')]231 elif key == 'QUESTION_TYPE':232 analysis['question_type'] = value.lower()233 234 # Default to CSIT if no department detected235 if not analysis['departments']:236 analysis['departments'] = ['CSIT']237 238 return QueryAnalysis(239 intent=analysis['intent'],240 departments=analysis['departments'],241 keywords=analysis['keywords'],242 question_type=analysis['question_type'],243 confidence=0.85 # High confidence in analysis244 )245 246 def generate_response(247 self,248 query: str,249 context_documents: List[Dict[str, Any]],250 query_analysis: QueryAnalysis,251 system_prompt: str252 ) -> ResponseGeneration:253 """254 Generate response using retrieved context and query analysis.255 256 Args:257 query: User's question258 context_documents: Retrieved documents with content and metadata259 query_analysis: Analyzed query metadata260 system_prompt: System-level instructions for response generation261 262 Returns:263 ResponseGeneration with formatted response264 """265 # Format context from documents266 context_text = self._format_context_documents(context_documents)267 268 # Build response prompt269 response_prompt = f"""{system_prompt}270 271QUERY: {query}272 273CONTEXT FROM VERIFIED SOURCES:274{context_text}275 276QUERY ANALYSIS:277- Intent: {query_analysis.intent}278- Departments: {', '.join(query_analysis.departments)}279- Question Type: {query_analysis.question_type}280 281INSTRUCTIONS:2821. Use ONLY information from the provided context2832. Cite sources: "According to the CSIT department website..." or "Based on the [year] prospectus..."2843. Be friendly but professional and factual2854. If information is incomplete, acknowledge it gracefully and provide contact info2865. Keep length appropriate: 150-300 words for most queries, longer for course listings2876. For historical events, mention the year clearly288 289Generate response:"""290 291 try:292 response_content = self._make_request(response_prompt)293 294 # Generate follow-up questions295 follow_up_questions = self._generate_followup_questions(query, response_content, query_analysis)296 297 # Extract sources used298 sources_used = [299 doc['metadata'].get('chunk_id', 'unknown')300 for doc in context_documents301 ]302 303 return ResponseGeneration(304 content=response_content.strip(),305 confidence=0.85,306 sources_used=sources_used,307 follow_up_questions=follow_up_questions308 )309 310 except Exception as e:311 logger.error(f"Response generation failed: {e}")312 return ResponseGeneration(313 content="I apologize, but I encountered an error generating a response. Please try again or contact the CSIT department office.",314 confidence=0.0,315 sources_used=[],316 follow_up_questions=[]317 )318 319 def _format_context_documents(self, documents: List[Dict[str, Any]]) -> str:320 """Format retrieved documents into context text for the LLM."""321 if not documents:322 return "No relevant context found."323 324 context_parts = []325 326 for i, doc in enumerate(documents, 1):327 metadata = doc.get('metadata', {})328 content = doc.get('document', '')329 330 # Format source information331 title = metadata.get('title', 'Unknown Document')332 category = metadata.get('category', 'N/A')333 departments = metadata.get('departments', [])334 335 context_part = f"""336--- Source {i}: {title} ---337Category: {category}338Departments: {', '.join(departments) if departments else 'N/A'}339 340Content:341{content[:1000]}...342"""343 context_parts.append(context_part.strip())344 345 return '\n\n'.join(context_parts)346 347 def _generate_followup_questions(348 self,349 query: str,350 response: str,351 analysis: QueryAnalysis352 ) -> List[str]:353 """Generate relevant follow-up questions."""354 followup_prompt = f"""Based on this conversation about NED University:355 356User Query: "{query}"357Intent: {analysis.intent}358Departments: {', '.join(analysis.departments)}359 360Response: "{response[:300]}..."361 362Suggest 3 relevant follow-up questions the user might want to ask.363Consider the intent and departments mentioned.364 365Format as:3661. Question one?3672. Question two?3683. Question three?369"""370 371 try:372 followup_response = self._make_request(followup_prompt)373 return self._parse_followup_questions(followup_response)374 except Exception as e:375 logger.warning(f"Follow-up generation failed: {e}")376 return []377 378 def _parse_followup_questions(self, response: str) -> List[str]:379 """Parse follow-up questions from LLM response."""380 lines = response.strip().split('\n')381 questions = []382 383 for line in lines:384 line = line.strip()385 # Look for numbered questions386 if line and (line[0].isdigit() or line.startswith('-')):387 # Remove numbering388 question = line.split('.', 1)[-1].strip()389 question = question.split(')', 1)[-1].strip() # Handle "1)" format390 391 if question and len(question) > 10: # Reasonable question length392 questions.append(question)393 394 return questions[:3] # Max 3 questions395 396# Global instance for reuse397_gemini_client = None398 399def get_gemini_client() -> GeminiClient:400 """401 Get or create the global Gemini client instance.402 Singleton pattern for efficient client reuse.403 """404 global _gemini_client405 if _gemini_client is None:406 _gemini_client = GeminiClient()407 return _gemini_client408