双方向音声チャットを備えた独自の AI をホストするのは、思っているより簡単です。

LLM と音声機能の統合により、パーソナライズされた顧客とのやり取りに新たな機会が生まれました。

このガイドでは、Python、Transformers、Qwen2-Audio-7B-Instruct、Bark を使用して双方向の音声インタラクションをサポートするローカル LLM サーバーの設定方法について説明します。

前提条件

始める前に、次のものをインストールしておく必要があります。

• Python : バージョン 3.9 以上。
• PyTorch : モデルを実行するため。
• トランスフォーマー: Qwen モデルへのアクセスを提供します。
• 高速化: 一部の環境では必須です。
• FFmpeg & pydub : オーディオ処理用。
• FastAPI : Web サーバーを作成します。
• Uvicorn : FastAPI を実行する ASGI サーバー。
• Bark : テキスト音声合成用。
• Multipart & Scipy : オーディオを操作します。

FFmpeg は、Linux ではapt install ffmpeg 、MacOS ではbrew install ffmpegでインストールできます。

pip を使用して Python の依存関係をインストールできます: pip install torch transformers accelerate pydub fastapi uvicorn bark python-multipart scipy

ステップ1: 環境の設定

まず、Python 環境をセットアップし、PyTorch デバイスを選択しましょう。

 import torch device = 'cuda' if torch.cuda.is_available() else 'cpu'

このコードは、CUDA 互換 (Nvidia) GPU が使用可能かどうかを確認し、それに応じてデバイスを設定します。

そのような GPU が利用できない場合は、PyTorch は代わりに CPU 上で実行されますが、これははるかに低速です。

新しい Apple Silicon デバイスの場合、デバイスをmpsに設定して Metal 上で PyTorch を実行することもできますが、PyTorch Metal の実装は包括的ではありません。

ステップ2: モデルの読み込み

ほとんどのオープンソース LLM は、テキスト入力とテキスト出力のみをサポートしています。ただし、音声入力音声出力システムを作成したいので、(1) 音声を LLM に取り込む前にテキストに変換し、(2) LLM 出力を音声に戻すという 2 つのモデルを追加で使用しなければなりません。

Qwen Audio のようなマルチモーダル LLM を使用すると、音声入力をテキスト応答に処理するモデルを 1 つだけ使用し、LLM 出力を音声に戻すには 2 番目のモデルのみを使用すればよいことになります。

このマルチモーダル アプローチは、処理時間と (V)RAM 消費の点でより効率的であるだけでなく、入力オーディオが何の摩擦もなく直接 LLM に送信されるため、通常はより良い結果が得られます。

RunpodやVastなどのクラウド GPU ホスト上で実行している場合は、モデルをダウンロードする前に、 export HF_HOME=/workspace/hfとexport XDG_CACHE_HOME=/workspace/barkを実行して、HuggingFace のホーム ディレクトリと Bark ディレクトリをボリューム ストレージに設定する必要があります。

 from transformers import AutoProcessor, Qwen2AudioForConditionalGeneration model_name = "Qwen/Qwen2-Audio-7B-Instruct" processor = AutoProcessor.from_pretrained(model_name) model = Qwen2AudioForConditionalGeneration.from_pretrained(model_name, device_map="auto").to(device)

ここでは、計算要件を減らすために、Qwen Audio モデル シリーズの小型 7B バリアントを使用することにしました。ただし、この記事を読んでいる頃には、Qwen はより強力で大型のオーディオ モデルをリリースしている可能性があります。HuggingFaceですべての Qwen モデルを表示して、最新モデルを使用していることを再確認できます。

実稼働環境では、スループットを大幅に向上させるために、 vLLMなどの高速推論エンジンを使用することをお勧めします。

ステップ3: Barkモデルの読み込み

Bark は、複数の言語とサウンド効果をサポートする最先端のオープンソースのテキスト読み上げ AI モデルです。

 from bark import SAMPLE_RATE, generate_audio, preload_models preload_models()

Bark 以外にも、オープンソースまたは独自のテキスト読み上げモデルを使用することもできます。独自のモデルの方がパフォーマンスは高いかもしれませんが、コストがかなり高いことに注意してください。TTSアリーナでは最新の比較が維持されています。

Qwen Audio 7B と Bark の両方をメモリにロードすると、(V)RAM の使用量はおよそ 24 GB になります。そのため、ハードウェアがこれをサポートしていることを確認してください。サポートしていない場合は、 Qwen モデルの量子化バージョンを使用してメモリを節約できます。

ステップ4: FastAPIサーバーのセットアップ

受信したオーディオまたはテキスト入力を処理し、オーディオ応答を返す 2 つのルートを持つ FastAPI サーバーを作成します。

 from fastapi import FastAPI, UploadFile, Form from fastapi.responses import StreamingResponse import uvicorn app = FastAPI() @app.post("/voice") async def voice_interaction(file: UploadFile): # TODO return @app.post("/text") async def text_interaction(text: str = Form(...)): # TODO return if __name__ == "__main__":  uvicorn.run(app, host="0.0.0.0", port=8000)

このサーバーは、 /voiceおよび/textエンドポイントで POST リクエストを介してオーディオ ファイルを受け入れます。

ステップ5: オーディオ入力の処理

ffmpeg を使用して、受信したオーディオを処理し、Qwen モデル用に準備します。

 from pydub import AudioSegment from io import BytesIO import numpy as np def audiosegment_to_float32_array(audio_segment: AudioSegment, target_rate: int = 16000) -> np.ndarray: audio_segment = audio_segment.set_frame_rate(target_rate).set_channels(1) samples = np.array(audio_segment.get_array_of_samples(), dtype=np.int16) samples = samples.astype(np.float32) / 32768.0 return samples def load_audio_as_array(audio_bytes: bytes) -> np.ndarray: audio_segment = AudioSegment.from_file(BytesIO(audio_bytes)) float_array = audiosegment_to_float32_array(audio_segment, target_rate=16000) return float_array

ステップ6: Qwenでテキスト応答を生成する

処理されたオーディオを使用して、Qwen モデルを使用してテキスト応答を生成できます。これには、テキストとオーディオの両方の入力を処理する必要があります。

プリプロセッサは、入力をモデルのチャット テンプレート (Qwen の場合は ChatML) に変換します。

 def generate_response(conversation): 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": audio_array = load_audio_as_array(ele["audio_url"]) audios.append(audio_array) if audios: inputs = processor( text=text, audios=audios, return_tensors="pt", padding=True ).to(device) else: inputs = processor( text=text, return_tensors="pt", padding=True ).to(device) generate_ids = model.generate(**inputs, max_length=256) generate_ids = generate_ids[:, inputs.input_ids.size(1):] response = processor.batch_decode( generate_ids, skip_special_tokens=True, clean_up_tokenization_spaces=False )[0] return response

model.generate関数の温度などの生成パラメータを自由に試してみてください。

ステップ 7: Bark でテキストを音声に変換する

最後に、生成されたテキスト応答を音声に変換します。

 from scipy.io.wavfile import write as write_wav def text_to_speech(text): audio_array = generate_audio(text) output_buffer = BytesIO() write_wav(output_buffer, SAMPLE_RATE, audio_array) output_buffer.seek(0) return output_buffer

ステップ8: APIにすべてを統合する

エンドポイントを更新して、オーディオまたはテキスト入力を処理し、応答を生成し、合成された音声を WAV ファイルとして返します。

 @app.post("/voice") async def voice_interaction(file: UploadFile): audio_bytes = await file.read() conversation = [ { "role": "user", "content": [ { "type": "audio", "audio_url": audio_bytes } ] } ] response_text = generate_response(conversation) audio_output = text_to_speech(response_text) return StreamingResponse(audio_output, media_type="audio/wav") @app.post("/text") async def text_interaction(text: str = Form(...)): conversation = [ {"role": "user", "content": [{"type": "text", "text": text}]} ] response_text = generate_response(conversation) audio_output = text_to_speech(response_text) return StreamingResponse(audio_output, media_type="audio/wav")

アシスタントの応答をより細かく制御するために、会話にシステム メッセージを追加することもできます。

ステップ9: テストする

次のようにcurl使用してサーバーに ping を実行できます。

 # Audio input curl -X POST http://localhost:8000/voice --output output.wav -F "[email protected]" # Text input curl -X POST http://localhost:8000/text --output output.wav -H "Content-Type: application/x-www-form-urlencoded" -d "text=Hey"

結論

これらの手順に従うことで、最先端のモデルを使用して双方向の音声対話が可能なシンプルなローカル サーバーをセットアップできました。このセットアップは、より複雑な音声対応アプリケーションを構築するための基盤として役立ちます。

アプリケーション

AI を活用した言語モデルを収益化する方法を検討している場合は、次の潜在的なアプリケーションを検討してください。

• チャットボット（例：キャラクター AI 、 NSFW AI チャット）
• 電話エージェント (例: Synthflow 、 Bland )
• カスタマーサポート自動化（例： Zendesk 、 Forethought ）
• 法務アシスタント ( Harvey AI 、 Leya AI )

完全なコード

import torch from fastapi import FastAPI, UploadFile, Form from fastapi.responses import StreamingResponse import uvicorn from transformers import AutoProcessor, Qwen2AudioForConditionalGeneration from bark import SAMPLE_RATE, generate_audio, preload_models from scipy.io.wavfile import write as write_wav from pydub import AudioSegment from io import BytesIO import numpy as np device = 'cuda' if torch.cuda.is_available() else 'cpu' model_name = "Qwen/Qwen2-Audio-7B-Instruct" processor = AutoProcessor.from_pretrained(model_name) model = Qwen2AudioForConditionalGeneration.from_pretrained(model_name, device_map="auto").to(device) preload_models() app = FastAPI() def audiosegment_to_float32_array(audio_segment: AudioSegment, target_rate: int = 16000) -> np.ndarray: audio_segment = audio_segment.set_frame_rate(target_rate).set_channels(1) samples = np.array(audio_segment.get_array_of_samples(), dtype=np.int16) samples = samples.astype(np.float32) / 32768.0 return samples def load_audio_as_array(audio_bytes: bytes) -> np.ndarray: audio_segment = AudioSegment.from_file(BytesIO(audio_bytes)) float_array = audiosegment_to_float32_array(audio_segment, target_rate=16000) return float_array def generate_response(conversation): 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": audio_array = load_audio_as_array(ele["audio_url"]) audios.append(audio_array) if audios: inputs = processor( text=text, audios=audios, return_tensors="pt", padding=True ).to(device) else: inputs = processor( text=text, return_tensors="pt", padding=True ).to(device) generate_ids = model.generate(**inputs, max_length=256) generate_ids = generate_ids[:, inputs.input_ids.size(1):] response = processor.batch_decode( generate_ids, skip_special_tokens=True, clean_up_tokenization_spaces=False )[0] return response def text_to_speech(text): audio_array = generate_audio(text) output_buffer = BytesIO() write_wav(output_buffer, SAMPLE_RATE, audio_array) output_buffer.seek(0) return output_buffer @app.post("/voice") async def voice_interaction(file: UploadFile): audio_bytes = await file.read() conversation = [ { "role": "user", "content": [ { "type": "audio", "audio_url": audio_bytes } ] } ] response_text = generate_response(conversation) audio_output = text_to_speech(response_text) return StreamingResponse(audio_output, media_type="audio/wav") @app.post("/text") async def text_interaction(text: str = Form(...)): conversation = [ {"role": "user", "content": [{"type": "text", "text": text}]} ] response_text = generate_response(conversation) audio_output = text_to_speech(response_text) return StreamingResponse(audio_output, media_type="audio/wav") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)