File size: 11,037 Bytes
1f0be7e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 | 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()
}
|