Spaces:
Runtime error
Runtime error
| # """ | |
| # Router Agent - The Coordinator | |
| # Classifies user queries and routes them to appropriate specialist agents | |
| # Now with Gemini API integration! | |
| # """ | |
| # import re | |
| # from .academic_agent import AcademicAgent | |
| # from .drug_info_agent import DrugInfoAgent | |
| # from .quiz_agent import QuizAgent | |
| # from .mnemonic_agent import MnemonicAgent | |
| # from .viva_agent import VivaAgent | |
| # class RouterAgent: | |
| # def __init__(self, gemini_model=None): | |
| # # Store Gemini model | |
| # self.model = gemini_model | |
| # # Initialize specialist agents with Gemini model | |
| # self.academic_agent = AcademicAgent(gemini_model) | |
| # self.drug_info_agent = DrugInfoAgent(gemini_model) | |
| # self.quiz_agent = QuizAgent(gemini_model) | |
| # self.mnemonic_agent = MnemonicAgent(gemini_model) | |
| # self.viva_agent = VivaAgent(gemini_model) | |
| # # Define keywords for each agent type (Free-tier friendly classification) | |
| # self.agent_keywords = { | |
| # 'drug_info': [ | |
| # 'drug', 'medicine', 'medication', 'side effects', 'dosage', | |
| # 'contraindication', 'interaction', 'pharmacology', 'therapeutic', | |
| # 'adverse', 'mechanism', 'action', 'indication', 'prescription' | |
| # ], | |
| # 'quiz_generation': [ | |
| # 'quiz', 'test', 'questions', 'mcq', 'multiple choice', | |
| # 'flashcard', 'practice', 'exam', 'assessment', 'evaluate' | |
| # ], | |
| # 'mnemonic_creation': [ | |
| # 'mnemonic', 'remember', 'memory', 'trick', 'acronym', | |
| # 'rhyme', 'shortcut', 'memorize', 'recall', 'aide' | |
| # ], | |
| # 'viva_practice': [ | |
| # 'viva', 'oral', 'interview', 'practice session', 'mock', | |
| # 'question answer', 'preparation', 'rehearse' | |
| # ] | |
| # } | |
| # def classify_query_with_ai(self, query): | |
| # """Use Gemini AI to classify queries more accurately""" | |
| # if not self.model: | |
| # return self.classify_query(query) # Fallback to keyword matching | |
| # try: | |
| # classification_prompt = f""" | |
| # You are a query classifier for a pharmacy education AI assistant. | |
| # Classify this user query into ONE of these categories: | |
| # 1. academic_query - General academic questions about pharmacy, chemistry, biology, mechanisms | |
| # 2. drug_info - Specific questions about drugs, medicines, side effects, dosages, interactions | |
| # 3. quiz_generation - Requests to create quizzes, tests, MCQs, practice questions | |
| # 4. mnemonic_creation - Requests for memory aids, mnemonics, acronyms, memory tricks | |
| # 5. viva_practice - Requests for mock interviews, viva practice, oral exam preparation | |
| # User Query: "{query}" | |
| # Respond with ONLY the category name (e.g., "academic_query") | |
| # """ | |
| # response = self.model.generate_content(classification_prompt) | |
| # classification = response.text.strip().lower() | |
| # # Validate the classification | |
| # valid_types = ['academic_query', 'drug_info', 'quiz_generation', 'mnemonic_creation', 'viva_practice'] | |
| # if classification in valid_types: | |
| # return classification | |
| # else: | |
| # return 'academic_query' # Default fallback | |
| # except Exception as e: | |
| # print(f"AI classification failed: {e}") | |
| # return self.classify_query(query) # Fallback to keyword matching | |
| # """ | |
| # Classify the user query into one of the agent categories | |
| # Uses keyword matching for free-tier efficiency | |
| # """ | |
| # query_lower = query.lower() | |
| # # Count keyword matches for each agent type | |
| # scores = {} | |
| # for agent_type, keywords in self.agent_keywords.items(): | |
| # score = sum(1 for keyword in keywords if keyword in query_lower) | |
| # scores[agent_type] = score | |
| # # Special pattern matching for better accuracy | |
| # if re.search(r'\b(what|explain|definition|mechanism|process|how does)\b', query_lower): | |
| # scores['drug_info'] += 1 if any(drug_word in query_lower for drug_word in ['drug', 'medicine', 'pharmacology']) else 0 | |
| # scores.setdefault('academic_query', 0) | |
| # scores['academic_query'] += 1 | |
| # if re.search(r'\b(create|make|generate|give me)\s+(quiz|questions|mcq)\b', query_lower): | |
| # scores['quiz_generation'] += 2 | |
| # if re.search(r'\b(help.*remember|memory.*trick|mnemonic.*for)\b', query_lower): | |
| # scores['mnemonic_creation'] += 2 | |
| # if re.search(r'\b(practice.*viva|mock.*interview|oral.*exam)\b', query_lower): | |
| # scores['viva_practice'] += 2 | |
| # # Find the highest scoring agent type | |
| # if max(scores.values()) == 0: | |
| # return 'academic_query' # Default to academic for general questions | |
| # return max(scores, key=scores.get) | |
| # def route_query(self, query): | |
| # """ | |
| # Route the query to the appropriate specialist agent | |
| # """ | |
| # agent_type = self.classify_query(query) | |
| # try: | |
| # if agent_type == 'drug_info': | |
| # response = self.drug_info_agent.process_query(query) | |
| # elif agent_type == 'quiz_generation': | |
| # response = self.quiz_agent.process_query(query) | |
| # elif agent_type == 'mnemonic_creation': | |
| # response = self.mnemonic_agent.process_query(query) | |
| # elif agent_type == 'viva_practice': | |
| # response = self.viva_agent.process_query(query) | |
| # else: # academic_query or default | |
| # response = self.academic_agent.process_query(query) | |
| # agent_type = 'academic_query' | |
| # # Add metadata to response | |
| # if isinstance(response, dict): | |
| # response['agent_type'] = agent_type | |
| # return response | |
| # else: | |
| # return { | |
| # 'message': response, | |
| # 'agent_type': agent_type, | |
| # 'success': True | |
| # } | |
| # except Exception as e: | |
| # return { | |
| # 'message': f"Router Error: {str(e)}", | |
| # 'agent_type': 'error', | |
| # 'success': False | |
| # } | |
| # # """ | |
| # # Router Agent - The Coordinator | |
| # # Classifies user queries and routes them to appropriate specialist agents | |
| # # """ | |
| # # import re | |
| # # from .academic_agent import AcademicAgent | |
| # # from .drug_info_agent import DrugInfoAgent | |
| # # from .quiz_agent import QuizAgent | |
| # # from .mnemonic_agent import MnemonicAgent | |
| # # from .viva_agent import VivaAgent | |
| # # class RouterAgent: | |
| # # def __init__(self): | |
| # # # Initialize specialist agents | |
| # # self.academic_agent = AcademicAgent() | |
| # # self.drug_info_agent = DrugInfoAgent() | |
| # # self.quiz_agent = QuizAgent() | |
| # # self.mnemonic_agent = MnemonicAgent() | |
| # # self.viva_agent = VivaAgent() | |
| # # # Define keywords for each agent type (Free-tier friendly classification) | |
| # # self.agent_keywords = { | |
| # # 'drug_info': [ | |
| # # 'drug', 'medicine', 'medication', 'side effects', 'dosage', | |
| # # 'contraindication', 'interaction', 'pharmacology', 'therapeutic', | |
| # # 'adverse', 'mechanism', 'action', 'indication', 'prescription' | |
| # # ], | |
| # # 'quiz_generation': [ | |
| # # 'quiz', 'test', 'questions', 'mcq', 'multiple choice', | |
| # # 'flashcard', 'practice', 'exam', 'assessment', 'evaluate' | |
| # # ], | |
| # # 'mnemonic_creation': [ | |
| # # 'mnemonic', 'remember', 'memory', 'trick', 'acronym', | |
| # # 'rhyme', 'shortcut', 'memorize', 'recall', 'aide' | |
| # # ], | |
| # # 'viva_practice': [ | |
| # # 'viva', 'oral', 'interview', 'practice session', 'mock', | |
| # # 'question answer', 'preparation', 'rehearse' | |
| # # ] | |
| # # } | |
| # # def classify_query(self, query): | |
| # # """ | |
| # # Classify the user query into one of the agent categories | |
| # # Uses keyword matching for free-tier efficiency | |
| # # """ | |
| # # query_lower = query.lower() | |
| # # # Count keyword matches for each agent type | |
| # # scores = {} | |
| # # for agent_type, keywords in self.agent_keywords.items(): | |
| # # score = sum(1 for keyword in keywords if keyword in query_lower) | |
| # # scores[agent_type] = score | |
| # # # Special pattern matching for better accuracy | |
| # # if re.search(r'\b(what|explain|definition|mechanism|process|how does)\b', query_lower): | |
| # # scores['drug_info'] += 1 if any(drug_word in query_lower for drug_word in ['drug', 'medicine', 'pharmacology']) else 0 | |
| # # scores.setdefault('academic_query', 0) | |
| # # scores['academic_query'] += 1 | |
| # # if re.search(r'\b(create|make|generate|give me)\s+(quiz|questions|mcq)\b', query_lower): | |
| # # scores['quiz_generation'] += 2 | |
| # # if re.search(r'\b(help.*remember|memory.*trick|mnemonic.*for)\b', query_lower): | |
| # # scores['mnemonic_creation'] += 2 | |
| # # if re.search(r'\b(practice.*viva|mock.*interview|oral.*exam)\b', query_lower): | |
| # # scores['viva_practice'] += 2 | |
| # # # Find the highest scoring agent type | |
| # # if max(scores.values()) == 0: | |
| # # return 'academic_query' # Default to academic for general questions | |
| # # return max(scores, key=scores.get) | |
| # # def route_query(self, query): | |
| # # """ | |
| # # Route the query to the appropriate specialist agent | |
| # # """ | |
| # # agent_type = self.classify_query(query) | |
| # # try: | |
| # # if agent_type == 'drug_info': | |
| # # response = self.drug_info_agent.process_query(query) | |
| # # elif agent_type == 'quiz_generation': | |
| # # response = self.quiz_agent.process_query(query) | |
| # # elif agent_type == 'mnemonic_creation': | |
| # # response = self.mnemonic_agent.process_query(query) | |
| # # elif agent_type == 'viva_practice': | |
| # # response = self.viva_agent.process_query(query) | |
| # # else: # academic_query or default | |
| # # response = self.academic_agent.process_query(query) | |
| # # agent_type = 'academic_query' | |
| # # # Add metadata to response | |
| # # if isinstance(response, dict): | |
| # # response['agent_type'] = agent_type | |
| # # return response | |
| # # else: | |
| # # return { | |
| # # 'message': response, | |
| # # 'agent_type': agent_type, | |
| # # 'success': True | |
| # # } | |
| # # except Exception as e: | |
| # # return { | |
| # # 'message': f"Router Error: {str(e)}", | |
| # # 'agent_type': 'error', | |
| # # 'success': False | |
| # # } | |
| # agents/router_agent.py | |
| """ | |
| Router Agent - Directs queries to the appropriate specialist agent. | |
| """ | |
| import re | |
| # Import all specialist agents | |
| from .academic_agent import AcademicAgent | |
| from .drug_info_agent import DrugInfoAgent | |
| from .mnemonic_agent import MnemonicAgent | |
| from .quiz_agent import QuizAgent | |
| from .viva_agent import VivaAgent | |
| class RouterAgent: | |
| def __init__(self, gemini_model=None): | |
| """ | |
| Initializes the router and all specialist agents, passing the model to them. | |
| """ | |
| self.model = gemini_model | |
| # Instantiate all agents | |
| self.academic_agent = AcademicAgent(gemini_model) | |
| self.drug_info_agent = DrugInfoAgent(gemini_model) | |
| self.mnemonic_agent = MnemonicAgent(gemini_model) | |
| self.quiz_agent = QuizAgent(gemini_model) | |
| self.viva_agent = VivaAgent(gemini_model) | |
| self.default_agent = self.academic_agent | |
| def route_query(self, query: str, file_context: str = "", viva_state: dict = None, chat_history: list = None): | |
| """ | |
| Determines the user's intent and routes the query to the correct agent. | |
| Args: | |
| query (str): The user's input query. | |
| file_context (str): Text content from any uploaded files. | |
| viva_state (dict): The current state of the viva session. | |
| Returns: | |
| dict: The response dictionary from the selected agent. | |
| """ | |
| query_lower = query.lower() | |
| # --- Intent Detection Logic --- | |
| # 1. Viva Agent: High priority to catch session-based commands | |
| # If a viva session is active, or user wants to start/end one. | |
| if viva_state and viva_state.get('active'): | |
| # The VivaAgent itself handles all logic when a session is active | |
| return self.viva_agent.process_query(query, file_context, viva_state) | |
| if any(cmd in query_lower for cmd in ["viva", "interview"]): | |
| return self.viva_agent.process_query(query, file_context, viva_state) | |
| # 2. Mnemonic Agent | |
| if any(cmd in query_lower for cmd in ["mnemonic", "memory aid", "remember"]): | |
| return self.mnemonic_agent.process_query(query, file_context) | |
| # 3. Quiz Agent | |
| if any(cmd in query_lower for cmd in ["quiz", "test me", "flashcard"]): | |
| return self.quiz_agent.process_query(query, file_context) | |
| # 4. Drug Info Agent | |
| # Uses keywords and also checks for common drug endings like 'ol', 'in', 'am' | |
| if any(cmd in query_lower for cmd in ["drug", "medicine", "medication", "side effect", "dosage", "interaction"]): | |
| return self.drug_info_agent.process_query(query, file_context) | |
| if re.search(r'\b(paracetamol|ibuprofen|metformin|aspirin|amoxicillin)\b', query_lower): | |
| return self.drug_info_agent.process_query(query, file_context) | |
| # 5. Default to Academic Agent | |
| # If no other intent is detected, it's likely a general academic question. | |
| return self.academic_agent.process_query(query, file_context,chat_history) |