"""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.primo_chat import fetch_primo_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, _isbn_search_value, _mask_quoted_search_text, _quoted_search_parts, _normalize_whitespace, _primo_clean_url, _sanitize_boolean_for_primo, _shared_build_primo_boolean_query, _shared_build_pubmed_url, _strip_resource_noise, _title_case_topic, ) from src.agentcore.rendering import _ai_tools_footer, _search_trace_block, _tool_urls logger = logging.getLogger(__name__) _YEAR = r"(?:19|20)\d{2}" _DATE_PATTERNS = ( ("range", re.compile(rf"\b(?:published\s+)?(?:(?:from|between)\s+)?({_YEAR})\s*(?:to|and|[-–—])\s*({_YEAR})\b", re.I)), ("since", re.compile(rf"\b(?:published\s+)?(since|from|after)\s+({_YEAR})(?:\s+(?:onwards?|and later|to present|to now))?\b", re.I)), ("before", re.compile(rf"\b(?:published\s+)?(before|until|through|up to)\s+({_YEAR})\b", re.I)), ("single", re.compile(rf"\b(?:during|published(?: in)?|in)\s+({_YEAR})\b", re.I)), ("relative", re.compile(r"\b(?:last|past)\s+([1-9]\d?)\s+years?\b", re.I)), ("decade", re.compile(r"\b(?:last|past) decade\b", re.I)), ("this_year", re.compile(r"\bthis year\b", re.I)), ("last_year", re.compile(r"\blast year\b", re.I)), ("recent", re.compile(r"\b(?:recent|latest|newest)\b", re.I)), ) def _has_recent_instruction(question: str) -> bool: return bool(_DATE_PATTERNS[-1][1].search(_mask_quoted_search_text(question))) def _title_search_value(question: str) -> str: """Select named-item searches, without treating quoted subject phrases as titles.""" q = _normalize_whitespace(question) cue = re.search( r"\b(?:(?:called|titled|entitled|named)\s+|" r"(?:only\s+)?(?:remember|recall|know).{0,35}\b(?:title|name)\s*[:=]?\s+|" r"title\s+(?:starts?|begins?)\s+with\s+|title\s*[:=]\s*|" r"(?:book|textbook|article|paper)\s+title\s*(?::|is)?\s+)", q, re.I, ) quoted = [part[1:-1] for i, part in enumerate(_quoted_search_parts(q)) if i % 2] if cue: if quoted: return quoted[0] return _strip_date_instructions(q[cue.end():]).strip(" :?.,") if quoted and re.search(r"\b(?:text\s?books?|e[- ]?books?|books?|articles?|papers?)\s+[\"“\']", q, re.I): return quoted[0] return "" def _strip_date_instructions(question: str) -> str: parts = _quoted_search_parts(question) for i in range(0, len(parts), 2): for _, pattern in _DATE_PATTERNS: parts[i] = pattern.sub(" ", parts[i]) parts[i] = re.sub(r"\b(?:peer[- ]reviewed|open[- ]access)\b", " ", parts[i], flags=re.I) parts[i] = re.sub(r"\b(?:published|publication dates?|publication years?)\s*$", " ", parts[i], flags=re.I) return _normalize_whitespace("".join(parts)) def _extract_topic_regex(question: str) -> str: exact_title = _title_search_value(question) if exact_title: return f'"{exact_title}"' q = _strip_date_instructions(_normalize_whitespace(question)) identifier = _isbn_search_value(q) if identifier: return identifier q = re.sub(r"^\s*(?:does|do)\s+(?:(?:the|our)\s+)?(?:(?:KU|Khalifa University)\s+)?library\s+have\s+", "", q, flags=re.I) q = re.sub(r"\s+(?:available\s+)?(?:in|at|from)\s+(?:(?:the|our)\s+)?(?:(?:KU|Khalifa University)\s+)?library\b", "", q, flags=re.I) q = _light_strip_retrieval_boilerplate(q) # Summary/research framing is stripped once; phrase contents remain intact. q = re.sub( r"^(?:give me\s+)?(?:a\s+)?(?:deep analysis|brief overview|brief summary|" r"summary|overview|analysis|literature review|research overview|research summary)\s+(?:of|on|about)\s+|" r"^(?:summari[sz]e|analy[sz]e|explain|describe)\s+|^(?:what is|what are)\s+", "", q, flags=re.I, ) q = _light_strip_retrieval_boilerplate(q) q = re.sub(r"^\s*(?:on|about|for)\s+", "", q, flags=re.I) q = re.sub(r"\s+(?:please|thanks|thank you)[.!?]?$", "", q, flags=re.I) material = r"(?:text\s?books?|e[- ]?books?|books?|articles?|papers?|journals?)" parts = _quoted_search_parts(q) if parts: parts[-1] = re.sub(rf"[,;]?\s+\b(?:not|no|without|exclud(?:e|ing))\s+(?:any\s+)?{material}[.!?]?$", "", parts[-1], flags=re.I) q = "".join(parts).strip(".?, ") # A trailing material noun is a request in "Electrical Engineering textbooks". # If a material+on/about phrase already specified scope, the trailing noun may # be the subject ("articles about Electrical Engineering textbooks"). explicit_scope = re.search(rf"\b{material}\s+(?:on|about|for)\b", _mask_quoted_search_text(question), re.I) if not explicit_scope: trailing = re.fullmatch(rf"(.+?)\s+{material}(?:\s+only)?", q, re.I) if trailing and not re.search(r"\b(?:of|on|about|for)\s*$", trailing.group(1), re.I): q = trailing.group(1) return _normalize_whitespace(q).strip(".?, ") async def _extract_topic(question: str, model: str) -> str: """Extract display keywords locally; the search planner handles spelling. A separate model call here delayed every search and could discard named entities through its former 2–7-keyword limit. """ 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.debug("Boolean search plan prepared") return result except Exception as e: logger.warning(f"LLM boolean query failed: {type(e).__name__} — using regex fallback") return _shared_build_primo_boolean_query(topic) def _derive_resource_type(question: str, existing: Optional[str] = None) -> str: q = _mask_quoted_search_text(question or "") book_word = r"(?:text\s?books?|e[- ]?books?|books?)" article_word = r"(?:articles?|papers?|stud(?:y|ies)|journals?)" books = bool(re.search(rf"\b{book_word}\b", q, re.I)) articles = bool(re.search(rf"\b{article_word}\b", q, re.I)) scope = re.search(rf"\b({book_word}|{article_word})(?:\s+(?:and|or)\s+({book_word}|{article_word}))?\s+(?:on|about|for)\b", q, re.I) if scope: scope_words = " ".join(value for value in scope.groups() if value) books = bool(re.search(rf"\b{book_word}\b", scope_words, re.I)) articles = bool(re.search(rf"\b{article_word}\b", scope_words, re.I)) if not articles and _isbn_search_value(question): books = True if re.search(rf"\b(?:no|not|without|exclude)\s+{article_word}\b|\b{book_word}\s+only\b|\bonly\s+{book_word}\b", q, re.I): articles = False if re.search(rf"\b(?:no|not|without|exclude)\s+{book_word}\b|\b{article_word}\s+only\b|\bonly\s+{article_word}\b", q, re.I): books = False if books and articles: return "both" if books: return "books" if articles: return "articles" return existing or "articles" def _parse_year_filters(question: str) -> Tuple[Optional[str], Optional[str]]: # Explicit publication dates win over "recent" or relative windows. Quoted # titles and phrases (including year-like titles) are never date instructions. q = _mask_quoted_search_text(question or "") current_year = datetime.now().year for kind, pattern in _DATE_PATTERNS: match = pattern.search(q) if not match: continue if kind == "range": return tuple(sorted((match.group(1), match.group(2)))) if kind == "since": start = int(match.group(2)) + (match.group(1).lower() == "after") return str(start), str(max(start, current_year)) if kind == "before": end = int(match.group(2)) - (match.group(1).lower() == "before") return None, str(end) if kind == "single": return match.group(1), match.group(1) if kind == "relative": return str(current_year - int(match.group(1)) + 1), str(current_year) if kind == "decade": return str(current_year - 9), str(current_year) if kind == "this_year": return str(current_year), str(current_year) if kind == "last_year": return str(current_year - 1), str(current_year - 1) if kind == "recent": return str(current_year - 4), str(current_year) return None, None def _usable_boolean_plan(value: str, subject: str) -> bool: """Reject model request scaffolding or broken Boolean syntax deterministically.""" if not value or len(value) > 1500 or value.count('"') % 2 or ";" in value: return False unquoted = _mask_quoted_search_text(value) depth = 0 for char in unquoted: depth += (char == "(") - (char == ")") if depth < 0: return False if depth: return False if re.search(r"\b(?:i want|i need|give me|show me|can you|find me|search for|textbooks? on|books? on|articles? on)\b", value, re.I): return False syntax = "".join(" TERM " if i % 2 else part for i, part in enumerate(_quoted_search_parts(value))) if re.search(r"\b(?:AND|OR|NOT)\s*$|^\s*(?:AND|OR)\b|\b(?:AND|OR)\s+(?:AND|OR)\b", syntax): return False for term in ("textbook", "textbooks", "ebook", "ebooks", "book", "books", "article", "articles"): if re.search(rf"\b{term}\b", value, re.I) and not re.search(rf"\b{term}\b", subject, re.I): return False # Date instructions belong in facets; actual subject/title years are retained. years = set(re.findall(r"\b(?:19|20)\d{2}\b", value)) if years - set(re.findall(r"\b(?:19|20)\d{2}\b", subject)): return False return True 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") elif context.year_to: filters.append(f"published through {context.year_to}") 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\": \"\"}\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: {type(exc).__name__} — failing open") return True, topic async def _build_search_plan(question: str, model: str) -> dict: settings = get_settings() raw_query = _normalize_whitespace(question or "") subject = _extract_topic_regex(raw_query) light_query = _light_strip_retrieval_boilerplate(raw_query) or raw_query identifier = _isbn_search_value(subject) exact_title = _title_search_value(raw_query) year_from, year_to = _parse_year_filters(raw_query) plan = { "corrected": raw_query, "natural": light_query, "topic": subject, "boolean": _shared_build_primo_boolean_query(subject), "year_from": year_from or "", "year_to": year_to or "", "peer_reviewed": bool(re.search(r"\bpeer[- ]reviewed\b", _mask_quoted_search_text(raw_query), re.I)), "open_access": bool(re.search(r"\bopen[- ]access\b", _mask_quoted_search_text(raw_query), re.I)), "sort": "date_d" if _has_recent_instruction(raw_query) else "rank", "query_field": "title" if exact_title else "any", } # Identifiers and explicitly named titles require literal matching, not model # expansion; a title like "1984" must not trigger the numeric-topic guard. if identifier or exact_title: plan["boolean"] = f'"{identifier or exact_title}"' plan["topic"] = identifier or exact_title plan["database_query"] = _clean_database_keywords(plan["boolean"]) return plan stripped = subject.strip() if len(stripped) < 2 or re.fullmatch(r"[\d\s.,\-/]+", stripped): return {**plan, "clarification_needed": True, "clarification_message": ( f"Could you clarify the subject you want to search for" f"{(' (' + _escape(stripped) + ')') if stripped else ''}? " "A topic, book title, or ISBN will help me find the right results." )} if settings.openai_api_key or settings.anthropic_api_key: try: llm = _get_llm(model, temperature=0, max_tokens=400) response = await llm.ainvoke([ {"role": "system", "content": _BUILD_SEARCH_PLAN_SYSTEM + ( "\nReturn valid:false ONLY for clearly meaningless random characters; " "otherwise valid:true. Recognisable topics, names and acronyms are valid. " "Current year: " + str(datetime.now().year) )}, {"role": "user", "content": json.dumps({"request": raw_query, "subject": subject})}, ]) raw = response.content.strip() start, end = raw.find("{"), raw.rfind("}") result = json.loads(raw[start:end + 1]) if start != -1 and end > start else None if not isinstance(result, dict): raise ValueError("Search plan must be an object") if result.get("valid") is False: return {**plan, "clarification_needed": True, "clarification_message": ( f"Could you check the spelling of {_escape(stripped)} " "or describe the research topic in a little more detail?" )} corrected = str(result.get("corrected") or raw_query).strip() corrected_subject = _extract_topic_regex(str(result.get("topic") or corrected)) raw_boolean = str(result.get("boolean") or "").strip() boolean = _sanitize_boolean_for_primo(raw_boolean) if _usable_boolean_plan(raw_boolean, subject) and _usable_boolean_plan(boolean, subject): plan["boolean"] = boolean # Labels must be subject-only too. Re-clean all model text. if corrected_subject and len(corrected_subject) <= 500: plan["topic"] = corrected_subject else: logger.warning("Rejected malformed or request-contaminated Boolean plan; using subject fallback") plan["corrected"] = corrected or raw_query plan["natural"] = str(result.get("natural") or light_query).strip()[:1000] # Only deterministic, explicit user instructions set filters. Model # guesses (e.g. the title 1984) must never invent publication dates. except Exception as exc: logger.warning("_build_search_plan LLM failed: %s — using subject fallback", type(exc).__name__) plan["database_query"] = _clean_database_keywords(plan["boolean"]) return plan async def _prepare_queries(question: str, context: SearchContextPayload, model: str, is_follow_up: bool) -> SearchContextPayload: # Filters do not need another model call or a rebuilt Boolean query. if is_follow_up: context.resource_type = _derive_resource_type(question, context.resource_type) if _has_recent_instruction(question): context.sort = "date_d" yf, yt = _parse_year_filters(question) if yf or yt: context.year_from, context.year_to = yf, yt if re.search(r"\bpeer[- ]reviewed\b", question, re.I): context.peer_reviewed = True if re.search(r"\bopen[- ]access\b", question, re.I): context.open_access = True if context.primo_boolean_query: context.ai_tool_query = _compose_ai_tool_query(question, context, True) return context elif context.primo_boolean_query and context.ai_tool_query: return context topic = context.display_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"] context.topic = plan.get("topic") or topic context.display_topic = context.topic context.query_field = plan.get("query_field", "any") if not is_follow_up: context.resource_type = _derive_resource_type(question, context.resource_type) context.sort = plan.get("sort", "rank") 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 asyncio.wait_for(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
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}"}, ]), timeout=4.0) intro = response.content.strip() return intro if intro else "" except Exception as e: logger.warning(f"_topic_intro failed: {type(e).__name__}") 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'' f'[{_escape(paper["citation_key"])}]' ) statements.append(f"{_escape(lead)} {citation}") if not statements: return f"Retrieved literature on {_escape(topic)} is available in the cited papers below." intro = f"Retrieved literature on {_escape(topic)}:

" return intro + "

".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'LeapSpace · ' f'Scopus AI · ' f'EBSCO AI · ' f'PRIMO AI Assistant · ' f'Consensus' ) more_results = f'PRIMO Library Discovery' if context.intent == "search_medical" and urls.get("pubmed"): more_results += f' · PubMed' return ( f"For deeper work on {topic}, continue with these AI research tools: {ai_tools}. " f"For more results, use {more_results}. Medical topics are strongest in PubMed, Embase, and CINAHL." ) return ( f"For deeper work on {topic}, continue with these AI research tools: {ai_tools}. " f"For more results, use {more_results}." ) async def _search_strategy_answer(question: str, model: str) -> Tuple[str, dict]: plan = await _build_search_plan(question, model) if plan.get("clarification_needed"): return plan.get("clarification_message", "Please describe the research topic."), plan topic = _escape(_light_strip_retrieval_boilerplate(question) or question) boolean = _escape(plan.get("boolean", "")) natural = _escape(plan.get("natural", "")) explanation = ( f"🔍 Search strategy for: {topic}

" f"1. Identify your key concepts
" f"Break the topic into 2–3 core concepts. For your question, I identified these concepts " f"and suggested synonyms for each using OR, then joined concepts with AND:

" f'{boolean}
' f"Why AND and OR? AND narrows — both concepts must appear. " f"OR broadens — any synonym counts. Quoting phrases like " f"\"machine learning\" keeps them together as an exact phrase.

" f"2. Apply filters
" f"Use database facets to limit by: peer-reviewed, date range, document type (article / review / book), " f"language, or open access. In PRIMO, these appear in the left sidebar after you search.

" f"3. Natural language query for AI tools
" f"For AI-powered tools (LeapSpace, Scopus AI, Consensus), use a conversational query:
" f'{natural}

' f"4. Iterate
" f"Too many results → add more AND terms or apply filters. " f"Too few → remove an AND group or use broader synonyms. " f"You can ask me to refine: \"narrow to peer-reviewed only\", " f"\"limit to last 5 years\", or \"suggest alternative keywords\"." ) return explanation, plan 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"Alternative search terms for: {_escape(topic)}

" "Try combining these broader, narrower, and related terms:
" "• Use broader terms if too few results
" "• Use narrower/specific terms if too many results
" "• Try acronyms and full forms (e.g. AI / Artificial Intelligence)
" "• Include British and American spellings (e.g. organisation/organization)" ) try: llm = _get_llm(model, temperature=0.4, max_tokens=300) response = await llm.ainvoke([ {"role": "system", "content": ( "You are an academic librarian. Given a research topic, generate a structured list of alternative " "search terms. Group them as: Broader terms, Narrower/specific terms, Related concepts, Acronyms/abbreviations. " "Format as HTML using for group labels and
for line breaks. " "Keep it concise — 3-4 terms per group maximum. No bullet points, use • instead." )}, {"role": "user", "content": f"Research topic: {topic}"}, ]) terms = response.content.strip() return ( f"💡 Alternative search terms for: {_escape(topic)}

" f"{terms}

" f"Tip: In PRIMO or databases, use OR between synonyms within a concept group, " f"and AND between different concept groups. Ask me to run a new search with any of these." ) except Exception as e: logger.warning(f"_alternative_terms_answer failed: {type(e).__name__}") return f"Alternative terms for {_escape(topic)}

Try synonyms, acronyms, broader/narrower terms, and related concepts in your search." def _citation_chain_answer() -> str: return ( "🔗 How to trace citations forward and backward from a paper

" "Backward citation (who does this paper cite?)
" "Read the paper's reference list — every source it cites is a potential lead. " "This gives you foundational and seminal works on the topic.

" "Forward citation (who has cited this paper since publication?)
" "Use these tools — paste in the DOI or title:
" '• Scopus ' '— search the article, then click "Cited by N documents"
' '• Web of Science ' '— search the article, click "Times Cited"
' '• Semantic Scholar ' '— free, excellent for CS and engineering, shows "Citations" tab
' '• OpenAlex ' '— fully open, API-accessible citation graph

' "Lateral search (similar papers)
" "In PRIMO, use \"Find Similar\". In Semantic Scholar, use \"Recommended Papers\". " "In Scopus, use \"Related Documents\".

" "Tip: Start with one highly cited foundational paper, trace forward to find the newest " "work, and backward to understand the theoretical roots." ) def _predatory_eval_answer() -> str: return ( "✅ How to tell if an article or journal is peer-reviewed, scholarly, or predatory

" "Is it peer-reviewed?
" "• Check the journal's website for a peer-review statement or editorial process description
" "• In PRIMO, tick the Peer-reviewed filter in the left sidebar
" "• Check if the journal is indexed in " 'Scopus or ' 'Web of Science — ' "indexed = generally peer-reviewed

" "Is it a legitimate journal?
" "• DOAJ " "— Directory of Open Access Journals (vetted, legitimate OA)
" "• Web of Science Master Journal List
" "• Scopus Source List
" "• Think · Check · Submit: thinkchecksubmit.org " "— a checklist to assess any journal

" "Warning signs of predatory journals
" "• Unsolicited email invitation to submit
" "• No clear peer-review process or very fast acceptance (days)
" "• High article processing charges (APCs) with no clear metrics
" "• Not indexed in Scopus or Web of Science
" "• Generic or misleading journal name ('International Journal of...')

" "Need help checking a specific journal? Ask our E-Resources Librarian: " "Rani Anand · rani.anand@ku.ac.ae" ) def _highly_cited_note(topic_escaped: str) -> str: return ( f"

📊 Finding highly cited papers on {topic_escaped}
" "PRIMO doesn't sort by citation count, but these tools do:
" '• Scopus ' '— search your topic → Sort by Cited by (highest)
' '• Web of Science ' '— search → Sort by Times Cited
' '• Semantic Scholar ' '— free, sort by Citation Count
' '• Google Scholar ' '— sort by Cited by (broader but includes grey literature)' ) async def _research_snapshot(context: SearchContextPayload, model: str) -> Tuple[str, List[dict], List[dict]]: """Research starting points with real PRIMO hits for catalogue searches.""" if context.clarification_needed: return context.clarification_message or "Please describe the research topic.", [], [] topic = context.topic source_name = "PRIMO" if context.source == "primo" else "PubMed" source_url = (_primo_clean_url(context) if context.source == "primo" else _shared_build_pubmed_url(topic, context.year_from, context.year_to, context.peer_reviewed, context.open_access)) answer = _search_trace_block(f"Research on {topic}", context) answer += ( f"📚 Research starting points: {_escape(topic)}

" "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." ) if context.source == "primo": answer += f'

PRIMO query:
{_escape(context.primo_boolean_query or "")}' answer += (f'

Search {source_name} for: {_escape(topic)} →') # Catalogue hits use the same prepared query as the link immediately above. if context.source == "primo": papers, panel_html = await fetch_primo_panel(context) else: papers, panel_html = await fetch_evidence_panel(topic, context=context) answer += panel_html answer += _ai_tools_footer(context) sources = [ {"title": paper.get("title", ""), "url": paper.get("primo_url") or paper.get("link", ""), "doi": paper.get("doi", ""), "provider": paper.get("_source", "")} for paper in papers ] return answer, papers, sources 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}") elif context.year_to: filters.append(f"through {context.year_to}") 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: cleaned = re.sub( r"\b(peer[- ]reviewed|open[- ]access|(?:last|past) \d+ years?|past decade|" r"summari[sz]e(?: this topic)?|overview|brief|use pubmed|search pubmed|" r"search primo|best databases?|deep research tools?)\b", " ", question or "", flags=re.I, ) cleaned = _strip_date_instructions(cleaned) extra_stop = {"can", "could", "would", "you", "i", "want", "need", "also", "just", "only", "now", "again", "please", "in", "between", "since", "after", "before", "through", "until", "during", "published"} def tokens(text): return {t for t in re.findall(r"[^\W_]+", text.lower()) if t not in REFINEMENT_STOP_WORDS | extra_stop and len(t) > 1 and not t.isdigit()} return bool(tokens(cleaned) - tokens(base_topic or "")) 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" yf, yt = _parse_year_filters(q) if yf or yt: return f"years_{yf or 'any'}_{yt or 'any'}" if re.search(r"\b(?:text\s?books?|e[- ]?books?|books?)\s+(?:instead|only)\b|\b(?:find|show|only)\s+(?:text\s?books?|e[- ]?books?|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 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: current_year = datetime.now().year 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]) if not 1 <= n <= 99: return context context.year_from = str(current_year - n + 1) context.year_to = str(current_year) except (IndexError, ValueError): pass elif action and re.fullmatch(r"years_(?:\d{4}|any)_(?:\d{4}|any)", action): _, yf, yt = action.split("_") context.year_from = None if yf == "any" else yf context.year_to = None if yt == "any" else yt 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": # A review article and a peer-reviewed article are different concepts. # Legacy action: retain the search without silently inventing a filter. context.resource_type = "articles" 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 asyncio.wait_for(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}"}, ]), timeout=4.0) 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: {type(e).__name__}") 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 {_escape(topic)}. " "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 prepared search on {topic} for the {location} and {resource}." else: action = f"I prepared a search for {resource} on {topic}." return action async def _run_search_mode( question: str, context: SearchContextPayload, model: str, is_follow_up: bool ) -> Tuple[str, List[dict], List[dict], str]: """Prepared query, matching source link, and bounded genuine catalogue hits.""" context = await _prepare_queries(question, context, model, is_follow_up) # Clarification gate if context.clarification_needed and context.clarification_message: return context.clarification_message, [], [], "" source_name = "PubMed" if context.source == "pubmed" else "PRIMO Library Discovery" source_url = ( _shared_build_pubmed_url(context.display_topic or context.topic, context.year_from, context.year_to, context.peer_reviewed, context.open_access) if context.source == "pubmed" else _primo_clean_url(context) ) topic = _escape(context.display_topic or context.topic) # PRIMO hits replace the article panel on catalogue searches. Run the # optional topic introduction concurrently; book searches need no intro call. async def optional_intro(): return "" if context.resource_type == "books" else await _topic_intro(context.display_topic or context.topic, model) panel_call = (fetch_primo_panel(context) if context.source == "primo" else fetch_evidence_panel(context.display_topic or context.topic, context=context)) intro, (papers, panel_html) = await asyncio.gather(optional_intro(), panel_call) answer = _search_trace_block(question, context) if intro: answer += f'
{intro}
' answer += _search_answer_intro(context, is_follow_up) if context.source == "primo": answer += f'

PRIMO query:
{_escape(context.primo_boolean_query or "")}' answer += ( f'

' f'' f'🔍 Search {source_name} for: {topic} →' ) # Keep this block adjacent to the primary search link: no tools, article # recommendations or filter guidance inserted between link and records. answer += panel_html if context.source == "pubmed": answer += '

Inside PubMed you can filter by publication date, article type, and free full text.

' if context.peer_reviewed: answer += '

PubMed has no peer-review filter. Use PRIMO’s Peer Reviewed facet and verify the publication with its journal.

' answer += _ai_tools_footer(context) sources = [ {"title": paper.get("title", ""), "url": paper.get("primo_url") or paper.get("link", ""), "doi": paper.get("doi", ""), "provider": paper.get("_source", "")} for paper in papers ] return answer, papers, sources, source_url def _sort_and_trim_contexts(contexts: List[SearchContextPayload]) -> List[SearchContextPayload]: dedup: Dict[str, SearchContextPayload] = {} for ctx in contexts: dedup[ctx.context_id] = ctx ordered = sorted(dedup.values(), key=lambda c: c.created_at, reverse=True) return ordered[:5]