Spaces:
Running
Running
File size: 6,907 Bytes
75033ed 2ecdea6 75033ed 2ecdea6 75033ed 2ecdea6 75033ed 2ecdea6 75033ed 2ecdea6 75033ed 2ecdea6 75033ed 2ecdea6 75033ed 2ecdea6 75033ed 2ecdea6 75033ed 2ecdea6 75033ed 2ecdea6 75033ed |
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 |
"""
Intent Classifier for Hybrid RAG + FSM Chatbot
Detects user intent to route between scenario flows and RAG queries
"""
from typing import Dict, Optional, List
import re
class IntentClassifier:
"""
Classify user intent using keyword matching
Routes to either:
- Scenario flows (scripted conversations)
- RAG queries (knowledge retrieval)
"""
def __init__(self, scenarios_dir: str = "scenarios"):
"""
Initialize with auto-loading triggers from scenario JSON files
Args:
scenarios_dir: Directory containing scenario JSON files
"""
# Auto-load scenario patterns from JSON files
self.scenario_patterns = self._load_scenario_patterns(scenarios_dir)
# General question patterns (RAG)
self.general_patterns = [
# Location
"ở đâu", "địa điểm", "location", "where",
"chỗ nào", "tổ chức tại",
# Time
"mấy giờ", "khi nào", "when", "time",
"bao giờ", "thời gian", "ngày nào",
# Info
"thông tin", "info", "information",
"chi tiết", "details", "về",
# Parking
"đậu xe", "parking", "gửi xe",
# Contact
"liên hệ", "contact", "số điện thoại",
# Events/content - NEW (Bug fix #3)
"sự kiện", "event", "đâu", "show nào",
"line-up", "lineup", "performer"
]
def _load_scenario_patterns(self, scenarios_dir: str) -> dict:
"""
Auto-load triggers from all scenario JSON files
Returns:
{"scenario_id": ["trigger1", "trigger2", ...]}
"""
import json
import os
patterns = {}
if not os.path.exists(scenarios_dir):
print(f"⚠ Scenarios directory not found: {scenarios_dir}")
return patterns
for filename in os.listdir(scenarios_dir):
if filename.endswith('.json'):
filepath = os.path.join(scenarios_dir, filename)
try:
with open(filepath, 'r', encoding='utf-8') as f:
scenario = json.load(f)
scenario_id = scenario.get('scenario_id')
triggers = scenario.get('triggers', [])
if scenario_id and triggers:
patterns[scenario_id] = triggers
print(f"✓ Loaded triggers for: {scenario_id} ({len(triggers)} patterns)")
except Exception as e:
print(f"⚠ Error loading {filename}: {e}")
return patterns
def classify(
self,
message: str,
conversation_state: Optional[Dict] = None
) -> str:
"""
Classify user intent with IMPROVED mid-scenario detection (Bug fix #3)
Returns:
- "scenario:{scenario_id}" - Trigger new scenario
- "scenario:continue" - Continue active scenario
- "rag:general" - General RAG query (no active scenario)
- "rag:with_resume" - RAG query mid-scenario (then resume)
"""
message_lower = message.lower().strip()
# Check if user is in active scenario
active_scenario = conversation_state.get('active_scenario') if conversation_state else None
if active_scenario:
# User is in a scenario - check if this is off-topic or continuation
# IMPROVED: Detect off-topic questions better
# Check for question words + patterns
question_indicators = ["?", "đâu", "gì", "sao", "where", "what", "how", "when"]
has_question = any(q in message_lower for q in question_indicators)
# Check if matches general patterns
matches_general = self._matches_any_pattern(message_lower, self.general_patterns)
# Short messages with questions are likely off-topic
word_count = len(message_lower.split())
is_short_question = word_count <= 4 and has_question
# Decision logic
if matches_general or is_short_question:
# User asking off-topic question → RAG with resume
print(f"🔀 Off-topic detected: '{message}' → rag:with_resume")
return "rag:with_resume"
else:
# Normal scenario continuation
return "scenario:continue"
# Not in scenario - check for scenario triggers
for scenario_id, patterns in self.scenario_patterns.items():
for pattern in patterns:
if pattern.lower() in message_lower:
return f"scenario:{scenario_id}"
# No scenario match - general RAG query
return "rag:general"
def _matches_any_pattern(self, message: str, patterns: List[str]) -> bool:
"""
Check if message matches any pattern in list
"""
for pattern in patterns:
# Simple substring match
if pattern in message:
return True
# Word boundary check
if re.search(rf'\b{re.escape(pattern)}\b', message, re.IGNORECASE):
return True
return False
def get_scenario_type(self, intent: str) -> Optional[str]:
"""
Extract scenario type from intent string
Args:
intent: "scenario:price_inquiry" or "scenario:continue"
Returns:
"price_inquiry" or None
"""
if not intent.startswith("scenario:"):
return None
parts = intent.split(":", 1)
if len(parts) < 2:
return None
scenario_type = parts[1]
if scenario_type == "continue":
return None
return scenario_type
def add_scenario_pattern(self, scenario_id: str, patterns: List[str]):
"""
Dynamically add new scenario patterns
"""
if scenario_id in self.scenario_patterns:
self.scenario_patterns[scenario_id].extend(patterns)
else:
self.scenario_patterns[scenario_id] = patterns
def add_general_pattern(self, patterns: List[str]):
"""
Dynamically add new general question patterns
"""
self.general_patterns.extend(patterns)
|