"""Source-aware retrieval for the ResearchBee chat assistant. The curated general knowledge lives in ``knowledge_base.json``. KU-specific policy and service content remains in ``ku_knowledge.py`` and is split into sections at runtime. Only the sections relevant to the latest user question are placed in the model context. This is intentionally a small, dependency-free hybrid lexical retriever. It is fast enough for a Hugging Face Space, transparent to librarians, and avoids sending the entire knowledge base with every chat request. It can later be replaced by a vector or full-text index without changing the public interface. """ from __future__ import annotations import json import math import re import calendar import unicodedata from dataclasses import dataclass from datetime import date, datetime, timedelta, timezone from functools import lru_cache from pathlib import Path from typing import Iterable, Sequence from ku_knowledge import ( KU_OA_KNOWLEDGE, KU_KNOWLEDGE_EXTRACTED, KU_KNOWLEDGE_REVIEW_CYCLE_DAYS, KU_KNOWLEDGE_SOURCE, ) ROOT = Path(__file__).resolve().parent KNOWLEDGE_PATH = ROOT / "knowledge_base.json" _TOKEN_RE = re.compile(r"[^\W_]+(?:\.[^\W_]+)?", re.UNICODE) _ARABIC_FOLD = str.maketrans({"أ": "ا", "إ": "ا", "آ": "ا", "ٱ": "ا", "ى": "ي", "ـ": ""}) # Retrieval translations only: these terms do not add policy facts. Phrase # matching handles Arabic articles and the common publisher transliterations. _ARABIC_TERMS: tuple[tuple[tuple[str, ...], tuple[str, ...]], ...] = ( (("جامعة خليفة", "جامعه خليفه", "خليفة", "خليفه"), ("ku", "khalifa", "university")), (("مكتبة", "المكتبة", "مكتبه", "المكتبه"), ("library",)), (("خزنة", "خزنه", "خزنا"), ("khazna", "repository")), (("رسوم النشر", "تكاليف النشر", "تمويل النشر", "دعم النشر", "رسوم معالجة المقالات", "رسوم", "تمويل"), ("apc", "fee", "funding")), (("اتفاقية", "اتفاقيات", "الاتفاقية", "الاتفاقيات"), ("agreement",)), (("الوصول المفتوح", "الوصول الحر", "النفاذ المفتوح", "النشر المفتوح"), ("open", "access")), (("مجلة", "مجلات", "المجلة", "المجلات", "دورية", "دوريات"), ("journal",)), (("اختيار", "اختار", "مناسبة", "مناسب"), ("suitable", "match")), (("السفير", "السيفير", "السيفر", "الزفير"), ("elsevier",)), (("وايلي", "ويلي"), ("wiley",)), (("سبرينجر", "سبرنجر", "شبرنجر", "سبرينغر"), ("springer",)), (("تايلور", "تيلور", "فرانسيس"), ("taylor",)), (("الكيميائية الامريكية", "الكيمياء الامريكية"), ("acs",)), (("الربع", "ربع", "تصنيف", "ترتيب", "التصنيف", "الترتيب"), ("quartile", "ranking")), (("سايت سكور", "سايتسكور"), ("citescore",)), (("معامل التاثير",), ("jif", "impact")), (("سكوبس", "سكوبوس"), ("scopus",)), (("مستودع", "مستودعات", "المستودع", "المستودعات", "ايداع", "الايداع"), ("repository", "deposit")), (("مؤسسي", "المؤسسي", "مؤسسية", "المؤسسية"), ("institutional",)), (("النسخة المقبولة", "المخطوطة المقبولة", "نسخة مقبولة", "المقالة المقبولة"), ("aam", "selfarchiving")), (("حظر", "الحظر", "الحجب", "حجب", "فترة الانتظار"), ("embargo",)), (("ترخيص", "الترخيص", "رخصة", "الرخصة"), ("licence",)), (("بيانات", "البيانات"), ("data", "dataset")), (("حساسة", "الحساسة", "شخصية", "الشخصية"), ("sensitive",)), (("خطة ادارة البيانات",), ("dmp", "rdm")), (("ذكاء اصطناعي", "الذكاء الاصطناعي"), ("ai",)), ) _SECTION_RE = re.compile( r"(?:^|\n)[━]{8,}\n(?P[^\n]+)\n[━]{8,}\n", re.MULTILINE, ) _STOP_WORDS = { "a", "about", "all", "also", "am", "an", "and", "any", "are", "as", "at", "be", "been", "before", "but", "by", "can", "could", "do", "does", "for", "from", "give", "help", "how", "i", "in", "is", "it", "me", "my", "of", "on", "or", "please", "should", "tell", "that", "the", "their", "there", "these", "this", "to", "use", "want", "what", "when", "where", "which", "who", "why", "with", "would", "you", } # Query terms that users commonly treat as interchangeable. Expansion is only # used for retrieval; it is never shown to the model as a factual assertion. _SYNONYM_GROUPS: tuple[tuple[str, ...], ...] = ( ("journal", "journals", "publication", "venue", "publish"), ("best", "suitable", "appropriate", "right", "match", "shortlist"), ("ranking", "rank", "metric", "metrics", "quartile", "percentile"), ("impact", "jif", "impactfactor", "citescore", "sjr", "snip", "jci"), ("repository", "repositories", "archive", "deposit"), ("dataset", "data", "rdm"), ("selfarchive", "selfarchiving", "green", "aam", "postprint", "embargo"), ("trusted", "trustworthy", "legitimate", "predatory", "scam"), ("apc", "fee", "funding", "agreement", "waiver", "discount"), ("ku", "khalifa", "university"), ("fair", "findable", "accessible", "interoperable", "reusable"), ("author", "authorship", "contributor", "credit", "orcid"), ) _CATEGORY_HINTS: dict[str, set[str]] = { "journals": {"journal", "journals", "publish", "venue", "scope", "shortlist"}, "metrics": {"metric", "ranking", "rank", "quartile", "percentile", "citescore", "jif", "jcr", "sjr", "snip"}, "integrity": {"predatory", "trusted", "trustworthy", "scam", "ethics", "peer", "editorial"}, "publishing": {"manuscript", "submission", "cover", "reviewer", "revision", "authorship", "orcid"}, "open-access": {"open", "access", "apc", "embargo", "aam", "preprint", "selfarchive", "licence", "license"}, "rdm": {"data", "dataset", "rdm", "fair", "metadata", "dmp", "sensitive"}, "repositories": {"repository", "repositories", "deposit", "zenodo", "dryad", "figshare", "khazna"}, "ai": {"ai", "chatgpt", "generative", "hallucination"}, "analytics": {"scopus", "scival", "citation", "analytics", "benchmarking", "fwci"}, "researchbee": {"researchbee", "service", "services", "tool", "tools", "chat"}, "ku": {"ku", "khalifa", "library"}, } @dataclass(frozen=True) class KnowledgeEntry: id: str category: str title: str summary: str topics: tuple[str, ...] sources: tuple[tuple[str, str], ...] authority: str reviewed: str review_cycle_days: int @dataclass(frozen=True) class RankedEntry: score: float entry: KnowledgeEntry def _fold_token(token: str) -> str: return token.lower().replace("-", "").replace("_", "") def _normalise_text(text: str) -> str: normalised = unicodedata.normalize("NFKC", text).casefold().translate(_ARABIC_FOLD) return "".join(ch for ch in normalised if unicodedata.category(ch) != "Mn") def _tokens(text: str) -> set[str]: normalised = _normalise_text(text) found = {_fold_token(m.group(0)) for m in _TOKEN_RE.finditer(normalised)} for phrases, equivalents in _ARABIC_TERMS: if any(re.search(r"(?<!\w)[وبكل]?" + re.escape(phrase) + r"(?!\w)", normalised) for phrase in phrases): found.update(equivalents) found = {t for t in found if t and t not in _STOP_WORDS} expanded = set(found) for group in _SYNONYM_GROUPS: folded = {_fold_token(item) for item in group} if found & folded: expanded.update(folded) return expanded @lru_cache(maxsize=1) def load_entries() -> tuple[KnowledgeEntry, ...]: """Load and validate the curated JSON knowledge base once per process.""" raw = json.loads(KNOWLEDGE_PATH.read_text(encoding="utf-8")) entries: list[KnowledgeEntry] = [] seen: set[str] = set() for item in raw.get("entries", []): entry_id = str(item.get("id", "")).strip() if not entry_id or entry_id in seen: raise ValueError(f"Invalid or duplicate knowledge entry id: {entry_id!r}") seen.add(entry_id) sources: list[tuple[str, str]] = [] for source in item.get("sources", []): name = str(source.get("name", "")).strip() url = str(source.get("url", "")).strip() if name and url.startswith(("https://", "http://")): sources.append((name, url)) entries.append( KnowledgeEntry( id=entry_id, category=str(item.get("category", "general")).strip() or "general", title=str(item.get("title", "")).strip(), summary=str(item.get("summary", "")).strip(), topics=tuple(str(x).strip() for x in item.get("topics", []) if str(x).strip()), sources=tuple(sources), authority=str(item.get("authority", "")).strip(), reviewed=str(item.get("reviewed", "")).strip(), review_cycle_days=max(1, int(item.get("review_cycle_days", 180))), ) ) if not entries: raise ValueError("knowledge_base.json contains no entries") return tuple(entries) @lru_cache(maxsize=1) def _document_frequency() -> dict[str, int]: frequency: dict[str, int] = {} for entry in load_entries(): doc_tokens = _tokens(" ".join((entry.title, entry.summary, " ".join(entry.topics)))) for token in doc_tokens: frequency[token] = frequency.get(token, 0) + 1 return frequency def _entry_score(query: str, query_tokens: set[str], entry: KnowledgeEntry) -> float: title_tokens = _tokens(entry.title) topic_text = " ".join(entry.topics) topic_tokens = _tokens(topic_text) summary_tokens = _tokens(entry.summary) category_tokens = _tokens(entry.category) df = _document_frequency() n_docs = len(load_entries()) score = 0.0 for token in query_tokens: idf = math.log((n_docs + 1) / (df.get(token, 0) + 1)) + 1.0 if token in title_tokens: score += 5.0 * idf if token in topic_tokens: score += 4.0 * idf if token in category_tokens: score += 2.5 * idf if token in summary_tokens: score += 1.2 * idf query_lower = query.lower().strip() for phrase in entry.topics: phrase_lower = phrase.lower() if len(phrase_lower) >= 4 and phrase_lower in query_lower: score += 10.0 + min(5.0, len(phrase_lower) / 10.0) hints = _CATEGORY_HINTS.get(entry.category, set()) if query_tokens & hints: score += 3.0 # These two records are useful only when directly relevant; avoid allowing # generic words such as "help" to crowd out substantive content. if entry.id == "researchbee-services" and not ( {"researchbee", "service", "services", "tool", "tools", "chat"} & query_tokens ): score *= 0.35 if entry.id == "ku-first-party-guides" and not ( {"ku", "khalifa", "library", "guide", "guides"} & query_tokens ): score *= 0.35 return score def retrieve_general_entries(query: str, max_entries: int = 6) -> tuple[RankedEntry, ...]: query_tokens = _tokens(query) if not query_tokens: default_ids = {"researchbee-services", "journal-selection-workflow", "repository-selection-workflow"} return tuple( RankedEntry(1.0, entry) for entry in load_entries() if entry.id in default_ids )[:max_entries] ranked = sorted( (RankedEntry(_entry_score(query, query_tokens, entry), entry) for entry in load_entries()), key=lambda item: (-item.score, item.entry.id), ) positive = [item for item in ranked if item.score >= 2.0] return tuple(positive[:max_entries]) @lru_cache(maxsize=1) def _ku_sections() -> tuple[tuple[str, str], ...]: matches = list(_SECTION_RE.finditer(KU_OA_KNOWLEDGE)) sections: list[tuple[str, str]] = [] for index, match in enumerate(matches): start = match.start() end = matches[index + 1].start() if index + 1 < len(matches) else len(KU_OA_KNOWLEDGE) title = match.group("title").strip() body = KU_OA_KNOWLEDGE[start:end].strip() sections.append((title, body)) return tuple(sections) def _ku_score(query: str, query_tokens: set[str], title: str, body: str) -> float: title_tokens = _tokens(title) body_tokens = _tokens(body) score = 0.0 for token in query_tokens: if token in title_tokens: score += 5.0 if token in body_tokens: score += 1.0 q = _normalise_text(query) + " " + " ".join(query_tokens) boosts: tuple[tuple[tuple[str, ...], tuple[str, ...]], ...] = ( (("apc", "funding", "pay", "agreement", "tier"), ("FUNDING", "PUBLISHER-SPECIFIC")), (("wiley", "elsevier", "springer", "taylor", "acs", "ieee"), ("PUBLISHER-SPECIFIC",)), (("khazna",), ("KHAZNA",)), (("journal", "citescore", "quartile", "top 15", "rating"), ("JOURNAL", "RATINGS")), (("scopus", "web of science", "scival", "analytics", "h-index", "h index"), ("RESEARCH ANALYTICS",)), (("researchbee", "what can", "service", "tool"), ("RESEARCHBEE TOOLS", "KEY CONTACTS")), (("contact", "librarian", "email", "support"), ("KEY CONTACTS",)), (("open access", "gold", "green", "hybrid"), ("WHAT IS OPEN ACCESS", "KU OPEN ACCESS POLICY")), ) upper_title = title.upper() for triggers, title_fragments in boosts: if any(trigger in q for trigger in triggers) and any(fragment in upper_title for fragment in title_fragments): score += 10.0 return score def retrieve_ku_sections(query: str, max_sections: int = 4) -> tuple[tuple[float, str, str], ...]: query_tokens = _tokens(query) ku_signal = bool({"ku", "khalifa", "apc", "khazna", "library", "researchbee"} & query_tokens) publisher_signal = bool(query_tokens & {"wiley", "elsevier", "springer", "taylor", "acs", "ieee"}) if not ku_signal and not publisher_signal: return () ranked = sorted( ((_ku_score(query, query_tokens, title, body), title, body) for title, body in _ku_sections()), key=lambda item: (-item[0], item[1]), ) selected = [item for item in ranked if item[0] >= 2.0] if not selected: selected = ranked[:2] return tuple(selected[:max_sections]) def _format_sources(sources: Sequence[tuple[str, str]]) -> str: if not sources: return "" return "\n".join(f"- {name}: {url}" for name, url in sources) def _freshness(value: str, cycle_days: int, *, as_of: date | None = None) -> str: """Evaluate the recorded date without inventing a more precise review date.""" today = as_of or datetime.now(timezone.utc).date() try: if re.fullmatch(r"\d{4}-\d{2}", value): year, month = map(int, value.split("-")) recorded = date(year, month, calendar.monthrange(year, month)[1]) else: recorded = date.fromisoformat(value) except (TypeError, ValueError): return "Review date unknown; current time-sensitive claims are unverified." if recorded > today: return "Recorded date is later than the current date; review status is unverified." if today > recorded + timedelta(days=cycle_days): return "Review overdue; historical guidance only. Current policies, fees and eligibility are unverified." return "Within the recorded review interval; this is curated guidance, not a live verification." def _format_entry(item: RankedEntry, *, as_of: date | None = None) -> str: entry = item.entry source_block = _format_sources(entry.sources) return ( f"### {entry.title}\n" f"{entry.summary}\n" f"Authority: {entry.authority or 'Curated guidance'}\n" f"Last reviewed: {entry.reviewed or 'Not recorded'}\n" f"Review status: {_freshness(entry.reviewed, entry.review_cycle_days, as_of=as_of)}\n" f"Sources:\n{source_block or '- No external source recorded'}" ) def get_chat_knowledge(query: str, *, context_query: str = "", as_of: date | None = None, max_entries: int = 6, max_ku_sections: int = 4, max_chars: int = 16000) -> str: """Return a compact, source-bearing context for one user question.""" retrieval_query = f"{query} {context_query[:1000]}".strip() if context_query else query general = retrieve_general_entries(retrieval_query, max_entries=max_entries) ku = retrieve_ku_sections(retrieval_query, max_sections=max_ku_sections) blocks = [ "CURATED KNOWLEDGE RETRIEVED FOR THIS QUESTION", ( "Use these records as guidance, not as permission to invent missing facts. " "For changing policies, fees, agreements, metrics, or repository terms, prefer a live " "source result when one is supplied and tell the user what still needs verification." ), ] # Put KU-specific provenance before section bodies and generic guidance so # truncation cannot remove the snapshot age while retaining funding claims. if ku: blocks.append( "\n## Relevant KU Library knowledge\n" f"Source: {KU_KNOWLEDGE_SOURCE}\n" "Snapshot extracted: May 2026 (month only; no subsequent policy review recorded).\n" f"Review interval: {KU_KNOWLEDGE_REVIEW_CYCLE_DAYS} days.\n" f"Review status: {_freshness(KU_KNOWLEDGE_EXTRACTED, KU_KNOWLEDGE_REVIEW_CYCLE_DAYS, as_of=as_of)}\n" "Statements such as 'active' and 'current' below describe the historical snapshot, not live eligibility.\n\n" + "\n\n".join(body for _score, _title, body in ku) ) if general: blocks.append("\n## Curated topic guidance\n" + "\n\n".join(_format_entry(item, as_of=as_of) for item in general)) if not general and not ku: blocks.append( "No specific curated record matched. Give only stable general guidance, say what cannot " "be confirmed, and direct the user to the relevant KU Library service." ) text = "\n\n".join(blocks).strip() if len(text) <= max_chars: return text return text[: max_chars - 120].rstrip() + "\n\n[Context truncated to the configured safety limit.]" def knowledge_status() -> dict[str, object]: """Small status block suitable for health/debug endpoints and tests.""" entries = load_entries() raw = json.loads(KNOWLEDGE_PATH.read_text(encoding="utf-8")) return { "loaded": True, "entries": len(entries), "schema_version": raw.get("schema_version"), "reviewed": raw.get("reviewed"), "source_file": KNOWLEDGE_PATH.name, "ku_policy_extracted": KU_KNOWLEDGE_EXTRACTED, "ku_policy_review_status": _freshness(KU_KNOWLEDGE_EXTRACTED, KU_KNOWLEDGE_REVIEW_CYCLE_DAYS), } __all__ = [ "KnowledgeEntry", "RankedEntry", "get_chat_knowledge", "knowledge_status", "load_entries", "retrieve_general_entries", "retrieve_ku_sections", ]