""" Khalifa University Library AI Agent MCP-style tool-calling backend with RAG, PRIMO, PubMed, Google Scholar, Consensus Tools: - search_primo: Search KU Library catalog - search_pubmed: Search biomedical literature - search_scholar: Search Google Scholar - search_consensus: Search Consensus (research papers) - get_library_info: RAG from KU library knowledge base Environment variables (HF Space Secrets): OPENAI_API_KEY — required (embeddings + ChatGPT) ANTHROPIC_API_KEY — optional (Claude answers) PRIMO_API_KEY — required (PRIMO search) """ import os import json import re import glob import time import sqlite3 import hashlib import secrets from datetime import datetime from contextlib import asynccontextmanager from fastapi import FastAPI, Request, Response, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, RedirectResponse from pydantic import BaseModel from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_community.vectorstores import FAISS from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document import httpx # ===== CONFIG ===== KNOWLEDGE_DIR = "knowledge" # Use /data for persistence across HF Space restarts — fall back to local if /data unavailable FAISS_INDEX_PATH = "/data/faiss_index" if os.path.exists("/data") else "faiss_index" DB_PATH = "/data/analytics.db" if os.path.exists("/data") else "analytics.db" CHUNK_SIZE = 800 CHUNK_OVERLAP = 100 TOP_K = 7 # ===== MEDICAL TOPIC DETECTION ===== MEDICAL_KEYWORDS = [ 'medicine','medical','health','clinical','nursing','pharmacy','biomedical', 'anatomy','physiology','epidemiology','oncology','cardiology','neuroscience', 'genomics','pathology','surgery','mental health','nutrition','public health', 'disease','therapy','treatment','diagnosis','patient','drug','pharmaceutical', 'hospital','cancer','diabetes','heart','brain','cell biology','immunology', 'biochemistry','microbiology','virology','pediatric','psychiatric','dental', 'ophthalmology','radiology','anesthesia','emergency medicine','geriatric' ] CURRENT_INFO_RE = re.compile(r"\b(current|currently|latest|today|recent|recently|now|this week|this month|news|update|updated|president|prime minister|ceo|minister|king|ruler|who won)\b", re.IGNORECASE) RESEARCH_CUE_RE = re.compile(r"\b(find|search|look for|show me|get me|give me|locate|recommend|suggest|articles?|papers?|books?|journals?|studies|literature|research|systematic review|evidence|sources|database|databases|peer reviewed|open access)\b", re.IGNORECASE) LIBRARY_CUE_RE = re.compile(r"\b(library|librarian|borrow|loan|renew|fine|study room|room booking|reserve a room|account|my library|interlibrary|ill|khazna|orcid|open access|apc|refworks|libkey|hours|location|contact|visitor|alumni|database access|off campus|remote access)\b", re.IGNORECASE) MEDICAL_SEARCH_RE = re.compile(r"\b(pubmed|embase|cinahl|clinicalkey|cochrane|uptodate|medline|systematic review|clinical trial|biomedical literature|medical literature)\b", re.IGNORECASE) SOCIAL_RE = re.compile(r"^(hi|hello|hey|good morning|good afternoon|good evening|how are you|how old are you|who are you|what are you|are you a bot|are you a robot|what can you do|thanks|thank you|ok|okay|bye|goodbye|lol|haha|hehe|you are silly|are you silly|are you dumb|stupid bot|dumb bot|idiot bot|joke|tell me a joke)\b", re.IGNORECASE) PURE_GREETING_RE = re.compile(r"^(hi|hello|hey|good morning|good afternoon|good evening)[!.\s]*$", re.IGNORECASE) HOURS_RE = re.compile( r"\b(" r"library hours?|opening hours?|closing hours?|open hours?|" r"open today|open tomorrow|open now|open on|open this|" r"closed today|closed tomorrow|closed on|closed this|" r"when.*open|when.*close|when.*does.*library|" r"what time.*open|what time.*close|what time.*library|" r"opening time|closing time|" r"is.*library.*open|is.*campus.*library|is.*habshan.*open|is.*san.*open|" r"is.*library.*closed|will.*library.*open|will.*it.*open|" r"does.*library.*open|library.*open.*today|library.*close.*today|" r"habshan.*open|san.*open|main campus.*open|" r"ramadan hours?|weekend hours?|friday hours?|saturday hours?|sunday hours?|" r"library.*schedule|schedule.*library|" r"what.*hours|till.*when|until.*when|" r"what day.*open|which day.*open|what day.*closed|which day.*closed" r")\b", re.IGNORECASE ) KU_GENERAL_RE = re.compile(r"\b(khalifa university|ku)\b.*\b(admission|admissions|program|programs|degree|degrees|college|colleges|school|schools|tuition|fees|scholarship|scholarships|hostel|housing|transport|ranking|rankings|president|vice president|chancellor|application|apply|registrar|academic calendar|semester dates|campus map)\b|\b(admission|admissions|program|programs|degree|degrees|college|colleges|school|schools|tuition|fees|scholarship|scholarships|hostel|housing|transport|ranking|rankings|president|vice president|chancellor|application|apply|registrar|academic calendar|semester dates|campus map)\b.*\b(khalifa university|ku)\b", re.IGNORECASE) LIBRARY_HOURS_URL = "https://library.ku.ac.ae/hours" KU_MAIN_URL = "https://www.ku.ac.ae/" GREETING_FOLLOWUP_RE = re.compile(r"^(yes|yes please|yes check|sure|okay|ok|yeah|yep|please do|go ahead)$", re.IGNORECASE) # Campus / location questions — always answered with hardcoded text, never from web KU_CAMPUS_RE = re.compile( r"\b(" r"where is the library|where is.*library|library.*location|location.*library|" r"which building|what building|which floor|" r"main campus library|san campus library|habshan library|sas al nakhl library|" r"arzanah library|" r"library.*address|address.*library|" r"how to get to.*library|how.*reach.*library|how.*find.*library|" r"library.*campus(es)?|campus(es)?.*library|" r"how many.*librar|how many.*branch|library.*branch(es)?|branch(es)?.*library|" r"which campus.*library|library.*which campus|" r"library.*map|map.*library" r")\b", re.IGNORECASE ) def _last_assistant_message(history): if not history: return "" for m in reversed(history): if m.get("role") == "assistant": return m.get("content", "") or "" return "" def _is_greeting_menu_followup(question: str, history) -> bool: q = (question or "").strip() if not GREETING_FOLLOWUP_RE.match(q): return False last = _last_assistant_message(history).lower() return ( "i’m libbee" in last or "i'm libbee" in last or "are you looking for one of these" in last or "khalifa university library ai assistant" in last ) def _greeting_menu_clarify_answer() -> str: return ( "Sure — what would you like help with?

" "You can choose one of these or type your question directly:
" "• Find articles and books
" "• Get full text for an article
" "• Submit an Interlibrary Loan (ILL) request
" "• Contact a librarian
" "• Check library hours" ) def _normalize_social_text(text: str) -> str: t = (text or "").strip().lower() t = re.sub(r"[^\w\s]", "", t) t = re.sub(r"\s+", " ", t).strip() return t def _fixed_social_answer(question: str) -> str | None: q = _normalize_social_text(question) if re.fullmatch(r"(how are you|hows it going|how are you doing|how do you do)", q): return ( "I’m doing well, thank you. I’m LibBee, the Khalifa University Library AI Assistant, " "and I’m here to help with articles, books, databases, full text, and library services." ) if re.fullmatch(r"(who are you|what are you|tell me about yourself)", q): return ( "I’m LibBee, the Khalifa University Library AI Assistant. " "I can help you find articles and books, search databases, get full text, " "answer library service questions, and guide you to the right librarian when needed." ) if re.fullmatch(r"(what can you do|what do you do|how can you help|what help can you provide)", q): return ( "I can help you search for articles and books, suggest databases, find full text, " "answer questions about library services, and connect you with the right Khalifa University Library staff member." ) if re.fullmatch(r"(are you a bot|are you a robot|are you ai|are you an ai)", q): return ( "Yes — I’m an AI assistant for Khalifa University Library. " "My job is to help with library research, databases, articles, books, full text, and library services." ) if re.fullmatch(r"(thank you|thanks|thx)", q): return "You’re welcome. I’m happy to help." if re.fullmatch(r"(bye|goodbye|see you)", q): return "Goodbye. If you need help later, I’ll be here." return None STAFF_DIRECTORY = [ # ── Library Leadership ─────────────────────────────────────────── { "full_name": "Dr. Abdulla Al Hefeiti", "role": "Library Director / Assistant Provost, Libraries", "details": "He can help with strategic and institutional matters, library leadership, and partnerships. Email: abdulla.alhefeiti@ku.ac.ae | Phone: +971 2 312 3331", "tokens": ["abdulla", "hefeiti", "abdulla al hefeiti", "al hefeiti"], }, # ── Specialist Librarians ──────────────────────────────────────── { "full_name": "Nikesh Narayanan", "role": "Research & Access Services Librarian", "details": ( "He can help with research support, Khazna institutional repository, ORCID, " "SciVal, Scopus, research impact, AI tools for research, bibliometrics, " "scholarly communication, and LibGuides. " "Email: nikesh.narayanan@ku.ac.ae | Phone: +971 2 312 3980" ), "tokens": ["nikesh", "narayanan", "nikesh narayanan"], }, { "full_name": "Rani Anand", "role": "E-Resources Librarian", "details": ( "She can help with database access problems, e-resources troubleshooting, " "remote access, vendor issues, patents, e-books, and bibliometrics. " "Email: rani.anand@ku.ac.ae | Phone: +971 2 312 3935" ), "tokens": ["rani", "anand", "rani anand"], }, { "full_name": "Jason Fetty", "role": "Medical Librarian", "details": ( "He can help with medical and health sciences research, systematic reviews, " "PubMed, Embase, CINAHL, UpToDate, and clinical databases. " "Email: jason.fetty@ku.ac.ae | Phone: +971 2 312 4722" ), "tokens": ["jason", "fetty", "jason fetty"], }, { "full_name": "Walter Brian Hall", "role": "Digital & Technology Services Librarian / Systems Librarian", "details": ( "He can help with library website, systems, technology, digital infrastructure, " "ORCID, STEAM, coding, Open Access publishing, APC funding, " "general library services, and public services. " "Email: walter.hall@ku.ac.ae | Open Access: openaccess@ku.ac.ae | " "Phone: +971 2 312 3163 | Mobile: 052-905-1491" ), "tokens": ["walter", "brian", "hall", "walter brian hall", "walter hall", "brian hall"], }, # ── Managers ──────────────────────────────────────────────────── { "full_name": "Alia Al-Harrasi", "role": "Manager, Technical Services", "details": ( "She can help with cataloguing, metadata, acquisitions processing, and technical services. " "Email: alia.alharrasi@ku.ac.ae | Phone: +971 2 312 3180" ), "tokens": ["alia", "harrasi", "al-harrasi", "alia al harrasi", "alia al-harrasi"], }, { "full_name": "Muna Ahmad Mohammad Al Blooshi", "role": "Manager, Public Services", "details": ( "She can help with circulation, general library services on Main Campus, and public services. " "Email: muna.alblooshi@ku.ac.ae | Phone: +971 2 312 4212" ), "tokens": ["muna", "blooshi", "al blooshi", "al-blooshi", "muna al blooshi"], }, # ── Library Information Senior Specialists / Specialists ───────── { "full_name": "Suaad Al Jneibi", "role": "Library Information Senior Specialist", "details": ( "She can help with Interlibrary Loan (ILL) requests, document delivery, " "and general library questions. " "Email: suaad.aljneibi@ku.ac.ae | Phone: +971 2 312 4278 | " "Location: SAN Habshan Building, GF Room 5017AWS01" ), "tokens": [ "suaad", "jneibi", "al jneibi", "suaad al jneibi", "suuad", "suaad jneibi", ], }, { "full_name": "Shamsa Saeed Salem Alhosani", "role": "Library Information Senior Specialist", "details": ( "She can help with library cataloguing and general library assistance. " "Email: shamsa.alhosani@ku.ac.ae | Phone: +971 2 312 3156 | " "Location: SAN Habshan Building, GF Room 5013WS04" ), "tokens": [ "shamsa", "alhosani", "shamsa alhosani", "shamsa saeed", "shamsa salem", "shamsa saeed salem alhosani", ], }, { "full_name": "William Nathaniel Hutchinson III", "role": "Library Information Senior Specialist", "details": ( "He can help with library cataloguing and general library assistance. " "Email: william.hutchinsoniii@ku.ac.ae | Phone: +971 2 312 3966" ), "tokens": [ "william", "hutchinson", "william hutchinson", "william nathaniel", "hutchinson iii", "william hutchinson iii", ], }, { "full_name": "Shaikha Mohamed Saleem Almenhali", "role": "Library Information Specialist", "details": ( "She can help with general library assistance. " "Email: shaikha.almenhali@ku.ac.ae | Phone: +971 2 312 5874" ), "tokens": [ "shaikha", "almenhali", "shaikha almenhali", "shaikha mohamed", "shaikha saleem", "shaikha mohamed saleem almenhali", ], }, { "full_name": "Wilfredo Arcilla", "role": "Library Information Specialist", "details": ( "He can help with general library assistance. " "Email: wilfredo.arcilla@ku.ac.ae | Phone: +971 2 312 3323 | " "Location: SAN Habshan Building, GF Room 5011WS05" ), "tokens": [ "wilfredo", "arcilla", "wilfredo arcilla", ], }, { "full_name": "Zainab Al Ali", "role": "Library Information Senior Specialist", "details": ( "She can help with general library assistance. " "Email: zainab.alali@ku.ac.ae | Phone: +971 2 312 4321 | " "Location: SAN Habshan Building, GF Room 5017AWS02" ), "tokens": [ "zainab", "al ali", "zainab al ali", "alali", "zainab alali", ], }, { "full_name": "Khawla Alhemeiri", "role": "Library Information Senior Specialist", "details": ( "She can help with general library assistance. " "Email: khawla.alhemeiri@ku.ac.ae | Phone: +971 2 312 3959 | " "Location: Main E Building, 1st Floor, Room E01035" ), "tokens": [ "khawla", "alhemeiri", "khawla alhemeiri", ], }, { "full_name": "Meera K. Alnaqbi", "role": "Library Information Specialist", "details": ( "She can help with acquisitions, new title requests, collection development, " "and general library assistance. " "Email: meera.alnaqbi@ku.ac.ae | Phone: +971 2 312 3375" ), "tokens": [ "meera", "alnaqbi", "meera alnaqbi", "meera k alnaqbi", ], }, ] def _normalize_name_query(text: str): return [t for t in re.sub(r"[^a-z0-9 ]+", " ", (text or "").lower()).split() if t] def _dedupe_keep_order(items): seen = set() out = [] for item in items: if item and item not in seen: seen.add(item) out.append(item) return out def _title_case_name(name: str) -> str: return re.sub(r"\s+", " ", (name or "").strip()).title() def _build_staff_tokens(full_name: str): honorifics = {"dr", "mr", "mrs", "ms", "prof"} raw_tokens = _normalize_name_query(full_name) core_tokens = [t for t in raw_tokens if t not in honorifics] token_lists = [raw_tokens, core_tokens] variants = [] for toks in token_lists: if not toks: continue variants.extend(toks) variants.append(" ".join(toks)) for n in (2, 3, 4): if len(toks) >= n: for i in range(len(toks) - n + 1): variants.append(" ".join(toks[i:i+n])) return _dedupe_keep_order(variants) def _parse_staff_directory_text(text: str): staff_entries = [] if not text: return staff_entries lines = [line.rstrip() for line in text.splitlines()] i = 0 while i < len(lines): line = lines[i].strip() is_name_line = ( line and line == line.upper() and not line.startswith("===") and not line.startswith("SOURCE:") and not line.startswith("TITLE:") and any(ch.isalpha() for ch in line) and len(line.split()) <= 10 ) if not is_name_line: i += 1 continue name_line = line block = [] i += 1 while i < len(lines): nxt = lines[i].strip() next_is_name = ( nxt and nxt == nxt.upper() and not nxt.startswith("===") and not nxt.startswith("SOURCE:") and not nxt.startswith("TITLE:") and any(ch.isalpha() for ch in nxt) and len(nxt.split()) <= 10 ) if next_is_name: break block.append(nxt) i += 1 role = "" email = "" phone = "" mobile = "" location = "" best_for = "" schedule = "" extra_bits = [] for raw in block: if not raw or raw.startswith("==="): continue low = raw.lower() if raw.startswith("Title:"): role = raw.split(":", 1)[1].strip() elif raw.startswith("Email:"): email = raw.split(":", 1)[1].strip() elif raw.startswith("Phone:") or raw.startswith("Work Phone:"): phone = raw.split(":", 1)[1].strip() elif raw.startswith("Mobile:"): mobile = raw.split(":", 1)[1].strip() elif raw.startswith("Location:"): location = raw.split(":", 1)[1].strip() elif raw.startswith("Best for:"): best_for = raw.split(":", 1)[1].strip() elif raw.startswith("Schedule appointment:"): schedule = raw.split(":", 1)[1].strip() elif any(low.startswith(prefix) for prefix in ["linkedin:", "orcid:"]): extra_bits.append(raw) else: extra_bits.append(raw) details_parts = [] if best_for: details_parts.append(f"Best for: {best_for}") if email: details_parts.append(f"Email: {email}") if phone: details_parts.append(f"Phone: {phone}") if mobile: details_parts.append(f"Mobile: {mobile}") if location: details_parts.append(f"Location: {location}") if schedule: details_parts.append(f"Schedule appointment: {schedule}") details_parts.extend(extra_bits) full_name = _title_case_name(name_line) staff_entries.append({ "full_name": full_name, "role": role or "Library staff member", "details": " | ".join(_dedupe_keep_order(details_parts)), "tokens": _build_staff_tokens(full_name), "source_title": "Khalifa University Library Staff Directory and Contacts", "source": "https://library.ku.ac.ae/librarystaff", }) return staff_entries def _load_staff_directory_from_kb(): entries = [] try: for filepath in glob.glob(os.path.join(KNOWLEDGE_DIR, "*.txt")): name = os.path.basename(filepath).lower() if "staff" not in name and "contact" not in name: continue with open(filepath, "r", encoding="utf-8") as f: content = f.read() if "staff directory" not in content.lower() and "library staff" not in content.lower(): continue entries.extend(_parse_staff_directory_text(content)) except Exception as e: print(f"Staff KB parse error: {e}") deduped = [] seen = set() for entry in entries: key = entry.get("full_name", "").lower() if key and key not in seen: seen.add(key) deduped.append(entry) return deduped def _staff_lookup_candidates(): return kb_staff_directory or STAFF_DIRECTORY def _role_key(name: str) -> str: return re.sub(r"[^a-z0-9]+", " ", (name or "").lower()).strip() def _build_role_aliases(staff: dict): role = (staff.get("role") or "").strip() details = (staff.get("details") or "").strip() full_name = staff.get("full_name", "") key = _role_key(full_name) aliases = [] if role: aliases.append(role) role_l = role.lower() parts = [p.strip() for p in re.split(r"[\/,]|\band\b", role_l) if p.strip()] aliases.extend(parts) if "manager" in role_l: mgr_tail = re.sub(r"^manager\s*,?\s*", "", role_l).strip() if mgr_tail: aliases.extend([ mgr_tail, f"{mgr_tail} librarian", f"{mgr_tail.rstrip('s')} librarian", f"{mgr_tail} manager", ]) if "librarian" in role_l: base = role_l.replace("librarian", "").replace("/", " ").strip(" ,") if base: aliases.extend([ base, f"{base} librarian", f"{base.rstrip('s')} librarian", ]) best_for = "" m = re.search(r"Best for:\s*([^|]+)", details, re.IGNORECASE) if m: best_for = m.group(1).strip() if best_for: aliases.extend([p.strip() for p in best_for.split(",") if p.strip()]) manual = { "nikesh narayanan": [ "research librarian", "access services librarian", "research and access services librarian", "research access librarian", "orcid librarian", "research impact librarian", "bibliometrics librarian", "scholarly communication librarian", "khazna librarian", "scival librarian", ], "walter brian hall": [ "systems librarian", "system librarian", "library systems librarian", "digital services librarian", "technology services librarian", "website librarian", "technology librarian", "digital librarian", "open access librarian", "open access", "oa librarian", "apc librarian", "article processing charge", "general librarian", "public services librarian", ], "rani anand": [ "e resources librarian", "e-resources librarian", "electronic resources librarian", "database librarian", "database access librarian", "resource access librarian", ], "jason fetty": [ "medical librarian", "health sciences librarian", "clinical librarian", "systematic review librarian", "pubmed librarian", ], "alia al harrasi": [ "acquisitions librarian", "acquisition librarian", "technical services librarian", "technical service librarian", "collection development librarian", "cataloguing librarian", "metadata librarian", ], "muna ahmad mohammad al blooshi": [ "public services librarian", "public service librarian", "circulation librarian", "access services manager", "service desk librarian", ], "dr abdulla al hefeiti": [ "library director", "director of library", "director libraries", "assistant provost libraries", ], "abdulla al hefeiti": [ "library director", "director of library", "director libraries", "assistant provost libraries", ], # ── New staff role aliases ────────────────────────────────── "suaad al jneibi": [ "ill librarian", "interlibrary loan librarian", "ill specialist", "ill contact", "document delivery", "ill staff", ], "shamsa saeed salem alhosani": [ "cataloguing specialist", "cataloging specialist", "cataloguer", "library cataloguing", "library cataloging", ], "william nathaniel hutchinson iii": [ "cataloguing specialist", "cataloging specialist", "cataloguer", "library cataloguing", "library cataloging", ], "meera k alnaqbi": [ "acquisitions specialist", "acquisitions staff", "acquisition specialist", "new title request", "book request", "purchase request", "collection development specialist", ], "meera k. alnaqbi": [ "acquisitions specialist", "acquisitions staff", "acquisition specialist", "new title request", "book request", "purchase request", "collection development specialist", ], } aliases.extend(manual.get(key, [])) normalized = [] for a in aliases: a = re.sub(r"[^a-z0-9&/ +()-]+", " ", (a or "").lower()) a = re.sub(r"\s+", " ", a).strip(" ,") if a: normalized.append(a) if a.endswith(" services"): normalized.append(a[:-1]) if a.endswith(" service"): normalized.append(a + " librarian") if a.endswith(" resources"): normalized.append(a[:-1]) return _dedupe_keep_order(normalized) def _match_staff_role(question: str): ql = re.sub(r"[^a-z0-9 ]+", " ", (question or "").lower()) ql = re.sub(r"\s+", " ", ql).strip() if not ql: return None role_trigger_re = re.compile( r"\b(librarian|library director|director|manager|services?|service|systems?|technology|website|" r"research|access|open access|orcid|bibliometric|bibliometrics|public services?|" r"technical services?|acquisitions?|collection development|catalogu(?:e|ing)|metadata|" r"database|e resources|e-resources|electronic resources|medical|clinical|systematic review|" r"circulation|interlibrary loan|ill|document delivery|" r"cataloging|cataloguer|general library|general service)\b" ) if not role_trigger_re.search(ql): return None best = None best_score = 0 for staff in _staff_lookup_candidates(): score = 0 for alias in _build_role_aliases(staff): if not alias: continue alias_words = alias.split() if alias in ql: score = max(score, 100 + len(alias_words)) elif len(alias_words) >= 2 and all(w in ql.split() for w in alias_words): score = max(score, 70 + len(alias_words)) else: overlap = sum(1 for w in alias_words if len(w) > 2 and w in ql.split()) if overlap >= 2: score = max(score, 40 + overlap) role_words = _normalize_name_query(staff.get("role", "")) if role_words: overlap = sum(1 for w in role_words if len(w) > 2 and w in ql.split()) if overlap >= 2: score = max(score, 20 + overlap) if score > best_score: best_score = score best = staff return best if best_score >= 42 else None def _match_staff_name(question: str): tokens = _normalize_name_query(question) if not tokens or len(tokens) > 5: return None ql = (question or "").strip().lower() blocked = ["who is", "who handles", "who can help", "contact", "librarian", "library"] if any(b in ql for b in blocked): return None for staff in _staff_lookup_candidates(): staff_tokens = set() for tok in staff.get("tokens", []): staff_tokens.update(_normalize_name_query(tok)) staff_tokens.add(" ".join(_normalize_name_query(tok))) if all(tok in staff_tokens for tok in tokens): return staff return None def _staff_name_answer(staff: dict, partial: str) -> str: return ( f"Did you mean {staff['full_name']}?

" f"{staff['full_name']} is the {staff['role']}. {staff['details']}" ) def _staff_role_answer(staff: dict, question: str) -> str: return ( f"For {question}, the best match is {staff['full_name']} — " f"{staff['role']}.

{staff['details']}" ) GROUNDED_LIBRARY_MAP = { "ill": "interlibrary loan ILL document delivery request article request book unavailable not in collection borrow from another library Suaad Al Jneibi", "fulltext": "full text libkey nomad article access pdf article not available no full text get full text request article", "primo": "PRIMO discovery catalog holdings publication finder library has this journal books articles search catalog", "circulation": "borrowing circulation renew renewal loan period due date hold request fines fees my library account checked out overdue", "research_help": "Ask a Librarian research consultation reference help subject guides research skills instruction tours workshops", "orcid_oa": "ORCID open access APC publishing research impact Scopus SciVal bibliometrics Walter Brian Hall openaccess@ku.ac.ae", "database_access": "database access e-resources remote access off campus login vendor issue Rani Anand e-books patents", "medical_help": "medical librarian Jason Fetty PubMed Embase CINAHL Cochrane UpToDate systematic review clinical databases", "systems_help": "Walter Brian Hall systems website technology digital services library systems coding STEAM open access APC", "acquisitions": "Alia Al-Harrasi Meera Alnaqbi acquisitions request title suggest a book collection development technical services cataloguing metadata", "cataloguing": "Shamsa Alhosani William Hutchinson Alia Al-Harrasi cataloguing cataloging metadata technical services", "rooms": "study room reserve a room room booking group study rooms libcal book room room reservation", "visitors_alumni": "visitors visitor registration alumni alumni services external visitors access policy", "printing": "printing scanning photocopying printers scanner copy services", "refworks": "RefWorks citation manager reference management citations bibliography write n cite", "hours_locations": "library hours opening hours closing hours locations main campus Habshan SAN Sas Al Nakhl", "standards": "ASTM Compass IEEE standards ASME ASCE engineering standards", "theses": "ProQuest Dissertations theses Khazna institutional repository dissertations theses", "metrics": "impact factor journal citation reports JCR CiteScore Scopus SciVal metrics", "chemistry": "ACS SciFindern RSC Reaxys chemistry database", "physics": "APS AIP IOPScience physics database", "engineering": "IEEE Xplore ACM ASCE ASME Knovel engineering databases", } # ===== GLOBALS ===== vectorstore = None http_client = None kb_staff_directory = [] # ===== ANALYTICS DB ===== def init_db(): conn = sqlite3.connect(DB_PATH) conn.execute('''CREATE TABLE IF NOT EXISTS queries ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT, question TEXT, tool_used TEXT, model TEXT, response_time REAL, result_count INTEGER, error TEXT )''') conn.execute('''CREATE TABLE IF NOT EXISTS bot_config ( key TEXT PRIMARY KEY, value TEXT, updated_at TEXT )''') conn.execute('''CREATE TABLE IF NOT EXISTS feedback ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT, question TEXT, feedback TEXT, answer_excerpt TEXT, intent TEXT )''') # Set defaults if not exist defaults = { "welcome_message": "Hi! I'm the Khalifa University Library AI Assistant. I can help you find articles, books, databases, and answer questions about library services. How can I help you today?", "bot_personality": "You are a helpful, friendly, and knowledgeable library assistant at Khalifa University, Abu Dhabi, UAE. KU = Khalifa University, NOT Kuwait University. Be concise (2-4 sentences). Always include relevant URLs.", "custom_instructions": "", "announcement": "", "max_results": "5", "default_model": "gpt", "maintenance_mode": "false", "maintenance_message": "The library chatbot is currently undergoing maintenance. Please try again later.", } for k, v in defaults.items(): conn.execute( "INSERT OR IGNORE INTO bot_config (key, value, updated_at) VALUES (?, ?, ?)", (k, v, datetime.utcnow().isoformat()) ) conn.commit() conn.close() def get_config(key, default=""): try: conn = sqlite3.connect(DB_PATH) row = conn.execute("SELECT value FROM bot_config WHERE key=?", (key,)).fetchone() conn.close() return row[0] if row else default except: return default def set_config(key, value): conn = sqlite3.connect(DB_PATH) conn.execute( "INSERT OR REPLACE INTO bot_config (key, value, updated_at) VALUES (?, ?, ?)", (key, value, datetime.utcnow().isoformat()) ) conn.commit() conn.close() # ===== ADMIN AUTH ===== # ADMIN_PASSWORD must be set as HF Space Secret — no insecure fallback ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "") if not ADMIN_PASSWORD: import warnings warnings.warn("ADMIN_PASSWORD secret is not set. Admin dashboard will be inaccessible until configured.") admin_sessions = {} # token -> expiry timestamp START_TIME = time.time() # for uptime tracking def create_session(): token = secrets.token_hex(32) admin_sessions[token] = time.time() + 86400 # 24 hour session return token def verify_session(request: Request): token = request.cookies.get("admin_session") if not token or token not in admin_sessions: return False if time.time() > admin_sessions[token]: del admin_sessions[token] return False return True def log_query(question, tool, model, response_time, result_count=0, error=None): try: conn = sqlite3.connect(DB_PATH) conn.execute( 'INSERT INTO queries (timestamp,question,tool_used,model,response_time,result_count,error) VALUES (?,?,?,?,?,?,?)', (datetime.utcnow().isoformat(), question, tool, model, response_time, result_count, error) ) conn.commit() conn.close() except Exception as e: print(f"Log error: {e}") # ===== RAG SETUP ===== def load_documents(): global kb_staff_directory docs = [] files = glob.glob(os.path.join(KNOWLEDGE_DIR, "*.txt")) kb_staff_directory = _load_staff_directory_from_kb() if kb_staff_directory: print(f"Loaded {len(kb_staff_directory)} staff entries from KB") for filepath in files: try: with open(filepath, "r", encoding="utf-8") as f: content = f.read() lines = content.split("\n", 3) source, title, text = "", "", content for line in lines[:2]: if line.startswith("SOURCE:"): source = line.replace("SOURCE:", "").strip() elif line.startswith("TITLE:"): title = line.replace("TITLE:", "").strip() if source or title: text = "\n".join(lines[2:]).strip() docs.append(Document(page_content=text, metadata={"source": source, "title": title})) except Exception as e: print(f"Error loading {filepath}: {e}") print(f"Loaded {len(docs)} documents") return docs def build_vectorstore(docs, force_rebuild=False): embeddings = OpenAIEmbeddings(model="text-embedding-3-small") index_file = os.path.join(FAISS_INDEX_PATH, "index.faiss") # On startup: load existing index if available (saves cost + time) if not force_rebuild and os.path.exists(index_file): print(f"Loading existing FAISS index from {FAISS_INDEX_PATH}") store = FAISS.load_local(FAISS_INDEX_PATH, embeddings, allow_dangerous_deserialization=True) print(f"Loaded existing index with {store.index.ntotal} chunks") return store # Force rebuild: always re-embed all documents from scratch print(f"Building new FAISS index from {len(docs)} documents...") splitter = RecursiveCharacterTextSplitter(chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP) chunks = splitter.split_documents(docs) print(f"Split into {len(chunks)} chunks, embedding now...") store = FAISS.from_documents(chunks, embeddings) os.makedirs(FAISS_INDEX_PATH, exist_ok=True) store.save_local(FAISS_INDEX_PATH) print(f"FAISS index saved with {store.index.ntotal} chunks") return store # ===== TOOL: SEARCH PRIMO ===== async def tool_search_primo(query, limit=5, peer_reviewed=False, open_access=False, year_from=None, year_to=None): api_key = os.environ.get("PRIMO_API_KEY") if not api_key: return {"error": "PRIMO_API_KEY not configured", "results": []} vid = "971KUOSTAR_INST:KU" facets = "" if peer_reviewed: facets += "&qInclude=facet_tlevel,exact,peer_reviewed" if open_access: facets += "&qInclude=facet_tlevel,exact,open_access" if year_from or year_to: yf = year_from or "1900" yt = year_to or str(datetime.now().year) facets += f"&multiFacets=facet_searchcreationdate,include,{yf}%7C,%7C{yt}" base = "https://api-eu.hosted.exlibrisgroup.com/primo/v1/search" qs = f"?vid={vid}&tab=Everything&scope=MyInst_and_CI&q=any,contains,{query}&lang=en&sort=rank&limit={limit}&offset=0&apikey={api_key}{facets}" async with httpx.AsyncClient(timeout=15) as client: for region in ["api-eu", "api-na", "api-ap"]: url = base.replace("api-eu", region) + qs try: r = await client.get(url, headers={"Accept": "application/json"}) if r.status_code == 200: data = r.json() total = data.get("info", {}).get("total", 0) results = [] for doc in data.get("docs", []): d = doc.get("pnx", {}).get("display", {}) a = doc.get("pnx", {}).get("addata", {}) s = doc.get("pnx", {}).get("search", {}) results.append({ "title": (d.get("title") or ["Untitled"])[0], "creator": "; ".join(d.get("creator") or d.get("contributor") or []) or "Unknown", "date": (s.get("creationdate") or a.get("risdate") or a.get("date") or [""])[0], "type": (d.get("type") or [""])[0], "source": (d.get("source") or a.get("jtitle") or [""])[0], "description": ((d.get("description") or [""])[0] or "")[:400], "doi": (a.get("doi") or [None])[0], }) return {"total": total, "results": results, "source": "PRIMO"} except Exception: continue return {"error": "PRIMO API unavailable", "results": [], "source": "PRIMO"} # ===== MCP HELPER ===== async def _call_mcp(api_key, prompt, mcp_url, mcp_name, timeout=30): """Call Anthropic API with an MCP server. Returns text content or raises.""" async with httpx.AsyncClient(timeout=timeout) as client: r = await client.post( "https://api.anthropic.com/v1/messages", headers={ "Content-Type": "application/json", "x-api-key": api_key, "anthropic-version": "2023-06-01", "anthropic-beta": "mcp-client-2025-04-04", }, json={ "model": "claude-sonnet-4-20250514", "max_tokens": 2000, "messages": [{"role": "user", "content": prompt}], "mcp_servers": [{"type": "url", "url": mcp_url, "name": mcp_name}], }, ) if r.status_code != 200: raise Exception(f"Anthropic API: {r.status_code}") data = r.json() parts = [] for block in data.get("content", []): if block.get("type") == "text": parts.append(block["text"]) elif block.get("type") == "mcp_tool_result": for item in block.get("content", []): if item.get("type") == "text": parts.append(item["text"]) return "\n".join(parts) async def _parse_results(api_key, raw_text, source_name, limit=5): """Parse raw MCP results into structured JSON.""" async with httpx.AsyncClient(timeout=20) as client: r = await client.post( "https://api.anthropic.com/v1/messages", headers={"Content-Type": "application/json", "x-api-key": api_key, "anthropic-version": "2023-06-01"}, json={ "model": "claude-sonnet-4-20250514", "max_tokens": 2000, "messages": [{"role": "user", "content": f"""Extract search results into a JSON array. Each item: title, creator, date, source, doi, link, description, type. Return ONLY valid JSON array. No markdown. If none, return []. Text: {raw_text[:3000]}"""}], }, ) if r.status_code != 200: return [] text = "".join(b["text"] for b in r.json().get("content", []) if b.get("type") == "text").strip() if text.startswith("```"): text = text.split("\n", 1)[1].rsplit("```", 1)[0].strip() results = json.loads(text) for item in results: item["_source"] = source_name if item.get("doi") and not item.get("link"): item["link"] = f"https://doi.org/{item['doi']}" return results[:limit] # ===== TOOL: SEARCH PUBMED (direct NCBI E-utilities — free, reliable) ===== async def tool_search_pubmed(query, limit=5): base = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils" try: async with httpx.AsyncClient(timeout=15) as client: r = await client.get(f"{base}/esearch.fcgi?db=pubmed&term={query}&retmax={limit}&retmode=json&sort=relevance") if r.status_code != 200: return {"error": "PubMed search failed", "results": [], "source": "PubMed"} ids = r.json().get("esearchresult", {}).get("idlist", []) if not ids: return {"total": 0, "results": [], "source": "PubMed"} r2 = await client.get(f"{base}/esummary.fcgi?db=pubmed&id={','.join(ids)}&retmode=json") if r2.status_code != 200: return {"error": "PubMed fetch failed", "results": [], "source": "PubMed"} data = r2.json().get("result", {}) results = [] for pid in ids: rec = data.get(pid, {}) if not isinstance(rec, dict): continue authors = ", ".join(a.get("name", "") for a in rec.get("authors", [])[:3]) if len(rec.get("authors", [])) > 3: authors += " et al." results.append({ "title": rec.get("title", ""), "creator": authors, "date": rec.get("pubdate", ""), "source": rec.get("fulljournalname", rec.get("source", "")), "doi": rec.get("elocationid", "").replace("doi: ", "") if "doi:" in rec.get("elocationid", "") else None, "pmid": pid, "link": f"https://pubmed.ncbi.nlm.nih.gov/{pid}/", "type": "Journal Article", "_source": "PubMed", }) total = int(r.json().get("esearchresult", {}).get("count", 0)) return {"total": total, "results": results, "source": "PubMed"} except Exception as e: return {"error": f"PubMed: {str(e)}", "results": [], "source": "PubMed"} # ===== TOOL: SEARCH CONSENSUS (via Semantic Scholar with consensus framing) ===== async def tool_search_consensus(query, limit=5): """ Consensus.app requires OAuth so we can't call it directly. Instead we search Semantic Scholar with the same query and return results alongside a direct Consensus deep-link so the user can also check the AI-synthesized answer there. """ scholar = await tool_search_scholar(query, limit) scholar["source"] = "Consensus / Semantic Scholar" scholar["consensus_url"] = f"https://consensus.app/results/?q={query}" scholar["message"] = "Results from Semantic Scholar. For AI-synthesized consensus, click the Consensus link." return scholar # ===== TOOL: SEARCH SEMANTIC SCHOLAR (direct API — free, no auth) ===== async def tool_search_scholar(query, limit=5): """Search 200M+ papers via Semantic Scholar API. Free, no API key needed.""" url = "https://api.semanticscholar.org/graph/v1/paper/search/bulk" params = { "query": query, "limit": min(limit, 20), "fields": "title,authors,year,venue,externalIds,abstract,citationCount,openAccessPdf,publicationTypes", } try: async with httpx.AsyncClient(timeout=15) as client: r = await client.get(url, params=params, headers={"Accept": "application/json"}) if r.status_code != 200: return {"error": f"Semantic Scholar API: {r.status_code}", "results": [], "source": "Semantic Scholar"} data = r.json() total = data.get("total", 0) results = [] for paper in data.get("data", [])[:limit]: authors = ", ".join(a.get("name", "") for a in (paper.get("authors") or [])[:3]) if len(paper.get("authors") or []) > 3: authors += " et al." ext_ids = paper.get("externalIds") or {} doi = ext_ids.get("DOI") link = None if paper.get("openAccessPdf", {}).get("url"): link = paper["openAccessPdf"]["url"] elif doi: link = f"https://doi.org/{doi}" else: s2_id = paper.get("paperId") if s2_id: link = f"https://www.semanticscholar.org/paper/{s2_id}" results.append({ "title": paper.get("title", ""), "creator": authors, "date": str(paper.get("year", "")), "source": paper.get("venue", ""), "description": ((paper.get("abstract") or "")[:400]), "doi": doi, "link": link, "citations": paper.get("citationCount", 0), "type": (paper.get("publicationTypes") or ["Article"])[0] if paper.get("publicationTypes") else "Article", "open_access": bool(paper.get("openAccessPdf")), "_source": "Semantic Scholar", }) return {"total": total, "results": results, "source": "Semantic Scholar"} except Exception as e: return {"error": f"Semantic Scholar: {str(e)}", "results": [], "source": "Semantic Scholar"} async def tool_library_info(question, history=None, model="gpt"): if not vectorstore: return {"answer": "Knowledge base not initialized.", "sources": [], "has_answer": False} # Build search query — use question + history context base_query = question if history: history_text = "\n".join(f"{'User' if m['role']=='user' else 'Assistant'}: {m['content']}" for m in history[-3:]) base_query = f"{history_text}\n{question}" # ── Semantic query expansion via lightweight LLM interpreter + grounded institutional map ── interp = await _interpret_semantics(question, history or []) grounding_text = " ".join(GROUNDED_LIBRARY_MAP.get(k, "") for k in interp.get("grounding_keys", [])) canonical_text = " ".join(interp.get("canonical_terms", [])) expanded_query = " ".join(x for x in [base_query, canonical_text, grounding_text] if x).strip() # ── FAISS scored search ── docs_with_scores = vectorstore.similarity_search_with_score(expanded_query, k=TOP_K) if not docs_with_scores: return {"answer": "", "sources": [], "has_answer": False} best_score = min(score for _, score in docs_with_scores) print(f"RAG best_score={best_score:.3f} for: {question[:60]}") # More forgiving threshold for library-service questions; still reject clearly unrelated queries score_threshold = 1.45 if interp.get("intent_hint") == "library_info": score_threshold = 1.7 if best_score > score_threshold: print(f"RAG skipped — score {best_score:.3f} exceeds {score_threshold:.2f} threshold") return {"answer": "", "sources": [], "has_answer": False} docs = [doc for doc, _ in docs_with_scores] context = "\n\n".join(d.page_content for d in docs) # If score is moderate (0.8–1.4), the match may be close but not exact. # Instruct the LLM to add a "Did you mean...?" clarification if it interprets the question loosely. moderate_match = 0.8 < best_score <= 1.4 did_you_mean_instruction = """ - If you are answering a question that uses different wording from what's in the context (e.g. user asked "research librarian" but context says "Research and Access Services Librarian"), start your answer with: "Did you mean [exact title from context]? If so, here is the information:" then give the answer.""" if moderate_match else "" prompt = f"""You are the Khalifa University Library AI Assistant in Abu Dhabi, UAE. KU means Khalifa University, NOT Kuwait University. RULES — follow exactly: 1. Answer ONLY using the context provided below. 2. If the context has relevant information → answer in 2-4 sentences, include URLs if present.{did_you_mean_instruction} 3. If the context does NOT contain the answer → output the single word: NO_LIBRARY_ANSWER - Do NOT write "I'm sorry" - Do NOT write "I don't have information" - Do NOT suggest contacting the library - Do NOT apologise - ONLY output: NO_LIBRARY_ANSWER Context: {context} Question: {question} Answer:""" # Respect the model selection — use Claude if requested, GPT otherwise use_claude = model == "claude" and os.environ.get("ANTHROPIC_API_KEY") if use_claude: from langchain_anthropic import ChatAnthropic llm = ChatAnthropic(model="claude-haiku-4-5-20251001", temperature=0.2, max_tokens=500) else: llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2, max_tokens=500) response = llm.invoke(prompt) answer = response.content.strip() # LLM signalled it couldn't answer from context if answer == "NO_LIBRARY_ANSWER" or answer.startswith("NO_LIBRARY_ANSWER"): return {"answer": "", "sources": [], "has_answer": False} # Catch cases where LLM ignored the NO_LIBRARY_ANSWER instruction # and generated a polite "I don't have info" response instead # Normalise smart/curly apostrophes to straight before pattern matching answer_normalised = answer.lower().replace("’", "'").replace("‘", "'").replace("“", '"').replace("”", '"') no_info_patterns = [ "i don't have", "i do not have", "i'm sorry", "i am sorry", "i apologize", "i apologise", "no specific information", "not available in", "not in the context", "context does not contain", "context doesn't contain", "i cannot find", "i could not find", "don't have information", "do not have information", "unable to find", "unable to provide", "not found in", "no information", "for further inquiries", "for more information, please contact", "please visit ask", "please contact the library", "recommend checking", "i'd suggest", "i would suggest contacting", ] if any(p in answer_normalised for p in no_info_patterns): print(f"RAG no-info pattern matched: '{answer_normalised[:80]}'") return {"answer": "", "sources": [], "has_answer": False} sources = [] seen = set() for d in docs: src = d.metadata.get("source", "") title = d.metadata.get("title", "") key = src or title if key and key not in seen: seen.add(key) sources.append({"title": title, "source": src}) return {"answer": answer, "sources": sources, "has_answer": True} # ===== DETECT MEDICAL TOPIC ===== def is_medical_topic(text): lower = text.lower() return any(kw in lower for kw in MEDICAL_KEYWORDS) # ===== STARTUP ===== @asynccontextmanager async def lifespan(app: FastAPI): global vectorstore, http_client print("=== Starting KU Library AI Agent ===") init_db() docs = load_documents() if docs: vectorstore = build_vectorstore(docs) print(f"Vector store ready with {vectorstore.index.ntotal} vectors") http_client = httpx.AsyncClient(timeout=15) yield await http_client.aclose() print("Shutting down...") # ===== FASTAPI ===== app = FastAPI(title="KU Library AI Agent", lifespan=lifespan) app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) # ===== MODELS ===== class ChatMessage(BaseModel): role: str content: str class SearchRequest(BaseModel): query: str source: str = "primo" # primo, pubmed, scholar, consensus, all model: str = "gpt" limit: int = 5 peer_reviewed: bool = False open_access: bool = False year_from: str | None = None year_to: str | None = None class RAGRequest(BaseModel): question: str model: str = "gpt" history: list[ChatMessage] = [] class AgentRequest(BaseModel): question: str model: str = "gpt" history: list[ChatMessage] = [] # ===== ENDPOINTS ===== @app.get("/") def health(): return { "status": "ok", "vectorstore_ready": vectorstore is not None, "tools": ["search_primo", "search_pubmed", "search_scholar", "search_consensus", "get_library_info"], "endpoints": ["/rag", "/search", "/agent", "/general", "/config", "/year"], "models": { "gpt": bool(os.environ.get("OPENAI_API_KEY")), "claude": bool(os.environ.get("ANTHROPIC_API_KEY")), }, } # Fix 6: authoritative server-side year — frontend uses this instead of client device clock @app.get("/year") def get_year(): now = datetime.utcnow() return { "year": now.year, "month": now.month, "date": now.strftime("%Y-%m-%d"), } # ---- Spell correction + Boolean building (uses HF Space OpenAI — separate quota from Cloudflare) ---- class CorrectRequest(BaseModel): query: str year: int = 2026 def _extract_json_object(text: str) -> dict: s, e = text.find("{"), text.rfind("}") if s == -1 or e == -1 or e <= s: raise ValueError("No JSON object found") return json.loads(text[s:e+1]) def _make_boolean(text: str) -> str: """Build a basic Boolean from text by extracting meaningful phrase groups.""" import re as re2 stop = re2.compile( r'^(of|on|in|the|a|an|and|or|for|to|with|by|from|at|is|are|was|were|be|been|' r'being|have|has|had|do|does|did|will|would|could|should|may|its|this|that|' r'these|those|about|impact|effect|role|influence|importance|use|application|' r'study|analysis|review|what|how|why|when|where|which)$', re2.IGNORECASE ) words = text.split() concepts, phrase = [], [] for w in words: clean = w.strip("'\"") if not stop.match(clean) and len(clean) > 2: phrase.append(clean) else: if phrase: concepts.append(' '.join(phrase)) phrase = [] if phrase: concepts.append(' '.join(phrase)) if len(concepts) >= 2: return ' AND '.join(f'"{c}"' for c in concepts) if len(concepts) == 1: return f'"{concepts[0]}"' return f'"{text}"' def _clean_database_keywords(boolean_query: str) -> str: return re.sub(r'\s+', ' ', re.sub(r'\b(AND|OR|NOT)\b|[()"]', ' ', boolean_query, flags=re.IGNORECASE)).strip() async def _build_search_plan(query: str, year: int = 2026) -> dict: prompt = f"""You are a search expert for Khalifa University Library. The user typed a query that may be messy, fragmented, or use chat phrasing. Create THREE forms: 1. corrected: spell-fixed version of the core topic (remove chat fragments like "and for", "also about") 2. natural: a MEANINGFUL research phrase for AI tools (Consensus, Perplexity, Semantic Scholar, LeapSpace). - Must be a proper research question or phrase — NOT raw keywords, NOT Boolean - Should add helpful context: "physics" → "recent advances in physics research" - If input is fragmented ("and for physics", "also about AI") extract topic and expand it - Aim for 5-10 words that a researcher would actually type into an AI tool 3. boolean: PRIMO/PubMed Boolean with AND/OR/parentheses for traditional database search Examples: Input: "and for physics" → corrected:"physics", natural:"recent advances and developments in physics research", boolean:("physics" OR "physical sciences") Input: "impuct glubal waming" → corrected:"impact global warming", natural:"impact of global warming on environment and climate", boolean:("global warming" OR "climate change") AND (impact OR effect) Input: "machne lerning helthcare" → corrected:"machine learning healthcare", natural:"machine learning applications in healthcare and medicine", boolean:("machine learning" OR "deep learning" OR AI) AND (healthcare OR clinical OR medical) Input: "renewable energy 2023" → corrected:"renewable energy 2023", natural:"renewable energy sources and sustainability research", boolean:("renewable energy" OR "clean energy" OR "solar energy"), year_from:"2023" Return ONLY valid JSON: {{"corrected":"spell-fixed core topic","natural":"meaningful 5-10 word research phrase","boolean":"(A OR B) AND (C OR D)","year_from":"","year_to":"","peer_reviewed":false,"open_access":false}} Query: "{query}""" try: llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, max_tokens=300) response = llm.invoke(prompt) result = _extract_json_object(response.content.strip()) corrected = result.get("corrected", query).strip() or query natural = result.get("natural", corrected).strip() or corrected boolean = (result.get("boolean") or "").strip() has_operators = bool(re.search(r'\b(AND|OR)\b', boolean)) if boolean else False has_parens = "(" in boolean if boolean else False if not has_operators or not has_parens: boolean = _make_boolean(corrected) return { "corrected": corrected, "natural": natural, "boolean": boolean, "database_query": _clean_database_keywords(boolean), "year_from": result.get("year_from", "") or "", "year_to": result.get("year_to", "") or "", "peer_reviewed": bool(result.get("peer_reviewed", False)), "open_access": bool(result.get("open_access", False)), } except Exception: boolean = _make_boolean(query) corrected = query.strip() or query return { "corrected": corrected, "natural": corrected, "boolean": boolean, "database_query": _clean_database_keywords(boolean), "year_from": "", "year_to": "", "peer_reviewed": False, "open_access": False, } def get_behavior_instructions() -> str: base = get_config("bot_personality", "You are a helpful, friendly, and knowledgeable library assistant at Khalifa University, Abu Dhabi, UAE. KU = Khalifa University, NOT Kuwait University. Be concise (2-4 sentences). Always include relevant URLs.") extra = get_config("custom_instructions", "") return (base + ("\n" + extra if extra else "")).strip() def _looks_social_or_greeting(question: str) -> bool: # SOCIAL_RE is anchored with ^ so .match() is the correct call. # Using .search() with ^ worked most of the time but could silently # fail when the engine didn't treat ^ as start-of-string in search mode # with certain Unicode inputs — .match() is unambiguous. return bool(SOCIAL_RE.match((question or "").strip())) async def _interpret_semantics(question: str, history=None) -> dict: q = (question or "").strip() ql = q.lower() canonical_terms = [] grounding_keys = [] social = False def add(key: str, *terms: str): if key and key not in grounding_keys: grounding_keys.append(key) for t in terms: if t and t not in canonical_terms: canonical_terms.append(t) if _looks_social_or_greeting(q): social = True return {"intent_hint": "general", "canonical_terms": canonical_terms, "grounding_keys": grounding_keys, "social": True} if _looks_library_hours_question(q): add("circulation", "library hours", "opening hours", "closing hours") return {"intent_hint": "library_info", "canonical_terms": canonical_terms, "grounding_keys": grounding_keys, "social": False} if _looks_nonlibrary_ku_question(q): return {"intent_hint": "general", "canonical_terms": canonical_terms, "grounding_keys": grounding_keys, "social": False} # Staff / role semantics if re.search(r"\b(systems? librarian|system librarian|website|digital services|library systems?|technology help)\b", ql): add("systems_help", "Walter Brian Hall", "Systems Librarian", "website", "technology") if re.search(r"\b(database access|e-?resources?|remote access|off campus|off-campus|login issue|access problem|vendor issue)\b", ql): add("database_access", "Rani Anand", "E-Resources", "database access") if re.search(r"\b(orcid|open access|apc|article processing charge|research impact|bibliometric|bibliometrics|scival|scopus metrics?)\b", ql): add("orcid_oa", "Walter Brian Hall", "ORCID", "Open Access", "APC", "openaccess@ku.ac.ae") if re.search(r"\b(research support|research impact|bibliometrics|scival|khazna|scholarly communication|libguides)\b", ql): add("research_help", "Nikesh Narayanan", "research support", "bibliometrics", "Khazna") if re.search(r"\b(medical librarian|pubmed help|embase|cinahl|cochrane|uptodate|systematic review|clinical databases?)\b", ql): add("medical_help", "Jason Fetty", "Medical Librarian", "PubMed", "systematic review") if re.search(r"\b(acquisitions?|collection development|suggest a book|request a title|new title request|purchase request|book request)\b", ql): add("acquisitions", "Alia Al-Harrasi", "Meera Alnaqbi", "Acquisitions", "collection development") if re.search(r"\b(catalogu(?:e|ing)|cataloging|metadata|cataloguer)\b", ql): add("cataloguing", "Shamsa Alhosani", "William Hutchinson", "cataloguing", "cataloging", "metadata") if re.search(r"\b(circulation|borrow|renew|due date|fine|hold request|my library account|public services?)\b", ql): add("circulation", "borrowing", "renewal", "My Library Account") if re.search(r"\b(ask a librarian|research help|research consultation|reference help|consultation)\b", ql): add("research_help", "Ask a Librarian", "research consultation") # Service / resource semantics if re.search(r"\b(inter\s*library loan|interlibrary loan|\bill\b|document delivery|full text unavailable|no full text|can't get (an )?article|cannot get (an )?article|article not available)\b", ql): add("ill", "Suaad Al Jneibi", "Interlibrary Loan", "ILL", "document delivery") add("fulltext", "full text", "LibKey Nomad", "article access") if re.search(r"\b(primo|catalog|discovery|holdings|publication finder|does the library have)\b", ql): add("primo", "PRIMO", "catalog", "Publication Finder") # Database recommendation semantics if re.search(r"\b(standards?|astm|ieee standards?|asme|asce)\b", ql): add("standards", "ASTM Compass", "IEEE standards", "ASME", "ASCE") if re.search(r"\b(dissertations?|theses|khazna)\b", ql): add("theses", "ProQuest Dissertations & Theses", "Khazna") if re.search(r"\b(impact factor|journal metrics?|jcr|citescore|sjr)\b", ql): add("metrics", "Journal Citation Reports", "CiteScore") if re.search(r"\b(chemistry|chemical)\b", ql): add("chemistry", "ACS", "SciFindern", "Reaxys") if re.search(r"\b(physics|optics?|photonics?)\b", ql): add("physics", "APS", "AIP", "IOPScience") if re.search(r"\b(engineering|civil engineering|mechanical engineering|electrical engineering|computer science|ai)\b", ql): add("engineering", "IEEE Xplore", "ACM", "Knovel", "ASCE", "ASME") # Name/title style library questions if re.search(r"\b(who is|who handles|who can help|whom should i contact|contact for)\b", ql) and (_looks_library_question(q) or grounding_keys): if not grounding_keys: add("research_help", "staff directory", "library staff") # Decide intent if _looks_medical_search(q): intent = "search_medical" elif _looks_research_question(q): intent = "search_academic" elif grounding_keys or _looks_library_question(q): intent = "library_info" elif _looks_current_question(q): intent = "general_recent" else: intent = "general" return { "intent_hint": intent, "canonical_terms": canonical_terms, "grounding_keys": grounding_keys, "social": social, } def _looks_library_question(question: str) -> bool: return bool(LIBRARY_CUE_RE.search(question)) def _looks_current_question(question: str) -> bool: return bool(CURRENT_INFO_RE.search(question)) def _looks_research_question(question: str) -> bool: q = question.lower() if RESEARCH_CUE_RE.search(q): return True return bool(re.search(r'\bimpact of\b|\beffects? of\b|\bcauses? of\b|\brelationship between\b|\bliterature on\b', q)) def _looks_medical_search(question: str) -> bool: q = question.lower() if not is_medical_topic(q): return False return bool(MEDICAL_SEARCH_RE.search(q) or _looks_research_question(q)) def _looks_library_hours_question(question: str) -> bool: q = question.strip().lower() # Must match hours pattern AND have at least one library/time/open signal return bool(HOURS_RE.search(q)) and any( kw in q for kw in ( 'library', 'campus', 'habshan', 'san', 'hours', 'open', 'close', 'schedule', 'till', 'until', 'today', 'tomorrow', 'friday', 'saturday', 'sunday', 'week', 'ramadan' ) ) def _looks_campus_question(question: str) -> bool: """Returns True for questions about KU library locations/branches/campuses.""" return bool(KU_CAMPUS_RE.search(question)) def _looks_nonlibrary_ku_question(question: str) -> bool: q = question.strip().lower() if 'library' in q or 'librarian' in q or 'primo' in q or 'database' in q or 'book' in q or 'article' in q: return False return bool(KU_GENERAL_RE.search(question)) def _hours_redirect_answer() -> str: """ Hardcoded library hours + always direct to the hours page. Hours are from the KB (ku_library_san_campus.txt). Always advise to check the link since times can change for exams, holidays, and Ramadan. """ return ( "Here are the current Khalifa University Library hours:

" "Habshan Library (SAN Campus):
" "Sunday – Wednesday: 8:00 AM – 7:00 PM
" "Thursday: 8:00 AM – 4:00 PM
" "Friday: Closed
" "Saturday: 12:00 PM – 8:00 PM

" "⚠️ Hours may change during exam periods, public holidays, and Ramadan. " "Always check the official Library Hours page for the latest times:
" f"{LIBRARY_HOURS_URL}" ) def _ku_campus_answer() -> str: """ Hardcoded answer for any question about KU library locations/campuses. Never answered from web — always this fixed text. """ return ( "Khalifa University Library has two branches:

" "1. Main Campus Library
" "Location: E Building, Main Campus, Abu Dhabi
" "Phone: +971 2 312 4604

" "2. Habshan Library (SAN Campus)
" "Location: Building 5, Ground Floor & First Floor, Sas Al Nakhl (SAN) Campus, Abu Dhabi
" "Phone: +971 2 312 3160

" "For general library enquiries: " "Ask a Librarian " "or email libse@ku.ac.ae." ) def _ku_general_redirect_answer() -> str: return ( f"I’m LibBee, the Khalifa University Library AI Assistant, so I’m best for library resources, services, databases, research help, and staff contacts.

" f"For general Khalifa University questions, please visit the main KU website: {KU_MAIN_URL}. " "A broader university chatbot is available there for non-library questions." ) @app.post("/correct") async def correct_query(req: CorrectRequest): """ Spell-correct a search query and build both natural and Boolean forms for traditional databases such as PRIMO and PubMed. Pre-cleans the query to strip chat prefixes and connectors before processing. """ # Strip chat prefixes, connectors, and follow-up fragments # so "and for physics" → "physics" before LLM processing raw = req.query.strip() cleaned = re.sub( r'^(find articles and books on|find articles on|find books on|' r'search for|look for|i want|i need|give me|show me|get me|' r'and\s+(also\s+)?(for|about|on|in|related to|regarding)|' r'also\s+(for|about|on|in)|what about|how about|or\s+about|' r'tell me about|more\s+(about|on))\s+', '', raw, flags=re.IGNORECASE ).strip() # Use cleaned query if it's non-empty, otherwise fall back to raw query_to_use = cleaned if cleaned else raw plan = await _build_search_plan(query_to_use, req.year) return { "corrected": plan["corrected"], "boolean": plan["boolean"], "natural": plan["natural"], "database_query": plan["database_query"], "year_from": plan["year_from"], "year_to": plan["year_to"], "peer_reviewed": plan["peer_reviewed"], "open_access": plan["open_access"], } # ---- General knowledge endpoint (uses HF Space OpenAI key — separate quota from Cloudflare) ---- class GeneralRequest(BaseModel): question: str history: list = [] model: str = "gpt" async def _answer_general(question: str, history=None) -> dict: """Answer general knowledge questions using live web search when available.""" if _looks_library_hours_question(question): return {"answer": _hours_redirect_answer(), "sources": [], "model": "library-hours-redirect"} if _looks_nonlibrary_ku_question(question): return {"answer": _ku_general_redirect_answer(), "sources": [], "model": "ku-general-redirect"} if _looks_social_or_greeting(question): if PURE_GREETING_RE.match(question.strip()): guided = ( "Hi! I’m LibBee, the Khalifa University Library AI Assistant.

" "I can help you with articles and books, databases, full text, Interlibrary Loan (ILL), " "library services, staff contacts, ORCID, RefWorks, and Open Access.

" "Are you looking for one of these? You can also type your question directly." ) return {"answer": guided, "sources": [], "model": "libbee-greeting"} try: behavior = get_behavior_instructions() llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.4, max_tokens=220) msgs = [] if behavior: msgs.append({"role": "system", "content": behavior + "\nFor social, greeting, or self-description questions, answer as LibBee in a warm, brief, natural way. Make it clear that you are LibBee, the Khalifa University Library AI Assistant, and that you help with books, articles, databases, library services, research support, and staff contacts. For questions like 'how are you', 'who are you', 'what are you', 'what can you do', 'are you a bot', playful teasing, or mild mockery, respond politely and briefly, without sounding defensive. If the question is not library-related, you may gently mention that you are specialized in library help. Match the language of the user question."}) for m in (history or [])[-4:]: if m.get("role") in ("user", "assistant"): msgs.append({"role": m["role"], "content": m.get("content", "")}) msgs.append({"role": "user", "content": question}) response = llm.invoke(msgs) return {"answer": response.content.strip(), "sources": [], "model": "gpt-4o-mini-social"} except Exception: return {"answer": "I’m LibBee, the Khalifa University Library chatbot. I’m here to help with library questions, books, articles, databases, full text, and library services. What would you like help with?", "sources": [], "model": "fallback-social"} try: import openai client = openai.OpenAI(api_key=os.environ.get("OPENAI_API_KEY")) messages = [] if history: for m in history[-5:]: role = m.get("role", "user") msg_content = m.get("content", "") if role in ("user", "assistant") and msg_content: messages.append({"role": role, "content": msg_content}) messages.append({"role": "user", "content": question}) response = client.responses.create( model="gpt-4o-mini", tools=[{"type": "web_search_preview"}], input=messages, ) answer = "" sources = [] for block in response.output: if hasattr(block, "content"): for item in block.content: if hasattr(item, "text"): answer += item.text if hasattr(item, "annotations"): for ann in item.annotations: if hasattr(ann, "url") and hasattr(ann, "title"): sources.append({"url": ann.url, "title": ann.title}) if not answer: answer = response.output_text if hasattr(response, "output_text") else "" return {"answer": answer.strip(), "sources": sources, "model": "gpt-4o-mini-web"} except Exception as e: print(f"Web search failed, falling back to plain GPT: {e}") try: llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3, max_tokens=500) msgs = [] for m in (history or [])[-5:]: if m.get("role") in ("user", "assistant"): msgs.append({"role": m["role"], "content": m.get("content", "")}) msgs.append({"role": "user", "content": question}) response = llm.invoke(msgs) return {"answer": response.content.strip(), "sources": [], "model": "gpt-4o-mini"} except Exception as e2: return {"answer": "", "sources": [], "error": str(e2)} @app.post("/general") async def general_query(req: GeneralRequest): return await _answer_general(req.question, req.history) # ---- Individual search endpoints ---- @app.post("/search") async def search(req: SearchRequest): start = time.time() source = req.source.lower() result = {} try: if source == "primo": result = await tool_search_primo(req.query, req.limit, req.peer_reviewed, req.open_access, req.year_from, req.year_to) elif source == "pubmed": result = await tool_search_pubmed(req.query, req.limit) elif source == "scholar": result = await tool_search_scholar(req.query, req.limit) elif source == "consensus": result = await tool_search_consensus(req.query, req.limit) elif source == "all": import asyncio tasks = [ tool_search_primo(req.query, req.limit, req.peer_reviewed, req.open_access, req.year_from, req.year_to), tool_search_pubmed(req.query, min(req.limit, 3)), tool_search_scholar(req.query, min(req.limit, 3)), ] results = await asyncio.gather(*tasks, return_exceptions=True) combined = [] for r in results: if isinstance(r, dict) and "results" in r: for item in r["results"]: item["_source"] = r.get("source", "unknown") combined.extend(r["results"]) result = {"total": len(combined), "results": combined, "source": "Multiple"} else: result = {"error": f"Unknown source: {source}", "results": []} elapsed = time.time() - start # Logging handled by Cloudflare D1 via worker /log endpoint (single source of truth) result["response_time"] = round(elapsed, 2) result["is_medical"] = is_medical_topic(req.query) return result except Exception as e: elapsed = time.time() - start return {"error": str(e), "results": [], "response_time": round(elapsed, 2)} # ---- Public log endpoint (frontend calls this for all queries) ---- class LogRequest(BaseModel): question: str tool: str = "unknown" model: str = "gpt" response_time: float = 0 result_count: int = 0 error: str | None = None @app.post("/log") async def log_endpoint(req: LogRequest): log_query(req.question, req.tool, req.model, req.response_time, req.result_count, req.error) return {"status": "ok"} # ---- RAG endpoint (backward compatible) ---- @app.post("/rag") async def rag_query(req: RAGRequest): start = time.time() try: history = [{"role": m.role, "content": m.content} for m in req.history] if req.history else None result = await tool_library_info(req.question, history, model=req.model) elapsed = time.time() - start # Logging handled by Cloudflare D1 via worker /log endpoint return { "answer": result["answer"], "sources": result["sources"], "has_answer": result.get("has_answer", True), "model_used": req.model, "response_time": round(elapsed, 2), } except Exception as e: elapsed = time.time() - start return {"answer": "", "sources": [], "has_answer": False, "error": str(e)} # ===== LLM FIRST-GATE CLASSIFIER ===== # # Architecture (executed in this order for every /agent request): # # Layer 0 — INSTANT free responses (no LLM, <1ms) # • Pure greetings → hardcoded LibBee greeting # • Exact social phrases → _fixed_social_answer() # • Staff name match → staff directory # • Staff role match → staff directory # • Greeting followup → clarify menu # • Library hours → hours redirect # • Non-library KU Q → KU website redirect # # Layer 1 — LLM CLASSIFIER (GPT-4o-mini, ~300ms, ~$0.000045/call) # ALL other messages hit this exactly once. # Returns: social | library_info | search_academic | # search_medical | general_recent | general # + optional casual_answer string for social intent # # Layer 2 — EXECUTE # social → _libbee_casual_response() # library_info → RAG pipeline (_interpret_semantics + tool_library_info) # search_* → PRIMO + PubMed/Scholar + RAG synthesis # general_recent → _answer_general (web search) # general → _answer_general _CLASSIFIER_SYSTEM = """You are the routing brain of LibBee, the Khalifa University Library AI Assistant (Abu Dhabi, UAE). Your only job: read the user message and return a single JSON object routing it to the right handler. INTENTS — pick exactly one: "social" — casual chat, greetings, emotions, compliments, complaints, vague help requests ("can you help me", "I'm frustrated", "you're great", "I give up", "help please", "I don't know what to do", "I am fed up", "I need some guidance", "are you there", jokes, small talk, anything with no library/research/academic content) "library_info" — questions about KU Library services, hours, rooms, borrowing, fines, ILL, staff, databases, ORCID, RefWorks, open access, APC, account, printing, remote access, LibKey, acquisitions, Khazna, thesis submission, library policies, alumni access "search_academic"— user wants to FIND articles / papers / books / literature on a specific academic topic (e.g. "find articles on renewable energy", "papers on machine learning in healthcare", "I need books about civil engineering", "literature review on AI ethics") "search_medical" — same as search_academic but topic is medical/clinical/nursing/pharmacy/health sciences "general_recent" — questions about current events, news, latest info, who holds a position now, real-time data (stock prices, today's weather, breaking news) "general" — factual/knowledge question NOT about KU Library and NOT a research search (explain a concept, historical facts, definitions, "what is quantum computing") RULES: 1. When in doubt between library_info and social, choose library_info if ANY library keyword is present. 2. "I don't know how to search for articles" = library_info (has library task content). 3. "I am fed up with searching" alone (no specific topic/database) = social (expressing frustration). 4. "I need help finding articles on diabetes" = search_medical. 5. KU = Khalifa University. Do NOT confuse with Kuwait University. For "social" intent ONLY: also include "casual_answer" — a warm 1-3 sentence LibBee response (friendly librarian tone, no markdown, no bullet points). Offer to help with library services. For all other intents: omit "casual_answer" entirely. Respond ONLY with valid JSON, no preamble, no markdown fences. Examples: {"intent":"social","casual_answer":"Of course, I\'m here to help! I\'m LibBee, the KU Library AI Assistant. What would you like to find today?"} {"intent":"social","casual_answer":"I completely understand that feeling — searching can be tricky at first. I\'m LibBee, the Khalifa University Library AI Assistant, and I can guide you through it step by step. What topic are you trying to research?"} {"intent":"library_info"} {"intent":"search_academic"} {"intent":"search_medical"} {"intent":"general_recent"} {"intent":"general"}""" async def _llm_classify(question: str, history: list) -> dict: """ Single LLM call that classifies intent for ALL non-trivial messages. ~300ms latency, ~$0.000045 per call at GPT-4o-mini pricing. Returns: {"intent": str, "casual_answer": str} Never raises — falls back to {"intent":"general","casual_answer":""} on error. """ # Build compact conversation context (last 2 turns, 100 chars each) ctx_parts = [] for m in history[-4:]: role = m.get("role", "") text = (m.get("content") or "")[:100] if role in ("user", "assistant") and text: ctx_parts.append(f"{role}: {text}") ctx = " | ".join(ctx_parts) if ctx_parts else "none" user_msg = ( "Message: " + question[:400] + "\n" + "Context: " + ctx + "\nJSON:" ) try: llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, max_tokens=150) response = llm.invoke([ {"role": "system", "content": _CLASSIFIER_SYSTEM}, {"role": "user", "content": user_msg}, ]) raw = response.content.strip() if raw.startswith("```"): raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip() result = json.loads(raw) intent = result.get("intent", "general") valid_intents = {"social","library_info","search_academic","search_medical","general_recent","general"} if intent not in valid_intents: intent = "general" return {"intent": intent, "casual_answer": result.get("casual_answer", "")} except Exception as e: print(f"[LibBee classifier error] {e} | question: {question[:80]}") return {"intent": "general", "casual_answer": ""} async def _libbee_casual_response(question: str, history: list, model: str, hint: str = "") -> str: """ Generate a warm, in-character LibBee response for social/casual messages. Uses the classifier hint when available (saves a second LLM call). Falls back to a hardcoded response on any error. """ # Use the classifier's casual_answer hint if it's good enough if hint and len(hint.strip()) > 25: return hint.strip() behavior = get_behavior_instructions() sys_prompt = ( f"{behavior}\n\n" "You are LibBee, the Khalifa University Library AI Assistant. " "The user has sent a casual, social, or emotional message. " "Respond warmly and naturally in 1-3 sentences as a friendly, knowledgeable librarian. " "Always identify yourself as LibBee and offer library help. " "If the user seems frustrated or confused, be empathetic and reassuring. " "If the user seems happy or is complimenting you, be cheerful and grateful. " "Never say \'I don\'t have information\'. Never use bullet points or markdown. " "Match the user\'s energy — if casual, be casual; if distressed, be warm and calm." ) msgs = [{"role": "system", "content": sys_prompt}] for m in history[-4:]: if m.get("role") in ("user", "assistant"): msgs.append({"role": m["role"], "content": (m.get("content") or "")[:200]}) msgs.append({"role": "user", "content": question}) use_claude = model == "claude" and os.environ.get("ANTHROPIC_API_KEY") try: if use_claude: from langchain_anthropic import ChatAnthropic llm = ChatAnthropic(model="claude-haiku-4-5-20251001", temperature=0.6, max_tokens=200) else: llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.6, max_tokens=200) return llm.invoke(msgs).content.strip() except Exception: return ( "I'm LibBee, the Khalifa University Library AI Assistant — " "always happy to help! What can I find for you today?" ) def _make_agent_response(answer, intent, tools_used, search_results, sources, model, elapsed, question, natural_query=None, database_query=None, original_question=None, is_follow_up=False, source_mode="", ask_librarian=None): """ Helper to build the standard /agent response dict. ask_librarian controls whether the frontend appends the "Ask a Librarian" footer block: None → frontend decides based on intent + tools_used (default behaviour) True → always show footer False → never show footer (e.g. hours/campus answers are self-contained) """ return { "answer": answer, "intent": intent, "tools_used": tools_used or [], "search_results": search_results or [], "sources": sources or [], "model_used": model, "response_time": round(elapsed, 2), "corrected_query": question, "natural_query": natural_query or question, "database_query": database_query or question, "original_question": original_question or question, "is_follow_up": is_follow_up, "source_mode": source_mode, "ask_librarian": ask_librarian, # None | True | False } # ---- Agent endpoint (Batch 3: tool-calling agent) ---- @app.post("/agent") async def agent_query(req: AgentRequest): """ LLM-first multi-tool agent. Execution order: Layer 0: Instant free handlers (pure greetings, exact social, staff, hours, KU redirect) Layer 1: LLM classifier → one GPT-4o-mini call for ALL other messages Layer 2: Execute the classified intent (social/library/search/general) """ start = time.time() question = req.question.strip() history = [{"role": m.role, "content": m.content} for m in req.history] if req.history else [] use_claude = req.model == "claude" and bool(os.environ.get("ANTHROPIC_API_KEY")) q_clean = question.strip() # ══════════════════════════════════════════════════════════════════ # LAYER 0 — INSTANT FREE HANDLERS (no LLM call, < 1ms each) # Only for responses that are 100% deterministic and zero-cost. # ══════════════════════════════════════════════════════════════════ # 0a. Pure greetings (hi / hello / hey / good morning …) if PURE_GREETING_RE.match(q_clean): elapsed = time.time() - start return _make_agent_response( answer=( "Hi! I'm LibBee, the Khalifa University Library AI Assistant.

" "I can help you find articles and books, search databases, access full text, " "request Interlibrary Loan (ILL), and answer questions about library services and staff.

" "Are you looking for one of these? You can also type your question directly." ), intent="social_greeting", tools_used=[], search_results=[], sources=[], model=req.model, elapsed=elapsed, question=question, is_follow_up=False, source_mode="social", ) # 0b. Exact social phrases (who are you, how are you, thanks, bye …) fixed = _fixed_social_answer(q_clean) if fixed: elapsed = time.time() - start return _make_agent_response( answer=fixed, intent="social_greeting", tools_used=[], search_results=[], sources=[], model=req.model, elapsed=elapsed, question=question, is_follow_up=False, source_mode="social", ) # 0c. Staff name direct lookup (e.g. "nikesh", "jason fetty") staff_match = _match_staff_name(question) if staff_match: elapsed = time.time() - start src_title = staff_match.get("source_title", "") src_url = staff_match.get("source", "") return _make_agent_response( answer=_staff_name_answer(staff_match, question), intent="library_info", tools_used=["staff_name_match"], search_results=[], sources=([{"title": src_title, "source": src_url}] if src_title or src_url else []), model=req.model, elapsed=elapsed, question=question, source_mode="staff_kb" if kb_staff_directory else "staff_directory", ) # 0d. Staff role lookup (e.g. "who is the medical librarian") staff_role_match = _match_staff_role(question) if staff_role_match: elapsed = time.time() - start src_title = staff_role_match.get("source_title", "") src_url = staff_role_match.get("source", "") return _make_agent_response( answer=_staff_role_answer(staff_role_match, question), intent="library_info", tools_used=["staff_role_match"], search_results=[], sources=([{"title": src_title, "source": src_url}] if src_title or src_url else []), model=req.model, elapsed=elapsed, question=question, source_mode="staff_kb" if kb_staff_directory else "staff_directory", ) # 0e. Greeting menu follow-up (yes / sure / okay after LibBee intro) if _is_greeting_menu_followup(question, history): elapsed = time.time() - start return _make_agent_response( answer=_greeting_menu_clarify_answer(), intent="social_greeting", tools_used=[], search_results=[], sources=[], model=req.model, elapsed=elapsed, question=question, is_follow_up=True, source_mode="social", ) # 0f. Library hours question (fast-path — hardcoded hours + link, never web) if _looks_library_hours_question(question): elapsed = time.time() - start return _make_agent_response( answer=_hours_redirect_answer(), intent="library_info", tools_used=["hours_hardcoded"], search_results=[], sources=[{"title": "Library Hours", "source": LIBRARY_HOURS_URL}], model=req.model, elapsed=elapsed, question=question, ask_librarian=False, # hours answer is self-contained ) # 0f2. Campus / library location question (hardcoded — never from web) if _looks_campus_question(question): elapsed = time.time() - start return _make_agent_response( answer=_ku_campus_answer(), intent="library_info", tools_used=["campus_hardcoded"], search_results=[], sources=[], model=req.model, elapsed=elapsed, question=question, ask_librarian=False, # answer already includes contact info ) # 0g. Non-library KU question (admissions, programs, fees …) if _looks_nonlibrary_ku_question(question): elapsed = time.time() - start return _make_agent_response( answer=_ku_general_redirect_answer(), intent="general", tools_used=["ku_general_redirect"], search_results=[], sources=[], model=req.model, elapsed=elapsed, question=question, ) # ══════════════════════════════════════════════════════════════════ # LAYER 1 — LLM CLASSIFIER # One GPT-4o-mini call classifies ALL messages that reached here. # Cost: ~$0.000045 per call. At 250 questions/day ≈ $1/month. # ══════════════════════════════════════════════════════════════════ cls_result = await _llm_classify(question, history) intent = cls_result.get("intent", "general") casual_hint = cls_result.get("casual_answer", "") # ══════════════════════════════════════════════════════════════════ # LAYER 2 — EXECUTE CLASSIFIED INTENT # ══════════════════════════════════════════════════════════════════ # ── Social / casual ────────────────────────────────────────────── if intent == "social": casual = await _libbee_casual_response(question, history, req.model, casual_hint) elapsed = time.time() - start return _make_agent_response( answer=casual, intent="social_greeting", tools_used=["libbee_casual"], search_results=[], sources=[], model=req.model, elapsed=elapsed, question=question, is_follow_up=False, source_mode="social", ) # ── Library info — RAG pipeline ─────────────────────────────────── if intent == "library_info": # Use _interpret_semantics to enrich the query with canonical library terms enriched_question = question try: interp = await _interpret_semantics(question, history) if interp.get("grounding_keys"): resolved = " ".join( [question] + interp.get("canonical_terms", []) + [GROUNDED_LIBRARY_MAP.get(k, "") for k in interp.get("grounding_keys", []) if GROUNDED_LIBRARY_MAP.get(k)] ) enriched_question = re.sub(r"\s+", " ", resolved).strip() except Exception: pass rag = await tool_library_info(enriched_question, history[-5:] if history else None, model=req.model) tools_used = ["get_library_info"] if rag.get("has_answer") and rag.get("answer"): # Good RAG answer — synthesise and return context_parts = [f"Library Knowledge Base:\n{rag['answer']}"] behavior = get_behavior_instructions() synthesis_prompt = ( f"{behavior}\n\n" "You are the Khalifa University Library AI Assistant (Abu Dhabi, UAE). KU = Khalifa University.\n" "Answer in 2-4 sentences. Include URLs from the context when relevant.\n" "Answer ONLY from the provided context. Do not add information not present in the context.\n\n" f"Context:\n{chr(10).join(context_parts)}\n\n" f"Question: {question}\nAnswer:" ) try: if use_claude: from langchain_anthropic import ChatAnthropic synth_llm = ChatAnthropic(model="claude-haiku-4-5-20251001", temperature=0.2, max_tokens=500) else: synth_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2, max_tokens=500) answer = synth_llm.invoke(synthesis_prompt).content.strip() except Exception: answer = rag["answer"] elapsed = time.time() - start return _make_agent_response( answer=answer, intent="library_info", tools_used=tools_used, search_results=[], sources=rag.get("sources", []), model=req.model, elapsed=elapsed, question=question, ask_librarian=False, # RAG answer from trusted KB — no footer needed ) else: # ── RAG miss — three-tier fallback for library questions ── # # Tier 1: Strict KU-library-only LLM prompt (no hallucination risk — # LLM is told to say it doesn't know rather than invent). # Tier 2: If tier 1 is vague/empty → web search scoped to library.ku.ac.ae # so all facts come from the real KU library website. # Tier 3: Always append "Ask a Librarian" + library homepage link. # # General web search (_answer_general) is NOT used here — that's only # for intent==general/general_recent where unrestricted search is correct. # ── Tier 1: Strict KU-only LLM ── ku_only_prompt = ( "You are LibBee, the Khalifa University Library AI Assistant (Abu Dhabi, UAE). " "KU = Khalifa University.\n\n" "STRICT RULES:\n" "1. Answer ONLY using your knowledge of Khalifa University Library services, " "databases, policies, staff, and resources.\n" "2. Be concise (2-4 sentences). Include real KU library URLs when relevant " "(e.g. https://library.ku.ac.ae/AskUs, https://library.ku.ac.ae/ill/, " "https://library.ku.ac.ae/eresources, https://library.ku.ac.ae/oa/).\n" "3. If you are not confident about a specific detail (phone number, exact policy, " "specific database URL), do NOT invent it — instead say you are not certain and " "direct the user to Ask a Librarian or the library website.\n" "4. Never make up staff names, contact details, or database names.\n" "5. Do NOT use general web knowledge or facts unrelated to KU Library.\n\n" f"Question: {question}\nAnswer:" ) tier1_answer = "" try: if use_claude: from langchain_anthropic import ChatAnthropic t1_llm = ChatAnthropic(model="claude-haiku-4-5-20251001", temperature=0.1, max_tokens=400) else: t1_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1, max_tokens=400) t1_response = t1_llm.invoke([ {"role": "system", "content": ku_only_prompt}, {"role": "user", "content": question}, ]) tier1_answer = t1_response.content.strip() except Exception as t1_err: print(f"[LibBee tier1 LLM error] {t1_err}") # ── Tier 2: If tier 1 is vague, search library.ku.ac.ae ── # Detect vague answers: short, uncertain, or explicitly admitting no knowledge vague_phrases = [ "i'm not sure", "i am not sure", "i don't have", "i do not have", "not certain", "cannot confirm", "not available", "unable to find", "please contact", "you may want to", "i would recommend checking", "i don't have specific", "i lack specific", ] is_vague = ( not tier1_answer or len(tier1_answer.split()) < 15 or any(p in tier1_answer.lower().replace("\u2019", "'") for p in vague_phrases) ) tier2_answer = "" tier2_sources = [] if is_vague: try: import openai as _openai _client = _openai.OpenAI(api_key=os.environ.get("OPENAI_API_KEY")) # Scope web search to KU library domain only scoped_query = f"site:library.ku.ac.ae {question}" _resp = _client.responses.create( model="gpt-4o-mini", tools=[{"type": "web_search_preview"}], input=[{ "role": "system", "content": ( "You are LibBee, the Khalifa University Library AI Assistant. " "Search the KU library website (library.ku.ac.ae) only. " "Answer in 2-4 sentences using only what you find on the KU library website. " "Always include the source URL. " "If nothing relevant is found on library.ku.ac.ae, say so briefly." ) }, {"role": "user", "content": scoped_query}], ) for block in _resp.output: if hasattr(block, "content"): for item in block.content: if hasattr(item, "text"): tier2_answer += item.text if hasattr(item, "annotations"): for ann in item.annotations: if hasattr(ann, "url") and hasattr(ann, "title"): if "library.ku.ac.ae" in getattr(ann, "url", ""): tier2_sources.append({"url": ann.url, "title": ann.title}) if not tier2_answer: tier2_answer = getattr(_resp, "output_text", "") or "" tier2_answer = tier2_answer.strip() except Exception as t2_err: print(f"[LibBee tier2 web search error] {t2_err}") # ── Combine best answer — ask_librarian=True for all miss tiers ── # The frontend will render the Ask a Librarian block. # Answer strings are clean — no embedded HTML footer. if tier2_answer and len(tier2_answer.split()) >= 10: # Web search on library.ku.ac.ae found something useful final_answer = tier2_answer final_sources = tier2_sources final_tools = tools_used + ["ku_web_search"] elif tier1_answer and len(tier1_answer.split()) >= 10: # Tier 1 KU-only LLM gave a decent answer final_answer = tier1_answer final_sources = [] final_tools = tools_used + ["ku_only_llm"] else: # Both failed — clean redirect only final_answer = ( "I don't have that specific information in my knowledge base right now. " "The KU Library team will be able to help directly." ) final_sources = [] final_tools = tools_used + ["redirected"] elapsed = time.time() - start return _make_agent_response( answer=final_answer, intent="library_info", tools_used=final_tools, search_results=[], sources=final_sources, model=req.model, elapsed=elapsed, question=question, ask_librarian=True, # Tier 1/2/3 — always show footer ) # ── Academic / medical search ───────────────────────────────────── if intent in ("search_academic", "search_medical"): import asyncio as _asyncio search_plan = await _build_search_plan(question) natural_query = search_plan["natural"] database_query = search_plan["database_query"] or search_plan["corrected"] tasks = [tool_search_primo(database_query, limit=5)] if intent == "search_medical": tasks.append(tool_search_pubmed(database_query, limit=3)) else: tasks.append(tool_search_scholar(natural_query, limit=3)) raw_results = await _asyncio.gather(*tasks, return_exceptions=True) combined = [] tools_used = [] for r in raw_results: if isinstance(r, dict) and r.get("results"): combined.extend(r["results"]) tools_used.append(r.get("source", "unknown")) rag = await tool_library_info(question, history[-3:] if history else None, model=req.model) tools_used.append("get_library_info") tools_used = list(dict.fromkeys(tools_used)) context_parts = [] if rag.get("answer"): context_parts.append(f"Library Knowledge Base:\n{rag['answer']}") if combined: top = combined[:3] res_text = "\n".join( f"- {r.get('title','')} by {r.get('creator','')} ({r.get('date','')})" for r in top ) context_parts.append(f"Search Results:\n{res_text}") context_parts.append(f"Natural query for AI tools: {natural_query}") context_parts.append(f"Database query for PRIMO/PubMed: {database_query}") behavior = get_behavior_instructions() synthesis_prompt = ( f"{behavior}\n\n" "You are the KU Library AI Assistant. Be concise (3-5 sentences).\n" "Briefly describe the search direction and mention 1-2 top results if present.\n\n" f"Context:\n{chr(10).join(context_parts) if context_parts else 'No additional context.'}\n\n" f"Question: {question}\nAnswer:" ) try: if use_claude: from langchain_anthropic import ChatAnthropic synth_llm = ChatAnthropic(model="claude-haiku-4-5-20251001", temperature=0.2, max_tokens=600) else: synth_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2, max_tokens=600) answer = synth_llm.invoke(synthesis_prompt).content.strip() except Exception as ex: answer = rag.get("answer", f"Error generating answer: {ex}") elapsed = time.time() - start return _make_agent_response( answer=answer, intent=intent, tools_used=tools_used, search_results=combined[:8], sources=rag.get("sources", []), model=req.model, elapsed=elapsed, question=question, natural_query=natural_query, database_query=database_query, ) # ── General / general_recent — web search or plain LLM ─────────── general = await _answer_general(question, history) gen_answer = general.get("answer", "").strip() # Guard: if general returned a "no info" string, replace with a helpful redirect no_info_phrases = [ "i don't have", "i do not have", "no specific information", "not available in my", "unable to find", "cannot find", ] gen_lower = gen_answer.lower().replace("\u2019", "'") if not gen_answer or any(p in gen_lower for p in no_info_phrases): gen_answer = ( "I'm not sure about that one. For library-specific questions, try " "Ask a Librarian. " "For general knowledge questions, you could also try a web search." ) elapsed = time.time() - start return _make_agent_response( answer=gen_answer, intent=intent, tools_used=["general_web" if intent == "general_recent" else "general"], search_results=[], sources=general.get("sources", []), model=req.model, elapsed=elapsed, question=question, ) # ---- Rebuild index (protected) ---- @app.post("/rebuild") async def rebuild_index(request: Request): if not verify_session(request): raise HTTPException(status_code=401, detail="Not authenticated") global vectorstore try: import shutil, glob as _glob # Delete old FAISS index files so we rebuild from scratch if os.path.exists(FAISS_INDEX_PATH): shutil.rmtree(FAISS_INDEX_PATH) print(f"Deleted old FAISS index at {FAISS_INDEX_PATH}") # List all knowledge files found kb_files = _glob.glob(os.path.join(KNOWLEDGE_DIR, "*.txt")) print(f"Knowledge files found: {len(kb_files)}") for f in kb_files: print(f" - {os.path.basename(f)} ({os.path.getsize(f)} bytes)") docs = load_documents() if not docs: return {"error": "No knowledge files found", "kb_dir": KNOWLEDGE_DIR, "files": []} # Force rebuild — ignore any cached index vectorstore = build_vectorstore(docs, force_rebuild=True) chunks = vectorstore.index.ntotal return { "status": "ok", "chunks": chunks, "documents": len(docs), "files": [os.path.basename(f) for f in kb_files], "kb_dir": KNOWLEDGE_DIR } except Exception as e: import traceback return {"error": str(e), "traceback": traceback.format_exc()} # ===== ADMIN LOGIN ===== @app.get("/admin/login", response_class=HTMLResponse) async def admin_login_page(): return """ KU Library AI — Admin Login

🔐 Admin Login

KU Library AI Dashboard
Incorrect password. Try again.
""" @app.post("/admin/auth") async def admin_auth(request: Request, response: Response): data = await request.json() if data.get("password") == ADMIN_PASSWORD: token = create_session() resp = Response(content=json.dumps({"status": "ok"}), media_type="application/json") resp.set_cookie("admin_session", token, httponly=True, max_age=86400, samesite="lax") return resp raise HTTPException(status_code=401, detail="Invalid password") @app.get("/admin/logout") async def admin_logout(request: Request): token = request.cookies.get("admin_session") if token and token in admin_sessions: del admin_sessions[token] resp = RedirectResponse(url="/admin/login") resp.delete_cookie("admin_session") return resp # ===== ADMIN DASHBOARD (protected) ===== WORKER_URL = "https://steep-haze-e1f8.nikeshn.workers.dev" @app.get("/admin", response_class=HTMLResponse) async def admin_dashboard(request: Request): if not verify_session(request): return RedirectResponse(url="/admin/login") config = {} conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row rows = conn.execute("SELECT key, value FROM bot_config").fetchall() conn.close() for r in rows: config[r["key"]] = r["value"] def esc(s): return (s or "").replace("&","&").replace("<","<").replace(">",">").replace('"',""") return f""" KU Library AI — Admin

📊 KU Library AI — Admin 🔒 Logout

Khalifa University Library AI Agent
📈 Analytics
🤖 Bot Behavior
📋 Query Log
⚙️ System
Loading analytics from Cloudflare D1…

🤖 Bot Behavior

Changes apply immediately.

First message users see.
Core instructions for every LLM call.
Extra guidance, e.g. "During Ramadan, mention reduced hours."
Shows at top of chat. Leave empty to hide.
Loading query log…

⚙️ Controls

🟢 Live Status

Loading status…

System Info

Knowledge Files{len(glob.glob(os.path.join(KNOWLEDGE_DIR,'*.txt')))} files
OpenAI API{'✅ Configured' if os.environ.get('OPENAI_API_KEY') else '❌ Missing'}
Anthropic API{'✅ Configured' if os.environ.get('ANTHROPIC_API_KEY') else '❌ Missing'}
PRIMO API{'✅ Configured' if os.environ.get('PRIMO_API_KEY') else '❌ Missing'}
Admin Password{'✅ Set' if os.environ.get('ADMIN_PASSWORD') else '❌ Missing'}
Cloudflare D1Connected via Worker
""" # ---- Restart Space endpoint ---- @app.post("/restart") async def restart_space(request: Request): if not verify_session(request): raise HTTPException(status_code=401, detail="Not authenticated") import threading def do_restart(): import time as t t.sleep(1) os.kill(os.getpid(), 9) # Force restart — HF Space will auto-restart threading.Thread(target=do_restart, daemon=True).start() return {"status": "restarting", "message": "Space is restarting. Please wait 30-60 seconds."} # ---- Status endpoint (no auth needed — for frontend health checks) ---- @app.get("/status") async def get_status(): # Read container-level memory from /sys/fs/cgroup (accurate for HF Spaces) # psutil reads the host machine memory which is misleading on shared containers try: # cgroup v2 (HF Spaces uses this) with open('/sys/fs/cgroup/memory.current') as f: mem_used_bytes = int(f.read().strip()) with open('/sys/fs/cgroup/memory.max') as f: val = f.read().strip() mem_total_bytes = int(val) if val != 'max' else 16 * 1024**3 # 16GB default mem_used_mb = round(mem_used_bytes / 1024 / 1024) mem_total_mb = round(mem_total_bytes / 1024 / 1024) mem_pct = round(mem_used_bytes / mem_total_bytes * 100) except: # Fallback: try cgroup v1 try: with open('/sys/fs/cgroup/memory/memory.usage_in_bytes') as f: mem_used_bytes = int(f.read().strip()) with open('/sys/fs/cgroup/memory/memory.limit_in_bytes') as f: mem_total_bytes = int(f.read().strip()) mem_used_mb = round(mem_used_bytes / 1024 / 1024) mem_total_mb = round(mem_total_bytes / 1024 / 1024) mem_pct = round(mem_used_bytes / mem_total_bytes * 100) except: mem_used_mb = mem_total_mb = mem_pct = 0 kb_files = glob.glob(os.path.join(KNOWLEDGE_DIR, "*.txt")) chunks = vectorstore.index.ntotal if vectorstore else 0 return { "status": "ok", "vectorstore": "ready" if vectorstore else "not_initialized", "chunks": chunks, "kb_files": len(kb_files), "openai": bool(os.environ.get("OPENAI_API_KEY")), "anthropic": bool(os.environ.get("ANTHROPIC_API_KEY")), "primo": bool(os.environ.get("PRIMO_API_KEY")), "mem_used_mb": mem_used_mb, "mem_total_mb": mem_total_mb, "mem_pct": mem_pct, "uptime_secs": round(time.time() - START_TIME), } # ---- Admin stats from D1 (kept for API compatibility) ---- @app.get("/admin/stats") async def admin_stats(request: Request): if not verify_session(request): raise HTTPException(status_code=401, detail="Not authenticated") # Stats now come from Cloudflare D1 via /analytics endpoint return {"message": "Use Cloudflare Worker /analytics endpoint for D1 data"} conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row total = conn.execute("SELECT COUNT(*) as c FROM queries").fetchone()["c"] today = conn.execute("SELECT COUNT(*) as c FROM queries WHERE timestamp >= date('now')").fetchone()["c"] tools = {r["tool_used"]: r["c"] for r in conn.execute("SELECT tool_used, COUNT(*) as c FROM queries GROUP BY tool_used").fetchall()} models = {r["model"]: r["c"] for r in conn.execute("SELECT model, COUNT(*) as c FROM queries GROUP BY model").fetchall()} conn.close() return {"total": total, "today": today, "tools": tools, "models": models} # ---- Admin Config GET/POST (protected) ---- @app.get("/admin/config") async def admin_get_config(request: Request): if not verify_session(request): raise HTTPException(status_code=401, detail="Not authenticated") conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row rows = conn.execute("SELECT key, value, updated_at FROM bot_config").fetchall() conn.close() return {r["key"]: {"value": r["value"], "updated_at": r["updated_at"]} for r in rows} @app.post("/admin/config") async def admin_set_config(request: Request): if not verify_session(request): raise HTTPException(status_code=401, detail="Not authenticated") data = await request.json() settings = data.get("settings", data) if isinstance(data, dict) else data for key, value in settings.items(): set_config(key, str(value)) return {"status": "ok", "updated": list(settings.keys())} class FeedbackRequest(BaseModel): question: str = "" feedback: str answer_excerpt: str = "" intent: str = "" @app.post("/feedback") async def feedback_submit(req: FeedbackRequest): try: conn = sqlite3.connect(DB_PATH) conn.execute( "INSERT INTO feedback (timestamp, question, feedback, answer_excerpt, intent) VALUES (?,?,?,?,?)", (datetime.utcnow().isoformat(), req.question or "", req.feedback or "", (req.answer_excerpt or "")[:500], req.intent or "") ) conn.commit() conn.close() return {"status": "ok"} except Exception as e: return {"status": "error", "error": str(e)} @app.get("/admin/feedback-stats") async def admin_feedback_stats(request: Request): if not verify_session(request): raise HTTPException(status_code=401, detail="Not authenticated") conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row summary = [dict(r) for r in conn.execute("SELECT feedback, COUNT(*) as c FROM feedback GROUP BY feedback ORDER BY c DESC").fetchall()] recent = [dict(r) for r in conn.execute("SELECT timestamp, question, feedback, intent FROM feedback ORDER BY id DESC LIMIT 30").fetchall()] weak = [dict(r) for r in conn.execute("SELECT question, COUNT(*) as c FROM feedback WHERE feedback='down' AND question != '' GROUP BY question ORDER BY c DESC LIMIT 10").fetchall()] conn.close() return {"summary": summary, "recent": recent, "weak": weak} # ---- Public config (for frontend) ---- @app.get("/config") async def public_config(): return { "welcome_message": get_config("welcome_message"), "bot_personality": get_config("bot_personality"), "custom_instructions": get_config("custom_instructions"), "announcement": get_config("announcement"), "default_model": get_config("default_model", "gpt"), "maintenance_mode": get_config("maintenance_mode", "false"), "maintenance_message": get_config("maintenance_message"), "max_results": get_config("max_results", "5"), } # ---- Clear logs ---- @app.post("/admin/clear-logs") async def clear_logs(): conn = sqlite3.connect(DB_PATH) conn.execute("DELETE FROM queries") conn.commit() conn.close() return {"status": "ok", "message": "All logs cleared"}