'
+ name + ''
@@ -2579,6 +2898,11 @@ def _is_database_recommendation_question(question: str) -> bool:
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)
@@ -2586,8 +2910,8 @@ def _safe_db_url(url: str) -> str:
return url
-# ── Subject → database mapping ────────────────────────────────────────────────
-
+# ── 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."),
@@ -2609,6 +2933,7 @@ _DB_SUBJECTS: Dict[str, List[Tuple[str, str, str]]] = {
("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."),
],
@@ -2616,6 +2941,7 @@ _DB_SUBJECTS: Dict[str, List[Tuple[str, str, str]]] = {
("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."),
],
@@ -2623,10 +2949,12 @@ _DB_SUBJECTS: Dict[str, List[Tuple[str, str, str]]] = {
("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."),
@@ -2670,19 +2998,24 @@ _DB_SUBJECTS: Dict[str, List[Tuple[str, str, str]]] = {
("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."),
@@ -2717,6 +3050,7 @@ _DB_SUBJECTS: Dict[str, List[Tuple[str, str, str]]] = {
("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": [
@@ -2724,16 +3058,21 @@ _DB_SUBJECTS: Dict[str, List[Tuple[str, str, str]]] = {
("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)."),
@@ -2748,6 +3087,7 @@ _DB_SUBJECTS: Dict[str, List[Tuple[str, str, str]]] = {
],
}
+# 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."),
@@ -2756,10 +3096,12 @@ _DB_GENERAL: List[Tuple[str, str, str]] = [
("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"),
@@ -2781,7 +3123,7 @@ def _match_db_subject(q: str) -> Optional[str]:
(["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"], "business"),
+ (["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"),
@@ -2794,7 +3136,13 @@ def _match_db_subject(q: str) -> Optional[str]:
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+",
@@ -2808,6 +3156,7 @@ def _extract_db_topic(question: str) -> str:
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"
@@ -2818,8 +3167,11 @@ def _database_recommendation_answer(question: str) -> str:
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}"
@@ -2828,6 +3180,7 @@ def _database_recommendation_answer(question: str) -> str:
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."
@@ -2843,7 +3196,10 @@ def _database_recommendation_answer(question: str) -> str:
# ── §13 Hardcoded Answers ─────────────────────────────────────────────────────
+# ── 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:
"
@@ -2869,7 +3225,13 @@ def _fulltext_chain_answer(question: str) -> str:
)
+# ── 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", ""))
@@ -2886,8 +3248,8 @@ async def _search_strategy_answer(question: str, model: str) -> Tuple[str, dict]
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, language, or open access. "
- f"In PRIMO, these appear in the left sidebar after you search.
"
+ 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}'
@@ -2900,7 +3262,10 @@ async def _search_strategy_answer(question: str, model: str) -> Tuple[str, dict]
return explanation, plan
-async def _alternative_terms_answer(question: str, context: Optional["SearchContextPayload"], model: str) -> str:
+# ── 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:
@@ -2935,6 +3300,8 @@ async def _alternative_terms_answer(question: str, context: Optional["SearchCont
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"
@@ -2959,6 +3326,8 @@ def _citation_chain_answer() -> str:
)
+# ── 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"
@@ -2974,11 +3343,12 @@ def _predatory_eval_answer() -> str:
"— Directory of Open Access Journals (vetted, legitimate OA)
"
"•
Web of Science Master Journal List"
"•
Scopus Source List"
- "• Think · Check · Submit:
thinkchecksubmit.org"
+ "• 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 APCs with no clear metrics
"
+ "• 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: "
@@ -2986,6 +3356,8 @@ def _predatory_eval_answer() -> str:
)
+# ── Theses / dissertations guide ─────────────────────────────────────────────
+
def _theses_answer() -> str:
return (
"
📘 Finding KU theses, dissertations, and institutional research"
@@ -3008,7 +3380,10 @@ def _theses_answer() -> str:
)
+# ── 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:
"
@@ -3023,7 +3398,13 @@ def _highly_cited_note(topic_escaped: str) -> str:
)
-async def _research_snapshot(context: "SearchContextPayload", model: str) -> Tuple[str, List[dict], List[dict]]:
+# ── Research Snapshot ─────────────────────────────────────────────────────────
+
+async def _research_snapshot(context: SearchContextPayload, model: str) -> Tuple[str, List[dict], List[dict]]:
+ """
+ Research snapshot — no API calls.
+ Shows PRIMO browser link + AI tools footer only.
+ """
topic = context.topic
primo_url = _primo_clean_url(context)
answer = _search_trace_block(f"Research on {topic}", context)
@@ -3038,866 +3419,1244 @@ async def _research_snapshot(context: "SearchContextPayload", model: str) -> Tup
answer += _ai_tools_footer(context)
return answer, [], []
-
# ── §14 Follow-up System ──────────────────────────────────────────────────────
-def _filters_summary(context: "SearchContextPayload") -> str:
- parts = []
+# ── Follow-up System ──────────────────────────────────────────────────────────
+
+def _filters_summary(context: SearchContextPayload) -> str:
+ filters: List[str] = []
if context.peer_reviewed:
- parts.append("peer-reviewed")
+ filters.append("peer reviewed only")
if context.open_access:
- parts.append("open access")
+ filters.append("open access")
if context.year_from and context.year_to:
- parts.append(f"{context.year_from}–{context.year_to}")
+ filters.append(f"{context.year_from}–{context.year_to}")
elif context.year_from:
- parts.append(f"from {context.year_from}")
- rtype = {"articles": "articles", "books": "books", "both": "articles and books"}.get(
- context.resource_type, "articles"
- )
- parts.append(rtype)
- return ", ".join(parts)
+ 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, context: Optional["SearchContextPayload"]) -> bool:
- if not context:
- return True
- q_lower = question.lower()
- existing_topic = (context.display_topic or context.topic or "").lower()
- topic_words = set(existing_topic.split())
- if any(word in q_lower for word in topic_words):
+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
- filter_terms = {
- "peer reviewed", "peer-reviewed", "open access", "last 5 years", "last 3 years",
- "last 2 years", "last year", "articles", "books", "ebooks", "journals",
- "make", "narrow", "refine", "filter", "limit", "show", "find", "books only",
- "articles only", "yes", "yes please", "ok", "sure", "please",
- }
- for term in filter_terms:
- if term in q_lower:
- return False
- significant_words = [
- w for w in q_lower.split()
- if w not in REFINEMENT_STOP_WORDS and w not in topic_words and len(w) > 3
- ]
- return bool(significant_words) and not any(kw in q_lower for kw in topic_words)
-
-
-def _parse_refinement_action(question: str) -> Dict[str, Any]:
- q = question.lower().strip()
- action: Dict[str, Any] = {}
- if re.search(r"\bbooks?\b", q):
- action["resource_type"] = "books"
- elif re.search(r"\b(articles?|papers?)\b", q):
- action["resource_type"] = "articles"
- if re.search(r"\bpeer[- ]reviewed?\b", q):
- action["peer_reviewed"] = True
- if re.search(r"\bopen[- ]access\b", q):
- action["open_access"] = True
+ 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"
+ # summarize_topic: ONLY match short/topicless follow-ups like "summarize this",
+ # "give me an overview", "what does the literature say" — NOT questions that
+ # include an explicit new topic like "summarize quantum physics".
+ # A question with 4+ non-stop content words almost certainly has its own topic
+ # and should be handled as a fresh query, not a follow-up.
+ if re.search(r"\b(summar(y|ize|ise)|overview|brief|what does the literature say|research snapshot)\b", q):
+ _STOP = {
+ 'a','an','the','of','on','in','for','to','with','by','from','at','is','are',
+ 'was','were','be','been','have','has','had','do','does','did','will','would',
+ 'could','should','may','its','this','that','these','those','about','me','my',
+ 'give','show','tell','please','can','you','i','need','want','get','find',
+ 'research','summary','overview','summarize','summarise','brief','literature',
+ }
+ content_words = [t for t in re.findall(r'[a-z0-9]+', q)
+ if t not in _STOP and (len(t) > 2 or t in {'ai','ml','nlp','cv','rl','dl','uae','ku','iot'})]
+ # Any explicit topic word → treat as a fresh query, not a follow-up
+ if content_words:
+ return None
+ return "summarize_topic"
if re.search(r"\bpubmed\b", q):
- action["use_pubmed"] = True
- yf, yt = _parse_year_filters(question)
- if yf:
- action["year_from"] = yf
- if yt:
- action["year_to"] = yt
- return action
+ 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"],
- context_id: Optional[str],
-) -> Optional["SearchContextPayload"]:
- if not client_state:
+def _resolve_base_context(client_state: Optional[ClientStatePayload]) -> Optional[SearchContextPayload]:
+ if not client_state or not client_state.recent_search_contexts:
return None
- if context_id:
- for ctx in client_state.recent_search_contexts:
- if ctx.context_id == context_id:
+ 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 client_state.recent_search_contexts:
+ for ctx in contexts:
if ctx.context_id == client_state.active_search_context_id:
return ctx
- if client_state.recent_search_contexts:
- return client_state.recent_search_contexts[-1]
- return None
+ contexts.sort(key=lambda c: c.created_at, reverse=True)
+ return contexts[0]
def _detect_follow_up(
- question: str,
- client_state: Optional["ClientStatePayload"],
- history: List["ChatMessage"],
-) -> Tuple[bool, Optional["SearchContextPayload"]]:
- if not client_state:
- return False, None
- if client_state.follow_up_action and client_state.follow_up_context_id:
- base = _resolve_base_context(client_state, client_state.follow_up_context_id)
- if base:
- return True, base
- context = _resolve_base_context(client_state, None)
- if not context:
- return False, None
- q_lower = question.lower()
- for kw in REFINEMENT_STOP_WORDS:
- if kw in q_lower:
- return True, context
- if re.search(r"\b(peer[- ]reviewed|open[- ]access|last \d+\s*years?|past \d+\s*years?|books?|articles?|papers?)\b", q_lower):
- return True, context
- if not _question_has_new_topic(question, context):
- return True, context
- return False, None
-
-
-def _clone_context(context: "SearchContextPayload") -> "SearchContextPayload":
- data = context.model_dump()
- data["context_id"] = str(uuid.uuid4())
- data["created_at"] = time.time()
- return SearchContextPayload(**data)
-
-
-def _apply_follow_up_action(
- question: str,
- context: "SearchContextPayload",
- action: Optional[str] = None,
-) -> "SearchContextPayload":
- new_ctx = _clone_context(context)
- refinement = _parse_refinement_action(question)
- if refinement.get("resource_type"):
- new_ctx.resource_type = refinement["resource_type"]
- if refinement.get("peer_reviewed"):
- new_ctx.peer_reviewed = True
- if refinement.get("open_access"):
- new_ctx.open_access = True
- if refinement.get("year_from"):
- new_ctx.year_from = refinement["year_from"]
- if refinement.get("year_to"):
- new_ctx.year_to = refinement["year_to"]
- if refinement.get("use_pubmed"):
- new_ctx.source = "pubmed"
- if action:
- if action == "refine_peer_reviewed":
- new_ctx.peer_reviewed = True
- elif action == "refine_open_access":
- new_ctx.open_access = True
- elif action == "switch_books":
- new_ctx.resource_type = "books"
- elif action == "switch_pubmed":
- new_ctx.source = "pubmed"
- elif action == "search_more":
- new_ctx.source = "both"
- return new_ctx
-
-
-def _generate_topic_follow_ups(context: "SearchContextPayload", n_results: int) -> List[dict]:
- topic = context.display_topic or context.topic
- suggestions = []
- if not context.peer_reviewed:
- suggestions.append({"label": "🎓 Peer-reviewed only", "action": "refine_peer_reviewed", "context_id": context.context_id})
- if not context.open_access:
- suggestions.append({"label": "🔓 Open access articles", "action": "refine_open_access", "context_id": context.context_id})
- if context.resource_type != "books":
- suggestions.append({"label": "📖 Books on this topic", "action": "switch_books", "context_id": context.context_id})
- if context.intent == "search_academic":
- suggestions.append({"label": f"🔍 Boolean strategy for {topic[:40]}", "action": "show_strategy", "context_id": context.context_id})
- suggestions.append({"label": "💡 Suggest alternative keywords", "action": "alt_terms", "context_id": context.context_id})
- if n_results == 0:
- suggestions.insert(0, {"label": "🌐 Broader search", "action": "search_more", "context_id": context.context_id})
- return suggestions[:4]
-
-
-def _search_follow_up(context: "SearchContextPayload") -> List[dict]:
- return _generate_topic_follow_ups(context, 3)
-
-
-def _library_follow_up() -> List[dict]:
- return [
- {"label": "📚 Find articles on a topic", "action": "search_academic"},
- {"label": "📖 Help with full text access", "action": "fulltext_help"},
- {"label": "🗄️ Database recommendations", "action": "database_rec"},
- {"label": "📧 Contact a librarian", "action": "contact"},
- ]
+ 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
-def _general_follow_up() -> List[dict]:
- return [
- {"label": "📚 Find articles on this topic", "action": "search_academic"},
- {"label": "🔍 Help me search a database", "action": "search_help"},
- {"label": "📧 Ask a librarian", "action": "contact"},
- ]
+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 []
-def _social_follow_up() -> List[dict]:
- return [
- {"label": "📚 Find research articles", "action": "search_academic"},
- {"label": "📖 Library hours", "action": "hours"},
- {"label": "🗄️ Database help", "action": "database_rec"},
- {"label": "📧 Ask a librarian", "action": "contact"},
- ]
+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,
+ })
+ 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 _search_answer_intro(context: "SearchContextPayload", result_count: int) -> str:
- topic = _escape(context.display_topic or context.topic)
- filters = _filters_summary(context)
- source_label = "PubMed" if context.source == "pubmed" else "KU Library PRIMO"
- if result_count == 0:
+
+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 (
- f"I searched
{source_label} for
{topic} ({filters}) "
- f"but no results matched. The platforms below have your query pre-loaded — try them directly."
+ "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"},
+ ],
)
- elif result_count == 1:
- return f"Here's 1 result from
{source_label} for
{topic} ({filters}):"
+ 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:
+ action = f"I updated your previous search on
{topic} in the
{location} and looked for
{resource}."
else:
- return f"Here are {result_count} results from
{source_label} for
{topic} ({filters}):"
+ action = f"I searched for
{resource} on
{topic}."
+ return action
async def _run_search_mode(
- question: str,
- context: "SearchContextPayload",
- model: str,
- is_follow_up: bool,
- history: List["ChatMessage"],
-) -> Tuple[str, List[dict], List[dict]]:
+ question: str, context: SearchContextPayload, model: str, is_follow_up: bool
+) -> Tuple[str, List[dict], List[dict], str]:
+ """
+ Search mode — builds PRIMO link + 3-sentence topic intro + AI tools footer.
+ No external academic API calls (v3.8).
+ Includes clarification gate for ambiguous/gibberish queries.
+ """
context = await _prepare_queries(question, context, model, is_follow_up)
- if context.clarification_needed:
- msg = context.clarification_message or (
- "Could you clarify your search topic a bit more? "
- "The more detail you give, the better I can build your search."
- )
- return msg, [], []
+ # Clarification gate
+ if context.clarification_needed and context.clarification_message:
+ return context.clarification_message, [], [], ""
- topic = context.display_topic or context.topic
- intro_task = asyncio.create_task(_topic_intro(topic, model))
+ source_url = _primo_clean_url(context)
+ topic = _escape(context.display_topic or context.topic)
- if context.source == "pubmed":
- search_data, source_url = await _search_pubmed_articles(context)
- else:
- search_data, source_url = await _search_primo_catalog(context)
+ # 3-sentence topic intro
+ intro = await _topic_intro(context.display_topic or context.topic, model)
- search_results: List[dict] = search_data.get("results", [])
- sources: List[dict] = []
- n_results = len(search_results)
+ answer = _search_trace_block(question, context)
+ if intro:
+ answer += f'
{intro}
'
+ answer += _search_answer_intro(context, is_follow_up)
+ answer += (
+ f'
'
+ f'
'
+ f'🔍 Search PRIMO Library Discovery for: {topic} →'
+ f'
'
+ f'💡 Inside PRIMO you can filter by Articles, Books, '
+ f'Peer Reviewed, date range, and more.
'
+ )
+ answer += _ai_tools_footer(context)
- try:
- intro_text = await intro_task
- except Exception:
- intro_text = ""
+ return answer, [], [], source_url
- answer = _search_trace_block(question if not is_follow_up else f"Find {context.resource_type} on {topic}", context)
+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]
- if intro_text:
- answer += (
- f'
{intro_text}
'
- )
- answer += _search_answer_intro(context, n_results) + "
"
-
- if not search_results:
- paper_results_ss = await _semantic_scholar_search(topic, limit=6)
- paper_results_oa = await _openalex_search(topic, limit=6)
- combined_papers = _sort_and_trim_contexts(paper_results_ss, paper_results_oa, limit=6)
- if combined_papers:
- cited_summary = await _generate_cited_summary(topic, combined_papers[:6], model)
- summary_block = cited_summary or _extractive_snapshot_from_papers(topic, combined_papers[:4])
- if summary_block:
- answer += (
- f'
'
- f'
'
- f'📖 Open-access literature summary
'
- f'
{summary_block}
'
- f'
'
- )
- for i, paper in enumerate(combined_papers[:4], 1):
- title_e = _escape(paper.get("title", "Untitled"))
- authors_e = _escape(paper.get("authors", ""))
- year_e = _escape(str(paper.get("year", "")))
- venue_e = _escape(paper.get("venue", ""))
- abstract_e = _escape(paper.get("abstract", "")[:220])
- link = paper.get("link", "#")
- source_label_paper = paper.get("_source", "Open Access")
- answer += (
- f'
'
- f'
'
- f'
{authors_e} · {year_e}'
- f'{" · " + venue_e if venue_e else ""}
'
- f'
{abstract_e}…
'
- f'
Source: {source_label_paper}
'
- f'
'
- )
- else:
- if HIGHLY_CITED_RE.search(question):
- answer += _highly_cited_note(_escape(topic)) + "
"
-
- paper_results_ss = await _semantic_scholar_search(topic, limit=6)
- paper_results_oa = await _openalex_search(topic, limit=6)
- combined_papers = _sort_and_trim_contexts(paper_results_ss, paper_results_oa, limit=6)
-
- if combined_papers and _is_summary_request(question) and n_results < 2:
- cited_summary = await _generate_cited_summary(topic, combined_papers[:6], model)
- if cited_summary:
- answer += (
- f'
'
- f'
'
- f'📖 Research summary
'
- f'
{cited_summary}
'
- f'
'
- )
+# ── §15 Main Routing ──────────────────────────────────────────────────────────
- for r in search_results:
- title_e = _escape(r.get("title", "Untitled"))
- authors_e = _escape(r.get("authors", r.get("creator", "")))
- year_e = _escape(str(r.get("year", "")))
- type_e = _escape(r.get("resource_type", ""))
- abstract_e = _escape(r.get("abstract", "")[:250])
- link = r.get("link", source_url)
- source_e = _escape(r.get("source", "PRIMO"))
- answer += (
- f'
'
- f'
'
- f'
'
- f'{authors_e}{" · " + year_e if year_e else ""}{" · " + type_e if type_e else ""}
'
- )
- if abstract_e:
- answer += f'
{abstract_e}…
'
- answer += (
- f'
Source: {source_e}
'
- f'
'
- )
+# ── Main Endpoint ──────────────────────────────────────────────────────────────
- sources = [{"url": source_url, "title": "KU Library PRIMO"}]
+def _log_agent_query(question: str, response: AgentResponse, start_time: float) -> None:
+ """Fire-and-forget query logging to MetricsService AND Cloudflare D1 via Worker."""
+ elapsed = time.time() - start_time
+ intent = response.intent or ""
+ tool = ", ".join(response.tools_used or [])[:60]
+ model_used = response.model_used or ""
+ result_count = len(response.search_results or [])
+ # Log to local MetricsService (in-memory counters)
+ try:
+ from app import get_metrics_service
+ get_metrics_service().log_query(
+ question=question,
+ intent=intent,
+ tool=tool,
+ model=model_used,
+ response_time=elapsed,
+ result_count=result_count,
+ )
+ except Exception:
+ pass
+ # Log to Cloudflare D1 via Worker (persistent across deploys)
+ try:
+ settings = get_settings()
+ worker_url = getattr(settings, "cloudflare_worker_url", "") or ""
+ if worker_url:
+ import asyncio as _al
+ async def _post_to_worker():
+ try:
+ async with httpx.AsyncClient(timeout=5) as _cl:
+ await _cl.post(
+ worker_url.rstrip("/") + "/log",
+ json={
+ "question": question[:500],
+ "tool": f"{intent}:{tool}"[:80],
+ "model": model_used,
+ "response_time": round(elapsed, 3),
+ "result_count": result_count,
+ },
+ )
+ except Exception:
+ pass
+ try:
+ loop = _al.get_event_loop()
+ if loop.is_running():
+ loop.create_task(_post_to_worker())
+ except Exception:
+ pass
+ except Exception:
+ pass
- answer += _ai_tools_footer(context)
- return answer, search_results, sources
-def _sort_and_trim_contexts(
- list_a: List[dict],
- list_b: List[dict],
- limit: int = 8,
-) -> List[dict]:
- seen_titles: set = set()
- merged: List[dict] = []
- for paper in list_a + list_b:
- title = paper.get("title", "").strip().lower()
- if title and title not in seen_titles:
- seen_titles.add(title)
- merged.append(paper)
- return merged[:limit]
+# ── §16 FastAPI Endpoint & Utility Routes ─────────────────────────────────────
+@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")
+ response = await _agent_query_inner(question, history, model, client_state, start_time)
+ # Strip legacy discovery system references (Summon etc.) from all responses
+ response.answer = _sanitize_llm_response(response.answer)
+ _log_agent_query(question, response, start_time)
+ # Append announcement banner if set in admin config
+ _rcfg = _get_runtime_config()
+ _ann = _rcfg.get("announcement", "").strip()
+ if _ann and response.intent not in ("social_greeting", "social"):
+ _ann_html = (
+ '
'
+ f'📢 {_ann}
'
+ )
+ response.answer = response.answer + _ann_html
+ return response
-# ── §15 Main Routing ──────────────────────────────────────────────────────────
-def _log_agent_query(
+async def _agent_query_inner(
question: str,
- intent: str,
+ history,
model: str,
- response_time: float,
- tools_used: List[str],
-) -> None:
- _safe_metrics_increment("total_queries")
- _safe_metrics_bucket("intents", intent)
- _safe_metrics_bucket("models", model)
+ client_state,
+ start_time: float,
+) -> AgentResponse:
-
-async def _agent_query_inner(
- request: "AgentRequest",
-) -> "AgentResponse":
- start_time = time.time()
- question = _normalize_whitespace(request.question)
- model = request.model or "gpt"
- history = request.history or []
- client_state = request.client_state
-
- answer = ""
- intent = "general"
- tools_used: List[str] = []
- search_results: List[dict] = []
- sources: List[dict] = []
- corrected_query = None
- natural_query = None
- database_query = None
- ask_librarian = False
- is_follow_up_flag = False
- follow_up_question: Optional[str] = None
- follow_up_suggestions: List[dict] = []
- search_context_dict: Optional[dict] = None
- ai_tool_query = None
- primo_boolean_query = None
- primo_search_url = None
- primo_ai_url_out = None
- pubmed_search_url_out = None
-
- # ── Injection guard ───────────────────────────────────────────────────────
- if _INJECTION_RE.search(question):
- answer = (
- "⚠️ I detected a potential prompt injection or jailbreak attempt. "
- "This has been logged. I'm LibBee, the KU Library AI Assistant, "
- "and I can only help with library services and research. "
- f'If you need help, please
Ask a Librarian.'
- )
- _safe_metrics_bucket("errors", "injection_attempt")
+ # ── Prompt injection detection ────────────────────────────────────────────
+ if question and _INJECTION_RE.search(question):
+ _safe_metrics_bucket("intents", "injection_blocked")
return AgentResponse(
- answer=answer, intent="blocked", tools_used=["injection_guard"],
- response_time=round(time.time() - start_time, 2),
+ answer=(
+ "I'm sorry, but I can't process that request. "
+ "I'm LibBee, the KU Library AI Assistant — here to help with "
+ "library services, research, and academic resources. "
+ f'How can I help you today?
Ask a Librarian'
+ ),
+ intent="blocked",
+ tools_used=["injection_guard"],
+ model_used=model,
+ response_time=time.time() - start_time,
)
- # ── Sensitive topics guard ────────────────────────────────────────────────
- if _SENSITIVE_TOPIC_RE.search(question):
- answer = (
- "I'm sorry, but I'm not able to provide negative commentary, criticism, "
- "or controversial opinions about Khalifa University, UAE institutions, "
- "government bodies, or religious topics. "
- "As an academic library assistant, I'm here to support your research and learning. "
- "If you have a library or research question, I'm happy to help! "
- f'You can also
Ask a Librarian for personal assistance.'
+ # ── Sensitive institutional topic gate ───────────────────────────────────
+ # Handled by LLM classifier (intent == "sensitive") further below.
+ # The regex gate (_SENSITIVE_TOPIC_RE) has been removed in favour of the
+ # LLM classifier which understands context and intent more accurately.
+ # ────────────────────────────────────────────────────��────────────────────
+
+ 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,
)
- intent = "sensitive"
+
+ if PURE_GREETING_RE.match(question):
+ follow_up_question, follow_up_suggestions = _social_follow_up()
return AgentResponse(
- answer=answer, intent=intent, tools_used=["sensitive_guard"],
- response_time=round(time.time() - start_time, 2),
+ answer=(
+ _get_runtime_config().get("welcome_message", "").strip()
+ or (
+ "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,
)
- # ── Pure greeting shortcut ────────────────────────────────────────────────
- if PURE_GREETING_RE.match(question.strip()):
- answer = (
- "
🐝 Hi! I'm LibBee, the Khalifa University Library AI Assistant.
"
- "I can help you with:
"
- "• Finding research articles, books, and journals
"
- "• Library services — hours, rooms, borrowing, ILL
"
- "• Databases, full text access, and research tools
"
- "• Building search strategies and understanding research topics
"
- "Are you looking for one of these, or just type your question?"
+ 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,
)
- intent = "social"
- tools_used = ["greeting"]
- follow_up_suggestions = _social_follow_up()
+
+ 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=answer, intent=intent, tools_used=tools_used,
+ 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,
- response_time=round(time.time() - start_time, 2),
)
- # ── Greeting follow-up shortcut ───────────────────────────────────────────
if _is_greeting_menu_followup(question, history):
- answer = _greeting_menu_clarify_answer()
- intent = "social"
- tools_used = ["greeting_clarify"]
+ follow_up_question, follow_up_suggestions = _social_follow_up()
return AgentResponse(
- answer=answer, intent=intent, tools_used=tools_used,
- response_time=round(time.time() - start_time, 2),
+ 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,
)
- # ── Staff name lookup ─────────────────────────────────────────────────────
- if should_attempt_staff_lookup(question):
- name_matches = match_staff_name(question)
- if name_matches:
- answer = staff_name_answer(name_matches)
- intent = "library_info"
- tools_used = ["staff_name_match"]
- follow_up_suggestions = _library_follow_up()
- return AgentResponse(
- answer=answer, intent=intent, tools_used=tools_used,
- follow_up_suggestions=follow_up_suggestions,
- response_time=round(time.time() - start_time, 2),
- )
- role_matches = match_staff_role(question)
- if role_matches:
- answer = staff_role_answer(role_matches)
- intent = "library_info"
- tools_used = ["staff_role_match"]
- follow_up_suggestions = _library_follow_up()
- return AgentResponse(
- answer=answer, intent=intent, tools_used=tools_used,
- follow_up_suggestions=follow_up_suggestions,
- response_time=round(time.time() - start_time, 2),
- )
-
- # ── Library hours ─────────────────────────────────────────────────────────
if _looks_library_hours_question(question):
- answer = await _hours_answer()
- intent = "library_info"
- tools_used = ["libcal_hours"]
- follow_up_suggestions = _library_follow_up()
+ follow_up_question, follow_up_suggestions = _library_follow_up(question)
+ hours_html = await _hours_answer()
return AgentResponse(
- answer=answer, intent=intent, tools_used=tools_used,
+ answer=_thought_block("hours", "libcal_live", question) + hours_html + _source_badge("ku_library", LIBRARY_HOURS_URL),
+ intent="library_info",
+ tools_used=["libcal_live_hours"],
+ ask_librarian=False,
+ model_used=model,
+ response_time=time.time() - start_time,
+ follow_up_question=follow_up_question,
follow_up_suggestions=follow_up_suggestions,
- response_time=round(time.time() - start_time, 2),
)
- # ── Library events / workshops ────────────────────────────────────────────
if EVENTS_RE.search(question):
- answer = await _events_answer()
- intent = "library_info"
- tools_used = ["libcal_events"]
- follow_up_suggestions = _library_follow_up()
+ follow_up_question, follow_up_suggestions = _library_follow_up(question)
+ events_html = await _events_answer()
return AgentResponse(
- answer=answer, intent=intent, tools_used=tools_used,
+ answer=_thought_block("library_info", "libcal_events", question,
+ "I fetched upcoming events from the KU LibCal events calendar.")
+ + events_html + _source_badge("ku_library", "https://kustar.libcal.com/calendar/events"),
+ intent="library_info",
+ tools_used=["libcal_live_events"],
+ ask_librarian=False,
+ model_used=model,
+ response_time=time.time() - start_time,
+ follow_up_question=follow_up_question,
follow_up_suggestions=follow_up_suggestions,
- response_time=round(time.time() - start_time, 2),
)
- # ── Campus location ───────────────────────────────────────────────────────
if _looks_campus_question(question):
- answer = _campus_answer()
- intent = "library_info"
- tools_used = ["campus_kb"]
- follow_up_suggestions = _library_follow_up()
+ follow_up_question, follow_up_suggestions = _library_follow_up(question)
return AgentResponse(
- answer=answer, intent=intent, tools_used=tools_used,
+ 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,
- response_time=round(time.time() - start_time, 2),
)
- # ── Database recommendation ───────────────────────────────────────────────
if _is_database_recommendation_question(question):
answer = _database_recommendation_answer(question)
- intent = "library_info"
- tools_used = ["database_recommendation"]
- follow_up_suggestions = _library_follow_up()
+ 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=answer, intent=intent, tools_used=tools_used,
+ 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,
- response_time=round(time.time() - start_time, 2),
)
- # ── Follow-up detection ───────────────────────────────────────────────────
- is_follow_up, base_context = _detect_follow_up(question, client_state, history)
- follow_up_action = client_state.follow_up_action if client_state else None
+ 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:
- is_follow_up_flag = True
- refined = _apply_follow_up_action(question, base_context, follow_up_action)
- intent = refined.intent
-
- # Handle specific follow-up action types
- if follow_up_action == "show_strategy" or BOOLEAN_STRATEGY_RE.search(question):
- strategy_text, plan = await _search_strategy_answer(
- refined.ai_tool_query or f"research on {refined.topic}", model
- )
- answer = strategy_text
- tools_used = ["strategy_builder"]
- follow_up_suggestions = _search_follow_up(refined)
- elif follow_up_action == "alt_terms" or ALT_TERMS_RE.search(question):
- answer = await _alternative_terms_answer(question, refined, model)
- tools_used = ["alt_terms"]
- follow_up_suggestions = _search_follow_up(refined)
- elif follow_up_action == "contact":
- answer = (
- "Here are the KU Library contact points:
"
- f'General:
Ask a Librarian · '
- '
libse@ku.ac.ae'
- 'ILL:
ill@ku.ac.ae'
+ _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,
)
- intent = "library_info"
- tools_used = ["contact_kb"]
- else:
- answer, search_results, sources = await _run_search_mode(
- question, refined, model, is_follow_up=True, history=history
+
+ 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,
)
- tools_used = ["primo_search" if refined.source != "pubmed" else "pubmed_search"]
- search_context_dict = _context_to_dict(refined)
- primo_search_url = _primo_clean_url(refined)
- primo_boolean_query = refined.primo_boolean_query
- ai_tool_query = refined.ai_tool_query
- if refined.intent == "search_medical":
- pubmed_search_url_out = _shared_build_pubmed_url(
- refined.topic, refined.year_from, refined.year_to, refined.peer_reviewed
+
+ 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)
)
- primo_ai_url_out = PRIMO_AI_URL + quote(refined.ai_tool_query or refined.topic)
- follow_up_suggestions = _search_follow_up(refined)
+ 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,
+ )
- _log_agent_query(question, intent, model, time.time() - start_time, tools_used)
+ 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=_sanitize_llm_response(answer),
- intent=intent,
- tools_used=tools_used,
+ 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=round(time.time() - start_time, 2),
- is_follow_up=is_follow_up_flag,
+ 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=search_context_dict,
- ai_tool_query=ai_tool_query,
- primo_boolean_query=primo_boolean_query,
- primo_search_url=primo_search_url,
- primo_ai_url=primo_ai_url_out,
- pubmed_search_url=pubmed_search_url_out,
+ search_context=_context_to_dict(context),
+ is_follow_up=True,
)
- # ── Hardcoded answers ─────────────────────────────────────────────────────
+ # ── 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):
- answer = _fulltext_chain_answer(question)
- intent = "library_info"
- tools_used = ["fulltext_chain"]
- follow_up_suggestions = _library_follow_up()
+ ft_answer = _fulltext_chain_answer(question)
+ follow_up_question, follow_up_suggestions = _library_follow_up(question)
return AgentResponse(
- answer=answer, intent=intent, tools_used=tools_used,
+ 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,
- response_time=round(time.time() - start_time, 2),
)
+ # Theses / dissertations / Khazna
if THESES_RE.search(question):
- answer = _theses_answer()
- intent = "library_info"
- tools_used = ["theses_kb"]
- follow_up_suggestions = _library_follow_up()
+ th_answer = _theses_answer()
+ follow_up_question, follow_up_suggestions = _library_follow_up(question)
return AgentResponse(
- answer=answer, intent=intent, tools_used=tools_used,
+ 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,
- response_time=round(time.time() - start_time, 2),
)
+ # Predatory / peer-review evaluation
if PREDATORY_RE.search(question):
- answer = _predatory_eval_answer()
- intent = "library_info"
- tools_used = ["predatory_kb"]
- follow_up_suggestions = _library_follow_up()
+ pr_answer = _predatory_eval_answer()
+ follow_up_question, follow_up_suggestions = _library_follow_up(question)
return AgentResponse(
- answer=answer, intent=intent, tools_used=tools_used,
+ 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,
- response_time=round(time.time() - start_time, 2),
)
+ # Citation chaining
if CITATION_CHAIN_RE.search(question):
- answer = _citation_chain_answer()
- intent = "library_info"
- tools_used = ["citation_chain_kb"]
- follow_up_suggestions = _library_follow_up()
+ cc_answer = _citation_chain_answer()
+ follow_up_question, follow_up_suggestions = _library_follow_up(question)
return AgentResponse(
- answer=answer, intent=intent, tools_used=tools_used,
+ 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,
- response_time=round(time.time() - start_time, 2),
)
- # ── LLM classification ────────────────────────────────────────────────────
- classified = await _llm_classify(question, history, model)
- intent = classified.get("intent", "general")
- casual_llm = classified.get("casual_answer", "")
+ # 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,
+ )
- # ── Social ────────────────────────────────────────────────────────────────
- if intent == "social":
- answer = await _libbee_casual_response(question, history, model, hint=casual_llm)
- tools_used = ["libbee_casual"]
- follow_up_suggestions = _social_follow_up()
- _log_agent_query(question, intent, model, time.time() - start_time, tools_used)
+ # 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=answer, intent=intent, tools_used=tools_used,
+ 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,
- response_time=round(time.time() - start_time, 2),
)
- # ── Sensitive ─────────────────────────────────────────────────────────────
- if intent == "sensitive":
+ # 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 = (
- "I'm sorry, but this falls outside what I can assist with. "
- "As the KU Library AI, I focus on academic research, library services, and information access. "
- "I'm not able to provide negative commentary, sexual content, political criticism, or offensive material. "
- f'For library assistance,
Ask a Librarian.'
+ 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,
)
- _log_agent_query(question, intent, model, time.time() - start_time, ["sensitive_guard"])
+
+ if intent == "sensitive":
+ _safe_metrics_bucket("intents", "sensitive_blocked")
+ follow_up_question, follow_up_suggestions = _social_follow_up()
return AgentResponse(
- answer=answer, intent=intent, tools_used=["sensitive_guard"],
- response_time=round(time.time() - start_time, 2),
+ answer=(
+ "That's not something I'm able to comment on. "
+ "As the KU Library AI Assistant, I'm here to support your academic "
+ "and research needs — finding articles, accessing databases, "
+ "library services, and research tools.
"
+ "Is there something I can help you with today? "
+ f'
💬 Ask a Librarian'
+ ),
+ intent="sensitive_blocked",
+ tools_used=["sensitive_intent_guard"],
+ model_used=model,
+ response_time=time.time() - start_time,
+ follow_up_question=follow_up_question,
+ follow_up_suggestions=follow_up_suggestions,
)
- # ── Library info — RAG ────────────────────────────────────────────────────
if intent == "library_info":
- tools_used.append("rag_search")
+ rag_results = []
try:
from app import get_rag_service
- rag_svc = get_rag_service()
- rag_results = await rag_svc.query(question, top_k=4)
+ rag_results = await get_rag_service().hybrid_search(question, top_k=4)
except Exception as e:
- logger.warning(f"RAG service error: {e}")
- rag_results = []
+ logger.error(f"RAG search failed: {e}")
- if rag_results:
- answer = await _rag_answer(question, rag_results, history, model)
+ 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 _llm_general_answer(question, history, model)
- tools_used.append("llm_general")
-
- answer += _contact_footer_for_query(question, None)
- answer += _thought_block("library_info", "rag_search", question)
- follow_up_suggestions = _library_follow_up()
- _log_agent_query(question, intent, model, time.time() - start_time, tools_used)
+ 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=_sanitize_llm_response(answer), intent=intent, tools_used=tools_used,
+ 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,
- response_time=round(time.time() - start_time, 2),
- )
-
- # ── Boolean strategy request ──────────────────────────────────────────────
- if BOOLEAN_STRATEGY_RE.search(question):
- strategy_text, plan = await _search_strategy_answer(question, model)
- _log_agent_query(question, "library_info", model, time.time() - start_time, ["strategy_builder"])
- return AgentResponse(
- answer=strategy_text, intent="library_info",
- tools_used=["strategy_builder"],
- response_time=round(time.time() - start_time, 2),
- )
-
- # ── Alt terms request ─────────────────────────────────────────────────────
- if ALT_TERMS_RE.search(question):
- answer = await _alternative_terms_answer(question, None, model)
- _log_agent_query(question, "library_info", model, time.time() - start_time, ["alt_terms"])
- return AgentResponse(
- answer=answer, intent="library_info",
- tools_used=["alt_terms"],
- response_time=round(time.time() - start_time, 2),
)
- # ── Academic and medical search ───────────────────────────────────────────
if intent in ("search_academic", "search_medical"):
- topic_raw = await _extract_topic(question, model)
- topic_raw = _normalize_whitespace(topic_raw).strip(".?") or question
+ # ── 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,
+ )
- is_medical = intent == "search_medical" or _looks_medical_search(question)
- resource_type = _derive_resource_type(question)
+ 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),
+ )
+ # No API — just build PRIMO links for articles and books
+ _ctx_articles = await _prepare_queries(question, _ctx_articles, model, False)
+ _src_url_a = _primo_clean_url(_ctx_articles)
+ _src_url_b = _primo_clean_url(_ctx_books)
+
+ _dual_answer = _search_trace_block(question, _ctx_articles)
+ _dual_answer += (
+ f"
📄 Articles on {_escape(topic)}"
+ f'
'
+ f'Search PRIMO for articles →'
+ )
+ _dual_answer += _source_badge("primo", _src_url_a)
+ _dual_answer += f"
📚 Books on {_escape(topic)}"
+ _dual_answer += (
+ f'
'
+ f'Search PRIMO for books →'
+ )
+ _dual_answer += _source_badge("primo", _src_url_b)
+ _dual_answer += _ai_tools_footer(_ctx_articles)
+
+ follow_up_question, follow_up_suggestions = await _search_follow_up(_ctx_articles, model)
+ return AgentResponse(
+ answer=_dual_answer,
+ intent=intent,
+ tools_used=["primo_browser_link"],
+ search_results=[],
+ sources=[],
+ 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_raw,
- display_topic=_title_case_topic(topic_raw),
+ topic=topic,
+ display_topic=topic,
intent=intent,
- source="pubmed" if is_medical else "primo",
- resource_type=resource_type,
+ 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)),
- )
- yf, yt = _parse_year_filters(question)
- if yf:
- context.year_from = yf
- if yt:
- context.year_to = yt
-
- answer, search_results, sources = await _run_search_mode(
- question, context, model, is_follow_up=False, history=history
+ open_access=bool(re.search(r"\bopen access\b", question, re.IGNORECASE)),
+ year_from=year_from,
+ year_to=year_to,
)
- search_context_dict = _context_to_dict(context)
- primo_search_url = _primo_clean_url(context)
- primo_boolean_query = context.primo_boolean_query
- ai_tool_query = context.ai_tool_query
- if is_medical:
- pubmed_search_url_out = _shared_build_pubmed_url(
- context.topic, context.year_from, context.year_to, context.peer_reviewed
+
+ 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),
)
- primo_ai_url_out = PRIMO_AI_URL + quote(context.ai_tool_query or context.topic)
- follow_up_suggestions = _search_follow_up(context)
- tools_used = ["primo_search" if context.source != "pubmed" else "pubmed_search",
- "semantic_scholar", "openalex"]
- _log_agent_query(question, intent, model, time.time() - start_time, tools_used)
+ 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=_sanitize_llm_response(answer),
+ answer=answer,
intent=intent,
- tools_used=tools_used,
+ tools_used=["primo_inline" if context.source == "primo" else "pubmed_inline"],
search_results=search_results,
sources=sources,
model_used=model,
- response_time=round(time.time() - start_time, 2),
- corrected_query=corrected_query,
- natural_query=ai_tool_query,
- database_query=primo_boolean_query,
- is_follow_up=is_follow_up_flag,
+ 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=search_context_dict,
- ai_tool_query=ai_tool_query,
- primo_boolean_query=primo_boolean_query,
- primo_search_url=primo_search_url,
- primo_ai_url=primo_ai_url_out,
- pubmed_search_url=pubmed_search_url_out,
+ search_context=_context_to_dict(context),
)
- # ── Live web search (general_recent) ─────────────────────────────────────
if intent == "general_recent":
- answer = await _web_search_answer(question, history, model)
- tools_used = ["web_search"]
- answer += _thought_block("general_recent", "web_search", question)
- follow_up_suggestions = _general_follow_up()
- _log_agent_query(question, intent, model, time.time() - start_time, tools_used)
+ 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=_sanitize_llm_response(answer), intent=intent, tools_used=tools_used,
+ 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,
- response_time=round(time.time() - start_time, 2),
)
- # ── General fallback ─────────────────────────────────────────────────────
- answer = await _llm_general_answer(question, history, model)
- tools_used = ["llm_general"]
- answer += _thought_block("general", "llm", question)
- follow_up_suggestions = _general_follow_up()
- _log_agent_query(question, "general", model, time.time() - start_time, tools_used)
+ # 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=_sanitize_llm_response(answer), intent="general", tools_used=tools_used,
+ 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,
- response_time=round(time.time() - start_time, 2),
)
-# ── §16 FastAPI Endpoints ─────────────────────────────────────────────────────
-
-@router.post("/query", response_model=AgentResponse)
-async def agent_query(request: AgentRequest, http_request: Request):
- settings = get_settings()
-
- if not request.question or not request.question.strip():
- return AgentResponse(
- answer="Please ask me something! Type your question or choose from the suggestions below.",
- intent="social",
- tools_used=[],
- response_time=0,
- )
-
- try:
- result = await _agent_query_inner(request)
- result.answer = _sanitize_llm_response(result.answer)
-
- cfg = _get_runtime_config()
- announcement = cfg.get("announcement", "").strip()
- if announcement:
- result.answer = (
- f'
'
- f'📢 {_escape(announcement)}
'
- ) + result.answer
-
- return result
-
- except Exception as e:
- logger.exception(f"agent_query error: {e}")
- return AgentResponse(
- answer=(
- "I encountered an unexpected error. Please try again in a moment. "
- f'If the problem persists,
Ask a Librarian.'
- ),
- intent="error",
- tools_used=["error_handler"],
- response_time=0,
- )
-
+# ── Utility Endpoints ──────────────────────────────────────────────────────────
@router.get("/test")
-async def test_agent():
- return {"status": "ok", "version": "3.8.1", "service": "LibBee Agent"}
+async def test():
+ return {
+ "status": "ok",
+ "message": "Agent endpoint is working!",
+ "staff_loaded": len(STAFF_DIRECTORY),
+ }
@router.get("/rag-status")
async def rag_status():
+ # FIX: use get_rag_service() accessor
try:
from app import get_rag_service
- svc = get_rag_service()
- return {
- "rag_ready": svc.is_ready(),
- "chunk_count": len(svc.bm25_corpus) if svc.is_ready() else 0,
- }
+ return get_rag_service().get_stats()
except Exception as e:
- return {"rag_ready": False, "error": str(e)}
+ return {"error": str(e), "ready": False}