import json import logging import random import re import unicodedata from difflib import SequenceMatcher from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from app.config import settings from app.guardrail.blacklist import Blacklist from app.model.intent_model import IntentModel from app.services.gemini_service import GeminiService logger = logging.getLogger(__name__) class InferenceService: """Inference engine untuk menentukan respons chatbot.""" MODEL_NAME = "logistic_regression_tfidf" FALLBACK_RESPONSES = [ ( "Maaf, saya belum cukup yakin memahami pertanyaan Anda. " "Coba sebutkan program atau informasi yang dicari, misalnya harga, " "jadwal, persyaratan, metode kelas, atau pendaftaran." ), ( "Pertanyaan Anda masih cukup umum. Mohon tambahkan nama program " "atau kebutuhan Anda agar saya dapat memberikan jawaban yang lebih tepat." ), ( "Saya belum menemukan jawaban yang paling sesuai. " "Anda dapat memperjelas pertanyaan atau menghubungi tim Phitagoras." ), ] BLOCKED_RESPONSE = ( "Maaf, saya tidak dapat membantu permintaan tersebut. " "Saya tetap dapat membantu informasi seputar pelatihan, sertifikasi, " "pendaftaran, dan layanan Phitagoras." ) SERVICE_UNAVAILABLE_RESPONSE = ( "Maaf, layanan chatbot sedang tidak tersedia. " "Silakan coba kembali beberapa saat lagi atau hubungi tim Phitagoras." ) EXACT_MATCH_CONFIDENCE = 1.0 FUZZY_MATCH_THRESHOLD = 0.93 FUZZY_MATCH_MIN_MARGIN = 0.08 FUZZY_MATCH_MIN_LENGTH = 8 DEFAULT_CONFIDENCE_MARGIN = 0.03 TOP_K_CANDIDATES = 3 INTENT_CTA = { "sertifikasi_k3_umum": "Konsultasikan program Ahli K3 Umum", "sertifikasi_k3_konstruksi": "Konsultasikan program K3 Konstruksi", "harga_pelatihan": "Minta penawaran harga terbaru", "durasi_pelatihan_k3": "Tanyakan durasi program", "syarat_peserta_k3": "Periksa persyaratan peserta", "instruktur_k3": "Tanyakan profil instruktur", "sertifikat_berlaku": "Konsultasikan masa berlaku sertifikat", "materi_pelatihan_k3": "Minta informasi silabus", "metode_pembelajaran_k3": "Pilih metode kelas", "jaminan_kelulusan": "Tanyakan ketentuan evaluasi", "sertifikat_lanjutan": "Konsultasikan program lanjutan", "lokasi_pelatihan": "Periksa lokasi kelas", "jadwal_pelatihan": "Lihat jadwal terbaru", "pendaftaran_peserta": "Mulai proses pendaftaran", "pembayaran_pelatihan": "Hubungi tim administrasi", "kerjasama_korporat": "Minta proposal pelatihan perusahaan", "kontak_phitagoras": "Hubungi tim Phitagoras", "testimoni_peserta": "Lihat pengalaman peserta", "out_of_domain": "Tanyakan tentang pelatihan K3", "kompetitor_comparison": "Bandingkan berdasarkan kebutuhan Anda", "informasi_umum": "Pelajari layanan Phitagoras", } INTENT_LABELS = { "sertifikasi_k3_umum": "Ahli K3 Umum", "sertifikasi_k3_konstruksi": "K3 Konstruksi", "harga_pelatihan": "Harga pelatihan", "durasi_pelatihan_k3": "Durasi pelatihan", "syarat_peserta_k3": "Persyaratan peserta", "instruktur_k3": "Instruktur", "sertifikat_berlaku": "Masa berlaku sertifikat", "materi_pelatihan_k3": "Materi pelatihan", "metode_pembelajaran_k3": "Metode pembelajaran", "jaminan_kelulusan": "Kelulusan dan remedial", "sertifikat_lanjutan": "Program lanjutan", "lokasi_pelatihan": "Lokasi pelatihan", "jadwal_pelatihan": "Jadwal pelatihan", "pendaftaran_peserta": "Pendaftaran peserta", "pembayaran_pelatihan": "Pembayaran", "kerjasama_korporat": "Pelatihan perusahaan", "kontak_phitagoras": "Kontak Phitagoras", "testimoni_peserta": "Testimoni peserta", "out_of_domain": "Di luar topik K3", "kompetitor_comparison": "Perbandingan penyelenggara", "informasi_umum": "Informasi Phitagoras", } def __init__( self, intents_path: Optional[str] = None, model_path: Optional[str] = None, confidence_threshold: Optional[float] = None, min_confidence_margin: Optional[float] = None, ) -> None: resolved_intents_path = self._resolve_project_path( intents_path or settings.INTENTS_PATH ) resolved_model_path = self._resolve_project_path( model_path or settings.MODEL_PATH ) configured_threshold = ( confidence_threshold if confidence_threshold is not None else settings.CONFIDENCE_THRESHOLD ) configured_margin = ( min_confidence_margin if min_confidence_margin is not None else getattr( settings, "MIN_CONFIDENCE_MARGIN", self.DEFAULT_CONFIDENCE_MARGIN, ) ) self.confidence_threshold = self._validate_probability( configured_threshold, "confidence_threshold", ) self.min_confidence_margin = self._validate_probability( configured_margin, "min_confidence_margin", ) self.intents_path = resolved_intents_path self.model_path = resolved_model_path self._random = random.SystemRandom() self.intents_data = self._load_intents(self.intents_path) self.pattern_index = self._build_pattern_index(self.intents_data) self.intent_model = IntentModel(model_path=str(self.model_path)) self.gemini_service = GeminiService() logger.info( "✓ Inference service ready: intents=%d, patterns=%d, " "threshold=%.2f, margin=%.2f, gemini=%s", len(self.intents_data), len(self.pattern_index), self.confidence_threshold, self.min_confidence_margin, self.gemini_service.is_available(), ) if not self.intent_model.is_loaded(): logger.error("❌ Intent model artifacts could not be loaded") @staticmethod def _validate_probability(value: Any, field_name: str) -> float: try: probability = float(value) except (TypeError, ValueError) as exc: raise ValueError( f"{field_name} harus berupa angka antara 0 dan 1" ) from exc if not 0.0 <= probability <= 1.0: raise ValueError( f"{field_name} harus berada pada rentang 0 sampai 1" ) return probability @staticmethod def _resolve_project_path(path_value: str) -> Path: path = Path(path_value).expanduser() if path.is_absolute(): return path.resolve() project_root = Path(__file__).resolve().parents[2] return (project_root / path).resolve() @staticmethod def _normalize_text(text: str) -> str: """ Normalize Unicode, case, punctuation, underscores, and whitespace. """ normalized = unicodedata.normalize("NFKC", text) normalized = normalized.lower().strip() normalized = normalized.replace("_", " ") normalized = re.sub( r"[^\w\s]", " ", normalized, flags=re.UNICODE, ) return " ".join(normalized.split()) def _load_intents( self, intents_path: Path, ) -> Dict[str, Dict[str, List[str]]]: try: with intents_path.open("r", encoding="utf-8") as file: data = json.load(file) except FileNotFoundError as exc: logger.exception("❌ Intents file not found: %s", intents_path) raise RuntimeError( f"Intents file tidak ditemukan: {intents_path}" ) from exc except json.JSONDecodeError as exc: logger.exception("❌ Invalid JSON in intents file: %s", intents_path) raise RuntimeError( f"Format JSON intents tidak valid: {intents_path}" ) from exc except OSError as exc: logger.exception("❌ Failed to read intents file: %s", intents_path) raise RuntimeError( f"Gagal membaca intents file: {intents_path}" ) from exc if not isinstance(data, dict) or not data: raise RuntimeError( "Intents data harus berupa object JSON yang tidak kosong" ) validated: Dict[str, Dict[str, List[str]]] = {} for intent_name, payload in data.items(): if not isinstance(intent_name, str) or not intent_name.strip(): raise RuntimeError( "Nama intent harus berupa string yang tidak kosong" ) if not isinstance(payload, dict): raise RuntimeError( f"Payload intent '{intent_name}' harus berupa object" ) patterns = payload.get("patterns") responses = payload.get("responses") if not isinstance(patterns, list) or not patterns: raise RuntimeError( f"Intent '{intent_name}' harus memiliki patterns" ) if not isinstance(responses, list) or not responses: raise RuntimeError( f"Intent '{intent_name}' harus memiliki responses" ) clean_patterns = [ pattern.strip() for pattern in patterns if isinstance(pattern, str) and pattern.strip() ] clean_responses = [ response.strip() for response in responses if isinstance(response, str) and response.strip() ] if not clean_patterns: raise RuntimeError( f"Intent '{intent_name}' tidak memiliki pattern yang valid" ) if not clean_responses: raise RuntimeError( f"Intent '{intent_name}' tidak memiliki response yang valid" ) validated[intent_name] = { "patterns": clean_patterns, "responses": clean_responses, } logger.info("✓ Intents data loaded: %d intents", len(validated)) return validated def _build_pattern_index( self, intents_data: Dict[str, Dict[str, List[str]]], ) -> Dict[str, str]: index: Dict[str, str] = {} for intent_name, payload in intents_data.items(): for pattern in payload["patterns"]: normalized = self._normalize_text(pattern) if normalized in index and index[normalized] != intent_name: raise RuntimeError( "Pattern duplikat ditemukan pada intent berbeda: " f"'{pattern}'" ) index[normalized] = intent_name return index def _find_exact_match(self, text: str) -> Optional[str]: return self.pattern_index.get(self._normalize_text(text)) def _find_fuzzy_match( self, text: str, ) -> Optional[Tuple[str, float, float, str]]: normalized = self._normalize_text(text) if len(normalized) < self.FUZZY_MATCH_MIN_LENGTH: return None ranked: List[Tuple[float, str, str]] = [] for pattern, intent_name in self.pattern_index.items(): score = SequenceMatcher( None, normalized, pattern, autojunk=False, ).ratio() ranked.append((score, intent_name, pattern)) ranked.sort(key=lambda item: item[0], reverse=True) if not ranked: return None best_score, best_intent, best_pattern = ranked[0] second_score = ranked[1][0] if len(ranked) > 1 else 0.0 margin = best_score - second_score if ( best_score >= self.FUZZY_MATCH_THRESHOLD and margin >= self.FUZZY_MATCH_MIN_MARGIN ): return best_intent, float(best_score), float(margin), best_pattern return None def _get_ranked_candidates( self, text: str, top_k: int = TOP_K_CANDIDATES, ) -> List[Dict[str, Any]]: if not self.intent_model.is_loaded(): return [] try: cleaned = self._normalize_text(text) vector = self.intent_model.vectorizer.transform([cleaned]) probabilities = self.intent_model.model.predict_proba(vector)[0] class_labels = self.intent_model.model.classes_ candidates: List[Dict[str, Any]] = [] for class_label, probability in zip(class_labels, probabilities): intent_name = self.intent_model.label_to_intent.get(class_label) if intent_name is None: try: intent_name = self.intent_model.label_to_intent.get( int(class_label) ) except (TypeError, ValueError): intent_name = None if not intent_name: continue candidates.append( { "intent": intent_name, "label": self.INTENT_LABELS.get( intent_name, intent_name.replace("_", " ").title(), ), "confidence": float(probability), } ) candidates.sort( key=lambda item: item["confidence"], reverse=True, ) return candidates[:max(1, top_k)] except Exception: logger.exception("❌ Failed to calculate ranked intent candidates") return [] @staticmethod def _confidence_margin(candidates: List[Dict[str, Any]]) -> float: if not candidates: return 0.0 if len(candidates) == 1: return float(candidates[0]["confidence"]) return float( candidates[0]["confidence"] - candidates[1]["confidence"] ) def _pick_response(self, intent: str) -> Optional[str]: intent_data = self.intents_data.get(intent) if not intent_data: return None responses = intent_data.get("responses", []) if not responses: return None return self._random.choice(responses) def _get_cta(self, intent: str) -> str: return self.INTENT_CTA.get( intent, "Konsultasikan kebutuhan Anda dengan tim Phitagoras", ) def _build_gemini_context( self, candidates: List[Dict[str, Any]], ) -> str: """ Build a small official context from responses of top local intents. """ context_parts: List[str] = [] top_k = max(1, settings.GEMINI_CONTEXT_TOP_K) for candidate in candidates[:top_k]: intent_name = candidate.get("intent") if not intent_name or intent_name == "out_of_domain": continue intent_data = self.intents_data.get(intent_name, {}) responses = intent_data.get("responses", []) if not responses: continue label = candidate.get( "label", self.INTENT_LABELS.get(intent_name, intent_name), ) confidence = float(candidate.get("confidence", 0.0)) context_parts.append( f"Topik: {label}\n" f"Confidence classifier: {confidence:.2%}\n" f"Informasi resmi: {responses[0]}" ) return "\n\n".join(context_parts) def _try_gemini_fallback( self, user_message: str, predicted_intent: Optional[str], confidence: float, candidates: List[Dict[str, Any]], margin: float, reason: str, ) -> Optional[Dict[str, Any]]: """ Use Gemini only for in-domain low-confidence or ambiguous queries. """ if not self.gemini_service.is_available(): return None if predicted_intent == "out_of_domain": return None if candidates and candidates[0].get("intent") == "out_of_domain": return None if confidence < settings.GEMINI_MIN_CLASSIFIER_CONFIDENCE: return None context = self._build_gemini_context(candidates) if not context: return None reply = self.gemini_service.generate_response( user_message=user_message, context=context, ) if not reply: return None logger.info( "✓ Gemini fallback used: intent=%s, reason=%s", predicted_intent, reason, ) return { "success": True, "intent": predicted_intent or "gemini_fallback", "confidence": float(confidence), "reply": reply, "cta": self._get_cta( predicted_intent or "informasi_umum" ), "metadata": { "status": "gemini_fallback", "provider": "google_gemini", "model": settings.GEMINI_MODEL, "classifier_model": self.MODEL_NAME, "classifier_confidence": float(confidence), "confidence_margin": float(margin), "fallback_reason": reason, "top_intents": candidates, }, } def _build_success_response( self, intent: str, confidence: float, match_type: str, metadata: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: reply = self._pick_response(intent) if not reply: return self._build_fallback_response( intent=intent, confidence=confidence, candidates=[], margin=0.0, reason="missing_intent_response", ) response_metadata: Dict[str, Any] = { "status": "success", "model": self.MODEL_NAME, "match_type": match_type, "confidence_threshold": self.confidence_threshold, } if match_type == "fuzzy_pattern": response_metadata["confidence_source"] = "text_similarity" elif match_type == "exact_pattern": response_metadata["confidence_source"] = "exact_match" else: response_metadata["confidence_source"] = "model_probability" if metadata: response_metadata.update(metadata) logger.info( "✓ Response selected: intent=%s, confidence=%.2f%%, match=%s", intent, confidence * 100, match_type, ) return { "success": True, "intent": intent, "confidence": float(confidence), "reply": reply, "cta": self._get_cta(intent), "metadata": response_metadata, } def _build_fallback_response( self, intent: Optional[str], confidence: float, candidates: List[Dict[str, Any]], margin: float, reason: str, ) -> Dict[str, Any]: suggestions = [ { "intent": candidate["intent"], "label": candidate["label"], } for candidate in candidates[:2] if candidate.get("intent") != "out_of_domain" ] return { "success": True, "intent": intent or "unknown", "confidence": float(confidence), "reply": self._random.choice(self.FALLBACK_RESPONSES), "cta": "Perjelas pertanyaan atau hubungi tim Phitagoras", "metadata": { "status": "low_confidence", "reason": reason, "model": self.MODEL_NAME, "confidence_threshold": self.confidence_threshold, "confidence_margin": float(margin), "minimum_confidence_margin": self.min_confidence_margin, "top_intents": candidates, "suggestions": suggestions, "gemini_available": self.gemini_service.is_available(), }, } def get_response(self, user_message: str) -> Dict[str, Any]: if not isinstance(user_message, str) or not user_message.strip(): return { "success": False, "intent": "invalid_input", "confidence": 0.0, "reply": "Silakan tuliskan pertanyaan terlebih dahulu.", "cta": "Tanyakan informasi pelatihan K3", "metadata": { "status": "invalid_input", "reason": "empty_message", }, } guardrail_result = Blacklist.process_input(user_message) sanitized = guardrail_result.get("sanitized", "").strip() if not sanitized: return { "success": False, "intent": "invalid_input", "confidence": 0.0, "reply": "Pesan tidak berisi teks yang dapat diproses.", "cta": "Tulis kembali pertanyaan Anda", "metadata": { "status": "invalid_input", "reason": "empty_after_sanitization", }, } logger.info("📥 Input received: length=%d", len(sanitized)) block_reason = guardrail_result.get("block_reason", "safe") is_competitor_mention = block_reason == "competitor_mention" sanitized_lower = sanitized.lower() contains_blocked_topic = any( topic in sanitized_lower for topic in getattr(Blacklist, "BLOCKED_TOPICS", []) ) allow_competitor_routing = ( is_competitor_mention and not contains_blocked_topic ) if guardrail_result.get("is_blocked") and not allow_competitor_routing: logger.warning("⛔ Input blocked: reason=%s", block_reason) return { "success": True, "intent": "blocked", "confidence": 1.0, "reply": self.BLOCKED_RESPONSE, "cta": "Tanyakan informasi pelatihan K3", "metadata": { "status": "blocked", "reason": block_reason, }, } if ( guardrail_result.get("is_sensitive") and not allow_competitor_routing and guardrail_result.get("override_response") ): return { "success": True, "intent": "sensitive_override", "confidence": 1.0, "reply": guardrail_result["override_response"], "cta": "Hubungi tim Phitagoras", "metadata": { "status": "sensitive_override", "match_type": "guardrail_override", }, } exact_intent = self._find_exact_match(sanitized) if exact_intent: return self._build_success_response( intent=exact_intent, confidence=self.EXACT_MATCH_CONFIDENCE, match_type="exact_pattern", metadata={ "confidence_margin": 1.0, "top_intents": [ { "intent": exact_intent, "label": self.INTENT_LABELS.get( exact_intent, exact_intent, ), "confidence": 1.0, } ], }, ) fuzzy_match = self._find_fuzzy_match(sanitized) if fuzzy_match: fuzzy_intent, fuzzy_score, fuzzy_margin, matched_pattern = ( fuzzy_match ) return self._build_success_response( intent=fuzzy_intent, confidence=fuzzy_score, match_type="fuzzy_pattern", metadata={ "confidence_margin": fuzzy_margin, "matched_pattern": matched_pattern, "fuzzy_threshold": self.FUZZY_MATCH_THRESHOLD, }, ) if not self.intent_model.is_loaded(): return { "success": False, "intent": None, "confidence": 0.0, "reply": self.SERVICE_UNAVAILABLE_RESPONSE, "cta": "Hubungi tim Phitagoras", "metadata": { "status": "service_unavailable", "error": "model_not_loaded", }, } try: classification = self.intent_model.classify( self._normalize_text(sanitized), threshold=self.confidence_threshold, ) except Exception: logger.exception("❌ Intent classification failed") return { "success": False, "intent": None, "confidence": 0.0, "reply": self.SERVICE_UNAVAILABLE_RESPONSE, "cta": "Hubungi tim Phitagoras", "metadata": { "status": "service_unavailable", "error": "classification_failed", }, } intent = classification.get("intent") confidence = float(classification.get("confidence", 0.0)) candidates = self._get_ranked_candidates(sanitized) margin = self._confidence_margin(candidates) logger.info( "🎯 Intent=%s, confidence=%.2f%%, margin=%.2f%%", intent, confidence * 100, margin * 100, ) intent_exists = bool(intent and intent in self.intents_data) passes_confidence = confidence >= self.confidence_threshold passes_margin = margin >= self.min_confidence_margin if not intent_exists: return self._build_fallback_response( intent=intent, confidence=confidence, candidates=candidates, margin=margin, reason="unknown_intent", ) if not passes_confidence: if settings.GEMINI_FALLBACK_ON_LOW_CONFIDENCE: gemini_response = self._try_gemini_fallback( user_message=sanitized, predicted_intent=intent, confidence=confidence, candidates=candidates, margin=margin, reason="below_confidence_threshold", ) if gemini_response: return gemini_response return self._build_fallback_response( intent=intent, confidence=confidence, candidates=candidates, margin=margin, reason="below_confidence_threshold", ) if not passes_margin: if settings.GEMINI_FALLBACK_ON_AMBIGUOUS: gemini_response = self._try_gemini_fallback( user_message=sanitized, predicted_intent=intent, confidence=confidence, candidates=candidates, margin=margin, reason="ambiguous_top_intents", ) if gemini_response: return gemini_response return self._build_fallback_response( intent=intent, confidence=confidence, candidates=candidates, margin=margin, reason="ambiguous_top_intents", ) return self._build_success_response( intent=intent, confidence=confidence, match_type="machine_learning", metadata={ "confidence_margin": margin, "minimum_confidence_margin": self.min_confidence_margin, "top_intents": candidates, }, ) def get_health_status(self) -> Dict[str, Any]: return { "model_loaded": self.intent_model.is_loaded(), "intents_loaded": len(self.intents_data), "patterns_loaded": len(self.pattern_index), "confidence_threshold": self.confidence_threshold, "minimum_confidence_margin": self.min_confidence_margin, "model_path": str(self.model_path), "intents_path": str(self.intents_path), "gemini": self.gemini_service.get_health_status(), }