"""Stage 4 orchestration: the seven-intent routing pipeline (_agent_query_inner) and D1 logging.""" import asyncio import html import json import logging import os import re import time import uuid from collections import deque 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.services.staff_service import ( STAFF_DIRECTORY, match_staff_name, match_staff_role, should_attempt_staff_lookup, staff_name_answer, staff_role_answer, ) from src.agentcore.models import AgentResponse, SearchContextPayload from src.agentcore.constants import ( ALT_TERMS_RE, ASK_LIBRARIAN_URL, BOOLEAN_STRATEGY_RE, CITATION_CHAIN_RE, DUAL_RESOURCE_RE, EVENTS_RE, FULLTEXT_REQUEST_RE, HIGHLY_CITED_RE, LIBRARY_HOURS_URL, PARTIAL_TITLE_RE, PREDATORY_RE, PURE_GREETING_RE, THESES_RE, _INJECTION_RE, ) from src.agentcore.utils import ( _escape, _get_runtime_config, _light_strip_retrieval_boilerplate, _normalize_whitespace, _primo_clean_url, _safe_metrics_bucket, _safe_metrics_increment, _title_case_topic, ) from src.agentcore.classify import ( _is_greeting_menu_followup, _is_summary_request, _llm_classify, _looks_campus_question, _looks_library_hours_question, ) from src.agentcore.libcal import _events_answer, _hours_answer from src.agentcore.rendering import ( _ai_tools_footer, _contact_footer_for_query, _search_trace_block, _source_badge, _thought_block, _tool_urls, ) from src.agentcore.intents_library import ( _campus_answer, _database_recommendation_answer, _fulltext_chain_answer, _greeting_menu_clarify_answer, _is_database_recommendation_question, _library_follow_up, _rag_answer, _theses_answer, ) from src.agentcore.intents_search import ( _alternative_terms_answer, _apply_follow_up_action, _citation_chain_answer, _context_to_dict, _derive_resource_type, _detect_follow_up, _extract_topic, _highly_cited_note, _parse_year_filters, _predatory_eval_answer, _prepare_queries, _research_snapshot, _run_search_mode, _search_follow_up, _search_strategy_answer, ) from src.agentcore.intents_general import ( _general_follow_up, _libbee_casual_response, _social_follow_up, _web_search_answer, ) logger = logging.getLogger(__name__) _EMAIL_RE = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+") _LONG_DIGITS_RE = re.compile(r"\d{6,}") _TAG_RE = re.compile(r"<[^>]+>") # Strong references to in-flight analytics POSTs. # # asyncio keeps only WEAK references to tasks: a task created with # create_task() and then dropped can be garbage-collected before it finishes, # silently cancelling the request. That is invisible when the network is fast # and drops writes unpredictably when the handshake is slow. Holding a # reference until the task completes is the documented fix. _ANALYTICS_TASKS = set() def _spawn_background(coro) -> bool: """Run coro in the background, keeping a strong reference. True if scheduled.""" try: loop = asyncio.get_running_loop() except RuntimeError: try: loop = asyncio.get_event_loop() except Exception: return False if not loop.is_running(): return False task = loop.create_task(coro) _ANALYTICS_TASKS.add(task) task.add_done_callback(_ANALYTICS_TASKS.discard) return True def _redact_for_analytics(text: str, max_len: int = 200) -> str: """Privacy pass before anything reaches the analytics store: e-mail addresses and long digit runs (IDs, phone numbers) are masked and the text is truncated. Raw personal identifiers must never be persisted.""" text = _EMAIL_RE.sub("[email]", text or "") text = _LONG_DIGITS_RE.sub("[number]", text) return text[:max_len] def _answer_excerpt(answer_html: str, max_len: int = 800) -> str: """Plain-text excerpt of the rendered answer for offline relevance grading by the daily KB-gap agent. HTML is stripped and the same redaction applied.""" text = _TAG_RE.sub(" ", answer_html or "") text = re.sub(r"\s+", " ", text).strip() return _redact_for_analytics(text, max_len) # ── Analytics retry buffer ─────────────────────────────────────────────────── # Transient failures reaching the Worker (rate limiting, connection resets) # previously meant the row was lost for good. Failed payloads are parked here # and re-attempted on subsequent queries, so a blocked window costs a delay # rather than data. Bounded so a long outage cannot grow memory without limit. _PENDING_LOGS: "deque" = deque(maxlen=200) _PENDING_DROPPED = {"count": 0} # Rolling record of the last few analytics-write attempts, so the outcome can be # inspected through /admin/log-status without access to container logs. _LOG_ATTEMPTS: "deque" = deque(maxlen=15) def _record_attempt(outcome: str, detail: str = "") -> None: _LOG_ATTEMPTS.append({ "at": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S"), "outcome": outcome, "detail": detail[:200], }) def analytics_log_status() -> dict: """Snapshot of the analytics write path for the admin dashboard.""" return { "pending_buffered": len(_PENDING_LOGS), "buffer_capacity": _PENDING_LOGS.maxlen, "recent_attempts": list(_LOG_ATTEMPTS)[::-1], } async def _send_log_payload(worker_url: str, token: str, payload: dict): """POST one analytics payload. Raises on transport failure; returns status.""" from src.services.http_client import make_client as _mk headers = {"Authorization": f"Bearer {token}"} if token else {} async with _mk(timeout=10) as client: resp = await client.post(worker_url.rstrip("/") + "/log", json=payload, headers=headers) return resp.status_code async def _flush_pending(worker_url: str, token: str, max_items: int = 5): """Re-attempt parked payloads oldest-first; stop at the first failure so a still-blocked Worker is not hammered.""" sent = 0 while _PENDING_LOGS and sent < max_items: payload = _PENDING_LOGS[0] try: status = await _send_log_payload(worker_url, token, payload) except Exception: return sent if status >= 400: return sent _PENDING_LOGS.popleft() sent += 1 if sent: logger.info(f"D1 logging: flushed {sent} buffered row(s); " f"{len(_PENDING_LOGS)} still pending") return sent def _log_agent_query(question: str, response: AgentResponse, start_time: float) -> None: elapsed = time.time() - start_time intent = response.intent or "" tool = ", ".join(response.tools_used or [])[:60] model_used = response.model_used or "" result_count = len(response.search_results or []) try: from app import get_metrics_service get_metrics_service().log_query( question=_redact_for_analytics(question), intent=intent, tool=tool, model=model_used, response_time=elapsed, result_count=result_count, ) except Exception: pass try: settings = get_settings() worker_url = getattr(settings, "cloudflare_worker_url", "") or "" if worker_url: _payload = { "question": _redact_for_analytics(question), "tool": f"{intent}:{tool}"[:80], "model": model_used, "response_time": round(elapsed, 3), "result_count": result_count, # consumed by the daily KB-gap / relevance agent "answer_excerpt": _answer_excerpt(response.answer), "source_ids": [ str(src.get("source") or src.get("id") or "")[:60] for src in (response.sources or [])[:8] ], } _token = getattr(settings, "cloudflare_worker_token", "") or "" async def _post_to_worker(): # Drain anything parked by earlier failures first, then send this row. try: await _flush_pending(worker_url, _token) except Exception: pass try: status = await _send_log_payload(worker_url, _token, _payload) except Exception as _exc: _PENDING_LOGS.append(_payload) _record_attempt("unreachable", f"{type(_exc).__name__}: {_exc}") logger.warning( f"D1 logging unreachable ({type(_exc).__name__}: {_exc}); buffered " f"({len(_PENDING_LOGS)} pending)" ) return if status == 401: _record_attempt("rejected_401", "token mismatch") logger.error( "D1 logging rejected (401). CLOUDFLARE_WORKER_TOKEN does " "not match the Worker's ANALYTICS_TOKEN secret." ) elif status in (403, 429): _PENDING_LOGS.append(_payload) _record_attempt("throttled", f"HTTP {status}") logger.warning( f"D1 logging throttled or blocked (HTTP {status}); buffered " f"({len(_PENDING_LOGS)} pending). Repeated 403/429 from " f"workers.dev usually means Cloudflare is rate-limiting this " f"host; a custom domain on the Worker avoids it." ) elif status >= 400: _PENDING_LOGS.append(_payload) _record_attempt("http_error", f"HTTP {status}") logger.warning(f"D1 logging failed: HTTP {status}; buffered") else: _record_attempt("ok", f"HTTP {status}") logger.debug("D1 logged OK") if not _spawn_background(_post_to_worker()): logger.warning("D1 logging skipped: no running event loop") except Exception: pass async def _agent_query_inner( question: str, history, model: str, client_state, start_time: float, ) -> AgentResponse: # ── Prompt injection detection ──────────────────────────────────────────── if question and _INJECTION_RE.search(question): _safe_metrics_bucket("intents", "injection_blocked") return AgentResponse( answer=( "I'm sorry, but I can't process that request. " "I'm LibBee, the KU Library AI Assistant — here to help with " "library services, research, and academic resources. " f'How can I help you today? Ask a Librarian' ), intent="blocked", tools_used=["injection_guard"], model_used=model, response_time=time.time() - start_time, ) if not question: follow_up_question, follow_up_suggestions = _social_follow_up() return AgentResponse( answer="Please type a question and I'll do my best to help!" + _source_badge("libbee"), intent="social", model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if PURE_GREETING_RE.match(question): follow_up_question, follow_up_suggestions = _social_follow_up() return AgentResponse( answer=( _get_runtime_config().get("welcome_message", "").strip() or ( "Hi! I'm LibBee, the Khalifa University Library AI Assistant.

" "I can help you find articles and books, search databases, access full text, " "request Interlibrary Loan (ILL), and answer questions about library services and staff." ) + _source_badge("libbee") ), intent="social_greeting", model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) staff_match = match_staff_name(question) if should_attempt_staff_lookup(question) else None if staff_match: _safe_metrics_bucket("intents", "staff_lookup") follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=_thought_block("staff_lookup", "staff_name_match", question) + staff_name_answer(staff_match) + _source_badge("ku_library", "https://library.ku.ac.ae/"), intent="library_info", tools_used=["staff_name_match"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) staff_role_match = match_staff_role(question) if should_attempt_staff_lookup(question) else None if staff_role_match: _safe_metrics_bucket("intents", "staff_lookup") follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=_thought_block("staff_lookup", "staff_role_match", question) + staff_role_answer(staff_role_match, question) + _source_badge("ku_library", "https://library.ku.ac.ae/"), intent="library_info", tools_used=["staff_role_match"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if _is_greeting_menu_followup(question, history): follow_up_question, follow_up_suggestions = _social_follow_up() return AgentResponse( answer=_thought_block("social_greeting", "greeting_menu", question) + _greeting_menu_clarify_answer() + _source_badge("libbee"), intent="social_greeting", is_follow_up=True, model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if _looks_library_hours_question(question): follow_up_question, follow_up_suggestions = _library_follow_up(question) hours_html = await _hours_answer() return AgentResponse( answer=_thought_block("hours", "libcal_live", question) + hours_html + _source_badge("ku_library", LIBRARY_HOURS_URL), intent="library_info", tools_used=["libcal_live_hours"], ask_librarian=False, model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if EVENTS_RE.search(question): follow_up_question, follow_up_suggestions = _library_follow_up(question) events_html = await _events_answer() return AgentResponse( answer=_thought_block("library_info", "libcal_events", question, "I fetched upcoming events from the KU LibCal events calendar.") + events_html + _source_badge("ku_library", "https://kustar.libcal.com/calendar/events"), intent="library_info", tools_used=["libcal_live_events"], ask_librarian=False, model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if _looks_campus_question(question): follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=_thought_block("campus", "campus_hardcoded", question) + _campus_answer() + _source_badge("ku_library", "https://library.ku.ac.ae/"), intent="library_info", tools_used=["campus_hardcoded"], ask_librarian=False, model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if _is_database_recommendation_question(question): answer = _database_recommendation_answer(question) answer += _contact_footer_for_query(question, None) answer += _source_badge("ku_library", "https://library.ku.ac.ae/eresources") follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=_thought_block("database_rec", "database_recommendation", question) + answer, intent="library_info", tools_used=["database_recommendation"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) is_follow_up, follow_up_action, base_context = _detect_follow_up(question, client_state) # ── Catalogue availability ──────────────────────────────────────────────── _CATALOGUE_RE = re.compile( r"\b(" r"available in (the )?(ku|khalifa) library" r"|in (the )?(ku|khalifa) library" r"|does (the )?(ku|khalifa) library have" r"|does (the )?library (have|carry|stock|hold)" r"|(ku|khalifa) library (has|have|carry|holds?|stocks?)" r"|ku library (books?|copies|titles?)" r")\b", re.IGNORECASE, ) _BOOK_RE = re.compile(r"\b(book|books|ebook|ebooks|textbook|textbooks|copy|copies|title|titles)\b", re.IGNORECASE) if _CATALOGUE_RE.search(question) and _BOOK_RE.search(question): from urllib.parse import quote as _quote _topic_raw = re.sub( r"\b(can you (help me )?find|find|search|look for|show me|help me find" r"|books? (available|in|on)|available in.{0,30}library|in the ku library" r"|on|about|regarding|related to|ku library|khalifa university library" r"|machine learning|please|thanks)\b", " ", question, flags=re.IGNORECASE, ) _topic = _normalize_whitespace(_topic_raw).strip(" ,.") _primo_books_url = ( "https://khalifa.primo.exlibrisgroup.com/discovery/search" f"?query=any,contains,{_quote(_topic or question)},AND" "&tab=Everything&search_scope=MyInst_and_CI" "&vid=971KUOSTAR_INST:KU&lang=en&mode=advanced" "&qInclude=facet_rtype,exact,books" ) follow_up_question, follow_up_suggestions = _library_follow_up(question) rag_results = [] try: from app import get_rag_service rag_results = await get_rag_service().hybrid_search(question, top_k=3) except Exception: pass if rag_results: rag_answer = await _rag_answer(question, rag_results, [], model) else: rag_answer = "" catalogue_answer = ( f"To find books available in the KU Library on {_escape(_topic or 'this topic')}, " f"search the KU Library catalogue directly:

" f'' f"📚 Search KU Library catalogue for books →

" f"The catalogue shows real-time availability — whether a copy is on the shelf, checked out, or available as an e-book." ) if rag_answer: catalogue_answer += f"

{rag_answer}" return AgentResponse( answer=_thought_block("library_info", "catalogue_books", question) + catalogue_answer + _source_badge("primo", _primo_books_url), intent="library_info", tools_used=["catalogue_books_redirect"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if is_follow_up and base_context: _safe_metrics_increment("follow_up_hits") context = _apply_follow_up_action(base_context, follow_up_action) if follow_up_action == "alt_terms": alt_answer = await _alternative_terms_answer(question, base_context, model) follow_up_question, follow_up_suggestions = await _search_follow_up(base_context, model) return AgentResponse( answer=_thought_block("library_info", "alt_terms", question) + alt_answer + _source_badge("libbee"), intent="library_info", tools_used=["alt_terms"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, search_context=_context_to_dict(base_context), is_follow_up=True, ) if follow_up_action in {"summarize_topic", "deep_research_tools", "best_databases"}: context = await _prepare_queries(question, context, model, True) if follow_up_action == "summarize_topic": answer, search_results, sources = await _research_snapshot(context, model) else: answer = ( f"{_search_trace_block(question, context)}" f"Research guide for {_escape(context.display_topic or context.topic)}

" f"Here are the best AI tools and resources for a deep dive on this topic." + _source_badge("libbee") + _ai_tools_footer(context) ) search_results = [] sources = [] follow_up_question, follow_up_suggestions = await _search_follow_up(context, model, summary_mode=True) _safe_metrics_increment("search_handoffs") return AgentResponse( answer=answer, intent=context.intent, tools_used=["follow_up_resolution", follow_up_action], search_results=search_results, sources=sources, model_used=model, response_time=time.time() - start_time, corrected_query=context.topic, natural_query=context.ai_tool_query, database_query=context.primo_boolean_query, ai_tool_query=context.ai_tool_query, primo_boolean_query=context.primo_boolean_query, primo_search_url=_primo_clean_url(context), primo_ai_url=_tool_urls(context).get("primo_ai"), pubmed_search_url=_tool_urls(context).get("pubmed"), follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, search_context=_context_to_dict(context), is_follow_up=True, ) answer, search_results, sources, source_url = await _run_search_mode(question, context, model, True) follow_up_question, follow_up_suggestions = await _search_follow_up(context, model) return AgentResponse( answer=answer, intent=context.intent, tools_used=["follow_up_resolution", follow_up_action or "contextual_refinement"], search_results=search_results, sources=sources, model_used=model, response_time=time.time() - start_time, corrected_query=context.topic, natural_query=context.ai_tool_query, database_query=context.primo_boolean_query, ai_tool_query=context.ai_tool_query, primo_boolean_query=context.primo_boolean_query, primo_search_url=source_url, primo_ai_url=_tool_urls(context).get("primo_ai"), pubmed_search_url=_tool_urls(context).get("pubmed"), follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, search_context=_context_to_dict(context), is_follow_up=True, ) # ── Pre-classifier specialist handlers ─────────────────────────────────── if FULLTEXT_REQUEST_RE.search(question): ft_answer = _fulltext_chain_answer(question) follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=_thought_block("library_info", "fulltext_chain", question, "Identified a full-text access request. Returning the KU full-text access chain.") + ft_answer + _source_badge("ku_library", "https://library.ku.ac.ae/ill/"), intent="library_info", tools_used=["fulltext_chain"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if THESES_RE.search(question): th_answer = _theses_answer() follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=_thought_block("library_info", "theses_guide", question, "Identified a theses/repository question.") + th_answer + _source_badge("ku_library", "https://khazna.ku.ac.ae"), intent="library_info", tools_used=["theses_guide"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if PREDATORY_RE.search(question): pr_answer = _predatory_eval_answer() follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=_thought_block("library_info", "predatory_eval", question, "Identified a question about evaluating journals/peer review.") + pr_answer + _source_badge("ku_library", "https://library.ku.ac.ae/eresources"), intent="library_info", tools_used=["predatory_eval"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if CITATION_CHAIN_RE.search(question): cc_answer = _citation_chain_answer() follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=_thought_block("library_info", "citation_chain", question, "Identified a citation tracing / forward-backward search question.") + cc_answer + _source_badge("ku_library", "https://library.ku.ac.ae/eresources"), intent="library_info", tools_used=["citation_chain"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if BOOLEAN_STRATEGY_RE.search(question): strategy_answer, _plan = await _search_strategy_answer(question, model) follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=_thought_block("library_info", "search_strategy", question, "Identified a search strategy / boolean building request.") + strategy_answer + _source_badge("libbee"), intent="library_info", tools_used=["search_strategy_explain"], model_used=model, response_time=time.time() - start_time, database_query=_plan.get("boolean"), natural_query=_plan.get("natural"), follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if ALT_TERMS_RE.search(question): alt_answer = await _alternative_terms_answer(question, None, model) follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=_thought_block("library_info", "alt_terms", question) + alt_answer + _source_badge("libbee"), intent="library_info", tools_used=["alt_terms"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) cls_result = await _llm_classify(question, history, model) intent = cls_result.get("intent", "general") _safe_metrics_bucket("intents", intent) casual_hint = cls_result.get("casual_answer", "") if intent == "social": answer = ( casual_hint if casual_hint and len(casual_hint.strip()) > 20 else await _libbee_casual_response(question, history, model, casual_hint) ) follow_up_question, follow_up_suggestions = _social_follow_up() return AgentResponse( answer=_thought_block("social", "libbee_casual", question) + answer + _source_badge("libbee"), intent="social_greeting", tools_used=["libbee_casual"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if intent == "sensitive": _safe_metrics_bucket("intents", "sensitive_blocked") follow_up_question, follow_up_suggestions = _social_follow_up() return AgentResponse( answer=( "That's not something I'm able to comment on. " "As the KU Library AI Assistant, I'm here to support your academic " "and research needs — finding articles, accessing databases, " "library services, and research tools.

" "Is there something I can help you with today? " f'💬 Ask a Librarian' ), intent="sensitive_blocked", tools_used=["sensitive_intent_guard"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if intent == "library_info": rag_results = [] try: from app import get_rag_service rag_results = await get_rag_service().hybrid_search(question, top_k=4) except Exception as e: logger.error(f"RAG search failed: {e}") if not rag_results: answer = ( "I couldn't find specific information on that in my knowledge base.

" "Please contact our librarians directly:
" '📧 libse@ku.ac.ae
' f'🔗 Ask a Librarian' ) answer += _source_badge("ku_library", ASK_LIBRARIAN_URL) sources = [] tools_used = ["rag_no_results"] thought = _thought_block("library_info", "rag_no_results", question, "No matching content was found — I am directing you to Ask a Librarian.") else: answer = await _rag_answer(question, rag_results, history, model) answer += _source_badge("rag", "https://library.ku.ac.ae/") sources = [ {"title": r.get("title", "Library KB"), "url": r.get("source", "")} for r in rag_results if r.get("source") ] tools_used = ["rag_hybrid_search"] thought = _thought_block("rag_search", "rag_hybrid_search", question, f"Hybrid FAISS + BM25 search returned {len(rag_results)} relevant chunk(s) from the KB.") follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=thought + answer, intent="library_info", tools_used=tools_used, sources=sources, ask_librarian=not bool(rag_results), model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) if intent in ("search_academic", "search_medical"): if PARTIAL_TITLE_RE.search(question): from urllib.parse import quote as _qt _title_frag = (re.search(r'"([^"]{4,80})"', question) or re.search(r'"([^"]{4,80})"', question)) _title_q = _title_frag.group(1) if _title_frag else _light_strip_retrieval_boilerplate(question) _title_url = ( "https://khalifa.primo.exlibrisgroup.com/discovery/search" f"?query=title,contains,{_qt(_title_q)},AND" "&tab=Everything&search_scope=MyInst_and_CI" "&vid=971KUOSTAR_INST:KU&lang=en&mode=advanced" ) follow_up_question, follow_up_suggestions = _library_follow_up(question) return AgentResponse( answer=( _thought_block("library_info", "title_search", question, "Identified a partial title search — routing to PRIMO title search.") + f"📖 Title search in KU Library

" f"I'll search the KU Library catalogue by title for: {_escape(_title_q)}

" f'' f"🔍 Search by title in PRIMO →

" "If you only remember part of the title, use PRIMO's advanced search and choose " "Title from the field dropdown — it supports partial matches.
" "You can also try " 'Google Scholar ' "with the partial title in quotes." + _source_badge("primo", _title_url) ), intent="library_info", tools_used=["title_search"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) topic = await _extract_topic(question, model) topic = _title_case_topic(topic) year_from, year_to = _parse_year_filters(question) if DUAL_RESOURCE_RE.search(question): import asyncio as _aio _ctx_articles = SearchContextPayload( context_id=str(uuid.uuid4()), topic=topic, display_topic=topic, intent=intent, source="primo", resource_type="articles", peer_reviewed=bool(re.search(r"\bpeer[- ]reviewed\b", question, re.IGNORECASE)), open_access=bool(re.search(r"\bopen access\b", question, re.IGNORECASE)), year_from=year_from, year_to=year_to, ) _ctx_books = SearchContextPayload( context_id=str(uuid.uuid4()), topic=topic, display_topic=topic, intent=intent, source="primo", resource_type="books", peer_reviewed=False, open_access=False, year_from=year_from, year_to=year_to, ) _ctx_articles, _ctx_books = await _aio.gather( _prepare_queries(question, _ctx_articles, model, False), _prepare_queries(question, _ctx_books, model, False), ) _src_url_a = _primo_clean_url(_ctx_articles) _src_url_b = _primo_clean_url(_ctx_books) _dual_answer = _search_trace_block(question, _ctx_articles) _dual_answer += ( f"📄 Articles on {_escape(topic)}
" f'' f'Search PRIMO for articles →' ) _dual_answer += _source_badge("primo", _src_url_a) _dual_answer += f"

📚 Books on {_escape(topic)}
" _dual_answer += ( f'' f'Search PRIMO for books →' ) _dual_answer += _source_badge("primo", _src_url_b) _dual_answer += _ai_tools_footer(_ctx_articles) follow_up_question, follow_up_suggestions = await _search_follow_up(_ctx_articles, model) return AgentResponse( answer=_dual_answer, intent=intent, tools_used=["primo_browser_link"], search_results=[], sources=[], model_used=model, response_time=time.time() - start_time, corrected_query=topic, natural_query=_ctx_articles.ai_tool_query, primo_boolean_query=_ctx_articles.primo_boolean_query, primo_search_url=_src_url_a, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, search_context=_context_to_dict(_ctx_articles), ) context = SearchContextPayload( context_id=str(uuid.uuid4()), topic=topic, display_topic=topic, intent=intent, source="primo", resource_type=_derive_resource_type(question), peer_reviewed=bool(re.search(r"\bpeer[- ]reviewed\b", question, re.IGNORECASE)), open_access=bool(re.search(r"\bopen access\b", question, re.IGNORECASE)), year_from=year_from, year_to=year_to, ) context = await _prepare_queries(question, context, model, False) summary_mode = _is_summary_request(question) deep_research_mode = bool(re.search( r"\b(deep research|deep dive|systematic review|full literature review|exhaustive)\b", question, re.IGNORECASE, )) is_highly_cited = bool(HIGHLY_CITED_RE.search(question)) if summary_mode or deep_research_mode: answer, search_results, sources = await _research_snapshot(context, model) if deep_research_mode and "Research guide" not in answer and "AI tools" not in answer: answer += f"

{_ai_tools_footer(context)}" if is_highly_cited: answer += _highly_cited_note(_escape(topic)) follow_up_question, follow_up_suggestions = await _search_follow_up(context, model, summary_mode=True) _safe_metrics_increment("search_handoffs") return AgentResponse( answer=answer, intent=intent, tools_used=["research_snapshot"], search_results=search_results, sources=sources, model_used=model, response_time=time.time() - start_time, corrected_query=context.topic, natural_query=context.ai_tool_query, database_query=context.primo_boolean_query, ai_tool_query=context.ai_tool_query, primo_boolean_query=context.primo_boolean_query, primo_search_url=_primo_clean_url(context), primo_ai_url=_tool_urls(context).get("primo_ai"), pubmed_search_url=_tool_urls(context).get("pubmed"), follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, search_context=_context_to_dict(context), ) answer, search_results, sources, source_url = await _run_search_mode(question, context, model, False) if is_highly_cited: answer += _highly_cited_note(_escape(topic)) follow_up_question, follow_up_suggestions = await _search_follow_up(context, model) return AgentResponse( answer=answer, intent=intent, tools_used=["primo_inline" if context.source == "primo" else "pubmed_inline"], search_results=search_results, sources=sources, model_used=model, response_time=time.time() - start_time, corrected_query=context.topic, natural_query=context.ai_tool_query, database_query=context.primo_boolean_query, ai_tool_query=context.ai_tool_query, primo_boolean_query=context.primo_boolean_query, primo_search_url=source_url, primo_ai_url=_tool_urls(context).get("primo_ai"), pubmed_search_url=_tool_urls(context).get("pubmed"), follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, search_context=_context_to_dict(context), ) if intent == "general_recent": answer = await _web_search_answer(question, history, model) + _source_badge("web_live") follow_up_question, follow_up_suggestions = _general_follow_up(question) return AgentResponse( answer=_thought_block("general_recent", "web_search", question) + answer, intent="general_recent", tools_used=["web_search"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, ) # general fallback answer = await _web_search_answer(question, history, model) + _source_badge("web") follow_up_question, follow_up_suggestions = _general_follow_up(question) return AgentResponse( answer=_thought_block("general", "web_search", question) + answer, intent="general", tools_used=["web_search"], model_used=model, response_time=time.time() - start_time, follow_up_question=follow_up_question, follow_up_suggestions=follow_up_suggestions, )