| """general / social intents: web-search answers, general LLM answers, casual persona replies.""" |
| 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, urlsplit |
|
|
| import httpx |
| from pydantic import BaseModel, ConfigDict, Field |
|
|
| from src.config import get_settings, LIBBEE_VERSION |
|
|
| from src.agentcore.models import ChatMessage |
| from src.agentcore.constants import KU_MAIN_URL, _GUARDRAIL, _URL_INSTRUCTION |
| from src.agentcore.utils import _build_history_messages, _get_llm, _get_runtime_config |
|
|
| logger = logging.getLogger(__name__) |
|
|
| async def _libbee_casual_response(question: str, history: List[ChatMessage], model: str, hint: str = "") -> str: |
| """LLM #1 — casual social response. v3.8.1: appends _GUARDRAIL + _URL_INSTRUCTION.""" |
| if hint and len(hint.strip()) > 20: |
| return hint.strip() |
| settings = get_settings() |
| if not settings.openai_api_key and not settings.anthropic_api_key: |
| return "I'm <strong>LibBee</strong>, the Khalifa University Library AI Assistant — always happy to help!" |
| try: |
| llm = _get_llm(model, temperature=0.6, max_tokens=160) |
| msgs = [{"role": "system", "content": ( |
| "You are LibBee, the Khalifa University Library AI Assistant. " |
| "Respond warmly and naturally in 1-3 sentences as a friendly librarian. " |
| "No markdown, no bullet points. " |
| + _GUARDRAIL + "\n\n" + _URL_INSTRUCTION |
| )}] |
| _cfg = _get_runtime_config() |
| _ci = _cfg.get("custom_instructions", "").strip() |
| if _ci: |
| msgs[0]["content"] += " " + _ci |
| msgs.extend(_build_history_messages(history)) |
| msgs.append({"role": "user", "content": question}) |
| response = await llm.ainvoke(msgs) |
| return response.content.strip() |
| except Exception: |
| return "I'm <strong>LibBee</strong>, the Khalifa University Library AI Assistant — always happy to help!" |
|
|
|
|
| async def _llm_general_answer(question: str, history: List[ChatMessage], model: str) -> str: |
| """LLM #3 — general factual answer. v3.8.1: appends _GUARDRAIL + _URL_INSTRUCTION.""" |
| settings = get_settings() |
| if not settings.openai_api_key and not settings.anthropic_api_key: |
| return ( |
| "I'm <strong>LibBee</strong>, the KU Library AI Assistant. " |
| f'For general university information, visit <a href="{KU_MAIN_URL}" target="_blank">ku.ac.ae</a>.' |
| ) |
| system = ( |
| "You are LibBee, the Khalifa University Library AI Assistant (Abu Dhabi, UAE). " |
| "Answer factually and concisely in 3-5 sentences. Use HTML <br> for line breaks. " |
| "KU means Khalifa University. If the topic is academic, briefly offer to help find library resources. " |
| "When referencing KU Library services, include these verified links where relevant: " |
| "Library homepage: https://library.ku.ac.ae | " |
| "Ask a Librarian: https://library.ku.ac.ae/AskUs | " |
| "Library hours: https://library.ku.ac.ae/hours | " |
| "ILL requests: https://library.ku.ac.ae/ill/ | " |
| "Study rooms: https://library.ku.ac.ae/rooms/ | " |
| "E-resources: https://library.ku.ac.ae/eresources | " |
| "Khazna repository: https://khazna.ku.ac.ae | " |
| "PRIMO discovery: https://khalifa.primo.exlibrisgroup.com/discovery/search?vid=971KUOSTAR_INST:KU. " |
| "Only use these exact URLs — never invent others. " |
| + _GUARDRAIL + "\n\n" + _URL_INSTRUCTION |
| ) |
| _cfg = _get_runtime_config() |
| _ci = _cfg.get("custom_instructions", "").strip() |
| if _ci: |
| system += " " + _ci |
| try: |
| llm = _get_llm(model, temperature=0.3, max_tokens=360) |
| msgs = [{"role": "system", "content": system}] |
| msgs.extend(_build_history_messages(history)) |
| msgs.append({"role": "user", "content": question}) |
| response = await llm.ainvoke(msgs) |
| return response.content.strip() |
| except Exception as e: |
| logger.error(f"_llm_general_answer error: {e}") |
| return f'Having trouble right now. Visit <a href="{KU_MAIN_URL}" target="_blank">ku.ac.ae</a>.' |
|
|
|
|
| def _response_field(value: Any, name: str, default: Any = None) -> Any: |
| """Read SDK objects and dictionaries without depending on a provider SDK.""" |
| return value.get(name, default) if isinstance(value, dict) else getattr(value, name, default) |
|
|
|
|
| def _citation_source(citation: Any) -> Optional[dict]: |
| url = str(_response_field(citation, "url", "") or "").strip() |
| try: |
| parsed = urlsplit(url) |
| if (parsed.scheme not in ("http", "https") or not parsed.hostname |
| or parsed.username or parsed.password or any(ord(c) < 32 for c in url)): |
| return None |
| except ValueError: |
| return None |
| return {"title": str(_response_field(citation, "title", "") or parsed.hostname), |
| "url": url, "source": "web_search"} |
|
|
|
|
| def _render_web_text(block: Any, sources: List[dict], anthropic: bool = False) -> str: |
| """Escape provider prose and attach real, clickable citations at their source span.""" |
| raw = str(_response_field(block, "text", "") or "") |
| if not raw: |
| return "" |
| citations = _response_field(block, "citations" if anthropic else "annotations", []) or [] |
| insertions: Dict[int, List[str]] = {} |
| for citation in citations: |
| kind = _response_field(citation, "type", "") |
| if kind != ("web_search_result_location" if anthropic else "url_citation"): |
| continue |
| source = _citation_source(citation) |
| if not source: |
| continue |
| existing = next((i for i, item in enumerate(sources) if item["url"] == source["url"]), None) |
| if existing is None: |
| sources.append(source) |
| existing = len(sources) - 1 |
| position = len(raw) if anthropic else _response_field(citation, "end_index", len(raw)) |
| if not isinstance(position, int) or not 0 <= position <= len(raw): |
| position = len(raw) |
| link = (f' <a href="{html.escape(source["url"], quote=True)}" target="_blank" ' |
| f'rel="noopener noreferrer" title="{html.escape(source["title"], quote=True)}">' |
| f'[{existing + 1}]</a>') |
| if link not in insertions.setdefault(position, []): |
| insertions[position].append(link) |
|
|
| def escaped(segment: str) -> str: |
| |
| segment = re.sub(r"\ue200cite[^\ue201]*\ue201", "", segment) |
| return html.escape(segment).replace("\n", "<br>") |
|
|
| answer, previous = [], 0 |
| for position in sorted(insertions): |
| answer.append(escaped(raw[previous:position])) |
| answer.extend(insertions[position]) |
| previous = position |
| answer.append(escaped(raw[previous:])) |
| return "".join(answer) |
|
|
|
|
| def _parse_web_response(response: Any, provider: str) -> dict: |
| """Only label an answer live when a search completed and returned cited sources.""" |
| sources: List[dict] = [] |
| parts: List[str] = [] |
| searched = False |
| if provider == "claude": |
| for block in _response_field(response, "content", []) or []: |
| kind = _response_field(block, "type", "") |
| if kind == "web_search_tool_result": |
| |
| result = _response_field(block, "content") |
| searched = searched or isinstance(result, list) |
| elif kind == "text": |
| parts.append(_render_web_text(block, sources, anthropic=True)) |
| finished = _response_field(response, "stop_reason", "end_turn") == "end_turn" |
| else: |
| for item in _response_field(response, "output", []) or []: |
| if (_response_field(item, "type", "") == "web_search_call" |
| and _response_field(item, "status", "") == "completed"): |
| searched = True |
| for block in _response_field(item, "content", []) or []: |
| if _response_field(block, "type", "") == "output_text": |
| parts.append(_render_web_text(block, sources)) |
| finished = _response_field(response, "status", "completed") == "completed" |
| live = searched and finished and bool(sources) |
| return {"answer": "".join(parts).strip(), "sources": sources, |
| "tools_used": ["web_search"] if searched else [], |
| "live_search": live, "model_used": provider} |
|
|
|
|
| async def _web_search_result(question: str, history: List[ChatMessage], model: str) -> dict: |
| """Return source evidence alongside text, with bounded calls and honest fallback metadata.""" |
| settings = get_settings() |
| system = ( |
| "You are LibBee, the Khalifa University Library AI Assistant (Abu Dhabi, UAE). " |
| "Search the web and answer using current sources. Be concise and factual. " |
| "Use plain text with paragraph breaks, no HTML or Markdown. " |
| "Cite factual claims using the search tool's citations; never invent sources or URLs. " |
| "The application renders citation links, so do not write HTML links yourself. " |
| + _GUARDRAIL |
| ) |
| messages = _build_history_messages(history) |
| messages.append({"role": "user", "content": question}) |
| if settings.anthropic_api_key and (model == "claude" or not settings.openai_api_key): |
| try: |
| import anthropic |
| async with anthropic.AsyncAnthropic( |
| api_key=settings.anthropic_api_key, timeout=16.0, max_retries=0 |
| ) as client: |
| response = await asyncio.wait_for(client.messages.create( |
| model=os.getenv("ANTHROPIC_WEB_MODEL", "claude-haiku-4-5-20251001"), |
| max_tokens=620, |
| system=system, |
| tools=[{"type": "web_search_20250305", "name": "web_search", "max_uses": 2}], |
| messages=messages, |
| ), timeout=18.0) |
| result = _parse_web_response(response, "claude") |
| if result["answer"] and result["live_search"]: |
| return result |
| logger.info("Claude web response had no complete cited search answer") |
| except Exception as exc: |
| logger.warning("Claude web search failed (%s)", type(exc).__name__) |
| if settings.openai_api_key: |
| try: |
| from openai import AsyncOpenAI |
| async with AsyncOpenAI( |
| api_key=settings.openai_api_key, timeout=16.0, max_retries=0 |
| ) as client: |
| response = await asyncio.wait_for(client.responses.create( |
| model=os.getenv("OPENAI_WEB_MODEL", "gpt-4.1-mini"), |
| |
| tools=[{"type": "web_search_preview"}], |
| tool_choice={"type": "web_search_preview"}, |
| instructions=system, |
| input=messages, |
| max_output_tokens=620, |
| store=False, |
| ), timeout=18.0) |
| result = _parse_web_response(response, "gpt") |
| if result["answer"] and result["live_search"]: |
| return result |
| logger.info("GPT web response had no complete cited search answer") |
| except Exception as exc: |
| logger.warning("GPT web search failed (%s)", type(exc).__name__) |
| fallback = await _llm_general_answer(question, history, model) |
| provider = "claude" if settings.anthropic_api_key and (model == "claude" or not settings.openai_api_key) else "gpt" |
| return { |
| "answer": "<em>I couldn’t verify current web information. The following is general guidance and may be out of date.</em><br><br>" + fallback, |
| "sources": [], "tools_used": ["general_knowledge"], "live_search": False, |
| "model_used": provider, |
| } |
|
|
|
|
| async def _web_search_answer(question: str, history: List[ChatMessage], model: str) -> str: |
| """Compatibility wrapper; orchestration uses _web_search_result for source metadata.""" |
| return (await _web_search_result(question, history, model))["answer"] |
|
|
|
|
| def _general_follow_up(question: str) -> Tuple[str, List[dict]]: |
| return ( |
| "Would you like a brief explanation, current web information, or KU Library resources on this topic?", |
| [ |
| {"label": "Shorter explanation", "question": f"Give me a shorter explanation of {question}"}, |
| {"label": "Current web information", "question": f"Show current web information on {question}"}, |
| {"label": "Find KU Library resources", "question": f"Find KU Library resources on {question}"}, |
| ], |
| ) |
|
|
|
|
| def _social_follow_up() -> Tuple[str, List[dict]]: |
| return ( |
| "What would you like help with next?", |
| [ |
| {"label": "Find articles on a topic", "question": "Find articles on a topic"}, |
| {"label": "Check a library service", "question": "Check a library service"}, |
| {"label": "Contact a librarian", "question": "Contact a librarian"}, |
| ], |
| ) |
|
|
|
|
|
|
