Update app.py

app.py

@@ -1,8 +1,209 @@
+import gradio as gr
+import edge_tts
+import asyncio
+import tempfile
 import os
+from huggingface_hub import InferenceClient
+import re
+from streaming_stt_nemo import Model
+import torch
+import random
+from openai import OpenAI
 import subprocess
-import gradio as gr
+import threading
+import queue
+import sounddevice as sd
+import numpy as np
+import wave
+import sys
+
+default_lang = "en"
+
+engines = { default_lang: Model(default_lang) }
+
+def transcribe(audio):
+    if audio is None:
+        return ""
+    lang = "en"
+    model = engines[lang]
+    text = model.stt_file(audio)[0]
+    return text
+
+HF_TOKEN = os.environ.get("HF_TOKEN", None)
+
+def client_fn(model):
+    if "Llama 3 8B Service" in model:
+        return OpenAI(
+            base_url="http://52.76.81.56:60002/v1",
+            api_key="token-abc123"
+        )
+    elif "Llama" in model:
+        return InferenceClient("meta-llama/Meta-Llama-3-8B-Instruct")
+    elif "Mistral" in model:
+        return InferenceClient("mistralai/Mistral-7B-Instruct-v0.2")
+    elif "Phi" in model:
+        return InferenceClient("microsoft/Phi-3-mini-4k-instruct")
+    elif "Mixtral" in model:
+        return InferenceClient("mistralai/Mixtral-8x7B-Instruct-v0.1")
+    else: 
+        return InferenceClient("microsoft/Phi-3-mini-4k-instruct")
+
+def randomize_seed_fn(seed: int) -> int:
+    seed = random.randint(0, 999999)
+    return seed
+
+system_instructions1 = """
+[SYSTEM] You are OPTIMUS Prime a personal AI voice assistant, Created by Jaward.
+Keep conversation friendly, short, clear, and concise. 
+Avoid unnecessary introductions and answer the user's questions directly. 
+Respond in a normal, conversational manner while being friendly and helpful.
+Remember previous parts of the conversation and use that context in your responses.
+Your creator Jaward is an AI Research Engineer at Linksoul AI. He is currently specializing in Artificial Intelligence (AI) research more specifically training and optimizing advance AI systems. He aspires to build not just human-like intelligence but AI Systems that augment human intelligence. He has contributed greatly to the opensource community with first-principles code implementations of AI/ML research papers. He did his first internship at Beijing Academy of Artificial Intelligence as an AI Researher where he contributed in cutting-edge AI research leading to him contributing to an insightful paper (AUTOAGENTS - A FRAMEWORK FOR AUTOMATIC AGENT GENERATION). The paper got accepted this year at IJCAI(International Joint Conference On AI). He is currently doing internship at LinkSoul AI - a small opensource AI Research startup in Beijing.
+[USER]
+"""
+
+conversation_history = []
+
+def models(text, model="Llama 3 8B Service", seed=42):
+    global conversation_history
+    seed = int(randomize_seed_fn(seed))
+    generator = torch.Generator().manual_seed(seed)  
+    
+    client = client_fn(model)
+    
+    if "Llama 3 8B Service" in model:
+        messages = [
+            {"role": "system", "content": system_instructions1},
+        ] + conversation_history + [
+            {"role": "user", "content": text}
+        ]
+        completion = client.chat.completions.create(
+            model="/data/shared/huggingface/hub/models--meta-llama--Meta-Llama-3-8B-Instruct/snapshots/c4a54320a52ed5f88b7a2f84496903ea4ff07b45/",
+            messages=messages
+        )
+        assistant_response = completion.choices[0].message.content
+        
+        # Update conversation history
+        conversation_history.append({"role": "user", "content": text})
+        conversation_history.append({"role": "assistant", "content": assistant_response})
+        
+        # Keep only the last 10 messages to avoid token limit issues
+        if len(conversation_history) > 20:
+            conversation_history = conversation_history[-20:]
+        
+        return assistant_response
+    else:
+        # For other models, we'll concatenate the conversation history into a single string
+        history_text = "\n".join([f"{'User' if msg['role'] == 'user' else 'Assistant'}: {msg['content']}" for msg in conversation_history])
+        formatted_prompt = f"{system_instructions1}\n\nConversation history:\n{history_text}\n\nUser: {text}\nOPTIMUS:"
+        
+        generate_kwargs = dict(
+            max_new_tokens=300,
+            seed=seed
+        )    
+        stream = client.text_generation(
+            formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=False)
+        output = ""
+        for response in stream:
+            if not response.token.text == "</s>":
+                output += response.token.text
+        
+        # Update conversation history
+        conversation_history.append({"role": "user", "content": text})
+        conversation_history.append({"role": "assistant", "content": output})
+        
+        # Keep only the last 10 messages to avoid token limit issues
+        if len(conversation_history) > 20:
+            conversation_history = conversation_history[-20:]
+        
+        return output
+
+# New global variables for audio processing
+RATE = 16000
+CHUNK = int(RATE / 10)  # 100ms
+audio_queue = queue.Queue()
+is_listening = False
+
+def audio_callback(indata, frames, time, status):
+    if status:
+        print(status, file=sys.stderr)
+    audio_queue.put(indata.copy())
+
+def process_audio_stream(model, seed):
+    global is_listening
+    audio_buffer = []
+    silence_threshold = 0.01
+    silence_duration = 0
+    max_silence = 2  # seconds
+
+    while True:
+        if not is_listening:
+            audio_buffer.clear()
+            silence_duration = 0
+            audio_queue.queue.clear()
+            continue
 
-# Supported languages
+        try:
+            chunk = audio_queue.get(timeout=1)
+            audio_buffer.append(chunk)
+
+            # Check for silence
+            if np.abs(chunk).mean() < silence_threshold:
+                silence_duration += CHUNK / RATE
+            else:
+                silence_duration = 0
+
+            if silence_duration > max_silence:
+                # Process the buffered audio
+                audio_data = np.concatenate(audio_buffer)
+                with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp_file:
+                    tmp_path = tmp_file.name
+                    with wave.open(tmp_path, 'wb') as wf:
+                        wf.setnchannels(1)
+                        wf.setsampwidth(2)
+                        wf.setframerate(RATE)
+                        wf.writeframes((audio_data * 32767).astype(np.int16).tobytes())
+
+                # Transcribe and process
+                user_input = transcribe(tmp_path)
+                if user_input:
+                    is_listening = False
+                    reply = models(user_input, model, seed)
+                    asyncio.run(respond_and_play(reply))
+                    is_listening = True
+
+                # Clear the buffer
+                audio_buffer.clear()
+                silence_duration = 0
+
+        except queue.Empty:
+            pass
+
+async def respond_and_play(text):
+    communicate = edge_tts.Communicate(text, voice="en-US-ChristopherNeural")
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp_file:
+        tmp_path = tmp_file.name
+        await communicate.save(tmp_path)
+    
+    # Play the audio
+    with wave.open(tmp_path, 'rb') as wf:
+        data = wf.readframes(wf.getnframes())
+        sd.play(np.frombuffer(data, dtype=np.int16), wf.getframerate())
+        sd.wait()
+
+def start_listening(model, seed):
+    global is_listening
+    is_listening = True
+    threading.Thread(target=process_audio_stream, args=(model, seed), daemon=True).start()
+    with sd.InputStream(callback=audio_callback, channels=1, samplerate=RATE, blocksize=CHUNK):
+        while is_listening:
+            sd.sleep(100)
+
+def stop_listening():
+    global is_listening
+    is_listening = False
+
+# Supported languages for seamless-expressive
 LANGUAGE_CODES = {
     "English": "eng",
     "Spanish": "spa",
@@ -12,58 +213,107 @@ LANGUAGE_CODES = {
     "Chinese": "cmn"
 }
 
-def transcribe(audio):
-    if audio is None:
-        return "No audio input detected. Please record or upload an audio file."
-    
-    try:
-        text = model.stt_file(audio)[0]
-        return text
-    except Exception as e:
-        return f"Error transcribing audio: {str(e)}"
-
 def translate_speech(audio_file, target_language):
+    """
+    Translate input speech (audio file) to the specified target language.
+    """
     if audio_file is None:
-        return "No audio input detected. Please record or upload an audio file."
+        return None
     
-    try:
-        language_code = LANGUAGE_CODES[target_language]
-        output_file = "translated_audio.wav"
-        
-        command = [
-            "expressivity_predict",
-            audio_file,
-            "--tgt_lang", language_code,
-            "--model_name", "seamless_expressivity",
-            "--vocoder_name", "vocoder_pretssel",
-            "--gated-model-dir", "seamlessmodel",
-            "--output_path", output_file
-        ]
-        
-        subprocess.run(command, check=True)
-
-        if os.path.exists(output_file):
-            print(f"File created successfully: {output_file}")
-            return output_file
-        else:
-            return "Error: Translated audio file not found."
-    except Exception as e:
-        return f"Error translating speech: {str(e)}"
-
-def create_interface():
-    inputs = [
-        gr.Audio(label="User", sources=["microphone"], type="filepath"),
-        gr.Dropdown(list(LANGUAGE_CODES.keys()), label="Target Language")
+    language_code = LANGUAGE_CODES[target_language]
+    output_file = "translated_audio.wav"
+    
+    command = [
+        "expressivity_predict",
+        audio_file,
+        "--tgt_lang", language_code,
+        "--model_name", "seamless_expressivity",
+        "--vocoder_name", "vocoder_pretssel",
+        "--gated-model-dir", "models",
+        "--output_path", output_file
     ]
+    
+    subprocess.run(command, check=True)
+
+    if os.path.exists(output_file):
+        print(f"File created successfully: {output_file}")
+        return output_file
+    else:
+        print(f"File not found: {output_file}")
+        return None
+
+def clear_history():
+    global conversation_history
+    conversation_history = []
+    return None, None, None, None
+
+def voice_assistant_tab():
+    return "# <center><b>Hello, I am Optimus Prime your personal AI voice assistant</b></center>"
+
+def speech_translation_tab():
+    return "# <center><b>Hear how you sound in another language</b></center>"
+
+with gr.Blocks(css="style.css") as demo:
+    description = gr.Markdown("# <center><b>Hello, I am Optimus Prime your personal AI voice assistant</b></center>")
+    
+    with gr.Tabs() as tabs:
+        with gr.TabItem("Voice Assistant") as voice_assistant:
+            select = gr.Dropdown([
+                'Llama 3 8B Service',
+                'Mixtral 8x7B',
+                'Llama 3 8B',
+                'Mistral 7B v0.3',
+                'Phi 3 mini',
+            ],
+            value="Llama 3 8B Service",
+            label="Model"
+            )
+            seed = gr.Slider(
+            label="Seed",
+            minimum=0,
+            maximum=999999,
+            step=1,
+            value=0,
+            visible=False
+            )
+            start_button = gr.Button("Start Listening")
+            stop_button = gr.Button("Stop Listening")
+            status = gr.Markdown("Status: Not listening")
+
+            start_button.click(
+                fn=lambda model, seed: start_listening(model, seed),
+                inputs=[select, seed],
+                outputs=[status],
+                _js="() => {document.getElementById('status').textContent = 'Status: Listening'}"
+            )
+            stop_button.click(
+                fn=stop_listening,
+                inputs=[],
+                outputs=[status],
+                _js="() => {document.getElementById('status').textContent = 'Status: Not listening'}"
+            )
+        
+        with gr.TabItem("Speech Translation") as speech_translation:
+            input_audio = gr.Audio(label="User", sources=["microphone"], type="filepath")
+            target_lang = gr.Dropdown(
+                choices=list(LANGUAGE_CODES.keys()),
+                value="German",
+                label="Target Language"
+            )
+            output_audio = gr.Audio(label="Translated Audio",
+                                    interactive=False,
+                                    autoplay=True,
+                                    elem_classes="audio")
+            
+            gr.Interface(
+                fn=translate_speech,
+                inputs=[input_audio, target_lang],
+                outputs=[output_audio],
+                live=True
+            )
 
-    return gr.Interface(
-        fn=translate_speech,
-        inputs=inputs,
-        outputs=gr.Audio(label="Translated Audio", interactive=False, autoplay=True, elem_classes="audio"),
-        title="Seamless Expressive Speech-To-Speech Translator",
-        description="Hear how you sound in another language.",
-    )
+    voice_assistant.select(fn=voice_assistant_tab, inputs=None, outputs=description)
+    speech_translation.select(fn=speech_translation_tab, inputs=None, outputs=description)
 
 if __name__ == "__main__":
-    iface = create_interface()
-    iface.launch()
+    demo.queue(max_size=200).launch()