""" ByteAstra — Core Tutoring Engine. This is the domain-agnostic orchestrator that powers every tutoring interaction. It takes a `domain` parameter and coordinates: 1. Loading domain configuration (prompts, thresholds, retrieval settings) 2. Condensing follow-up queries that contain pronouns into standalone queries 3. Retrieving relevant chunks from ChromaDB 4. TWO-TIER grounding check: Tier 1 (Pre-emptive): If no chunks clear the relevance threshold, skip the LLM entirely and serve the canned refusal directly. Tier 2 (Post-hoc): If the LLM produces OUT_OF_SYLLABUS despite chunks clearing the threshold, the stream buffer catches it before any text reaches the client. 5. Assembling the prompt (system + context + history + query) 6. Streaming the LLM completion with human-readable citation expansion IMPORTANT: This file must contain ZERO domain-specific logic. If you find yourself writing `if domain == "ayurveda":`, that belongs in the domain's YAML config file, not here. """ from __future__ import annotations import logging import re import json import asyncio from dataclasses import dataclass from functools import lru_cache from pathlib import Path from typing import AsyncGenerator import yaml from app.services import rag, llm logger = logging.getLogger(__name__) # Path to domain config files (relative to this module) _DOMAINS_DIR = Path(__file__).parent.parent / "domains" OUT_OF_SYLLABUS_SENTINEL = "OUT_OF_SYLLABUS" # Pronouns that indicate a follow-up query needing condensation _FOLLOW_UP_PRONOUNS = frozenset([ "it", "its", "they", "them", "their", "theirs", "this", "that", "these", "those", "he", "she", "his", "her", "what about", "and what", "how about", ]) @dataclass class TutoringResult: """Final assembled result from the engine (non-streaming path).""" content: str citations: list[rag.RetrievedChunk] is_grounded: bool def _levenshtein_distance(s1: str, s2: str) -> int: """Calculate the Levenshtein distance between two strings.""" if len(s1) < len(s2): return _levenshtein_distance(s2, s1) if len(s2) == 0: return len(s1) previous_row = range(len(s2) + 1) for i, c1 in enumerate(s1): current_row = [i + 1] for j, c2 in enumerate(s2): insertions = previous_row[j + 1] + 1 deletions = current_row[j] + 1 substitutions = previous_row[j] + (c1 != c2) current_row.append(min(insertions, deletions, substitutions)) previous_row = current_row return previous_row[-1] def _clean_stem_word(w: str) -> str: """Normalize and stem a word to handle plurals simply.""" w = w.lower().strip() if w.endswith("ies") and len(w) > 4: w = w[:-3] + "y" elif w.endswith("es") and len(w) > 3: w = w[:-2] elif w.endswith("s") and len(w) > 3: w = w[:-1] return w def _sanskrit_phonetic_hash(w: str) -> str: """ Generate a simple phonetic key for a Sanskrit transliterated word. Aligns common spelling differences (e.g. sh/s, v/b, cch/ch, ri/r). """ w = w.lower().strip() w = w.replace("sh", "s") w = w.replace("s'", "s") w = w.replace("ś", "s") w = w.replace("ṣ", "s") w = w.replace("v", "b") w = w.replace("cch", "ch") w = w.replace("chh", "ch") w = w.replace("ri", "r") w = w.replace("ṛ", "r") w = w.replace("dh", "d") w = w.replace("th", "t") w = w.replace("ph", "p") w = w.replace("kh", "k") w = w.replace("gh", "g") w = w.replace("bh", "b") w = w.replace("jh", "j") w = w.replace("aa", "a") w = w.replace("ā", "a") w = w.replace("ee", "i") w = w.replace("ī", "i") w = w.replace("oo", "u") w = w.replace("ū", "u") return w # Load synonyms dictionary _SYNONYMS_FILE = _DOMAINS_DIR / "ayurveda_synonyms.json" _SYNONYMS = {} try: if _SYNONYMS_FILE.exists(): with open(_SYNONYMS_FILE, "r", encoding="utf-8") as f: _SYNONYMS = json.load(f) logger.info("Loaded %d synonym mappings.", len(_SYNONYMS)) except Exception as e: logger.error("Failed to load synonyms file: %s", e) @lru_cache(maxsize=16) def _load_domain_config(domain: str) -> dict: """ Load and cache a domain's YAML config. Raises FileNotFoundError if the domain config doesn't exist. """ config_path = _DOMAINS_DIR / f"{domain}.yaml" if not config_path.exists(): raise FileNotFoundError( f"No domain config found for '{domain}'. " f"Expected file at: {config_path}" ) with open(config_path, "r", encoding="utf-8") as f: config = yaml.safe_load(f) logger.info("Loaded domain config: %s", domain) return config def _needs_condensation(query: str) -> bool: """ Detect if a query is a follow-up that requires pronoun resolution. Returns True if the query contains any follow-up pronouns at the start or as isolated words. """ q_lower = query.lower().strip() tokens = set(q_lower.split()) # Check for standalone pronouns if tokens & _FOLLOW_UP_PRONOUNS: return True # Check for multi-word phrases for phrase in ("what about", "and what", "how about"): if q_lower.startswith(phrase): return True return False async def condense_query(query: str, history: list[dict], llm_cfg: dict) -> str: """ Reformulate a pronoun-containing follow-up query into a standalone query by asking the LLM to resolve contextual references using the recent history. Returns the condensed query string. On failure (LLM offline / empty output), falls back to the original query unchanged. The intermediate condensed string is logged at INFO level so callers and tests can assert on it without modifying this function's signature. """ if not _needs_condensation(query): return query # Build a minimal condensation prompt using only the last 4 turns of history recent = history[-4:] if len(history) >= 4 else history condensation_prompt = [ { "role": "system", "content": ( "You are a query rewriter. Your ONLY job is to rewrite the follow-up question " "to resolve pronouns (like 'it', 'that', 'this', 'they') using the conversation history.\n\n" "CRITICAL RULES:\n" "1. Preserve the exact question and all new topics introduced in the follow-up.\n" "2. Output ONLY the rewritten question as a single sentence. Do not add any preamble, explanation, or notes.\n\n" "EXAMPLE:\n" "History:\n" "User: What are the qualities of Pitta dosha?\n" "Follow-up: How does that relate to Ama (toxins)?\n" "Output: How does Pitta dosha relate to Ama (toxins)?" ), } ] condensation_prompt.extend(recent) condensation_prompt.append({"role": "user", "content": f"Follow-up question to rewrite: {query}"}) condensed_tokens: list[str] = [] try: async for token in llm.stream_completion( condensation_prompt, model=llm_cfg.get("model_name"), temperature=0.0, max_tokens=128, ): condensed_tokens.append(token) except Exception as exc: logger.warning("condense_query failed, using original query: %s", exc) return query condensed = "".join(condensed_tokens).strip() if not condensed: logger.warning("condense_query returned empty string, using original query.") return query logger.info("condense_query: '%s' → '%s'", query, condensed) return condensed def make_citations_readable(text: str, domain_config: dict) -> str: """ Expand abbreviated citation codes in `text` to human-readable form using mappings loaded from the domain config's `citation_mappings` section. Two levels of expansion are applied: 1. Composite codes (e.g. CS.Su.8 → "Charaka Samhita, Sutrasthana, Chapter 8") Matched via a dynamically compiled regex from the YAML keys. 2. Bare book+chapter codes (e.g. CS.19 → "Charaka Samhita, Chapter 19") Applied as a fallback after composite expansion, so it never clobbers a composite match. Mapping keys starting with "_" (internal config keys like _book_names) are skipped when building the citation regex. """ citation_mappings: dict = domain_config.get("citation_mappings", {}) if not citation_mappings: return text # ── Tier 1: Composite sthana codes ───────────────────────────────────────── # Sort by key length descending so longer (more specific) patterns match first sorted_keys = sorted( (k for k in citation_mappings if not k.startswith("_")), key=len, reverse=True, ) for key in sorted_keys: full_name = citation_mappings[key] # Match KEY followed by an optional chapter number: e.g. CS.Su or CS.Su.8 pattern = re.compile( r"\b" + re.escape(key) + r"(?:\.(\d+))?\b" ) def _replace_composite(m: re.Match, _full_name: str = full_name) -> str: chapter_num = m.group(1) if chapter_num: return f"{_full_name}, Chapter {chapter_num}" return _full_name text = pattern.sub(_replace_composite, text) # ── Tier 2: Bare book+chapter fallback ───────────────────────────────────── # Handles bare codes like CS.19, SS.5, AH.3 that have no sthana subdivision. book_names: dict = citation_mappings.get("_book_names", {}) if book_names: bare_abbrevs = "|".join(re.escape(abbr) for abbr in book_names.keys()) bare_pattern = re.compile(r"\b(" + bare_abbrevs + r")\.(\d+)\b") def _replace_bare(m: re.Match) -> str: abbr = m.group(1) chapter_num = m.group(2) book = book_names.get(abbr, abbr) return f"{book}, Chapter {chapter_num}" text = bare_pattern.sub(_replace_bare, text) return text def estimate_tokens(text: str) -> int: # Heuristic for Qwen/llama token counts (~4 chars/token) return max(1, len(text) // 4) def condense_context(chunks: list[rag.RetrievedChunk], max_context_tokens: int = 700) -> tuple[list[rag.RetrievedChunk], int]: selected = [] running = 0 for chunk in chunks: cost = estimate_tokens(chunk.content) if running + cost > max_context_tokens: break selected.append(chunk) running += cost return selected, running def _build_prompt( domain_config: dict, query: str, chunks: list[rag.RetrievedChunk], history: list[dict], ) -> list[dict]: """ Assemble the full message list for the LLM. Format: [system_message, ...filtered_history, context_injection + user_message] NOTE on history deduplication: main.py persists the user's current message to MongoDB *before* calling get_recent_history(). This means the history list already contains the current user turn. We filter it out here (by removing any trailing user-role message that matches the query exactly) to avoid sending the query twice in the prompt. """ prompts = domain_config.get("prompts", {}), # Handle the case where yaml loaded as a tuple due to trailing comma bug guard if isinstance(prompts, tuple): prompts = prompts[0] system_prompt = prompts.get("system", "You are a helpful tutoring assistant.") context_header = prompts.get("context_header", "--- CONTEXT ---") # ── Deduplicate: remove the current query from the tail of history ────────── filtered_history = list(history) if ( filtered_history and filtered_history[-1].get("role") == "user" and filtered_history[-1].get("content", "").strip() == query.strip() ): filtered_history = filtered_history[:-1] # ── Build context block from retrieved chunks ──────────────────────────────── context_parts = [context_header] for i, chunk in enumerate(chunks, 1): source_info = f"Source: {chunk.source}" if chunk.chapter: source_info += f", {chunk.chapter}" if chunk.section: source_info += f", {chunk.section}" context_parts.append(f"\n{chunk.content}\n({source_info})") context_block = "\n".join(context_parts) # ── Tutor synthesis instruction (reinforces system prompt at query level) ───────────── synthesis_instruction = ( "\n\n[INSTRUCTION] The context above contains raw textbook extracts with noise (chapter titles, author names, navigation text). " "IGNORE all chapter/section titles, author/publication lines, and '[back to top]' links. " "EXTRACT and EXPLAIN the actual Ayurvedic content: definitions, qualities (Gunas), sub-types, functions, and clinical significance. " "Write a detailed, comprehensive BAMS exam answer (aiming for 600-800 words) starting DIRECTLY with the topic (not a book title). " "List ALL items in any classification exhaustively (e.g. all 5 Vata sub-types, all 7 Dhatus, all 3 Nyayas). " "Explain each concept thoroughly using BAMS curriculum knowledge. Treat parallel classical theories, therapies, or sub-doshas as independent sibling concepts and do NOT nest them under each other. " "Use Sanskrit terms with English meanings in brackets. " "End with exactly ONE 'Sources:' section at the very end." ) # Final user message: context block + question + tutor instruction user_message_with_context = ( f"{context_block}\n\n--- STUDENT QUESTION ---\n{query}{synthesis_instruction}" ) messages = [{"role": "system", "content": system_prompt}] messages.extend(filtered_history) messages.append({"role": "user", "content": user_message_with_context}) return messages def is_conversational_query(query: str) -> bool: """Check if the user query is conversational (greeting, thanks, identity, help, creator).""" q = query.lower().strip().strip("?!.,🌿") greetings = { "hello", "hi", "hey", "hola", "greetings", "good morning", "good afternoon", "good evening", "namaste", "pranam", "helo", "helllo", "hellllo" } identities = { "who are you", "what is your name", "what are you", "tell me about yourself", "tell me about you", "what can you do", "how can you help", "how do you help", "who created you", "who made you", "who created this", "who made this", "who developed you", "who developed this", "creator", "developer", "risu solutions" } thanks = {"thank you", "thanks", "thank you so much", "dhanyavad", "dhanyawad", "thank"} help_queries = {"help", "how to use this", "how to use", "how does this work"} # Word boundary matches to prevent substring false positives if any(re.search(r"\b" + re.escape(g) + r"\b", q) for g in greetings) or re.search(r"\btutor\b", q): return True if any(re.search(r"\b" + re.escape(i) + r"\b", q) for i in identities) or re.search(r"\bwhat can you do\b", q) or re.search(r"\bhow can you help\b", q): return True if any(re.search(r"\b" + re.escape(t) + r"\b", q) for t in thanks): return True if any(re.search(r"\b" + re.escape(h) + r"\b", q) for h in help_queries): return True # Check if the query is very short and contains basic words words = q.split() if len(words) <= 3 and any(w in greetings or w in help_queries or w in {"ok", "okay", "yes", "no", "sure"} for w in words): return True return False def get_conversational_response_fallback(query: str) -> str: """Return a high-quality static response for conversational queries if the LLM server is offline.""" q = query.lower().strip().strip("?!.,🌿") greetings = { "hello", "hi", "hey", "hola", "greetings", "good morning", "good afternoon", "good evening", "namaste", "pranam", "helo", "helllo" } identities = { "who are you", "what is your name", "what are you", "tell me about yourself", "tell me about you", "what can you do", "how can you help", "how do you help", "who created", "who made", "developer", "creator", "risu solutions" } thanks = {"thank you", "thanks", "thank you so much", "dhanyavad", "dhanyawad"} if any(g in q for g in greetings) or "tutor" in q: return "Hello! I am ByteAstra, your BAMS tutoring assistant developed by Risu Solutions. How can I help you study the BAMS syllabus today?" if any(i in q for i in identities) or "what can you do" in q or "help" in q: return "I am ByteAstra, an AI-powered BAMS tutor developed by Risu Solutions. I help BAMS students study classical Ayurvedic texts like the Charaka Samhita, Sushruta Samhita, and Ashtanga Hridaya, providing grounded answers with verified citations." if any(t in q for t in thanks): return "You're very welcome! Let me know if you have any other questions from the syllabus." return "Hello! I am ByteAstra, your BAMS tutoring assistant developed by Risu Solutions. Ask me any question from the BAMS syllabus to get started." async def stream_answer( domain: str, query: str, history: list[dict], ) -> AsyncGenerator[dict, None]: """ Main streaming entry point for the engine. Yields dicts with shape: {"type": "status", "engine_mode": "hybrid"|"fallback"} {"type": "citations", "citations": [...], "is_grounded": True/False} {"type": "delta", "content": "..."} (repeated) {"type": "done"} On error: {"type": "error", "error": "...message"} Grounding architecture — TWO TIERS: ┌─────────────────────────────────────────────────────────────────────┐ │ Tier 1 (Pre-emptive): Retrieval threshold check BEFORE the LLM. │ │ • If no chunk exceeds relevance_threshold → serve canned refusal │ │ directly, without calling the LLM at all. │ │ • Eliminates state-bleed: LLM never inspects old history turns. │ ├─────────────────────────────────────────────────────────────────────┤ │ Tier 2 (Stream buffer): Safety net for chunks that pass threshold │ │ but the model still judges insufficient for the specific query. │ │ • Buffer stream until first newline OR 40 characters. │ │ • Strip whitespace; if buffer contains OUT_OF_SYLLABUS sentinel: │ │ - Cancel stream, emit is_grounded=False, empty citations, │ │ stream the clean canned refusal message. │ │ • Otherwise flush buffer and continue normal streaming. │ └─────────────────────────────────────────────────────────────────────┘ """ try: config = _load_domain_config(domain) except FileNotFoundError as e: yield {"type": "error", "error": str(e)} return try: retrieval_cfg = config.get("retrieval", {}) collection_name: str = retrieval_cfg.get("collection_name", f"domain_{domain}") top_k: int = retrieval_cfg.get("top_k", 3) # fewer chunks = smaller prompt threshold: float = retrieval_cfg.get("relevance_threshold", 0.45) history_turns: int = retrieval_cfg.get("history_turns", 3) # 3 turns = much faster llm_cfg = config.get("llm", {}) out_of_syllabus_msg: str = config.get("prompts", {}).get( "out_of_syllabus", "I couldn't find this in the current knowledge base.", ) is_online = await llm.check_connection() # ── Step 1: Adversarial speculation check ────────────────────────────────── # If the user explicitly asks us to guess, speculate, or claims the info is not in the textbook, # we refuse immediately to prevent state-bleed and speculation. def _is_adversarial(q: str) -> bool: q_low = q.lower() adversarial_indicators = [ "guess anyway", "best guess", "speculate", "speculation", "not in your textbook", "not in the textbook", "outside the textbook", "bypass constraints" ] return any(ind in q_low for ind in adversarial_indicators) if _is_adversarial(query): logger.info("Adversarial speculation detected in query, triggering immediate out-of-syllabus refusal.") engine_mode = "hybrid" if is_online else "fallback" yield {"type": "status", "engine_mode": engine_mode} yield {"type": "citations", "citations": [], "is_grounded": False} for word in out_of_syllabus_msg.split(" "): yield {"type": "delta", "content": word + " "} yield {"type": "done"} return # ── Step 1.2: Early Conversational Query Check ────────────────────────────── if is_conversational_query(query): logger.info("Conversational query detected: '%s'", query[:80]) engine_mode = "hybrid" if is_online else "fallback" yield {"type": "status", "engine_mode": engine_mode} yield {"type": "citations", "citations": [], "is_grounded": True} if is_online: conv_prompt = [ { "role": "system", "content": ( "You are ByteAstra, a helpful AI tutoring assistant for BAMS (Bachelor of " "Ayurvedic Medicine and Surgery) students, developed by the engineering team at Risu Solutions.\n\n" "Your purpose is to guide students through the BAMS syllabus and classical Ayurvedic texts " "(like Charaka Samhita, Sushruta Samhita, and Ashtanga Hridaya) by providing detailed, exam-quality answers with verified citations.\n\n" "Introduce yourself as ByteAstra, created by Risu Solutions, and explain your purpose. " "Keep your response brief, friendly, and polite." ) } ] conv_prompt.extend(history[-6:]) conv_prompt.append({"role": "user", "content": query}) async for token in llm.stream_completion( conv_prompt, model=llm_cfg.get("model_name"), temperature=0.7, max_tokens=128, # conversational — keep short ): yield {"type": "delta", "content": token} else: fallback_msg = get_conversational_response_fallback(query) for word in fallback_msg.split(" "): yield {"type": "delta", "content": word + " "} await asyncio.sleep(0.01) yield {"type": "done"} return # Compute query embeddings for cache and search embedding_fn = rag._get_embedding_fn() query_embeddings = await asyncio.to_thread(embedding_fn, [query]) # ── Step 1.5: Semantic Cache Check ────────────────────────────────────────── cached_val = await check_semantic_cache(query_embeddings, f"answer_cache_{domain}") if cached_val: logger.info("Semantic cache HIT for query: %s", query[:80]) engine_mode = "hybrid" if is_online else "fallback" yield {"type": "status", "engine_mode": engine_mode} yield { "type": "citations", "citations": cached_val["citations"], "is_grounded": cached_val["is_grounded"] } yield {"type": "delta", "content": cached_val["answer"]} yield {"type": "done"} return # Query condensation DISABLED on CPU deployment (saves a full LLM round trip). # ── Step 2: Tier 1 — Pre-emptive grounding gate ────────────────────────────── chunks = await asyncio.to_thread( rag.retrieve, query, collection_name, top_k=top_k, relevance_threshold=threshold ) # ── Lexical query-to-context overlap verification (adversarial defense) ── if chunks: # Extract significant terms from query q_words = re.findall(r"\b[a-zA-Z]{3,}\b", query.lower()) generic_words = { # Question words / Pronouns / Conjunctions / Auxiliaries "what", "how", "does", "that", "relate", "explain", "describe", "discuss", "ayurveda", "ayurvedic", "according", "textbook", "samhita", "charaka", "sushruta", "chapter", "section", "treatment", "concept", "theory", "definition", "disease", "biological", "meaning", "define", "list", "give", "write", "detail", "with", "from", "when", "where", "which", "into", "your", "tell", "about", "study", "student", "syllabus", "text", "body", "human", "are", "the", "and", "for", "its", "why", "who", "whom", "whose", "this", "these", "those", "can", "could", "would", "should", "will", "shall", "been", "have", "has", "had", "was", "were", "srotas", "dosha", "dhatu", "mala", "agni", "vata", "pitta", "kapha", "tridosha", "difference", "between", "concept", "relation", "relationship", "function", "functions", "location", "locations", "subtype", "subtypes", "role", "roles", "effect", "effects", "cause", "causes", "symptom", "symptoms", "significance", "primary", "secondary", "major", "minor", "possible", "balance", "imbalance", "cured", "cure", "treat", "treatment", "therapy", "therapies", "therapeusis", "normal", "abnormal", "qualities", "quality", "sutra", "sutrasthana", "nidana", "chikitsa", "kalpa", "siddhi", # Common general English verbs (and variations) "interact", "interacts", "interacting", "interaction", "interactions", "affect", "affects", "affecting", "affected", "occur", "occurs", "occurring", "occurred", "happen", "happens", "happening", "happened", "manifest", "manifests", "manifesting", "manifested", "manifestation", "manifestations", "produce", "produces", "producing", "produced", "result", "results", "resulting", "resulted", "lead", "leads", "leading", "cause", "causes", "causing", "caused", "connect", "connects", "connecting", "connected", "connection", "connections", "compare", "compares", "comparing", "compared", "contrast", "contrasts", "contrasting", "contrasted", "differentiate", "differentiates", "differentiating", "distinguish", "distinguishes", "distinguishing", "identify", "identifies", "identifying", "show", "shows", "showing", "ask", "asks", "asking", "answer", "answers", "answering", "say", "says", "said", "saying", # Common general nouns / adjectives "concepts", "theories", "system", "systems", "process", "processes", "mechanism", "mechanisms", "action", "actions", "type", "types", "classification", "classifications", "class", "classes", "category", "categories", "group", "groups", "term", "terms", "word", "words", "name", "names", "definition", "definitions", "meaning", "meanings", "explanation", "explanations", "description", "descriptions", "difference", "differences", "relation", "relations", "relationship", "relationships", "association", "associations", "similar", "similarity", "similarities", "common", "various", "several", "many", "person", "patient", "individual", "people", "life", "health", "diseases", "disorder", "disorders", "illness", "illnesses", "condition", "conditions", "state", "states" } significant_query_words = [w for w in q_words if w not in generic_words] # Combine all retrieved content for search combined_chunks_text = " ".join(c.content.lower() for c in chunks) # Tokenize and stem all context words context_words_cleaned = {_clean_stem_word(w) for w in re.findall(r"\b[a-zA-Z]{3,}\b", combined_chunks_text)} # Check overlap of significant query words with the retrieved context missing_words = [] matched_count = 0 for word in significant_query_words: stemmed_q_word = _clean_stem_word(word) # Expand query word with its synonyms equivalent_words = {stemmed_q_word} for w_lookup in (word.lower(), stemmed_q_word): if w_lookup in _SYNONYMS: for syn in _SYNONYMS[w_lookup]: equivalent_words.add(_clean_stem_word(syn)) matched = False for eq_word in equivalent_words: eq_phonetic = _sanskrit_phonetic_hash(eq_word) for cw in context_words_cleaned: # Direct match or substring if eq_word in cw or cw in eq_word: matched = True break # Levenshtein spelling match limit = 1 if len(eq_word) <= 5 else 2 if _levenshtein_distance(eq_word, cw) <= limit: matched = True break # Phonetic spelling match (e.g. vasti / basti) if eq_phonetic == _sanskrit_phonetic_hash(cw): matched = True break if matched: break if matched: matched_count += 1 else: missing_words.append(word) # Require at least 50% of the significant words to match (or at least 1) min_required = max(1, len(significant_query_words) // 2) if significant_query_words else 0 if significant_query_words and matched_count < min_required: logger.info( "Lexical validation failed: matched only %d/%d significant terms (missing: %s). Setting chunks to empty.", matched_count, len(significant_query_words), missing_words ) chunks = [] if not chunks: # Check for conversational query before declaring out-of-syllabus if is_conversational_query(query): engine_mode = "hybrid" if is_online else "fallback" yield {"type": "status", "engine_mode": engine_mode} yield {"type": "citations", "citations": [], "is_grounded": True} if is_online: conv_prompt = [ { "role": "system", "content": ( "You are ByteAstra, a helpful AI tutoring assistant for BAMS (Bachelor of " "Ayurvedic Medicine and Surgery) students, developed by the engineering team at Risu Solutions.\n\n" "Your purpose is to guide students through the BAMS syllabus and classical Ayurvedic texts " "(like Charaka Samhita, Sushruta Samhita, and Ashtanga Hridaya) by providing detailed, exam-quality answers with verified citations.\n\n" "Introduce yourself as ByteAstra, created by Risu Solutions, and explain your purpose. " "Keep your response brief, friendly, and polite." ) } ] conv_prompt.extend(history[-6:]) conv_prompt.append({"role": "user", "content": query}) async for token in llm.stream_completion( conv_prompt, model=llm_cfg.get("model_name"), temperature=0.7, max_tokens=128, # conversational — keep short ): yield {"type": "delta", "content": token} else: fallback_msg = get_conversational_response_fallback(query) for word in fallback_msg.split(" "): yield {"type": "delta", "content": word + " "} await asyncio.sleep(0.01) yield {"type": "done"} return # Real out-of-syllabus: Tier 1 gate — no LLM call at all logger.info( "Tier 1 grounding gate triggered: no chunks above threshold for query '%s'", query[:80], ) engine_mode = "hybrid" if is_online else "fallback" yield {"type": "status", "engine_mode": engine_mode} yield {"type": "citations", "citations": [], "is_grounded": False} for word in out_of_syllabus_msg.split(" "): yield {"type": "delta", "content": word + " "} yield {"type": "done"} return # ── Step 3: Chunks cleared threshold — proceed with LLM ───────────────────── engine_mode = "hybrid" if is_online else "fallback" yield {"type": "status", "engine_mode": engine_mode} if not is_online: # LLM offline — serve canned fallback directly (no stream buffer needed) yield {"type": "citations", "citations": [], "is_grounded": False} for word in out_of_syllabus_msg.split(" "): yield {"type": "delta", "content": word + " "} yield {"type": "done"} return # Build prompt (history is trimmed + deduplicated inside _build_prompt) trimmed_history = history[-(history_turns * 2):] condensed_chunks, _ = condense_context(chunks, max_context_tokens=1400) messages = _build_prompt(config, query, condensed_chunks, trimmed_history) # ── Step 4: Tier 2 — Stream buffer for OUT_OF_SYLLABUS detection ──────────── # We do NOT emit citations before we know the model is grounded. # Buffer the first tokens until we hit a newline or accumulate 40 characters. buffer: list[str] = [] buffer_done = False is_grounded_confirmed = False citation_emitted = False async def _emit_citations(grounded: bool) -> None: """Helper to emit the citations event exactly once.""" nonlocal citation_emitted if citation_emitted: return citation_emitted = True if grounded: yield_data = { "type": "citations", "citations": [c.to_citation().model_dump() for c in condensed_chunks], "is_grounded": True, } else: yield_data = {"type": "citations", "citations": [], "is_grounded": False} # We can't yield from a nested function, so return the dict for the caller return yield_data async def _stream_with_buffer() -> AsyncGenerator[dict, None]: """ Inner generator that wraps the LLM stream with the Tier 2 buffer logic. Emits citations event AFTER we've confirmed the model is not refusing. """ nonlocal buffer_done, is_grounded_confirmed, citation_emitted raw_stream = llm.stream_completion( messages, model=llm_cfg.get("model_name"), temperature=llm_cfg.get("temperature"), max_tokens=llm_cfg.get("max_tokens"), ) full_response_text = [] tokens_yielded = 0 async for token in raw_stream: tokens_yielded += 1 if not buffer_done: buffer.append(token) combined = "".join(buffer) # Check termination conditions: newline hit OR 30+ chars accumulated hit_newline = "\n" in combined hit_length = len(combined.replace(" ", "").replace("\n", "")) >= 30 if hit_newline or hit_length: buffer_done = True normalized = combined.strip().replace("\n", "").upper() if OUT_OF_SYLLABUS_SENTINEL in normalized: # Tier 2 gate triggered: model said OUT_OF_SYLLABUS logger.info( "Tier 2 stream buffer triggered: model emitted OUT_OF_SYLLABUS " "for query '%s'", query[:80] ) is_grounded_confirmed = False citation_emitted = True yield {"type": "citations", "citations": [], "is_grounded": False} for word in out_of_syllabus_msg.split(" "): yield {"type": "delta", "content": word + " "} return # Cancel the stream — don't continue else: # Model is grounded — emit citations, then flush buffer is_grounded_confirmed = True citation_emitted = True yield { "type": "citations", "citations": [c.to_citation().model_dump() for c in condensed_chunks], "is_grounded": True, } # Flush buffer expanded = make_citations_readable(combined, config) full_response_text.append(expanded) yield {"type": "delta", "content": expanded} # Still accumulating buffer — don't yield yet else: # Post-buffer: emit tokens with citation expansion applied expanded = make_citations_readable(token, config) full_response_text.append(expanded) yield {"type": "delta", "content": expanded} # Edge case: stream ended while we were still buffering (very short response) if not buffer_done and buffer: buffer_done = True combined = "".join(buffer) normalized = combined.strip().replace("\n", "").upper() if OUT_OF_SYLLABUS_SENTINEL in normalized: is_grounded_confirmed = False citation_emitted = True yield {"type": "citations", "citations": [], "is_grounded": False} for word in out_of_syllabus_msg.split(" "): yield {"type": "delta", "content": word + " "} else: is_grounded_confirmed = True if not citation_emitted: citation_emitted = True yield { "type": "citations", "citations": [c.to_citation().model_dump() for c in condensed_chunks], "is_grounded": True, } expanded = make_citations_readable(combined, config) full_response_text.append(expanded) yield {"type": "delta", "content": expanded} # Empty stream fallback: if we got absolutely no tokens from the LLM, # we automatically fall back to the raw RAG context from the textbook. if tokens_yielded == 0: logger.warning("Local LLM server returned 0 tokens. Performing automatic textbook fallback.") is_grounded_confirmed = True if not citation_emitted: citation_emitted = True yield { "type": "citations", "citations": [c.to_citation().model_dump() for c in condensed_chunks], "is_grounded": True, } # Format fallback content from condensed_chunks context_parts = [] for chunk in condensed_chunks: context_parts.append(chunk.content) context_text = "\n\n".join(context_parts).strip() intro = "*[Connected directly to textbook database]*\n\nAccording to the classical texts:\n\n" for word in intro.split(" "): yield {"type": "delta", "content": word + " "} await asyncio.sleep(0.01) for part in context_text.split("\n\n"): for word in part.split(" "): yield {"type": "delta", "content": word + " "} await asyncio.sleep(0.01) yield {"type": "delta", "content": "\n\n"} # Cache the fallback response await write_semantic_cache( query, query_embeddings, intro + context_text, [c.to_citation().model_dump() for c in condensed_chunks], True, f"answer_cache_{domain}" ) elif is_grounded_confirmed: # Cache the generated LLM response final_text = "".join(full_response_text) await write_semantic_cache( query, query_embeddings, final_text, [c.to_citation().model_dump() for c in condensed_chunks], True, f"answer_cache_{domain}" ) async for event in _stream_with_buffer(): yield event yield {"type": "done"} except Exception as e: logger.exception("Error in stream_answer processing:") yield {"type": "error", "error": str(e)} ANSWER_CACHE_COLLECTION = "answer_cache" SIMILARITY_THRESHOLD = 0.92 async def check_semantic_cache(query_embeddings: list[list[float]], collection_name: str = ANSWER_CACHE_COLLECTION) -> dict | None: try: from app.services.rag import get_or_create_collection collection = get_or_create_collection(collection_name) # Query the cache collection using the query embeddings results = await asyncio.to_thread( collection.query, query_embeddings=query_embeddings, n_results=1, include=["documents", "metadatas", "distances"] ) if not results["distances"] or not results["distances"][0]: return None distance = results["distances"][0][0] similarity = 1.0 - distance if similarity >= SIMILARITY_THRESHOLD: meta = results["metadatas"][0][0] citations_raw = meta.get("citations", "[]") try: citations = json.loads(citations_raw) except Exception: citations = [] return { "answer": meta.get("answer", ""), "citations": citations, "is_grounded": meta.get("is_grounded", True) } except Exception as e: logger.warning("Semantic cache lookup failed: %s", e) return None async def write_semantic_cache( query: str, query_embeddings: list[list[float]], answer: str, citations: list[dict], is_grounded: bool = True, collection_name: str = ANSWER_CACHE_COLLECTION ) -> None: try: from app.services.rag import get_or_create_collection collection = get_or_create_collection(collection_name) doc_id = hashlib.sha256(query.encode("utf-8")).hexdigest() # Write/Update the cache record await asyncio.to_thread( collection.upsert, ids=[doc_id], embeddings=query_embeddings, documents=[query], metadatas=[{ "answer": answer, "citations": json.dumps(citations), "is_grounded": is_grounded, "query": query }] ) logger.info("Saved response to semantic cache for query: %s", query[:50]) except Exception as e: logger.warning("Failed to write to semantic cache: %s", e)