diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -1,219 +1,3446 @@ -"""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 """ -import asyncio +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 logging +import re import time -from contextlib import asynccontextmanager +import uuid +from datetime import datetime +from typing import Any, Dict, List, Literal, Optional, Tuple +from urllib.parse import quote -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 +import httpx +from fastapi import APIRouter, Request +from pydantic import BaseModel, ConfigDict, Field from src.config import get_settings -from src.services.cache_service import CacheService -from src.services.metrics_service import MetricsService -from src.services.rag_service import RAGService +from src.services.staff_service import ( + STAFF_DIRECTORY, + match_staff_name, + match_staff_role, + should_attempt_staff_lookup, + staff_name_answer, + staff_role_answer, +) -logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) +router = APIRouter() -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() +# ── Pydantic Models ─────────────────────────────────────────────────────────── +class ChatMessage(BaseModel): + role: str + content: str -# ── Public accessors (used by routers to avoid circular imports) ─────────────── -def get_rag_service() -> RAGService: - return rag_service +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) -def get_metrics_service() -> MetricsService: - return metrics_service +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 -# ── Background flush task ────────────────────────────────────────────────────── +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}' + ) + + +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.' + ) + + +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" + ) + + +# ── LLM Helpers ─────────────────────────────────────────────────────────────── + +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, + ) + + +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.' + ) + - # Build staff index +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." + ) try: - build_staff_index() - logger.info("Staff index built") - except Exception as exc: - logger.error("Staff index build failed: %s", exc) + 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}") - # Initialize RAG if settings.openai_api_key: try: - 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) + 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}" else: - logger.warning("OPENAI_API_KEY not set — RAG disabled") + base = _normalize_whitespace(question).strip(".?") + if len(base.split()) < 4: + base = f"Find research on {topic}" - # Start background metrics flush - flush_task = asyncio.create_task(_metrics_flush_loop(interval=30)) - logger.info("LibBee startup complete") + if filters: + return f"{base} with {', '.join(filters)}." + return base.rstrip(".") + "." - yield - # Graceful shutdown — flush remaining metrics - flush_task.cancel() +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) try: - await asyncio.wait_for(metrics_service.flush(), timeout=5.0) + 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) except Exception: - pass - logger.info("LibBee shutting down") + return _normalize_whitespace(text) -app = FastAPI( - title="LibBee - KU Library AI", - description="Khalifa University Library AI Assistant", - version="3.1.0", - lifespan=lifespan, -) +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() -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, -) +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. + +1. corrected: lightly edit the FULL request for spelling, grammar, and clarity. + Preserve every substantive concept, constraint, and task. Do NOT collapse. -@app.middleware("http") -async def operational_guardrails(request: Request, call_next): +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 + +Return ONLY valid JSON: +{{"corrected":"...","natural":"...","boolean":"(\"A\" OR \"B\") AND (\"C\" OR \"D\")","year_from":"","year_to":"","peer_reviewed":false,"open_access":false}}""" + + +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. + """ settings = get_settings() + raw_query = _normalize_whitespace(question or "") + light_query = _light_strip_retrieval_boilerplate(raw_query) or raw_query - # 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."}, - ) - 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."}, + 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 "#") ) - bucket.append(now) + 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 [] + +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: - 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."}) + 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 "" -from src.api import admin, agent, feedback, search # noqa: E402 +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]) -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 ──────────────────────────────────────────── -@app.get("/") -def health_check(): +async def _search_primo_catalog(context: SearchContextPayload, limit: int = 5) -> Tuple[dict, str]: + from src.api.search import SearchRequest, search_primo settings = get_settings() - 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"], + 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 -@app.get("/config") -def public_config(): +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."), + } + 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 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 { - "welcome_message": "Hi! I'm LibBee, the Khalifa University Library AI Assistant.", - "max_results": settings.max_results, - "maintenance_mode": settings.maintenance_mode, + "status": "ok", + "message": "Agent endpoint is working!", + "staff_loaded": len(STAFF_DIRECTORY), } -@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")} +@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}