import os import warnings from typing import Dict, Tuple # Suppress deprecation warning for google.generativeai warnings.filterwarnings("ignore", category=FutureWarning, module="google.generativeai") import google.generativeai as genai from backend.memory import ConversationMemory from backend.knowledge_base import KnowledgeBase from backend.intent_detector import IntentDetector from backend.lead_scorer import LeadScorer from backend.storage import Storage class ConversationEngine: """Main conversation orchestration engine.""" def __init__(self): """Initialize conversation engine.""" self.api_key = os.getenv("GEMINI_API_KEY") if self.api_key: genai.configure(api_key=self.api_key) self.model = genai.GenerativeModel("gemini-2.5-flash") if self.api_key else None self.memory = ConversationMemory(max_messages=10) self.knowledge_base = KnowledgeBase("data/faq.json") self.intent_detector = IntentDetector(api_key=self.api_key) self.lead_scorer = LeadScorer() self.storage = Storage("data") # Flow state self.current_intent = "unknown" self.lead_data = {} self.support_data = {} self.lead_questions = [ "What type of business are you in or what industry?", "What's the main problem you're trying to solve?", "What's your approximate budget range?", "What's the best email to reach you?" ] self.lead_question_index = 0 self.support_questions = [ "Could you describe the issue you're experiencing?", "Do you have an order ID or reference number?" ] self.support_question_index = 0 def process_message(self, user_message: str) -> Tuple[str, Dict]: """ Process user message and generate response. Args: user_message: User's input Returns: Tuple of (response, metadata) """ # Add to memory self.memory.add_message("user", user_message) # Detect intent intent, confidence = self.intent_detector.detect_intent(user_message) if self.current_intent == "unknown" and confidence > 0.6: self.current_intent = intent self.memory.set_intent(intent) # Extract name if not already extracted if not self.memory.user_name: name = self.intent_detector.extract_name(user_message) if name: self.memory.set_user_name(name) # Route to appropriate handler if self.current_intent == "lead": response, metadata = self._handle_lead_flow(user_message) elif self.current_intent == "support": response, metadata = self._handle_support_flow(user_message) else: response, metadata = self._handle_general_flow(user_message) # Add response to memory self.memory.add_message("assistant", response) return response, metadata def _handle_lead_flow(self, user_message: str) -> Tuple[str, Dict]: """Handle lead qualification flow.""" metadata = {"intent": "lead", "type": ""} # Try to extract email email = self.intent_detector.extract_email(user_message) if email: self.lead_data["email"] = email # Ask sequential questions if self.lead_question_index < len(self.lead_questions): question = self.lead_questions[self.lead_question_index] self.lead_question_index += 1 # Store answer if self.lead_question_index == 1: self.lead_data["business_type"] = user_message metadata["type"] = "business_info" elif self.lead_question_index == 2: self.lead_data["problem"] = user_message metadata["type"] = "problem_info" elif self.lead_question_index == 3: self.lead_data["budget"] = user_message metadata["type"] = "budget_info" elif self.lead_question_index == 4: if email: self.lead_data["email"] = email metadata["type"] = "contact_info" # Check if qualification complete if self.lead_question_index >= len(self.lead_questions): response = self._finalize_lead() metadata["lead_finalized"] = True # Save lead self.lead_data["conversation"] = self.memory.get_context() self.lead_data["user_name"] = self.memory.user_name lead_score = self.lead_scorer.score_lead(self.memory.get_context(), self.lead_data) self.lead_data["scoring"] = lead_score lead_id = self.storage.save_lead(self.lead_data) metadata["lead_id"] = lead_id else: response = f"{question}" return response, metadata else: response = f"{self.lead_questions[-1]}" return response, metadata def _handle_support_flow(self, user_message: str) -> Tuple[str, Dict]: """Handle support ticket flow.""" metadata = {"intent": "support", "type": ""} # Ask support questions if self.support_question_index < len(self.support_questions): question = self.support_questions[self.support_question_index] self.support_question_index += 1 if self.support_question_index == 1: self.support_data["issue_description"] = user_message metadata["type"] = "issue_info" elif self.support_question_index == 2: self.support_data["reference_id"] = user_message metadata["type"] = "reference_info" if self.support_question_index >= len(self.support_questions): response = self._finalize_support() metadata["ticket_finalized"] = True # Save support ticket self.support_data["conversation"] = self.memory.get_context() self.support_data["user_name"] = self.memory.user_name ticket_id = self.storage.save_support_ticket(self.support_data) metadata["ticket_id"] = ticket_id else: response = f"{question}" return response, metadata else: response = f"{self.support_questions[-1]}" return response, metadata def _handle_general_flow(self, user_message: str) -> Tuple[str, Dict]: """Handle general queries with FAQ and LLM.""" metadata = {"intent": "general", "used_faq": False} # Try to answer from FAQ faq_result = self.knowledge_base.retrieve_answer(user_message) if faq_result: metadata["used_faq"] = True return faq_result["answer"], metadata # Generate response using Gemini if self.model: return self._generate_with_gemini(user_message, metadata) else: return self._generate_grounded_response(user_message, metadata) def _generate_with_gemini(self, user_message: str, metadata: Dict) -> Tuple[str, Dict]: """Generate response using Gemini API.""" try: context = self.memory.get_formatted_context() user_name = f" {self.memory.user_name}" if self.memory.user_name else "" prompt = f"""You are a helpful and professional customer support AI assistant for a website support + lead qualification service. IMPORTANT RULES: 1. Be helpful, friendly, and natural 2. NEVER make up information or hallucinate details 3. If you don't know something, say "I don't have that information, but I can connect you with our support team." 4. Keep answers concise (2-3 sentences max for general queries) 5. If the user seems interested in your services, ask about their needs User name: {user_name if user_name else "Visitor"} Previous conversation: {context} Current user message: {user_message} Respond conversationally and helpfully. Keep your response brief and natural.""" response = self.model.generate_content(prompt, safety_settings=[ {"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_NONE"}, {"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_NONE"}, {"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "BLOCK_NONE"}, {"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "BLOCK_NONE"}, ]) return response.text, metadata except Exception as e: print(f"Error generating response: {e}") return self._generate_grounded_response(user_message, metadata) def _generate_grounded_response(self, user_message: str, metadata: Dict) -> Tuple[str, Dict]: """Generate grounded response without API.""" response = "Thank you for your message. To better assist you, could you tell me more about what you're looking for? Are you interested in our services, or do you have a support question?" return response, metadata def _finalize_lead(self) -> str: """Generate lead finalization message.""" user_name = self.memory.user_name or "there" return f"Thank you for the information! I've captured your details. One of our specialists will reach out to you shortly to discuss how we can help your {self.lead_data.get('business_type', 'business')} succeed. We appreciate your interest!" def _finalize_support(self) -> str: """Generate support ticket finalization message.""" user_name = self.memory.user_name or "there" return f"Thank you for providing those details. Your support ticket has been created and our team will investigate your issue. You'll receive an update shortly via email. We appreciate your patience!" def reset(self): """Reset conversation state for new session.""" self.memory.clear() self.current_intent = "unknown" self.lead_data = {} self.support_data = {} self.lead_question_index = 0 self.support_question_index = 0 def get_debug_info(self) -> Dict: """Get debug information about current session.""" return { "current_intent": self.current_intent, "user_name": self.memory.user_name, "lead_data": self.lead_data, "support_data": self.support_data, "message_count": len(self.memory.get_context()), "memory_summary": self.memory.get_summary() }