"""library_info intent handlers: campus, RAG answers, database recommendations, full-text chain, theses."""
import asyncio
import html
import json
import logging
import os
import re
import time
import uuid
from datetime import datetime
from typing import Any, Dict, List, Literal, Optional, Tuple
from urllib.parse import quote
import httpx
from pydantic import BaseModel, ConfigDict, Field
from src.config import get_settings, LIBBEE_VERSION
from src.services.staff_service import (
STAFF_DIRECTORY,
match_staff_name,
match_staff_role,
should_attempt_staff_lookup,
staff_name_answer,
staff_role_answer,
)
from src.agentcore.models import ChatMessage
from src.agentcore.constants import (
ASK_LIBRARIAN_URL,
DATABASE_RECOMMEND_RE,
_DB_GENERAL,
_DB_SUBJECTS,
_GUARDRAIL,
_URL_INSTRUCTION,
)
from src.agentcore.utils import (
_build_history_messages,
_escape,
_get_llm,
_get_runtime_config,
_normalize_whitespace,
_strip_resource_noise,
)
from src.agentcore.classify import _looks_library_hours_question
logger = logging.getLogger(__name__)
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"
)
async def _rag_answer(question: str, rag_results: List[dict], history: List[ChatMessage], model: str) -> str:
"""LLM #2 — RAG answer from KB chunks. v3.8.1: appends _GUARDRAIL + _URL_INSTRUCTION."""
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. "
"IMPORTANT — LINKS: When the context contains a URL (starting with https:// or http://), "
"include it as a clickable HTML link using link text. "
"For example, if context mentions 'library.ku.ac.ae/ill/' include it as a link. "
"NEVER invent or guess URLs that do not appear in the context. "
"Only link to URLs that are explicitly present in the context below. "
f"If the context does not contain enough information, say so and suggest Ask a Librarian at "
f'Ask a Librarian or '
f'libse@ku.ac.ae.\n\n'
f"CONTEXT:\n{context}\n\n"
+ _GUARDRAIL + "\n\n" + _URL_INSTRUCTION
)
_cfg = _get_runtime_config()
_ci = _cfg.get("custom_instructions", "").strip()
if _ci:
system += "\n\nAdditional instructions: " + _ci
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.'
)
def _is_database_recommendation_question(question: str) -> bool:
return bool(DATABASE_RECOMMEND_RE.search(question or ""))
def _match_db_subject(q: str) -> Optional[str]:
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:
q = _normalize_whitespace(question or "")
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
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)
subject_key = _match_db_subject(q)
dbs = _DB_SUBJECTS.get(subject_key, _DB_GENERAL) if subject_key else _DB_GENERAL
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:
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
def _fulltext_chain_answer(question: str) -> str:
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"
)
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"
)
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"},
],
)