はじめに
この記事は、AIアクセラレータLLM8850を搭載したRaspberry Pi 5で、VOICEVOX音声のAIと日本語でおしゃべりすることを目的としています。
※ Geminiによるイメージ図
ハードウェア構成
-
Raspberry Pi 5（8GB）
2GBだと厳しいですが4GBならいけるかもしれません。（未確認 後述） -
Raspberry Pi 5用アクティブクーラー
動作中は頻繁に70度に達して冷却、を繰り返していました。
80度まで上がって処理が落ちない程度の冷却が必須です。 -
LLM8850
MakerFareTokyo2025で購入しました。
国内ではSWITCH SCIENCEさんの販売のみ？ ※執筆中現在売り切れ中
SWITCH SCIENCE LLM8850販売ページ -
Raspberry Pi M.2 HAT+
M.2変換はいろいろありますがLLM8850は電力や帯域幅の制約があり、公式の変換ボードを使うのがよいそうです。参考
秋月電子 M.2 Hat+ -
USBスピーカー&マイク
Raspberry Pi 5だとオーディオジャックがついてません。
Bluetoothだとすっきりしますがいろいろ設定が必要そうなのでUSBにしました。 -
電源
LLM8850の公式ガイドにはDC 5V@3Aとありますが、Raspberry Pi 5でも同時にVOICEVOXをフル動作させると4Aを超えることがあり、電圧低下も見られました。
5.1V@5AのACアダプタを使っておくのが無難と思われます。
※ワットチェッカーを使って計測していましたが、ワットチェッカー接続時のみ低電圧警告と強制シャットダウンが発生していました。電圧低下はこれのケーブルのせいかも -
USB3.0接続SSD
数GBのファイルの読み書きが頻繁に行われるので、マイクロSDではなくSSD起動にしておくことをお勧めします。
デュアルM.2 HatでM.2 SSDと併用、はできなそうです。
LLM8850について
LLM8850はM5Stack社から今年10月？に発売された、M.2対応のAIアクセラレータです。
Raspberry Pi 5にM.2 HAT+経由で接続することができます。
Raspberry Pi 5で使えるAIアクセラレータとしてはAI HAT+がありましたが、画像処理に特化していました。
LLM8850は画像処理に加えて、LLMや音声処理もできるようになったもの、と思えばいいかと思います。
LLM8850の使い方
基本的なLLM8850の環境設定は公式チュートリアルや他記事を参考にしてください。
本記事では、LLM8850を使った標準入出力方式でのLLM動作確認ができていることを前提にしています。
LLMモデルのAPI化
公式ページにいくつかモデルごとのサンプルがありますが、このページにAPI化の方法が載っています。
https://docs.m5stack.com/ja/guide/ai_accelerator/llm-8850/m5_llm_8850_qwen3_1.7b
サンプルページのモデルはQwen3-1.7Bですが、本記事では別のQwen3-4BをAPI化してみます。
1. Qwen3 4Bモデルを入手
公式ガイドにリンクはないですが、こちらにあるモデルなら動くのではないかと思います。
Qwen2.5 7Bモデルなんてのもありますが2.0 tokens/secはさすがに・・。
$ git clone https://huggingface.co/AXERA-TECH/Qwen3-4B
$ cd Qwen3-4B
2. python仮想環境の作成と依存パッケージのインストール
$ python -m venv env
$ source env/bin/activate
(env) $ pip install transformers jinja2 torch
3. API用調整
実行ファイルの権限追加、および実行スクリプトをRaspberry Pi 5用に調整します。
$ chmod +x main_api_axcl_aarch64
$ cp run_qwen3_4b_int8_ctx_axcl_x86_api.sh run_qwen3_4b_int8_ctx_axcl_aarch64_api.sh
$ chmod +x run_qwen3_4b_int8_ctx_axcl_aarch64_api.sh
$ vi run_qwen3_4b_int8_ctx_axcl_aarch64_api.sh
# 変更前のファイル内容 (一部抜粋)
-./main_api_axcl_x86 \
- --system_prompt "You are Qwen, a helpful assistant." \
- --devices=0,1
+./main_api_axcl_aarch64 \
+ --system_prompt "あなたはユーザーの質問に正確に答えるAIアシスタントです。" \
+ --devices=0
4. 実行
標準入出力方式のチュートリアルと同様に複数ターミナルで起動します。
ターミナル1 （標準入出力方式と同じ）
(env) $ python qwen3_tokenizer_uid.py
ターミナル2 （apiとついてる方のファイルであることに注意）
$ ./run_qwen3_4b_int8_ctx_axcl_aarch64_api.sh
ターミナル3
リクエストすると返答の生成開始
$ curl -X POST "http://localhost:8000/api/generate" \
-H "Content-Type: application/json" \
-d '{
"prompt": "あなたの名前は？",
"temperature": 0.7,
"top-k": 40
}'
生成中の文章をポーリングして取得
curl "http://localhost:8000/api/generate_provider"
実行例
動かない時
- run_qwen3_4b_int8_ctx_axcl_aarch64_api.sh 実行時に止まってしまう場合
[I][ Init][ 263]: LLM init ok
Server running on port 8000...
[I][ run][ 80]: AXCLWorker exit with devid 0
OpenAIのAPIとは共存できないようなので止める必要があります。
$ sudo systemctl stop llm-openai-api
VOICEVOX
1. Dockerのインストール
$ curl -fsSL https://get.docker.com -o get-docker.sh
$ sudo sh get-docker.sh
$ sudo gpasswd -a $USER docker
$ newgrp docker
$ docker run --rm hello-world
2. VOICEVOXエンジンの取得と起動
デフォルトだとCPUコアの半分しか使わないので、Raspberry Pi 5の4コアすべて使う設定にします。
$ docker pull voicevox/voicevox_engine:cpu-ubuntu20.04-latest
$ docker run --rm -it -d -e VV_CPU_NUM_THREADS=4 -p ':50021:50021' voicevox/voicevox_engine:cpu-ubuntu20.04-latest
3. テスト実行
スピーカーからずんだもんの声が聞こえれば成功です
$ curl -s -X POST "localhost:50021/audio_query?speaker=1" --get --data-urlencode "text=これはテストなのだ" > query.json \
&& curl -s -H "Content-Type: application/json" -X POST -d @query.json "localhost:50021/synthesis?speaker=1" > audio.wav \
&& aplay audio.wav
メインプログラム
構成要素
| 機能 | 使用技術 | 役割 | 最適化ポイント |
|---|---|---|---|
| 音声認識 (STT) | Faster-Whisper | マイク音声をテキスト化 | CPU最適化 (compute_type="int8") |
| 大規模言語モデル (LLM) | LLM8850 API (localhost:8000) | 質問応答、推論 | ポーリングによるストリーミング |
| 音声合成 (TTS) | VOICEVOX (localhost:50021) | テキストを高品質な音声へ | 4コアフル活用 (VV_CPU_NUM_THREADS=4) |
| 並列処理 | Python threading & queue | 合成と再生の非同期化 | キューイングによる再生の逐次実行 |
処理の流れ
- 5秒間録音
- Faster-Whisperで音声認識
- 認識結果をユーザープロンプトとしてLLMにAPIリクエスト
- 生成結果をポーリングして細かく取得（ストリーミング処理）
- 生成テキストがチャンクで区切れたら不要なタグを除去し、VOICEVOXにリクエストして読み上げ音声を生成。
WAVファイルを再生キューに追加（非同期）。 - 再生キュー管理
最初の文のWAVファイルは、2文目の合成が完了するまで再生を遅延（途切れ防止の工夫）。
バックグラウンドの再生ワーカーが、キューからWAVファイルを順番に取り出し、再生が完了するまで逐次実行（再生の重なり防止）。 - すべての音声読み上げが終わったら（再生キューが空になったら）、LLMキャッシュをリセット。
- 1に戻る
python環境構築
$ cd
$ mkdir speakllm
$ cd speakllm
$ python -m venv env
$ source env/bin/activate
(env) $ pip install pyaudio numpy requests faster_whisper
実行
$ vi speakllm.py
実行スクリプト(クリックして展開) ※ほぼGemini製
speakllm.py
import pyaudio
import numpy as np
import wave
import os
import requests
import json
import subprocess
import time
import threading
import queue
import re
from faster_whisper import WhisperModel
# ===============================================
# 1. 設定
# ===============================================
CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
RECORD_SECONDS = 5
WAVE_RECORD_FILENAME = "temp_recording.wav"
MODEL_SIZE = "small"
LLM_API_BASE = "http://localhost:8000"
LLM_GENERATE_URL = f"{LLM_API_BASE}/api/generate"
LLM_POLLING_URL = f"{LLM_API_BASE}/api/generate_provider"
LLM_RESET_URL = f"{LLM_API_BASE}/api/reset"
LLM_TEMP = 0.7
LLM_TOP_K = 40
# 繰り返し防止
LLM_REPETITION_PENALTY = 1.2
# システムプロンプト ここでAIの性格付け等を記入
SYSTEM_PROMPT = "あなたは「ずんだもん」という名前の妖精です。枝豆や「ずんだ餅」が好物です。一人称は「僕」、語尾に「〜のだ」「〜なのだ」をつけてフレンドリーに話してください。性別不詳ですが男の子っぽい話し方をしてください。回答は簡潔にし、同じ内容や表現を繰り返さないでください。"
VOICEVOX_API_URL = "http://localhost:50021"
# 話者ID ずんだもん(NORMAL)の場合3
SPEAKER_ID = 3
# 読み上げ速度
SPEED_SCALE = 0.9
playback_queue = queue.Queue()
TEMP_FILE_PATH = "temp_files"
SHOULD_STOP_WORKER = threading.Event()
CONVERSATION_HISTORY = [
{"role": "system", "content": SYSTEM_PROMPT}
]
# ===============================================
# --- Whisper モデルのロード ---
print(f"Loading Whisper model: {MODEL_SIZE}...")
try:
WHISPER_MODEL = WhisperModel(MODEL_SIZE, device="cpu", compute_type="int8")
print("Whisper Model loaded successfully.")
except Exception as e:
print(f"Error loading Whisper model: {e}")
exit()
# ===============================================
## 2. ヘルパー関数群
# ===============================================
def transcribe_audio():
p = pyaudio.PyAudio()
stream = p.open(format=FORMAT, channels=CHANNELS, rate=RATE, input=True, frames_per_buffer=CHUNK)
print(f"\n🎙️ Recording for {RECORD_SECONDS} seconds. Speak now...")
frames = []
for _ in range(0, int(RATE / CHUNK * RECORD_SECONDS)):
data = stream.read(CHUNK, exception_on_overflow=False)
frames.append(data)
print("Recording finished. 📝 Transcribing...")
stream.stop_stream()
stream.close()
p.terminate()
with wave.open(WAVE_RECORD_FILENAME, 'wb') as wf:
wf.setnchannels(CHANNELS)
wf.setsampwidth(p.get_sample_size(FORMAT))
wf.setframerate(RATE)
wf.writeframes(b''.join(frames))
recognized_text = ""
try:
segments, _ = WHISPER_MODEL.transcribe(WAVE_RECORD_FILENAME, beam_size=5, language="ja")
for segment in segments:
recognized_text += segment.text
except Exception as e:
print(f"Error during transcription: {e}")
if os.path.exists(WAVE_RECORD_FILENAME):
os.remove(WAVE_RECORD_FILENAME)
return recognized_text.strip()
def voicevox_to_speech(text, output_filename):
if not text.strip(): return False
query_url = f"{VOICEVOX_API_URL}/audio_query"
query_params = {"text": text, "speaker": SPEAKER_ID}
try:
query_response = requests.post(query_url, params=query_params)
query_response.raise_for_status()
except requests.exceptions.RequestException:
print(f"VOICEVOX audio_query エラー。エンジン起動を確認してください。")
return False
query_data = query_response.json()
query_data['speedScale'] = SPEED_SCALE
synthesis_url = f"{VOICEVOX_API_URL}/synthesis"
synthesis_params = {"speaker": SPEAKER_ID}
try:
synthesis_response = requests.post(
synthesis_url,
params=synthesis_params,
json=query_data,
timeout=30
)
synthesis_response.raise_for_status()
except requests.exceptions.RequestException as e:
print(f"VOICEVOX synthesis エラー: {e}")
return False
with open(output_filename, 'wb') as f:
f.write(synthesis_response.content)
return True
def play_audio(filename):
playback_queue.put(filename)
def playback_worker():
print(" [Worker] Playback worker started.")
while not SHOULD_STOP_WORKER.is_set():
try:
filename = playback_queue.get(timeout=0.1)
if os.path.exists(filename):
try:
subprocess.run(["aplay", filename], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except FileNotFoundError:
print("エラー: 'aplay' コマンドが見つかりません。")
except subprocess.CalledProcessError:
pass
finally:
os.remove(filename)
playback_queue.task_done()
except queue.Empty:
continue
except Exception as e:
print(f"致命的な再生エラー: {e}")
break
while not playback_queue.empty():
filename = playback_queue.get_nowait()
if os.path.exists(filename):
os.remove(filename)
print(" [Worker] Playback worker stopped.")
def reset_llm_context():
"""LLM8850 APIにリセットコマンドを送信し、ローカルの履歴をリセットする"""
global CONVERSATION_HISTORY
print("\n⚠️ LLMサーバーのコンテキストをリセットしています...")
reset_payload = {
"system_prompt": SYSTEM_PROMPT
}
try:
response = requests.post(LLM_RESET_URL, json=reset_payload, timeout=10)
response.raise_for_status()
CONVERSATION_HISTORY = [
{"role": "system", "content": SYSTEM_PROMPT}
]
print("✅ LLMサーバーのキャッシュとローカル履歴をリセットしました。")
except requests.exceptions.RequestException as e:
print(f"❌ LLMリセットAPIとの通信エラー: {e}。推論の継続は保証できません。")
# ===============================================
## 3. コアロジック: LLM8850 ポーリング処理
# ===============================================
def build_llm_prompt_from_history(current_prompt):
"""会話履歴をLLM8850 APIが処理できる単一のプロンプト文字列に整形する"""
global CONVERSATION_HISTORY
full_prompt = ""
for message in CONVERSATION_HISTORY:
role = message["role"]
content = message["content"]
if role == "user":
full_prompt += f"ユーザー: {content}\n"
elif role == "assistant":
full_prompt += f"アシスタント: {content}\n"
full_prompt += f"ユーザー: {current_prompt}\nアシスタント:"
return full_prompt.strip()
def stream_and_speak(current_prompt):
"""
LLM8850 APIにリクエストを開始し、ポーリングで応答を逐次取得し、再生キューに入れる
"""
global TEMP_FILE_PATH, CONVERSATION_HISTORY
if not os.path.exists(TEMP_FILE_PATH):
os.makedirs(TEMP_FILE_PATH)
llm_input_prompt = build_llm_prompt_from_history(current_prompt)
print(f"\n🤖 Starting generation on LLM8850...")
generate_payload = {
"prompt": llm_input_prompt,
"temperature": LLM_TEMP,
"top-k": LLM_TOP_K,
"repetition-penalty": LLM_REPETITION_PENALTY
}
try:
response = requests.post(LLM_GENERATE_URL, json=generate_payload, timeout=60)
response.raise_for_status()
start_result = response.json()
if start_result.get("status") != "ok":
print(f"エラー: LLM推論開始に失敗しました: {start_result}")
return None
except requests.exceptions.RequestException as e:
print(f"エラー: LLM推論開始APIとの通信エラー: {e}")
return None
# 3. LLM応答のポーリング (API Step 2: /api/generate_provider)
print("--- アシスタント応答 ---")
full_response = ""
current_chunk = ""
punctuation = "。、？！\n"
POLLING_INTERVAL = 0.1
tts_chunk_count = 0
while True:
try:
poll_response = requests.get(LLM_POLLING_URL, timeout=10)
poll_response.raise_for_status()
data = poll_response.json()
is_done = data.get("done", False)
text_chunk = data.get("response", "")
if text_chunk:
full_response += text_chunk
current_chunk += text_chunk
# 句読点、または完了信号でチャンクを処理
if is_done or any(p in current_chunk for p in punctuation):
tts_text = ""
if is_done:
tts_text = current_chunk
current_chunk = ""
else:
break_index = -1
if "。" in current_chunk:
break_index = current_chunk.rfind("。")
else:
for p in punctuation:
if p in current_chunk:
break_index = max(break_index, current_chunk.rfind(p))
if break_index != -1:
tts_text = current_chunk[:break_index + 1]
current_chunk = current_chunk[break_index + 1:]
# TTS生成リクエスト (VOICEVOX) とキューイング
if tts_text.strip():
filtered_text = re.sub(r'<[^>]+>.*?</[^>]+>', '', tts_text, flags=re.DOTALL)
filtered_text = re.sub(r'<[^>]+?>', '', filtered_text)
if filtered_text.strip():
print(f"{filtered_text}")
tts_chunk_count += 1
temp_wav_file = os.path.join(TEMP_FILE_PATH, f"temp_tts_{int(time.time() * 1000)}.wav")
if voicevox_to_speech(filtered_text, temp_wav_file):
if tts_chunk_count > 1:
play_audio(temp_wav_file)
else:
playback_queue.put(temp_wav_file)
if is_done:
if tts_chunk_count > 0:
try:
first_chunk_file = playback_queue.get_nowait()
play_audio(first_chunk_file)
playback_queue.task_done()
except queue.Empty:
pass
break
time.sleep(POLLING_INTERVAL)
except requests.exceptions.RequestException as e:
print(f"\nエラー: LLM応答ポーリング中に通信エラーが発生しました: {e}")
break
except json.JSONDecodeError:
time.sleep(POLLING_INTERVAL)
continue
# 4. 会話履歴の更新と終了処理
CONVERSATION_HISTORY.append({"role": "user", "content": current_prompt})
if full_response:
CONVERSATION_HISTORY.append({"role": "assistant", "content": full_response})
reset_llm_context()
# キューが空になるのを待つ
playback_queue.join()
# 一時ファイルディレクトリの削除
if os.path.exists(TEMP_FILE_PATH) and not os.listdir(TEMP_FILE_PATH):
os.rmdir(TEMP_FILE_PATH)
return full_response
# ===============================================
## 4. メイン処理 (継続会話ループ付き)
# ===============================================
if __name__ == "__main__":
# --- 1. 再生ワーカーの開始 ---
worker_thread = threading.Thread(target=playback_worker)
worker_thread.start()
print("システム起動。再生ワーカーをバックグラウンドで開始しました。")
# 起動時にLLMコンテキストをリセットし、システムプロンプトを適用
reset_llm_context()
try:
while True:
# 2. 音声認識を実行
user_prompt = transcribe_audio()
print("--- ユーザーの質問 (STT) ---")
print(user_prompt if user_prompt else "音声が認識されませんでした。")
print("---------------------------\n")
if not user_prompt:
print("音声が認識されませんでした。再度お話しください。")
continue
# 終了キーワードのチェック
if user_prompt in ["終了", "さようなら", "おしまい"]:
print("会話を終了します。")
break
# 3. 認識したテキストをLLMに入力し、ポーリング & 再生
stream_and_speak(user_prompt)
except KeyboardInterrupt:
print("\nユーザーにより処理が中断されました。")
finally:
print("システムのシャットダウン処理中...")
# --- 4. 再生ワーカーの停止 ---
SHOULD_STOP_WORKER.set()
worker_thread.join()
# 最終的なクリーンアップ
if os.path.exists(WAVE_RECORD_FILENAME):
os.remove(WAVE_RECORD_FILENAME)
if os.path.exists(TEMP_FILE_PATH) and not os.listdir(TEMP_FILE_PATH):
os.rmdir(TEMP_FILE_PATH)
print("全体の処理が完了しました。")
(env) $ python speakllm.py
5秒間の録音が始まったらマイクに向かって喋ってみてください。
🎙️ Recording for 5 seconds. Speak now...
ずんだもんの声で返答がスピーカーから聞こえてきたら成功です。
おつかれさまでした。
備考
メモリ消費について
ざっくりhtopで計測した結果
| システム | 動作 | メモリ消費量 |
|---|---|---|
| LLM8850 | 起動（モデル転送）時 | 1.7GB |
| 起動後 | 0.9GB | |
| VOICEVOX | 音声リクエスト時 | 0.2GB〜1GB以上 ※テキスト量に依存 |
| Faster Whisper | 初回起動後 | 0.6GB |
トータルのメモリ使用量(OS使用含む)は、LLMモデル転送時に2GBちょっと、会話中は2GB前後というところでした。
Raspberry Pi 5の2GBモデルでは足りず、4GBモデルでなら動くかもしれない、というところです。
読み上げ速度について
本当はもう少し早く喋らせたいところですが、Qwen3-4Bモデルの生成速度が3.5〜4tokens/sec程度のため、若干読み上げに追いついていませんでした。
実行スクリプトで読み上げ速度を0.9にしてちょうどよい速度バランスになりました。
継続会話について
おしゃべりといいつつ一方的なユーザー発話→LLM解答の繰り返しになってしまいました。
これはLLM解答が終わる度にリセットリクエストを送っているためです。
LLM8850側でトークンキャッシュがあるらしいのですが、やたら小さいのかすぐオーバーフローしてしまっていました。
何かオプションで増やせるかもしれません。
[E][SetKVCache][ 627]: precompute_len(388) + input_num_token(258) > _attr.prefill_max_kv_cache_num_grp[3]