import gradio as gr import torch import os import numpy as np import soundfile as sf import tempfile import json import uuid import re import time from datetime import datetime from transformers import ( AutoModelForSpeechSeq2Seq, AutoProcessor, pipeline, AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer ) from peft import PeftModel from huggingface_hub import snapshot_download, HfApi, hf_hub_download import shutil from threading import Thread # ── Config ──────────────────────────────────────────────────────── WHISPER_BASE = "openai/whisper-large-v3-turbo" WHISPER_REPO = "mozed54/whisper-finetuned" GEMMA_BASE = "google/gemma-2-2b-it" GEMMA_REPO = "mozed54/modelV3" FEEDBACK_REPO = "mozed54/whisper-feedback" FEEDBACK2_REPO = "mozed54/gemma-feedback" HF_TOKEN = os.environ.get("HF_TOKEN") device = "cuda" if torch.cuda.is_available() else "cpu" torch_dtype = torch.float16 if device == "cuda" else torch.float32 # ── Chargement Whisper ──────────────────────────────────────────── print("Chargement Whisper...") whisper_base = AutoModelForSpeechSeq2Seq.from_pretrained( WHISPER_BASE, torch_dtype=torch_dtype, use_safetensors=True, ) whisper_adapter = snapshot_download( repo_id=WHISPER_REPO, repo_type="dataset", token=HF_TOKEN if HF_TOKEN else None, ) whisper_model = PeftModel.from_pretrained(whisper_base, whisper_adapter) whisper_model = whisper_model.merge_and_unload() whisper_model.to(device) whisper_processor = AutoProcessor.from_pretrained(WHISPER_BASE) whisper_pipe = pipeline( "automatic-speech-recognition", model=whisper_model, tokenizer=whisper_processor.tokenizer, feature_extractor=whisper_processor.feature_extractor, torch_dtype=torch_dtype, device=device, generate_kwargs={"language": "arabic", "task": "transcribe"}, ) print("Whisper pret !") # ── Chargement Gemma ────────────────────────────────────────────── print("Chargement Gemma...") gemma_tokenizer = AutoTokenizer.from_pretrained(GEMMA_BASE, token=HF_TOKEN) gemma_base = AutoModelForCausalLM.from_pretrained( GEMMA_BASE, torch_dtype=torch_dtype, token=HF_TOKEN, ) gemma_adapter = snapshot_download( repo_id=GEMMA_REPO, repo_type="dataset", token=HF_TOKEN if HF_TOKEN else None, ) gemma_model = PeftModel.from_pretrained(gemma_base, gemma_adapter, is_trainable=False) gemma_model = gemma_model.merge_and_unload() gemma_model.to(device) print("Gemma pret !") # ── État global ─────────────────────────────────────────────────── current_audio = {"path": None, "filename": None} current_exchange = {"transcription": None, "reponse": None} stop_generation = {"value": False} audio_state = {"mic_ready": False, "file_ready": False} def btn_states(): mic_label = "🎤 Transcrire la voix" file_label = "📂 Transcrire le fichier" return ( gr.update(interactive=audio_state["mic_ready"], value=mic_label), gr.update(interactive=audio_state["file_ready"], value=file_label), ) def btn_states_disabled(): """Désactive les deux boutons de transcription (traitement en cours).""" return ( gr.update(interactive=False, value="🎤 Transcrire la voix"), gr.update(interactive=False, value="📂 Transcrire le fichier"), ) # ── Helpers ─────────────────────────────────────────────────────── def generate_audio_name(): dt = datetime.utcnow().strftime("%Y-%m-%d_%H-%M-%S") uid = uuid.uuid4().hex[:6] return f"{dt}{uid}.wav" def process_audio(audio): if audio is None: return None if isinstance(audio, tuple): sample_rate, data = audio if data.ndim > 1: data = data.mean(axis=1) data = data.astype(np.float32) if np.abs(data).max() > 1.0: data = data / 32768.0 tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) sf.write(tmp.name, data, sample_rate) return tmp.name return audio def append_and_upload(feedback_file, entry, repo_id): api = HfApi(token=HF_TOKEN) local = f"/tmp/{repo_id.replace('/', '')}.jsonl" try: existing = hf_hub_download( repo_id=repo_id, filename="feedback.jsonl", repo_type="dataset", token=HF_TOKEN, ) shutil.copy(existing, local) except Exception: pass with open(local, "a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") api.upload_file( path_or_fileobj=local, path_in_repo="feedback.jsonl", repo_id=repo_id, repo_type="dataset", token=HF_TOKEN, ) # ── Prompt Gemma ────────────────────────────────────────────────── SYSTEM_PROMPT = ( "أنت مساعد ذكاء اصطناعي متخصص في الرد على التعليقات المكتوبة بالدارجة الجزائرية " "عن معالجة النصوص الصوتية. " "إذا كان النص يتعلق بخدمات الاتصالات (مودم، فيبر، شكوى، تعبئة)، قم بالرد عليه مباشرة " "باسم خدمة العملاء بلهجة جزائرية مهذبة ومفيدة. " "أما إذا كان النص عبارة عن سؤال أو استفسار عام في أي مجال آخر، فأجب عليه مباشرة " "وبشكل طبيعي بالدارجة الجزائرية. " "إذا كان النص غير مفهوم، اعتذر بلطف واطلب التوضيح. " "يُمنع الرد إذا كان النص يحتوي على شتائم، اعتذر بأدب. " "مهم جداً: اكتب الرد المباشر فقط. لا تكتب أي تحليل، لا تكتب 'الفئة'، لا تكتب 'الملخص'، " "ولا تكتب 'الرد المقترح'. اكتب نص الرد فقط مباشرة بدون أي مقدمات." ) def build_prompt(transcription: str) -> str: messages = [ { "role": "user", "content": f"{SYSTEM_PROMPT}\n\nالنص: {transcription}", } ] return gemma_tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True, ) # ── Filtre caractères non-arabes ────────────────────────────────── def filter_arabic_only(text: str) -> str: allowed = re.compile( r'[^\u0600-\u06FF\u0750-\u077F\u08A0-\u08FF' r'\uFB50-\uFDFF\uFE70-\uFEFF' r'\u0020\u00A0' r'\u060C\u061B\u061F\u06D4' r'\u0021-\u002F\u003A-\u0040' r'\u005B-\u0060\u007B-\u007E' r'\n]' ) return allowed.sub('', text).strip() def extract_reply(text: str) -> str: lines = text.split('\n') for i, line in enumerate(lines): clean_line = re.sub(r'[*_()\[\]\s]+', ' ', line).strip() if not re.search(r'الرد', clean_line): continue sep = ':' if ':' in line else ('：' if '：' in line else None) if sep: parts = line.split(sep, 1) if len(parts) >= 2: reply = parts[1].strip().strip('"\'\u201c\u201d\u2018\u2019\u00ab\u00bb') reply = re.sub(r'[*_`#]+', '', reply).strip() if len(reply) > 3: return reply next_parts = [] for j in range(i + 1, len(lines)): nl = lines[j].strip().strip('"\'\u201c\u201d\u2018\u2019\u00ab\u00bb') nl = re.sub(r'[*_`#*]+', '', nl).strip() if nl: next_parts.append(nl) if next_parts: reply = ' '.join(next_parts) if len(reply) > 3: return reply return text def truncate_gemma_output(text: str) -> str: stop_tokens = [ "", "", "\nسؤال:", "\nسوال:", "\nالسؤال:", "User:", "user:", "\nmodel\n", "###", "---", ] for token in stop_tokens: if token in text: text = text.split(token)[0].strip() text = text.strip() text = extract_reply(text) text = filter_arabic_only(text) sentences = re.split(r'(?<=[.!?؟])\s+', text) if len(sentences) > 4: text = " ".join(sentences[:4]).strip() if len(text) > 500: text = text[:500].rsplit(" ", 1)[0].strip() + "..." return text # ── Génération Gemma avec streaming ─────────────────────────────── def generate_gemma_response(transcription: str, progress=None): stop_generation["value"] = False if progress: progress(0.6, desc="🤖 Préparation du prompt...") prompt = build_prompt(transcription) inputs = gemma_tokenizer(prompt, return_tensors="pt").to(device) streamer = TextIteratorStreamer( gemma_tokenizer, skip_prompt=True, skip_special_tokens=True, decode_kwargs={"skip_special_tokens": True}, ) generation_kwargs = dict( **inputs, max_new_tokens=200, do_sample=True, temperature=0.2, top_p=0.9, top_k=40, repetition_penalty=1.2, no_repeat_ngram_size=4, pad_token_id=gemma_tokenizer.eos_token_id, eos_token_id=gemma_tokenizer.eos_token_id, streamer=streamer, ) if progress: progress(0.75, desc="🤖 Génération en cours...") thread = Thread(target=gemma_model.generate, kwargs=generation_kwargs) thread.start() generated_text = "" for new_text in streamer: if stop_generation["value"]: break generated_text += new_text truncated = truncate_gemma_output(generated_text) yield truncated thread.join() final = truncate_gemma_output(generated_text) yield final # ── ÉTAPE 1 : Transcription Whisper uniquement ──────────────────── def transcribe_only(audio, progress=gr.Progress()): path = process_audio(audio) if path is None: mic_u, file_u = btn_states() yield ( "Aucun audio recu.", gr.update(visible=False), gr.update(visible=False), mic_u, file_u, ) return mic_u, file_u = btn_states_disabled() yield ( gr.update(), gr.update(visible=False), gr.update(visible=False), mic_u, file_u, ) try: progress(0.3, desc="🎙 Transcription audio...") result = whisper_pipe(path, return_timestamps=False) text = result["text"].strip() current_audio["path"] = path current_audio["filename"] = generate_audio_name() current_exchange["transcription"] = text current_exchange["reponse"] = None progress(1.0, desc="✅ Transcription terminée") yield ( text, gr.update(visible=False), gr.update(visible=False), gr.update(interactive=False), gr.update(interactive=False), ) except Exception as e: mic_u, file_u = btn_states() yield ( f"Erreur Whisper : {str(e)}", gr.update(visible=False), gr.update(visible=False), mic_u, file_u, ) # ── ÉTAPE 2 : Réponse Gemma ─────────────────────────────────────── def respond_gemma(progress=gr.Progress()): transcription = current_exchange.get("transcription") if not transcription: mic_u, file_u = btn_states() yield ( "⚠️ Pas de transcription disponible.", gr.update(visible=False), gr.update(visible=False), mic_u, file_u, ) return try: progress(0.5, desc="🤖 Gemma réfléchit...") last_partial = "" for partial in generate_gemma_response(transcription, progress=progress): if stop_generation["value"]: return last_partial = partial yield ( partial, gr.update(visible=False), gr.update(visible=False), gr.update(interactive=False), gr.update(interactive=False), ) if stop_generation["value"]: return current_exchange["reponse"] = last_partial progress(1.0, desc="✅ Terminé") mic_u, file_u = btn_states() yield ( last_partial, gr.update(visible=True), gr.update(visible=True), mic_u, file_u, ) except Exception as e: mic_u, file_u = btn_states() yield ( f"Erreur Gemma : {str(e)}", gr.update(visible=False), gr.update(visible=False), mic_u, file_u, ) # ── Régénération Gemma seule ────────────────────────────────────── def regenerate_gemma_only(progress=gr.Progress()): transcription = current_exchange.get("transcription") if not transcription: mic_u, file_u = btn_states() yield "", gr.update(visible=False), mic_u, file_u return # Désactive les boutons de transcription pendant la régénération mic_u, file_u = btn_states_disabled() yield gr.update(), gr.update(visible=False), mic_u, file_u try: progress(0.3, desc="🔄 Régénération...") last_partial = "" for partial in generate_gemma_response(transcription, progress=progress): if stop_generation["value"]: return last_partial = partial yield partial, gr.update(visible=False), gr.update(interactive=False), gr.update(interactive=False) if stop_generation["value"]: return current_exchange["reponse"] = last_partial progress(1.0, desc="✅ Prêt") mic_u, file_u = btn_states() yield last_partial, gr.update(visible=True), mic_u, file_u except Exception as e: mic_u, file_u = btn_states() yield f"Erreur : {str(e)}", gr.update(visible=False), mic_u, file_u # ── Gestion de l'enregistrement micro (attente de chargement) ───── def on_audio_recorded(audio): if audio is None: audio_state["mic_ready"] = False # Audio supprimé -> bouton inactif return gr.update(interactive=False, value="🎤 Transcrire la voix") audio_state["mic_ready"] = False # On affiche un état "chargement" pendant l'attente yield gr.update(interactive=False, value="⏳ Chargement de l'audio...") time.sleep(2) audio_state["mic_ready"] = True yield gr.update(interactive=True, value="🎤 Transcrire la voix") def on_file_uploaded(audio): """ Appelé dès qu'un fichier audio est chargé (ou supprimé) dans le composant upload. On désactive le bouton 'Transcrire le fichier' si aucun fichier n'est présent, et on attend 2 secondes avant de le réactiver pour laisser le temps au fichier de se charger côté navigateur. """ if audio is None: audio_state["file_ready"] = False return gr.update(interactive=False, value="📂 Transcrire le fichier") audio_state["file_ready"] = False yield gr.update(interactive=False, value="⏳ Chargement du fichier...") time.sleep(2) audio_state["file_ready"] = True yield gr.update(interactive=True, value="📂 Transcrire le fichier") # ── Reset ───────────────────────────────────────────────────────── def reset_all(): stop_generation["value"] = True current_audio["path"] = None current_audio["filename"] = None current_exchange["transcription"] = None current_exchange["reponse"] = None audio_state["mic_ready"] = False audio_state["file_ready"] = False return ( None, None, "", "", gr.update(visible=False), gr.update(visible=False), None, None, "", "", gr.update(interactive=False, value="🎤 Transcrire la voix"), gr.update(interactive=False, value="📂 Transcrire le fichier"), ) # ── Feedback Whisper ────────────────────────────────────────────── def save_feedback_whisper(transcription, note): if not transcription or note is None: return "⚠️ Transcription ou note manquante." try: api = HfApi(token=HF_TOKEN) ts = datetime.utcnow().isoformat() audio_nm = current_audio["filename"] or "inconnu.wav" entry = { "timestamp": ts, "audio_file": audio_nm, "transcription": transcription, "note": int(note), } append_and_upload("feedback.jsonl", entry, FEEDBACK_REPO) if current_audio["path"] and os.path.exists(current_audio["path"]): api.upload_file( path_or_fileobj=current_audio["path"], path_in_repo=f"audios/{audio_nm}", repo_id=FEEDBACK_REPO, repo_type="dataset", token=HF_TOKEN, ) return f"✅ Feedback enregistre — {'⭐' * int(note)}" except Exception as e: return f"❌ Erreur : {str(e)}" # ── Feedback Gemma ──────────────────────────────────────────────── def save_feedback_gemma(note): t = current_exchange["transcription"] r = current_exchange["reponse"] if not t or not r or note is None: return "⚠️ Effectue une transcription d'abord." try: ts = datetime.utcnow().isoformat() entry = { "timestamp": ts, "transcription": t, "reponse": r, "note": int(note), } append_and_upload("feedback.jsonl", entry, FEEDBACK2_REPO) return f"✅ Feedback enregistre — {'⭐' * int(note)}" except Exception as e: return f"❌ Erreur : {str(e)}" # ── CSS ─────────────────────────────────────────────────────────── css = """ @import url("https://fonts.googleapis.com/css2?family=DM+Sans:wght@400;500;600&display=swap"); body, .gradio-container { font-family: "DM Sans", sans-serif !important; background: #0f0f13 !important; } .gradio-container { max-width: 900px !important; margin: 0 auto !important; padding: 2rem 1.5rem !important; } #app-header { text-align: center; margin-bottom: 2rem; padding: 2rem 0 1.5rem; border-bottom: 1px solid #1e1e2a; } #app-header h1 { font-size: 2rem; font-weight: 600; color: #f0f0f5; margin: 0 0 0.4rem; } #app-header p { font-size: 0.9rem; color: #6b6b80; margin: 0; } .section-label { font-size: 0.72rem; font-weight: 600; color: #4a4a60; letter-spacing: 0.12em; text-transform: uppercase; margin-bottom: 0.6rem; } /* Bouton Transcrire le fichier - BLEU */ #file-btn { background: linear-gradient(135deg, #5a5aff 0%, #4040d9 100%) !important; color: #fff !important; border: none !important; border-radius: 10px !important; font-weight: 600 !important; font-size: 0.95rem !important; padding: 0.75rem 1.4rem !important; box-shadow: 0 4px 14px rgba(90, 90, 255, 0.35) !important; transition: all 0.2s ease !important; } #file-btn:hover:not(:disabled) { background: linear-gradient(135deg, #4848e0 0%, #3030c0 100%) !important; box-shadow: 0 6px 20px rgba(90, 90, 255, 0.45) !important; transform: translateY(-1px) !important; } /* Boutons Transcrire - état désactivé / chargement */ #mic-btn:disabled, #mic-btn[disabled], #file-btn:disabled, #file-btn[disabled] { opacity: 0.55 !important; cursor: not-allowed !important; filter: grayscale(40%) !important; } /* Boutons Envoyer feedback - ORANGE/JAUNE */ #btn-whisper-feedback, #btn-gemma-feedback { background: linear-gradient(135deg, #f59e0b 0%, #d97706 100%) !important; color: #1a1000 !important; border: none !important; border-radius: 10px !important; font-weight: 600 !important; font-size: 0.88rem !important; padding: 0.6rem 1.4rem !important; letter-spacing: 0.03em !important; box-shadow: 0 3px 12px rgba(245, 158, 11, 0.35) !important; transition: all 0.18s ease !important; cursor: pointer !important; margin-top: 0.5rem !important; } #btn-whisper-feedback:hover, #btn-gemma-feedback:hover { background: linear-gradient(135deg, #fbbf24 0%, #f59e0b 100%) !important; box-shadow: 0 5px 18px rgba(245, 158, 11, 0.5) !important; transform: translateY(-1px) !important; } #btn-whisper-feedback:active, #btn-gemma-feedback:active { transform: translateY(0) !important; box-shadow: 0 2px 8px rgba(245, 158, 11, 0.3) !important; } button.danger { background: #dc2626 !important; color: #fff !important; border: none !important; border-radius: 10px !important; font-weight: 500 !important; padding: 0.7rem 1.4rem !important; } button.danger:hover { background: #b91c1c !important; } button.regenerate { background: #7c3aed !important; color: #fff !important; border: none !important; border-radius: 10px !important; font-weight: 500 !important; padding: 0.5rem 1rem !important; } button.regenerate:hover { background: #6d28d9 !important; } #transcription-box textarea { background: #0f0f13 !important; border: 1.5px solid #1e1e2a !important; border-radius: 14px !important; color: #e0e0ec !important; font-size: 1rem !important; line-height: 1.7 !important; padding: 1.2rem !important; direction: rtl !important; text-align: right !important; min-height: 120px !important; } #transcription-box.has-text textarea { border-color: #22c97a !important; box-shadow: 0 0 0 3px rgba(34,201,122,0.12) !important; } #gemma-box textarea { background: #0f0f13 !important; border: 1.5px solid #1e1e2a !important; border-radius: 14px !important; color: #e0e0ec !important; font-size: 1rem !important; line-height: 1.7 !important; padding: 1.2rem !important; direction: rtl !important; text-align: right !important; min-height: 120px !important; } #gemma-box.has-text textarea { border-color: #a855f7 !important; box-shadow: 0 0 0 3px rgba(168,85,247,0.12) !important; } .feedback-panel { background: #16161f; border: 1px solid #1e1e2a; border-radius: 14px; padding: 1.2rem 1.5rem; margin-top: 0.8rem; } @media (max-width: 768px) { .gr-audio canvas, [data-testid="audio"] canvas, .waveform-component canvas { display: none !important; } .gr-audio, [data-testid="audio"] { -webkit-transform: translate3d(0, 0, 0); transform: translate3d(0, 0, 0); -webkit-backface-visibility: hidden; backface-visibility: hidden; } } footer { display: none !important; } """ # ── JS ──────────────────────────────────────────────────────────── js = """ function() { const addListener = (id, cls) => { const setup = () => { const box = document.querySelector(id + " textarea"); const wrapper = document.querySelector(id); if (!box || !wrapper) return setTimeout(() => setup(), 100); const toggleClass = () => { if (box.value && box.value.trim().length > 0) { wrapper.classList.add(cls); } else { wrapper.classList.remove(cls); } }; box.addEventListener('input', toggleClass); box.addEventListener('change', toggleClass); toggleClass(); }; setup(); }; addListener("#transcription-box", "has-text"); addListener("#gemma-box", "has-text"); return []; } """ # ── Interface ───────────────────────────────────────────────────── with gr.Blocks(title="Whisper + Gemma · Darija", css=css, js=js) as demo: gr.HTML('

🎙 Whisper · Gemma · Darija

Transcription + Réponse — dialecte algérien

') gr.HTML('
Source audio
') with gr.Row(): with gr.Column(scale=1): mic_input = gr.Audio(sources=["microphone"], type="numpy", label="🎤 Enregistrer ma voix") mic_btn = gr.Button("🎤 Transcrire la voix", variant="primary", interactive=False, elem_id="mic-btn") with gr.Column(scale=1): file_input = gr.Audio(sources=["upload"], type="numpy", label="📂 Charger un fichier audio") file_btn = gr.Button("📂 Transcrire le fichier", elem_id="file-btn", interactive=False) gr.HTML("
") with gr.Row(): with gr.Column(scale=1): gr.HTML('
🎙 Transcription
') whisper_out = gr.Textbox( label="", lines=5, placeholder="النص سيظهر هنا", interactive=False, elem_id="transcription-box", show_label=False, ) with gr.Column(visible=False, elem_classes="feedback-panel") as fb_whisper: gr.HTML('
Evaluer Transcription
') note_whisper = gr.Radio(choices=["1","2","3","4","5"], label="", info="1 = tres mauvais · 5 = parfait", show_label=False) btn_whisper = gr.Button("Envoyer", elem_id="btn-whisper-feedback") status_whisper = gr.Textbox(label="", show_label=False, interactive=False, lines=1) with gr.Column(scale=1): gr.HTML('
🤖 Réponse
') gemma_out = gr.Textbox( label="", lines=5, placeholder="الجواب سيظهر هنا", interactive=False, elem_id="gemma-box", show_label=False, ) with gr.Row(): regenerate_btn = gr.Button("🔄 Régénérer", variant="regenerate", size="sm") with gr.Column(visible=False, elem_classes="feedback-panel") as fb_gemma: gr.HTML('
Evaluer Réponse
') note_gemma = gr.Radio(choices=["1","2","3","4","5"], label="", info="1 = tres mauvais · 5 = parfait", show_label=False) btn_gemma = gr.Button("Envoyer", elem_id="btn-gemma-feedback") status_gemma = gr.Textbox(label="", show_label=False, interactive=False, lines=1) gr.HTML("
") with gr.Row(): with gr.Column(): reset_btn = gr.Button("🔄 Réinitialiser tout", variant="danger") mic_input.change( fn=on_audio_recorded, inputs=[mic_input], outputs=[mic_btn], ) file_input.change( fn=on_file_uploaded, inputs=[file_input], outputs=[file_btn], ) mic_btn.click( fn=transcribe_only, inputs=[mic_input], outputs=[whisper_out, fb_whisper, fb_gemma, mic_btn, file_btn], ).then( fn=respond_gemma, inputs=[], outputs=[gemma_out, fb_whisper, fb_gemma, mic_btn, file_btn], ) file_btn.click( fn=transcribe_only, inputs=[file_input], outputs=[whisper_out, fb_whisper, fb_gemma, mic_btn, file_btn], ).then( fn=respond_gemma, inputs=[], outputs=[gemma_out, fb_whisper, fb_gemma, mic_btn, file_btn], ) regenerate_btn.click( fn=regenerate_gemma_only, inputs=[], outputs=[gemma_out, fb_gemma, mic_btn, file_btn], ) btn_whisper.click(fn=save_feedback_whisper, inputs=[whisper_out, note_whisper], outputs=[status_whisper]) btn_gemma.click(fn=save_feedback_gemma, inputs=[note_gemma], outputs=[status_gemma]) reset_btn.click( fn=reset_all, outputs=[ mic_input, file_input, whisper_out, gemma_out, fb_whisper, fb_gemma, note_whisper, note_gemma, status_whisper, status_gemma, mic_btn, file_btn, ], ) demo.launch()