app.py

import gradio as gr
import time
import transformers
from transformers import Qwen2AudioForConditionalGeneration, AutoProcessor
from io import BytesIO
from urllib.request import urlopen
import librosa
import os, json
from sys import argv
from vllm import LLM, SamplingParams
import vllm
import re


def load_model_processor(model_path):
    processor = AutoProcessor.from_pretrained(model_path)
    llm = LLM(
        model=model_path, trust_remote_code=True, gpu_memory_utilization=0.8,  
        enforce_eager=True,  device = "cuda",
        limit_mm_per_prompt={"audio": 5},
    )
    return llm, processor

model_path1 = "SeaLLMs/SeaLLMs-Audio-7B"
model1, processor1 = load_model_processor(model_path1)

def response_to_audio_conv(conversation, model=None, processor=None, temperature = 0.7,repetition_penalty=1.1, top_p = 0.5,max_new_tokens = 2048):
    turn = conversation[-1]
    if turn["role"] == "user":
        for content in turn['content']:
            if content["type"] == "text":
                if contains_chinese(content["text"]):
                    return "ERROR! This demo does not support Chinese!"
    
    
    text = processor.apply_chat_template(conversation, add_generation_prompt=True, tokenize=False)
    audios = []
    for message in conversation:
        if isinstance(message["content"], list):
            for ele in message["content"]:
                if ele["type"] == "audio":
                    if ele['audio_url'] != None:
                        audios.append(librosa.load(
                            ele['audio_url'], 
                            sr=processor.feature_extractor.sampling_rate)[0]
                        )

    sampling_params = SamplingParams(
        temperature=temperature, max_tokens=max_new_tokens, repetition_penalty=repetition_penalty, top_p=top_p, top_k=20,
        stop_token_ids=[],
    )

    input = {
            'prompt': text,
            'multi_modal_data': {
                'audio': [(audio, 16000) for audio in audios]
            }
            }

    output = model.generate([input], sampling_params=sampling_params)[0]
    response = output.outputs[0].text
    if contains_chinese(response):
        return "ERROR! This demo does not support Chinese! Try a different instruction/prompt!"
    return response

def print_like_dislike(x: gr.LikeData):
    print(x.index, x.value, x.liked)

def contains_chinese(text):
    # Regular expression for Chinese characters
    chinese_char_pattern = re.compile(r'[\u4e00-\u9fff]')
    return bool(chinese_char_pattern.search(text))

def add_message(history, message):
    paths = []
    for turn in history: 
        if turn['role'] == "user" and type(turn['content']) != str:
            paths.append(turn['content'][0])
    for x in message["files"]:
        if x not in paths:
            history.append({"role": "user", "content": {"path": x}})
    if message["text"] is not None:
        history.append({"role": "user", "content": message["text"]})
    return history, gr.MultimodalTextbox(value=None, interactive=False)

def format_user_messgae(message):
    if type(message['content']) == str:
        return {"role": "user", "content": [{"type": "text", "text": message['content']}]}
    else:
        return {"role": "user", "content": [{"type": "audio", "audio_url": message['content'][0]}]}

def history_to_conversation(history):
    conversation = []
    audio_paths = []
    for turn in history:
        if turn['role'] == "user":
            if not turn['content']: 
                continue
            turn = format_user_messgae(turn)
            if turn['content'][0]['type'] == 'audio':
                if turn['content'][0]['audio_url'] in audio_paths:
                    continue
                else: 
                    audio_paths.append(turn['content'][0]['audio_url'])

            if len(conversation) > 0 and conversation[-1]["role"] == "user":
                conversation[-1]['content'].append(turn['content'][0])
            else:
                conversation.append(turn)
        else:
            conversation.append(turn)
    
    print(json.dumps(conversation, indent=4, ensure_ascii=False))
    return conversation

def bot(history: list, temperature = 0.7,repetition_penalty=1.1, top_p = 0.5,
                    max_new_tokens = 2048):
    conversation = history_to_conversation(history)
    response = response_to_audio_conv(conversation, model=model1, processor=processor1, temperature = temperature,repetition_penalty=repetition_penalty, top_p = top_p, max_new_tokens = max_new_tokens)
    # response = "Nice to meet you!"
    print("Bot:",response)

    history.append({"role": "assistant", "content": ""})
    for character in response:
        history[-1]["content"] += character
        time.sleep(0.01)
        yield history

with gr.Blocks() as demo:
    gr.HTML("""<p align="center"><img src="https://DAMO-NLP-SG.github.io/SeaLLMs-Audio/static/images/seallm-audio-logo.png" style="height: 80px"/><p>""")
    gr.HTML("""<h1 align="center" id="space-title">SeaLLMs-Audio-Demo</h1>""")

    gr.HTML(
        """<div style="text-align: center; font-size: 16px;">
        This WebUI is based on <a href="https://huggingface.co/SeaLLMs/SeaLLMs-Audio-7B">SeaLLMs-Audio-7B</a>, developed by Alibaba DAMO Academy.<br>
        You can interact with the chatbot in <b>English, Indonesian, Thai, or Vietnamese</b>.<br>
        For the input, you can provide <b>audio and/or text</b>.
        </div>"""
    )
    
    gr.HTML(
        """<div style="text-align: center; font-size: 16px;">
        <a href="https://DAMO-NLP-SG.github.io/SeaLLMs-Audio/">[Website]</a> &nbsp; 
        <a href="https://huggingface.co/SeaLLMs/SeaLLMs-Audio-7B">[Model🤗]</a> &nbsp; 
        <a href="https://github.com/DAMO-NLP-SG/SeaLLMs-Audio">[Github]</a>
        </div>"""
    )

    # gr.Markdown(insturctions)
    # with gr.Row():
    #     with gr.Column():
    #         temperature = gr.Slider(minimum=0, maximum=1, value=0.3, step=0.1, label="Temperature")
    #     with gr.Column():
    #         top_p = gr.Slider(minimum=0.1, maximum=1, value=0.5, step=0.1, label="Top P")
    #     with gr.Column():
    #         repetition_penalty = gr.Slider(minimum=0, maximum=2, value=1.1, step=0.1, label="Repetition Penalty")
    
    chatbot = gr.Chatbot(elem_id="chatbot", bubble_full_width=False, type="messages")

    chat_input = gr.MultimodalTextbox(
        interactive=True,
        file_count="single",
        file_types=['.wav'],
        placeholder="Enter message (optional) ...",
        show_label=False,
        sources=["microphone", "upload"],
    )

    chat_msg = chat_input.submit(
        add_message, [chatbot, chat_input], [chatbot, chat_input]
    )
    bot_msg = chat_msg.then(bot, chatbot, chatbot, api_name="bot_response")
    # bot_msg = chat_msg.then(bot, [chatbot, temperature, repetition_penalty, top_p], chatbot, api_name="bot_response")
    bot_msg.then(lambda: gr.MultimodalTextbox(interactive=True), None, [chat_input])

    # chatbot.like(print_like_dislike, None, None, like_user_message=True)

    clear_button = gr.ClearButton([chatbot, chat_input])

demo.launch(share=True)
demo.queue(default_concurrency_limit=40).launch(share=True)