Prof-chaos-5
feat: improve TTS pronunciation for Hinglish and refine language rules to mandate English technical terminology
c0d1884 | """ | |
| 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( | |
| '<div style="text-align:center;padding:10px 0;">' | |
| '<h1 style="font-size:32px;background:linear-gradient(135deg,#10b981,#3b82f6);' | |
| '-webkit-background-clip:text;-webkit-text-fill-color:transparent;margin:0;">' | |
| '📚 Sakhi — AI Learning Companion</h1></div>' | |
| ) | |
| 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, | |
| ) |