| """ |
| 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 |
|
|
| |
| KNOWLEDGE_DIR = "knowledge" |
| |
| 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_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) |
|
|
| |
| 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 <strong>libbee</strong>" in last |
| or "i'm <strong>libbee</strong>" 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?<br><br>" |
| "You can choose one of these or type your question directly:<br>" |
| "• Find articles and books<br>" |
| "• Get full text for an article<br>" |
| "• Submit an Interlibrary Loan (ILL) request<br>" |
| "• Contact a librarian<br>" |
| "• 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 <strong>LibBee</strong>, 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 <strong>LibBee</strong>, 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 = [ |
| |
| { |
| "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"], |
| }, |
| |
| { |
| "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"], |
| }, |
| |
| { |
| "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"], |
| }, |
| |
| { |
| "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", |
| ], |
| |
| "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 <strong>{staff['full_name']}</strong>?<br><br>" |
| f"<strong>{staff['full_name']}</strong> is the <strong>{staff['role']}</strong>. {staff['details']}" |
| ) |
|
|
| def _staff_role_answer(staff: dict, question: str) -> str: |
| return ( |
| f"For <strong>{question}</strong>, the best match is <strong>{staff['full_name']}</strong> — " |
| f"<strong>{staff['role']}</strong>.<br><br>{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", |
| } |
|
|
| |
| vectorstore = None |
| http_client = None |
| kb_staff_directory = [] |
|
|
|
|
| |
| 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 |
| )''') |
| |
| 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_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 = {} |
| START_TIME = time.time() |
|
|
| def create_session(): |
| token = secrets.token_hex(32) |
| admin_sessions[token] = time.time() + 86400 |
| 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}") |
|
|
|
|
| |
| 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") |
|
|
| |
| 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 |
|
|
| |
| 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 |
|
|
|
|
| |
| 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"} |
|
|
|
|
| |
| 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] |
|
|
|
|
| |
| 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"} |
|
|
|
|
| |
| 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 |
|
|
|
|
| |
| 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} |
|
|
| |
| 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}" |
|
|
| |
| 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() |
|
|
| |
| 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]}") |
|
|
| |
| 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) |
|
|
| |
| |
| 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:""" |
|
|
| |
| 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() |
|
|
| |
| if answer == "NO_LIBRARY_ANSWER" or answer.startswith("NO_LIBRARY_ANSWER"): |
| return {"answer": "", "sources": [], "has_answer": False} |
|
|
| |
| |
| |
| 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} |
|
|
|
|
| |
| def is_medical_topic(text): |
| lower = text.lower() |
| return any(kw in lower for kw in MEDICAL_KEYWORDS) |
|
|
|
|
| |
| @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...") |
|
|
|
|
| |
| app = FastAPI(title="KU Library AI Agent", lifespan=lifespan) |
| app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) |
|
|
|
|
| |
| class ChatMessage(BaseModel): |
| role: str |
| content: str |
|
|
| class SearchRequest(BaseModel): |
| query: str |
| source: str = "primo" |
| 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] = [] |
|
|
|
|
| |
| @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")), |
| }, |
| } |
|
|
|
|
| |
| @app.get("/year") |
| def get_year(): |
| now = datetime.utcnow() |
| return { |
| "year": now.year, |
| "month": now.month, |
| "date": now.strftime("%Y-%m-%d"), |
| } |
|
|
|
|
| |
| 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: |
| |
| |
| |
| |
| 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} |
|
|
| |
| 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") |
|
|
| |
| 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") |
|
|
| |
| 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") |
|
|
| |
| 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") |
|
|
| |
| 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() |
| |
| 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 <strong>Khalifa University Library</strong> hours:<br><br>" |
| "<strong>Habshan Library (SAN Campus):</strong><br>" |
| "Sunday – Wednesday: 8:00 AM – 7:00 PM<br>" |
| "Thursday: 8:00 AM – 4:00 PM<br>" |
| "Friday: Closed<br>" |
| "Saturday: 12:00 PM – 8:00 PM<br><br>" |
| "⚠️ Hours may change during exam periods, public holidays, and Ramadan. " |
| "Always check the official Library Hours page for the latest times:<br>" |
| f"<a href=\"{LIBRARY_HOURS_URL}\" target=\"_blank\">{LIBRARY_HOURS_URL}</a>" |
| ) |
|
|
|
|
| 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 <strong>two branches</strong>:<br><br>" |
| "<strong>1. Main Campus Library</strong><br>" |
| "Location: E Building, Main Campus, Abu Dhabi<br>" |
| "Phone: <a href=\"tel:+97123124604\">+971 2 312 4604</a><br><br>" |
| "<strong>2. Habshan Library (SAN Campus)</strong><br>" |
| "Location: Building 5, Ground Floor & First Floor, Sas Al Nakhl (SAN) Campus, Abu Dhabi<br>" |
| "Phone: <a href=\"tel:+97123123160\">+971 2 312 3160</a><br><br>" |
| "For general library enquiries: " |
| "<a href=\"https://library.ku.ac.ae/AskUs\" target=\"_blank\">Ask a Librarian</a> " |
| "or email <a href=\"mailto:libse@ku.ac.ae\">libse@ku.ac.ae</a>." |
| ) |
|
|
| def _ku_general_redirect_answer() -> str: |
| return ( |
| f"I’m <strong>LibBee</strong>, the Khalifa University <strong>Library</strong> AI Assistant, so I’m best for library resources, services, databases, research help, and staff contacts.<br><br>" |
| f"For general Khalifa University questions, please visit the main KU website: <a href=\"{KU_MAIN_URL}\" target=\"_blank\">{KU_MAIN_URL}</a>. " |
| "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. |
| """ |
| |
| |
| 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() |
| |
| 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"], |
| } |
|
|
| |
| 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 <strong>LibBee</strong>, the Khalifa University Library AI Assistant.<br><br>" |
| "I can help you with articles and books, databases, full text, Interlibrary Loan (ILL), " |
| "library services, staff contacts, ORCID, RefWorks, and Open Access.<br><br>" |
| "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 <strong>LibBee</strong>, 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) |
|
|
| |
| @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 |
| |
| 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)} |
|
|
|
|
| |
| 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"} |
|
|
|
|
| |
| @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 |
| |
| 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)} |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| _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. |
| """ |
| |
| 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. |
| """ |
| |
| 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 <strong>LibBee</strong>, 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, |
| } |
|
|
|
|
| |
| @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() |
|
|
| |
| |
| |
| |
|
|
| |
| if PURE_GREETING_RE.match(q_clean): |
| elapsed = time.time() - start |
| return _make_agent_response( |
| answer=( |
| "Hi! I'm <strong>LibBee</strong>, the Khalifa University Library AI Assistant.<br><br>" |
| "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.<br><br>" |
| "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", |
| ) |
|
|
| |
| 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", |
| ) |
|
|
| |
| 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", |
| ) |
|
|
| |
| 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", |
| ) |
|
|
| |
| 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", |
| ) |
|
|
| |
| 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, |
| ) |
|
|
| |
| 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, |
| ) |
|
|
| |
| 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, |
| ) |
|
|
| |
| |
| |
| |
| |
|
|
| cls_result = await _llm_classify(question, history) |
| intent = cls_result.get("intent", "general") |
| casual_hint = cls_result.get("casual_answer", "") |
|
|
| |
| |
| |
|
|
| |
| 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", |
| ) |
|
|
| |
| if intent == "library_info": |
| |
| 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"): |
| |
| 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, |
| ) |
| else: |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| |
| 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}") |
|
|
| |
| |
| 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")) |
| |
| 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}") |
|
|
| |
| |
| |
| if tier2_answer and len(tier2_answer.split()) >= 10: |
| |
| final_answer = tier2_answer |
| final_sources = tier2_sources |
| final_tools = tools_used + ["ku_web_search"] |
| elif tier1_answer and len(tier1_answer.split()) >= 10: |
| |
| final_answer = tier1_answer |
| final_sources = [] |
| final_tools = tools_used + ["ku_only_llm"] |
| else: |
| |
| 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, |
| ) |
|
|
| |
| 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 = await _answer_general(question, history) |
| gen_answer = general.get("answer", "").strip() |
| |
| 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 " |
| "<a href=\"https://library.ku.ac.ae/AskUs\" target=\"_blank\">Ask a Librarian</a>. " |
| "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, |
| ) |
|
|
|
|
|
|
| |
| @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 |
|
|
| |
| if os.path.exists(FAISS_INDEX_PATH): |
| shutil.rmtree(FAISS_INDEX_PATH) |
| print(f"Deleted old FAISS index at {FAISS_INDEX_PATH}") |
|
|
| |
| 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": []} |
|
|
| |
| 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()} |
|
|
|
|
| |
| @app.get("/admin/login", response_class=HTMLResponse) |
| async def admin_login_page(): |
| return """<!DOCTYPE html> |
| <html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"> |
| <title>KU Library AI — Admin Login</title> |
| <style> |
| *{box-sizing:border-box;margin:0;padding:0} |
| body{font-family:'Segoe UI',system-ui,sans-serif;background:#f0f4f8;display:flex;justify-content:center;align-items:center;min-height:100vh} |
| .login-box{background:#fff;border-radius:16px;padding:40px;width:360px;box-shadow:0 4px 24px rgba(0,0,0,.1);text-align:center} |
| h1{color:#003366;font-size:1.3rem;margin-bottom:6px} |
| .subtitle{color:#6b7280;font-size:.84rem;margin-bottom:24px} |
| input{width:100%;padding:12px 14px;border:1.5px solid #d1d5db;border-radius:10px;font-size:.92rem;margin-bottom:14px} |
| input:focus{outline:none;border-color:#003366} |
| .btn{width:100%;padding:12px;background:#003366;color:#fff;border:none;border-radius:10px;font-weight:700;font-size:.92rem;cursor:pointer} |
| .btn:hover{background:#004488} |
| .error{color:#dc2626;font-size:.82rem;margin-bottom:10px;display:none} |
| </style></head><body> |
| <div class="login-box"> |
| <h1>🔐 Admin Login</h1> |
| <div class="subtitle">KU Library AI Dashboard</div> |
| <div class="error" id="err">Incorrect password. Try again.</div> |
| <form onsubmit="return doLogin(event)"> |
| <input type="password" id="pw" placeholder="Enter admin password" autofocus> |
| <button type="submit" class="btn">Login</button> |
| </form> |
| </div> |
| <script> |
| async function doLogin(e){ |
| e.preventDefault(); |
| const pw=document.getElementById('pw').value; |
| const r=await fetch('/admin/auth',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({password:pw})}); |
| if(r.ok){window.location.href='/admin';} |
| else{document.getElementById('err').style.display='block';document.getElementById('pw').value='';document.getElementById('pw').focus();} |
| return false; |
| } |
| </script></body></html>""" |
|
|
|
|
| @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 |
|
|
|
|
| |
| 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"""<!DOCTYPE html> |
| <html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"> |
| <title>KU Library AI — Admin</title> |
| <style> |
| *{{box-sizing:border-box;margin:0;padding:0}} |
| body{{font-family:'Segoe UI',system-ui,sans-serif;background:#f0f4f8;color:#1a1a2e;padding:20px;max-width:1200px;margin:0 auto}} |
| h1{{color:#003366;font-size:1.3rem;margin-bottom:4px}} |
| .sub{{color:#6b7280;font-size:.82rem;margin-bottom:14px}} |
| .tabs{{display:flex;gap:2px;border-bottom:2px solid #d1d5db;margin-bottom:14px}} |
| .tab{{padding:8px 16px;cursor:pointer;font-weight:600;font-size:.84rem;color:#6b7280;border:none;background:none;border-bottom:3px solid transparent;margin-bottom:-2px}} |
| .tab.active{{color:#003366;border-bottom-color:#003366}} |
| .tc{{display:none}}.tc.active{{display:block}} |
| .grid{{display:grid;grid-template-columns:repeat(auto-fit,minmax(160px,1fr));gap:10px;margin-bottom:16px}} |
| .stat{{background:#fff;border-radius:10px;padding:12px;border:1px solid #d1d5db;text-align:center}} |
| .stat .n{{font-size:1.6rem;font-weight:800;color:#003366}}.stat .l{{font-size:.76rem;color:#6b7280}} |
| .card{{background:#fff;border-radius:10px;padding:14px;border:1px solid #d1d5db;margin-bottom:12px}} |
| .card h2{{font-size:.9rem;color:#003366;margin-bottom:8px}} |
| table{{width:100%;border-collapse:collapse;font-size:.78rem}} |
| th{{background:#003366;color:#fff;padding:6px 8px;text-align:left}} |
| td{{padding:5px 8px;border-bottom:1px solid #e5e7eb}} |
| tr:hover{{background:#f9fafb}} |
| .badge{{padding:2px 7px;border-radius:8px;font-size:.68rem;font-weight:600}} |
| .err{{color:#dc2626;font-size:.72rem}} |
| canvas{{max-height:180px}} |
| .fg{{margin-bottom:12px}} |
| .fg label{{display:block;font-weight:700;font-size:.82rem;color:#003366;margin-bottom:3px}} |
| .fg .h{{font-size:.72rem;color:#6b7280;margin-bottom:3px}} |
| textarea,input[type=text],select{{width:100%;padding:9px;border:1.5px solid #d1d5db;border-radius:8px;font-family:inherit;font-size:.82rem;resize:vertical}} |
| textarea:focus,input:focus,select:focus{{outline:none;border-color:#003366}} |
| .btn{{padding:9px 20px;border:none;border-radius:8px;font-weight:700;cursor:pointer;font-size:.84rem}} |
| .bp{{background:#003366;color:#fff}}.bp:hover{{background:#004488}} |
| .bg{{background:#059669;color:#fff}} |
| .br{{background:#dc2626;color:#fff}} |
| .toggle{{display:flex;align-items:center;gap:8px}} |
| .toggle input{{width:18px;height:18px;accent-color:#003366}} |
| .st{{padding:8px 14px;border-radius:8px;font-size:.8rem;font-weight:600;display:none;margin-top:8px}} |
| .st.ok{{background:#d1fae5;color:#065f46;display:block}} |
| .st.er{{background:#fecaca;color:#991b1b;display:block}} |
| .two{{display:grid;grid-template-columns:1fr 1fr;gap:12px}} |
| @media(max-width:768px){{.two{{grid-template-columns:1fr}}}} |
| #loading{{text-align:center;padding:40px;color:#6b7280}} |
| </style> |
| <script src="https://cdnjs.cloudflare.com/ajax/libs/Chart.js/4.4.1/chart.umd.min.js"></script> |
| </head><body> |
| <h1>📊 KU Library AI — Admin <span style="float:right;display:flex;gap:6px;align-items:center"><button onclick="location.reload()" class="btn bp" style="padding:5px 12px;font-size:.76rem">🔄 Refresh</button><a href="/admin/logout" style="font-size:.76rem;color:#dc2626;text-decoration:none;padding:5px 12px;border:1.5px solid #dc2626;border-radius:8px;font-weight:600">🔒 Logout</a></span></h1> |
| <div class="sub">Khalifa University Library AI Agent</div> |
| |
| <div class="tabs"> |
| <div class="tab active" onclick="st(this,'analytics')">📈 Analytics</div> |
| <div class="tab" onclick="st(this,'behavior')">🤖 Bot Behavior</div> |
| <div class="tab" onclick="st(this,'queries')">📋 Query Log</div> |
| <div class="tab" onclick="st(this,'system')">⚙️ System</div> |
| </div> |
| |
| <div class="tc active" id="t-analytics"><div id="loading">Loading analytics from Cloudflare D1…</div></div> |
| |
| <div class="tc" id="t-behavior"> |
| <div class="card"><h2>🤖 Bot Behavior</h2><p style="font-size:.8rem;color:#6b7280;margin-bottom:12px">Changes apply immediately.</p> |
| <div class="fg"><label>Welcome Message</label><div class="h">First message users see.</div><textarea id="c-welcome_message" rows="3">{esc(config.get('welcome_message',''))}</textarea></div> |
| <div class="fg"><label>Bot Personality / System Prompt</label><div class="h">Core instructions for every LLM call.</div><textarea id="c-bot_personality" rows="4">{esc(config.get('bot_personality',''))}</textarea></div> |
| <div class="fg"><label>Custom Instructions</label><div class="h">Extra guidance, e.g. "During Ramadan, mention reduced hours."</div><textarea id="c-custom_instructions" rows="3">{esc(config.get('custom_instructions',''))}</textarea></div> |
| <div class="fg"><label>📢 Announcement Banner</label><div class="h">Shows at top of chat. Leave empty to hide.</div><textarea id="c-announcement" rows="2">{esc(config.get('announcement',''))}</textarea></div> |
| <div class="two"><div class="fg"><label>Default Model</label><select id="c-default_model"><option value="gpt" {'selected' if config.get('default_model')=='gpt' else ''}>ChatGPT</option><option value="claude" {'selected' if config.get('default_model')=='claude' else ''}>Claude</option></select></div> |
| <div class="fg"><label>Max Results</label><select id="c-max_results"><option value="3" {'selected' if config.get('max_results')=='3' else ''}>3</option><option value="5" {'selected' if config.get('max_results')=='5' else ''}>5</option><option value="10" {'selected' if config.get('max_results')=='10' else ''}>10</option></select></div></div> |
| <div id="sv" class="st"></div><button class="btn bp" onclick="saveConfig()">💾 Save</button> |
| </div></div> |
| |
| <div class="tc" id="t-queries"><div id="recent-loading">Loading query log…</div></div> |
| |
| <div class="tc" id="t-system"> |
| <div class="card"><h2>⚙️ Controls</h2> |
| <div class="fg"><div class="toggle"><input type="checkbox" id="c-maintenance_mode" {'checked' if config.get('maintenance_mode')=='true' else ''}><label><strong>Maintenance Mode</strong></label></div></div> |
| <div class="fg"><label>Maintenance Message</label><textarea id="c-maintenance_message" rows="2">{esc(config.get('maintenance_message',''))}</textarea></div> |
| <div style="display:flex;gap:8px;margin-top:10px;flex-wrap:wrap"> |
| <button class="btn bp" onclick="saveConfig()">💾 Save Config</button> |
| <button class="btn bg" onclick="rebuildIdx()">🔄 Rebuild RAG</button> |
| <button class="btn" style="background:#dc2626;color:#fff;border:none;padding:8px 14px;border-radius:8px;font-weight:600;cursor:pointer" onclick="restartSpace()">♻️ Restart Space</button> |
| </div> |
| <div id="sys" class="st"></div> |
| </div> |
| |
| <div class="card"><h2>🟢 Live Status</h2> |
| <div id="status-panel" style="font-size:.85rem">Loading status…</div> |
| </div> |
| |
| <div class="card"><h2>System Info</h2><table> |
| <tr><td><strong>Knowledge Files</strong></td><td>{len(glob.glob(os.path.join(KNOWLEDGE_DIR,'*.txt')))} files</td></tr> |
| <tr><td><strong>OpenAI API</strong></td><td>{'✅ Configured' if os.environ.get('OPENAI_API_KEY') else '❌ Missing'}</td></tr> |
| <tr><td><strong>Anthropic API</strong></td><td>{'✅ Configured' if os.environ.get('ANTHROPIC_API_KEY') else '❌ Missing'}</td></tr> |
| <tr><td><strong>PRIMO API</strong></td><td>{'✅ Configured' if os.environ.get('PRIMO_API_KEY') else '❌ Missing'}</td></tr> |
| <tr><td><strong>Admin Password</strong></td><td>{'✅ Set' if os.environ.get('ADMIN_PASSWORD') else '❌ Missing'}</td></tr> |
| <tr><td><strong>Cloudflare D1</strong></td><td>Connected via Worker</td></tr> |
| </table></div></div> |
| |
| <script> |
| const W='{WORKER_URL}'; |
| function st(el,t){{document.querySelectorAll('.tab').forEach(e=>e.classList.remove('active'));document.querySelectorAll('.tc').forEach(e=>e.classList.remove('active'));el.classList.add('active');document.getElementById('t-'+t).classList.add('active');}} |
| |
| // Fetch analytics from Cloudflare D1 |
| fetch(W+'/analytics').then(r=>r.json()).then(d=>{{ |
| const el=document.getElementById('t-analytics'); |
| el.innerHTML=` |
| <div class="grid"> |
| <div class="stat"><div class="n">${{d.total}}</div><div class="l">Total</div></div> |
| <div class="stat"><div class="n">${{d.today}}</div><div class="l">Today</div></div> |
| <div class="stat"><div class="n">${{d.week}}</div><div class="l">This Week</div></div> |
| <div class="stat"><div class="n">${{(d.avg_time||0).toFixed(1)}}s</div><div class="l">Avg Time</div></div> |
| <div class="stat"><div class="n">${{d.errors}}</div><div class="l">Errors</div></div> |
| </div> |
| <div class="two"> |
| <div class="card"><h2>Tool Usage</h2><table><tr><th>Tool</th><th>Count</th></tr>${{d.tools.map(t=>`<tr><td>${{t.tool_used}}</td><td>${{t.c}}</td></tr>`).join('')}}</table></div> |
| <div class="card"><h2>Model Usage</h2><table><tr><th>Model</th><th>Count</th></tr>${{d.models.map(m=>`<tr><td>${{m.model}}</td><td>${{m.c}}</td></tr>`).join('')}}</table></div> |
| </div> |
| <div class="two"> |
| <div class="card"><h2>Hourly</h2><canvas id="hc"></canvas></div> |
| <div class="card"><h2>Daily (14d)</h2><canvas id="dc"></canvas></div> |
| </div> |
| <div class="card"><h2>Popular (Top 20)</h2><table><tr><th>Question</th><th>Count</th></tr>${{d.popular.map(p=>`<tr><td>${{(p.question||'').substring(0,70)}}</td><td>${{p.c}}</td></tr>`).join('')}}</table></div>`; |
| if(d.hourly.length)new Chart(document.getElementById('hc'),{{type:'bar',data:{{labels:d.hourly.map(h=>h.hour+':00'),datasets:[{{label:'Q',data:d.hourly.map(h=>h.c),backgroundColor:'#003366'}}]}},options:{{responsive:true,plugins:{{legend:{{display:false}}}}}}}}); |
| if(d.daily.length)new Chart(document.getElementById('dc'),{{type:'line',data:{{labels:d.daily.map(x=>(x.day||'').slice(5)),datasets:[{{label:'Q',data:d.daily.map(x=>x.c),borderColor:'#003366',backgroundColor:'rgba(0,51,102,0.1)',fill:true,tension:.3}}]}},options:{{responsive:true,plugins:{{legend:{{display:false}}}}}}}}); |
| }}).catch(e=>{{document.getElementById('t-analytics').innerHTML='<div class="card" style="color:#dc2626">Failed to load analytics: '+e.message+'<br>Make sure D1 is initialized: <a href="'+W+'/analytics/init" target="_blank">Click here to init DB</a></div>';}}); |
| |
| |
| // Fetch local feedback stats |
| fetch('/admin/feedback-stats').then(r=>r.json()).then(d=>{{ |
| const host=document.getElementById('t-analytics'); |
| if(!host) return; |
| const s=(d.summary||[]).reduce((a,x)=>{{a[x.feedback]=x.c;return a;}},{{}}); |
| const card=document.createElement('div'); |
| card.className='card'; |
| card.innerHTML=`<h2>👍 User Feedback</h2> |
| <div class="grid" style="margin-bottom:10px"> |
| <div class="stat"><div class="n">${{s.up||0}}</div><div class="l">Helpful</div></div> |
| <div class="stat"><div class="n">${{s.down||0}}</div><div class="l">Not Helpful</div></div> |
| </div> |
| <div class="two"> |
| <div><h2 style="margin-bottom:6px">Recent feedback</h2><table><tr><th>Time</th><th>Question</th><th>Feedback</th></tr>${{(d.recent||[]).map(r=>`<tr><td>${{(r.timestamp||'').substring(0,19)}}</td><td>${{(r.question||'').substring(0,60)}}</td><td>${{r.feedback==='up'?'👍':'👎'}}</td></tr>`).join('')}}</table></div> |
| <div><h2 style="margin-bottom:6px">Top weak questions</h2><table><tr><th>Question</th><th>Count</th></tr>${{(d.weak||[]).map(r=>`<tr><td>${{(r.question||'').substring(0,70)}}</td><td>${{r.c}}</td></tr>`).join('')}}</table></div> |
| </div>`; |
| host.appendChild(card); |
| }}).catch(()=>{{}}); |
| |
| // Fetch recent queries |
| fetch(W+'/analytics/recent').then(r=>r.json()).then(d=>{{ |
| const el=document.getElementById('t-queries'); |
| el.innerHTML='<div class="card"><h2>Recent (50)</h2><table><tr><th>Time</th><th>Question</th><th>Tool</th><th>Model</th><th>Results</th><th>Status</th></tr>'+(d.results||[]).map(r=>`<tr><td>${{(r.timestamp||'').substring(0,19)}}</td><td title="${{r.question}}">${{(r.question||'').substring(0,55)}}</td><td><span class="badge" style="background:#e0e7ff">${{r.tool_used}}</span></td><td>${{r.model}}</td><td>${{r.result_count}}</td><td class="err">${{r.error||'✓'}}</td></tr>`).join('')+'</table></div>'; |
| }}).catch(e=>{{document.getElementById('t-queries').innerHTML='<div class="card" style="color:#dc2626">'+e.message+'</div>';}}); |
| |
| async function saveConfig(){{ |
| const keys=['welcome_message','bot_personality','custom_instructions','announcement','default_model','max_results','maintenance_message']; |
| const data={{}};keys.forEach(k=>{{data[k]=document.getElementById('c-'+k).value;}}); |
| data.maintenance_mode=document.getElementById('c-maintenance_mode').checked?'true':'false'; |
| const r=await fetch('/admin/config',{{method:'POST',headers:{{'Content-Type':'application/json'}},body:JSON.stringify(data)}}); |
| const res=await r.json(); |
| const s=document.getElementById('sv')||document.getElementById('sys'); |
| s.className=res.status==='ok'?'st ok':'st er'; s.textContent=res.status==='ok'?'✅ Saved!':'❌ Error'; |
| setTimeout(()=>s.style.display='none',3000); |
| }} |
| async function rebuildIdx(){{ |
| const s=document.getElementById('sys');s.className='st ok';s.textContent='🔄 Rebuilding...';s.style.display='block'; |
| const r=await fetch('/rebuild',{{method:'POST'}});const res=await r.json(); |
| s.textContent=res.status==='ok'?'✅ Rebuilt: '+res.chunks+' chunks from '+res.documents+' files':'❌ '+(res.error||JSON.stringify(res)); |
| s.className=res.status==='ok'?'st ok':'st er'; |
| loadStatus(); // refresh status after rebuild |
| }} |
| |
| async function restartSpace(){{ |
| if(!confirm('Restart the HF Space? The service will be unavailable for ~30-60 seconds.')) return; |
| const s=document.getElementById('sys');s.className='st ok';s.textContent='♻️ Restarting...';s.style.display='block'; |
| try {{ |
| await fetch('/restart',{{method:'POST'}}); |
| s.textContent='♻️ Space restarting — please wait 30-60 seconds then refresh this page.'; |
| // Poll until back online |
| let attempts=0; |
| const poll=setInterval(async()=>{{ |
| attempts++; |
| try{{ |
| const r=await fetch('/status'); |
| if(r.ok){{clearInterval(poll);s.textContent='✅ Space is back online!';s.className='st ok';loadStatus();}} |
| }}catch(e){{ |
| s.textContent=`♻️ Still restarting... (${{attempts*5}}s elapsed)`; |
| }} |
| if(attempts>24){{clearInterval(poll);s.textContent='⚠️ Restart taking longer than expected. Try refreshing the page.';}} |
| }},5000); |
| }}catch(e){{s.textContent='❌ '+e.message;s.className='st er';}} |
| }} |
| |
| async function loadStatus(){{ |
| const el=document.getElementById('status-panel'); |
| if(!el) return; |
| try{{ |
| const r=await fetch('/status'); |
| if(!r.ok){{el.innerHTML='<span style="color:#dc2626">❌ HF Space not responding</span>';return;}} |
| const d=await r.json(); |
| const uptimeMin=Math.round((d.uptime_secs||0)/60); |
| const memColor=d.mem_pct>85?'#dc2626':d.mem_pct>70?'#d97706':'#16a34a'; |
| el.innerHTML=` |
| <table style="width:100%"> |
| <tr><td><strong>Space Status</strong></td><td><span style="color:#16a34a;font-weight:700">✅ Running</span></td></tr> |
| <tr><td><strong>Vectorstore</strong></td><td>${{d.vectorstore==='ready'?'✅ Ready ('+d.chunks+' chunks, '+d.kb_files+' files)':'❌ Not initialized'}}</td></tr> |
| <tr><td><strong>OpenAI API</strong></td><td>${{d.openai?'✅ Configured':'❌ Missing'}}</td></tr> |
| <tr><td><strong>Anthropic API</strong></td><td>${{d.anthropic?'✅ Configured':'❌ Missing'}}</td></tr> |
| <tr><td><strong>PRIMO API</strong></td><td>${{d.primo?'✅ Configured':'❌ Missing'}}</td></tr> |
| <tr><td><strong>Memory Usage</strong></td><td><span style="color:${{memColor}};font-weight:700">${{d.mem_pct}}%</span> (${{d.mem_used_mb}}MB / ${{d.mem_total_mb}}MB)</td></tr> |
| <tr><td><strong>Uptime</strong></td><td>${{uptimeMin < 60 ? uptimeMin+'m' : Math.floor(uptimeMin/60)+'h '+uptimeMin%60+'m'}}</td></tr> |
| </table> |
| <div style="margin-top:8px;font-size:.74rem;color:#6b7280">Last checked: ${{new Date().toLocaleTimeString()}}</div>`; |
| }}catch(e){{ |
| el.innerHTML='<span style="color:#dc2626">❌ Cannot reach HF Space: '+e.message+'</span>'; |
| }} |
| }} |
| |
| // Load status on page load and every 30 seconds |
| loadStatus(); |
| setInterval(loadStatus, 30000); |
| </script> |
| </body></html>""" |
|
|
|
|
| |
| @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) |
| threading.Thread(target=do_restart, daemon=True).start() |
| return {"status": "restarting", "message": "Space is restarting. Please wait 30-60 seconds."} |
|
|
|
|
| |
| @app.get("/status") |
| async def get_status(): |
| |
| |
| try: |
| |
| 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 |
| 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: |
| |
| 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), |
| } |
|
|
|
|
| |
| @app.get("/admin/stats") |
| async def admin_stats(request: Request): |
| if not verify_session(request): |
| raise HTTPException(status_code=401, detail="Not authenticated") |
| |
| 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} |
|
|
|
|
| |
| @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} |
|
|
| |
| @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"), |
| } |
|
|
|
|
| |
| @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"} |
|
|
