Create app.py

app.py

@@ -0,0 +1,163 @@
+import gradio as gr
+from transformers import AutoModelForCausalLM, AutoTokenizer
+import torch
+import os
+import copy
+import re
+import secrets
+from pathlib import Path
+from pydub import AudioSegment
+
+# Initialize the model and tokenizer
+torch.manual_seed(420)
+tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen-Audio-Chat", trust_remote_code=True)
+model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen-Audio-Chat", device_map="cuda", trust_remote_code=True).eval()
+
+def _parse_text(text):
+    lines = text.split("\n")
+    lines = [line for line in lines if line != ""]
+    count = 0
+    for i, line in enumerate(lines):
+        if "```" in line:
+            count += 1
+            items = line.split("`")
+            if count % 2 == 1:
+                lines[i] = f'<pre><code class="language-{items[-1]}">'
+            else:
+                lines[i] = f"<br></code></pre>"
+        else:
+            if i > 0:
+                if count % 2 == 1:
+                    line = line.replace("`", r"\`")
+                    line = line.replace("<", "&lt;")
+                    line = line.replace(">", "&gt;")
+                    line = line.replace(" ", "&nbsp;")
+                    line = line.replace("*", "&ast;")
+                    line = line.replace("_", "&lowbar;")
+                    line = line.replace("-", "&#45;")
+                    line = line.replace(".", "&#46;")
+                    line = line.replace("!", "&#33;")
+                    line = line.replace("(", "&#40;")
+                    line = line.replace(")", "&#41;")
+                    line = line.replace("$", "&#36;")
+                lines[i] = "<br>" + line
+    text = "".join(lines)
+    return text
+    
+def predict(_chatbot, task_history):
+    if not task_history:
+        return _chatbot
+
+    query = task_history[-1][0]
+    history_cp = copy.deepcopy(task_history)
+    history_filter = []
+    audio_idx = 1
+    pre = ""
+    last_audio = None
+
+    for i, (q, a) in enumerate(history_cp):
+        if isinstance(q, (tuple, list)):
+            last_audio = q[0]
+            q = f'Audio {audio_idx}: <audio>{q[0]}</audio>'
+            pre += q + '\n'
+            audio_idx += 1
+        else:
+            pre += q
+            history_filter.append((pre, a))
+            pre = ""
+
+    history, message = history_filter[:-1], history_filter[-1][0]
+    response, history = model.chat(tokenizer, message, history=history)
+
+    ts_pattern = r"<\|\d{1,2}\.\d+\|>"
+    all_time_stamps = re.findall(ts_pattern, response)
+    if (len(all_time_stamps) > 0) and (len(all_time_stamps) % 2 ==0) and last_audio:
+        ts_float = [ float(t.replace("<|","").replace("|>","")) for t in all_time_stamps]
+        ts_float_pair = [ts_float[i:i + 2] for i in range(0,len(all_time_stamps),2)]
+        # 读取音频文件
+        format = os.path.splitext(last_audio)[-1].replace(".","")
+        audio_file = AudioSegment.from_file(last_audio, format=format)
+        chat_response_t = response.replace("<|", "").replace("|>", "")
+        chat_response = chat_response_t
+        temp_dir = secrets.token_hex(20)
+        temp_dir = Path(uploaded_file_dir) / temp_dir
+        temp_dir.mkdir(exist_ok=True, parents=True)
+            # 截取音频文件
+        for pair in ts_float_pair:
+            audio_clip = audio_file[pair[0] * 1000: pair[1] * 1000]
+                # 保存音频文件
+            name = f"tmp{secrets.token_hex(5)}.{format}"
+            filename = temp_dir / name
+            audio_clip.export(filename, format=format)
+            _chatbot[-1] = (_parse_text(query), chat_response)
+            _chatbot.append((None, (str(filename),)))
+    else:
+        _chatbot[-1] = (_parse_text(query), response)
+        full_response = _parse_text(response)
+        task_history[-1] = (query, full_response)
+        print("Qwen-Audio-Chat: " + _parse_text(full_response))
+        return _chatbot
+
+def regenerate(_chatbot, task_history):
+    if not task_history:
+        return _chatbot
+    item = task_history[-1]
+    if item[1] is None:
+        return _chatbot
+    task_history[-1] = (item[0], None)
+    chatbot_item = _chatbot.pop(-1)
+    if chatbot_item[0] is None:
+            _chatbot[-1] = (_chatbot[-1][0], None)
+    else:
+        _chatbot.append((chatbot_item[0], None))
+    return predict(_chatbot, task_history)
+
+def add_text(history, task_history, text):
+    history = history + [(_parse_text(text), None)]
+    task_history = task_history + [(text, None)]
+    return history, task_history, ""
+
+def add_file(history, task_history, file):
+    history = history + [((file.name,), None)]
+    task_history = task_history + [((file.name,), None)]
+    return history, task_history
+
+def add_mic(history, task_history, file):
+    if file is None:
+        return history, task_history
+    os.rename(file, file + '.wav')
+    print("add_mic file:", file)
+    print("add_mic history:", history)
+    print("add_mic task_history:", task_history)
+    # history = history + [((file.name,), None)]
+    # task_history = task_history + [((file.name,), None)]
+    task_history = task_history + [((file + '.wav',), None)]
+    history = history + [((file + '.wav',), None)]
+    print("task_history", task_history)
+    return history, task_history
+
+def reset_user_input():
+    return gr.update(value="")
+
+def reset_state(task_history):
+    task_history.clear()
+    return []
+        
+iface = gr.Interface(
+    fn=chat_with_model,
+    inputs=[
+        gr.inputs.Audio(label="Audio Input"),
+        gr.inputs.Textbox(label="Text Query"),
+        gr.State()  
+    ],
+    outputs=[
+        "text",  
+        gr.State()  
+    ],
+    title="Audio-Text Interaction Model",
+    description="This model can process an audio input along with a text query and provide a response.",
+    theme="default",
+    allow_flagging="never"
+)
+
+iface.launch()