Update app_qwen_tts.py

app_qwen_tts.py

@@ -2,9 +2,11 @@ import os
 import torch
 import gradio as gr
 import numpy as np
-import soundfile as sf
+import requests
 from transformers import AutoTokenizer, AutoModelForCausalLM
 from sentence_transformers import SentenceTransformer
+import base64
+import io
 
 # =========================================================
 # Configuration
@@ -14,24 +16,22 @@ DOC_FILE = "general.md"
 
 MAX_NEW_TOKENS = 200
 TOP_K = 3
-TTS_API_URL = "https://rahul7star-Chatterbox-Multilingual-TTS-API.hf.space/tts"  # FastAPI TTS endpoint
+
+# FastAPI TTS endpoint
+TTS_API_URL = "https://rahul7star-Chatterbox-Multilingual-TTS-API.hf.space/tts"
 
 # =========================================================
 # Paths
 # =========================================================
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 DOC_PATH = os.path.join(BASE_DIR, DOC_FILE)
-
 if not os.path.exists(DOC_PATH):
     raise RuntimeError(f"❌ {DOC_FILE} not found next to app.py")
 
 # =========================================================
 # Load Qwen Model
 # =========================================================
-tokenizer = AutoTokenizer.from_pretrained(
-    MODEL_ID, trust_remote_code=True
-)
-
+tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, trust_remote_code=True)
 model = AutoModelForCausalLM.from_pretrained(
     MODEL_ID,
     device_map="auto",
@@ -90,25 +90,17 @@ def extract_final_answer(text: str) -> str:
 # =========================================================
 def answer_question(question):
     context = retrieve_context(question)
-
     messages = [
-        {
-            "role": "system",
-            "content": (
-                "You are a strict document-based Q&A assistant.\n"
-                "Answer ONLY the question.\n"
-                "Do NOT repeat the context or the question.\n"
-                "Respond in 1–2 sentences.\n"
-                "If the answer is not present, say:\n"
-                "'I could not find this information in the document.'"
-            )
-        },
-        {
-            "role": "user",
-            "content": f"Context:\n{context}\n\nQuestion:\n{question}"
-        }
+        {"role": "system", "content": (
+            "You are a strict document-based Q&A assistant.\n"
+            "Answer ONLY the question.\n"
+            "Do NOT repeat the context or the question.\n"
+            "Respond in 1–2 sentences.\n"
+            "If the answer is not present, say:\n"
+            "'I could not find this information in the document.'"
+        )},
+        {"role": "user", "content": f"Context:\n{context}\n\nQuestion:\n{question}"}
     ]
-
     prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
     inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
 
@@ -119,68 +111,62 @@ def answer_question(question):
             temperature=0.3,
             do_sample=True
         )
-
     decoded = tokenizer.decode(output[0], skip_special_tokens=True)
     return extract_final_answer(decoded)
 
 # =========================================================
-#  CPU-friendly TTS
-
+# TTS via API (returns NumPy audio)
 # =========================================================
-def tts_via_api(text: str, language_id="en", mode="Speak 🗣️", exaggeration=0.5, temperature=0.8, cfg_weight=0.5):
-    payload = {
-        "text": text,
-        "language_id": language_id,
-        "mode": mode,
-        "exaggeration": exaggeration,
-        "temperature": temperature,
-        "cfg_weight": cfg_weight
-    }
+def tts_via_api(text: str):
     try:
+        payload = {"text": text}
         resp = requests.post(TTS_API_URL, json=payload, timeout=60)
         resp.raise_for_status()
         data = resp.json()
         audio_b64 = data.get("audio", "")
         if not audio_b64:
             return None
-        return f"data:audio/wav;base64,{audio_b64}"
+        # Decode base64 audio to bytes
+        audio_bytes = base64.b64decode(audio_b64.split(",")[-1])
+        # Convert to np.float32
+        import soundfile as sf
+        wav, sr = sf.read(io.BytesIO(audio_bytes), dtype='float32')
+        return wav, sr
     except Exception as e:
         print(f"TTS API error: {e}")
         return None
 
 # =========================================================
-# Gradio Chatbot function
+# Gradio chat function
 # =========================================================
 def chat(user_message, history):
     if not user_message.strip():
         return "", history
-
     try:
-        # 1️⃣ Generate text answer
+        # Text answer
         answer_text = answer_question(user_message)
 
-        # 2️⃣ Generate audio
-        sr, wav = tts_via_api(answer_text)
-
-        # Save temp wav for Gradio audio player
-        audio_path = "/tmp/output.wav"
-        import soundfile as sf
-        sf.write(audio_path, wav, sr)
-
-        # 3️⃣ Append as tuple (text + audio)
-        history.append((user_message, [answer_text, audio_path]))
-
+        # TTS
+        tts_result = tts_via_api(answer_text)
+        if tts_result is not None:
+            wav, sr = tts_result
+            # Gradio can take NumPy array + sample rate directly
+            audio_output = (sr, wav)
+        else:
+            audio_output = None
+
+        # Append tuple with text and playable audio
+        history.append((user_message, [f"**Bot:** {answer_text}", audio_output]))
     except Exception as e:
         print(e)
-        history.append((user_message, ["⚠️ Error generating answer or audio."]))
-
+        history.append((user_message, ["⚠️ Error generating answer or audio.", None]))
     return "", history
 
 def reset_chat():
     return []
 
 # =========================================================
-# Build UI
+# Gradio UI
 # =========================================================
 def build_ui():
     with gr.Blocks(theme=gr.themes.Soft()) as demo: