""" Sahel-Voice-Lab — Internal Edition (Phase 2: Voice Output) Stack (100% non-Meta): STT : openai/whisper-large-v3-turbo LLM : Qwen/Qwen2.5-72B-Instruct (or LLM_MODEL_ID env var) TTS : MALIBA-AI/bambara-tts (Bambara) | ous-sow/fula-tts (Fula, after training) Store: HF Dataset ous-sow/sahel-agri-feedback → vocabulary.jsonl Flow: 1. User presses Push-to-Talk → records audio 2. Whisper transcribes to text 3. MemoryManager injects current vocabulary into Gemma's system prompt 4. Gemma returns structured JSON: teaching → MemoryManager.add_word_pair() → push to Hub question → answer using vocabulary conversation → natural reply 5. UI shows Gemma's reply + last 5 learned words """ from __future__ import annotations import logging import os import sys import threading from pathlib import Path logger = logging.getLogger(__name__) import gradio as gr ROOT = Path(__file__).parent sys.path.insert(0, str(ROOT)) # ── Env ─────────────────────────────────────────────────────────────────────── HF_TOKEN = os.environ.get("HF_TOKEN") FEEDBACK_REPO_ID = os.environ.get("FEEDBACK_REPO_ID", "ous-sow/sahel-agri-feedback") WHISPER_MODEL_ID = os.environ.get("WHISPER_MODEL_ID", "openai/whisper-large-v3-turbo") LLM_MODEL_ID = os.environ.get("LLM_MODEL_ID", "Qwen/Qwen2.5-72B-Instruct") LANGUAGE_NAMES = { "bam": "Bambara", "ful": "Fula / Pular", "fr": "French", "en": "English", } # ── Singletons ──────────────────────────────────────────────────────────────── from src.memory.memory_manager import MemoryManager from src.llm.gemma_client import GemmaClient from src.tts.waxal_tts import WaxalTTSEngine _memory = MemoryManager(repo_id=FEEDBACK_REPO_ID, hf_token=HF_TOKEN) _gemma = GemmaClient(model_id=LLM_MODEL_ID, hf_token=HF_TOKEN) _tts = WaxalTTSEngine() # Whisper — loaded lazily in background _whisper_model = None _whisper_processor = None _whisper_lock = threading.Lock() _whisper_status = "not loaded" # ── Whisper loading ─────────────────────────────────────────────────────────── def _do_load_whisper() -> None: global _whisper_model, _whisper_processor, _whisper_status import torch try: from transformers.models.whisper import WhisperProcessor, WhisperForConditionalGeneration except ImportError: from transformers.models.whisper.processing_whisper import WhisperProcessor from transformers.models.whisper.modeling_whisper import WhisperForConditionalGeneration _whisper_status = "loading…" try: _whisper_processor = WhisperProcessor.from_pretrained( WHISPER_MODEL_ID, token=HF_TOKEN ) _whisper_model = WhisperForConditionalGeneration.from_pretrained( WHISPER_MODEL_ID, token=HF_TOKEN ) _whisper_model.eval() _whisper_status = f"ready ({WHISPER_MODEL_ID})" except Exception as exc: _whisper_status = f"error: {exc}" def _ensure_whisper() -> str: global _whisper_status with _whisper_lock: if _whisper_model is None and "loading" not in _whisper_status: _whisper_status = "loading…" threading.Thread(target=_do_load_whisper, daemon=True).start() return _whisper_status def _whisper_status_label() -> str: s = _ensure_whisper() if "ready" in s: return f"🟢 STT {s}" if "loading" in s: return f"🟡 STT {s}" if "error" in s: return f"🔴 STT {s}" return f"⚪ STT {s}" def _transcribe(audio_path: str, language_hint: str) -> str: """Run Whisper STT. Returns transcribed text.""" if _whisper_model is None: return "" import torch, librosa audio_np, _ = librosa.load(audio_path, sr=16_000, mono=True) with _whisper_lock: inputs = _whisper_processor.feature_extractor( audio_np, sampling_rate=16_000, return_tensors="pt" ) input_features = inputs.input_features # Whisper doesn't have Bambara / Fula tokens — let it auto-detect if language_hint in ("bam", "ful"): forced_ids = None else: try: forced_ids = _whisper_processor.get_decoder_prompt_ids( language=language_hint, task="transcribe" ) except Exception: forced_ids = None with torch.no_grad(): predicted_ids = _whisper_model.generate( input_features, forced_decoder_ids=forced_ids, max_new_tokens=256, ) return _whisper_processor.batch_decode(predicted_ids, skip_special_tokens=True)[0].strip() # ── Core pipeline ───────────────────────────────────────────────────────────── def _run_llm_and_tts( transcript: str, lang_code: str, history: list, source_label: str, ) -> tuple: """ Shared core: Gemma → memory update → TTS. Returns: (history, recent_words_md, status_msg, audio_tuple_or_None) """ # 1. Ask Gemma (with vocabulary context) vocab_ctx = _memory.get_vocabulary_context() llm_result = _gemma.chat(transcript, vocab_ctx) intent = llm_result.get("intent", "conversation") response = llm_result.get("response", "…") # 2. Persist teaching intent to memory if intent == "teaching": word = llm_result.get("word", transcript) lang = llm_result.get("language", lang_code) trans = llm_result.get("translation", "") trans_l = llm_result.get("translation_language", "en") if word and trans: _memory.add_word_pair(word, lang, trans, trans_l, source="user_taught") # 3. TTS — speak the response if language supported audio_out = None tts_result = _tts.synthesize(response, lang_code) if tts_result is not None: audio_out = WaxalTTSEngine.audio_to_gradio(*tts_result) # 4. Update chat history history = list(history or []) history.append({"role": "user", "content": f"[{LANGUAGE_NAMES.get(lang_code, lang_code)}] {transcript}"}) history.append({"role": "assistant", "content": response}) tts_status = "" if audio_out else " (TTS not available for this language yet)" status_msg = { "teaching": f"✅ Word learned and saved!{tts_status}", "question": f"💬 Answered from vocabulary.{tts_status}", "conversation": f"💬 Replied.{tts_status}", "error": "⚠️ LLM error.", }.get(intent, f"💬 Replied.{tts_status}") return history, _render_recent_words(), status_msg, audio_out def process_audio(audio_path, language_label: str, history: list) -> tuple: """ Full pipeline: audio → Whisper STT → Gemma → TTS. Returns: (history, recent_words_md, status_msg, audio_out) """ try: if audio_path is None: return history, _render_recent_words(), "⚠️ No audio recorded.", None lang_code = _label_to_code(language_label) status = _ensure_whisper() if _whisper_model is None: return history, _render_recent_words(), f"⏳ {status} — wait a moment and try again.", None transcript = _transcribe(audio_path, lang_code) if not transcript: return history, _render_recent_words(), "⚠️ Could not transcribe audio.", None return _run_llm_and_tts(transcript, lang_code, history, "voice") except Exception as exc: logger.exception("process_audio error") return history, _render_recent_words(), f"❌ Error: {exc}", None def process_text(text: str, language_label: str, history: list) -> tuple: """Text input path — Gemma → TTS. Returns: (history, recent_words_md, status_msg, audio_out)""" try: if not text.strip(): return history, _render_recent_words(), "⚠️ Please type something.", None lang_code = _label_to_code(language_label) return _run_llm_and_tts(text.strip(), lang_code, history, "text") except Exception as exc: logger.exception("process_text error") return history, _render_recent_words(), f"❌ Error: {exc}", None # ── Helpers ─────────────────────────────────────────────────────────────────── LANGUAGE_CHOICES = ["Bambara (bam)", "Fula (ful)", "French (fr)", "English (en)"] def _label_to_code(label: str) -> str: mapping = { "Bambara (bam)": "bam", "Fula (ful)": "ful", "French (fr)": "fr", "English (en)": "en", } return mapping.get(label, "bam") def _render_recent_words() -> str: recent = _memory.get_recent(5) if not recent: return "_No words learned yet. Start teaching me! Say something like: **'I ni ce means hello in Bambara'**_" lines = ["### 📖 Last 5 words learned\n"] for e in reversed(recent): lang = LANGUAGE_NAMES.get(e.get("language", "?"), e.get("language", "?")) word = e.get("word", "") tr = e.get("translation", "") tr_l = e.get("translation_language", "") lines.append(f"**{word}** `[{lang}]` → {tr} `({tr_l})`") return "\n\n".join(lines) # ── UI ──────────────────────────────────────────────────────────────────────── def build_ui() -> gr.Blocks: with gr.Blocks(title="Sahel-Voice-Lab", theme=gr.themes.Soft()) as demo: gr.Markdown( "# 🌍 Sahel-Voice-Lab — Internal Edition\n" "**Phase 1 · The Memory Loop** \n" "Teach me Bambara and Fula — I will remember every word you share." ) with gr.Row(): # ── Left column: input + voice output ──────────────────────────── with gr.Column(scale=2): def _full_status() -> str: stt = _whisper_status_label() tts = _tts.get_status() bam = "🟢" if tts["bam"] == "ready" else ("🟡" if "not" in tts["bam"] else "🔴") ful = "🟢" if tts["ful"] == "ready" else ("🟡" if "not" in tts["ful"] else "🔴") return f"{stt} | TTS Bambara {bam} | TTS Fula {ful}" status_box = gr.Textbox( value=_full_status(), label="System status", interactive=False, max_lines=1, ) status_timer = gr.Timer(value=4) status_timer.tick(fn=_full_status, outputs=status_box) language_dd = gr.Dropdown( choices=LANGUAGE_CHOICES, value="Bambara (bam)", label="Language you are speaking", ) with gr.Tab("🎙️ Push-to-Talk"): audio_input = gr.Audio( sources=["microphone"], type="filepath", label="Hold to record — release to send", ) talk_btn = gr.Button("▶ Send Recording", variant="primary", size="lg") with gr.Tab("⌨️ Type instead"): text_input = gr.Textbox( lines=3, placeholder=( "Type a message or teach me a word.\n" "Examples:\n" " 'I ni ce means hello in Bambara'\n" " 'Jam waali veut dire bonjour en Fula'\n" " 'How do you say rain in Bambara?'" ), label="Message", ) text_btn = gr.Button("▶ Send", variant="primary") action_status = gr.Textbox( label="Last action", interactive=False, max_lines=1 ) # Voice response output audio_output = gr.Audio( label="🔊 Voice response", autoplay=True, interactive=False, visible=True, ) gr.Markdown( "**Teaching tips:**\n" "- *'I ni ce means hello in Bambara'*\n" "- *'Jam waali veut dire bonjour en Fula'*\n" "- *'How do you say rain in Bambara?'*\n\n" "Every new word is saved to the Hub automatically.\n\n" "**TTS note:** Bambara voice is ready. " "Fula voice requires running `notebooks/train_fula_tts.ipynb` on Kaggle first." ) # ── Right column: memory + chat ─────────────────────────────────── with gr.Column(scale=3): recent_words = gr.Markdown(value=_render_recent_words()) gr.Markdown("---") chatbot = gr.Chatbot( label="Conversation", height=420, type="messages", bubble_full_width=False, ) clear_btn = gr.Button("🗑️ Clear conversation", size="sm", variant="secondary") # ── Wiring ──────────────────────────────────────────────────────────── history_state = gr.State([]) talk_btn.click( fn=process_audio, inputs=[audio_input, language_dd, history_state], outputs=[history_state, recent_words, action_status, audio_output], ).then( fn=lambda h: h, inputs=[history_state], outputs=[chatbot], ) text_btn.click( fn=process_text, inputs=[text_input, language_dd, history_state], outputs=[history_state, recent_words, action_status, audio_output], ).then( fn=lambda h: (h, ""), inputs=[history_state], outputs=[chatbot, text_input], ) text_input.submit( fn=process_text, inputs=[text_input, language_dd, history_state], outputs=[history_state, recent_words, action_status, audio_output], ).then( fn=lambda h: (h, ""), inputs=[history_state], outputs=[chatbot, text_input], ) clear_btn.click( fn=lambda: ([], _render_recent_words(), "", None), outputs=[history_state, recent_words, action_status, audio_output], ).then(fn=lambda: [], outputs=[chatbot]) return demo # ── Entry point ─────────────────────────────────────────────────────────────── # Load vocabulary at startup (background — non-blocking for the UI) threading.Thread(target=_memory.load, daemon=True).start() # Begin loading Whisper immediately _ensure_whisper() # Preload TTS models in background _tts.preload() if __name__ == "__main__": from dotenv import load_dotenv load_dotenv() HF_TOKEN = os.environ.get("HF_TOKEN") FEEDBACK_REPO_ID = os.environ.get("FEEDBACK_REPO_ID", "ous-sow/sahel-agri-feedback") WHISPER_MODEL_ID = os.environ.get("WHISPER_MODEL_ID", "openai/whisper-large-v3-turbo") LLM_MODEL_ID = os.environ.get("LLM_MODEL_ID", "Qwen/Qwen2.5-72B-Instruct") _memory._hf_token = HF_TOKEN _memory._repo_id = FEEDBACK_REPO_ID _gemma._hf_token = HF_TOKEN print(f"STT model : {WHISPER_MODEL_ID}") print(f"LLM model : {LLM_MODEL_ID}") print(f"Store : {FEEDBACK_REPO_ID}") print(f"HF_TOKEN : {'set' if HF_TOKEN else 'NOT SET — Hub push disabled'}") print() demo = build_ui() demo.launch( server_port=7860, inbrowser=False, share=False, show_api=False, ssr_mode=False, )