Spaces:
Running
Running
| """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<title>[^\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"}, | |
| } | |
| 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 | |
| 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 | |
| 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) | |
| 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]) | |
| 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", | |
| ] | |