| """search_academic / search_medical machinery: topic extraction, boolean building, URLs, research snapshot, follow-up state.""" |
| import asyncio |
| import html |
| import json |
| import logging |
| import os |
| import re |
| import time |
| import uuid |
| from datetime import datetime |
| from typing import Any, Dict, List, Literal, Optional, Tuple |
| from urllib.parse import quote |
|
|
| import httpx |
| from pydantic import BaseModel, ConfigDict, Field |
|
|
| from src.config import get_settings, LIBBEE_VERSION |
|
|
| from src.agentcore.models import ClientStatePayload, SearchContextPayload |
| from src.agentcore.scholarly import fetch_evidence_panel |
| from src.agentcore.constants import ( |
| ALT_TERMS_RE, |
| CURRENT_YEAR, |
| REFINEMENT_STOP_WORDS, |
| REVIEW_ONLY_RE, |
| _BOOLEAN_SYSTEM, |
| _BUILD_SEARCH_PLAN_SYSTEM, |
| _GUARDRAIL, |
| _URL_INSTRUCTION, |
| ) |
| from src.agentcore.utils import ( |
| _clean_database_keywords, |
| _escape, |
| _get_llm, |
| _light_strip_retrieval_boilerplate, |
| _normalize_whitespace, |
| _primo_clean_url, |
| _sanitize_boolean_for_primo, |
| _shared_build_primo_boolean_query, |
| _strip_resource_noise, |
| _title_case_topic, |
| ) |
| from src.agentcore.rendering import _ai_tools_footer, _search_trace_block, _tool_urls |
|
|
| logger = logging.getLogger(__name__) |
|
|
| def _extract_topic_regex(question: str) -> str: |
| q = _normalize_whitespace(question) |
| prefixes = [ |
| "can u give me a deep analysis of ", "can you give me a deep analysis of ", |
| "give me a deep analysis of ", "deep analysis of ", |
| "give me a summary of ", "give me a summary on ", "give me a summary about ", |
| "give me an overview of ", "give me an overview on ", |
| "give me a brief overview of ", "give me a brief summary of ", |
| "summarize the latest studies in ", "summarize latest studies in ", |
| "summarize the latest research on ", "summarize latest research on ", |
| "summarize recent studies in ", "summarize recent research on ", |
| "latest studies in ", "latest research on ", "recent studies in ", "recent research on ", |
| "summarize ", "summarise ", "summary of ", "summary on ", "summary about ", |
| "overview of ", "overview on ", "brief overview of ", "brief summary of ", |
| "analysis of ", "analyze ", "analyse ", "explain ", "tell me about ", |
| "what is ", "what are ", "describe ", "review of ", "literature review on ", |
| "recent advances in ", "recent developments in ", "state of the art in ", |
| "state of research on ", "research overview of ", "research summary of ", |
| "find peer reviewed articles on ", "find peer reviewed papers on ", |
| "find articles on ", "find papers on ", "find books on ", "find articles about ", |
| "find papers about ", "search for articles on ", "search for papers on ", |
| "i need peer reviewed articles on ", "i need articles on ", "i need papers on ", |
| "i need books on ", "give me articles on ", "show me articles on ", |
| "get me articles on ", "look for articles on ", "articles on ", "papers on ", |
| "books on ", "research on ", "literature on ", "find ", "search for ", |
| "i need ", "give me ", "show me ", "get me ", |
| ] |
| lower = q.lower() |
| for prefix in prefixes: |
| if lower.startswith(prefix): |
| q = q[len(prefix):] |
| break |
| q = re.sub(r"\s+(please|thanks|thank you)\.?$", "", q, flags=re.IGNORECASE) |
| q = re.sub(r"\b(peer[- ]reviewed|open access|last \d+ years?|past \d+ years?)\b", "", q, flags=re.IGNORECASE) |
| q = _strip_resource_noise(q) |
| return _normalize_whitespace(q).strip(".?") |
|
|
|
|
| async def _extract_topic(question: str, model: str) -> str: |
| settings = get_settings() |
| if not settings.openai_api_key and not settings.anthropic_api_key: |
| return _extract_topic_regex(question) |
| try: |
| llm = _get_llm(model, temperature=0, max_tokens=32) |
| response = await llm.ainvoke([ |
| {"role": "system", "content": ( |
| "Extract only the core research topic from the user's question. " |
| "Return 2-7 keywords ONLY — no punctuation, no sentence, no explanation. " |
| "CRITICAL: do NOT include words like articles, papers, books, journals, " |
| "peer-reviewed, open-access, studies, or research in your output. " |
| "Those are filters, not topic words. Return ONLY the subject matter.\n" |
| "Examples:\n" |
| "'find articles on indian politics' → indian politics\n" |
| "'peer reviewed papers on climate change 2020' → climate change\n" |
| "'books on structural engineering UAE' → structural engineering UAE\n" |
| "'give me a summary of quantum physics advances' → quantum physics advances\n" |
| "'i need peer reviewed research on machine learning healthcare' → machine learning healthcare" |
| )}, |
| {"role": "user", "content": question}, |
| ]) |
| topic = response.content.strip().strip('"').strip("'").strip(".") |
| topic = _strip_resource_noise(topic) |
| topic = _normalize_whitespace(topic).strip(".?") |
| if topic and len(topic) > 2: |
| return topic |
| except Exception as e: |
| logger.warning(f"Topic extraction failed: {e}") |
| return _extract_topic_regex(question) |
|
|
|
|
| async def _llm_build_boolean_query(topic: str, model: str) -> str: |
| settings = get_settings() |
| if not settings.openai_api_key and not settings.anthropic_api_key: |
| return _shared_build_primo_boolean_query(topic) |
| try: |
| llm = _get_llm(model, temperature=0, max_tokens=120) |
| response = await llm.ainvoke([ |
| {"role": "system", "content": _BOOLEAN_SYSTEM}, |
| {"role": "user", "content": f"Input: {topic.strip()}"}, |
| ]) |
| result = response.content.strip() |
| if not result or len(result) > 500: |
| raise ValueError("Bad LLM boolean output") |
| result = re.sub(r"^```[a-z]*\n?", "", result).rstrip("`").strip() |
| if '"' not in result and ' AND ' not in result and ' OR ' not in result: |
| result = f'"{result}"' |
| logger.info(f"LLM boolean: {result!r} ← topic: {topic!r}") |
| return result |
| except Exception as e: |
| logger.warning(f"LLM boolean query failed: {e} — using regex fallback") |
| return _shared_build_primo_boolean_query(topic) |
|
|
|
|
| def _derive_resource_type(question: str, existing: Optional[str] = None) -> str: |
| q = (question or "").lower() |
| if any(token in q for token in ["book", "books", "ebook", "ebooks"]): |
| return "books" |
| if any(token in q for token in ["article", "articles", "paper", "papers", "study", "studies", "journal", "journals"]): |
| return "articles" |
| return existing or "articles" |
|
|
|
|
| def _parse_year_filters(question: str) -> Tuple[Optional[str], Optional[str]]: |
| q = (question or "").lower() |
| match = re.search(r"\b(last|past)\s+(\d{1,2})\s+years?\b", q) |
| if match: |
| years = int(match.group(2)) |
| return str(CURRENT_YEAR - years + 1), str(CURRENT_YEAR) |
| between = re.search(r"\b(?:from|between)\s+(20\d{2}|19\d{2})\s+(?:to|and|-)\s+(20\d{2}|19\d{2})\b", q) |
| if between: |
| return between.group(1), between.group(2) |
| years = re.findall(r"\b(19\d{2}|20\d{2})\b", q) |
| if len(years) >= 2: |
| return min(years), max(years) |
| if len(years) == 1 and any(word in q for word in ["since", "from", "after"]): |
| return years[0], str(CURRENT_YEAR) |
| return None, None |
|
|
|
|
| def _compose_ai_tool_query(question: str, context: SearchContextPayload, is_follow_up: bool) -> str: |
| topic = context.display_topic or context.topic |
| topic = _title_case_topic(topic) |
| scope = context.resource_type |
| filters = [] |
| if context.peer_reviewed: |
| filters.append("peer-reviewed") |
| if context.open_access: |
| filters.append("open-access") |
| if context.year_from and context.year_to: |
| filters.append(f"published between {context.year_from} and {context.year_to}") |
| elif context.year_from: |
| filters.append(f"published from {context.year_from} onward") |
| if is_follow_up: |
| if scope == "books": |
| base = f"Find books on {topic}" |
| elif scope == "both": |
| base = f"Find articles and books on {topic}" |
| else: |
| base = f"Find research articles on {topic}" |
| else: |
| base = _normalize_whitespace(question).strip(".?") |
| if len(base.split()) < 4: |
| base = f"Find research on {topic}" |
| if filters: |
| return f"{base} with {', '.join(filters)}." |
| return base.rstrip(".") + "." |
|
|
|
|
| def _context_to_dict(context: SearchContextPayload) -> dict: |
| return context.model_dump() |
|
|
|
|
| async def _validate_topic(topic: str, model: str) -> Tuple[bool, str]: |
| settings = get_settings() |
| if not settings.openai_api_key and not settings.anthropic_api_key: |
| return True, topic |
| try: |
| llm = _get_llm(model, temperature=0, max_tokens=40) |
| response = await llm.ainvoke([ |
| {"role": "system", "content": ( |
| "You are a topic validator for an academic library search system. " |
| "Given a word or phrase, decide if it is a recognisable research topic, " |
| "subject area, acronym, proper noun, or concept — even if misspelled.\n\n" |
| "Rules:\n" |
| "- If recognisable or a fixable typo: return JSON {\"valid\": true, \"corrected\": \"<corrected spelling>\"}\n" |
| "- If gibberish, random characters, or completely unrecognisable: return JSON {\"valid\": false, \"corrected\": \"\"}\n" |
| "- Acronyms like NLP, ML, AI, CRISPR, IoT are always valid.\n" |
| "- Proper nouns (country names, people, organisations) are always valid.\n" |
| "- Misspellings like 'machne lernig' → corrected: 'machine learning' → valid: true\n" |
| "- Random strings like 'sdfmdnoc', 'xyzabc123', 'qwerty' → valid: false\n" |
| "Return ONLY valid JSON. No explanation." |
| )}, |
| {"role": "user", "content": f"Topic: {topic}"}, |
| ]) |
| raw = response.content.strip() |
| if raw.startswith("```"): |
| raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip() |
| s, e = raw.find("{"), raw.rfind("}") |
| if s != -1 and e > s: |
| result = json.loads(raw[s:e + 1]) |
| is_valid = bool(result.get("valid", True)) |
| corrected = str(result.get("corrected") or topic).strip() |
| return is_valid, corrected |
| return True, topic |
| except Exception as exc: |
| logger.warning(f"_validate_topic failed: {exc} — failing open") |
| return True, topic |
|
|
|
|
| async def _build_search_plan(question: str, model: str) -> dict: |
| settings = get_settings() |
| raw_query = _normalize_whitespace(question or "") |
| light_query = _light_strip_retrieval_boilerplate(raw_query) or raw_query |
| _stripped = light_query.strip() |
| _is_pure_number = bool(re.fullmatch(r"[\d\s\.\,\-\/]+", _stripped)) |
| _is_too_short = len(_stripped) < 2 |
| _is_single_char = bool(re.fullmatch(r"[a-zA-Z0-9]", _stripped)) |
| _content_words = [w for w in _stripped.split() if len(w) > 1 and not re.fullmatch(r"[\d\.\,\-]+", w)] |
| _no_content = len(_content_words) == 0 and len(_stripped) > 0 |
| if _is_pure_number or _is_too_short or _is_single_char or _no_content: |
| _display = _stripped or raw_query |
| return { |
| "corrected": raw_query, "natural": light_query, "boolean": "", |
| "database_query": "", "year_from": "", "year_to": "", |
| "peer_reviewed": False, "open_access": False, |
| "clarification_needed": True, |
| "clarification_message": ( |
| f"I want to make sure I search for the right thing. " |
| f"Could you clarify what <strong>{_escape(_display)}</strong> refers to? " |
| f"For example, is it a course code, a specific topic name, a year, " |
| f"or something else? The more detail you provide, " |
| f"the better I can build your search." |
| ), |
| } |
| if settings.openai_api_key or settings.anthropic_api_key: |
| _valid, _corrected_topic = await _validate_topic(_stripped, model) |
| if not _valid: |
| return { |
| "corrected": raw_query, "natural": light_query, "boolean": "", |
| "database_query": "", "year_from": "", "year_to": "", |
| "peer_reviewed": False, "open_access": False, |
| "clarification_needed": True, |
| "clarification_message": ( |
| f"I couldn't recognise <strong>{_escape(_stripped)}</strong> as a research topic. " |
| f"Could you check the spelling, or describe what you're looking for in more detail? " |
| f"For example: <em>\"find research on machine learning\"</em> or " |
| f"<em>\"articles on renewable energy in UAE\"</em>." |
| ), |
| } |
| if _corrected_topic and _corrected_topic.lower() != _stripped.lower(): |
| logger.info(f"_validate_topic corrected: {_stripped!r} → {_corrected_topic!r}") |
| light_query = _corrected_topic |
| if settings.openai_api_key or settings.anthropic_api_key: |
| try: |
| llm = _get_llm(model, temperature=0, max_tokens=300) |
| response = await llm.ainvoke([ |
| {"role": "system", "content": _BUILD_SEARCH_PLAN_SYSTEM}, |
| {"role": "user", "content": f'Query: "{light_query}"'}, |
| ]) |
| raw = response.content.strip() |
| if raw.startswith("```"): |
| raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip() |
| s, e = raw.find("{"), raw.rfind("}") |
| if s != -1 and e > s: |
| result = json.loads(raw[s:e + 1]) |
| else: |
| raise ValueError("No JSON found in response") |
| corrected = (result.get("corrected") or raw_query).strip() or raw_query |
| natural = (result.get("natural") or corrected).strip() or corrected |
| boolean = (result.get("boolean") or "").strip() |
| has_ops = bool(re.search(r"\b(AND|OR)\b", boolean)) if boolean else False |
| has_parens = "(" in boolean if boolean else False |
| if not has_ops or not has_parens: |
| boolean = _shared_build_primo_boolean_query(corrected) |
| boolean = _sanitize_boolean_for_primo(boolean) |
| year_from = str(result.get("year_from") or "").strip() |
| year_to = str(result.get("year_to") or "").strip() |
| if not year_from and not year_to: |
| yf, yt = _parse_year_filters(raw_query) |
| year_from = yf or "" |
| year_to = yt or "" |
| return { |
| "corrected": corrected, |
| "natural": natural, |
| "boolean": boolean, |
| "database_query": _clean_database_keywords(boolean), |
| "year_from": year_from, |
| "year_to": year_to, |
| "peer_reviewed": bool(result.get("peer_reviewed", False)), |
| "open_access": bool(result.get("open_access", False)), |
| } |
| except Exception as e: |
| logger.warning(f"_build_search_plan LLM failed: {e} — using regex fallback") |
| boolean = _shared_build_primo_boolean_query(light_query) |
| yf, yt = _parse_year_filters(raw_query) |
| pr = bool(re.search(r"\bpeer[- ]reviewed\b", raw_query, re.IGNORECASE)) |
| oa = bool(re.search(r"\bopen[- ]access\b", raw_query, re.IGNORECASE)) |
| return { |
| "corrected": raw_query, |
| "natural": light_query, |
| "boolean": boolean, |
| "database_query": _clean_database_keywords(boolean), |
| "year_from": yf or "", |
| "year_to": yt or "", |
| "peer_reviewed": pr, |
| "open_access": oa, |
| } |
|
|
|
|
| async def _prepare_queries(question: str, context: SearchContextPayload, model: str, is_follow_up: bool) -> SearchContextPayload: |
| topic = _strip_resource_noise(context.display_topic or context.topic) or context.topic |
| q = question if not is_follow_up else f"Find {context.resource_type} on {topic}" |
| plan = await _build_search_plan(q, model) |
| if plan.get("clarification_needed"): |
| context.clarification_needed = True |
| context.clarification_message = plan.get("clarification_message", "") |
| return context |
| context.ai_tool_query = plan["natural"] |
| context.primo_boolean_query = plan["boolean"] |
| if not context.year_from and plan.get("year_from"): |
| context.year_from = plan["year_from"] |
| if not context.year_to and plan.get("year_to"): |
| context.year_to = plan["year_to"] |
| if not context.peer_reviewed and plan.get("peer_reviewed"): |
| context.peer_reviewed = True |
| if not context.open_access and plan.get("open_access"): |
| context.open_access = True |
| return context |
|
|
|
|
| async def _topic_intro(topic: str, model: str) -> str: |
| """LLM #5 — 3-sentence topic intro. v3.8.1: appends _GUARDRAIL + _URL_INSTRUCTION.""" |
| settings = get_settings() |
| if not settings.openai_api_key and not settings.anthropic_api_key: |
| return "" |
| try: |
| llm = _get_llm(model, temperature=0.2, max_tokens=180) |
| response = await llm.ainvoke([ |
| {"role": "system", "content": ( |
| "You are LibBee, the Khalifa University Library AI Assistant. " |
| "Write exactly 3 clear, factual sentences introducing the given research topic " |
| "for a university student or researcher. " |
| "Cover: what the topic is, why it matters, and one key area of current research interest. " |
| "Use HTML <br> for line breaks only if needed. No markdown, no bullet points, no headings. " |
| "Be concise and informative. " |
| + _GUARDRAIL + "\n\n" + _URL_INSTRUCTION |
| )}, |
| {"role": "user", "content": f"Research topic: {topic}"}, |
| ]) |
| intro = response.content.strip() |
| return intro if intro else "" |
| except Exception as e: |
| logger.warning(f"_topic_intro failed: {e}") |
| return "" |
|
|
|
|
|
|
| def _extractive_snapshot_from_papers(topic: str, papers: List[dict]) -> str: |
| if not papers: |
| return "" |
| statements: List[str] = [] |
| for paper in papers[:4]: |
| sentences = re.split(r"(?<=[.!?])\s+", paper.get("abstract", "")) |
| lead = sentences[0].strip() if sentences else "" |
| if not lead: |
| continue |
| citation = ( |
| f'<a href="{paper["link"]}" target="_blank" ' |
| f'style="color:#1e40af;font-weight:600;text-decoration:none">' |
| f'[{_escape(paper["citation_key"])}]</a>' |
| ) |
| statements.append(f"{_escape(lead)} {citation}") |
| if not statements: |
| return f"Recent literature on <strong>{_escape(topic)}</strong> is available in the cited papers below." |
| intro = f"Recent literature on <strong>{_escape(topic)}</strong>:<br><br>" |
| return intro + "<br><br>".join(statements[:4]) |
|
|
|
|
| def _database_guidance_for_topic(context: SearchContextPayload) -> str: |
| topic = _escape(context.display_topic or context.topic) |
| urls = _tool_urls(context) |
| ai_tools = ( |
| f'<a href="{urls["leapspace"]}" target="_blank" style="color:#9a3412;font-weight:700;text-decoration:none">LeapSpace</a> · ' |
| f'<a href="{urls["scopus_ai"]}" target="_blank" style="color:#1e40af;font-weight:700;text-decoration:none">Scopus AI</a> · ' |
| f'<a href="{urls["ebsco_ai"]}" target="_blank" style="color:#9d174d;font-weight:700;text-decoration:none">EBSCO AI</a> · ' |
| f'<a href="{urls["primo_ai"]}" target="_blank" style="color:#5b21b6;font-weight:700;text-decoration:none">PRIMO AI Assistant</a> · ' |
| f'<a href="{urls["consensus"]}" target="_blank" style="color:#86198f;font-weight:700;text-decoration:none">Consensus</a>' |
| ) |
| more_results = f'<a href="{urls["primo_discovery"]}" target="_blank" style="color:#5b21b6;font-weight:700;text-decoration:none">PRIMO Library Discovery</a>' |
| if context.intent == "search_medical" and urls.get("pubmed"): |
| more_results += f' · <a href="{urls["pubmed"]}" target="_blank" style="color:#065f46;font-weight:700;text-decoration:none">PubMed</a>' |
| return ( |
| f"For deeper work on <strong>{topic}</strong>, continue with these AI research tools: {ai_tools}. " |
| f"For more results, use {more_results}. Medical topics are strongest in PubMed, Embase, and CINAHL." |
| ) |
| return ( |
| f"For deeper work on <strong>{topic}</strong>, continue with these AI research tools: {ai_tools}. " |
| f"For more results, use {more_results}." |
| ) |
|
|
|
|
| async def _search_strategy_answer(question: str, model: str) -> Tuple[str, dict]: |
| plan = await _build_search_plan(question, model) |
| topic = _escape(_light_strip_retrieval_boilerplate(question) or question) |
| boolean = _escape(plan.get("boolean", "")) |
| natural = _escape(plan.get("natural", "")) |
| explanation = ( |
| f"<strong>🔍 Search strategy for: {topic}</strong><br><br>" |
| f"<strong>1. Identify your key concepts</strong><br>" |
| f"Break the topic into 2–3 core concepts. For your question, I identified these concepts " |
| f"and suggested synonyms for each using OR, then joined concepts with AND:<br><br>" |
| f'<code style="display:block;padding:8px 12px;background:#1a1a2e;color:#C8A951;' |
| f'border-radius:8px;font-size:.82rem;word-break:break-all">{boolean}</code><br>' |
| f"<strong>Why AND and OR?</strong> AND narrows — both concepts must appear. " |
| f"OR broadens — any synonym counts. Quoting phrases like " |
| f"<code>\"machine learning\"</code> keeps them together as an exact phrase.<br><br>" |
| f"<strong>2. Apply filters</strong><br>" |
| 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.<br><br>" |
| f"<strong>3. Natural language query for AI tools</strong><br>" |
| f"For AI-powered tools (LeapSpace, Scopus AI, Consensus), use a conversational query:<br>" |
| f'<em style="color:var(--color-text-secondary)">{natural}</em><br><br>' |
| f"<strong>4. Iterate</strong><br>" |
| f"Too many results → add more AND terms or apply filters. " |
| f"Too few → remove an AND group or use broader synonyms. " |
| f"You can ask me to refine: <em>\"narrow to peer-reviewed only\"</em>, " |
| f"<em>\"limit to last 5 years\"</em>, or <em>\"suggest alternative keywords\"</em>." |
| ) |
| return explanation, plan |
|
|
|
|
| async def _alternative_terms_answer(question: str, context: Optional[SearchContextPayload], model: str) -> str: |
| topic = (context.display_topic or context.topic) if context else question |
| settings = get_settings() |
| if not settings.openai_api_key and not settings.anthropic_api_key: |
| return ( |
| f"<strong>Alternative search terms for: {_escape(topic)}</strong><br><br>" |
| "Try combining these broader, narrower, and related terms:<br>" |
| "• Use broader terms if too few results<br>" |
| "• Use narrower/specific terms if too many results<br>" |
| "• Try acronyms and full forms (e.g. AI / Artificial Intelligence)<br>" |
| "• Include British and American spellings (e.g. organisation/organization)" |
| ) |
| try: |
| llm = _get_llm(model, temperature=0.4, max_tokens=300) |
| response = await llm.ainvoke([ |
| {"role": "system", "content": ( |
| "You are an academic librarian. Given a research topic, generate a structured list of alternative " |
| "search terms. Group them as: Broader terms, Narrower/specific terms, Related concepts, Acronyms/abbreviations. " |
| "Format as HTML using <strong> for group labels and <br> for line breaks. " |
| "Keep it concise — 3-4 terms per group maximum. No bullet points, use • instead." |
| )}, |
| {"role": "user", "content": f"Research topic: {topic}"}, |
| ]) |
| terms = response.content.strip() |
| return ( |
| f"<strong>💡 Alternative search terms for: {_escape(topic)}</strong><br><br>" |
| f"{terms}<br><br>" |
| f"<strong>Tip:</strong> In PRIMO or databases, use OR between synonyms within a concept group, " |
| f"and AND between different concept groups. Ask me to run a new search with any of these." |
| ) |
| except Exception as e: |
| logger.warning(f"_alternative_terms_answer failed: {e}") |
| return f"<strong>Alternative terms for {_escape(topic)}</strong><br><br>Try synonyms, acronyms, broader/narrower terms, and related concepts in your search." |
|
|
|
|
| def _citation_chain_answer() -> str: |
| return ( |
| "<strong>🔗 How to trace citations forward and backward from a paper</strong><br><br>" |
| "<strong>Backward citation (who does this paper cite?)</strong><br>" |
| "Read the paper's reference list — every source it cites is a potential lead. " |
| "This gives you foundational and seminal works on the topic.<br><br>" |
| "<strong>Forward citation (who has cited this paper since publication?)</strong><br>" |
| "Use these tools — paste in the DOI or title:<br>" |
| '• <a href="https://www-scopus-com.khalifa.idm.oclc.org/pages/ai" target="_blank"><strong>Scopus</strong></a> ' |
| '— search the article, then click "Cited by N documents"<br>' |
| '• <a href="https://www.webofscience.com" target="_blank"><strong>Web of Science</strong></a> ' |
| '— search the article, click "Times Cited"<br>' |
| '• <a href="https://www.semanticscholar.org" target="_blank"><strong>Semantic Scholar</strong></a> ' |
| '— free, excellent for CS and engineering, shows "Citations" tab<br>' |
| '• <a href="https://openalex.org" target="_blank"><strong>OpenAlex</strong></a> ' |
| '— fully open, API-accessible citation graph<br><br>' |
| "<strong>Lateral search (similar papers)</strong><br>" |
| "In PRIMO, use <em>\"Find Similar\"</em>. In Semantic Scholar, use <em>\"Recommended Papers\"</em>. " |
| "In Scopus, use <em>\"Related Documents\"</em>.<br><br>" |
| "<strong>Tip:</strong> Start with one highly cited foundational paper, trace forward to find the newest " |
| "work, and backward to understand the theoretical roots." |
| ) |
|
|
|
|
| def _predatory_eval_answer() -> str: |
| return ( |
| "<strong>✅ How to tell if an article or journal is peer-reviewed, scholarly, or predatory</strong><br><br>" |
| "<strong>Is it peer-reviewed?</strong><br>" |
| "• Check the journal's website for a peer-review statement or editorial process description<br>" |
| "• In PRIMO, tick the <em>Peer-reviewed</em> filter in the left sidebar<br>" |
| "• Check if the journal is indexed in " |
| '<a href="https://www-scopus-com.khalifa.idm.oclc.org" target="_blank">Scopus</a> or ' |
| '<a href="https://www.webofscience.com" target="_blank">Web of Science</a> — ' |
| "indexed = generally peer-reviewed<br><br>" |
| "<strong>Is it a legitimate journal?</strong><br>" |
| "• <a href=\"https://doaj.org\" target=\"_blank\"><strong>DOAJ</strong></a> " |
| "— Directory of Open Access Journals (vetted, legitimate OA)<br>" |
| "• <a href=\"https://mjl.clarivate.com\" target=\"_blank\"><strong>Web of Science Master Journal List</strong></a><br>" |
| "• <a href=\"https://www.scopus.com/sources\" target=\"_blank\"><strong>Scopus Source List</strong></a><br>" |
| "• Think · Check · Submit: <a href=\"https://thinkchecksubmit.org\" target=\"_blank\">thinkchecksubmit.org</a> " |
| "— a checklist to assess any journal<br><br>" |
| "<strong>Warning signs of predatory journals</strong><br>" |
| "• Unsolicited email invitation to submit<br>" |
| "• No clear peer-review process or very fast acceptance (days)<br>" |
| "• High article processing charges (APCs) with no clear metrics<br>" |
| "• Not indexed in Scopus or Web of Science<br>" |
| "• Generic or misleading journal name ('International Journal of...')<br><br>" |
| "<strong>Need help checking a specific journal?</strong> Ask our E-Resources Librarian: " |
| "<strong>Rani Anand</strong> · <a href=\"mailto:rani.anand@ku.ac.ae\">rani.anand@ku.ac.ae</a>" |
| ) |
|
|
|
|
| def _highly_cited_note(topic_escaped: str) -> str: |
| return ( |
| f"<br><br><strong>📊 Finding highly cited papers on {topic_escaped}</strong><br>" |
| "PRIMO doesn't sort by citation count, but these tools do:<br>" |
| '• <a href="https://www-scopus-com.khalifa.idm.oclc.org" target="_blank"><strong>Scopus</strong></a> ' |
| '— search your topic → Sort by <em>Cited by (highest)</em><br>' |
| '• <a href="https://www.webofscience.com" target="_blank"><strong>Web of Science</strong></a> ' |
| '— search → Sort by <em>Times Cited</em><br>' |
| '• <a href="https://www.semanticscholar.org" target="_blank"><strong>Semantic Scholar</strong></a> ' |
| '— free, sort by <em>Citation Count</em><br>' |
| '• <a href="https://scholar.google.com" target="_blank"><strong>Google Scholar</strong></a> ' |
| '— sort by <em>Cited by</em> (broader but includes grey literature)' |
| ) |
|
|
|
|
| async def _research_snapshot(context: SearchContextPayload, model: str) -> Tuple[str, List[dict], List[dict]]: |
| """Research snapshot = prepared platform links + a live open-evidence panel. |
| |
| The evidence panel queries OpenAlex first (Semantic Scholar as fallback), |
| then enriches results with legal open-access links via Unpaywall and fills |
| metadata gaps via Crossref. Every external call is best-effort: if the open |
| indexes are unreachable the handler degrades to the prepared-links block |
| (the entire pre-3.8 behaviour), so the user always gets an answer. |
| """ |
| topic = context.topic |
| primo_url = _primo_clean_url(context) |
| answer = _search_trace_block(f"Research on {topic}", context) |
| answer += ( |
| f"<strong>📚 Research starting points: {_escape(topic)}</strong><br><br>" |
| "Your query has been prepared and pre-loaded across multiple platforms. " |
| "Click any platform below to search instantly, or use the boolean search tip " |
| "to search directly in any KU database." |
| f'<br><br><a href="{primo_url}" target="_blank" style="color:#003366;font-weight:700">' |
| f'Search PRIMO for: {_escape(topic)} →</a>' |
| ) |
|
|
| |
| papers, panel_html = await fetch_evidence_panel(topic) |
| answer += panel_html |
|
|
| answer += _ai_tools_footer(context) |
| return answer, papers, [] |
|
|
|
|
| def _filters_summary(context: SearchContextPayload) -> str: |
| filters: List[str] = [] |
| if context.peer_reviewed: |
| filters.append("peer reviewed only") |
| if context.open_access: |
| filters.append("open access") |
| if context.year_from and context.year_to: |
| filters.append(f"{context.year_from}–{context.year_to}") |
| elif context.year_from: |
| filters.append(f"from {context.year_from}") |
| if context.resource_type == "books": |
| filters.append("books") |
| elif context.resource_type == "articles": |
| filters.append("articles") |
| elif context.resource_type == "both": |
| filters.append("articles and books") |
| return ", ".join(filters) |
|
|
|
|
| def _question_has_new_topic(question: str, base_topic: str) -> bool: |
| q = (question or "").lower() |
| cleaned = re.sub( |
| r"\b(peer[- ]reviewed|open access|last \d+ years?|past \d+ years?|books? instead|articles? instead|" |
| r"summari[sz]e( this topic)?|overview|brief|use pubmed|search pubmed|search primo|best databases?|deep research tools?)\b", |
| " ", q, |
| ) |
| tokens = [t for t in re.findall(r"[a-z0-9]+", cleaned) if t not in REFINEMENT_STOP_WORDS and len(t) > 2] |
| if len(tokens) < 3: |
| return False |
| base_tokens = {t for t in re.findall(r"[a-z0-9]+", (base_topic or "").lower()) if t not in REFINEMENT_STOP_WORDS} |
| overlap = sum(1 for t in tokens if t in base_tokens) |
| return overlap <= max(1, min(2, len(base_tokens))) |
|
|
|
|
| def _parse_refinement_action(question: str) -> Optional[str]: |
| q = (question or "").lower().strip() |
| if re.search(r"\b(peer[- ]reviewed|peer reviewed only|peer reviewd|peer review only)\b", q): |
| return "peer_reviewed_only" |
| if re.search(r"\b(open access|oa only|only open access)\b", q): |
| return "open_access_only" |
| if re.search(r"\b(last|past)\s+5\s+years?\b", q): |
| return "last_5_years" |
| if re.search(r"\b(last|past)\s+10\s+years?\b|\bpast decade\b", q): |
| return "last_10_years" |
| 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" |
| if REVIEW_ONLY_RE.search(q): |
| return "review_articles_only" |
| if ALT_TERMS_RE.search(q): |
| return "alt_terms" |
| if re.search(r"\b(summar(y|ize|ise)|overview|brief|what does the literature say|research snapshot)\b", q): |
| _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'})] |
| if content_words: |
| return None |
| return "summarize_topic" |
| if re.search(r"\bpubmed\b", q): |
| return "search_pubmed" |
| if re.search(r"\bprimo\b", q): |
| return "search_primo" |
| if re.search(r"\b(best database|best databases|which database|which databases)\b", q): |
| return "best_databases" |
| if re.search(r"\b(deep research|deep dive|full literature review|exhaustive)\b", q): |
| return "deep_research_tools" |
| return None |
|
|
|
|
| def _resolve_base_context(client_state: Optional[ClientStatePayload]) -> Optional[SearchContextPayload]: |
| if not client_state or not client_state.recent_search_contexts: |
| return None |
| contexts = list(client_state.recent_search_contexts) |
| if client_state.follow_up_context_id: |
| for ctx in contexts: |
| if ctx.context_id == client_state.follow_up_context_id: |
| return ctx |
| if client_state.active_search_context_id: |
| for ctx in contexts: |
| if ctx.context_id == client_state.active_search_context_id: |
| return ctx |
| contexts.sort(key=lambda c: c.created_at, reverse=True) |
| return contexts[0] |
|
|
|
|
| def _detect_follow_up( |
| question: str, client_state: Optional[ClientStatePayload] |
| ) -> Tuple[bool, Optional[str], Optional[SearchContextPayload]]: |
| base_context = _resolve_base_context(client_state) |
| if not base_context: |
| return False, None, None |
| explicit_action = (client_state.follow_up_action if client_state else None) or None |
| if explicit_action: |
| return True, explicit_action, base_context |
| action = _parse_refinement_action(question) |
| if not action: |
| return False, None, None |
| if _question_has_new_topic(question, base_context.topic): |
| return False, None, None |
| return True, action, base_context |
|
|
|
|
| def _clone_context(base_context: SearchContextPayload) -> SearchContextPayload: |
| return SearchContextPayload.model_validate(base_context.model_dump()) |
|
|
|
|
| def _apply_follow_up_action(base_context: SearchContextPayload, action: Optional[str]) -> SearchContextPayload: |
| context = _clone_context(base_context) |
| context.context_id = str(uuid.uuid4()) |
| context.created_at = time.time() |
| if action == "peer_reviewed_only": |
| context.peer_reviewed = True |
| elif action == "open_access_only": |
| context.open_access = True |
| elif action == "last_5_years": |
| context.year_from = str(CURRENT_YEAR - 4) |
| context.year_to = str(CURRENT_YEAR) |
| elif action == "last_10_years": |
| context.year_from = str(CURRENT_YEAR - 9) |
| context.year_to = str(CURRENT_YEAR) |
| elif action == "last_3_years": |
| context.year_from = str(CURRENT_YEAR - 2) |
| context.year_to = str(CURRENT_YEAR) |
| elif action == "last_2_years": |
| context.year_from = str(CURRENT_YEAR - 1) |
| context.year_to = str(CURRENT_YEAR) |
| elif action and action.startswith("last_") and action.endswith("_years"): |
| 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": |
| context.resource_type = "articles" |
| context.peer_reviewed = True |
| elif action == "search_pubmed": |
| context.source = "pubmed" |
| context.intent = "search_medical" |
| context.resource_type = "articles" |
| elif action == "search_primo": |
| context.source = "primo" |
| return context |
|
|
|
|
| async def _generate_topic_follow_ups(topic: str, model: str) -> List[dict]: |
| settings = get_settings() |
| if not settings.openai_api_key and not settings.anthropic_api_key: |
| return [] |
| try: |
| llm = _get_llm(model, temperature=0.5, max_tokens=200) |
| response = await llm.ainvoke([ |
| {"role": "system", "content": ( |
| "You are a research librarian helping a user explore a topic more deeply. " |
| "Given a research topic, generate exactly 3 short follow-up questions that explore " |
| "DIFFERENT ASPECTS of the topic — such as subtopics, methodological angles, " |
| "applications, comparisons, or related fields. " |
| "Do NOT suggest filter actions like peer-reviewed, date ranges, or format changes. " |
| "Return ONLY a JSON array of 3 strings, each under 12 words. " |
| "No explanation, no preamble, no markdown.\n" |
| "Example for 'quantum computing':\n" |
| '[\"What are the main hardware approaches in quantum computing?\", ' |
| '"How does quantum error correction work?\", ' |
| '"Applications of quantum computing in cryptography\"]' |
| )}, |
| {"role": "user", "content": f"Topic: {topic}"}, |
| ]) |
| raw = response.content.strip() |
| if raw.startswith("```"): |
| raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip() |
| questions = json.loads(raw) |
| if not isinstance(questions, list): |
| return [] |
| return [ |
| {"label": q.strip("?") + "?", "question": q.strip("?") + "?"} |
| for q in questions[:3] |
| if isinstance(q, str) and q.strip() |
| ] |
| except Exception as e: |
| logger.warning(f"Topic follow-up generation failed: {e}") |
| return [] |
|
|
|
|
| async def _search_follow_up( |
| context: SearchContextPayload, model: str, summary_mode: bool = False |
| ) -> Tuple[str, List[dict]]: |
| topic = context.display_topic or context.topic |
| suggestions: List[dict] = [] |
| topic_follow_ups = await _generate_topic_follow_ups(topic, model) |
| suggestions.extend(topic_follow_ups) |
|
|
| filter_suggestions: List[dict] = [] |
| if summary_mode: |
| filter_suggestions.append({ |
| "label": f"Show KU-accessible articles on {topic}", |
| "question": f"Find KU-accessible articles on {topic}", |
| "action": "articles_only", |
| "context_id": context.context_id, |
| }) |
| 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 <strong>{_escape(topic)}</strong>. " |
| "Or I can refine the search with filters — just ask." |
| ) |
| return question_text, suggestions[:5] |
|
|
|
|
| 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 <strong>{topic}</strong> in the <strong>{location}</strong> and looked for <strong>{resource}</strong>." |
| else: |
| action = f"I searched for <strong>{resource}</strong> on <strong>{topic}</strong>." |
| return action |
|
|
|
|
| async def _run_search_mode( |
| 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 and context.clarification_message: |
| return context.clarification_message, [], [], "" |
|
|
| source_url = _primo_clean_url(context) |
| topic = _escape(context.display_topic or context.topic) |
|
|
| |
| |
| |
| intro, (papers, panel_html) = await asyncio.gather( |
| _topic_intro(context.display_topic or context.topic, model), |
| fetch_evidence_panel(context.display_topic or context.topic), |
| ) |
|
|
| answer = _search_trace_block(question, context) |
| if intro: |
| answer += f'<div style="margin-bottom:12px;color:#374151;font-size:.88rem;line-height:1.7">{intro}</div>' |
| answer += _search_answer_intro(context, is_follow_up) |
| answer += ( |
| f'<br><br>' |
| f'<a href="{source_url}" target="_blank" ' |
| f'style="display:inline-block;padding:9px 18px;background:#003366;color:#fff;' |
| f'border-radius:8px;font-weight:700;text-decoration:none;font-size:.88rem">' |
| f'🔍 Search PRIMO Library Discovery for: {topic} →</a>' |
| f'<br><div style="margin-top:6px;font-size:.78rem;color:#6b7280">' |
| f'💡 Inside PRIMO you can filter by <strong>Articles</strong>, <strong>Books</strong>, ' |
| f'<strong>Peer Reviewed</strong>, date range, and more.</div>' |
| ) |
| answer += panel_html |
| answer += _ai_tools_footer(context) |
|
|
| return answer, papers, [], source_url |
|
|
|
|
| def _sort_and_trim_contexts(contexts: List[SearchContextPayload]) -> List[SearchContextPayload]: |
| dedup: Dict[str, SearchContextPayload] = {} |
| for ctx in contexts: |
| dedup[ctx.context_id] = ctx |
| ordered = sorted(dedup.values(), key=lambda c: c.created_at, reverse=True) |
| return ordered[:5] |
|
|
|
|
|
|
