| """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 <strong>two branches</strong>:<br><br>" |
| "<strong>1. Main Campus Library</strong><br>" |
| "Location: E Building, Main Campus, Abu Dhabi<br>" |
| 'Reference desk: <a href="tel:+97123124604">+971 2 312 4604</a><br><br>' |
| "<strong>2. Habshan Library (SAN Campus)</strong><br>" |
| "Location: Building 5, Ground Floor & First Floor, Sas Al Nakhl (SAN) Campus, Abu Dhabi<br>" |
| 'Service desk: <a href="tel:+97123123160">+971 2 312 3160</a><br><br>' |
| "Both branches offer study spaces, computer workstations, printing, scanning, " |
| "and access to physical collections.<br><br>" |
| f'For general library enquiries: <a href="{ASK_LIBRARIAN_URL}" target="_blank">Ask a Librarian</a> ' |
| 'or email <a href="mailto:libse@ku.ac.ae">libse@ku.ac.ae</a>.' |
| ) |
|
|
|
|
| def _greeting_menu_clarify_answer() -> str: |
| return ( |
| "Sure — what would you like help with?<br><br>" |
| "You can choose one of these or type your question directly:<br>" |
| "• Find articles and books<br>" |
| "• Get full text for an article<br>" |
| "• Submit an Interlibrary Loan (ILL) request<br>" |
| "• Contact a librarian<br>" |
| "• Check library hours" |
| ) |
|
|
|
|
| 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 <br> 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 <a href=\"URL\" target=\"_blank\">link text</a>. " |
| "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'<a href="{ASK_LIBRARIAN_URL}" target="_blank">Ask a Librarian</a> or ' |
| f'<a href="mailto:libse@ku.ac.ae">libse@ku.ac.ae</a>.\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 <a href="{ASK_LIBRARIAN_URL}" target="_blank">Ask a Librarian</a>.' |
| ) |
|
|
|
|
| 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"<strong>Recommended databases for {topic}</strong><br><br>" |
| answer += "Here are the strongest KU-subscribed starting points:<br>" |
| for name, url, why in dbs: |
| answer += f'<br>• <a href="{url}" target="_blank"><strong>{_escape(name)}</strong></a> — {_escape(why)}' |
| if not subject_key: |
| answer += ( |
| "<br><br>💡 For subject-specific databases, ask me: " |
| "<em>\"best databases for [your subject]\"</em> — e.g. physics, chemistry, medicine, civil engineering." |
| ) |
| answer += ( |
| f'<br><br>Browse all 50+ KU databases: ' |
| f'<a href="https://library.ku.ac.ae/eresources" target="_blank">library.ku.ac.ae/eresources</a>' |
| ) |
| answer += "<br><br>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 ( |
| "<strong>📄 How to get full text for an article</strong><br><br>" |
| "Follow this chain in order — each step is faster than the next:<br><br>" |
| "<strong>Step 1 — Search KU PRIMO</strong><br>" |
| "Find the article in <a href=\"https://khalifa.primo.exlibrisgroup.com/discovery/search" |
| "?vid=971KUOSTAR_INST:KU\" target=\"_blank\">KU Library Discovery</a>. " |
| "If KU subscribes, you'll see a <em>Full Text Available</em> or <em>Online Access</em> link. " |
| "Click it — it passes through the IDM proxy automatically when on campus or logged in.<br><br>" |
| "<strong>Step 2 — Try LibKey Nomad (browser extension)</strong><br>" |
| "<a href=\"https://libkey.io/libraries/3025/\" target=\"_blank\">LibKey Nomad</a> 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.<br><br>" |
| "<strong>Step 3 — Try the publisher or open access</strong><br>" |
| "Visit the DOI link directly. Many authors post their accepted manuscript on " |
| "<a href=\"https://arxiv.org\" target=\"_blank\">arXiv</a>, " |
| "<a href=\"https://www.researchgate.net\" target=\"_blank\">ResearchGate</a>, or their institutional page. " |
| "<a href=\"https://openalex.org\" target=\"_blank\">OpenAlex</a> also links to many legal open-access PDFs.<br><br>" |
| "<strong>Step 4 — Submit an Interlibrary Loan (ILL) request</strong><br>" |
| "If none of the above work, KU Library can borrow the article from another library — usually within 1–5 working days. " |
| "Submit here: <a href=\"https://library.ku.ac.ae/ill/\" target=\"_blank\">library.ku.ac.ae/ill/</a><br>" |
| "Contact: <strong>Suaad Al Jneibi</strong> · " |
| "<a href=\"mailto:suaad.aljneibi@ku.ac.ae\">suaad.aljneibi@ku.ac.ae</a> · +971 2 312 4278" |
| ) |
|
|
|
|
| def _theses_answer() -> str: |
| return ( |
| "<strong>📘 Finding KU theses, dissertations, and institutional research</strong><br><br>" |
| "<strong>Khalifa University Khazna Repository</strong><br>" |
| "Khazna is KU's institutional repository — it holds KU theses, dissertations, " |
| "faculty publications, conference papers, and open-access research outputs.<br>" |
| '🔗 <a href="https://khazna.ku.ac.ae" target="_blank">khazna.ku.ac.ae</a><br><br>' |
| "<strong>ProQuest Dissertations & Theses Global</strong><br>" |
| "KU subscribes to ProQuest D&T Global — the world's largest database of " |
| "theses and dissertations from universities worldwide.<br>" |
| '🔗 <a href="https://khalifa.idm.oclc.org/login?url=https://www.proquest.com/pqdtglobal" target="_blank">' |
| "Access via KU Library</a><br><br>" |
| "<strong>Other sources</strong><br>" |
| '• <a href="https://ethos.bl.uk" target="_blank">EThOS</a> — UK theses (British Library)<br>' |
| '• <a href="https://www.dart-europe.org" target="_blank">DART-Europe</a> — European theses<br>' |
| '• <a href="https://ndltd.org" target="_blank">NDLTD</a> — Global networked digital library of theses<br><br>' |
| "For help with Khazna deposits or ORCID integration, contact: " |
| "<strong>Nikesh Narayanan</strong> · " |
| "<a href=\"mailto:nikesh.narayanan@ku.ac.ae\">nikesh.narayanan@ku.ac.ae</a>" |
| ) |
|
|
|
|
| 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"}, |
| ], |
| ) |
|
|
|
|
|
|
