diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bc67918..79f1d06 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,10 +28,10 @@ jobs: pip install pytest flake8 - name: Run linting - run: flake8 main.py stellar.py safety tests/redteam study.py fiqh.py hadith.py confidence.py review.py review_store.py tafsir.py semantic_cache.py tests/test_confidence.py tests/test_review_queue.py tests/test_tafsir.py scripts/build_hadith_data.py scripts/build_surah_index.py --max-line-length=120 --ignore=E501,W503 + run: flake8 main.py stellar.py safety tests/redteam study.py fiqh.py hadith.py confidence.py review.py review_store.py tafsir.py semantic_cache.py intent.py tests/test_confidence.py tests/test_review_queue.py tests/test_tafsir.py tests/test_intent.py scripts/build_hadith_data.py scripts/build_surah_index.py --max-line-length=120 --ignore=E501,W503 - name: Check syntax - run: python -m compileall -q main.py stellar.py safety tests/redteam study.py fiqh.py hadith.py confidence.py review.py review_store.py tafsir.py semantic_cache.py scripts/build_hadith_data.py scripts/build_surah_index.py + run: python -m compileall -q main.py stellar.py safety tests/redteam study.py fiqh.py hadith.py confidence.py review.py review_store.py tafsir.py semantic_cache.py intent.py scripts/build_hadith_data.py scripts/build_surah_index.py - name: Run offline safety and red-team tests run: pytest -q tests/redteam @@ -56,3 +56,6 @@ jobs: - name: Run tafsir tests run: pytest -q tests/test_tafsir.py + + - name: Run intent classification tests + run: pytest -q tests/test_intent.py diff --git a/intent.py b/intent.py new file mode 100644 index 0000000..08f0ec6 --- /dev/null +++ b/intent.py @@ -0,0 +1,484 @@ +""" +Question-understanding pipeline: intent classification, clarifying questions, +answer-length calibration, and suggested follow-ups. + +Provides a deterministic short-circuit for trivial messages (salam variants, +very short greetings), a Gemini structured-output call for everything else, +per-intent generation-parameter tables, ambiguity detection, and +defensive follow-up parsing. +""" + +import json +import logging +import os +import re +from dataclasses import dataclass +from enum import Enum +from typing import Any, Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Intent Taxonomy +# --------------------------------------------------------------------------- + + +class Intent(str, Enum): + GREETING_SMALLTALK = "greeting_smalltalk" + FACTUAL_KNOWLEDGE = "factual_knowledge" + FIQH_RULING = "fiqh_ruling" + PERSONAL_GUIDANCE = "personal_guidance" + PLATFORM_QUESTION = "platform_question" + OUT_OF_SCOPE = "out_of_scope" + + +# --------------------------------------------------------------------------- +# Classification result +# --------------------------------------------------------------------------- + + +@dataclass +class ClassificationResult: + intent: Intent + needs_clarification: bool = False + clarifying_question: str = "" + ambiguity_confidence: float = 0.0 + """How likely the question is ambiguous, 0..1. Only meaningful when + needs_clarification is True.""" + + +# --------------------------------------------------------------------------- +# Deterministic short-circuit — trivial messages are classified without an +# LLM call so the common greeting case adds nearly zero latency. +# --------------------------------------------------------------------------- + +# Patterns that match salam / greeting variants +_SALAM_PATTERNS = [ + re.compile(r"^\s*(?:as[sz]ala[mu]|salam|salaam|sallam)", + re.IGNORECASE), + re.compile(r"^\s*(?:wa\s*)?(?:'?alaykum|'?aleikum)", + re.IGNORECASE), + re.compile(r"^\s*(?:hello|hi\b|hey|greetings|good\s+(?:morning|afternoon|evening|day))", + re.IGNORECASE), + re.compile(r"^\s*(?:assalamu[_\s]?alaykum|assalamo[_\s]?alaikum|as-salam)", + re.IGNORECASE), +] + +# Very short messages (<=3 words) that are clearly just greetings +_GREETING_WORDS = { + "hi", "hello", "hey", "salam", "salaam", "assalamu", "alaykum", + "assalamo", "alaikum", "as-salam", "marhaba", "ahlan", + "good", "morning", "afternoon", "evening", "peace", +} + +# Short message length threshold — messages at or below this word count +# go through the deterministic check if they match greeting-like patterns. +_SHORT_MSG_THRESHOLD = 5 + + +def _is_trivial_greeting(message: str) -> bool: + """Return True if *message* is almost certainly a greeting / salam. + + This is intentionally conservative — borderline cases fall through to + the LLM classifier so we never misclassify a real question as a greeting. + """ + stripped = message.strip() + if not stripped: + return False + + words = stripped.split() + word_count = len(words) + + # Very short pure-greeting messages + if word_count <= 3 and words[0].lower().strip("?!.,") in _GREETING_WORDS: + return True + + # Salam pattern match + if word_count <= _SHORT_MSG_THRESHOLD: + for pat in _SALAM_PATTERNS: + if pat.match(stripped): + return True + + return False + + +# --------------------------------------------------------------------------- +# Ambiguity hints — keywords/phrases that suggest the question could have +# multiple valid interpretations. +# --------------------------------------------------------------------------- + +_AMBIGUITY_TRIGGERS: List[Tuple[re.Pattern, str]] = [ + (re.compile(r"\b(?:is\s+\w+\s+haram|is\s+\w+\s+halal)\b", re.IGNORECASE), + "This question depends on context and scholarly school of thought. " + "Could you specify which madhhab or situation you are asking about?"), + (re.compile(r"\bwhat\s+breaks\b", re.IGNORECASE), + "Are you asking about what invalidates wudu, breaks the fast, or " + "nullifies something else?"), + (re.compile(r"\b(?:can\s+I|is\s+it\s+permissible)\b", re.IGNORECASE), + "Could you provide more context about the specific situation you " + "are asking about?"), + (re.compile(r"\bhow\s+(?:to|do\s+I)\b", re.IGNORECASE), + "Which specific act or practice are you asking about?"), +] + + +def _detect_ambiguity(message: str) -> Tuple[bool, str]: + """Determine whether *message* seems underspecified / ambiguous. + + Returns (needs_clarification, clarifying_question). + """ + for pat, question in _AMBIGUITY_TRIGGERS: + if pat.search(message): + return True, question + return False, "" + + +# Default fallback clarifying question +_FALLBACK_CLARIFY = ( + "Could you please provide more detail so I can give you the most " + "accurate answer?" +) + + +# --------------------------------------------------------------------------- +# Classifier — uses Gemini structured output for non-trivial messages. +# --------------------------------------------------------------------------- + +_CLASSIFIER_PROMPT = """You are an intent classifier for an Islamic-knowledge AI assistant. +Analyze the user's message and output a JSON object with the following fields: + +1. "intent": one of "greeting_smalltalk", "factual_knowledge", "fiqh_ruling", + "personal_guidance", "platform_question", "out_of_scope" +2. "needs_clarification": true or false — set to true ONLY if the question is + genuinely ambiguous or underspecified. Be conservative: when in doubt, set false. +3. "clarifying_question": if needs_clarification is true, provide a single short + clarifying question. Otherwise, set to "". +4. "ambiguity_confidence": a float from 0.0 to 1.0 indicating how ambiguous + the message is. 0.0 = not ambiguous, 1.0 = very ambiguous. + +Intent descriptions: +- greeting_smalltalk: Simple greetings, salam exchanges, casual talk +- factual_knowledge: Questions about aqeedah, seerah, Islamic definitions, history +- fiqh_ruling: Questions about halal/haram, rulings, jurisprudence +- personal_guidance: Emotional/spiritual support, du'a requests, personal advice +- platform_question: Questions about the Deen Bridge platform, courses, zakat feature +- out_of_scope: Non-Islamic topics, harmful requests, nonsense + +Respond with ONLY a valid JSON object, no other text. + +User message: {message}""" + +# Threshold for "would adding the LLM call be worthwhile"? If the message +# passes the deterministic greeting check, we skip LLM entirely. +# Default ambiguity threshold — needs to be high to avoid over-clarifying. +AMBIGUITY_THRESHOLD = float(os.getenv("AMBIGUITY_THRESHOLD", "0.8")) + + +def classify_intent(message: str, model_callable=None) -> ClassificationResult: + """Classify *message* into an Intent. + + *model_callable* should be a callable that accepts a prompt string and + returns a response object with a ``.text`` attribute (e.g. a + ``genai.GenerativeModel.generate_content`` method). If ``None``, the + function still works for trivial messages but will raise for non-trivial + ones (useful for testing the short-circuit path). + """ + # 1. Deterministic short-circuit for trivial greetings + if _is_trivial_greeting(message): + return ClassificationResult( + intent=Intent.GREETING_SMALLTALK, + ) + + # 2. LLM-based classification + return _llm_classify(message, model_callable) + + +def _llm_classify(message: str, model_callable) -> ClassificationResult: + """Call the LLM to classify *message*.""" + prompt = _CLASSIFIER_PROMPT.format(message=message) + + if model_callable is None: + raise RuntimeError( + "classify_intent requires a model_callable for non-trivial messages." + ) + + try: + response = model_callable(prompt) + raw = response.text.strip() + + # Strip markdown code fences if present + if raw.startswith("```"): + # Find the first ``` and the closing ``` + start = raw.find("\n") + 1 if "\n" in raw else raw.find("```") + 3 + end = raw.rfind("```") + if end > start: + raw = raw[start:end].strip() + + data = json.loads(raw) + except (json.JSONDecodeError, AttributeError, ValueError) as exc: + logger.warning("Intent classifier returned unparseable output: %s", exc) + return ClassificationResult( + intent=Intent.FACTUAL_KNOWLEDGE, # safe fallback + ) + + # Extract and validate intent + intent_str = data.get("intent", "factual_knowledge") + try: + intent = Intent(intent_str) + except ValueError: + logger.warning("Unknown intent '%s'; falling back to factual_knowledge", intent_str) + intent = Intent.FACTUAL_KNOWLEDGE + + # Ambiguity + needs_clarification = data.get("needs_clarification", False) + clarifying_question = data.get("clarifying_question", "") + ambiguity_confidence = float(data.get("ambiguity_confidence", 0.0)) + + # Apply threshold — if confidence is below threshold, don't clarify + if ambiguity_confidence < AMBIGUITY_THRESHOLD: + needs_clarification = False + clarifying_question = "" + + # Fallback if needs_clarification is True but no question provided + if needs_clarification and not clarifying_question: + clarifying_question = _FALLBACK_CLARIFY + + return ClassificationResult( + intent=intent, + needs_clarification=needs_clarification, + clarifying_question=clarifying_question, + ambiguity_confidence=ambiguity_confidence, + ) + + +# --------------------------------------------------------------------------- +# Per-intent generation parameters and instruction snippets +# --------------------------------------------------------------------------- + + +@dataclass +class IntentConfig: + instruction_snippet: str + """Additional instruction appended to ISLAMIC_CONTEXT for this intent.""" + + temperature: float = 0.7 + top_p: float = 0.8 + top_k: int = 40 + max_output_tokens: int = 2048 + """Default generation parameters, all env-overridable per intent.""" + + def effective_params(self) -> Dict[str, Any]: + """Return generation config dict, respecting env overrides.""" + return { + "temperature": float(os.getenv( + f"TEMP_{self.max_output_tokens}", + str(self.temperature), + )), + "top_p": float(os.getenv( + f"TOP_P_{self.max_output_tokens}", + str(self.top_p), + )), + "top_k": int(os.getenv( + f"TOP_K_{self.max_output_tokens}", + str(self.top_k), + )), + "max_output_tokens": int(os.getenv( + f"MAX_TOKENS_{self.max_output_tokens}", + str(self.max_output_tokens), + )), + } + + +# Default per-intent config table +INTENT_CONFIGS: Dict[Intent, IntentConfig] = { + Intent.GREETING_SMALLTALK: IntentConfig( + instruction_snippet=( + "This is a greeting or casual small-talk. Respond warmly in " + "1-2 sentences, return the salam/salutation properly, and " + "invite the user to ask an Islamic knowledge question." + ), + temperature=0.5, + max_output_tokens=256, + ), + Intent.FACTUAL_KNOWLEDGE: IntentConfig( + instruction_snippet=( + "Provide a structured, thorough answer grounded in authentic " + "Islamic sources. Organize the answer clearly with headings " + "or sections where appropriate. Cite Quran surah:ayah and " + "authentic Hadith where possible. Follow with 2-3 suggested " + "follow-up questions in a delimited block." + ), + temperature=0.7, + max_output_tokens=2048, + ), + Intent.FIQH_RULING: IntentConfig( + instruction_snippet=( + "Provide a fiqh ruling based on authentic Islamic sources. " + "Acknowledge differences of scholarly opinion where relevant. " + "Be balanced and cite sources. Follow with 2-3 suggested " + "follow-up questions in a delimited block." + ), + temperature=0.6, + max_output_tokens=2048, + ), + Intent.PERSONAL_GUIDANCE: IntentConfig( + instruction_snippet=( + "Respond with a compassionate, supportive tone. Make du'a " + "for the user where fitting. For serious personal matters, " + "gently recommend consulting a qualified scholar or " + "professional. Keep the answer concise and comforting." + ), + temperature=0.8, + max_output_tokens=1024, + ), + Intent.PLATFORM_QUESTION: IntentConfig( + instruction_snippet=( + "Answer concisely and practically about the Deen Bridge " + "platform, its courses, features, or the zakat calculator. " + "Keep it brief and actionable." + ), + temperature=0.5, + max_output_tokens=1024, + ), + Intent.OUT_OF_SCOPE: IntentConfig( + instruction_snippet=( + "Politely explain that this question is outside your scope " + "as an Islamic-knowledge assistant. If the question seems " + "harmful or inappropriate, gently redirect. Do not attempt " + "to answer the question." + ), + temperature=0.5, + max_output_tokens=512, + ), +} + + +def get_intent_config(intent: Intent) -> IntentConfig: + """Return the generation config for *intent*, falling back to factual_knowledge.""" + return INTENT_CONFIGS.get(intent, INTENT_CONFIGS[Intent.FACTUAL_KNOWLEDGE]) + + +# --------------------------------------------------------------------------- +# Suggested follow-ups — parse a delimited block from the model response +# --------------------------------------------------------------------------- + +# Delimiter used by the model to wrap follow-up suggestions. +# Using a distinctive marker unlikely to appear in natural prose. +FOLLOWUP_START = "" +FOLLOWUP_END = "" +# Alternative delimiter (XML-style) for robustness +FOLLOWUP_START_ALT = "[[FOLLOWUPS]]" +FOLLOWUP_END_ALT = "[[/FOLLOWUPS]]" + + +def parse_followups(response_text: str) -> List[str]: + """Extract 2-3 suggested follow-up questions from *response_text*. + + Returns an empty list if parsing fails — never raises. + """ + if not response_text: + return [] + + text = response_text + + # Try primary delimiter + followups = _extract_delimited_block(text, FOLLOWUP_START, FOLLOWUP_END) + + # Try alternative delimiter + if not followups: + followups = _extract_delimited_block(text, FOLLOWUP_START_ALT, FOLLOWUP_END_ALT) + + return followups + + +def _extract_delimited_block(text: str, start_delim: str, end_delim: str) -> List[str]: + """Extract and parse a delimited block of follow-up questions.""" + start_idx = text.find(start_delim) + if start_idx == -1: + return [] + + end_idx = text.find(end_delim, start_idx + len(start_delim)) + if end_idx == -1: + return [] + + block = text[start_idx + len(start_delim):end_idx].strip() + if not block: + return [] + + # Split by newlines and extract list items + items = [] + for line in block.split("\n"): + line = line.strip() + if not line: + continue + # Strip leading numbering, bullets, quotes + cleaned = _clean_followup_line(line) + if cleaned: + items.append(cleaned) + + # Return at most 5 items + return items[:5] + + +def _clean_followup_line(line: str) -> str: + """Clean a single follow-up line: strip numbering, bullets, quotes.""" + # Remove leading numbering like "1.", "2)", "1." + line = re.sub(r'^\s*\d+[.)]\s*', '', line).strip() + # Remove leading bullets (including Unicode bullet \u2022) + line = re.sub(r'^\s*[\*\-\u2022]\s*', '', line).strip() + # Remove surrounding quotes + line = re.sub(r'^["\'](.*)["\']$', r'\1', line).strip() + return line + + +def strip_followup_block(response_text: str) -> str: + """Remove the follow-up delimited block from *response_text* so it never + leaks into the visible response. + """ + text = response_text + + # Primary delimiter + text = _remove_block(text, FOLLOWUP_START, FOLLOWUP_END) + # Alternative delimiter + text = _remove_block(text, FOLLOWUP_START_ALT, FOLLOWUP_END_ALT) + + return text.strip() + + +def _remove_block(text: str, start_delim: str, end_delim: str) -> str: + start_idx = text.find(start_delim) + if start_idx == -1: + return text + + end_idx = text.find(end_delim, start_idx + len(start_delim)) + if end_idx == -1: + # No closing delimiter — remove from start to end + return text[:start_idx].strip() + + # Remove the entire block including delimiters + return text[:start_idx].strip() + text[end_idx + len(end_delim):] + + +# --------------------------------------------------------------------------- +# No-double-clarification guard +# --------------------------------------------------------------------------- + + +def should_clarify( + classification: ClassificationResult, + last_classification: Optional[ClassificationResult], +) -> bool: + """Return True if the assistant should ask a clarifying question. + + Never clarifies twice in a row for the same session: if the *last* + classification already had ``needs_clarification`` set, this returns + False so the assistant answers the follow-up reply directly. + """ + if not classification.needs_clarification: + return False + + # Don't clarify twice in a row + if last_classification and last_classification.needs_clarification: + return False + + return True diff --git a/main.py b/main.py index 91d7310..3207be4 100644 --- a/main.py +++ b/main.py @@ -6,7 +6,7 @@ import os from dotenv import load_dotenv import logging -from typing import Any, List, Optional +from typing import Any, Dict, List, Optional import uuid from stellar import router as stellar_router @@ -25,6 +25,15 @@ normalize_madhhab, ) from hadith import HADITH_ADAB_CONTEXT, HadithReference, annotate as annotate_hadith, build_caution_note +from intent import ( + Intent, + ClassificationResult, + classify_intent, + get_intent_config, + parse_followups, + strip_followup_block, + should_clarify, +) from study import router as study_router from tafsir import ( TafsirContext, @@ -92,6 +101,8 @@ # Store active chats active_chats = {} +# Track the last classification per session to avoid double-clarification +_last_classifications: Dict[str, Optional[ClassificationResult]] = {} # Islamic context and safety instructions ISLAMIC_CONTEXT = """You are an AI assistant specialized in providing Islamic knowledge and guidance. @@ -138,6 +149,10 @@ class ChatResponse(BaseModel): hadith_references: Optional[List[HadithReference]] = None tafsir: Optional[TafsirInfo] = None confidence: Optional[ConfidenceAssessment] = None + intent: Optional[str] = None + needs_clarification: bool = False + clarifying_question: str = "" + suggested_followups: List[str] = [] def classify_for_safety(prompt: str, candidate_ids: List[str]): @@ -230,6 +245,54 @@ async def chat(request: ChatRequest, http_request: Request, fastapi_response: Re is_fiqh = classify_fiqh(request.prompt) fiqh_info = FiqhInfo(is_fiqh_question=is_fiqh, madhhab_requested=madhhab) + # --- Intent classification --- + # A lightweight classifier (deterministic short-circuit for greetings, + # one Gemini call for everything else) that guides answer shape, length, + # and whether to ask a clarifying question before answering. + import time + _intent_start = time.monotonic() + classifier_model = genai.GenerativeModel( + model_name="gemini-1.5-flash", + generation_config=genai.types.GenerationConfig( + temperature=0.1, + max_output_tokens=256, + ), + ) + classification = classify_intent( + request.prompt, + model_callable=classifier_model.generate_content, + ) + logger.info( + "Intent: %s | chat_id=%s | ambiguous=%s | latency=%.0fms", + classification.intent.value, + chat_id, + classification.needs_clarification, + (time.monotonic() - _intent_start) * 1000, + ) + + # --- Clarify-before-answering --- + # When a question is genuinely ambiguous, ask one clarifying question + # instead of guessing. The no-double-clarification guard ensures we + # never ask twice in a row for the same session. + last_cls = _last_classifications.get(chat_id) + if should_clarify(classification, last_cls): + _last_classifications[chat_id] = classification + clarifying_question = classification.clarifying_question or ( + "Could you please provide more detail so I can help you better?" + ) + return ChatResponse( + response=clarifying_question, + chat_id=chat_id, + history=[], + intent=classification.intent.value, + needs_clarification=True, + clarifying_question=clarifying_question, + ) + _last_classifications[chat_id] = classification + + # --- Per-intent generation config --- + intent_config = get_intent_config(classification.intent) + # --- Tafsir retrieval for verse-explanation questions --- # Detection is offline (regex + the bundled surah index), so a # non-tafsir prompt costs nothing. @@ -274,6 +337,7 @@ async def chat(request: ChatRequest, http_request: Request, fastapi_response: Re history=cached.history, fiqh=fiqh_info, hadith_references=annotate_hadith(cached.response), + intent=classification.intent.value, ) elif is_bypass: semantic_cache.bypasses += 1 @@ -288,7 +352,9 @@ def generate(safety_prompt: str) -> str: ) active_chats[chat_id] = model.start_chat(history=[]) + # Build system context with per-intent instruction snippet system_context = ISLAMIC_CONTEXT + HADITH_ADAB_CONTEXT + system_context += f"\n\nINTENT-SPECIFIC GUIDANCE:\n{intent_config.instruction_snippet}" if is_fiqh: system_context += FIQH_IKHTILAF_CONTEXT if madhhab: @@ -301,10 +367,10 @@ def generate(safety_prompt: str) -> str: response = active_chats[chat_id].send_message( full_prompt, generation_config={ - "temperature": 0.7, - "top_p": 0.8, - "top_k": 40, - "max_output_tokens": 2048, + "temperature": intent_config.temperature, + "top_p": intent_config.top_p, + "top_k": intent_config.top_k, + "max_output_tokens": intent_config.max_output_tokens, } ) if not response.text: @@ -348,6 +414,12 @@ def generate(safety_prompt: str) -> str: response_text = safety_result.text if safety_result else generated_text + # --- Suggested follow-ups --- + # Parse follow-up questions from the model response, then strip the + # raw delimited block so it never leaks into the visible answer. + suggested_followups = parse_followups(response_text) + response_text = strip_followup_block(response_text) + # --- Hadith authenticity grading --- # Baked into response_text *before* the cache write so a cached hit # replays the same caution the user originally saw. @@ -359,13 +431,26 @@ def generate(safety_prompt: str) -> str: # --- Confidence, abstention, and scholar escalation --- # is_religious and is_high_stakes reuse classification that already ran # this turn (the fiqh classifier and the hadith annotator) rather than - # adding a competing classifier. self_consistency (#ai-18) and - # citation_verification (#40) are passed through when those components - # supply them; until then they are simply absent from the average. + # adding a competing classifier. The intent classifier enriches this: + # fiqh_ruling and out_of_scope are treated as high-stakes, while + # factual_knowledge and personal_guidance are religious content. + # self_consistency (#ai-18) and citation_verification (#40) are passed + # through when those components supply them; until then they are simply + # absent from the average. + is_high_stakes = is_fiqh or classification.intent in ( + Intent.FIQH_RULING, + Intent.OUT_OF_SCOPE, + ) + is_religious = ( + is_fiqh + or bool(hadith_refs) + or classification.intent + in (Intent.FACTUAL_KNOWLEDGE, Intent.PERSONAL_GUIDANCE, Intent.FIQH_RULING) + ) signals = build_signals( response_text, - is_religious=is_fiqh or bool(hadith_refs), - is_high_stakes=is_fiqh, + is_religious=is_religious, + is_high_stakes=is_high_stakes, ) assessment = assess(signals) answer_before_policy = response_text @@ -426,6 +511,8 @@ def generate(safety_prompt: str) -> str: hadith_references=hadith_refs, tafsir=tafsir_info, confidence=assessment, + intent=classification.intent.value, + suggested_followups=suggested_followups, ) except Exception as e: @@ -439,9 +526,9 @@ async def delete_chat(chat_id: str): try: if chat_id in active_chats: del active_chats[chat_id] - logger.info(f"Deleted chat session: {chat_id}") - return {"message": "Chat session deleted successfully"} - return {"message": "Chat session not found"} + _last_classifications.pop(chat_id, None) + logger.info(f"Deleted chat session: {chat_id}") + return {"message": "Chat session deleted successfully"} except Exception as e: error_msg = f"❌ Error deleting chat: {str(e)}" logger.error(error_msg) diff --git a/tests/test_intent.py b/tests/test_intent.py new file mode 100644 index 0000000..022758f --- /dev/null +++ b/tests/test_intent.py @@ -0,0 +1,565 @@ +""" +Tests for the question-understanding pipeline: intent classification, +classifying questions, answer-length calibration, and suggested follow-ups. + +All tests are offline (mocked Gemini client where needed). +""" + +import json + +import pytest + +from intent import ( + Intent, + ClassificationResult, + classify_intent, + _is_trivial_greeting, + _detect_ambiguity, + get_intent_config, + parse_followups, + strip_followup_block, + should_clarify, + AMBIGUITY_THRESHOLD, +) + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +class FakeModelResponse: + """Simulates genai's response object for testing.""" + + def __init__(self, text: str): + self.text = text + + +class FakeModel: + """Simulates model.generate_content for testing.""" + + def __init__(self, response_text: str): + self.response_text = response_text + + def generate_content(self, prompt: str): + return FakeModelResponse(self.response_text) + + +# Successful classification responses +CLASSIFICATION_FIXTURES = { + "greeting": { + "intent": "greeting_smalltalk", + "needs_clarification": False, + "clarifying_question": "", + "ambiguity_confidence": 0.0, + }, + "factual": { + "intent": "factual_knowledge", + "needs_clarification": False, + "clarifying_question": "", + "ambiguity_confidence": 0.1, + }, + "fiqh": { + "intent": "fiqh_ruling", + "needs_clarification": False, + "clarifying_question": "", + "ambiguity_confidence": 0.2, + }, + "ambiguous": { + "intent": "fiqh_ruling", + "needs_clarification": True, + "clarifying_question": "Which madhhab are you asking about?", + "ambiguity_confidence": 0.85, + }, + "personal": { + "intent": "personal_guidance", + "needs_clarification": False, + "clarifying_question": "", + "ambiguity_confidence": 0.1, + }, + "platform": { + "intent": "platform_question", + "needs_clarification": False, + "clarifying_question": "", + "ambiguity_confidence": 0.0, + }, + "out_of_scope": { + "intent": "out_of_scope", + "needs_clarification": False, + "clarifying_question": "", + "ambiguity_confidence": 0.0, + }, +} + + +def _make_model(intent_key: str, wrap_markdown: bool = False) -> FakeModel: + """Create a FakeModel that returns the given classification fixture.""" + data = CLASSIFICATION_FIXTURES[intent_key] + raw = json.dumps(data) + if wrap_markdown: + raw = f"```json\n{raw}\n```" + return FakeModel(raw) + + +# --------------------------------------------------------------------------- +# Deterministic short-circuit tests +# --------------------------------------------------------------------------- + + +class TestDeterministicShortCircuit: + """The common greeting case must add nearly zero latency — no LLM call.""" + + def test_salam_exact(self): + """'Assalamu alaykum' should be classified without an LLM call.""" + result = _is_trivial_greeting("Assalamu alaykum") + assert result is True + + def test_salam_variants(self): + """Common salam variants must be caught.""" + for msg in [ + "Salam", + "Salaam", + "Assalamo Alaikum", + "as-salam", + "As-salamu alaykum", + "Wa alaykum assalam", + ]: + assert _is_trivial_greeting(msg), f"Failed for: {msg}" + + def test_english_greetings(self): + """Common English greetings must be caught when short.""" + for msg in ["Hi", "Hello", "Hey", "Good morning", "Peace"]: + result = _is_trivial_greeting(msg) + assert result is True, f"Failed for: {msg}" + + def test_longer_greeting_not_caught(self): + """A longer message that starts with a greeting word should NOT be + short-circuited, as it likely contains a substantive question.""" + msg = "Hi I have a question about wudu" + assert _is_trivial_greeting(msg) is False + + def test_substantive_message_not_greeting(self): + """A substantive question should NOT be caught by the short-circuit.""" + for msg in [ + "What is the ruling on interest?", + "Can you explain Surah Al-Fatiha?", + "How should I pray?", + ]: + assert _is_trivial_greeting(msg) is False, f"Failed for: {msg}" + + def test_empty_message_not_greeting(self): + assert _is_trivial_greeting("") is False + + def test_classify_trivial_greeting_no_model(self): + """classify_intent with a trivial greeting should NOT require a model.""" + result = classify_intent("Assalamu alaykum") + assert result.intent == Intent.GREETING_SMALLTALK + assert result.needs_clarification is False + + +# --------------------------------------------------------------------------- +# LLM-based classification tests +# --------------------------------------------------------------------------- + + +class TestLLMClassification: + """Tests that exercise the Gemini-based classifier path.""" + + def test_classify_factual_knowledge(self): + model = _make_model("factual") + result = classify_intent("What does Islam say about charity?", model.generate_content) + assert result.intent == Intent.FACTUAL_KNOWLEDGE + assert result.needs_clarification is False + + def test_classify_fiqh_ruling(self): + model = _make_model("fiqh") + result = classify_intent("Is eating pork haram?", model.generate_content) + assert result.intent == Intent.FIQH_RULING + + def test_classify_personal_guidance(self): + model = _make_model("personal") + result = classify_intent("I'm going through a hard time, please make dua", model.generate_content) + assert result.intent == Intent.PERSONAL_GUIDANCE + + def test_classify_platform_question(self): + model = _make_model("platform") + result = classify_intent("How does the zakat calculator work?", model.generate_content) + assert result.intent == Intent.PLATFORM_QUESTION + + def test_classify_out_of_scope(self): + model = _make_model("out_of_scope") + result = classify_intent("What is the meaning of life according to Nietzsche?", model.generate_content) + assert result.intent == Intent.OUT_OF_SCOPE + + def test_classify_ambiguous_question(self): + model = _make_model("ambiguous") + result = classify_intent("Is music haram?", model.generate_content) + assert result.intent == Intent.FIQH_RULING + assert result.needs_clarification is True + assert len(result.clarifying_question) > 0 + + def test_unknown_intent_falls_back_to_factual(self): + """If the LLM returns an invalid intent, fallback to factual_knowledge.""" + model = FakeModel(json.dumps({"intent": "cooking_recipe", "needs_clarification": False})) + result = classify_intent("How do I make biryani?", model.generate_content) + assert result.intent == Intent.FACTUAL_KNOWLEDGE + + def test_malformed_json_uses_safe_fallback(self): + """Unparseable JSON from the classifier should fall back gracefully.""" + model = FakeModel("this is not json") + result = classify_intent("What is the meaning of life?", model.generate_content) + # Falls back to factual_knowledge + assert result.intent == Intent.FACTUAL_KNOWLEDGE + assert result.needs_clarification is False + + def test_markdown_wrapped_json(self): + """Handle markdown code fences around the JSON output.""" + model = _make_model("factual", wrap_markdown=True) + result = classify_intent("Explain Surah Ikhlas", model.generate_content) + assert result.intent == Intent.FACTUAL_KNOWLEDGE + + def test_ambiguity_below_threshold_not_clarified(self): + """When ambiguity_confidence is below threshold, don't clarify.""" + data = CLASSIFICATION_FIXTURES["ambiguous"].copy() + data["ambiguity_confidence"] = AMBIGUITY_THRESHOLD - 0.1 + data["needs_clarification"] = True + model = FakeModel(json.dumps(data)) + result = classify_intent("Is music haram?", model.generate_content) + assert result.needs_clarification is False, ( + "Should not clarify when confidence below threshold" + ) + + def test_needs_clarification_without_question_falls_back(self): + """If needs_clarification is True but no clarifying_question given, + the module provides a safe fallback.""" + data = CLASSIFICATION_FIXTURES["ambiguous"].copy() + data["clarifying_question"] = "" + data["ambiguity_confidence"] = 0.9 + model = FakeModel(json.dumps(data)) + result = classify_intent("Is music haram?", model.generate_content) + assert result.needs_clarification is True + assert len(result.clarifying_question) > 0 + + +# --------------------------------------------------------------------------- +# Per-intent generation config tests +# --------------------------------------------------------------------------- + + +class TestIntentConfig: + """Each intent maps to an appropriate config (answer shape, length, tone).""" + + def test_greeting_config_is_short(self): + cfg = get_intent_config(Intent.GREETING_SMALLTALK) + assert cfg.max_output_tokens <= 256 + assert cfg.temperature <= 0.7 + assert "greeting" in cfg.instruction_snippet.lower() or "salam" in cfg.instruction_snippet.lower() + + def test_factual_config_is_thorough(self): + cfg = get_intent_config(Intent.FACTUAL_KNOWLEDGE) + assert cfg.max_output_tokens >= 1024 + assert "structured" in cfg.instruction_snippet.lower() + + def test_fiqh_config_cites_sources(self): + cfg = get_intent_config(Intent.FIQH_RULING) + assert cfg.max_output_tokens >= 1024 + assert "fiqh" in cfg.instruction_snippet.lower() or "ruling" in cfg.instruction_snippet.lower() + + def test_personal_guidance_config_compassionate(self): + cfg = get_intent_config(Intent.PERSONAL_GUIDANCE) + assert cfg.max_output_tokens <= 1024 + assert "compassionate" in cfg.instruction_snippet.lower() or "supportive" in cfg.instruction_snippet.lower() + + def test_platform_config_is_concise(self): + cfg = get_intent_config(Intent.PLATFORM_QUESTION) + assert cfg.max_output_tokens <= 1024 + assert "concisely" in cfg.instruction_snippet.lower() or "actionable" in cfg.instruction_snippet.lower() + + def test_out_of_scope_config_deflects(self): + cfg = get_intent_config(Intent.OUT_OF_SCOPE) + assert "outside your scope" in cfg.instruction_snippet.lower() + + def test_unknown_intent_falls_back_to_factual(self): + cfg = get_intent_config(Intent.GREETING_SMALLTALK) + assert cfg is not None + + def test_all_intents_have_configs(self): + for intent in [ + Intent.GREETING_SMALLTALK, + Intent.FACTUAL_KNOWLEDGE, + Intent.FIQH_RULING, + Intent.PERSONAL_GUIDANCE, + Intent.PLATFORM_QUESTION, + Intent.OUT_OF_SCOPE, + ]: + cfg = get_intent_config(intent) + assert cfg is not None, f"Missing config for {intent}" + assert len(cfg.instruction_snippet) > 0 + + +# --------------------------------------------------------------------------- +# Follow-up parsing tests +# --------------------------------------------------------------------------- + + +class TestFollowUpParsing: + """Defensive parse-or-degrade pattern: never raises, never leaks delimiters.""" + + def test_parse_well_formed_followups(self): + text = ( + "Here is the answer about wudu.\n\n" + "\n" + "1. What invalidates wudu?\n" + "2. How is tayammum performed?\n" + "3. What are the sunnah acts of wudu?\n" + "" + ) + result = parse_followups(text) + assert len(result) == 3 + assert "What invalidates wudu?" in result + assert "How is tayammum performed?" in result + assert "What are the sunnah acts of wudu?" in result + + def test_parse_alternative_delimiter(self): + text = ( + "Here is the answer.\n\n" + "[[FOLLOWUPS]]\n" + "* What is the first question?\n" + "* What is the second question?\n" + "[[/FOLLOWUPS]]" + ) + result = parse_followups(text) + assert len(result) == 2 + assert "What is the first question?" in result + + def test_missing_block_returns_empty(self): + text = "Just a plain answer with no follow-ups." + result = parse_followups(text) + assert result == [] + + def test_malformed_block_returns_empty(self): + """A block that has a start delimiter but no end returns empty list.""" + text = "Answer here.\n\n\n1. Orphan question" + result = parse_followups(text) + assert result == [] + + def test_empty_block_returns_empty(self): + text = "Answer.\n\n\n\n" + result = parse_followups(text) + assert result == [] + + def test_strip_followup_block(self): + text = ( + "Visible answer here.\n\n" + "\n" + "1. Follow-up?\n" + "" + ) + stripped = strip_followup_block(text) + assert stripped == "Visible answer here." + assert "FOLLOWUPS" not in stripped + + def test_strip_followup_block_alternative_delimiter(self): + text = ( + "Visible answer.\n\n" + "[[FOLLOWUPS]]\n" + "1. Follow-up?\n" + "[[/FOLLOWUPS]]" + ) + stripped = strip_followup_block(text) + assert stripped == "Visible answer." + assert "FOLLOWUPS" not in stripped + + def test_strip_does_not_remove_content_without_block(self): + text = "Just a plain answer with no follow-ups." + assert strip_followup_block(text) == text + + def test_mixed_bullets_and_numbers(self): + text = ( + "Answer.\n\n" + "\n" + "- What invalidates wudu?\n" + "* How is tayammum performed?\n" + "1. What are the sunnah acts?\n" + "" + ) + result = parse_followups(text) + assert len(result) == 3 + assert "What invalidates wudu?" in result + assert "How is tayammum performed?" in result + assert "What are the sunnah acts?" in result + + def test_empty_text_returns_empty(self): + assert parse_followups("") == [] + assert parse_followups(None) == [] + + def test_maximum_five_followups(self): + """At most 5 follow-ups are returned, even if more are present.""" + items = "\n".join(f"{i}. Question {i}?" for i in range(1, 10)) + text = f"Answer.\n\n\n{items}\n" + result = parse_followups(text) + assert len(result) <= 5 + + +# --------------------------------------------------------------------------- +# No-double-clarification guard tests +# --------------------------------------------------------------------------- + + +class TestNoDoubleClarification: + """Never clarify twice in a row for the same session.""" + + def test_first_clarification_allowed(self): + cls = ClassificationResult( + intent=Intent.FIQH_RULING, + needs_clarification=True, + clarifying_question="Which madhhab?", + ) + assert should_clarify(cls, None) is True + + def test_second_clarification_blocked(self): + cls = ClassificationResult( + intent=Intent.FIQH_RULING, + needs_clarification=True, + clarifying_question="Which madhhab?", + ) + last = ClassificationResult( + intent=Intent.FIQH_RULING, + needs_clarification=True, + ) + assert should_clarify(cls, last) is False + + def test_nonambiguous_not_clarified(self): + cls = ClassificationResult( + intent=Intent.FACTUAL_KNOWLEDGE, + needs_clarification=False, + ) + assert should_clarify(cls, None) is False + + def test_different_intent_allows_clarification(self): + """If last was a different intent, clarification is allowed again + (e.g. the user changed the subject).""" + cls = ClassificationResult( + intent=Intent.FIQH_RULING, + needs_clarification=True, + clarifying_question="Which madhhab?", + ) + last = ClassificationResult( + intent=Intent.GREETING_SMALLTALK, + needs_clarification=False, + ) + assert should_clarify(cls, last) is True + + +# --------------------------------------------------------------------------- +# Ambiguity detection tests +# --------------------------------------------------------------------------- + + +class TestAmbiguityDetection: + """Offline keyword-based ambiguity detection.""" + + def test_haram_question_detected(self): + needs, question = _detect_ambiguity("Is music haram?") + assert needs is True + assert len(question) > 0 + + def test_what_breaks_detected(self): + needs, question = _detect_ambiguity("What breaks the fast?") + assert needs is True + + def test_specific_question_not_ambiguous(self): + needs, question = _detect_ambiguity("What are the five pillars of Islam?") + # The five pillars question is not inherently ambiguous + # It may or may not match triggers depending on the implementation + # Just check that it returns a valid result + assert isinstance(needs, bool) + assert isinstance(question, str) + + def test_greeting_not_ambiguous(self): + needs, question = _detect_ambiguity("Assalamu alaykum") + assert needs is False + assert isinstance(question, str) + + +# --------------------------------------------------------------------------- +# Edge cases +# --------------------------------------------------------------------------- + + +class TestEdgeCases: + """Edge cases and boundary conditions.""" + + def test_non_trivial_message_requires_model(self): + """classify_intent should raise when called on non-trivial message + without a model_callable.""" + with pytest.raises(RuntimeError): + classify_intent("What is the ruling on interest?", None) + + def test_long_message_classified(self): + model = _make_model("factual") + long_msg = "Can you explain " + "the concept of " * 50 + "tawheed?" + result = classify_intent(long_msg, model.generate_content) + assert result.intent in Intent + + def test_special_characters(self): + model = _make_model("factual") + result = classify_intent("What does Quran 2:255 mean?", model.generate_content) + assert result.intent == Intent.FACTUAL_KNOWLEDGE + + def test_bullet_followup_with_different_markers(self): + """Test that Unicode bullet markers are handled correctly.""" + text = ( + "Answer.\n\n" + "\n" + "• First question?\n" + "• Second question?\n" + "" + ) + result = parse_followups(text) + # Unicode bullet (U+2022) is now handled + assert len(result) == 2 + assert "First question?" in result + assert "Second question?" in result + + +# --------------------------------------------------------------------------- +# Integration: end-to-end intent flow +# --------------------------------------------------------------------------- + + +class TestIntentFlow: + """Higher-level test of the complete classify_intent → config → clarify flow.""" + + def test_greeting_flow(self): + """Greeting → short answer, no clarification, appropriate config.""" + result = classify_intent("Assalamu alaykum") + assert result.intent == Intent.GREETING_SMALLTALK + assert result.needs_clarification is False + + cfg = get_intent_config(result.intent) + assert cfg.max_output_tokens <= 256 + # No clarifying question + assert not should_clarify(result, None) + + def test_ambiguous_flow(self): + """Ambiguous fiqh question → clarification requested, config is fiqh.""" + model = _make_model("ambiguous") + result = classify_intent("Is music haram?", model.generate_content) + assert result.intent == Intent.FIQH_RULING + assert result.needs_clarification is True + + # First turn: clarification allowed + assert should_clarify(result, None) is True + + # Second turn: clarification blocked (same session) + assert should_clarify(result, result) is False + + def test_factual_followups_config(self): + """Factual knowledge config should mention follow-ups in instruction.""" + cfg = get_intent_config(Intent.FACTUAL_KNOWLEDGE) + assert "follow-up" in cfg.instruction_snippet.lower() + + def test_fiqh_followups_config(self): + """Fiqh ruling config should mention follow-ups in instruction.""" + cfg = get_intent_config(Intent.FIQH_RULING) + assert "follow-up" in cfg.instruction_snippet.lower()