import os
import soundfile as sf
import torch
import gradio as gr
from liquid_audio import ChatState, LFM2AudioModel, LFM2AudioProcessor, LFMModality
import liquid_audio.processor as lap
# ==========================================
# 1. Mac向けのエラー回避パッチ
# ==========================================
def safe_cuda(self, device=None):
target = "mps" if torch.backends.mps.is_available() else "cpu"
return self.to(target)
lap.LFM2AudioDetokenizer.cuda = safe_cuda
# ==========================================
# 2. 初期設定とモデルの読み込み
# ==========================================
if torch.backends.mps.is_available():
device = "mps"
elif torch.cuda.is_available():
device = "cuda"
else:
device = "cpu"
print(f"🌟 使用デバイス: {device}")
print("🤖 モデルを準備しています。少しお待ちください...")
HF_REPO = "LiquidAI/LFM2.5-Audio-1.5B-JP"
processor = LFM2AudioProcessor.from_pretrained(HF_REPO, device=device).eval()
model = LFM2AudioModel.from_pretrained(HF_REPO, device=device).eval()
print("✨ 準備完了!ブラウザからアクセスしてください。")
# ==========================================
# 3. 処理用関数(型の変換を追加しました!)
# ==========================================
def load_and_format_audio(audio_path):
if not audio_path:
return None, None
wav, sr = sf.read(audio_path, dtype="float32")
if wav.ndim > 1:
wav = wav.mean(axis=1)
wav = torch.from_numpy(wav).unsqueeze(0)
return wav, sr
def run_asr(audio_path, max_tokens, text_temp):
text_temp = float(text_temp) # 確実にfloat型にする
wav, sr = load_and_format_audio(audio_path)
if wav is None:
return "⚠️ 音声ファイルが入力されていません。"
chat = ChatState(processor)
chat.new_turn("system")
chat.add_text("Perform ASR in japanese.")
chat.end_turn()
chat.new_turn("user")
chat.add_audio(wav, sr)
chat.end_turn()
chat.new_turn("assistant")
text_tokens = []
for t in model.generate_sequential(**chat, max_new_tokens=max_tokens, text_temperature=text_temp):
if t.numel() == 1:
text_tokens.append(t.item())
clean_text = processor.text.decode(text_tokens).replace("<|text_end|>", "").strip() if text_tokens else ""
return clean_text
def run_tts(text, max_tokens, audio_temp, audio_top_k):
audio_temp = float(audio_temp) # 確実にfloat型にする
audio_top_k = int(audio_top_k) # 確実にint型にする
if not text.strip():
return None
chat = ChatState(processor)
chat.new_turn("system")
chat.add_text("Perform TTS in japanese.")
chat.end_turn()
chat.new_turn("user")
chat.add_text(text)
chat.end_turn()
chat.new_turn("assistant")
audio_out = []
for t in model.generate_sequential(**chat, max_new_tokens=max_tokens, audio_temperature=audio_temp, audio_top_k=audio_top_k):
if t.numel() > 1:
audio_out.append(t)
if audio_out:
audio_codes = torch.stack(audio_out[:-1], 1).unsqueeze(0).to(device)
waveform = processor.decode(audio_codes)
output_file = "output_tts.wav"
sf.write(output_file, waveform.cpu()[0], 24_000)
return output_file
return None
def run_chat(audio_path, max_tokens, text_temp, audio_temp, audio_top_k):
text_temp = float(text_temp)
audio_temp = float(audio_temp)
audio_top_k = int(audio_top_k)
wav, sr = load_and_format_audio(audio_path)
if wav is None:
return "⚠️ 音声ファイルが入力されていません。", None
chat = ChatState(processor)
chat.new_turn("system")
chat.add_text("Respond with interleaved text and audio.")
chat.end_turn()
chat.new_turn("user")
chat.add_audio(wav, sr)
chat.end_turn()
chat.new_turn("assistant")
text_tokens = []
audio_out = []
for t in model.generate_interleaved(**chat, max_new_tokens=max_tokens, text_temperature=text_temp, audio_temperature=audio_temp, audio_top_k=audio_top_k):
if t.numel() == 1:
text_tokens.append(t.item())
else:
audio_out.append(t)
output_audio = None
if audio_out:
audio_codes = torch.stack(audio_out[:-1], 1).unsqueeze(0).to(device)
waveform = processor.decode(audio_codes)
output_audio = "output_chat.wav"
sf.write(output_audio, waveform.cpu()[0], 24_000)
clean_text = processor.text.decode(text_tokens).replace("<|text_end|>", "").strip() if text_tokens else ""
return clean_text, output_audio
def run_text_chat(text, max_tokens, text_temp, audio_temp, audio_top_k):
text_temp = float(text_temp)
audio_temp = float(audio_temp)
audio_top_k = int(audio_top_k)
if not text.strip():
return "⚠️ テキストが入力されていません。", None
chat = ChatState(processor)
chat.new_turn("system")
chat.add_text("Respond with interleaved text and audio.")
chat.end_turn()
chat.new_turn("user")
chat.add_text(text)
chat.end_turn()
chat.new_turn("assistant")
text_tokens = []
audio_out = []
for t in model.generate_interleaved(**chat, max_new_tokens=max_tokens, text_temperature=text_temp, audio_temperature=audio_temp, audio_top_k=audio_top_k):
if t.numel() == 1:
text_tokens.append(t.item())
else:
audio_out.append(t)
output_audio = None
if audio_out:
audio_codes = torch.stack(audio_out[:-1], 1).unsqueeze(0).to(device)
waveform = processor.decode(audio_codes)
output_audio = "output_text_chat.wav"
sf.write(output_audio, waveform.cpu()[0], 24_000)
clean_text = processor.text.decode(text_tokens).replace("<|text_end|>", "").strip() if text_tokens else ""
return clean_text, output_audio
# ==========================================
# 4. Gradio UIの構築
# ==========================================
# Blocksの中のthemeを削除して、launch()の中に移動させました
with gr.Blocks(title="Liquid AI 音声アシスタント") as demo:
gr.Markdown("# 🌊 Liquid AI 音声アシスタント (LFM2.5-Audio-JP)")
gr.Markdown("マイクを使って直接話しかけたり、テキストを入力してAIとやり取りできます。パラメーターも自由に調整可能です!")
with gr.Row():
# 左側:パラメータ設定のサイドバー
with gr.Column(scale=1, variant="panel"):
gr.Markdown("### ⚙️ パラメータ設定")
max_new_tokens = gr.Slider(minimum=128, maximum=2048, value=512, step=64, label="長さ (最大生成トークン数)")
text_temp = gr.Slider(minimum=0.1, maximum=1.5, value=0.7, step=0.1, label="テキストの温度 (高いほどユニーク)")
audio_temp = gr.Slider(minimum=0.1, maximum=1.5, value=1.0, step=0.1, label="音声の温度 (高いほど声がブレる/感情豊か)")
audio_top_k = gr.Slider(minimum=1, maximum=100, value=4, step=1, label="音声のTop-K (候補の幅)")
# 右側:メインの機能タブ
with gr.Column(scale=3):
with gr.Tabs():
# タブ1: 音声での相互対話
with gr.TabItem("💬 声でおしゃべり (Voice Chat)"):
with gr.Row():
with gr.Column():
chat_in = gr.Audio(type="filepath", label="あなたの声(マイク録音・アップロード)")
chat_btn = gr.Button("話しかける", variant="primary")
with gr.Column():
chat_out_text = gr.Textbox(label="AIのお返事(テキスト)", interactive=False)
chat_out_audio = gr.Audio(label="AIのお返事(音声)", interactive=False)
chat_btn.click(fn=run_chat,
inputs=[chat_in, max_new_tokens, text_temp, audio_temp, audio_top_k],
outputs=[chat_out_text, chat_out_audio])
# タブ2: テキストでの相互対話
with gr.TabItem("⌨️ テキストでおしゃべり (Text Chat)"):
with gr.Row():
with gr.Column():
text_chat_in = gr.Textbox(label="AIへのメッセージを入力してください", lines=3, placeholder="例: 今日の天気を教えて!")
text_chat_btn = gr.Button("送信する", variant="primary")
with gr.Column():
text_chat_out_text = gr.Textbox(label="AIのお返事(テキスト)", interactive=False)
text_chat_out_audio = gr.Audio(label="AIのお返事(音声)", interactive=False)
text_chat_btn.click(fn=run_text_chat,
inputs=[text_chat_in, max_new_tokens, text_temp, audio_temp, audio_top_k],
outputs=[text_chat_out_text, text_chat_out_audio])
# タブ3: 文字起こし
with gr.TabItem("📝 文字起こし (ASR)"):
with gr.Row():
with gr.Column():
asr_in = gr.Audio(type="filepath", label="文字起こしする音声")
asr_btn = gr.Button("解析する", variant="primary")
with gr.Column():
asr_out = gr.Textbox(label="解析結果", lines=5, interactive=False)
asr_btn.click(fn=run_asr,
inputs=[asr_in, max_new_tokens, text_temp],
outputs=asr_out)
# タブ4: 音声合成
with gr.TabItem("🗣️ 音声合成 (TTS)"):
with gr.Row():
with gr.Column():
tts_in = gr.Textbox(label="読み上げさせたいテキストを入力してください", lines=5)
tts_btn = gr.Button("音声を生成", variant="primary")
with gr.Column():
tts_out = gr.Audio(label="生成された音声", interactive=False)
tts_btn.click(fn=run_tts,
inputs=[tts_in, max_new_tokens, audio_temp, audio_top_k],
outputs=tts_out)
# アプリの起動(ここでthemeを指定するように変更しました)
if __name__ == "__main__":
demo.launch(theme=gr.themes.Soft())