| """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 |
|
|
| 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>.' |
|
|
|
|
| async def _web_search_answer(question: str, history: List[ChatMessage], model: str) -> str: |
| """LLM #4 — web search answer. v3.8.1: appends _GUARDRAIL + _URL_INSTRUCTION.""" |
| settings = get_settings() |
| _WEB_SYSTEM = ( |
| "You are LibBee, the Khalifa University Library AI Assistant (Abu Dhabi, UAE). " |
| "Answer using current web search. Be concise and factual. Use HTML <br> for line breaks. " |
| + _GUARDRAIL + "\n\n" + _URL_INSTRUCTION |
| ) |
| if model == "claude" and settings.anthropic_api_key: |
| try: |
| import anthropic |
| client = anthropic.Anthropic(api_key=settings.anthropic_api_key) |
| response = client.messages.create( |
| model="claude-haiku-4-5-20251001", |
| max_tokens=520, |
| system=_WEB_SYSTEM, |
| tools=[{"type": "web_search_20250305", "name": "web_search"}], |
| messages=[{"role": "user", "content": question}], |
| ) |
| text = "".join(block.text for block in response.content if hasattr(block, "text")) |
| if text.strip(): |
| return text.strip() |
| except Exception as e: |
| logger.warning(f"Claude web search failed: {e}") |
| if settings.openai_api_key: |
| try: |
| from openai import OpenAI |
| client = OpenAI(api_key=settings.openai_api_key) |
| response = client.responses.create( |
| model="gpt-4o-mini", |
| tools=[{"type": "web_search_preview"}], |
| instructions=_WEB_SYSTEM, |
| input=question, |
| ) |
| text = "" |
| for item in response.output: |
| if hasattr(item, "content"): |
| for block in item.content: |
| if hasattr(block, "text"): |
| text += block.text |
| if text.strip(): |
| return text.strip() |
| except Exception as e: |
| logger.warning(f"GPT web search failed: {e}") |
| return await _llm_general_answer(question, history, model) |
|
|
|
|
| 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"}, |
| ], |
| ) |
|
|
|
|
|
|
