error404-api / app.py
benjamin5607's picture
Update app.py
a23a872 verified
Raw History Blame Contribute Delete
6.97 kB
import os, re, uuid, torch, scipy.io.wavfile, edge_tts, asyncio
import numpy as np
import gradio as gr
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
from groq import Groq
# ๋ชจ๋ธ ๊ด€๋ฆฌ ์‹ฑ๊ธ€ํ†ค
class ModelManager:
_llm_pipeline = None
_music_pipeline = None
_groq_client = None
@classmethod
def get_qwen(cls):
if cls._llm_pipeline is None:
model_id = "Qwen/Qwen2.5-0.5B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForCausalLM.from_pretrained(model_id, device_map="cpu", torch_dtype=torch.float32)
cls._llm_pipeline = pipeline("text-generation", model=model, tokenizer=tokenizer)
return cls._llm_pipeline
@classmethod
def get_groq(cls):
key = os.getenv("GROQ_API_KEY")
if cls._groq_client is None and key:
cls._groq_client = Groq(api_key=key)
return cls._groq_client
@classmethod
def get_music(cls):
if cls._music_pipeline is None:
cls._music_pipeline = pipeline("text-to-audio", "facebook/musicgen-small", device="cpu")
return cls._music_pipeline
# ๋ฉค๋ฒ„๋ณ„ ๊ณ ์œ  ์„ฑ๊ฒฉ/๋งํˆฌ ํŽ˜๋ฅด์†Œ๋‚˜
PERSONA_MAP = {
"์„œ์œค (Korea)": "์„œ์ •์ ์ด๋ฉด์„œ๋„ ๊ฐ•๋ ฌํ•œ K-Rock ๋ฆฌ๋”. ์ง„์ง€ํ•˜๊ณ  ์ฒ ํ•™์ ์ด๋ฉฐ ๋”ฐ๋œปํ•˜๊ฒŒ ์กฐ์–ธํ•จ.",
"Chloe (USA)": "์ž์œ ๋ถ„๋ฐฉํ•œ ์บ˜๋ฆฌํฌ๋‹ˆ์•„ ํŽ‘ํฌ ๋ฝ์Šคํƒ€. ๊ฑฐ์นจ์—†๊ณ  ์ฟจํ•˜๋ฉฐ ์—๋„ˆ์ œํ‹ฑํ•œ ๋งํˆฌ.",
"Naomi (Japan)": "์™„๋ฒฝ์ฃผ์˜ J-Rock ๊ธฐํƒ€๋ฆฌ์ŠคํŠธ. ์„ฌ์„ธํ•˜๊ณ  ์˜ˆ์˜ ๋ฐ”๋ฅด๋ฉฐ ์Œ์•… ์ด๋ก ์— ๋‚ ์นด๋กœ์›€.",
"Beatrice (Brazil)": "์—ด์ •์ ์ธ ์‚ผ๋ฐ” ๋ฝ์ปค. ๊ธ์ •์ ์ด๊ณ  ๋ฆฌ๋“ฌ๊ฐ์ด ๋„˜์น˜๋ฉฐ ์ธ์ƒ์˜ ์ฆ๊ฑฐ์›€์„ ๊ฐ•์กฐํ•จ.",
"Elena (Spain)": "๋“œ๋ผ๋งˆํ‹ฑํ•œ ๊ฐ์„ฑ์˜ ์†Œ์œ ์ž. ํ”Œ๋ผ๋ฉฉ์ฝ”์˜ ์ •์—ด๊ณผ ๋ฝ์˜ ํŒŒ๊ดด๋ ฅ์„ ๋™์‹œ์— ๊ฐ€์ง.",
"Amira (Egypt)": "์‹ ๋น„๋กœ์šด ์˜ค๋ฆฌ์—”ํƒˆ ๋ฝ์ปค. ๊นŠ์€ ์ง€ํ˜œ์™€ ๊ณ ์ „์ ์ธ ๋ฌด๊ฒŒ๊ฐ์„ ๋‹ด์•„ ์กฐ์–ธํ•จ.",
"Liwei (China)": "์ „ํ†ต๊ณผ ํ˜„๋Œ€์˜ ์กฐํ™”๋ฅผ ์ค‘์‹œํ•˜๋Š” ํ“จ์ „ ๋ฝ์ปค. ์ ˆ์ œ๋˜๊ณ  ํž˜ ์žˆ๋Š” ๋งํˆฌ.",
"Sophie (France)": "์˜ˆ์ˆ ์  ์ž์กด์‹ฌ์ด ๊ฐ•ํ•œ ์•„๋ฐฉ๊ฐ€๋ฅด๋“œ ๋ฝ์ปค. ์‹œ์ ์ด๊ณ  ์„ธ๋ จ๋œ ํ‘œํ˜„์„ ์ฆ๊ฒจ ์”€."
}
# ์–ธ์–ด๋ณ„ ๋ฉค๋ฒ„/๋ณด์ด์Šค ๋งคํ•‘
LANG_MEMBER_MAP = {
"Korean": {"name": "์„œ์œค (Korea)", "voice": "ko-KR-SunHiNeural"},
"English": {"name": "Chloe (USA)", "voice": "en-US-AriaNeural"},
"Japanese": {"name": "Naomi (Japan)", "voice": "ja-JP-NanamiNeural"},
"Portuguese": {"name": "Beatrice (Brazil)", "voice": "pt-BR-FranciscaNeural"},
"Spanish": {"name": "Elena (Spain)", "voice": "es-ES-ElviraNeural"},
"Arabic": {"name": "Amira (Egypt)", "voice": "ar-EG-SalmaNeural"},
"Chinese": {"name": "Liwei (China)", "voice": "zh-CN-XiaoxiaoNeural"},
"French": {"name": "Sophie (France)", "voice": "fr-FR-DeniseNeural"}
}
async def band_consulting(user_input, selected_lang, g_inst, b_inst, d_inst, chords):
try:
req_id = str(uuid.uuid4())[:8]
voice_path = f"/tmp/v_{req_id}.mp3"
music_path = f"/tmp/m_{req_id}.wav"
m_info = LANG_MEMBER_MAP.get(selected_lang, LANG_MEMBER_MAP["Korean"])
persona = PERSONA_MAP.get(m_info['name'], "์—ด์ •์ ์ธ ๋ฝ์Šคํƒ€")
# ๋ณธ๋ฌธ ์š”์•ฝ ๋ฐ ์ƒ์„ธ ํƒญ ๋ถ„๋ฆฌ ์ง€์‹œ ํ”„๋กฌํ”„ํŠธ
system_prompt = f"""You are the rock star '{m_info['name']}'. Respond ONLY in {selected_lang}.
Persona: {persona}.
CRITICAL RULES:
1. MAIN ADVICE: Provide exactly 3 to 5 deep, soulful sentences for the main chat.
2. [TAB] SECTION: Put ALL long explanations, music theory, Drum patterns (e.g., H|x-x-x-x| S|--o---o-|), and detailed Guitar Solo tabs here. This must be very technical and professional.
3. [TRANSLATION]: English translation of your 3-5 sentence MAIN ADVICE only.
4. [MUSIC]: English prompt for MusicGen reflecting: Guitar:{g_inst}, Bass:{b_inst}, Drums:{d_inst}, Chords:{chords}
"""
ai_text_raw = ""
groq_client = ModelManager.get_groq()
if groq_client:
try:
res = groq_client.chat.completions.create(
messages=[{"role": "system", "content": system_prompt}, {"role": "user", "content": user_input}],
model="llama-3.3-70b-versatile"
)
ai_text_raw = res.choices[0].message.content
except Exception as e:
print(f"Groq Error: {e}")
if not ai_text_raw:
qwen = ModelManager.get_qwen()
input_t = f"<|im_start|>system\n{system_prompt}<|im_end|>\n<|im_start|>user\n{user_input}<|im_end|>\nassistant\n"
out = qwen(input_t, max_new_tokens=1024)
ai_text_raw = out[0]['generated_text'].split("assistant\n")[-1]
# ํŒŒ์‹ฑ ๋กœ์ง
tab_match = re.search(r'\[TAB\](.*?)(\[|$)', ai_text_raw, re.DOTALL | re.IGNORECASE)
music_match = re.search(r'\[MUSIC:(.*?)\]', ai_text_raw, re.IGNORECASE)
trans_match = re.search(r'\[TRANSLATION\](.*?)(\[|$)', ai_text_raw, re.DOTALL | re.IGNORECASE)
tab_display = tab_match.group(1).strip() if tab_match else "No Detailed Data"
eng_translation = trans_match.group(1).strip() if trans_match else ""
# ์Œ์„ฑ์—์„œ ๋ฒˆ์—ญ๋ฌธ/ํƒญ/์Œ์•…ํ”„๋กฌํ”„ํŠธ ์™„๋ฒฝ ์ œ๊ฑฐ
speech_text = re.sub(r'\[TAB\].*?(\[|$)', '', ai_text_raw, flags=re.DOTALL | re.IGNORECASE)
speech_text = re.sub(r'\[MUSIC:.*?\]', '', speech_text, flags=re.IGNORECASE)
speech_text = re.sub(r'\[TRANSLATION\].*?(\[|$)', '', speech_text, flags=re.DOTALL | re.IGNORECASE).strip()
# TTS ์ƒ์„ฑ
tts_text = re.sub(r'[\*\#\-\_\~\|]', '', speech_text)
communicate = edge_tts.Communicate(tts_text, m_info["voice"])
await communicate.save(voice_path)
# MusicGen ์Œ์•… ์ƒ์„ฑ
music_gen = ModelManager.get_music()
music_p = music_match.group(1).strip() if music_match else "rock music"
music_output = music_gen(music_p, forward_params={"max_new_tokens": 512})
audio_data = np.squeeze(music_output["audio"])
audio_int16 = (audio_data * 32767).astype(np.int16)
scipy.io.wavfile.write(music_path, music_output["sampling_rate"], audio_int16)
return speech_text, voice_path, music_path, tab_display, eng_translation
except Exception as e:
print(f"Final Error: {e}")
return f"๋ฐด๋“œ ์‹œ์Šคํ…œ ์—๋Ÿฌ: {str(e)}", None, None, "No Data", "Error occurred"
with gr.Blocks() as demo:
inputs = [gr.Textbox(visible=False) for _ in range(6)]
outputs = [
gr.Textbox(visible=False),
gr.Audio(visible=False),
gr.Audio(visible=False),
gr.Textbox(visible=False),
gr.Textbox(visible=False)
]
btn = gr.Button("API", visible=False)
btn.click(band_consulting, inputs, outputs, api_name="predict")
demo.queue().launch()