diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -1,3446 +1,219 @@ +"""LibBee - Main Application Entry Point +Phase 3: production hardening, persistence, metrics, and safer staff routing + +Fixes applied (v3.1): +1. _request_log replaced with cachetools.TTLCache — no more unbounded IP dict +2. MetricsService background flush task started in lifespan +3. Circular import risk reduced — rag_service / metrics_service exposed via + module-level accessors get_rag_service() / get_metrics_service() so routers + can import the accessor instead of doing a deferred 'from app import ...' +4. Startup errors no longer silently swallow RAG init exceptions """ -agent.py — LibBee v3.1 - -Fixes applied over v3: -1. CRITICAL: Removed broken import of build_primo_boolean_query, - build_primo_discovery_url, build_pubmed_url from search.py — those - functions do not exist there. Replaced with self-contained implementations - directly in this file. This was the primary cause of the 500 errors. - -2. Circular import hardened: 'from app import rag_service / metrics_service' - replaced with 'from app import get_rag_service / get_metrics_service' - accessors that are safe to call inside request handlers. - -3. Deprecated Claude model string updated: - claude-3-haiku-20240307 → claude-haiku-4-5-20251001 - -4. _llm_classify and all LLM call sites wrapped with explicit try/except - so any LLM failure falls back to rule-based logic, never 500s. - -5. AgentRequest model field validated: model is now Literal["gpt", "claude"] - so malformed values are caught at the Pydantic layer. - -6. _web_search_answer: updated claude model string to match fix #3. - -7. _contact_footer_for_query: removed broken _find_staff_by_token("muna") - default — Muna is not in STAFF_DIRECTORY, so default now falls through - to no specialist rather than silently returning None. -""" -import html -import json +import asyncio import logging -import re import time -import uuid -from datetime import datetime -from typing import Any, Dict, List, Literal, Optional, Tuple -from urllib.parse import quote +from contextlib import asynccontextmanager -import httpx -from fastapi import APIRouter, Request -from pydantic import BaseModel, ConfigDict, Field +from fastapi import FastAPI, Request +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse + +try: + from cachetools import TTLCache + _ttl_cache_available = True +except ImportError: + _ttl_cache_available = False from src.config import get_settings -from src.services.staff_service import ( - STAFF_DIRECTORY, - match_staff_name, - match_staff_role, - should_attempt_staff_lookup, - staff_name_answer, - staff_role_answer, -) +from src.services.cache_service import CacheService +from src.services.metrics_service import MetricsService +from src.services.rag_service import RAGService +logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) -router = APIRouter() - - -# ── Pydantic Models ─────────────────────────────────────────────────────────── - -class ChatMessage(BaseModel): - role: str - content: str - - -class SearchContextPayload(BaseModel): - context_id: str - topic: str - display_topic: Optional[str] = None - intent: str = "search_academic" - source: str = "primo" - resource_type: str = "articles" # "articles" | "books" | "both" - peer_reviewed: bool = False - open_access: bool = False - year_from: Optional[str] = None - year_to: Optional[str] = None - ai_tool_query: Optional[str] = None - primo_boolean_query: Optional[str] = None - created_at: float = Field(default_factory=time.time) - - -class ClientStatePayload(BaseModel): - session_id: Optional[str] = None - active_search_context_id: Optional[str] = None - recent_search_contexts: List[SearchContextPayload] = Field(default_factory=list) - follow_up_action: Optional[str] = None - follow_up_context_id: Optional[str] = None - - -class AgentRequest(BaseModel): - question: str - # FIX: validated literal — malformed model values caught at Pydantic layer - model: Literal["gpt", "claude"] = "gpt" - history: List[ChatMessage] = Field(default_factory=list) - client_state: Optional[ClientStatePayload] = None - - -class AgentResponse(BaseModel): - model_config = ConfigDict(extra="ignore") - - answer: str - intent: str = "general" - tools_used: List[str] = Field(default_factory=list) - search_results: List[dict] = Field(default_factory=list) - sources: List[dict] = Field(default_factory=list) - model_used: str = "gpt" - response_time: float = 0 - corrected_query: Optional[str] = None - natural_query: Optional[str] = None - database_query: Optional[str] = None - ask_librarian: Optional[bool] = None - is_follow_up: bool = False - follow_up_question: Optional[str] = None - follow_up_suggestions: List[dict] = Field(default_factory=list) - search_context: Optional[dict] = None - ai_tool_query: Optional[str] = None - primo_boolean_query: Optional[str] = None - primo_search_url: Optional[str] = None - primo_ai_url: Optional[str] = None - pubmed_search_url: Optional[str] = None - - -# ── Constants ───────────────────────────────────────────────────────────────── - -LIBRARY_HOURS_URL = "https://library.ku.ac.ae/hours" -KU_MAIN_URL = "https://www.ku.ac.ae/" -ASK_LIBRARIAN_URL = "https://library.ku.ac.ae/AskUs" -PRIMO_BASE_URL = "https://khalifa.primo.exlibrisgroup.com/discovery/search?vid=971KUOSTAR_INST:KU" -PRIMO_AI_URL = "https://khalifa.primo.exlibrisgroup.com/discovery/researchAssistant?vid=971KUOSTAR_INST:KU&query=" -HISTORY_WINDOW = 6 -CURRENT_YEAR = datetime.utcnow().year - -_RESOURCE_TYPE_NOISE = re.compile( - r"\b(articles?|papers?|books?|ebooks?|journals?|publications?|monographs?" - r"|peer[- ]reviewed|open[- ]access|studies|study|research)\b", - re.IGNORECASE, -) - -SOCIAL_RE = re.compile( - r"^(hi|hello|hey|good morning|good afternoon|good evening|how are you|hows it going" - r"|how are you doing|who are you|what are you|tell me about yourself|what can you do" - r"|what do you do|how can you help|what help can you provide|are you a bot|are you a robot" - r"|are you ai|are you an ai|thank you|thanks|thx|bye|goodbye|see you|lol|haha" - r"|you are silly|are you silly|are you dumb|stupid bot|dumb bot|idiot bot" - r"|joke|tell me a joke)$", - re.IGNORECASE, -) -PURE_GREETING_RE = re.compile( - r"^(hi|hello|hey|good morning|good afternoon|good evening)[!.\s]*$", - re.IGNORECASE, -) -GREETING_FOLLOWUP_RE = re.compile( - r"^(yes|yes please|yes check|sure|okay|ok|yeah|yep|please do|go ahead)$", - re.IGNORECASE, -) -HOURS_RE = re.compile( - r"\b(library hours?|opening hours?|closing hours?|open hours?|open today|open tomorrow" - r"|open now|open on|open this|closed today|closed tomorrow|closed on|closed this" - r"|when.*open|when.*close|when.*does.*library|what time.*open|what time.*close" - r"|what time.*library|opening time|closing time|is.*library.*open|is.*campus.*library" - r"|is.*habshan.*open|is.*san.*open|is.*library.*closed|will.*library.*open" - r"|will.*it.*open|does.*library.*open|library.*open.*today|library.*close.*today" - r"|habshan.*open|san.*open|main campus.*open|ramadan hours?|weekend hours?" - r"|friday hours?|saturday hours?|sunday hours?|library.*schedule|schedule.*library" - r"|what.*hours|till.*when|until.*when|what day.*open|which day.*open" - r"|what day.*closed|which day.*closed)\b", - re.IGNORECASE, -) -KU_CAMPUS_RE = re.compile( - r"\b(where is the library|where is.*library|library.*location|location.*library" - r"|which building|what building|which floor|main campus library|san campus library" - r"|habshan library|sas al nakhl library|library.*address|address.*library" - r"|how to get to.*library|how.*reach.*library|how.*find.*library" - r"|library.*campus(es)?|campus(es)?.*library|how many.*librar|how many.*branch" - r"|library.*branch(es)?|branch(es)?.*library|which campus.*library" - r"|library.*which campus|library.*map|map.*library|how many libraries" - r"|what libraries|which libraries|(ku|khalifa).*(has|have|got).*(librar|branch)" - r"|(librar|branch).*(ku|khalifa).*(has|have)" - r"|(all|both|two|2).*(libraries|library|branches)" - r"|tell me about.*(campus|libraries|branches)|describe.*(campus|libraries)" - r"|(what|where).*(campus|campuses).*(librar))\b", - re.IGNORECASE, -) -RESEARCH_CUE_RE = re.compile( - r"\b(find|search|look for|show me|get me|give me|locate|recommend|suggest" - r"|articles?|papers?|books?|journals?|studies|literature|research|systematic review" - r"|evidence|sources|database|databases|peer reviewed|open access|advances?" - r"|approach(es)?|method(s|ology)?|technique[s]?|application[s]?|framework[s]?" - r"|model[s]?|algorithm[s]?|publication[s]?|recent work|current state|state of the art)\b", - re.IGNORECASE, -) -LIBRARY_CUE_RE = re.compile( - r"\b(library|librarian|borrow|loan|renew|fine|study room|room booking|reserve a room" - r"|account|my library|interlibrary|ill|khazna|orcid|open access|apc|refworks" - r"|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" - r"|clinical trial|biomedical literature|medical literature)\b", - re.IGNORECASE, -) -CURRENT_INFO_RE = re.compile( - r"\b(current|currently|latest|today|recent|recently|now|this week|this month" - r"|news|update|updated|prime minister|ceo|minister|king|ruler|who won)\b", - re.IGNORECASE, -) -SUMMARY_RE = re.compile( - r"\b(summar(y|ize|ise|izing|ising)|overview|explain|what is|what are|describe" - r"|tell me about|give me a summary|give me an overview|give me a brief" - r"|brief (summary|overview|explanation|description)" - r"|research (overview|summary|landscape|state)" - r"|state of (the art|research|the field)" - r"|recent (advances?|developments?|progress|trends?)" - r"|current (state|trends?|advances?|research)" - r"|deep analysis|analysis of|review of|literature review|deep research|deep dive" - r"|what do we know about|how does|how do)\b", - re.IGNORECASE, -) -DATABASE_RECOMMEND_RE = re.compile( - r"(" - # explicit recommendation phrasing - r"\b(best database|best databases|which database|which databases" - r"|database for|databases for" - r"|where should i search|where should i start)\b" - # subscription / access phrasing - r"|databases?\s+(does|did|do)\s+.{0,40}subscri" - r"|\bsubscribed\s+databases?\b" - r"|\blibrary\s+subscri" - r"|\bwhat\s+databases?\b" - r"|\bdatabases?\s+(available|accessible)\b" - r"|\bdatabases?\s+(at|in|through|via)\s+(ku|khalifa)\b" - r"|\bku\s+(library\s+)?databases?\b" - r"|\bkhalifa.{0,20}databases?\b" - r"|\blibrary\s+databases?\b" - r"|\be.?resources?\s+(available|subscri)\b" - r"|\baccess\s+to\s+databases?\b" - r"|\blist\s+of\s+databases?\b" - # comparison phrasing ← NEW - r"|\b(compare|comparison|difference|differences)\s+(between\s+)?(\w+\s+)?(and\s+)?\w*\s*database" - r"|\bdifference\s+between.{0,60}database" - r"|\bscopus\s+(vs|versus|or)\s+(web of science|wos|google scholar)\b" - r"|\b(web of science|wos)\s+(vs|versus|or)\s+(scopus|google scholar)\b" - r")", - re.IGNORECASE, -) - -# ── New pattern detection regexes ──────────────────────────────────────────── - -HIGHLY_CITED_RE = re.compile( - r"\b(highly cited|most cited|top cited|citation count|most influential" - r"|most referenced|high impact papers?|most downloaded)\b", - re.IGNORECASE, -) - -BOOLEAN_STRATEGY_RE = re.compile( - r"\b(search strategy|boolean (search|operator|query|logic)" - r"|build.{0,20}search|help.{0,20}search strategy" - r"|how.{0,30}search.{0,30}(database|boolean|keyword)" - r"|refine.{0,30}(topic|question|research question|searchable)" - r"|turn.{0,30}(topic|question).{0,20}search" - r"|keywords? for.{0,30}search|search terms?" - r"|explain.{0,20}keyword|why.{0,20}(chose|choose|selected|pick).{0,20}keyword)\b", - re.IGNORECASE, -) - -FULLTEXT_REQUEST_RE = re.compile( - r"\b(" - # PDF / access problem — only fires when there's an explicit access failure - r"(can.t|cannot|couldn.t|don.t have|doesn.t have|no).{0,30}(access|pdf|full.?text)" - r"|(pdf|full.?text).{0,20}(not available|unavailable|can.t find|not found|missing)" - r"|found.{0,30}citation.{0,20}(but|not).{0,20}(pdf|full.?text)" - r"|article.{0,20}(behind.{0,10}paywall|not available|no access|access denied)" - r"|ku.{0,20}(doesn.t|does not|no).{0,20}(have|access).{0,20}(article|journal|pdf)" - r"|(how|where).{0,20}(get|find|access|download).{0,20}(full.?text|pdf).{0,20}article" - r"|get full.?text|access full.?text|download.{0,10}(article|pdf)" - r"|ku.{0,15}(don.t|doesn.t|no).{0,15}(subscribe|have access)" - r")\b", - re.IGNORECASE, -) - -PARTIAL_TITLE_RE = re.compile( - r"\b(only (remember|recall|know).{0,20}(title|name)" - r"|title (starts?|begins?).{0,15}with" - r"|article.{0,20}(called|titled|named|starts? with)" - r"|find.{0,20}(article|paper).{0,20}(called|titled|named)" - r"|looking for.{0,20}(paper|article).{0,20}(about|called))\b", - re.IGNORECASE, -) - -DUAL_RESOURCE_RE = re.compile( - r"\b(articles?.{0,30}(and|as well as|plus).{0,30}books?" - r"|books?.{0,30}(and|as well as|plus).{0,30}articles?" - r"|papers?.{0,30}(and|as well as|plus).{0,30}textbooks?" - r"|textbooks?.{0,30}(and|as well as|plus).{0,30}papers?" - r"|(literature review|articles?).{0,50}(background|textbook|introductory book)" - r"|(textbook|book).{0,50}(literature review|research articles?))\b", - re.IGNORECASE, -) - -THESES_RE = re.compile( - r"\b(thes(is|es)|dissertation|doctoral|phd thesis|master.s thesis" - r"|khazna|ku repository|institutional repository" - r"|proquest dissertations?)\b", - re.IGNORECASE, -) - -PREDATORY_RE = re.compile( - r"\b(predatory (journal|publisher|publication)" - r"|how.{0,20}(know|tell|check).{0,30}(peer.?reviewed|scholarly|legitimate|reputable)" - r"|is.{0,30}(journal|article).{0,30}(legitimate|reputable|trusted|peer.?reviewed)" - r"|peer.?reviewed vs|distinguish.{0,20}(peer|scholarly|predatory)" - r"|journal (quality|credibility|ranking|whitelist|blacklist)" - r"|beall.s list|doaj|cabells|think.check.submit)\b", - re.IGNORECASE, -) - -CITATION_CHAIN_RE = re.compile( - r"\b(cited by|forward citation|backward citation|reference.{0,15}(chain|mining)" - r"|trace.{0,30}(reference|citation)|who cited|papers? that cite" - r"|related (articles?|papers?).{0,20}(paper|article)" - r"|find similar.{0,20}(paper|article))\b", - re.IGNORECASE, -) - -REVIEW_ONLY_RE = re.compile( - r"\b(review (articles?|papers?|only)|only (review|systematic review)" - r"|limit.{0,20}review|show.{0,20}review|filter.{0,20}review" - r"|systematic reviews? only|literature reviews? only)\b", - re.IGNORECASE, -) - -ALT_TERMS_RE = re.compile( - r"\b(alternative (search terms?|keywords?|terms?)" - r"|other (search terms?|keywords?|ways? to search)" - r"|more (keywords?|search terms?)|suggest.{0,20}keywords?" - r"|related (terms?|keywords?|synonyms?)|synonyms? for.{0,20}search" - r"|different (keywords?|terms?|search))\b", - re.IGNORECASE, -) - -# ───────────────────────────────────────────────────────────────────────────── - -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", - "crispr", "gene therapy", "gene editing", "genomic medicine", "precision medicine", -] - -REFINEMENT_STOP_WORDS = { - "make", "it", "them", "those", "these", "the", "a", "an", "for", "to", "only", "show", - "find", "give", "me", "please", "and", "with", "within", "limit", "narrow", "search", - "results", "on", "about", "of", "this", "that", "topic", "same", "using", "from", - "peer", "reviewed", "review", "open", "access", "books", "book", "articles", "article", - "papers", "paper", "years", "year", "last", "past", "recent", "instead", "summary", - "overview", "brief", "pubmed", "primo", "database", "databases", "best", "deep", "research", - "tools", "tool", "guide", "ku", "library", -} - -_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 - "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/GET articles/papers/books on a topic, OR wants a summary/overview/analysis/explanation of an academic/research topic - "search_medical" — same as search_academic but topic is medical/clinical/nursing/pharmacy/health sciences - "general_recent" — questions about TRULY live/real-time data: breaking news, live sports scores, today's weather, current stock prices - "general" — ALL other factual questions: university info, people, history, science definitions, technology explanations - -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. - 3. "I am fed up with searching" alone = social. - 4. "I need help finding articles on diabetes" = search_medical. - 5. KU = Khalifa University. Do NOT confuse with Kuwait University. - 6. ANY summary, overview, analysis, deep dive, review, or explanation of an academic/research topic = search_academic or search_medical. - 7. Questions about people, university leadership, history, science definitions = general. - 8. Only use general_recent for truly live data changing by the minute. - -For "social" intent ONLY: include "casual_answer" — a warm 1-3 sentence LibBee response (no markdown, no bullets). -For all other intents: omit "casual_answer". - -Respond ONLY with valid JSON. No preamble, no markdown fences. -""" - - -# ── Utilities ───────────────────────────────────────────────────────────────── - -def _escape(text: str) -> str: - return html.escape(text or "") - - -def _normalize_whitespace(text: str) -> str: - return re.sub(r"\s+", " ", (text or "").strip()) - - -def _title_case_topic(topic: str) -> str: - return _normalize_whitespace(topic).strip().strip(".?") - - -def _strip_resource_noise(text: str) -> str: - cleaned = _RESOURCE_TYPE_NOISE.sub(" ", text or "") - cleaned = re.sub(r"\s+(AND|OR)\s+(AND|OR)\s+", " AND ", cleaned, flags=re.IGNORECASE) - cleaned = re.sub(r"^\s*(AND|OR)\s+", "", cleaned, flags=re.IGNORECASE) - cleaned = re.sub(r"\s+(AND|OR)\s*$", "", cleaned, flags=re.IGNORECASE) - return _normalize_whitespace(cleaned) - - -def _source_badge(source: str, url: str = "") -> str: - styles = { - "rag": ("📖", "KU Library Knowledge Base", "#fef3c7", "#92400e"), - "primo": ("📚", "KU Library PRIMO", "#eff6ff", "#1e40af"), - "semantic": ("🎓", "Semantic Scholar", "#f0fdf4", "#166534"), - "pubmed": ("🏥", "PubMed", "#d1fae5", "#065f46"), - "web": ("🌐", "Web Search", "#f5f3ff", "#4c1d95"), - "web_live": ("🌐", "Web Search · Live", "#fdf4ff", "#86198f"), - "libbee": ("🐝", "LibBee", "#fef9c3", "#713f12"), - "ku_library": ("📋", "KU Library", "#eff6ff", "#1e40af"), - "llm": ("🤖", "AI Knowledge Base", "#f3f4f6", "#374151"), - } - icon, label, bg, fg = styles.get(source, ("ℹ️", source, "#f3f4f6", "#374151")) - inner = f"{icon} Source: {label}" - if url: - inner = f'{inner}' - return ( - f'
{inner}
' - ) - - -# FIX: use get_metrics_service() accessor to avoid circular import fragility -def _safe_metrics_increment(key: str) -> None: - try: - from app import get_metrics_service - get_metrics_service().incr(key) - except Exception: - return - - -def _safe_metrics_bucket(bucket: str, key: str) -> None: - try: - from app import get_metrics_service - get_metrics_service().incr_bucket(bucket, key) - except Exception: - return - - -def _build_history_messages(history: List[ChatMessage]) -> List[dict]: - msgs = [] - for m in history[-HISTORY_WINDOW:]: - if m.role in ("user", "assistant") and m.content: - msgs.append({"role": m.role, "content": m.content[:300]}) - return msgs - - -def _is_summary_request(question: str) -> bool: - return bool(SUMMARY_RE.search(question or "")) - - -def _is_greeting_menu_followup(question: str, history: List[ChatMessage]) -> bool: - if not GREETING_FOLLOWUP_RE.match((question or "").strip()): - return False - for m in reversed(history): - if m.role == "assistant": - last = m.content.lower() - return "i'm libbee" in last or "are you looking for one of these" in last - return False - - -def _looks_library_hours_question(question: str) -> bool: - q = (question or "").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: - return bool(KU_CAMPUS_RE.search(question or "")) - - -def _looks_medical_search(question: str) -> bool: - q = (question or "").lower() - if not any(kw in q for kw in MEDICAL_KEYWORDS): - return False - return bool(MEDICAL_SEARCH_RE.search(q) or RESEARCH_CUE_RE.search(q) or SUMMARY_RE.search(q)) - - -def _looks_research_question(question: str) -> bool: - q = question or "" - return bool(RESEARCH_CUE_RE.search(q) or SUMMARY_RE.search(q)) - - -def _rule_based_classify(question: str) -> Dict[str, str]: - q = (question or "").lower() - if SOCIAL_RE.match((question or "").strip()): - return { - "intent": "social", - "casual_answer": ( - "Hello! I'm LibBee, the KU Library AI Assistant. " - "I'm here to help you with articles, books, databases, and library services. " - "What would you like to find today?" - ), - } - if _looks_medical_search(question): - return {"intent": "search_medical"} - if _looks_research_question(question): - return {"intent": "search_academic"} - if LIBRARY_CUE_RE.search(q): - return {"intent": "library_info"} - if CURRENT_INFO_RE.search(q): - return {"intent": "general_recent"} - return {"intent": "general"} - - -def _hours_answer() -> str: - """ - Return a rich, day-aware library hours answer with correct Normal and Semester Break schedules. - Always links to https://library.ku.ac.ae/hours for live schedule. - """ - from datetime import datetime, timezone, timedelta - # UAE is UTC+4 - uae_now = datetime.now(timezone(timedelta(hours=4))) - day_names = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"] - today = day_names[uae_now.weekday()] - - # Normal hours (classes in session) - _NORMAL = { - "Monday": ("Open", "8:00 AM – 9:00 PM"), - "Tuesday": ("Open", "8:00 AM – 9:00 PM"), - "Wednesday": ("Open", "8:00 AM – 9:00 PM"), - "Thursday": ("Open", "8:00 AM – 5:00 PM"), - "Friday": ("Open", "7:30 AM – 12:30 PM"), - "Saturday": ("Closed", ""), - "Sunday": ("Open", "12:00 PM – 8:00 PM"), - } - # Semester break hours (classes not in session) - _BREAK = { - "Monday": ("Open", "8:00 AM – 4:00 PM"), - "Tuesday": ("Open", "8:00 AM – 4:00 PM"), - "Wednesday": ("Open", "8:00 AM – 4:00 PM"), - "Thursday": ("Open", "8:00 AM – 4:00 PM"), - "Friday": ("Open", "7:30 AM – 12:00 PM"), - "Saturday": ("Closed", ""), - "Sunday": ("Closed", ""), - } - - def _day_row(day: str, status: str, hrs: str) -> str: - if status == "Closed": - color = "#dc2626" - label = "Closed" - elif "12:00 PM" in hrs or "12:30 PM" in hrs: - color = "#d97706" - label = hrs - else: - color = "#16a34a" - label = hrs - return ( - f"{day}" - f"{label}" - ) - - def _today_line(schedule: dict) -> str: - status, hrs = schedule.get(today, ("Unknown", "")) - if status == "Closed": - return (f'📅 Today ({today}): ' - f'Closed') - elif status == "Open": - return (f'📅 Today ({today}): ' - f'Open · {hrs}') - return f'📅 Today ({today}): Check live hours page.' - - def _schedule_table(schedule: dict) -> str: - rows = "".join( - _day_row(d, *schedule[d]) - for d in ["Monday","Tuesday","Wednesday","Thursday","Friday","Saturday","Sunday"] - ) - return ( - "" - + rows + "
" - ) - - return ( - "Here are the Khalifa University Library hours:

" - f"{_today_line(_NORMAL)}
" - "(based on normal semester schedule — see note below)

" - - "📚 Normal Hours " - "(when classes are in session)

" - + _schedule_table(_NORMAL) + - - "

🏖️ Semester Break Hours " - "(when classes are not in session)

" - + _schedule_table(_BREAK) + - - "

📍 Both campuses follow the same schedule:
" - "" - "Main Campus Library · E Building · " - '+971 2 312 4604
' - "Habshan Library · Building 5, SAN Campus · " - '+971 2 312 3160' - "

" - "⚠️ Hours change during Ramadan, public holidays, and exam periods. " - "Always check the live schedule for the most current times:
" - f'{LIBRARY_HOURS_URL}' - ) +rag_service = RAGService() +cache_service = CacheService() +metrics_service = MetricsService(get_settings().metrics_path) +# ── Rate-limit request log ───────────────────────────────────────────────────── +# TTLCache(maxsize, ttl) auto-evicts entries older than ttl seconds. +# Falls back to a plain dict with manual sweep if cachetools not installed. +if _ttl_cache_available: + _request_log = TTLCache(maxsize=10_000, ttl=120) + _request_log_lock = asyncio.Lock() +else: + from collections import defaultdict, deque + _request_log = defaultdict(deque) # type: ignore[assignment] + _request_log_lock = asyncio.Lock() -def _campus_answer() -> str: - return ( - "Khalifa University Library has two branches:

" - "1. Main Campus Library
" - "Location: E Building, Main Campus, Abu Dhabi
" - 'Reference desk: +971 2 312 4604

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

' - "Both branches offer study spaces, computer workstations, printing, scanning, " - "and access to physical collections.

" - f'For general library enquiries: Ask a Librarian ' - 'or email libse@ku.ac.ae.' - ) +# ── Public accessors (used by routers to avoid circular imports) ─────────────── -def _greeting_menu_clarify_answer() -> str: - return ( - "Sure — what would you like help with?

" - "You can choose one of these or type your question directly:
" - "• Find articles and books
" - "• Get full text for an article
" - "• Submit an Interlibrary Loan (ILL) request
" - "• Contact a librarian
" - "• Check library hours" - ) +def get_rag_service() -> RAGService: + return rag_service -# ── LLM Helpers ─────────────────────────────────────────────────────────────── +def get_metrics_service() -> MetricsService: + return metrics_service -def _get_llm(model: str, temperature: float, max_tokens: int): - settings = get_settings() - # FIX: updated model string — claude-3-haiku-20240307 is deprecated - if model == "claude" and settings.anthropic_api_key: - from langchain_anthropic import ChatAnthropic - return ChatAnthropic( - model="claude-haiku-4-5-20251001", - temperature=temperature, - max_tokens=max_tokens, - anthropic_api_key=settings.anthropic_api_key, - ) - from langchain_openai import ChatOpenAI - return ChatOpenAI( - model="gpt-4o-mini", - temperature=temperature, - max_tokens=max_tokens, - openai_api_key=settings.openai_api_key, - ) +# ── Background flush task ────────────────────────────────────────────────────── -async def _llm_classify(question: str, history: List[ChatMessage], model: str) -> Dict[str, str]: - settings = get_settings() - if not settings.openai_api_key and not settings.anthropic_api_key: - return _rule_based_classify(question) - # FIX: always fall back to rule-based on any exception — never propagate - try: - ctx_parts = [ - f"{m.role}: {m.content[:100]}" - for m in history[-HISTORY_WINDOW:] - if m.role in ("user", "assistant") - ] - ctx = " | ".join(ctx_parts) if ctx_parts else "none" - llm = _get_llm(model, temperature=0, max_tokens=120) - response = await llm.ainvoke([ - {"role": "system", "content": _CLASSIFIER_SYSTEM}, - {"role": "user", "content": f"Message: {question[:400]}\nContext: {ctx}\nJSON:"}, - ]) - 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") - if intent not in {"social", "library_info", "search_academic", "search_medical", "general_recent", "general"}: - intent = "general" - return {"intent": intent, "casual_answer": result.get("casual_answer", "")} - except Exception as e: - logger.warning(f"LLM classifier error: {e} — falling back to rule-based") - return _rule_based_classify(question) - +async def _metrics_flush_loop(interval: int = 30) -> None: + """Flush in-memory metric counters to disk every `interval` seconds.""" + while True: + await asyncio.sleep(interval) + try: + await metrics_service.flush() + except Exception as exc: + logger.warning("Metrics flush error: %s", exc) -async def _libbee_casual_response(question: str, history: List[ChatMessage], model: str, hint: str = "") -> str: - if hint and len(hint.strip()) > 20: - return hint.strip() - settings = get_settings() - if not settings.openai_api_key and not settings.anthropic_api_key: - return "I'm LibBee, the Khalifa University Library AI Assistant — always happy to help!" - try: - llm = _get_llm(model, temperature=0.6, max_tokens=160) - msgs = [{"role": "system", "content": ( - "You are LibBee, the Khalifa University Library AI Assistant. " - "Respond warmly and naturally in 1-3 sentences as a friendly librarian. " - "No markdown, no bullet points." - )}] - msgs.extend(_build_history_messages(history)) - msgs.append({"role": "user", "content": question}) - response = await llm.ainvoke(msgs) - return response.content.strip() - except Exception: - return "I'm LibBee, the Khalifa University Library AI Assistant — always happy to help!" +@asynccontextmanager +async def lifespan(app: FastAPI): + from src.services.staff_service import build_staff_index -async def _rag_answer(question: str, rag_results: List[dict], history: List[ChatMessage], model: str) -> str: settings = get_settings() - if not settings.openai_api_key and not settings.anthropic_api_key: - return rag_results[0]["content"] if rag_results else "" - context = "\n\n---\n\n".join(r["content"] for r in rag_results) - system = ( - "You are LibBee, the Khalifa University Library AI Assistant. " - "Answer the user's question using ONLY the context provided below. " - "Be concise, friendly, and use HTML
for line breaks where needed. " - f"If the context does not contain enough information, say so and suggest Ask a Librarian at {ASK_LIBRARIAN_URL}. " - "Never invent facts or URLs.\n\n" - f"CONTEXT:\n{context}" - ) - try: - llm = _get_llm(model, temperature=0.2, max_tokens=380) - msgs = [{"role": "system", "content": system}] - msgs.extend(_build_history_messages(history)) - msgs.append({"role": "user", "content": question}) - response = await llm.ainvoke(msgs) - return response.content.strip() - except Exception as e: - logger.error(f"_rag_answer error: {e}") - return ( - "I'm having trouble generating an answer right now. " - f'Please try Ask a Librarian.' - ) - -async def _llm_general_answer(question: str, history: List[ChatMessage], model: str) -> str: - settings = get_settings() - if not settings.openai_api_key and not settings.anthropic_api_key: - return ( - "I'm LibBee, the KU Library AI Assistant. " - f'For general university information, visit ku.ac.ae.' - ) - system = ( - "You are LibBee, the Khalifa University Library AI Assistant (Abu Dhabi, UAE). " - "Answer factually and concisely in 3-5 sentences. Use HTML
for line breaks. " - "KU means Khalifa University. If the topic is academic, briefly offer to help find library resources." - ) + # Build staff index try: - llm = _get_llm(model, temperature=0.3, max_tokens=360) - msgs = [{"role": "system", "content": system}] - msgs.extend(_build_history_messages(history)) - msgs.append({"role": "user", "content": question}) - response = await llm.ainvoke(msgs) - return response.content.strip() - except Exception as e: - logger.error(f"_llm_general_answer error: {e}") - return f'Having trouble right now. Visit ku.ac.ae.' - - -async def _web_search_answer(question: str, history: List[ChatMessage], model: str) -> str: - settings = get_settings() - # FIX: updated model string to match _get_llm fix - if model == "claude" and settings.anthropic_api_key: - try: - import anthropic - client = anthropic.Anthropic(api_key=settings.anthropic_api_key) - response = client.messages.create( - model="claude-haiku-4-5-20251001", - max_tokens=520, - system=( - "You are LibBee, the Khalifa University Library AI Assistant (Abu Dhabi, UAE). " - "Answer using current web search. Be concise and factual. Use HTML
for line breaks." - ), - tools=[{"type": "web_search_20250305", "name": "web_search"}], - messages=[{"role": "user", "content": question}], - ) - text = "".join(block.text for block in response.content if hasattr(block, "text")) - if text.strip(): - return text.strip() - except Exception as e: - logger.warning(f"Claude web search failed: {e}") + build_staff_index() + logger.info("Staff index built") + except Exception as exc: + logger.error("Staff index build failed: %s", exc) + # Initialize RAG if settings.openai_api_key: try: - from openai import OpenAI - client = OpenAI(api_key=settings.openai_api_key) - response = client.responses.create( - model="gpt-4o-mini", - tools=[{"type": "web_search_preview"}], - instructions=( - "You are LibBee, the Khalifa University Library AI Assistant (Abu Dhabi, UAE). " - "Answer using current web search. Be concise and factual. Use HTML
for line breaks." - ), - input=question, - ) - text = "" - for item in response.output: - if hasattr(item, "content"): - for block in item.content: - if hasattr(block, "text"): - text += block.text - if text.strip(): - return text.strip() - except Exception as e: - logger.warning(f"GPT web search failed: {e}") - - return await _llm_general_answer(question, history, model) - - -# ── Topic Extraction ────────────────────────────────────────────────────────── - -def _extract_topic_regex(question: str) -> str: - q = _normalize_whitespace(question) - prefixes = [ - "can u give me a deep analysis of ", "can you give me a deep analysis of ", - "give me a deep analysis of ", "deep analysis of ", - "give me a summary of ", "give me a summary on ", "give me a summary about ", - "give me an overview of ", "give me an overview on ", - "give me a brief overview of ", "give me a brief summary of ", - "summarize the latest studies in ", "summarize latest studies in ", - "summarize the latest research on ", "summarize latest research on ", - "summarize recent studies in ", "summarize recent research on ", - "latest studies in ", "latest research on ", "recent studies in ", "recent research on ", - "summarize ", "summarise ", "summary of ", "summary on ", "summary about ", - "overview of ", "overview on ", "brief overview of ", "brief summary of ", - "analysis of ", "analyze ", "analyse ", "explain ", "tell me about ", - "what is ", "what are ", "describe ", "review of ", "literature review on ", - "recent advances in ", "recent developments in ", "state of the art in ", - "state of research on ", "research overview of ", "research summary of ", - "find peer reviewed articles on ", "find peer reviewed papers on ", - "find articles on ", "find papers on ", "find books on ", "find articles about ", - "find papers about ", "search for articles on ", "search for papers on ", - "i need peer reviewed articles on ", "i need articles on ", "i need papers on ", - "i need books on ", "give me articles on ", "show me articles on ", - "get me articles on ", "look for articles on ", "articles on ", "papers on ", - "books on ", "research on ", "literature on ", "find ", "search for ", - "i need ", "give me ", "show me ", "get me ", - ] - lower = q.lower() - for prefix in prefixes: - if lower.startswith(prefix): - q = q[len(prefix):] - break - q = re.sub(r"\s+(please|thanks|thank you)\.?$", "", q, flags=re.IGNORECASE) - q = re.sub(r"\b(peer[- ]reviewed|open access|last \d+ years?|past \d+ years?)\b", "", q, flags=re.IGNORECASE) - q = _strip_resource_noise(q) - return _normalize_whitespace(q).strip(".?") - - -async def _extract_topic(question: str, model: str) -> str: - settings = get_settings() - if not settings.openai_api_key and not settings.anthropic_api_key: - return _extract_topic_regex(question) - try: - llm = _get_llm(model, temperature=0, max_tokens=32) - response = await llm.ainvoke([ - {"role": "system", "content": ( - "Extract only the core research topic from the user's question. " - "Return 2-7 keywords ONLY — no punctuation, no sentence, no explanation. " - "CRITICAL: do NOT include words like articles, papers, books, journals, " - "peer-reviewed, open-access, studies, or research in your output. " - "Those are filters, not topic words. Return ONLY the subject matter.\n" - "Examples:\n" - "'find articles on indian politics' → indian politics\n" - "'peer reviewed papers on climate change 2020' → climate change\n" - "'books on structural engineering UAE' → structural engineering UAE\n" - "'give me a summary of quantum physics advances' → quantum physics advances\n" - "'i need peer reviewed research on machine learning healthcare' → machine learning healthcare" - )}, - {"role": "user", "content": question}, - ]) - topic = response.content.strip().strip('"').strip("'").strip(".") - topic = _strip_resource_noise(topic) - topic = _normalize_whitespace(topic).strip(".?") - if topic and len(topic) > 2: - return topic - except Exception as e: - logger.warning(f"Topic extraction failed: {e}") - return _extract_topic_regex(question) - - -# ── PRIMO / PubMed URL + Query Builders ─────────────────────────────────────── -# FIX: these were previously imported from search.py as -# build_primo_boolean_query / build_primo_discovery_url / build_pubmed_url -# but those exports do not exist in search.py. -# They are now self-contained here. - -def _shared_build_primo_boolean_query(topic: str) -> str: - """ - Regex fallback for boolean query construction — used only when LLM is unavailable. - Keeps multi-word phrases quoted, strips stop words, ANDs remaining concepts. - Does NOT split known compound terms. - """ - clean = _strip_resource_noise(topic) - if not clean: - clean = topic - clean = _normalize_whitespace(clean) - - # Treat the whole cleaned topic as a single phrase if ≤ 4 words - words = clean.split() - if len(words) <= 4: - return f'"{clean}"' - - # For longer topics: split on stop words, quote resulting multi-word chunks - _BOOL_STOP = re.compile( - r"\b(of|on|in|the|a|an|and|or|for|to|with|by|from|at|is|are|was|were|" - r"be|been|have|has|had|do|does|did|will|would|could|should|may|its|" - r"this|that|these|those|about|impact|role|effect|use|analysis|review|" - r"what|how|why|when|where|which|between|within|across|among|using|" - r"based|related|towards|toward|during|after|before|over|under)\b", - re.IGNORECASE, - ) - # Split on stop words, keep non-stop chunks - parts = _BOOL_STOP.split(clean) - concepts = [_normalize_whitespace(p) for p in parts if _normalize_whitespace(p) and len(_normalize_whitespace(p)) > 2] - if not concepts: - return f'"{clean}"' - if len(concepts) == 1: - return f'"{concepts[0]}"' - # Quote multi-word concepts, leave single words bare - quoted = [f'"{c}"' if ' ' in c else c for c in concepts[:4]] - return " AND ".join(quoted) - - -_BOOLEAN_SYSTEM = """You are an expert academic database search specialist building PRIMO/ExLibris boolean queries. - -Given a research topic or question, output ONLY a valid boolean search string. - -RULES: -1. Identify 2-4 key CONCEPTS in the query. -2. Multi-word concepts MUST be quoted: "machine learning" not machine AND learning. -3. Use OR between synonyms of the same concept: ("machine learning" OR "deep learning" OR "neural networks") -4. Use AND to connect different concepts: ("climate change" OR "global warming") AND ("renewable energy" OR "clean energy") -5. NEVER split a natural phrase with AND: "global warming" stays "global warming", NOT global AND warming. -6. NEVER include stop words (of, the, a, in, on, for, to, with, by, impact, role, effect, use, study, analysis, review) as standalone terms. -7. For medical/clinical topics, include MeSH-style synonyms. -8. Keep to a maximum of 3 concept groups joined by AND. -9. Return ONLY the boolean string. No explanation, no preamble, no markdown. - -EXAMPLES: -Input: global warming -Output: ("global warming" OR "climate change" OR "greenhouse effect") - -Input: machine learning in healthcare -Output: ("machine learning" OR "deep learning" OR "artificial intelligence") AND ("healthcare" OR "medicine" OR "clinical") - -Input: impact of social media on mental health -Output: ("social media" OR "social networks") AND ("mental health" OR "psychological wellbeing" OR "depression") - -Input: structural engineering UAE -Output: ("structural engineering" OR "structural analysis") AND ("UAE" OR "United Arab Emirates" OR "Gulf") - -Input: CRISPR gene editing applications -Output: ("CRISPR" OR "CRISPR-Cas9" OR "gene editing") AND ("applications" OR "therapeutic" OR "treatment") - -Input: renewable energy policy -Output: ("renewable energy" OR "clean energy" OR "sustainable energy") AND ("policy" OR "regulation" OR "governance") -""" - - -async def _llm_build_boolean_query(topic: str, model: str) -> str: - """ - Use LLM to generate a high-quality PRIMO boolean query with: - - Quoted multi-word phrases - - OR synonyms within concept groups - - AND between distinct concepts - Falls back to regex version if LLM fails. - """ - settings = get_settings() - if not settings.openai_api_key and not settings.anthropic_api_key: - return _shared_build_primo_boolean_query(topic) - try: - llm = _get_llm(model, temperature=0, max_tokens=120) - response = await llm.ainvoke([ - {"role": "system", "content": _BOOLEAN_SYSTEM}, - {"role": "user", "content": f"Input: {topic.strip()}"}, - ]) - result = response.content.strip() - # Sanity checks — must contain quotes and AND/OR or be a short phrase - if not result or len(result) > 500: - raise ValueError("Bad LLM boolean output") - # Strip any markdown the LLM snuck in - result = re.sub(r"^```[a-z]*\n?", "", result).rstrip("`").strip() - # Must have at least one quote or look like a valid query - if '"' not in result and ' AND ' not in result and ' OR ' not in result: - # Wrap bare output as a phrase - result = f'"{result}"' - logger.info(f"LLM boolean: {result!r} ← topic: {topic!r}") - return result - except Exception as e: - logger.warning(f"LLM boolean query failed: {e} — using regex fallback") - return _shared_build_primo_boolean_query(topic) - - -def _make_primo_boolean_query(context: SearchContextPayload) -> str: - """Sync wrapper — returns regex boolean. Async LLM version called from _prepare_queries.""" - topic = _title_case_topic(context.display_topic or context.topic) - topic = _strip_resource_noise(topic) or (context.topic or "library search") - return _shared_build_primo_boolean_query(topic) - - -def _shared_build_primo_discovery_url( - boolean_query: str, - resource_type: str = "articles", - peer_reviewed: bool = False, - open_access: bool = False, - year_from: Optional[str] = None, - year_to: Optional[str] = None, -) -> str: - """Build a full PRIMO discovery URL with facets.""" - base = ( - "https://khalifa.primo.exlibrisgroup.com/discovery/search" - f"?vid=971KUOSTAR_INST:KU&tab=Everything&scope=MyInst_and_CI" - f"&query=any,contains,{quote(boolean_query)}" - f"&lang=en&search_scope=MyInst_and_CI&sortby=rank&mode=advanced" - ) - facets = [] - if resource_type == "articles": - facets.append("facet_rtype,include,articles") - elif resource_type == "books": - facets.append("facet_rtype,include,books") - if peer_reviewed: - facets.append("facet_tlevel,include,peer_reviewed") - if open_access: - facets.append("facet_tlevel,include,online_resources") - if year_from or year_to: - yf = year_from or "0001" - yt = year_to or "9999" - facets.append(f"facet_searchcreationdate,include,{yf}|,|{yt}") - for facet in facets: - base += f"&multiFacets={quote(facet)}" - return base - - -def _shared_build_pubmed_url( - topic: str, - year_from: Optional[str] = None, - year_to: Optional[str] = None, - peer_reviewed: bool = False, -) -> str: - """Build a PubMed search URL.""" - clean = _strip_resource_noise(topic) - term = clean or topic - if peer_reviewed: - term = f"({term}) AND Journal Article[pt]" - url = f"https://pubmed.ncbi.nlm.nih.gov/?term={quote(term)}" - if year_from or year_to: - yf = year_from or "1900" - yt = year_to or str(CURRENT_YEAR) - url += f"&filter=datesearch.y_{yf}-{yt}" - return url - - -# ── Query Building ──────────────────────────────────────────────────────────── - -def _derive_resource_type(question: str, existing: Optional[str] = None) -> str: - q = (question or "").lower() - if any(token in q for token in ["book", "books", "ebook", "ebooks"]): - return "books" - if any(token in q for token in ["article", "articles", "paper", "papers", "study", "studies", "journal", "journals"]): - return "articles" - return existing or "articles" - - -def _parse_year_filters(question: str) -> Tuple[Optional[str], Optional[str]]: - q = (question or "").lower() - match = re.search(r"\b(last|past)\s+(\d{1,2})\s+years?\b", q) - if match: - years = int(match.group(2)) - return str(CURRENT_YEAR - years + 1), str(CURRENT_YEAR) - between = re.search(r"\b(?:from|between)\s+(20\d{2}|19\d{2})\s+(?:to|and|-)\s+(20\d{2}|19\d{2})\b", q) - if between: - return between.group(1), between.group(2) - years = re.findall(r"\b(19\d{2}|20\d{2})\b", q) - if len(years) >= 2: - return min(years), max(years) - if len(years) == 1 and any(word in q for word in ["since", "from", "after"]): - return years[0], str(CURRENT_YEAR) - return None, None - - -def _make_primo_boolean_query(context: SearchContextPayload) -> str: - topic = _title_case_topic(context.display_topic or context.topic) - topic = _strip_resource_noise(topic) or (context.topic or "library search") - return _shared_build_primo_boolean_query(topic) - - -def _primo_clean_url(context: SearchContextPayload) -> str: - boolean_query = context.primo_boolean_query or _make_primo_boolean_query(context) - return _shared_build_primo_discovery_url( - boolean_query, - resource_type=context.resource_type, - peer_reviewed=context.peer_reviewed, - open_access=context.open_access, - year_from=context.year_from, - year_to=context.year_to, - ) - - -def _compose_ai_tool_query(question: str, context: SearchContextPayload, is_follow_up: bool) -> str: - topic = context.display_topic or context.topic - topic = _title_case_topic(topic) - scope = context.resource_type - filters = [] - if context.peer_reviewed: - filters.append("peer-reviewed") - if context.open_access: - filters.append("open-access") - if context.year_from and context.year_to: - filters.append(f"published between {context.year_from} and {context.year_to}") - elif context.year_from: - filters.append(f"published from {context.year_from} onward") - - if is_follow_up: - if scope == "books": - base = f"Find books on {topic}" - elif scope == "both": - base = f"Find articles and books on {topic}" - else: - base = f"Find research articles on {topic}" + await rag_service.initialize(openai_api_key=settings.openai_api_key) + logger.info("RAG service ready — chunks: %d", len(rag_service.bm25_corpus)) + except Exception as exc: + logger.error("RAG initialization failed: %s", exc, exc_info=True) else: - base = _normalize_whitespace(question).strip(".?") - if len(base.split()) < 4: - base = f"Find research on {topic}" + logger.warning("OPENAI_API_KEY not set — RAG disabled") - if filters: - return f"{base} with {', '.join(filters)}." - return base.rstrip(".") + "." + # Start background metrics flush + flush_task = asyncio.create_task(_metrics_flush_loop(interval=30)) + logger.info("LibBee startup complete") + yield -async def _grammar_refine_query(text: str, model: str) -> str: - settings = get_settings() - if not settings.openai_api_key and not settings.anthropic_api_key: - return _normalize_whitespace(text) + # Graceful shutdown — flush remaining metrics + flush_task.cancel() try: - llm = _get_llm(model, temperature=0, max_tokens=60) - response = await llm.ainvoke([ - {"role": "system", "content": "Rewrite the user's search question in clear grammatical English. Keep the meaning exactly the same. Return one sentence only."}, - {"role": "user", "content": text}, - ]) - refined = _normalize_whitespace(response.content) - return refined or _normalize_whitespace(text) + await asyncio.wait_for(metrics_service.flush(), timeout=5.0) except Exception: - return _normalize_whitespace(text) - - -def _light_strip_retrieval_boilerplate(text: str) -> str: - """Strip common retrieval command prefixes before building a search plan.""" - cleaned = re.sub(r"^\s*(please\s+)?(?:can you|could you|would you)\s+", "", (text or "").strip(), flags=re.IGNORECASE) - cleaned = re.sub(r"^\s*(please\s+)?help me\s+", "", cleaned, flags=re.IGNORECASE) - cleaned = re.sub(r"^\s*please\s+", "", cleaned, flags=re.IGNORECASE) - cleaned = re.sub(r"\s+(please|thanks|thank you|asap)$", "", cleaned, flags=re.IGNORECASE) - return re.sub(r"\s+", " ", cleaned).strip() - - -def _sanitize_boolean_for_primo(boolean: str) -> str: - """Clean LLM-generated boolean so it works in PRIMO advanced search URLs.""" - if not boolean: - return boolean - # Single quotes → double quotes around terms - boolean = re.sub(r"'([^']+)'", r'"\1"', boolean) - # Remove pure-year groups like (2025 OR 2026) — years belong in date facets - boolean = re.sub( - r"\(\s*(?:(?:19|20)\d{2}\s*(?:OR\s*(?:19|20)\d{2}\s*)*)\)", - "", boolean, flags=re.IGNORECASE, - ) - # Remove dangling AND/OR - boolean = re.sub(r"^\s*(AND|OR)\s*", "", boolean, flags=re.IGNORECASE) - boolean = re.sub(r"\s*(AND|OR)\s*$", "", boolean, flags=re.IGNORECASE) - # Collapse double AND/OR - boolean = re.sub(r"\b(AND|OR)\s+(AND|OR)\b", r"\1", boolean, flags=re.IGNORECASE) - return re.sub(r"\s+", " ", boolean).strip() - - -def _clean_database_keywords(boolean_query: str) -> str: - """Strip boolean operators/punctuation to get plain keyword string.""" - return re.sub(r"\s+", " ", - re.sub(r"\b(AND|OR|NOT)\b|[()\"]", " ", - boolean_query, flags=re.IGNORECASE)).strip() - - -def _context_to_dict(context: SearchContextPayload) -> dict: - return context.model_dump() - - -_BUILD_SEARCH_PLAN_SYSTEM = f"""You are a search expert for Khalifa University Library. - -The user typed a research request. Create THREE forms preserving the full intent. + pass + logger.info("LibBee shutting down") -1. corrected: lightly edit the FULL request for spelling, grammar, and clarity. - Preserve every substantive concept, constraint, and task. Do NOT collapse. -2. natural: write a natural-language research query for AI tools such as Consensus, - Scopus AI, PRIMO AI, and LeapSpace. Keep full context. Remove only retrieval - boilerplate like "find me", "show me", "can you find". - -3. boolean: write a PRIMO/PubMed Boolean query with AND/OR/parentheses. - -Also detect: year_from, year_to (e.g. "last 24 months" → year_from={{CURRENT_YEAR-2}}), -peer_reviewed, open_access. - -BOOLEAN RULES — follow strictly: -- Use DOUBLE quotes around multi-word phrases: ("smart city" OR "urban mobility") -- NEVER use single quotes -- Max 3 AND groups. Each group max 4 OR terms. -- NEVER put years as Boolean terms — put them in year_from/year_to -- Keep broad enough to return results - -EXAMPLES: -Input: "find articles on global warming and renewable energy" -→ boolean:("global warming" OR "climate change") AND ("renewable energy" OR "clean energy") -→ year_from:"", peer_reviewed:false - -Input: "peer reviewed articles last 5 years on machine learning healthcare" -→ boolean:("machine learning" OR "deep learning" OR "AI") AND ("healthcare" OR "medicine" OR "clinical") -→ year_from:"{CURRENT_YEAR - 5}", peer_reviewed:true - -Input: "summarize shift in personalized genomic medicine from N-of-1 trials last 24 months" -→ boolean:("personalized genomic medicine" OR "precision medicine") AND ("N-of-1 trials" OR "scalable precision") -→ year_from:"{CURRENT_YEAR - 2}", peer_reviewed:true +app = FastAPI( + title="LibBee - KU Library AI", + description="Khalifa University Library AI Assistant", + version="3.1.0", + lifespan=lifespan, +) -Return ONLY valid JSON: -{{"corrected":"...","natural":"...","boolean":"(\"A\" OR \"B\") AND (\"C\" OR \"D\")","year_from":"","year_to":"","peer_reviewed":false,"open_access":false}}""" +app.add_middleware( + CORSMiddleware, + allow_origins=[ + "https://nikeshn.github.io", + "http://localhost:8080", + "http://localhost:3000", + "http://127.0.0.1:5500", + ], + allow_credentials=True, + allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], + allow_headers=["*"], + max_age=86400, +) -async def _build_search_plan(question: str, model: str) -> dict: - """ - LLM-driven search plan — ported from app81.py and adapted for LibBee. - Returns: corrected, natural, boolean, database_query, year_from, year_to, - peer_reviewed, open_access. - Falls back to regex boolean if LLM fails. - """ +@app.middleware("http") +async def operational_guardrails(request: Request, call_next): settings = get_settings() - raw_query = _normalize_whitespace(question or "") - light_query = _light_strip_retrieval_boilerplate(raw_query) or raw_query - - if settings.openai_api_key or settings.anthropic_api_key: - try: - llm = _get_llm(model, temperature=0, max_tokens=300) - response = await llm.ainvoke([ - {"role": "system", "content": _BUILD_SEARCH_PLAN_SYSTEM}, - {"role": "user", "content": f'Query: "{light_query}"'}, - ]) - raw = response.content.strip() - if raw.startswith("```"): - raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip() - # Extract JSON object - s, e = raw.find("{"), raw.rfind("}") - if s != -1 and e > s: - result = json.loads(raw[s:e + 1]) - else: - raise ValueError("No JSON found in response") - - corrected = (result.get("corrected") or raw_query).strip() or raw_query - natural = (result.get("natural") or corrected).strip() or corrected - boolean = (result.get("boolean") or "").strip() - - # Validate boolean — must have operators and structure - has_ops = bool(re.search(r"\b(AND|OR)\b", boolean)) if boolean else False - has_parens = "(" in boolean if boolean else False - if not has_ops or not has_parens: - boolean = _shared_build_primo_boolean_query(corrected) - - boolean = _sanitize_boolean_for_primo(boolean) - - year_from = str(result.get("year_from") or "").strip() - year_to = str(result.get("year_to") or "").strip() - # If LLM didn't detect year filters, fall back to regex - if not year_from and not year_to: - yf, yt = _parse_year_filters(raw_query) - year_from = yf or "" - year_to = yt or "" - - return { - "corrected": corrected, - "natural": natural, - "boolean": boolean, - "database_query": _clean_database_keywords(boolean), - "year_from": year_from, - "year_to": year_to, - "peer_reviewed": bool(result.get("peer_reviewed", False)), - "open_access": bool(result.get("open_access", False)), - } - except Exception as e: - logger.warning(f"_build_search_plan LLM failed: {e} — using regex fallback") - - # Regex fallback - boolean = _shared_build_primo_boolean_query(light_query) - yf, yt = _parse_year_filters(raw_query) - pr = bool(re.search(r"\bpeer[- ]reviewed\b", raw_query, re.IGNORECASE)) - oa = bool(re.search(r"\bopen[- ]access\b", raw_query, re.IGNORECASE)) - return { - "corrected": raw_query, - "natural": light_query, - "boolean": boolean, - "database_query": _clean_database_keywords(boolean), - "year_from": yf or "", - "year_to": yt or "", - "peer_reviewed": pr, - "open_access": oa, - } - - -async def _prepare_queries(question: str, context: SearchContextPayload, model: str, is_follow_up: bool) -> SearchContextPayload: - """ - Build all query forms using _build_search_plan. - Updates context.ai_tool_query, primo_boolean_query, and if plan detects - year/peer_reviewed/open_access that aren't already set, applies them. - """ - topic = _strip_resource_noise(context.display_topic or context.topic) or context.topic - q = question if not is_follow_up else f"Find {context.resource_type} on {topic}" - - plan = await _build_search_plan(q, model) - - context.ai_tool_query = plan["natural"] - context.primo_boolean_query = plan["boolean"] - - # Apply plan's filter detections only if context doesn't already have them set - if not context.year_from and plan.get("year_from"): - context.year_from = plan["year_from"] - if not context.year_to and plan.get("year_to"): - context.year_to = plan["year_to"] - if not context.peer_reviewed and plan.get("peer_reviewed"): - context.peer_reviewed = True - if not context.open_access and plan.get("open_access"): - context.open_access = True - - return context - - -# ── Semantic Scholar + Summary ─────────────────────────────────────���────────── - -async def _semantic_scholar_search(topic: str, limit: int = 8) -> List[dict]: - clean_topic = _strip_resource_noise(topic) - if not clean_topic: - clean_topic = topic - try: - params = { - "query": clean_topic, - "limit": limit, - "fields": "title,authors,year,venue,abstract,externalIds,openAccessPdf", - } - async with httpx.AsyncClient(timeout=15) as client: - r = await client.get("https://api.semanticscholar.org/graph/v1/paper/search", params=params) - if r.status_code != 200: - logger.warning(f"Semantic Scholar {r.status_code} for '{clean_topic}'") - return [] - data = r.json() - results = [] - for paper in data.get("data", []): - abstract = paper.get("abstract") or "" - if not abstract or len(abstract) < 50: - continue - author_list = paper.get("authors") or [] - first_author_last = author_list[0].get("name", "Unknown").split()[-1] if author_list else "Unknown" - year = paper.get("year") or "n.d." - paper_id = paper.get("paperId", "") - doi = (paper.get("externalIds") or {}).get("DOI", "") - pdf_url = (paper.get("openAccessPdf") or {}).get("url", "") - link = ( - pdf_url - or (f"https://doi.org/{doi}" if doi else "") - or (f"https://www.semanticscholar.org/paper/{paper_id}" if paper_id else "#") + # Maintenance mode + if settings.maintenance_mode and request.url.path not in {"/", "/config"}: + return JSONResponse( + status_code=503, + content={"detail": "LibBee is in maintenance mode. Please try again shortly."}, + ) + + # Rate limiting + client_ip = request.client.host if request.client else "unknown" + now = time.time() + + async with _request_log_lock: + if _ttl_cache_available: + # TTLCache stores a list of timestamps per IP + timestamps = _request_log.get(client_ip, []) + timestamps = [t for t in timestamps if now - t < 60] + if len(timestamps) >= settings.rate_limit_per_minute: + metrics_service.incr_bucket("errors", "rate_limit") + return JSONResponse( + status_code=429, + content={"detail": "Too many requests. Please slow down and try again."}, ) - results.append({ - "title": paper.get("title", "Untitled"), - "authors": ( - ", ".join(a.get("name", "") for a in author_list[:3]) - + (" et al." if len(author_list) > 3 else "") - ), - "first_author": first_author_last, - "year": year, - "citation_key": f"{first_author_last}, {year}", - "venue": paper.get("venue", ""), - "abstract": abstract[:700], - "link": link, - "_source": "Semantic Scholar", - }) - return results - except Exception as e: - logger.error(f"Semantic Scholar search failed: {e}") - return [] - + timestamps.append(now) + _request_log[client_ip] = timestamps + else: + # Fallback: deque-based sliding window + from collections import deque + bucket = _request_log[client_ip] # type: ignore[index] + while bucket and now - bucket[0] > 60: + bucket.popleft() + if len(bucket) >= settings.rate_limit_per_minute: + metrics_service.incr_bucket("errors", "rate_limit") + return JSONResponse( + status_code=429, + content={"detail": "Too many requests. Please slow down and try again."}, + ) + bucket.append(now) -async def _generate_cited_summary(topic: str, papers: List[dict], model: str) -> str: - settings = get_settings() - if not settings.openai_api_key and not settings.anthropic_api_key: - return "" - if not papers: - return "" - context_text = "" - for i, paper in enumerate(papers, 1): - context_text += ( - f"[{i}] {paper['citation_key']}: {paper['title']}\n" - f"Abstract: {paper['abstract']}\n\n" - ) - system = ( - "You are a research librarian writing a concise academic literature snapshot. " - "Using ONLY the numbered references provided, write a 4-6 sentence summary on the given topic. " - "Use in-text citations as [1], [2], [3] etc. referring to the numbered references. " - "Cite at least 4 different references. " - "Use HTML

for paragraph breaks. Do not use bullet points or markdown. " - "Only state what is explicitly supported by the abstracts. " - "Do NOT invent content." - ) try: - llm = _get_llm(model, temperature=0.2, max_tokens=560) - response = await llm.ainvoke([ - {"role": "system", "content": system}, - {"role": "user", "content": f"Topic: {topic}\n\nReferences:\n{context_text}\nWrite the summary."}, - ]) - summary = response.content.strip() - for i in range(len(papers), 0, -1): - paper = papers[i - 1] - title_escaped = _escape(paper["title"]) - cite_html = ( - f'' - f'[{paper["citation_key"]}]' - ) - summary = summary.replace(f"[{i}]", cite_html) - return summary - except Exception as e: - logger.error(f"Summary generation failed: {e}") - return "" + response = await call_next(request) + return response + except Exception as exc: + logger.exception("Unhandled application error: %s", exc) + metrics_service.incr_bucket("errors", "unhandled_exception") + return JSONResponse(status_code=500, content={"detail": "Unexpected server error."}) -def _extractive_snapshot_from_papers(topic: str, papers: List[dict]) -> str: - if not papers: - return "" - statements: List[str] = [] - for paper in papers[:4]: - sentences = re.split(r"(?<=[.!?])\s+", paper.get("abstract", "")) - lead = sentences[0].strip() if sentences else "" - if not lead: - continue - citation = ( - f'' - f'[{_escape(paper["citation_key"])}]' - ) - statements.append(f"{_escape(lead)} {citation}") - if not statements: - return f"Recent literature on {_escape(topic)} is available in the cited papers below." - intro = f"Recent literature on {_escape(topic)}:

" - return intro + "

".join(statements[:4]) +from src.api import admin, agent, feedback, search # noqa: E402 +app.include_router(agent.router, prefix="/agent", tags=["Agent"]) +app.include_router(search.router, prefix="/search", tags=["Search"]) +app.include_router(admin.router, prefix="/admin", tags=["Admin"]) +app.include_router(feedback.router, prefix="/feedback", tags=["Feedback"]) -# ── PRIMO / PubMed Search Wrappers ──────────────────────────────────────────── -async def _search_primo_catalog(context: SearchContextPayload, limit: int = 5) -> Tuple[dict, str]: - from src.api.search import SearchRequest, search_primo +@app.get("/") +def health_check(): settings = get_settings() - clean_topic = _strip_resource_noise(context.topic) - if not clean_topic: - clean_topic = context.topic - result = await search_primo( - SearchRequest( - query=clean_topic, - source="primo", - limit=limit, - peer_reviewed=context.peer_reviewed, - open_access=context.open_access, - year_from=context.year_from, - year_to=context.year_to, - resource_type=context.resource_type, - ), - settings.primo_api_key, - ) - source_url = _primo_clean_url(context) - return result, source_url - - -async def _search_pubmed_articles(context: SearchContextPayload, limit: int = 5) -> Tuple[dict, str]: - from src.api.search import SearchRequest, search_pubmed - clean_topic = _strip_resource_noise(context.topic) - if not clean_topic: - clean_topic = context.topic - result = await search_pubmed( - SearchRequest( - query=clean_topic, - source="pubmed", - limit=limit, - peer_reviewed=context.peer_reviewed, - year_from=context.year_from, - year_to=context.year_to, - ) - ) - pubmed_url = _shared_build_pubmed_url( - clean_topic, context.year_from, context.year_to, context.peer_reviewed - ) - return result, pubmed_url - - -# ── AI Tools / Database Guidance ───────────────────────────────────────────── - -def _tool_urls(context: SearchContextPayload) -> Dict[str, str]: - ai_query = quote(context.ai_tool_query or context.display_topic or context.topic) - primo_ai_query = quote(context.ai_tool_query or context.display_topic or context.topic) - urls: Dict[str, str] = { - "leapspace": f"https://www.sciencedirect.com/leapspace?query={ai_query}", - "scopus_ai": f"https://www-scopus-com.khalifa.idm.oclc.org/pages/ai?query={ai_query}", - "ebsco_ai": ( - f"https://research-ebsco-com.khalifa.idm.oclc.org/c/yst6t2/search/results" - f"?q={ai_query}" - f"&autocorrect=y" - f"&expanders=concept" - f"&limiters=None" - f"&searchMode=enhanced" - f"&searchSegment=all-results" - ), - "primo_ai": f"{PRIMO_AI_URL}{primo_ai_query}", - "consensus": f"https://consensus.app/results/?q={ai_query}", - "primo_discovery": _primo_clean_url(context), - } - if context.intent == "search_medical": - urls["pubmed"] = _shared_build_pubmed_url( - context.topic, context.year_from, context.year_to, context.peer_reviewed - ) - return urls - - -def _database_guidance_for_topic(context: SearchContextPayload) -> str: - topic = _escape(context.display_topic or context.topic) - urls = _tool_urls(context) - ai_tools = ( - f'LeapSpace · ' - f'Scopus AI · ' - f'EBSCO AI · ' - f'PRIMO AI Assistant · ' - f'Consensus' - ) - more_results = f'PRIMO Library Discovery' - if context.intent == "search_medical" and urls.get("pubmed"): - more_results += f' · PubMed' - return ( - f"For deeper work on {topic}, continue with these AI research tools: {ai_tools}. " - f"For more results, use {more_results}. Medical topics are strongest in PubMed, Embase, and CINAHL." - ) - return ( - f"For deeper work on {topic}, continue with these AI research tools: {ai_tools}. " - f"For more results, use {more_results}." - ) - - -def _find_staff_by_token(token: str) -> Optional[dict]: - token = (token or "").lower() - for staff in STAFF_DIRECTORY: - hay = (staff.get("full_name", "") + " " + staff.get("role", "")).lower() - if token in hay: - return staff - return None - - -def _contact_footer_for_query(question: str, context: Optional[SearchContextPayload]) -> str: - q = (question or "").lower() - # FIX: removed broken default to _find_staff_by_token("muna") — Muna is not in STAFF_DIRECTORY - specialist: Optional[dict] = None - if any(t in q for t in ["database access", "proxy", "full text", "eresource", "e-resource", "remote access", "access problem"]): - specialist = _find_staff_by_token("rani") - elif any(t in q for t in ["website", "system", "technical", "login issue", "page not loading"]): - specialist = _find_staff_by_token("walter") - elif context and context.intent == "search_medical": - specialist = _find_staff_by_token("jason") - elif context and context.intent == "search_academic": - specialist = _find_staff_by_token("nikesh") - - footer = ( - "

Library contact points
" - f'General help: Ask a Librarian · ' - 'libse@ku.ac.ae' - ) - if specialist: - footer += ( - f'
Relevant contact: {_escape(specialist.get("full_name", ""))} — ' - f'{_escape(specialist.get("role", ""))}
' - f'{_escape(specialist.get("email", ""))}' - ) - phone = specialist.get("phone") or specialist.get("work_phone") - if phone: - footer += f' · {_escape(phone)}' - return footer - - -def _thought_block(intent: str, tool: str, question: str, extra: str = "") -> str: - """ - Lightweight collapsible reasoning block shown at the top of every non-search response. - """ - labels = { - "library_info": ("📋", "Library information lookup", "I searched the KU Library knowledge base for relevant content."), - "general": ("🌐", "General knowledge / web search", "I answered using a web search for current information."), - "general_recent": ("🔴", "Live web search", "I ran a live web search for the most up-to-date information."), - "social_greeting": ("🐝", "Conversational response", "I responded directly as LibBee, the KU Library AI Assistant."), - "social": ("🐝", "Conversational response", "I responded directly as LibBee, the KU Library AI Assistant."), - "staff_lookup": ("👤", "Staff directory lookup", "I matched your request to the KU Library staff directory."), - "hours": ("⏰", "Library hours lookup", "I retrieved the current library hours from the KU Library knowledge base."), - "campus": ("📍", "Campus information lookup", "I retrieved KU Library campus information from the knowledge base."), - "database_rec": ("🗄️", "Database recommendation", "I matched your subject to the KU Library e-resources database."), - "rag_search": ("📖", "Knowledge base search", "I ran a hybrid semantic + keyword search across the KU Library knowledge base."), + return { + "status": "ok", + "version": "3.1.0", + "service": "LibBee KU Library AI", + "rag_ready": rag_service.is_ready(), + "maintenance_mode": settings.maintenance_mode, + "endpoints": ["/agent", "/search", "/admin", "/feedback"], } - icon, label, reasoning = labels.get(intent, ("🤖", "AI response", "I processed your question and generated a response.")) - if extra: - reasoning += " " + extra - - return ( - f'
' - f'' - f'{icon} My approach — click to see' - f'
' - f'Question: {_escape(question)}
' - f'Classified as: {label}
' - f'How I answered: {reasoning}' - f'
' - ) - - -def _search_trace_block(question: str, context: SearchContextPayload) -> str: - """ - Richer collapsible reasoning block for search responses. - Shows topic, boolean query, and reasoning. Also hints that AI tools are at the end. - """ - topic = _escape(context.display_topic or context.topic) - primo_boolean = _escape(context.primo_boolean_query or "") - source_label = "PubMed" if context.source == "pubmed" else "KU PRIMO Library Discovery" - - intent_label = { - "search_academic": "academic literature search", - "search_medical": "medical / clinical literature search", - }.get(context.intent, "library search") - - filters_parts = [] - if context.peer_reviewed: - filters_parts.append("peer reviewed only") - if context.open_access: - filters_parts.append("open access") - if context.year_from and context.year_to: - filters_parts.append(f"{context.year_from}–{context.year_to}") - elif context.year_from: - filters_parts.append(f"from {context.year_from}") - rtype = {"articles": "articles", "books": "books", "both": "articles and books"}.get( - context.resource_type, "articles" - ) - filters_parts.append(rtype) - filters_str = ", ".join(filters_parts) - - reasoning = ( - f"I classified this as a {intent_label}, " - f"extracted the topic {topic}, " - f"built an LLM-optimised boolean query with quoted phrases and OR synonyms, " - f"then searched {source_label} for {filters_str}." - ) - if context.intent == "search_medical" and context.source != "pubmed": - reasoning += " For medical topics, PubMed and Embase are also recommended." - - block = ( - f'
' - f'' - f'🔍 How I searched — click to expand' - f'' - f'
' - f'Your question: {_escape(question)}
' - f'Interpreted topic: {topic}
' - ) - if primo_boolean: - block += ( - f'Boolean query sent to {source_label}:
' - f'{primo_boolean}' - ) - block += f'Reasoning: {reasoning}
' - block += '✨ AI research tools for deeper exploration are listed at the end of this answer.' - block += '
' - return block - - -def _ai_tools_footer(context: SearchContextPayload) -> str: - """ - AI tools recommendation panel — always rendered at the very end of the answer. - Structured as KU-subscribed tools first, then freemium, then more results. - """ - urls = _tool_urls(context) - topic_enc = _escape(context.ai_tool_query or context.display_topic or context.topic) - is_medical = context.intent == "search_medical" - - ku_tools = [ - ("🔬 LeapSpace", urls["leapspace"], "leapspace", "#9a3412"), - ("🔭 Scopus AI", urls["scopus_ai"], "scopus", "#1e40af"), - ("📰 EBSCO AI", urls["ebsco_ai"], "ebscoai", "#9d174d"), - ("⭐ PRIMO AI Assistant", urls["primo_ai"], "primoai", "#5b21b6"), - ] - free_tools = [ - ("🧬 Consensus", urls["consensus"], "consensus", "#86198f"), - ("🌐 OpenAlex", f"https://openalex.org/works?page=1&sort=relevance_score:desc&search={topic_enc}", "openalex", "#0c4a6e"), - ("🔎 Perplexity", f"https://www.perplexity.ai/search?q={topic_enc}", "perplexity", "#4c1d95"), - ] - more = [("📚 PRIMO Library Discovery", urls["primo_discovery"], "primoai", "#003366")] - if is_medical and urls.get("pubmed"): - more.append(("🏥 PubMed", urls["pubmed"], "pubmed_badge", "#065f46")) - - def chip(label, url, cls, color): - return ( - f'{label}' - ) - - html = ( - '
' - '
' - '✨ For deeper research — AI tools
' - '
🔐 KU subscribed
' - '
' - ) - for label, url, cls, color in ku_tools: - html += chip(label, url, cls, color) - html += ( - '
🆓 Freemium tools
' - '
' - ) - for label, url, cls, color in free_tools: - html += chip(label, url, cls, color) - html += ( - '
📚 More results
' - '
' - ) - for label, url, cls, color in more: - html += chip(label, url, cls, color) - html += '
' - return html - - -def _is_database_recommendation_question(question: str) -> bool: - return bool(DATABASE_RECOMMEND_RE.search(question or "")) - - -def _safe_db_url(url: str) -> str: - """ - Encode the inner URL in IDM proxy links so the frontend fmt() auto-linkify - regex cannot match a bare https:// inside an href attribute and break the HTML. - e.g. href="...?url=https://foo.com" → href="...?url=https%3A%2F%2Ffoo.com" - """ - from urllib.parse import quote as _quote - if "?url=" in url: - base, inner = url.split("?url=", 1) - return base + "?url=" + _quote(inner, safe="") - return url - - -# ── Subject → database mapping sourced from KU KB ──────────────────────────── -# All IDM proxy URLs are encoded via _safe_db_url to prevent HTML breakage. -_DB_SUBJECTS: Dict[str, List[Tuple[str, str, str]]] = { - "physics": [ - ("APS Journals", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://journals.aps.org"), "American Physical Society — core physics journals."), - ("AIP Publishing", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://pubs.aip.org"), "Core physics and applied physics literature."), - ("IOPScience", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://iopscience.iop.org"), "Core physics journal platform."), - ("arXiv", "https://arxiv.org", "Open-access preprints in physics and maths."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text across applied and interdisciplinary physics."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Discovery and citation tracking."), - ], - "chemistry": [ - ("ACS Publications", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://pubs.acs.org"), "Core chemistry journals from the American Chemical Society."), - ("SciFinder-n", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://scifinder-n.cas.org"), "Substance, reaction, reference, and patent searching."), - ("Reaxys", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.reaxys.com"), "Reaction and property searching."), - ("RSC Journals", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://pubs.rsc.org"), "Royal Society of Chemistry journals."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text chemistry journals and books."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Discovery and citation tracking."), - ], - "chemical_engineering": [ - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Strong full text for chemical engineering journals and books."), - ("ACS Publications", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://pubs.acs.org"), "Core chemistry and chemical engineering coverage."), - ("SciFinder-n", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://scifinder-n.cas.org"), "Chemical substances, reactions, references, and patents."), - ("Reaxys", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.reaxys.com"), "Reaction and property searching."), - ("Knovel", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://app.knovel.com"), "Engineering handbooks and technical reference data."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery and citation tracking."), - ], - "computer_science": [ - ("ACM Digital Library", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://dl.acm.org"), "Core computer science literature."), - ("IEEE Xplore", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://ieeexplore.ieee.org"), "Strong for CS, AI, networks, and systems."), - ("arXiv", "https://arxiv.org", "Important preprint source for CS, AI, and ML."), - ("INSPEC", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.theiet.org/publishing/inspec"), "Indexing for computing and engineering topics."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text across CS and applied computing."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery and citation tracking."), - ], - "robotics": [ - ("IEEE Xplore", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://ieeexplore.ieee.org"), "Core source for robotics, AI, automation, and intelligent systems."), - ("ACM Digital Library", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://dl.acm.org"), "AI, ML, and computing research."), - ("arXiv", "https://arxiv.org", "Essential for current AI, robotics, and ML preprints."), - ("INSPEC", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.theiet.org/publishing/inspec"), "Indexing for robotics, computing, electronics, and control."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text for AI, robotics, and applied engineering."), - ], - "electrical_engineering": [ - ("IEEE Xplore", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://ieeexplore.ieee.org"), "Core electrical engineering database."), - ("INSPEC", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.theiet.org/publishing/inspec"), "Strong indexing for EE, electronics, and physics."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text for electrical engineering and control."), - ("ASTM Compass", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://compass.astm.org"), "Standards and testing methods."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery and citation tracking."), - ], - "mechanical_engineering": [ - ("ASME Digital Library", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://asmedigitalcollection.asme.org"), "Core mechanical engineering literature."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text across mechanics, thermal, and manufacturing topics."), - ("Knovel", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://app.knovel.com"), "Engineering reference data and handbooks."), - ("ASTM Compass", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://compass.astm.org"), "Standards and materials testing information."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery and citation tracking."), - ], - "civil_engineering": [ - ("ASCE Library", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://ascelibrary.org"), "Core civil and structural engineering journals and conference papers."), - ("ICE Virtual Library", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.icevirtuallibrary.com"), "Civil engineering literature and proceedings."), - ("ASTM Compass", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://compass.astm.org"), "Standards and testing methods for materials and engineering."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text across civil engineering areas."), - ("Knovel", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://app.knovel.com"), "Handbooks and design reference data."), - ], - "aerospace": [ - ("AIAA", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://arc.aiaa.org/"), "Core aerospace journals, conference papers, and standards."), - ("ASME Digital Library", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://asmedigitalcollection.asme.org"), "Thermal, mechanical, and fluid systems relevant to aerospace."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Strong full text for aerospace, mechanics, and materials."), - ("IEEE Xplore", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://ieeexplore.ieee.org"), "Avionics, control systems, and guidance."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery and citation tracking."), - ], - "petroleum": [ - ("OnePetro", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://onepetro.org"), "Core petroleum engineering and oil and gas literature."), - ("EarthDoc", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.earthdoc.org"), "Geoscience and petroleum-related literature."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text for petroleum, energy, and engineering topics."), - ("Knovel", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://app.knovel.com"), "Engineering handbooks and oil/gas reference material."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery and citation tracking."), - ], - "energy": [ - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Strong energy engineering and sustainability full text."), - ("OnePetro", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://onepetro.org"), "Energy sector and petroleum crossover topics."), - ("IEEE Xplore", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://ieeexplore.ieee.org"), "Power systems, electrical energy, and smart grids."), - ("Knovel", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://app.knovel.com"), "Engineering handbooks and energy reference content."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery and citation tracking."), - ], - "mathematics": [ - ("MathSciNet", "https://mathscinet.ams.org", "Core mathematics review and discovery database."), - ("SpringerLink", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://link.springer.com"), "Strong mathematics and statistics book and journal coverage."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text for mathematics, statistics, and data science."), - ("JSTOR", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.jstor.org"), "Foundational mathematics and statistics literature."), - ("arXiv", "https://arxiv.org", "Preprints in mathematics, statistics, and data science."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery and citation tracking."), - ], - "materials": [ - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Strong materials science full text."), - ("SpringerLink", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://link.springer.com"), "Strong materials and nanoscience coverage."), - ("AIP Publishing", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://pubs.aip.org"), "Applied materials and physics crossover."), - ("IOPScience", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://iopscience.iop.org"), "Materials and physics crossover."), - ("Knovel", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://app.knovel.com"), "Reference data and engineering materials handbooks."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery and citation tracking."), - ], - "nuclear": [ - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Strong nuclear engineering and energy full text."), - ("AIP Publishing", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://pubs.aip.org"), "Nuclear and applied physics crossover."), - ("Knovel", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://app.knovel.com"), "Engineering reference and handbooks."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Discovery and citation tracking."), - ("Web of Science", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.webofscience.com"), "Broad citation discovery."), - ], - "quantum_computing": [ - ("IEEE Xplore", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://ieeexplore.ieee.org"), "Quantum computing engineering implementations."), - ("ACM Digital Library", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://dl.acm.org"), "Algorithms, theory, and computing literature."), - ("APS Journals", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://journals.aps.org"), "Quantum information and foundational physics."), - ("arXiv", "https://arxiv.org", "Essential for current quantum computing preprints."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Discovery and citation tracking."), - ], - "biomedical": [ - ("PubMed", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://pubmed.ncbi.nlm.nih.gov"), "Core biomedical and life sciences literature."), - ("IEEE Xplore", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://ieeexplore.ieee.org"), "Biomedical engineering and medical devices."), - ("Embase", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.embase.com"), "Biomedical and pharmacological literature."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text for biomedical engineering and applied sciences."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Discovery and citation tracking."), - ], - "biology": [ - ("PubMed", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://pubmed.ncbi.nlm.nih.gov"), "Core life sciences and biomedical literature."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text biology and molecular sciences content."), - ("Embase", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.embase.com"), "Biomedical crossover topics."), - ("PubMed Central", "https://www.ncbi.nlm.nih.gov/pmc", "Open full-text biomedical articles — no login required."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Discovery and citation tracking."), - ], - "medical": [ - ("PubMed @ KU", "http://www.ncbi.nlm.nih.gov/pubmed/?otool=iaekustlib", "Core medical literature — use this KU link for full-text access."), - ("Embase", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.embase.com"), "Biomedical and pharmacological literature."), - ("CINAHL Ultimate", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://search.ebscohost.com/login.aspx?authtype=ip,uid&profile=ehost&defaultdb=cul"), "Nursing and allied health literature."), - ("Cochrane Library", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.cochranelibrary.com"), "Evidence synthesis and systematic reviews — gold standard."), - ("UpToDate", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.uptodate.com"), "Point-of-care clinical decision support."), - ("ClinicalKey Student", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.clinicalkey.com/student/login"), "Clinical reference and learning resource."), - ], - "business": [ - ("Business Source Complete", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://search.ebscohost.com"), "Core business and management literature."), - ("Emerald Insight", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.emerald.com/insight"), "Management, strategy, and leadership journals."), - ("ProQuest Databases", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://search.proquest.com"), "Broad business and interdisciplinary discovery."), - ("JSTOR", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.jstor.org"), "Business, economics, and policy literature."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery and citation tracking."), - ], - "earth_science": [ - ("GeoscienceWorld", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://pubs.geoscienceworld.org"), "Core earth science journals and conference content."), - ("EarthDoc", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.earthdoc.org"), "Geoscience and subsurface literature."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Full text for geoscience and planetary topics."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Discovery and citation tracking."), - ("Web of Science", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.webofscience.com"), "Broad citation discovery."), - ], - "environment": [ - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Strong sustainability and environmental full text."), - ("ProQuest Databases", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://search.proquest.com"), "Environmental policy and interdisciplinary discovery."), - ("Web of Science", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.webofscience.com"), "Broad citation discovery."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery and citation tracking."), - ("Business Source Complete", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://search.ebscohost.com"), "Sustainability, policy, and corporate responsibility."), - ], - "engineering_management": [ - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Broad discovery across engineering and management research."), - ("Emerald Insight", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.emerald.com/insight"), "Management, operations, and leadership literature."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Engineering and management full text."), - ("Business Source Complete", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://search.ebscohost.com"), "Business, management, and strategy literature."), - ("ProQuest Databases", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://search.proquest.com"), "Broad business and interdisciplinary discovery."), - ], - "metrics": [ - ("Journal Citation Reports", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://jcr.clarivate.com"), "Impact factors and journal rankings (Web of Science)."), - ("Scopus / CiteScore", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "CiteScore and source-level metrics."), - ("SciVal", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scival.com"), "Research performance analytics."), - ("Web of Science", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.webofscience.com"), "Citation discovery and journal evaluation context."), - ], - "theses": [ - ("ProQuest Dissertations & Theses Global", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.proquest.com/pqdtglobal"), "International dissertations and theses."), - ("Khazna", "https://khazna.ku.ac.ae/", "KU institutional repository and research outputs."), - ("PRIMO", "https://khalifa.primo.exlibrisgroup.com/discovery/search?vid=971KUOSTAR_INST:KU", "Discovery for dissertations, theses, and KU holdings."), - ], -} - -# General / fallback databases (used when no subject match found) -_DB_GENERAL: List[Tuple[str, str, str]] = [ - ("PRIMO (KU Library Discovery)", "https://khalifa.primo.exlibrisgroup.com/discovery/search?vid=971KUOSTAR_INST:KU", "Best starting point — search all KU books, articles, and journals."), - ("Scopus", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.scopus.com"), "Largest abstract and citation database — all disciplines."), - ("Web of Science", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.webofscience.com"), "Premier citation index — all disciplines."), - ("ScienceDirect", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.sciencedirect.com"), "Large full-text platform across science, engineering, and medicine."), - ("SpringerLink", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://link.springer.com"), "Springer journals and books across all disciplines."), - ("ProQuest Databases", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://search.proquest.com"), "Broad multidisciplinary discovery."), - ("Emerald Insight", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.emerald.com/insight"), "Social sciences, management, and interdisciplinary research."), - ("JSTOR", _safe_db_url("https://khalifa.idm.oclc.org/login?url=https://www.jstor.org"), "Foundational and archival literature across many fields."), -] - - -def _match_db_subject(q: str) -> Optional[str]: - """Map lowercased question to a _DB_SUBJECTS key using keyword matching.""" - checks = [ - (["quantum computing", "quantum algorithm", "quantum cryptography", "quantum information"], "quantum_computing"), - (["robotics", "autonomous system", "intelligent system"], "robotics"), - (["artificial intelligence", "machine learning", "deep learning", "nlp", "natural language processing", "neural network", "cybersecurity", "computer science", "software engineering", "algorithm", "data structure"], "computer_science"), - (["computer engineering", "embedded system", "computer architecture", "networking", "digital system"], "computer_science"), - (["aerospace", "aerodynamic", "propulsion", "aircraft", "aviation", "flight system"], "aerospace"), - (["biomedical engineering", "medical device", "biomechanic", "bioinstrument", "tissue engineering", "biomedical imaging"], "biomedical"), - (["chemical engineering", "process engineering", "reaction engineering", "separation", "catalysis"], "chemical_engineering"), - (["chemistry", "organic chemistry", "analytical chemistry", "inorganic", "chemical", "chemical properties"], "chemistry"), - (["civil engineering", "structural engineering", "geotechnical", "transportation engineering", "water resources", "construction"], "civil_engineering"), - (["electrical engineering", "circuit", "power system", "signal processing", "control system", "communication system", "electronics"], "electrical_engineering"), - (["energy engineering", "renewable energy", "sustainable energy", "clean energy", "energy transition", "solar", "wind power"], "energy"), - (["engineering management", "engineering systems", "systems engineering", "operations management", "innovation management"], "engineering_management"), - (["mechanical engineering", "thermofluid", "thermodynamics", "fluid mechanics", "manufacturing", "mechatronics"], "mechanical_engineering"), - (["nuclear engineering", "reactor", "radiation", "nuclear material", "nuclear safety"], "nuclear"), - (["petroleum engineering", "oil and gas", "drilling", "reservoir", "subsurface", "petroleum"], "petroleum"), - (["earth science", "geology", "geophysics", "geochemistry", "planetary science", "geoscience"], "earth_science"), - (["environment", "sustainability", "climate change", "sustainable development", "conservation", "environmental policy", "ecology"], "environment"), - (["materials science", "nanotechnology", "nanomaterial", "biomaterial", "advanced material", "functional material"], "materials"), - (["cell biology", "molecular biology", "genetics", "genomics", "biotechnology", "bioinformatics", "microbiology", "virology"], "biology"), - (["medicine", "medical", "health science", "clinical", "nursing", "pharmacy", "biomedical literature", "pharmacology", "clinical trial", "systematic review", "evidence-based", "patient care"], "medical"), - (["business", "management", "finance", "economics", "accounting", "entrepreneurship", "marketing", "strategy", "leadership", "international relations", "political science", "social science", "humanities", "law", "education", "psychology", "sociology"], "business"), - (["physics", "optics", "photonics", "quantum physics", "condensed matter", "theoretical physics", "astrophysics"], "physics"), - (["mathematics", "statistics", "data science", "calculus", "algebra", "optimization", "probability", "mathematical model"], "mathematics"), - (["impact factor", "journal metrics", "citescore", "jcr", "quartile", "research impact", "scival", "bibliometric"], "metrics"), - (["dissertation", "thesis", "doctoral", "master thesis", "repository", "phd thesis"], "theses"), - ] - for keywords, key in checks: - if any(kw in q for kw in keywords): - return key - return None - - -def _extract_db_topic(question: str) -> str: - """ - Extract the subject from a database-recommendation question. - e.g. "best databases for physics" → "physics" - "which database should I use for machine learning" → "machine learning" - """ - q = _normalize_whitespace(question or "") - # Strip leading boilerplate phrases - patterns = [ - r"^(best|good|recommended?)\s+databases?\s+for\s+", - r"^which\s+databases?\s+(for|should i use for|is best for|are best for)\s+", - r"^databases?\s+for\s+", - r"^where should i (search|start)\s+(for\s+)?", - r"^what (database|databases)\s+(for|should i use for)\s+", - ] - lower = q.lower() - for pat in patterns: - m = re.match(pat, lower) - if m: - q = q[m.end():] - break - # Strip trailing noise - q = re.sub(r"\s+(please|thanks|thank you)\.?$", "", q, flags=re.IGNORECASE) - q = _strip_resource_noise(q) - return _normalize_whitespace(q).strip(".?") or "this subject" - -def _database_recommendation_answer(question: str) -> str: - q = (question or "").lower() - subject = _extract_db_topic(question) - topic = _escape(subject) - # Match to a specific subject list; fall back to general KU databases - subject_key = _match_db_subject(q) - dbs = _DB_SUBJECTS.get(subject_key, _DB_GENERAL) if subject_key else _DB_GENERAL - - # Cap at 5 results for readability - dbs = list(dbs)[:5] - - answer = f"Recommended databases for {topic}

" - answer += "Here are the strongest KU-subscribed starting points:
" - for name, url, why in dbs: - answer += f'
• {_escape(name)} — {_escape(why)}' - - if not subject_key: - # Generic topic — add a note that subject-specific guides exist - answer += ( - "

💡 For subject-specific databases, ask me: " - "\"best databases for [your subject]\" — e.g. physics, chemistry, medicine, civil engineering." - ) - - answer += ( - f'

Browse all 50+ KU databases: ' - f'library.ku.ac.ae/eresources' - ) - answer += "

Want me to run a direct search on any of these for a specific topic?" - return answer - - -# ── Full-text access chain ──────────────────────────────────────────────────── - -def _fulltext_chain_answer(question: str) -> str: - """Step-by-step guide for getting full text when PDF is unavailable.""" - return ( - "📄 How to get full text for an article

" - "Follow this chain in order — each step is faster than the next:

" - "Step 1 — Search KU PRIMO
" - "Find the article in KU Library Discovery. " - "If KU subscribes, you'll see a Full Text Available or Online Access link. " - "Click it — it passes through the IDM proxy automatically when on campus or logged in.

" - "Step 2 — Try LibKey Nomad (browser extension)
" - "LibKey Nomad is a free browser " - "extension that detects articles you're viewing and fetches KU's full text automatically. " - "Install it once and it works on every publisher site.

" - "Step 3 — Try the publisher or open access
" - "Visit the DOI link directly. Many authors post their accepted manuscript on " - "arXiv, " - "ResearchGate, or their institutional page. " - "OpenAlex also links to many legal open-access PDFs.

" - "Step 4 — Submit an Interlibrary Loan (ILL) request
" - "If none of the above work, KU Library can borrow the article from another library — usually within 1–5 working days. " - "Submit here: library.ku.ac.ae/ill/
" - "Contact: Suaad Al Jneibi · " - "suaad.aljneibi@ku.ac.ae · +971 2 312 4278" - ) - - -# ── Search strategy / Boolean explanation ───────────────────────────────────── - -async def _search_strategy_answer(question: str, model: str) -> Tuple[str, dict]: - """ - Build a search plan AND explain it step by step. - Returns (explanation_html, plan_dict). - """ - plan = await _build_search_plan(question, model) - topic = _escape(_light_strip_retrieval_boilerplate(question) or question) - boolean = _escape(plan.get("boolean", "")) - natural = _escape(plan.get("natural", "")) - - explanation = ( - f"🔍 Search strategy for: {topic}

" - f"1. Identify your key concepts
" - f"Break the topic into 2–3 core concepts. For your question, I identified these concepts " - f"and suggested synonyms for each using OR, then joined concepts with AND:

" - f'{boolean}
' - f"Why AND and OR? AND narrows — both concepts must appear. " - f"OR broadens — any synonym counts. Quoting phrases like " - f"\"machine learning\" keeps them together as an exact phrase.

" - f"2. Apply filters
" - f"Use database facets to limit by: peer-reviewed, date range, document type (article / review / book), " - f"language, or open access. In PRIMO, these appear in the left sidebar after you search.

" - f"3. Natural language query for AI tools
" - f"For AI-powered tools (LeapSpace, Scopus AI, Consensus), use a conversational query:
" - f'{natural}

' - f"4. Iterate
" - f"Too many results → add more AND terms or apply filters. " - f"Too few → remove an AND group or use broader synonyms. " - f"You can ask me to refine: \"narrow to peer-reviewed only\", " - f"\"limit to last 5 years\", or \"suggest alternative keywords\"." - ) - return explanation, plan - - -# ── Alternative search terms ────────────────────────────────────────────────── - -async def _alternative_terms_answer(question: str, context: Optional[SearchContextPayload], model: str) -> str: - """Generate alternative keywords/synonyms for the current topic without re-running the search.""" - topic = (context.display_topic or context.topic) if context else question +@app.get("/config") +def public_config(): settings = get_settings() - if not settings.openai_api_key and not settings.anthropic_api_key: - return ( - f"Alternative search terms for: {_escape(topic)}

" - "Try combining these broader, narrower, and related terms:
" - "• Use broader terms if too few results
" - "• Use narrower/specific terms if too many results
" - "• Try acronyms and full forms (e.g. AI / Artificial Intelligence)
" - "• Include British and American spellings (e.g. organisation/organization)" - ) - try: - llm = _get_llm(model, temperature=0.4, max_tokens=300) - response = await llm.ainvoke([ - {"role": "system", "content": ( - "You are an academic librarian. Given a research topic, generate a structured list of alternative " - "search terms. Group them as: Broader terms, Narrower/specific terms, Related concepts, Acronyms/abbreviations. " - "Format as HTML using for group labels and
for line breaks. " - "Keep it concise — 3-4 terms per group maximum. No bullet points, use • instead." - )}, - {"role": "user", "content": f"Research topic: {topic}"}, - ]) - terms = response.content.strip() - return ( - f"💡 Alternative search terms for: {_escape(topic)}

" - f"{terms}

" - f"Tip: In PRIMO or databases, use OR between synonyms within a concept group, " - f"and AND between different concept groups. Ask me to run a new search with any of these." - ) - except Exception as e: - logger.warning(f"_alternative_terms_answer failed: {e}") - return f"Alternative terms for {_escape(topic)}

Try synonyms, acronyms, broader/narrower terms, and related concepts in your search." - - -# ── Citation chaining guide ─────────────────────────────────────────────────── - -def _citation_chain_answer() -> str: - return ( - "🔗 How to trace citations forward and backward from a paper

" - "Backward citation (who does this paper cite?)
" - "Read the paper's reference list — every source it cites is a potential lead. " - "This gives you foundational and seminal works on the topic.

" - "Forward citation (who has cited this paper since publication?)
" - "Use these tools — paste in the DOI or title:
" - '• Scopus ' - '— search the article, then click "Cited by N documents"
' - '• Web of Science ' - '— search the article, click "Times Cited"
' - '• Semantic Scholar ' - '— free, excellent for CS and engineering, shows "Citations" tab
' - '• OpenAlex ' - '— fully open, API-accessible citation graph

' - "Lateral search (similar papers)
" - "In PRIMO, use \"Find Similar\". In Semantic Scholar, use \"Recommended Papers\". " - "In Scopus, use \"Related Documents\".

" - "Tip: Start with one highly cited foundational paper, trace forward to find the newest " - "work, and backward to understand the theoretical roots." - ) - - -# ── Predatory / peer-review evaluation guide ───────────────────────────────── - -def _predatory_eval_answer() -> str: - return ( - "✅ How to tell if an article or journal is peer-reviewed, scholarly, or predatory

" - "Is it peer-reviewed?
" - "• Check the journal's website for a peer-review statement or editorial process description
" - "• In PRIMO, tick the Peer-reviewed filter in the left sidebar
" - "• Check if the journal is indexed in " - 'Scopus or ' - 'Web of Science — ' - "indexed = generally peer-reviewed

" - "Is it a legitimate journal?
" - "• DOAJ " - "— Directory of Open Access Journals (vetted, legitimate OA)
" - "• Web of Science Master Journal List
" - "• Scopus Source List
" - "• Think · Check · Submit: thinkchecksubmit.org " - "— a checklist to assess any journal

" - "Warning signs of predatory journals
" - "• Unsolicited email invitation to submit
" - "• No clear peer-review process or very fast acceptance (days)
" - "• High article processing charges (APCs) with no clear metrics
" - "• Not indexed in Scopus or Web of Science
" - "• Generic or misleading journal name ('International Journal of...')

" - "Need help checking a specific journal? Ask our E-Resources Librarian: " - "Rani Anand · rani.anand@ku.ac.ae" - ) - - -# ── Theses / dissertations guide ───────────────────────────────────────────── - -def _theses_answer() -> str: - return ( - "📘 Finding KU theses, dissertations, and institutional research

" - "Khalifa University Khazna Repository
" - "Khazna is KU's institutional repository — it holds KU theses, dissertations, " - "faculty publications, conference papers, and open-access research outputs.
" - '🔗 khazna.ku.ac.ae

' - "ProQuest Dissertations & Theses Global
" - "KU subscribes to ProQuest D&T Global — the world's largest database of " - "theses and dissertations from universities worldwide.
" - '🔗 ' - "Access via KU Library

" - "Other sources
" - '• EThOS — UK theses (British Library)
' - '• DART-Europe — European theses
' - '• NDLTD — Global networked digital library of theses

' - "For help with Khazna deposits or ORCID integration, contact: " - "Nikesh Narayanan · " - "nikesh.narayanan@ku.ac.ae" - ) - - -# ── Highly-cited papers guide ───────────────────────────────────────────────── - -def _highly_cited_note(topic_escaped: str) -> str: - """Append to a standard search answer when highly-cited filter is detected.""" - return ( - f"

📊 Finding highly cited papers on {topic_escaped}
" - "PRIMO doesn't sort by citation count, but these tools do:
" - '• Scopus ' - '— search your topic → Sort by Cited by (highest)
' - '• Web of Science ' - '— search → Sort by Times Cited
' - '• Semantic Scholar ' - '— free, sort by Citation Count
' - '• Google Scholar ' - '— sort by Cited by (broader but includes grey literature)' - ) - - -# ── Research Snapshot ───────────────────────────────────────────────────────── - -async def _research_snapshot(context: SearchContextPayload, model: str) -> Tuple[str, List[dict], List[dict]]: - """ - Research snapshot: - - Reasoning block at the top (collapsible). - - Cited summary in the middle. - - PRIMO results below that. - - AI tools recommendation panel always at the very end. - - Silent fallback if Semantic Scholar returns nothing. - """ - import asyncio as _asyncio - topic = context.topic - references: List[dict] = [] - - # Parallel fetch - papers, (primo_result, primo_url) = await _asyncio.gather( - _semantic_scholar_search(topic, limit=8), - _search_primo_catalog(context, limit=6), - ) - search_results = primo_result.get("results", []) or [] - - # ── Top: reasoning block ────────────────────────────────────────────────── - q_display = context.ai_tool_query or f"Research on {topic}" - answer = _search_trace_block(q_display, context) - - if papers: - summary = await _generate_cited_summary(topic, papers, model) - if not summary: - _safe_metrics_increment("summary_fallbacks") - summary = _extractive_snapshot_from_papers(topic, papers) - - # ── Middle: cited summary ───────────────────────────────────────────── - answer += f"📖 Research Snapshot: {_escape(topic)}

{summary}" - - if search_results: - answer += ( - "

KU Library results: " - "KU-accessible items are shown below." - ) - - answer += "

Selected references:
" - for i, paper in enumerate(papers[:6], 1): - answer += ( - f'
[{i}] ' - f'{_escape(paper["title"])}
' - f'' - f'{_escape(paper["authors"])} · {paper["year"]} · {_escape(paper["venue"] or "")}' - f"" - ) - answer += _source_badge("semantic", "https://www.semanticscholar.org/") - if search_results: - answer += _source_badge("primo", primo_url) - - # ── Bottom: AI tools ────────────────────────────────────────────────── - answer += _ai_tools_footer(context) - references = [{"title": p["title"], "url": p["link"]} for p in papers[:6]] - return answer, search_results, references - - # Semantic Scholar returned nothing — silent fallback - if search_results: - answer += ( - f"📚 Research starting points: {_escape(topic)}

" - "Here are KU-accessible results you can open immediately." - ) - else: - answer += ( - f"📚 Research starting points: {_escape(topic)}

" - "I didn't find direct results right now — try the AI research tools below." - ) - answer += _source_badge("primo", primo_url) - # ── Bottom: AI tools ────────────────────────────────────────────────────── - answer += _ai_tools_footer(context) - return answer, search_results, [] - - -# ── Follow-up System ────────────────────────────────────────────────────────── - -def _filters_summary(context: SearchContextPayload) -> str: - filters: List[str] = [] - if context.peer_reviewed: - filters.append("peer reviewed only") - if context.open_access: - filters.append("open access") - if context.year_from and context.year_to: - filters.append(f"{context.year_from}–{context.year_to}") - elif context.year_from: - filters.append(f"from {context.year_from}") - if context.resource_type == "books": - filters.append("books") - elif context.resource_type == "articles": - filters.append("articles") - elif context.resource_type == "both": - filters.append("articles and books") - return ", ".join(filters) - - -def _question_has_new_topic(question: str, base_topic: str) -> bool: - q = (question or "").lower() - cleaned = re.sub( - r"\b(peer[- ]reviewed|open access|last \d+ years?|past \d+ years?|books? instead|articles? instead|" - r"summari[sz]e( this topic)?|overview|brief|use pubmed|search pubmed|search primo|best databases?|deep research tools?)\b", - " ", q, - ) - tokens = [t for t in re.findall(r"[a-z0-9]+", cleaned) if t not in REFINEMENT_STOP_WORDS and len(t) > 2] - if len(tokens) < 3: - return False - base_tokens = {t for t in re.findall(r"[a-z0-9]+", (base_topic or "").lower()) if t not in REFINEMENT_STOP_WORDS} - overlap = sum(1 for t in tokens if t in base_tokens) - return overlap <= max(1, min(2, len(base_tokens))) - - -def _parse_refinement_action(question: str) -> Optional[str]: - q = (question or "").lower().strip() - if re.search(r"\b(peer[- ]reviewed|peer reviewed only|peer reviewd|peer review only)\b", q): - return "peer_reviewed_only" - if re.search(r"\b(open access|oa only|only open access)\b", q): - return "open_access_only" - if re.search(r"\b(last|past)\s+5\s+years?\b", q): - return "last_5_years" - if re.search(r"\b(last|past)\s+10\s+years?\b|\bpast decade\b", q): - return "last_10_years" - # "last 3 years", "last 2 years" etc. - m = re.search(r"\b(last|past)\s+(\d{1,2})\s+years?\b", q) - if m: - return f"last_{m.group(2)}_years" - if re.search(r"\bbooks? instead\b|\bbooks? only\b|\bfind books?\b|\bshow books?\b", q): - return "books_only" - if re.search(r"\barticles? instead\b|\barticles? only\b|\bpapers? only\b", q): - return "articles_only" - if re.search(r"\b(?:both|articles and books|books and articles)\b", q): - return "both_resources" - # Review articles only — NEW - if REVIEW_ONLY_RE.search(q): - return "review_articles_only" - # Alternative search terms — NEW - if ALT_TERMS_RE.search(q): - return "alt_terms" - if re.search(r"\b(summar(y|ize|ise)|overview|brief|what does the literature say|research snapshot)\b", q): - return "summarize_topic" - if re.search(r"\bpubmed\b", q): - return "search_pubmed" - if re.search(r"\bprimo\b", q): - return "search_primo" - if re.search(r"\b(best database|best databases|which database|which databases)\b", q): - return "best_databases" - if re.search(r"\b(deep research|deep dive|full literature review|exhaustive)\b", q): - return "deep_research_tools" - return None - - -def _resolve_base_context(client_state: Optional[ClientStatePayload]) -> Optional[SearchContextPayload]: - if not client_state or not client_state.recent_search_contexts: - return None - contexts = list(client_state.recent_search_contexts) - if client_state.follow_up_context_id: - for ctx in contexts: - if ctx.context_id == client_state.follow_up_context_id: - return ctx - if client_state.active_search_context_id: - for ctx in contexts: - if ctx.context_id == client_state.active_search_context_id: - return ctx - contexts.sort(key=lambda c: c.created_at, reverse=True) - return contexts[0] - - -def _detect_follow_up( - question: str, client_state: Optional[ClientStatePayload] -) -> Tuple[bool, Optional[str], Optional[SearchContextPayload]]: - base_context = _resolve_base_context(client_state) - if not base_context: - return False, None, None - explicit_action = (client_state.follow_up_action if client_state else None) or None - if explicit_action: - return True, explicit_action, base_context - action = _parse_refinement_action(question) - if not action: - return False, None, None - if _question_has_new_topic(question, base_context.topic): - return False, None, None - return True, action, base_context - - -def _clone_context(base_context: SearchContextPayload) -> SearchContextPayload: - return SearchContextPayload.model_validate(base_context.model_dump()) - - -def _apply_follow_up_action(base_context: SearchContextPayload, action: Optional[str]) -> SearchContextPayload: - context = _clone_context(base_context) - context.context_id = str(uuid.uuid4()) - context.created_at = time.time() - if action == "peer_reviewed_only": - context.peer_reviewed = True - elif action == "open_access_only": - context.open_access = True - elif action == "last_5_years": - context.year_from = str(CURRENT_YEAR - 4) - context.year_to = str(CURRENT_YEAR) - elif action == "last_10_years": - context.year_from = str(CURRENT_YEAR - 9) - context.year_to = str(CURRENT_YEAR) - elif action == "last_3_years": - context.year_from = str(CURRENT_YEAR - 2) - context.year_to = str(CURRENT_YEAR) - elif action == "last_2_years": - context.year_from = str(CURRENT_YEAR - 1) - context.year_to = str(CURRENT_YEAR) - elif action and action.startswith("last_") and action.endswith("_years"): - # Generic "last N years" — parse N - try: - n = int(action.split("_")[1]) - context.year_from = str(CURRENT_YEAR - n + 1) - context.year_to = str(CURRENT_YEAR) - except (IndexError, ValueError): - pass - elif action == "books_only": - context.resource_type = "books" - elif action == "articles_only": - context.resource_type = "articles" - elif action == "both_resources": - context.resource_type = "both" - elif action == "review_articles_only": - # Map to PRIMO facet for review articles — stored as resource_type hint - context.resource_type = "articles" - context.peer_reviewed = True # reviews are implicitly peer-reviewed - # Store the review filter flag in the context notes field if available - # The PRIMO query will add facet_rtype=reviews - elif action == "search_pubmed": - context.source = "pubmed" - context.intent = "search_medical" - context.resource_type = "articles" - elif action == "search_primo": - context.source = "primo" - return context - - -async def _generate_topic_follow_ups(topic: str, model: str) -> List[dict]: - settings = get_settings() - if not settings.openai_api_key and not settings.anthropic_api_key: - return [] - try: - llm = _get_llm(model, temperature=0.5, max_tokens=200) - response = await llm.ainvoke([ - {"role": "system", "content": ( - "You are a research librarian helping a user explore a topic more deeply. " - "Given a research topic, generate exactly 3 short follow-up questions that explore " - "DIFFERENT ASPECTS of the topic — such as subtopics, methodological angles, " - "applications, comparisons, or related fields. " - "Do NOT suggest filter actions like peer-reviewed, date ranges, or format changes. " - "Return ONLY a JSON array of 3 strings, each under 12 words. " - "No explanation, no preamble, no markdown.\n" - "Example for 'quantum computing':\n" - '[\"What are the main hardware approaches in quantum computing?\", ' - '"How does quantum error correction work?\", ' - '"Applications of quantum computing in cryptography\"]' - )}, - {"role": "user", "content": f"Topic: {topic}"}, - ]) - raw = response.content.strip() - if raw.startswith("```"): - raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip() - questions = json.loads(raw) - if not isinstance(questions, list): - return [] - return [ - {"label": q.strip("?") + "?", "question": q.strip("?") + "?"} - for q in questions[:3] - if isinstance(q, str) and q.strip() - ] - except Exception as e: - logger.warning(f"Topic follow-up generation failed: {e}") - return [] - - -async def _search_follow_up( - context: SearchContextPayload, model: str, summary_mode: bool = False -) -> Tuple[str, List[dict]]: - topic = context.display_topic or context.topic - suggestions: List[dict] = [] - topic_follow_ups = await _generate_topic_follow_ups(topic, model) - suggestions.extend(topic_follow_ups) - - filter_suggestions: List[dict] = [] - if summary_mode: - filter_suggestions.append({ - "label": f"Show KU-accessible articles on {topic}", - "question": f"Find KU-accessible articles on {topic}", - "action": "articles_only", - "context_id": context.context_id, - }) - filter_suggestions.append({ - "label": f"Limit {topic} to the last 5 years", - "question": f"Limit {topic} to the last 5 years", - "action": "last_5_years", - "context_id": context.context_id, - }) - else: - if not context.peer_reviewed and context.source == "primo": - filter_suggestions.append({ - "label": "Peer reviewed only", - "question": f"Make the {topic} search peer reviewed only", - "action": "peer_reviewed_only", - "context_id": context.context_id, - }) - if not (context.year_from and context.year_to == str(CURRENT_YEAR)): - filter_suggestions.append({ - "label": "Last 5 years", - "question": f"Limit {topic} to the last 5 years", - "action": "last_5_years", - "context_id": context.context_id, - }) - if context.intent == "search_medical" and context.source != "pubmed": - filter_suggestions.append({ - "label": "Search PubMed instead", - "question": f"Search PubMed for {topic}", - "action": "search_pubmed", - "context_id": context.context_id, - }) - filter_suggestions.append({ - "label": "Research snapshot", - "question": f"Give me a brief research snapshot on {topic}", - "action": "summarize_topic", - "context_id": context.context_id, - }) - - suggestions.extend(filter_suggestions[:2]) - question_text = ( - f"Here are some related angles you might want to explore on {_escape(topic)}. " - "Or I can refine the search with filters — just ask." - ) - return question_text, suggestions[:5] - - -def _library_follow_up(question: str) -> Tuple[str, List[dict]]: - q = (question or "").lower() - if _looks_library_hours_question(question): - return ( - "Would you like the library contact details too?", - [ - {"label": "Show library contact details", "question": "Show the library contact details"}, - {"label": "Open the live library hours page", "question": "Open the live library hours page"}, - ], - ) - if "study room" in q or "room" in q: - return ( - "Would you like the study-room booking link?", - [ - {"label": "Study-room booking link", "question": "Show the study-room booking link"}, - {"label": "Library services for both campuses", "question": "Show library services for both campuses"}, - ], - ) - if any(token in q for token in ["database", "databases", "search", "articles", "research"]): - return ( - "Would you like me to recommend the best KU databases for this topic?", - [ - {"label": "Recommend best KU databases", "question": "Recommend the best KU databases for this topic"}, - {"label": "How to access databases off campus", "question": "How do I access databases off campus?"}, - ], - ) - return ( - "Would you like a related library service or staff contact?", - [ - {"label": "Best staff contact", "question": "Show the best staff contact for this"}, - {"label": "KU library page for this topic", "question": "Open the KU library page for this topic"}, - ], - ) - - -def _general_follow_up(question: str) -> Tuple[str, List[dict]]: - return ( - "Would you like a brief explanation, current web information, or KU Library resources on this topic?", - [ - {"label": "Shorter explanation", "question": f"Give me a shorter explanation of {question}"}, - {"label": "Current web information", "question": f"Show current web information on {question}"}, - {"label": "Find KU Library resources", "question": f"Find KU Library resources on {question}"}, - ], - ) - - -def _social_follow_up() -> Tuple[str, List[dict]]: - return ( - "What would you like help with next?", - [ - {"label": "Find articles on a topic", "question": "Find articles on a topic"}, - {"label": "Check a library service", "question": "Check a library service"}, - {"label": "Contact a librarian", "question": "Contact a librarian"}, - ], - ) - - -def _search_answer_intro(context: SearchContextPayload, is_follow_up: bool) -> str: - topic = _escape(context.display_topic or context.topic) - resource = {"books": "books", "articles": "articles", "both": "articles and books"}.get(context.resource_type, "articles") - location = "PubMed" if context.source == "pubmed" else "KU Library catalogue" - if is_follow_up: - return f"I updated your previous search on {topic} in the {location} and looked for {resource}." - return f"I searched the {location} for {resource} on {topic}." - - -async def _run_search_mode( - question: str, context: SearchContextPayload, model: str, is_follow_up: bool -) -> Tuple[str, List[dict], List[dict], str]: - context = await _prepare_queries(question, context, model, is_follow_up) - - if context.source == "pubmed": - result, source_url = await _search_pubmed_articles(context, limit=6) - source_name = "pubmed" - else: - result, source_url = await _search_primo_catalog(context, limit=6) - source_name = "primo" - - search_results = result.get("results", []) or [] - total = result.get("total", len(search_results)) - - # ── Top: reasoning / how-I-searched block ──────────────────────────────── - answer = _search_trace_block(question, context) - - # ── Middle: answer sentence + result count ──────────────────────────────── - answer += _search_answer_intro(context, is_follow_up) - - filters = _filters_summary(context) - if filters: - answer += f"

Active filters: {_escape(filters)}." - - if search_results: - answer += f"

Found {total:,} result(s) — top {len(search_results)} shown below." - else: - answer += "

No matching results found right now — try broadening your terms or use the AI tools below." - - if context.intent == "search_medical" and context.source != "pubmed": - answer += "

For the strongest medical evidence trail, also check PubMed or Embase." - - answer += _source_badge(source_name, source_url) - - # ── Bottom: AI tools recommendation panel ───────────────────────────────── - answer += _ai_tools_footer(context) - - sources = [] - for item in search_results[:6]: - link = item.get("primo_url") or item.get("link") - if link: - sources.append({"title": item.get("title", "Result"), "url": link}) - return answer, search_results, sources, source_url - - -def _sort_and_trim_contexts(contexts: List[SearchContextPayload]) -> List[SearchContextPayload]: - dedup: Dict[str, SearchContextPayload] = {} - for ctx in contexts: - dedup[ctx.context_id] = ctx - ordered = sorted(dedup.values(), key=lambda c: c.created_at, reverse=True) - return ordered[:5] - - -# ── Main Endpoint ────────────────────────────────────────────────────────────── - -@router.post("", response_model=AgentResponse) -async def agent_query(request: Request, req: AgentRequest): - start_time = time.time() - question = _normalize_whitespace(req.question) - history = req.history - model = req.model - client_state = req.client_state - _safe_metrics_increment("agent_requests") - - if not question: - follow_up_question, follow_up_suggestions = _social_follow_up() - return AgentResponse( - answer="Please type a question and I'll do my best to help!" + _source_badge("libbee"), - intent="social", - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - if PURE_GREETING_RE.match(question): - follow_up_question, follow_up_suggestions = _social_follow_up() - return AgentResponse( - answer=( - "Hi! I'm LibBee, the Khalifa University Library AI Assistant.

" - "I can help you find articles and books, search databases, access full text, request Interlibrary Loan (ILL), " - "and answer questions about library services and staff." - + _source_badge("libbee") - ), - intent="social_greeting", - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - staff_match = match_staff_name(question) if should_attempt_staff_lookup(question) else None - if staff_match: - _safe_metrics_bucket("intents", "staff_lookup") - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=_thought_block("staff_lookup", "staff_name_match", question) + staff_name_answer(staff_match) + _source_badge("ku_library", "https://library.ku.ac.ae/"), - intent="library_info", - tools_used=["staff_name_match"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - staff_role_match = match_staff_role(question) if should_attempt_staff_lookup(question) else None - if staff_role_match: - _safe_metrics_bucket("intents", "staff_lookup") - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=_thought_block("staff_lookup", "staff_role_match", question) + staff_role_answer(staff_role_match, question) + _source_badge("ku_library", "https://library.ku.ac.ae/"), - intent="library_info", - tools_used=["staff_role_match"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - if _is_greeting_menu_followup(question, history): - follow_up_question, follow_up_suggestions = _social_follow_up() - return AgentResponse( - answer=_thought_block("social_greeting", "greeting_menu", question) + _greeting_menu_clarify_answer() + _source_badge("libbee"), - intent="social_greeting", - is_follow_up=True, - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - if _looks_library_hours_question(question): - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=_thought_block("hours", "hours_hardcoded", question) + _hours_answer() + _source_badge("ku_library", LIBRARY_HOURS_URL), - intent="library_info", - tools_used=["hours_hardcoded"], - ask_librarian=False, - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - if _looks_campus_question(question): - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=_thought_block("campus", "campus_hardcoded", question) + _campus_answer() + _source_badge("ku_library", "https://library.ku.ac.ae/"), - intent="library_info", - tools_used=["campus_hardcoded"], - ask_librarian=False, - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - if _is_database_recommendation_question(question): - answer = _database_recommendation_answer(question) - answer += _contact_footer_for_query(question, None) - answer += _source_badge("ku_library", "https://library.ku.ac.ae/eresources") - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=_thought_block("database_rec", "database_recommendation", question) + answer, - intent="library_info", - tools_used=["database_recommendation"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - is_follow_up, follow_up_action, base_context = _detect_follow_up(question, client_state) - - # ── Catalogue availability: "books available in the KU library on X" ───────── - # Catches questions asking what the library *holds*, not topic searches. - # Must fire BEFORE the LLM classifier which would route these to search_academic. - _CATALOGUE_RE = re.compile( - r"\b(" - r"available in (the )?(ku|khalifa) library" - r"|in (the )?(ku|khalifa) library" - r"|does (the )?(ku|khalifa) library have" - r"|does (the )?library (have|carry|stock|hold)" - r"|(ku|khalifa) library (has|have|carry|holds?|stocks?)" - r"|ku library (books?|copies|titles?)" - r")\b", - re.IGNORECASE, - ) - _BOOK_RE = re.compile(r"\b(book|books|ebook|ebooks|textbook|textbooks|copy|copies|title|titles)\b", re.IGNORECASE) - - if _CATALOGUE_RE.search(question) and _BOOK_RE.search(question): - # Route as library_info → RAG will handle or fall back to Ask a Librarian - # Also build a PRIMO books URL so the user can search directly - from urllib.parse import quote as _quote - _topic_raw = re.sub( - r"\b(can you (help me )?find|find|search|look for|show me|help me find" - r"|books? (available|in|on)|available in.{0,30}library|in the ku library" - r"|on|about|regarding|related to|ku library|khalifa university library" - r"|machine learning|please|thanks)\b", - " ", question, flags=re.IGNORECASE, - ) - _topic = _normalize_whitespace(_topic_raw).strip(" ,.") - _primo_books_url = ( - "https://khalifa.primo.exlibrisgroup.com/discovery/search" - f"?query=any,contains,{_quote(_topic or question)},AND" - "&tab=Everything&search_scope=MyInst_and_CI" - "&vid=971KUOSTAR_INST:KU&lang=en&mode=advanced" - "&qInclude=facet_rtype,exact,books" - ) - follow_up_question, follow_up_suggestions = _library_follow_up(question) - - # Try RAG first for any policy/service context - rag_results = [] - try: - from app import get_rag_service - rag_results = await get_rag_service().hybrid_search(question, top_k=3) - except Exception: - pass - - if rag_results: - rag_answer = await _rag_answer(question, rag_results, [], model) - else: - rag_answer = "" - - catalogue_answer = ( - f"To find books available in the KU Library on {_escape(_topic or 'this topic')}, " - f"search the KU Library catalogue directly:

" - f'' - f"📚 Search KU Library catalogue for books →

" - f"The catalogue shows real-time availability — whether a copy is on the shelf, checked out, or available as an e-book." - ) - if rag_answer: - catalogue_answer += f"

{rag_answer}" - - return AgentResponse( - answer=_thought_block("library_info", "catalogue_books", question) + catalogue_answer + _source_badge("primo", _primo_books_url), - intent="library_info", - tools_used=["catalogue_books_redirect"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - if is_follow_up and base_context: - _safe_metrics_increment("follow_up_hits") - context = _apply_follow_up_action(base_context, follow_up_action) - - if follow_up_action in {"summarize_topic", "deep_research_tools", "best_databases"}: - context = await _prepare_queries(question, context, model, True) - if follow_up_action == "summarize_topic": - answer, search_results, sources = await _research_snapshot(context, model) - else: - # deep_research_tools / best_databases follow-up - answer = ( - f"{_search_trace_block(question, context)}" - f"Research guide for {_escape(context.display_topic or context.topic)}

" - f"Here are the best AI tools and resources for a deep dive on this topic." - + _source_badge("libbee") - + _ai_tools_footer(context) - ) - search_results = [] - sources = [] - follow_up_question, follow_up_suggestions = await _search_follow_up(context, model, summary_mode=True) - _safe_metrics_increment("search_handoffs") - return AgentResponse( - answer=answer, - intent=context.intent, - tools_used=["follow_up_resolution", follow_up_action], - search_results=search_results, - sources=sources, - model_used=model, - response_time=time.time() - start_time, - corrected_query=context.topic, - natural_query=context.ai_tool_query, - database_query=context.primo_boolean_query, - ai_tool_query=context.ai_tool_query, - primo_boolean_query=context.primo_boolean_query, - primo_search_url=_primo_clean_url(context), - primo_ai_url=_tool_urls(context).get("primo_ai"), - pubmed_search_url=_tool_urls(context).get("pubmed"), - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - search_context=_context_to_dict(context), - is_follow_up=True, - ) - - answer, search_results, sources, source_url = await _run_search_mode(question, context, model, True) - follow_up_question, follow_up_suggestions = await _search_follow_up(context, model) - return AgentResponse( - answer=answer, - intent=context.intent, - tools_used=["follow_up_resolution", follow_up_action or "contextual_refinement"], - search_results=search_results, - sources=sources, - model_used=model, - response_time=time.time() - start_time, - corrected_query=context.topic, - natural_query=context.ai_tool_query, - database_query=context.primo_boolean_query, - ai_tool_query=context.ai_tool_query, - primo_boolean_query=context.primo_boolean_query, - primo_search_url=source_url, - primo_ai_url=_tool_urls(context).get("primo_ai"), - pubmed_search_url=_tool_urls(context).get("pubmed"), - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - search_context=_context_to_dict(context), - is_follow_up=True, - ) - - if is_follow_up and base_context: - _safe_metrics_increment("follow_up_hits") - context = _apply_follow_up_action(base_context, follow_up_action) - - # ── Follow-up: alternative search terms (no new search) ─────────────── - if follow_up_action == "alt_terms": - alt_answer = await _alternative_terms_answer(question, base_context, model) - follow_up_question, follow_up_suggestions = await _search_follow_up(base_context, model) - return AgentResponse( - answer=_thought_block("library_info", "alt_terms", question) + alt_answer + _source_badge("libbee"), - intent="library_info", - tools_used=["alt_terms"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - search_context=_context_to_dict(base_context), - is_follow_up=True, - ) - - if follow_up_action in {"summarize_topic", "deep_research_tools", "best_databases"}: - context = await _prepare_queries(question, context, model, True) - if follow_up_action == "summarize_topic": - answer, search_results, sources = await _research_snapshot(context, model) - else: - # deep_research_tools / best_databases follow-up - answer = ( - f"{_search_trace_block(question, context)}" - f"Research guide for {_escape(context.display_topic or context.topic)}

" - f"Here are the best AI tools and resources for a deep dive on this topic." - + _source_badge("libbee") - + _ai_tools_footer(context) - ) - search_results = [] - sources = [] - follow_up_question, follow_up_suggestions = await _search_follow_up(context, model, summary_mode=True) - _safe_metrics_increment("search_handoffs") - return AgentResponse( - answer=answer, - intent=context.intent, - tools_used=["follow_up_resolution", follow_up_action], - search_results=search_results, - sources=sources, - model_used=model, - response_time=time.time() - start_time, - corrected_query=context.topic, - natural_query=context.ai_tool_query, - database_query=context.primo_boolean_query, - ai_tool_query=context.ai_tool_query, - primo_boolean_query=context.primo_boolean_query, - primo_search_url=_primo_clean_url(context), - primo_ai_url=_tool_urls(context).get("primo_ai"), - pubmed_search_url=_tool_urls(context).get("pubmed"), - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - search_context=_context_to_dict(context), - is_follow_up=True, - ) - - answer, search_results, sources, source_url = await _run_search_mode(question, context, model, True) - follow_up_question, follow_up_suggestions = await _search_follow_up(context, model) - return AgentResponse( - answer=answer, - intent=context.intent, - tools_used=["follow_up_resolution", follow_up_action or "contextual_refinement"], - search_results=search_results, - sources=sources, - model_used=model, - response_time=time.time() - start_time, - corrected_query=context.topic, - natural_query=context.ai_tool_query, - database_query=context.primo_boolean_query, - ai_tool_query=context.ai_tool_query, - primo_boolean_query=context.primo_boolean_query, - primo_search_url=source_url, - primo_ai_url=_tool_urls(context).get("primo_ai"), - pubmed_search_url=_tool_urls(context).get("pubmed"), - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - search_context=_context_to_dict(context), - is_follow_up=True, - ) - - # ── Pre-classifier specialist handlers ─────────────────────────────────── - # These fire BEFORE the LLM classifier for patterns the classifier handles - # poorly (too often routes to search_academic for library-service questions). - - # Full-text / ILL chain - if FULLTEXT_REQUEST_RE.search(question): - ft_answer = _fulltext_chain_answer(question) - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=_thought_block("library_info", "fulltext_chain", question, - "Identified a full-text access request. Returning the KU full-text access chain.") + ft_answer + _source_badge("ku_library", "https://library.ku.ac.ae/ill/"), - intent="library_info", - tools_used=["fulltext_chain"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - # Theses / dissertations / Khazna - if THESES_RE.search(question): - th_answer = _theses_answer() - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=_thought_block("library_info", "theses_guide", question, - "Identified a theses/repository question.") + th_answer + _source_badge("ku_library", "https://khazna.ku.ac.ae"), - intent="library_info", - tools_used=["theses_guide"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - # Predatory / peer-review evaluation - if PREDATORY_RE.search(question): - pr_answer = _predatory_eval_answer() - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=_thought_block("library_info", "predatory_eval", question, - "Identified a question about evaluating journals/peer review.") + pr_answer + _source_badge("ku_library", "https://library.ku.ac.ae/eresources"), - intent="library_info", - tools_used=["predatory_eval"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - # Citation chaining - if CITATION_CHAIN_RE.search(question): - cc_answer = _citation_chain_answer() - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=_thought_block("library_info", "citation_chain", question, - "Identified a citation tracing / forward-backward search question.") + cc_answer + _source_badge("ku_library", "https://library.ku.ac.ae/eresources"), - intent="library_info", - tools_used=["citation_chain"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - # Search strategy / boolean explanation - if BOOLEAN_STRATEGY_RE.search(question): - strategy_answer, _plan = await _search_strategy_answer(question, model) - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=_thought_block("library_info", "search_strategy", question, - "Identified a search strategy / boolean building request.") + strategy_answer + _source_badge("libbee"), - intent="library_info", - tools_used=["search_strategy_explain"], - model_used=model, - response_time=time.time() - start_time, - database_query=_plan.get("boolean"), - natural_query=_plan.get("natural"), - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - # Alternative search terms (standalone, not follow-up) - if ALT_TERMS_RE.search(question): - alt_answer = await _alternative_terms_answer(question, None, model) - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=_thought_block("library_info", "alt_terms", question) + alt_answer + _source_badge("libbee"), - intent="library_info", - tools_used=["alt_terms"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - # FIX: _llm_classify always falls back to rule-based on any exception - cls_result = await _llm_classify(question, history, model) - intent = cls_result.get("intent", "general") - _safe_metrics_bucket("intents", intent) - casual_hint = cls_result.get("casual_answer", "") - - if intent == "social": - answer = ( - casual_hint - if casual_hint and len(casual_hint.strip()) > 20 - else await _libbee_casual_response(question, history, model, casual_hint) - ) - follow_up_question, follow_up_suggestions = _social_follow_up() - return AgentResponse( - answer=_thought_block("social", "libbee_casual", question) + answer + _source_badge("libbee"), - intent="social_greeting", - tools_used=["libbee_casual"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - if intent == "library_info": - rag_results = [] - try: - from app import get_rag_service - rag_results = await get_rag_service().hybrid_search(question, top_k=4) - except Exception as e: - logger.error(f"RAG search failed: {e}") - - if not rag_results: - answer = ( - "I couldn't find specific information on that in my knowledge base.

" - "Please contact our librarians directly:
" - '📧 libse@ku.ac.ae
' - f'🔗 Ask a Librarian' - ) - answer += _source_badge("ku_library", ASK_LIBRARIAN_URL) - sources = [] - tools_used = ["rag_no_results"] - thought = _thought_block("library_info", "rag_no_results", question, - "No matching content was found — I am directing you to Ask a Librarian.") - else: - answer = await _rag_answer(question, rag_results, history, model) - answer += _source_badge("rag", "https://library.ku.ac.ae/") - sources = [ - {"title": r.get("title", "Library KB"), "url": r.get("source", "")} - for r in rag_results if r.get("source") - ] - tools_used = ["rag_hybrid_search"] - thought = _thought_block("rag_search", "rag_hybrid_search", question, - f"Hybrid FAISS + BM25 search returned {len(rag_results)} relevant chunk(s) from the KB.") - - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=thought + answer, - intent="library_info", - tools_used=tools_used, - sources=sources, - ask_librarian=not bool(rag_results), - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - if intent in ("search_academic", "search_medical"): - # ── Partial title search — route to PRIMO title query ───────────────── - if PARTIAL_TITLE_RE.search(question): - from urllib.parse import quote as _qt - # Extract quoted fragment or the whole question as title search - _title_frag = (re.search(r'"([^"]{4,80})"', question) or - re.search(r'"([^"]{4,80})"', question)) - _title_q = _title_frag.group(1) if _title_frag else _light_strip_retrieval_boilerplate(question) - _title_url = ( - "https://khalifa.primo.exlibrisgroup.com/discovery/search" - f"?query=title,contains,{_qt(_title_q)},AND" - "&tab=Everything&search_scope=MyInst_and_CI" - "&vid=971KUOSTAR_INST:KU&lang=en&mode=advanced" - ) - follow_up_question, follow_up_suggestions = _library_follow_up(question) - return AgentResponse( - answer=( - _thought_block("library_info", "title_search", question, - "Identified a partial title search — routing to PRIMO title search.") + - f"📖 Title search in KU Library

" - f"I'll search the KU Library catalogue by title for: {_escape(_title_q)}

" - f'' - f"🔍 Search by title in PRIMO →

" - "If you only remember part of the title, use PRIMO's advanced search and choose " - "Title from the field dropdown — it supports partial matches.
" - "You can also try " - 'Google Scholar ' - "with the partial title in quotes." + - _source_badge("primo", _title_url) - ), - intent="library_info", - tools_used=["title_search"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - topic = await _extract_topic(question, model) - topic = _title_case_topic(topic) - year_from, year_to = _parse_year_filters(question) - - # ── Dual resource type: articles AND books in one question ───────────── - if DUAL_RESOURCE_RE.search(question): - import asyncio as _aio - # Build two contexts: articles and books - _ctx_articles = SearchContextPayload( - context_id=str(uuid.uuid4()), topic=topic, display_topic=topic, - intent=intent, source="primo", resource_type="articles", - peer_reviewed=bool(re.search(r"\bpeer[- ]reviewed\b", question, re.IGNORECASE)), - open_access=bool(re.search(r"\bopen access\b", question, re.IGNORECASE)), - year_from=year_from, year_to=year_to, - ) - _ctx_books = SearchContextPayload( - context_id=str(uuid.uuid4()), topic=topic, display_topic=topic, - intent=intent, source="primo", resource_type="books", - peer_reviewed=False, open_access=False, - year_from=year_from, year_to=year_to, - ) - _ctx_articles, _ctx_books = await _aio.gather( - _prepare_queries(question, _ctx_articles, model, False), - _prepare_queries(question, _ctx_books, model, False), - ) - (_res_a, _src_url_a), (_res_b, _src_url_b) = await _aio.gather( - _search_primo_catalog(_ctx_articles, limit=4), - _search_primo_catalog(_ctx_books, limit=3), - ) - _results_a = _res_a.get("results", []) - _results_b = _res_b.get("results", []) - _total_a = _res_a.get("total", len(_results_a)) - _total_b = _res_b.get("total", len(_results_b)) - - _dual_answer = _search_trace_block(question, _ctx_articles) - _dual_answer += ( - f"📄 Articles on {_escape(topic)} " - f"— {_total_a:,} result(s) found, top {len(_results_a)} shown below.
" - ) - if not _results_a: - _dual_answer += "No article results found — try broadening your search.
" - _dual_answer += _source_badge("primo", _src_url_a) - - _dual_answer += f"

📚 Books on {_escape(topic)} " - _dual_answer += f"— {_total_b:,} result(s) found, top {len(_results_b)} shown below.
" - if not _results_b: - _dual_answer += "No book results found — try the KU Library catalogue directly.
" - _dual_answer += _source_badge("primo", _src_url_b) - _dual_answer += _ai_tools_footer(_ctx_articles) - - _combined = _results_a + _results_b - follow_up_question, follow_up_suggestions = await _search_follow_up(_ctx_articles, model) - return AgentResponse( - answer=_dual_answer, - intent=intent, - tools_used=["primo_dual_resource"], - search_results=_combined[:8], - sources=[{"title": r.get("title", "Result"), "url": r.get("primo_url") or r.get("link", "")} - for r in _combined[:6] if r.get("primo_url") or r.get("link")], - model_used=model, - response_time=time.time() - start_time, - corrected_query=topic, - natural_query=_ctx_articles.ai_tool_query, - primo_boolean_query=_ctx_articles.primo_boolean_query, - primo_search_url=_src_url_a, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - search_context=_context_to_dict(_ctx_articles), - ) - - context = SearchContextPayload( - context_id=str(uuid.uuid4()), - topic=topic, - display_topic=topic, - intent=intent, - source="primo", - resource_type=_derive_resource_type(question), - peer_reviewed=bool(re.search(r"\bpeer[- ]reviewed\b", question, re.IGNORECASE)), - open_access=bool(re.search(r"\bopen access\b", question, re.IGNORECASE)), - year_from=year_from, - year_to=year_to, - ) - - context = await _prepare_queries(question, context, model, False) - summary_mode = _is_summary_request(question) - deep_research_mode = bool(re.search( - r"\b(deep research|deep dive|systematic review|full literature review|exhaustive)\b", - question, re.IGNORECASE, - )) - is_highly_cited = bool(HIGHLY_CITED_RE.search(question)) - - if summary_mode or deep_research_mode: - answer, search_results, sources = await _research_snapshot(context, model) - if deep_research_mode and "Research guide" not in answer and "AI tools" not in answer: - answer += f"

{_ai_tools_footer(context)}" - if is_highly_cited: - answer += _highly_cited_note(_escape(topic)) - follow_up_question, follow_up_suggestions = await _search_follow_up(context, model, summary_mode=True) - _safe_metrics_increment("search_handoffs") - return AgentResponse( - answer=answer, - intent=intent, - tools_used=["research_snapshot"], - search_results=search_results, - sources=sources, - model_used=model, - response_time=time.time() - start_time, - corrected_query=context.topic, - natural_query=context.ai_tool_query, - database_query=context.primo_boolean_query, - ai_tool_query=context.ai_tool_query, - primo_boolean_query=context.primo_boolean_query, - primo_search_url=_primo_clean_url(context), - primo_ai_url=_tool_urls(context).get("primo_ai"), - pubmed_search_url=_tool_urls(context).get("pubmed"), - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - search_context=_context_to_dict(context), - ) - - answer, search_results, sources, source_url = await _run_search_mode(question, context, model, False) - # Append highly-cited guidance if requested - if is_highly_cited: - answer += _highly_cited_note(_escape(topic)) - follow_up_question, follow_up_suggestions = await _search_follow_up(context, model) - return AgentResponse( - answer=answer, - intent=intent, - tools_used=["primo_inline" if context.source == "primo" else "pubmed_inline"], - search_results=search_results, - sources=sources, - model_used=model, - response_time=time.time() - start_time, - corrected_query=context.topic, - natural_query=context.ai_tool_query, - database_query=context.primo_boolean_query, - ai_tool_query=context.ai_tool_query, - primo_boolean_query=context.primo_boolean_query, - primo_search_url=source_url, - primo_ai_url=_tool_urls(context).get("primo_ai"), - pubmed_search_url=_tool_urls(context).get("pubmed"), - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - search_context=_context_to_dict(context), - ) - - if intent == "general_recent": - answer = await _web_search_answer(question, history, model) + _source_badge("web_live") - follow_up_question, follow_up_suggestions = _general_follow_up(question) - return AgentResponse( - answer=_thought_block("general_recent", "web_search", question) + answer, - intent="general_recent", - tools_used=["web_search"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - # general fallback - answer = await _web_search_answer(question, history, model) + _source_badge("web") - follow_up_question, follow_up_suggestions = _general_follow_up(question) - return AgentResponse( - answer=_thought_block("general", "web_search", question) + answer, - intent="general", - tools_used=["web_search"], - model_used=model, - response_time=time.time() - start_time, - follow_up_question=follow_up_question, - follow_up_suggestions=follow_up_suggestions, - ) - - -# ── Utility Endpoints ────────────────────────────────────────────────────────── - -@router.get("/test") -async def test(): return { - "status": "ok", - "message": "Agent endpoint is working!", - "staff_loaded": len(STAFF_DIRECTORY), + "welcome_message": "Hi! I'm LibBee, the Khalifa University Library AI Assistant.", + "max_results": settings.max_results, + "maintenance_mode": settings.maintenance_mode, } -@router.get("/rag-status") -async def rag_status(): - # FIX: use get_rag_service() accessor - try: - from app import get_rag_service - return get_rag_service().get_stats() - except Exception as e: - return {"error": str(e), "ready": False} +@app.get("/year") +def get_year(): + from datetime import datetime + now = datetime.utcnow() + return {"year": now.year, "month": now.month, "date": now.strftime("%Y-%m-%d")}