""" Conversational Voice Bot Service Makes the bot more interactive by asking for clarification when context is unclear """ import logging from typing import Dict, Any, List, Optional import re logger = logging.getLogger("voicebot") class ConversationalService: def __init__(self): self.conversation_history = {} def analyze_query_clarity(self, query: str, search_results: List[Dict[str, Any]]) -> Dict[str, Any]: """ Analyze if the query is clear enough or needs clarification Returns clarification questions if needed """ query_lower = query.lower().strip() # Remove filler words filler_words = ['um', 'uh', 'like', 'you know', 'actually', 'basically'] clean_query = query_lower for filler in filler_words: clean_query = clean_query.replace(filler, '') clarity_analysis = { 'is_clear': True, 'confidence': 1.0, 'clarification_needed': False, 'clarification_questions': [], 'suggested_responses': [], 'query_type': 'specific' } # Check for vague queries vague_patterns = [ r'\b(what|how|tell me|explain)\s+(about|regarding)?\s*$', r'^(help|info|information)$', r'^(what|how)$', r'\b(anything|something|stuff)\b', r'^(yes|no|okay|ok)$', r'\b(this|that|it)\b.*\?$' ] is_vague = any(re.search(pattern, clean_query) for pattern in vague_patterns) # Check for ambiguous pension queries if 'pension' in query_lower: pension_ambiguity = self._check_pension_ambiguity(query_lower) if pension_ambiguity['needs_clarification']: clarity_analysis.update(pension_ambiguity) return clarity_analysis else: # Pension query is clear - don't ask for clarification clarity_analysis.update({ 'is_clear': True, 'confidence': 0.9, 'clarification_needed': False, 'query_type': 'pension_clear' }) return clarity_analysis # Check for ambiguous procurement queries elif any(word in query_lower for word in ['tender', 'procurement', 'bid']): procurement_ambiguity = self._check_procurement_ambiguity(query_lower) if procurement_ambiguity['needs_clarification']: clarity_analysis.update(procurement_ambiguity) return clarity_analysis # Check for too generic queries elif is_vague or len(clean_query.split()) <= 2: clarity_analysis.update({ 'is_clear': False, 'confidence': 0.3, 'clarification_needed': True, 'query_type': 'vague', 'clarification_questions': [ "I'd be happy to help! Could you be more specific about what you're looking for?", "Are you asking about:", "• Pension rules and benefits?", "• Leave policies and applications?", "• Procurement and tender processes?", "• Salary and allowances?", "• Or something else?" ], 'suggested_responses': [ "Please let me know which topic interests you most, and I'll provide detailed information." ] }) return clarity_analysis # Check if search results are relevant if search_results: relevance_score = self._calculate_relevance_score(query_lower, search_results) if relevance_score < 0.1: # Low relevance - made much more permissive clarity_analysis.update({ 'is_clear': False, 'confidence': relevance_score, 'clarification_needed': True, 'query_type': 'low_relevance', 'clarification_questions': [ f"I found some information, but I want to make sure I understand your question correctly.", f"When you asked '{query}', did you mean:", "• Something specific about government policies?", "• A particular process or procedure?", "• Information for a specific situation?", "", "Could you provide a bit more context about what you're trying to accomplish?" ] }) return clarity_analysis def _check_pension_ambiguity(self, query: str) -> Dict[str, Any]: """Check if pension query needs clarification""" # Specific pension queries that are clear clear_pension_terms = [ 'pension', 'retirement', 'gratuity', 'superannuation', 'pension calculation', 'pension formula', 'pension eligibility', 'pension application', 'family pension', 'commutation', 'pension documents', 'pension process', 'pension rules', 'pension increment', 'pension impact', 'old age', 'benefits' ] if any(term in query for term in clear_pension_terms): return {'needs_clarification': False} # Disable overly aggressive clarification for pension queries # Users asking about "pension rules" should get direct answers # generic_pension_patterns = [ # r'^pension\??$', # r'^what.*pension\??$', # r'^tell me.*pension\??$', # r'^pension.*rules?\??$', # r'^how.*pension\??$' # ] # Always provide direct answers for pension-related queries # if any(re.search(pattern, query) for pattern in generic_pension_patterns): # return clarification logic - DISABLED to provide direct answers return {'needs_clarification': False} def _check_procurement_ambiguity(self, query: str) -> Dict[str, Any]: """Check if procurement query needs clarification""" clear_procurement_terms = [ 'tender process', 'bid submission', 'gem portal', 'msme benefits', 'vendor registration', 'procurement threshold', 'tender documents' ] if any(term in query for term in clear_procurement_terms): return {'needs_clarification': False} generic_procurement_patterns = [ r'^tender\??$', r'^procurement\??$', r'^bid\??$', r'^what.*tender\??$', r'^how.*procurement\??$' ] if any(re.search(pattern, query) for pattern in generic_procurement_patterns): return { 'needs_clarification': True, 'is_clear': False, 'confidence': 0.4, 'clarification_needed': True, 'query_type': 'procurement_generic', 'clarification_questions': [ "I can help with procurement and tendering! To provide the right information, could you clarify:", "", "🏢 **What specifically about procurement?**", "• How to participate in tenders?", "• Procurement rules and thresholds?", "• GeM portal registration?", "• MSME benefits in procurement?", "• Vendor empanelment process?", "• Bid preparation and submission?", "", "Are you a vendor looking to participate, or an officer managing procurement?" ], 'suggested_responses': [ "Please specify your role and what aspect of procurement you need help with." ] } return {'needs_clarification': False} def _calculate_relevance_score(self, query: str, search_results: List[Dict[str, Any]]) -> float: """Calculate how relevant search results are to the query""" if not search_results: return 0.0 query_words = set(query.lower().split()) query_words = {word for word in query_words if len(word) > 2} # Remove short words total_relevance = 0.0 for result in search_results: content = result.get('content', '').lower() filename = result.get('filename', '').lower() # Count query word matches in content content_matches = sum(1 for word in query_words if word in content) filename_matches = sum(1 for word in query_words if word in filename) # Calculate relevance score for this result max_possible_matches = len(query_words) if max_possible_matches > 0: relevance = (content_matches + filename_matches * 2) / (max_possible_matches * 2) total_relevance += relevance # Average relevance across all results return total_relevance / len(search_results) if search_results else 0.0 def generate_conversational_response(self, query: str, search_results: List[Dict[str, Any]], session_id: str = None) -> Dict[str, Any]: """ Generate a conversational response that asks for clarification when needed """ clarity = self.analyze_query_clarity(query, search_results) if clarity['clarification_needed']: # Generate clarification response clarification_text = "\n".join(clarity['clarification_questions']) response = { 'needs_clarification': True, 'response': clarification_text, 'type': 'clarification_request', 'suggested_follow_ups': clarity.get('suggested_responses', []), 'query_type': clarity.get('query_type', 'unclear'), 'confidence': clarity.get('confidence', 0.5) } # Store conversation context for follow-up if session_id: self.conversation_history[session_id] = { 'last_query': query, 'awaiting_clarification': True, 'clarification_type': clarity.get('query_type', 'unclear'), 'timestamp': __import__('time').time() } return response else: # Query is clear, proceed with normal response return { 'needs_clarification': False, 'response': None, # Will be filled by normal RAG processing 'type': 'information_request', 'confidence': clarity.get('confidence', 1.0) } def handle_follow_up(self, query: str, session_id: str) -> Dict[str, Any]: """Handle follow-up queries after clarification request""" if session_id not in self.conversation_history: return {'is_follow_up': False} context = self.conversation_history[session_id] if not context.get('awaiting_clarification', False): return {'is_follow_up': False} # Check if this is a clarification response query_lower = query.lower().strip() # Clear clarification state context['awaiting_clarification'] = False # Enhanced query based on clarification clarification_type = context.get('clarification_type', '') original_query = context.get('last_query', '') enhanced_query = f"{original_query} {query}" return { 'is_follow_up': True, 'enhanced_query': enhanced_query, 'context_type': clarification_type, 'original_query': original_query } def generate_friendly_greeting(self) -> str: """Generate a friendly greeting that encourages conversation""" greetings = [ "Hello! I'm here to help you with government policies and procedures. What would you like to know about?", "Hi there! I can assist you with information about pensions, procurement, leave policies, and more. What's on your mind?", "Welcome! I'm your government assistant. Feel free to ask me about any policies, rules, or procedures you need help with.", "Hello! I'm ready to help you navigate government policies and processes. What information are you looking for today?" ] import random return random.choice(greetings) def generate_helpful_response(self, response_text: str, sources: List[Dict[str, Any]]) -> str: """Make responses more helpful and conversational""" # Add conversational elements if response_text: # Add follow-up encouragement follow_up_phrases = [ "\n\nIs there anything specific about this topic you'd like me to explain further?", "\n\nDo you have any follow-up questions about this information?", "\n\nWould you like me to provide more details on any particular aspect?", "\n\nIs there a specific situation you're dealing with that I can help you navigate?" ] import random follow_up = random.choice(follow_up_phrases) response_text += follow_up return response_text # Global instance conversational_service = ConversationalService()