| """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) |
| |
| 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(".?, ") |
| |
| |
| |
| 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]]: |
| |
| |
| 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 |
| |
| 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\": \"<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: {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", |
| } |
| |
| |
| 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 <strong>{_escape(stripped)}</strong> " |
| "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 |
| |
| 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] |
| |
| |
| 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: |
| |
| 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 <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}"}, |
| ]), 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'<a href="{_escape(paper.get("link", ""))}" target="_blank" rel="noopener noreferrer" ' |
| 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"Retrieved literature on <strong>{_escape(topic)}</strong> is available in the cited papers below." |
| intro = f"Retrieved 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) |
| 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"<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: {type(e).__name__}") |
| 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://go.openathens.net/redirector/ku.ac.ae?url=https%3A%2F%2Fwww.scopus.com%2Fpages%2Fai" 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://go.openathens.net/redirector/ku.ac.ae?url=https%3A%2F%2Fwww.scopus.com" 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://go.openathens.net/redirector/ku.ac.ae?url=https%3A%2F%2Fwww.scopus.com" 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 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"<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." |
| ) |
| if context.source == "primo": |
| answer += f'<br><br><strong>PRIMO query:</strong><br><code>{_escape(context.primo_boolean_query or "")}</code>' |
| answer += (f'<br><br><a href="{_escape(source_url)}" target="_blank" rel="noopener noreferrer" ' |
| f'style="color:#003366;font-weight:700">Search {source_name} for: {_escape(topic)} →</a>') |
|
|
| |
| 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": |
| |
| |
| 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 <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 prepared search on <strong>{topic}</strong> for the <strong>{location}</strong> and <strong>{resource}</strong>." |
| else: |
| action = f"I prepared a search 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]: |
| """Prepared query, matching source link, and bounded genuine catalogue hits.""" |
| context = await _prepare_queries(question, context, model, is_follow_up) |
|
|
| |
| 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) |
|
|
| |
| |
| 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'<div style="margin-bottom:12px;color:#374151;font-size:.88rem;line-height:1.7">{intro}</div>' |
| answer += _search_answer_intro(context, is_follow_up) |
| if context.source == "primo": |
| answer += f'<br><br><strong>PRIMO query:</strong><br><code>{_escape(context.primo_boolean_query or "")}</code>' |
| answer += ( |
| f'<br><br>' |
| f'<a href="{_escape(source_url)}" target="_blank" rel="noopener noreferrer" ' |
| 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 {source_name} for: {topic} →</a>' |
| ) |
| |
| |
| answer += panel_html |
| if context.source == "pubmed": |
| answer += '<p>Inside PubMed you can filter by publication date, article type, and free full text.</p>' |
| if context.peer_reviewed: |
| answer += '<p>PubMed has no peer-review filter. Use PRIMO’s Peer Reviewed facet and verify the publication with its journal.</p>' |
| 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] |
|
|
|
|
