""" Sakhi — AI Learning Companion (App UI & Orchestrator) Preserves robust Gradio UI and Mermaid JS while adding Hands-Free Wake Word and Intents. Wake word fixes applied: 1. Language changed from hi-IN to en-IN (Hinglish-friendly) 2. Wake word + command in same breath now works (no longer discarded) 3. rec.onend now calls startListeningLoop() instead of rec.start() on a dead object 4. All wake word detection gated on isFinal only (no interim false triggers) Gradio 6 migration notes: - gr.Chatbot now requires the "messages" format: a flat list of {"role": "user"|"assistant", "content": str} dicts, NOT [user, bot] pairs. - theme/css/js/head moved from Blocks() constructor to launch(). - show_api removed from launch(). """ import logging import time import re import os import tempfile import gradio as gr from gtts import gTTS from config import APP_TITLE, MAX_CHAT_HISTORY from llm_client import LLMClient from intent_engine import IntentDetector from rag_engine import RAGEngine from diagram_builder import render_mermaid_html from quiz_engine import ( QuizState, detect_quiz_trigger, parse_quiz_response, format_quiz_question_html, format_answer_feedback_html, format_quiz_results_html, ) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # NOTE: The gradio_client schema monkeypatch from the 4.44.x debugging era has been # removed now that the Space runs Gradio 6.19.0, where that bug no longer applies. def _strip_json_from_chunk(text: str) -> str: """Remove JSON payloads that leak into streaming spoken-text chunks. Handles: ```json fences, bare `json{`, and raw `{"spoken"` / `{"display"` starts. Returns only the portion of text *before* the JSON begins. """ if not text: return text # 1. Fenced JSON blocks for marker in ('```json', '```'): idx = text.find(marker) if idx != -1: return text[:idx] # 2. Bare "json{" (some models emit this without fences) import re as _re m = _re.search(r'\bjson\s*\{', text, _re.IGNORECASE) if m: return text[:m.start()] # 3. Raw JSON object starting with known keys for key in ('{"spoken"', '{"display"', '{ "spoken"', '{ "display"'): idx = text.find(key) if idx != -1: return text[:idx] return text def _clean_accumulated_spoken(spoken: str) -> str: """Scrub any trailing JSON that leaked into the accumulated spoken text.""" if not spoken: return spoken import re as _re # Strip from first JSON-like object boundary for pattern in (r'```json.*', r'```.*', r'\bjson\s*\{.*', r'\{\s*"(?:spoken|display)".*'): spoken = _re.sub(pattern, '', spoken, flags=_re.DOTALL | _re.IGNORECASE) return spoken.rstrip() llm = LLMClient() rag = RAGEngine() intent_detector = IntentDetector(llm) def _clean_tts_text(text: str) -> str: """Fix common gTTS Hinglish mispronunciations.""" import re # Replace standalone "ab" with "abb" so gTTS doesn't read it as the musical note "A-flat" text = re.sub(r'\bab\b', 'abb', text, flags=re.IGNORECASE) # Also replace "Ab" at the start of sentences text = re.sub(r'\bAb\b', 'Abb', text) return text def text_to_audio(text: str) -> str | None: logger.info(f"text_to_audio called | raw_len={len(text) if text else 0} | preview={(text[:80] if text else None)!r}") if not text or not text.strip(): logger.warning("text_to_audio: empty/whitespace text, returning None") return None cleaned = _clean_tts_text(text.strip()) try: tts = gTTS(text=cleaned, lang='hi') tmp_path = os.path.join(tempfile.gettempdir(), f"sakhi_tts_{int(time.time()*1000)}.mp3") tts.save(tmp_path) size = os.path.getsize(tmp_path) if os.path.exists(tmp_path) else 0 logger.info(f"text_to_audio: saved {tmp_path} ({size} bytes)") if size < 512: logger.error(f"gTTS tiny file | text_preview={cleaned[:120]!r}") return None return tmp_path except Exception as e: logger.error(f"gTTS error | text_preview={cleaned[:120]!r} | error={type(e).__name__}: {e}") return None def format_chat_history(chat_history: list) -> str: """chat_history is now a flat list of {"role": ..., "content": ...} dicts (Gradio 6 messages format), not [user, bot] pairs.""" if not chat_history: return "No previous conversation." lines = [] # MAX_CHAT_HISTORY was previously counted in turns (pairs); each turn is now # 2 messages, so take the last MAX_CHAT_HISTORY*2 messages to preserve behavior. for msg in chat_history[-MAX_CHAT_HISTORY * 2:]: if not isinstance(msg, dict): continue role = msg.get("role") content = msg.get("content", "") if not content: continue speaker = "Student" if role == "user" else "Sakhi" lines.append(f"{speaker}: {content}") return "\n".join(lines) if lines else "No previous conversation." # ─── Core Processing ──────────────────────────────────────────────────────── def process_text_input(text, chat_history, q_state, last_ctx): if not text or not text.strip(): yield (create_status_html("ready"), gr.update(), "", None, chat_history, q_state, last_ctx, text, chat_history) return chat_history.append({"role": "user", "content": text}) chat_history.append({"role": "assistant", "content": ""}) intent_data = intent_detector.detect(text) intent = intent_data.get("intent", "Explain") topic = intent_data.get("topic", text) is_quiz, q_topic, num_questions = detect_quiz_trigger(text) if intent == "Quiz" or is_quiz: yield from process_quiz_input(text, q_topic or topic, num_questions, chat_history, q_state, last_ctx) return if q_state.is_active: yield from process_quiz_answer(text, chat_history, q_state, last_ctx) return yield from process_explain(text, intent, chat_history, q_state, last_ctx) def process_explain(transcript, intent, chat_history, q_state, last_ctx): yield (create_status_html("thinking"), gr.update(), "", None, chat_history, q_state, last_ctx, "", chat_history) is_greeting = transcript.strip().lower() in ["hi", "hello", "hey", "hi sakhi", "नमस्ते"] if not isinstance(last_ctx, dict): last_ctx = {"question": "", "rag_context": "", "spoken": "", "memories": []} if "memories" not in last_ctx or not isinstance(last_ctx.get("memories"), list): last_ctx["memories"] = [] search_query = transcript if not is_greeting and last_ctx.get("memories"): search_query = f"{last_ctx['memories'][-1]} {transcript}" chunks = [] if is_greeting else rag.retrieve(search_query) base_rag_context = rag.format_context(chunks) history_str = format_chat_history(chat_history) recent_memories = last_ctx.get("memories", [])[-6:] memory_section = "".join([f"Memory {i+1}: {m}\n" for i, m in enumerate(recent_memories)]) rag_context_full = ( base_rag_context + "\n\n" + "Previous Summaries:\n" + memory_section if memory_section else base_rag_context ) history_str = format_chat_history(chat_history) last_ctx["question"] = transcript last_ctx["rag_context"] = rag_context_full last_ctx["spoken"] = "" spoken_full = "" display_raw = "" diagram_json = None try: for part in llm.generate_spoken_and_display_stream(transcript, intent, rag_context_full, history_str): if not isinstance(part, dict): continue if part.get("type") == "chunk": text = _strip_json_from_chunk(part.get("text", "") or "") if text.strip(): spoken_full += text chat_history[-1]["content"] = _clean_accumulated_spoken(spoken_full) yield (create_status_html("speaking"), gr.update(), text, None, chat_history, q_state, last_ctx, "", chat_history) elif part.get("type") == "result": combined = part.get("result") or {} spoken_full = (combined.get("spoken") or spoken_full) or "" display_raw = combined.get("display") or "" diagram_json = combined.get("diagram") chat_history[-1]["content"] = spoken_full last_ctx["spoken"] = spoken_full try: summary = llm.summarize_text(spoken_full) if summary: last_ctx.setdefault("memories", []).append(summary) except Exception as e: logger.error(f"Failed to summarize response: {e}") diagram_html = "" if diagram_json: try: # Extract raw mermaid code directly from the new JSON format mermaid_code = diagram_json.get("code", "") if mermaid_code: diagram_html = render_mermaid_html(mermaid_code) except Exception: diagram_html = "" display_html = create_display_html(display_raw, diagram_html) audio_path = text_to_audio(spoken_full) logger.info(f"process_explain: yielding audio_path={audio_path!r}") yield (create_status_html("ready"), display_html, spoken_full, audio_path, chat_history, q_state, last_ctx, "", chat_history) break except Exception as e: logger.error(f"Streaming combined generation failed: {e}") spoken_full = "Mujhe maaf kijiye, thoda technical issue ho gaya hai. Dobara try karen." display_html = create_display_html("", "") chat_history[-1]["content"] = spoken_full yield (create_status_html("ready"), display_html, spoken_full, None, chat_history, q_state, last_ctx, "", chat_history) def process_quiz_input(transcript, topic, num_questions, chat_history, q_state, last_ctx): generic_triggers = { "quiz", "start quiz", "a quiz", "quiz please", "start", "take a quiz", "do a quiz", "start the quiz", "it", "this", "on it", "on this", "about it", "about this", "is", "ispe", "ispar", "iss", "iss pe", "iss par", "is pe", "is par", "isme", "ismein", "is topic par", "is topic pe", "is baare mein", "is bare me", "isko" } lower_topic = topic.strip().lower() if isinstance(topic, str) else "" is_generic = not topic if lower_topic in generic_triggers: is_generic = True else: clean = lower_topic.replace("take a", "").replace("give me a", "").replace("start", "").replace("quiz", "") clean = clean.replace("about", "").replace("on", "").replace("it", "").replace("this", "") clean = clean.replace("ispar", "").replace("ispe", "").replace("is", "").replace("par", "").replace("pe", "").replace("baare mein", "").replace("topic", "") clean = clean.strip() if not clean: is_generic = True if is_generic: if isinstance(last_ctx, dict): prev = last_ctx.get("question") or last_ctx.get("rag_context") if prev and isinstance(prev, str) and prev.strip() and prev.strip().lower() not in generic_triggers: topic = prev else: topic = transcript else: topic = transcript yield (create_status_html("thinking"), create_display_html("Generating Quiz... ⏳", ""), "", None, chat_history, q_state, last_ctx, "", chat_history) if not isinstance(last_ctx, dict): last_ctx = {"question": "", "rag_context": "", "spoken": "", "memories": []} if is_generic and last_ctx.get("rag_context"): rag_context_full = last_ctx.get("rag_context") else: search_query = topic if last_ctx.get("memories"): search_query = f"{last_ctx['memories'][-1]} {topic}" chunks = rag.retrieve(search_query) base_rag_context = rag.format_context(chunks) recent_memories = last_ctx.get("memories", [])[-6:] memory_section = "".join([f"Memory {i+1}: {m}\n" for i, m in enumerate(recent_memories)]) rag_context_full = ( base_rag_context + "\n\n" + "Previous Summaries:\n" + memory_section if memory_section else base_rag_context ) quiz_data = llm.generate_quiz(topic, num_questions or 5, rag_context_full) if not quiz_data: err_msg = "Abhi quiz generate nahi ho paya. Kya hum dobara try karein?" chat_history[-1]["content"] = err_msg yield (create_status_html("ready"), create_display_html("Could not generate quiz.", ""), err_msg, text_to_audio(err_msg), chat_history, q_state, last_ctx, "", chat_history) return questions, quiz_intro, quiz_summary = parse_quiz_response(quiz_data) q_state.reset() q_state.questions = questions q_state.total = len(questions) q_state.is_active = True q_state.topic = topic if questions: question_html = format_quiz_question_html(questions[0], 0, len(questions)) q_text = questions[0].spoken_question or questions[0].question spoken = q_text chat_history[-1]["content"] = spoken # Use pre-generated quiz summary for memory (no extra LLM call) memory_entry = quiz_summary or f"Quiz started on {topic} with {len(questions)} questions." last_ctx.setdefault("memories", []).append(memory_entry) display_html = create_display_html("", question_html) yield (create_status_html("quiz"), display_html, spoken, text_to_audio(spoken), chat_history, q_state, last_ctx, "", chat_history) else: chat_history[-1]["content"] = "No questions found." yield (create_status_html("ready"), create_display_html("No questions found.", ""), "Error.", text_to_audio("Error."), chat_history, q_state, last_ctx, "", chat_history) def process_quiz_answer(transcript, chat_history, q_state, last_ctx): if not q_state.current_question: q_state.is_active = False yield (create_status_html("ready"), create_display_html("Quiz Finished!", ""), "Quiz khatam!", text_to_audio("Quiz khatam!"), chat_history, q_state, last_ctx, "", chat_history) return # Grab the question BEFORE check_answer increments current_index answered_q = q_state.current_question is_correct, explanation = q_state.check_answer(transcript) if is_correct is None: # User said nothing or just the wake word; do not advance the quiz. spoken = explanation chat_history[-1]["content"] = spoken # Re-render current question next_html = format_quiz_question_html(answered_q, q_state.current_index, q_state.total) display_html = create_display_html("", next_html) yield (create_status_html("quiz"), display_html, spoken, text_to_audio(spoken), chat_history, q_state, last_ctx, "", chat_history) return feedback_html = format_answer_feedback_html( is_correct, answered_q.correct, explanation, transcript, ) # Use pre-generated spoken feedback from the quiz generation call if is_correct: spoken = answered_q.feedback_correct or f"Bilkul sahi! {explanation}" else: spoken = answered_q.feedback_incorrect or f"Galat! Sahi answer hai {answered_q.correct}. {explanation}" # Use pre-generated per-question summary (no LLM call needed) memory_entry = answered_q.summary if memory_entry: last_ctx.setdefault("memories", []).append(memory_entry) if q_state.is_finished: results_html = format_quiz_results_html(q_state) # Display only results, hide feedback text as requested display_html = create_display_html("", results_html) spoken += f" Quiz khatam ho gaya! Tumhara score hai {q_state.score} out of {q_state.total}. Koi aur topic padhna hai?" q_state.is_active = False chat_history[-1]["content"] = spoken yield (create_status_html("ready"), display_html, spoken, text_to_audio(spoken), chat_history, q_state, last_ctx, "", chat_history) else: next_q = q_state.current_question next_html = format_quiz_question_html(next_q, q_state.current_index, q_state.total) # Display only the next question, hide feedback text as requested display_html = create_display_html("", next_html) q_text = next_q.spoken_question or f"Agla sawaal: {next_q.question}" spoken += f" {q_text}" chat_history[-1]["content"] = spoken yield (create_status_html("quiz"), display_html, spoken, text_to_audio(spoken), chat_history, q_state, last_ctx, "", chat_history) from ui_assets import create_status_html, create_display_html, create_welcome_html, CUSTOM_CSS, HEAD_HTML, CUSTOM_JS def create_app(): with gr.Blocks() as app: chat_history = gr.State([]) q_state = gr.State(QuizState()) last_ctx = gr.State({"question": "", "rag_context": "", "spoken": "", "memories": []}) # Hidden I/O elements spoken_text = gr.Textbox(elem_id="spoken-text-output", elem_classes=["hidden-text"]) hidden_quiz_input = gr.Textbox(elem_id="hidden-quiz-input", elem_classes=["hidden-text"]) audio_output = gr.Audio(visible=True, autoplay=True, elem_id="main-audio-out") gr.HTML( '
' '

' '📚 Sakhi — AI Learning Companion

' ) with gr.Row(elem_classes=["main-row"]): with gr.Column(scale=2): status_display = gr.HTML(value=create_status_html("ready")) main_display = gr.HTML(value=create_welcome_html()) with gr.Row(): clear_btn = gr.Button("🗑️ Clear Session", variant="stop") with gr.Column(scale=1): hands_free_toggle = gr.Checkbox( label="🎙️ Hands-Free Mode (Say 'Hi Sakhi')", value=False, elem_id="hands-free-toggle", ) # Gradio 6.19: "messages" format ({"role": ..., "content": ...} dicts) # is the ONLY supported format — the `type` kwarg was removed entirely, # so it must not be passed. chatbot = gr.Chatbot(label="Chat History", height=300) audio_input = gr.Audio(sources=["microphone"], type="filepath", label="Manual Voice Input") with gr.Row(): text_input = gr.Textbox( label="Type or Edit Input", placeholder="Ask a question...", lines=1, elem_id="main-text-input", scale=4, ) text_submit_btn = gr.Button("Send", variant="primary", elem_id="text-submit-btn", scale=1) inputs_list = [text_input, chat_history, q_state, last_ctx] outputs_list = [status_display, main_display, spoken_text, audio_output, chat_history, q_state, last_ctx, text_input, chatbot] audio_input.stop_recording( fn=lambda audio: (llm.transcribe_audio(audio) if audio else "", None), inputs=[audio_input], outputs=[text_input, audio_input], ) text_submit_btn.click(fn=process_text_input, inputs=inputs_list, outputs=outputs_list) text_input.submit(fn=process_text_input, inputs=inputs_list, outputs=outputs_list) hidden_quiz_input.change( fn=process_text_input, inputs=[hidden_quiz_input, chat_history, q_state, last_ctx], outputs=outputs_list, ) clear_btn.click( fn=lambda: ( [], QuizState(), {"question": "", "rag_context": "", "spoken": ""}, create_status_html("ready"), create_welcome_html(), [], "", "", None, ), outputs=[chat_history, q_state, last_ctx, status_display, main_display, chatbot, text_input, spoken_text, audio_output], ) return app from build_index import ensure_index_exists if __name__ == "__main__": logger.info("Starting Sakhi Initialization...") try: import threading import build_index def _ensure_index_bg(): try: build_index.ensure_index_exists() except Exception as e: logger.exception("Background FAISS build failed: %s", e) t = threading.Thread(target=_ensure_index_bg, daemon=True) t.start() # Also initialize the RAG transformer model in background to reduce first-use latency def _init_rag_bg(): try: rag.initialize() logger.info("RAG transformer initialized in background.") except Exception as e: logger.exception("RAG background initialization failed: %s", e) r = threading.Thread(target=_init_rag_bg, daemon=True) r.start() except Exception as e: logger.error(f"Failed to start background index builder: {e}") logger.info("Launching the Web UI...") app = create_app() app.launch( server_name="0.0.0.0", server_port=7860, theme=gr.themes.Soft(), css=CUSTOM_CSS, js=CUSTOM_JS, head=HEAD_HTML, )