Spaces:
Sleeping
Sleeping
| # =========================== | |
| # ACC FAQ Chatbot β Final One-Paste Colab | |
| # =========================== | |
| # ---------- Imports ---------- | |
| import os, re, json, csv, random, datetime, zipfile, xml.etree.ElementTree as ET | |
| from typing import List, Dict, Tuple | |
| import numpy as np | |
| import gradio as gr | |
| import docx # python-docx | |
| from sklearn.feature_extraction.text import TfidfVectorizer | |
| from sklearn.metrics.pairwise import cosine_similarity | |
| from sklearn.decomposition import TruncatedSVD | |
| from sklearn.preprocessing import normalize as sk_normalize | |
| import nltk | |
| from nltk.corpus import wordnet as wn, stopwords | |
| from nltk.stem import PorterStemmer | |
| from rapidfuzz import process, fuzz | |
| from sentence_transformers import SentenceTransformer | |
| import faiss | |
| # ---------- NLTK (safe if cached) ---------- | |
| try: _ = wn.all_synsets() | |
| except LookupError: | |
| nltk.download("wordnet"); nltk.download("omw-1.4") | |
| try: _ = stopwords.words("english") | |
| except LookupError: | |
| nltk.download("stopwords") | |
| # ---------- Config ---------- | |
| SUMMARY_SENTENCES = 2 # default TL;DR length | |
| UNMATCHED_LOG_PATH = "unmatched_queries_log.csv" | |
| SVD_COMPONENTS = 300 # 200β400 typically good | |
| RANDOM_SEED = 13 | |
| np.random.seed(RANDOM_SEED); random.seed(RANDOM_SEED) | |
| EMBEDDINGS_ENABLED = True # MiniLM + FAISS layer | |
| # ---------- YOUR SITE LINKS (EDIT THESE) ---------- | |
| SITE_LINKS = { | |
| "home": "https://your-site.example/", # main / homepage | |
| "apply": "https://your-site.example/apply", | |
| "grants": "https://your-site.example/apply", | |
| "donate": "https://your-site.example/donate", | |
| "about": "https://your-site.example/about", | |
| "contact":"https://your-site.example/contact", | |
| "faq": "https://your-site.example/faq", | |
| } | |
| # ---------- Small talk ---------- | |
| SMALL_TALK = { | |
| "greetings": [" hi ", " hello ", " hey ", " salaam ", " salam ", " assalam "], | |
| "how_are_you": [" how are you", " how are u", " how r u", " howβs it going", " hows it going"], | |
| "goodbye": [" bye", " goodbye", " see you", " take care"], | |
| "thanks": [" thanks", " thank you", " thx", " shukran", " jazakallah", " jazak allah", " jazakallahu khair"] | |
| } | |
| SMALL_TALK_RESPONSES = { | |
| "greetings": [ | |
| "π Hello! How can I help you today?", | |
| "π€ Hi there! What would you like to know?", | |
| "π Hey! Iβm your ACC assistant β ask me anything.", | |
| ], | |
| "how_are_you": [ | |
| "π Iβm doing great, thanks for asking! How can I support you today?", | |
| "π Always ready to help with ACC info β what would you like to know?", | |
| ], | |
| "goodbye": [ | |
| "π Goodbye! Wishing you the best.", | |
| "π€² Take care! Iβm here whenever you need me.", | |
| ], | |
| "thanks": [ | |
| "π Youβre most welcome!", | |
| "π€ Glad I could help.", | |
| ], | |
| } | |
| # ---------- (Optional) Quick keyword map for ultra-common snippets ---------- | |
| KEYWORD_MAP = { | |
| "donation": "π You can donate via Credit/Debit Card, Zelle (donate@acceducate.org), Bank Transfer, or by mailing checks.", | |
| "zelle": "π² Donate via Zelle using donate@acceducate.org.", | |
| "check": "βοΈ Mail checks to: 7750 N MacArthur Blvd, Ste 120-282, Irving, TX 75063.", | |
| "zakat": "π Zakat is one of Islamβs five pillars. ACC ensures zakat goes to eligible students.", | |
| "riba": "β Riba means interest, prohibited in Islam. ACC provides riba-free loans.", | |
| "interest": "β Interest (riba) is haram. ACC ensures loans are interest-free.", | |
| "loan": "π΅ ACC offers interest-free loans to students.", | |
| "minimum loan": "π’ Minimum loan: $1,000.", | |
| "maximum loan": "π’ Maximum loan: $10,000.", | |
| "repay": "π Repayment begins 6 months after graduation with flexible monthly installments.", | |
| "about acc": "π ACC is a nonprofit (founded 2013) providing riba-free loans.", | |
| "what is acc": "π ACC is a nonprofit (founded 2013) providing riba-free loans.", | |
| "scam": "β ACC is a registered 501(c)(3) nonprofit. It is legitimate and transparent.", | |
| "email": "π§ adviser@acceducate.org.", | |
| "mail": "π§ adviser@acceducate.org.", | |
| "address": "π Office: 955 W John Carpenter Fwy #100, Irving, TX 75039.", | |
| "facebook": "π https://www.facebook.com/acontinuouscharity", | |
| "instagram": "πΈ https://www.instagram.com/acontinuouscharity", | |
| "youtube": "βΆοΈ https://www.youtube.com/channel/UCkq8wAvAqt54yzjzL_QIq3w", | |
| "twitter": "π¦ https://x.com/accnational", | |
| } | |
| # ---------- Donation / Repayment context cues ---------- | |
| DONATION_CUES = [" donate"," donation"," give"," giving"," zakat"," sadaqah"," sadaqa"," appeal"," campaign"," fundraiser"," payment options"," apple pay"," card"," zelle"," bank transfer"," check"," pay to donate"," ways to give"] | |
| REPAYMENT_CUES = [" repay"," repayment"," installments"," monthly payment"," pay back"," due"," loan payment"," when do i pay"," when to repay"] | |
| # ---------- Navigation cues (includes Home & βshopβ redirects) ---------- | |
| NAV_CUES = [ | |
| ("home", ["home", "homepage", "main page", "start page", "website home", | |
| "what is this website about", "what does this website do", "what is this site", | |
| "buy coffee", "buy clothes", "do you sell products", "store", "shop", "shopping", "purchase", "products", "where is the shop"]), | |
| ("apply", ["apply online","application portal","start application","apply now","grant application","grant page","take me to the grant","grant"]), | |
| ("donate", ["donate now","donation page","donate online","give now","take me to donate","payment page"]), | |
| ("about", ["about us","about acc","learn about acc","organization info"]), | |
| ("contact",["contact","contact us","reach you","email you","support email"]), | |
| ("faq", ["faq","faqs","help center","help desk"]), | |
| ] | |
| # ---------- Built-in Add-on entries (loan-side + website redirects) ---------- | |
| HOME_URL = SITE_LINKS.get("home","") | |
| APPLY_URL = SITE_LINKS.get("apply","") | |
| ADDON_ENTRIES = [ | |
| # Loan-side | |
| ("When do I start repayment?", "Repayment begins 6 months after graduation. Payments are made in manageable monthly installments."), | |
| ("When does repayment start?", "Repayment begins 6 months after graduation. Payments are made in manageable monthly installments."), | |
| ("Is there a grace period?", "Yes. There is a 6-month grace period after graduation before repayment begins."), | |
| ("How much are monthly installments?", "Installments are structured to be manageable; exact amounts depend on the total borrowed and agreed schedule."), | |
| ("Can I pay early or extra?", "Yes. You may make extra or early payments. Early repayment reduces your outstanding balance sooner."), | |
| ("How do I make repayments?", "Follow the repayment instructions provided by ACC. If you have any issues, please contact adviser@acceducate.org."), | |
| ("Payment options to repay a loan", "Repayments follow the instructions provided by ACC and are made in monthly installments. Contact adviser@acceducate.org if you need help."), | |
| ("What happens if I canβt pay?", "If you experience a hardship, contact adviser@acceducate.org to discuss available options per ACC policy."), | |
| ("Who can apply for a loan?", "Students pursuing accredited education who meet ACCβs criteria may apply. Documentation and an interview are typically required."), | |
| ("Am I eligible for ACC loan?", "Eligibility depends on ACCβs criteria for students pursuing accredited education. Applicants provide documentation and complete an interview."), | |
| ("How do I apply for a loan?", "Complete the application on the ACC website, submit required documents, and complete the interview process."), | |
| ("Where can I apply for a loan?", f"You can start your application online here: {APPLY_URL}" if APPLY_URL else "You can start your application on the ACC website."), | |
| ("What is the minimum loan amount?", "The minimum loan amount is $1,000."), | |
| ("What is the maximum loan amount?", "The maximum loan amount is $10,000."), | |
| ("Is there any interest?", "No. ACC provides riba-free (interest-free) loans."), | |
| ("Are ACC loans halal?", "Yes. ACC loans are riba-free (interest-free)."), | |
| ("How long does the application review take?", "Processing times can vary. Youβll be contacted by ACC as your application is reviewed and next steps are scheduled."), | |
| ("Are there application deadlines?", "Application timelines may vary. Please check the application page for current information."), | |
| # Website redirects to Home (with blurb) | |
| ("What is this website about?", f"ACC (A Continuous Charity) is a 501(c)(3) nonprofit founded in 2013 that provides riba-free (interest-free) loans to students. Learn more on our homepage: {HOME_URL}"), | |
| ("What does this website do?", f"ACC supports students with interest-free (riba-free) educational loans funded by donations. Visit our homepage for an overview: {HOME_URL}"), | |
| ("What is this site?", f"This is ACCβs official site for information, donations, and student support. Start here: {HOME_URL}"), | |
| ("Can we buy coffee here?", f"No β this is not an e-commerce site. ACC is a nonprofit aiding students with riba-free loans. Please see our homepage: {HOME_URL}"), | |
| ("Can we buy clothes here?", f"No β this is not an online store. ACC is a nonprofit supporting students with riba-free loans. Learn more: {HOME_URL}"), | |
| ("Do you sell products?", f"No β ACC is a nonprofit organization and does not operate an online store. See our homepage for mission and programs: {HOME_URL}"), | |
| ("Where is the shop?", f"ACC does not have an online shop. To learn about our mission and programs, please visit the homepage: {HOME_URL}"), | |
| ] | |
| # ---------- Loaders ---------- | |
| def load_csv(path: str) -> List[Dict[str, str]]: | |
| import pandas as pd | |
| qas = [] | |
| df = pd.read_csv(path) | |
| # normalize headers | |
| cols = {c.lower(): c for c in df.columns} | |
| qcol = cols.get("question") or cols.get("q") or cols.get("title") | |
| acol = cols.get("answer") or cols.get("a") or cols.get("content") or cols.get("text") | |
| if not qcol or not acol: | |
| raise ValueError("CSV must have 'question' and 'answer' columns.") | |
| for q, a in df[[qcol, acol]].itertuples(index=False): | |
| if isinstance(q, str) and isinstance(a, str) and q.strip() and a.strip(): | |
| qas.append({"question": q.strip(), "answer": a.strip()}) | |
| return qas | |
| def load_file(path: str) -> List[Dict[str, str]]: | |
| pl = path.lower() | |
| if pl.endswith(".csv"): return load_csv(path) | |
| raise ValueError("Unsupported file format. Use .docx, .csv, or .json") | |
| # ---------- Text utils ---------- | |
| STOP = set(stopwords.words("english")) | |
| STEM = PorterStemmer() | |
| def normalize_basic(text: str) -> str: | |
| t = " " + (text or "").lower() + " " | |
| t = re.sub(r"[^a-z0-9 ]+", " ", t) | |
| t = re.sub(r"\s+", " ", t).strip() | |
| return t | |
| def tokenize_and_stem(text: str) -> List[str]: | |
| t = normalize_basic(text) | |
| tokens = [w for w in t.split() if w not in STOP] | |
| return [STEM.stem(w) for w in tokens] | |
| def normalize_for_index(text: str) -> str: | |
| rep = { | |
| r"\bmin\b": "minimum", | |
| r"\bmax\b": "maximum", | |
| r"\bdon\b": "donation", | |
| r"\bfund\b": "donation", | |
| r"\bmoney\b": "donation", | |
| r"\brecurring\b": "monthly", | |
| r"\bacc\b": "a continuous charity", | |
| r"\bmail\b": "email", | |
| r"\bzelle\b": "zelle", | |
| r"\briba\b": "riba", | |
| } | |
| t = " " + (text or "").lower() + " " | |
| for k, v in rep.items(): | |
| t = re.sub(k, f" {v} ", t) | |
| tokens = tokenize_and_stem(t) | |
| return " ".join(tokens) | |
| def split_sentences(text: str) -> List[str]: | |
| parts = re.split(r'(?<=[.!?])\s+', (text or "").strip()) | |
| return [p.strip() for p in parts if p.strip()] | |
| def short_summary(answer: str, n: int = SUMMARY_SENTENCES) -> str: | |
| sents = split_sentences(answer) | |
| if len(sents) <= n: | |
| return (answer or "").strip() | |
| return " ".join(sents[:n]).strip() | |
| # ---------- Intent / Nav helpers ---------- | |
| def _detect_nav_target(text: str): | |
| q = " " + (text or "").lower().strip() + " " | |
| for key, triggers in NAV_CUES: | |
| for t in triggers: | |
| if f" {t} " in q or q.strip() == t: | |
| return key | |
| if " take me to " in q: | |
| for key in SITE_LINKS: | |
| if key in q: return key | |
| return None | |
| # ---------- Chatbot ---------- | |
| class FAQChatbot: | |
| def __init__(self, qa_list: List[Dict[str, str]]): | |
| assert qa_list, "QA list is empty." | |
| # De-dupe and clean | |
| seen = set(); uniq = [] | |
| for qa in qa_list: | |
| q = (qa.get("question") or "").strip() | |
| a = (qa.get("answer") or "").strip() | |
| if not q or not a: continue | |
| if q.lower() in seen: continue | |
| seen.add(q.lower()); uniq.append({"question": q, "answer": a}) | |
| # Add-on merge (runtime; only if not present) | |
| for q_add, a_add in ADDON_ENTRIES: | |
| if q_add.strip().lower() not in seen: | |
| uniq.append({"question": q_add.strip(), "answer": a_add.strip()}) | |
| seen.add(q_add.strip().lower()) | |
| self.qa_list = uniq | |
| self.questions = [x["question"] for x in uniq] | |
| self.answers = [x["answer"] for x in uniq] | |
| # Precompute normalized forms | |
| self.norm_qs_basic = [normalize_basic(q) for q in self.questions] | |
| self.norm_qs_idx = [normalize_for_index(q) for q in self.questions] | |
| self.q_tokens = [tokenize_and_stem(q) for q in self.questions] | |
| # Vectorizers & matrices | |
| self.vec_word = TfidfVectorizer(analyzer="word", ngram_range=(1, 2), min_df=2, sublinear_tf=True) | |
| self.vec_char = TfidfVectorizer(analyzer="char", ngram_range=(3, 5), min_df=2) | |
| self.mat_word = self.vec_word.fit_transform(self.norm_qs_idx) | |
| self.mat_char = self.vec_char.fit_transform(self.norm_qs_idx) | |
| # Semantic layer (LSA) | |
| max_comp = max(2, min(SVD_COMPONENTS, self.mat_word.shape[1] - 1)) | |
| self.svd = TruncatedSVD(n_components=max_comp) | |
| self.mat_sem = sk_normalize(self.svd.fit_transform(self.mat_word)) | |
| # Embeddings + FAISS | |
| self._faiss = None; self._emb_qs = None; self._enc = None | |
| if EMBEDDINGS_ENABLED and len(self.questions) > 0: | |
| self._enc = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2") | |
| self._emb_qs = self._enc.encode(self.questions, normalize_embeddings=True).astype("float32") | |
| d = self._emb_qs.shape[1] | |
| self._faiss = faiss.IndexFlatIP(d) | |
| self._faiss.add(self._emb_qs) | |
| # UI voice | |
| self.response_styles = [ | |
| "π Of course, let me explain:", | |
| "π Absolutely, hereβs the info:", | |
| "π Great question! Hereβs what I found:", | |
| "π€ Glad you asked! Hereβs what I know:", | |
| "π Hereβs a quick summary:", | |
| ] | |
| # Keyword override | |
| self.keyword_map = KEYWORD_MAP.copy() | |
| self._kw_keys = list(self.keyword_map.keys()) | |
| self._keyword_matcher = lambda q: (lambda m: (m[0], m[1] / 100.0) if m else ("", 0.0))( | |
| process.extractOne(q, self._kw_keys, scorer=fuzz.token_set_ratio) | |
| ) | |
| # Trigger index for common features | |
| self.trigger_index = self._build_trigger_index(top_k=300) | |
| self._last_intent = None | |
| self._log_buf = [] | |
| def _build_trigger_index(self, top_k=300) -> Dict[str, int]: | |
| vec = TfidfVectorizer(analyzer="word", ngram_range=(1, 2), min_df=5) | |
| X = vec.fit_transform(self.norm_qs_idx) | |
| if X.shape[1] == 0: | |
| return {} | |
| vocab = np.array(vec.get_feature_names_out()) | |
| scores = np.asarray(X.sum(axis=0)).ravel() | |
| top_idx = scores.argsort()[::-1][:top_k] | |
| Xd = X.toarray() | |
| triggers = {} | |
| for i in top_idx: | |
| feat = vocab[i] | |
| col = Xd[:, i] | |
| best_q = int(col.argmax()) | |
| triggers[feat] = best_q | |
| return triggers | |
| def _exact_or_near_match(self, qn_basic: str) -> Tuple[int, float]: | |
| try: | |
| idx = self.norm_qs_basic.index(qn_basic) | |
| return idx, 1.0 | |
| except ValueError: | |
| pass | |
| q_tokens = tokenize_and_stem(qn_basic) | |
| best_idx, best_score = -1, 0.0 | |
| for i, toks in enumerate(self.q_tokens): | |
| s = len(set(q_tokens) & set(toks)) / max(1, len(set(q_tokens) | set(toks))) | |
| if s > best_score: | |
| best_idx, best_score = i, s | |
| return best_idx, best_score | |
| def _keyword_override(self, raw_q: str) -> Tuple[str, float]: | |
| return self._keyword_matcher(raw_q) | |
| def _trigger_lookup(self, qn_idx: str) -> Tuple[int, float]: | |
| tokens = set(qn_idx.split()) | |
| cand = [self.trigger_index[t] for t in tokens if t in self.trigger_index] | |
| if not cand: | |
| return -1, 0.0 | |
| best_idx = max(set(cand), key=cand.count) | |
| conf = min(0.55, 0.25 + 0.05 * len(cand)) | |
| return best_idx, conf | |
| def _semantic_scores(self, qn_idx: str): | |
| q_w = self.vec_word.transform([qn_idx]) | |
| q_sem = sk_normalize(self.svd.transform(q_w)) | |
| sims_sem = cosine_similarity(q_sem, self.mat_sem)[0] | |
| return sims_sem | |
| def _semantic_topk(self, query: str, k: int = 5): | |
| if self._faiss is None: | |
| return np.array([], dtype=int), np.array([], dtype=float) | |
| qv = self._enc.encode([query], normalize_embeddings=True).astype("float32") | |
| sims, idxs = self._faiss.search(qv, min(k, len(self.questions))) | |
| return idxs[0], sims[0] | |
| def _log_unmatched(self, raw_q: str, suggestions: List[str], scores: List[float]): | |
| self._log_buf.append([ | |
| datetime.datetime.utcnow().isoformat(), raw_q, | |
| suggestions[0] if len(suggestions) > 0 else "", f"{scores[0]:.4f}" if len(scores) > 0 else "", | |
| suggestions[1] if len(suggestions) > 1 else "", f"{scores[1]:.4f}" if len(scores) > 1 else "", | |
| suggestions[2] if len(suggestions) > 2 else "", f"{scores[2]:.4f}" if len(scores) > 2 else "", | |
| ]) | |
| if len(self._log_buf) >= 25: | |
| self._flush_log() | |
| def _flush_log(self): | |
| exists = os.path.exists(UNMATCHED_LOG_PATH) | |
| with open(UNMATCHED_LOG_PATH, "a", encoding="utf-8", newline="") as f: | |
| w = csv.writer(f) | |
| if not exists: | |
| w.writerow(["timestamp","query","suggestion_1","score_1","suggestion_2","score_2","suggestion_3","score_3"]) | |
| w.writerows(self._log_buf) | |
| self._log_buf.clear() | |
| def _context_flags(self, qlow: str) -> Tuple[bool, bool]: | |
| is_donation = any(c in qlow for c in DONATION_CUES) | |
| is_repayment = any(c in qlow for c in REPAYMENT_CUES) | |
| if is_donation and not is_repayment: | |
| self._last_intent = "donation" | |
| elif is_repayment and not is_donation: | |
| self._last_intent = "repayment" | |
| return is_donation, is_repayment | |
| def _render(self, ans: str): | |
| intro = random.choice(self.response_styles) | |
| sents = split_sentences(ans) | |
| if len(ans) > 180 and len(sents) > SUMMARY_SENTENCES: | |
| tl = short_summary(ans, SUMMARY_SENTENCES) | |
| if tl.strip().lower() != sents[0].strip().lower(): | |
| return f"{intro}\n**TL;DR:** {tl}\n\nπ {ans}" | |
| return f"{intro}\nπ {ans}" | |
| def get_answer(self, query: str) -> str: | |
| raw_q = (query or "").strip() | |
| if not raw_q: | |
| return "β οΈ Please type a question so I can assist you better." | |
| qlow = " " + raw_q.lower() + " " | |
| # small talk | |
| for cat, trigs in SMALL_TALK.items(): | |
| if any(t in qlow for t in trigs): | |
| return random.choice(SMALL_TALK_RESPONSES[cat]) | |
| # identity | |
| if any(p in qlow for p in [" who are you", " what are you", " tell me about yourself"]): | |
| return random.choice([ | |
| "π€ Iβm ACCβs virtual assistant, here to answer your questions about donations, loans, and more.", | |
| "π Iβm an AI chatbot trained on ACCβs FAQs to guide you with accurate answers.", | |
| "π Iβm your online assistant for A Continuous Charity, here to help you understand ACC.", | |
| ]) | |
| # navigation (incl. home & ecommerce) | |
| nav = _detect_nav_target(raw_q) | |
| if nav and nav in SITE_LINKS and SITE_LINKS[nav]: | |
| label = { | |
| "home":"Home","apply":"Apply Online","grants":"Grant Application", | |
| "donate":"Donate Now","about":"About Us","contact":"Contact Us","faq":"FAQ" | |
| }.get(nav, nav.title()) | |
| intro = random.choice(self.response_styles) | |
| if nav == "home": | |
| blurb = "π ACC (A Continuous Charity) is a 501(c)(3) nonprofit founded in 2013 that provides riba-free (interest-free) loans to students." | |
| return f"{intro}\n{blurb}\nπ **[{label}]({SITE_LINKS[nav]})**" | |
| return f"{intro}\nπ **[{label}]({SITE_LINKS[nav]})**" | |
| # credibility/safety | |
| if any(k in qlow for k in [" scam", " fraud", " fake", " legit", " legitimate", " trust ", " real "]): | |
| return self._render("β ACC is a registered **501(c)(3)** nonprofit. It is legitimate and transparent.") | |
| # topic-ish fallback: payment options ambiguity | |
| is_donation, is_repayment = self._context_flags(qlow) | |
| if " payment option" in qlow or " pay " in qlow or qlow.strip() == "payment options?": | |
| if is_donation or self._last_intent == "donation": | |
| return self._render( | |
| "π **Ways to donate to ACC**\n\nβ’ Credit/Debit Card (online)\nβ’ Zelle: donate@acceducate.org\nβ’ Bank Transfer (contact us)\nβ’ Check by Mail: 7750 N MacArthur Blvd, Ste 120-282, Irving, TX 75063\n\nβΉοΈ Apple Pay may appear on supported devices/browsers." | |
| ) | |
| if is_repayment or self._last_intent == "repayment": | |
| return self._render("π **Repayment** begins **6 months after graduation** with monthly installments.") | |
| # retrieval pipeline | |
| qn_basic = normalize_basic(raw_q) | |
| qn_idx = normalize_for_index(raw_q) | |
| # exact / near-exact | |
| near_idx, near_score = self._exact_or_near_match(qn_basic) | |
| if near_score >= 0.90: | |
| return self._render(self.answers[near_idx]) | |
| # keyword quick win | |
| key, key_score = self._keyword_override(qlow) | |
| if key_score >= 0.60 and key in self.keyword_map: | |
| return self._render(self.keyword_map[key]) | |
| # trigger lookup | |
| trig_idx, trig_conf = self._trigger_lookup(qn_idx) | |
| if trig_idx >= 0 and trig_conf >= 0.45: | |
| return self._render(self.answers[trig_idx]) | |
| # TF-IDF + char + LSA | |
| q_w = self.vec_word.transform([qn_idx]) | |
| q_c = self.vec_char.transform([qn_idx]) | |
| sims_word = cosine_similarity(q_w, self.mat_word)[0] | |
| sims_char = cosine_similarity(q_c, self.mat_char)[0] | |
| sims_sem = self._semantic_scores(qn_idx) | |
| sims_blend_lsa = 0.50 * sims_sem + 0.30 * sims_word + 0.20 * sims_char | |
| # Embeddings + FAISS blend | |
| final_scores = sims_blend_lsa.copy() | |
| if EMBEDDINGS_ENABLED: | |
| idx_e, sc_e = self._semantic_topk(raw_q, k=max(5, min(10, len(self.questions)))) | |
| emb = np.zeros_like(final_scores) | |
| if len(idx_e) > 0: | |
| emb[idx_e] = sc_e | |
| final_scores = 0.55 * emb + 0.30 * sims_sem + 0.10 * sims_word + 0.05 * sims_char | |
| best_idx = int(np.argmax(final_scores)) | |
| best_score = float(final_scores[best_idx]) | |
| # decisions | |
| if near_score >= 0.75: | |
| return self._render(self.answers[near_idx]) | |
| if best_score >= (0.40 if EMBEDDINGS_ENABLED else 0.36): | |
| return self._render(self.answers[best_idx]) | |
| # suggestions + log | |
| if best_score >= (0.22 if EMBEDDINGS_ENABLED else 0.18): | |
| top = np.argsort(final_scores)[::-1][:3] | |
| suggestions = [self.questions[i] for i in top] | |
| scores = [float(final_scores[i]) for i in top] | |
| self._log_unmatched(raw_q, suggestions, scores) | |
| msg = "π€ Iβm not completely sure β did you mean one of these?\n\n" | |
| msg += "\n".join([f"πΉ {s}" for s in suggestions]) | |
| msg += "\n\nYou can click one of these or rephrase your question." | |
| return msg | |
| self._log_unmatched(raw_q, [], []) | |
| return "π I couldnβt find anything close. Could you rephrase your question or add a bit more detail?" | |
| # ---------- Build bot from path ---------- | |
| def build_bot_from_path(path: str) -> FAQChatbot: | |
| qas = load_file(path) | |
| return FAQChatbot(qas) | |
| # ---------- Default dataset candidates ---------- | |
| CANDIDATES = [ | |
| "dataset.csv", # fixed typo β | |
| ] | |
| DEFAULT_DATA = next((p for p in CANDIDATES if os.path.exists(p)), None) | |
| bot = None | |
| if DEFAULT_DATA: | |
| try: | |
| bot = build_bot_from_path(DEFAULT_DATA) | |
| print(f"β Loaded dataset: {DEFAULT_DATA} (QAs: {len(bot.qa_list)})") | |
| except Exception as e: | |
| print("β Default load failed:", e) | |
| # ---------- Gradio UI ---------- | |
| with gr.Blocks() as demo: | |
| gr.Markdown("## π€ ACC Online Assistant") | |
| summary_n = gr.Slider(1, 5, value=SUMMARY_SENTENCES, step=1, label="TL;DR sentences") | |
| status = gr.Markdown(value=f"β Loaded default dataset ({len(bot.qa_list)} QAs)" if bot else "β Failed to load dataset.") | |
| chatbot = gr.Chatbot( | |
| height=520, | |
| show_label=False, | |
| value=[["π€ Assistant", "π Hello! Iβm your ACC Online Assistant. Ask me anything."]] | |
| ) | |
| msg = gr.Textbox(label="π¬ Ask me anything about ACC", placeholder="Type your question here...") | |
| clear = gr.Button("Clear") | |
| _bot = {"inst": bot} | |
| # Respond to user query | |
| def respond(message, chat_history, n_sentences=None): | |
| global SUMMARY_SENTENCES | |
| SUMMARY_SENTENCES = int(n_sentences) | |
| if _bot["inst"] is None: | |
| return "", chat_history + [["π€ Assistant", "β οΈ No dataset loaded."]] | |
| reply = _bot["inst"].get_answer(message) | |
| chat_history.append(["π You", message]) | |
| chat_history.append(["π€ Assistant", reply]) | |
| return "", chat_history | |
| # Clear chat | |
| def do_clear(): | |
| return "", [["π€ Assistant", "π Hello! Iβm your ACC Online Assistant. Ask me anything."]] | |
| # Input bindings | |
| msg.submit(respond, [msg, chatbot, summary_n], [msg, chatbot]) | |
| clear.click(do_clear, [], [msg, chatbot], queue=False) | |
| # ---------- Launch ---------- | |
| demo.launch(server_name="0.0.0.0", server_port=7860) | |