Upload backend/core/chatbot.py with huggingface_hub

backend/core/chatbot.py

@@ -0,0 +1,349 @@
+"""
+Chatbot with ML-based intent classification for natural language queries.
+"""
+import re
+import unicodedata
+from typing import Dict, List, Tuple, Any, Optional
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.naive_bayes import MultinomialNB
+from sklearn.pipeline import Pipeline
+import numpy as np
+from .models import Procedure, Fine, Office, Advisory
+from .search_ml import search_with_ml, expand_query_with_synonyms
+
+
+# Training data for intent classification
+INTENT_TRAINING_DATA = {
+    "search_fine": [
+        "mức phạt", "phạt bao nhiêu", "tiền phạt", "vi phạm giao thông",
+        "vượt đèn đỏ", "nồng độ cồn", "không đội mũ bảo hiểm",
+        "mức phạt là gì", "phạt như thế nào", "hành vi vi phạm",
+        "điều luật", "nghị định", "mức xử phạt"
+    ],
+    "search_procedure": [
+        "thủ tục", "làm thủ tục", "hồ sơ", "điều kiện",
+        "thủ tục cư trú", "thủ tục ANTT", "thủ tục PCCC",
+        "cần giấy tờ gì", "làm như thế nào", "quy trình",
+        "thời hạn", "lệ phí", "nơi nộp"
+    ],
+    "search_office": [
+        "địa chỉ", "điểm tiếp dân", "công an", "phòng ban",
+        "số điện thoại", "giờ làm việc", "nơi tiếp nhận",
+        "đơn vị nào", "ở đâu", "liên hệ"
+    ],
+    "search_advisory": [
+        "cảnh báo", "lừa đảo", "scam", "thủ đoạn",
+        "cảnh giác", "an toàn", "bảo mật"
+    ],
+    "general_query": [
+        "xin chào", "giúp tôi", "tư vấn", "hỏi",
+        "thông tin", "tra cứu", "tìm kiếm"
+    ]
+}
+
+# Response templates
+RESPONSE_TEMPLATES = {
+    "search_fine": "Tôi tìm thấy {count} mức phạt liên quan đến '{query}':",
+    "search_procedure": "Tôi tìm thấy {count} thủ tục liên quan đến '{query}':",
+    "search_office": "Tôi tìm thấy {count} đơn vị liên quan đến '{query}':",
+    "search_advisory": "Tôi tìm thấy {count} cảnh báo liên quan đến '{query}':",
+    "general_query": "Tôi có thể giúp bạn tra cứu thông tin về thủ tục, mức phạt, đơn vị hoặc cảnh báo. Bạn muốn tìm gì?",
+    "no_results": "Xin lỗi, tôi không tìm thấy thông tin liên quan đến '{query}'. Vui lòng thử lại với từ khóa khác.",
+    "greeting": "Xin chào! Tôi có thể giúp bạn tra cứu thông tin về thủ tục hành chính, mức phạt giao thông, danh bạ đơn vị và cảnh báo an ninh. Bạn cần tìm gì?",
+}
+
+
+class Chatbot:
+    def __init__(self):
+        self.intent_classifier = None
+        self.vectorizer = None
+        self._train_classifier()
+    
+    def _train_classifier(self):
+        """Train intent classification model."""
+        try:
+            # Prepare training data
+            texts = []
+            labels = []
+            
+            for intent, examples in INTENT_TRAINING_DATA.items():
+                for example in examples:
+                    texts.append(self._preprocess_text(example))
+                    labels.append(intent)
+            
+            if not texts:
+                return
+            
+            # Create and train pipeline
+            self.intent_classifier = Pipeline([
+                ('tfidf', TfidfVectorizer(
+                    analyzer='word',
+                    ngram_range=(1, 2),
+                    min_df=1,
+                    lowercase=True,
+                    token_pattern=r'\b\w+\b'
+                )),
+                ('clf', MultinomialNB())
+            ])
+            
+            self.intent_classifier.fit(texts, labels)
+        except Exception as e:
+            print(f"Error training classifier: {e}")
+            self.intent_classifier = None
+    
+    def _preprocess_text(self, text: str) -> str:
+        """Preprocess text for classification - keep Vietnamese characters."""
+        if not text:
+            return ""
+        text = text.lower().strip()
+        # Only remove punctuation marks, keep all letters (including Vietnamese) and numbers
+        # Remove: !"#$%&'()*+,-./:;<=>?@[\]^_`{|}~
+        text = re.sub(r'[!"#$%&\'()*+,\-./:;<=>?@\[\\\]^_`{|}~]', ' ', text)
+        text = re.sub(r'\s+', ' ', text)
+        return text.strip()
+
+    def _remove_accents(self, text: str) -> str:
+        """Remove diacritics for accent-insensitive matching."""
+        if not text:
+            return ""
+        normalized = unicodedata.normalize("NFD", text)
+        return "".join(ch for ch in normalized if unicodedata.category(ch) != "Mn")
+
+    def _keyword_in(self, query_lower: str, query_ascii: str, keyword: str) -> bool:
+        """Check keyword presence in either original or accent-free text."""
+        kw_lower = keyword.lower()
+        if kw_lower in query_lower:
+            return True
+        kw_ascii = self._remove_accents(kw_lower)
+        return kw_ascii in query_ascii
+    
+    def classify_intent(self, query: str) -> Tuple[str, float]:
+        """
+        Classify user intent from query.
+        Returns (intent, confidence_score)
+        """
+        # Use keyword-based classification first (more reliable for Vietnamese)
+        keyword_intent, keyword_confidence = self._keyword_based_intent(query)
+        
+        # ALWAYS use keyword-based for now (more reliable for Vietnamese)
+        # Special handling for greeting - only if really simple
+        if keyword_intent == "greeting":
+            query_lower = query.lower().strip()
+            query_ascii = self._remove_accents(query_lower)
+            query_words = query_lower.split()
+            # Double-check: if query has fine keywords, it's NOT a greeting
+            fine_indicators = ["phạt", "mức", "vuot", "vượt", "đèn", "den", "vi phạm", "vi pham"]
+            if any(self._keyword_in(query_lower, query_ascii, indicator) for indicator in fine_indicators):
+                # Re-check with fine keywords
+                for kw in ["mức phạt", "vi phạm", "đèn đỏ", "vượt đèn", "muc phat", "vuot den", "phat", "vuot", "den", "muc"]:
+                    if self._keyword_in(query_lower, query_ascii, kw):
+                        return ("search_fine", 0.9)
+            # Only return greeting if query is very short (<= 3 words)
+            if len(query_words) > 3:
+                # If long query classified as greeting, it's probably wrong - use general
+                return ("general_query", 0.5)
+        
+        # For all other intents, use keyword-based result
+        return (keyword_intent, max(keyword_confidence, 0.8))
+    
+    def _keyword_based_intent(self, query: str) -> Tuple[str, float]:
+        """Fallback keyword-based intent classification."""
+        # Use original query (lowercase) to preserve Vietnamese characters
+        query_lower = query.lower().strip()
+        query_ascii = self._remove_accents(query_lower)
+        query_words = query_lower.split()
+        
+        # Check for keywords - prioritize fine-related queries FIRST
+        # Check on original query to preserve Vietnamese characters
+        # Check longer phrases first, then single words
+        fine_keywords = ["mức phạt", "vi phạm", "đèn đỏ", "nồng độ cồn", "mũ bảo hiểm", "tốc độ", "bằng lái", "vượt đèn", "mức phạt vượt"]
+        fine_keywords_ascii = [self._remove_accents(kw) for kw in fine_keywords]
+        fine_single_words = ["phạt", "vượt", "đèn", "mức", "phat", "vuot", "den"]
+        
+        # Check multi-word keywords first
+        has_fine_keywords = False
+        for kw, kw_ascii in zip(fine_keywords, fine_keywords_ascii):
+            if self._keyword_in(query_lower, query_ascii, kw) or kw_ascii in query_ascii:
+                return ("search_fine", 0.95)  # Very high confidence
+        # Then check single words - check ALL of them, not just first match
+        for kw in fine_single_words:
+            if self._keyword_in(query_lower, query_ascii, kw):
+                has_fine_keywords = True
+                # Return immediately if found
+                return ("search_fine", 0.9)
+        
+        has_procedure_keywords = any(
+            self._keyword_in(query_lower, query_ascii, kw) for kw in
+            ["thủ tục", "hồ sơ", "điều kiện", "cư trú", "antt", "pccc", "thu tuc", "ho so", "dieu kien", "cu tru"]
+        )
+        if has_procedure_keywords:
+            return ("search_procedure", 0.8)
+        
+        has_office_keywords = any(
+            self._keyword_in(query_lower, query_ascii, kw) for kw in
+            ["địa chỉ", "điểm tiếp dân", "công an", "số điện thoại", "giờ làm việc", "dia chi", "diem tiep dan", "cong an", "so dien thoai", "gio lam viec"]
+        )
+        if has_office_keywords:
+            return ("search_office", 0.8)
+        
+        has_advisory_keywords = any(
+            self._keyword_in(query_lower, query_ascii, kw) for kw in
+            ["cảnh báo", "lừa đảo", "scam", "canh bao", "lua dao"]
+        )
+        if has_advisory_keywords:
+            return ("search_advisory", 0.8)
+        
+        # Only treat as greeting if it's VERY short (<= 3 words) and ONLY contains greeting words
+        # AND does NOT contain any other keywords
+        has_any_keyword = (has_fine_keywords or has_procedure_keywords or 
+                          has_office_keywords or has_advisory_keywords)
+        
+        if (len(query_words) <= 3 and 
+            any(self._keyword_in(query_lower, query_ascii, kw) for kw in ["xin chào", "chào", "hello", "hi", "xin chao", "chao"]) and
+            not has_any_keyword):
+            return ("greeting", 0.9)
+        
+        return ("general_query", 0.5)
+    
+    def extract_keywords(self, query: str) -> List[str]:
+        """Extract keywords from query for search."""
+        # Remove common stopwords
+        stopwords = {"là", "gì", "bao nhiêu", "như thế nào", "ở đâu", "của", "và", "hoặc", "tôi", "bạn"}
+        
+        words = re.findall(r'\b\w+\b', query.lower())
+        keywords = [w for w in words if w not in stopwords and len(w) > 2]
+        
+        return keywords
+    
+    def search_by_intent(self, intent: str, query: str, limit: int = 5) -> Dict[str, Any]:
+        """Search based on classified intent."""
+        # Use original query for better matching, especially for Vietnamese text
+        keywords = query.strip()
+        # Also try with extracted keywords as fallback
+        extracted = " ".join(self.extract_keywords(query))
+        if extracted and len(extracted) > 2:
+            keywords = f"{keywords} {extracted}"
+        
+        results = []
+        
+        if intent == "search_fine":
+            qs = Fine.objects.all()
+            text_fields = ["name", "code", "article", "decree", "remedial"]
+            search_results = search_with_ml(qs, keywords, text_fields, top_k=limit, min_score=0.1)
+            results = [{"type": "fine", "data": {
+                "id": f.id,
+                "name": f.name,
+                "code": f.code,
+                "min_fine": float(f.min_fine) if f.min_fine else None,
+                "max_fine": float(f.max_fine) if f.max_fine else None,
+                "article": f.article,
+                "decree": f.decree,
+            }} for f in search_results]
+        
+        elif intent == "search_procedure":
+            qs = Procedure.objects.all()
+            text_fields = ["title", "domain", "conditions", "dossier"]
+            search_results = search_with_ml(qs, keywords, text_fields, top_k=limit, min_score=0.1)
+            results = [{"type": "procedure", "data": {
+                "id": p.id,
+                "title": p.title,
+                "domain": p.domain,
+                "level": p.level,
+            }} for p in search_results]
+        
+        elif intent == "search_office":
+            qs = Office.objects.all()
+            text_fields = ["unit_name", "address", "district", "service_scope"]
+            search_results = search_with_ml(qs, keywords, text_fields, top_k=limit, min_score=0.1)
+            results = [{"type": "office", "data": {
+                "id": o.id,
+                "unit_name": o.unit_name,
+                "address": o.address,
+                "district": o.district,
+                "phone": o.phone,
+                "working_hours": o.working_hours,
+            }} for o in search_results]
+        
+        elif intent == "search_advisory":
+            qs = Advisory.objects.all()
+            text_fields = ["title", "summary"]
+            search_results = search_with_ml(qs, keywords, text_fields, top_k=limit, min_score=0.1)
+            results = [{"type": "advisory", "data": {
+                "id": a.id,
+                "title": a.title,
+                "summary": a.summary,
+            }} for a in search_results]
+        
+        return {
+            "intent": intent,
+            "query": query,
+            "keywords": keywords,
+            "results": results,
+            "count": len(results)
+        }
+    
+    def generate_response(self, query: str) -> Dict[str, Any]:
+        """
+        Generate chatbot response for user query.
+        Returns dict with message, intent, and results.
+        """
+        query = query.strip()
+        
+        # Classify intent FIRST
+        intent, confidence = self.classify_intent(query)
+        
+        # Only handle greetings if it's REALLY a simple greeting (very short, no other keywords)
+        query_lower = query.lower().strip()
+        query_words = query_lower.split()
+        
+        # Check if it contains keywords that indicate it's NOT a greeting
+        has_fine_keywords = any(kw in query_lower for kw in ["phạt", "mức phạt", "vi phạm", "đèn đỏ", "nồng độ cồn", "mũ bảo hiểm", "tốc độ", "vượt"])
+        has_procedure_keywords = any(kw in query_lower for kw in ["thủ tục", "hồ sơ", "điều kiện", "cư trú", "antt", "pccc"])
+        has_office_keywords = any(kw in query_lower for kw in ["địa chỉ", "công an", "số điện thoại", "giờ làm việc"])
+        has_advisory_keywords = any(kw in query_lower for kw in ["cảnh báo", "lừa đảo", "scam"])
+        
+        # Only treat as greeting if it's very short AND has no other keywords AND classified as greeting
+        is_simple_greeting = (len(query_words) <= 3 and 
+                             any(greeting in query_lower for greeting in ["xin chào", "chào", "hello", "hi"]) and
+                             not (has_fine_keywords or has_procedure_keywords or has_office_keywords or has_advisory_keywords))
+        
+        if is_simple_greeting and intent == "greeting":
+            return {
+                "message": RESPONSE_TEMPLATES["greeting"],
+                "intent": "greeting",
+                "results": [],
+                "count": 0
+            }
+        
+        # Search based on intent
+        search_result = self.search_by_intent(intent, query, limit=5)
+        
+        # Generate response message
+        if search_result["count"] > 0:
+            template = RESPONSE_TEMPLATES.get(intent, RESPONSE_TEMPLATES["general_query"])
+            message = template.format(
+                count=search_result["count"],
+                query=query
+            )
+        else:
+            message = RESPONSE_TEMPLATES["no_results"].format(query=query)
+        
+        return {
+            "message": message,
+            "intent": intent,
+            "confidence": confidence,
+            "results": search_result["results"],
+            "count": search_result["count"]
+        }
+
+
+# Global chatbot instance
+_chatbot_instance = None
+
+def get_chatbot() -> Chatbot:
+    """Get or create chatbot instance."""
+    global _chatbot_instance
+    if _chatbot_instance is None:
+        _chatbot_instance = Chatbot()
+    return _chatbot_instance
+