注釈

こんにちは、SunFounder Raspberry Pi & Arduino & ESP32 Enthusiast Community on Facebookへようこそ！他の愛好家と一緒に、Raspberry Pi、Arduino、ESP32の世界により深く入り込みましょう。

参加する理由

  • 専門家サポート: 購入後の問題や技術的な課題を、コミュニティと私たちのチームの助けを借りて解決します。

  • 学習と共有: ヒントやチュートリアルを交換して、スキルを向上させましょう。

  • 限定プレビュー: 新製品の発表や先行プレビューに早期アクセスできます。

  • 特別割引: 最新製品を特別割引でお楽しみいただけます。

  • 季節限定キャンペーンとプレゼント: プレゼント企画やホリデーキャンペーンに参加しましょう。

👉 一緒に発見し、創造する準備はできましたか？ [こちら] をクリックして、今すぐ参加しましょう！

6. ローカル音声チャットボット

このレッスンでは、これまで学んだ 音声認識（STT）、 音声合成（TTS）、そして ローカルLLM（Ollama） を組み合わせて、 Fusion HAT+ 上で完全オフラインで動作する 音声チャットボット を構築します。

ワークフローはシンプルです：

  1. 聞く — マイクがあなたの音声を取得し、 Vosk で文字起こしします。

  2. 考える — テキストは Ollama 上で動作するローカル LLM （例： llama3.2:3b ）に送信されます。

  3. 話す — チャットボットが Piper TTS を使って音声で応答します。

これにより、リアルタイムで理解し応答できる ハンズフリーの会話型ロボット を実現できます。

開始前に

以下の準備が完了していることを確認してください：

サンプルの実行

  1. サンプルスクリプトを開きます：

    cd ~/ai-lab-kit/llm/
sudo nano local_voice_chatbot.py

  2. 必要に応じてパラメータを更新します：

    • stt = Vosk(language="en-us") : 使用するアクセントや言語パックに合わせて変更します（例： en-us 、 zh-cn 、 es ）。

    • tts.set_model("en_US-amy-low") : 1. Piper のテスト で確認した Piper の音声モデルに置き換えます。

    • llm = Ollama(ip="localhost", model="llama3.2:3b") : ip と model の両方を自分の環境に合わせて更新します。

      • ip: Ollama を 同じ Pi 上で実行している場合は localhost を使います。LAN 内の別のコンピュータで Ollama を実行している場合は、Ollama で Expose to network を有効にし、そのコンピュータの LAN IP を ip に設定します。

      • model: Ollama でダウンロードまたは有効化したモデル名と完全に一致している必要があります。

  3. スクリプトを実行します：

    cd ~/ai-lab-kit/llm/
sudo python3 local_voice_chatbot.py

  4. 実行後、次のような動作になります：

    • ボットが音声付きのウェルカムメッセージで挨拶します。

    • 音声入力を待機します。

    • Vosk が音声をテキストに変換します。

    • テキストは Ollama に送信され、応答がストリーミングで返されます。

    • 応答は整形され（内部的な思考出力を除去）、Piper によって音声で読み上げられます。

    • Ctrl+C でいつでもプログラムを停止できます。

コード

import re
import time
from fusion_hat.llm import Ollama
from fusion_hat.stt import Vosk
from fusion_hat.tts import Piper

# Initialize speech recognition
stt = Vosk(language="en-us")

# Initialize TTS
tts = Piper()
tts.set_model("en_US-amy-low")

# Instructions for the LLM
INSTRUCTIONS = (
    "You are a helpful assistant. Answer directly in plain English. "
    "Do NOT include any hidden thinking, analysis, or tags like <think>."
)
WELCOME = "Hello! I'm your voice chatbot. Speak when you're ready."

# Initialize Ollama connection
llm = Ollama(ip="localhost", model="llama3.2:3b")
llm.set_max_messages(20)
llm.set_instructions(INSTRUCTIONS)

# Utility: clean hidden reasoning
def strip_thinking(text: str) -> str:
    if not text:
        return ""
    text = re.sub(r"<\s*think[^>]*>.*?<\s*/\s*think\s*>", "", text, flags=re.DOTALL|re.IGNORECASE)
    text = re.sub(r"<\s*thinking[^>]*>.*?<\s*/\s*thinking\s*>", "", text, flags=re.DOTALL|re.IGNORECASE)
    text = re.sub(r"```(?:\s*thinking)?\s*.*?```", "", text, flags=re.DOTALL|re.IGNORECASE)
    text = re.sub(r"\[/?thinking\]", "", text, flags=re.IGNORECASE)
    return re.sub(r"\s+\n", "\n", text).strip()

def main():
    print(WELCOME)
    tts.say(WELCOME)

    try:
        while True:
            print("\n🎤 Listening... (Press Ctrl+C to stop)")

            # Collect final transcript from Vosk
            text = ""
            for result in stt.listen(stream=True):
                if result["done"]:
                    text = result["final"].strip()
                    print(f"[YOU] {text}")
                else:
                    print(f"[YOU] {result['partial']}", end="\r", flush=True)

            if not text:
                print("[INFO] Nothing recognized. Try again.")
                time.sleep(0.1)
                continue

            # Query Ollama with streaming
            reply_accum = ""
            response = llm.prompt(text, stream=True)
            for next_word in response:
                if next_word:
                    print(next_word, end="", flush=True)
                    reply_accum += next_word
            print("")

            # Clean and speak
            clean = strip_thinking(reply_accum)
            if clean:
                tts.say(clean)
            else:
                tts.say("Sorry, I didn't catch that.")

            time.sleep(0.05)

    except KeyboardInterrupt:
        print("\n[INFO] Stopping...")
    finally:
        tts.say("Goodbye!")
        print("Bye.")

if __name__ == "__main__":
    main()

コードの解説

インポートとグローバル設定

import re
import time
from fusion_hat.llm import Ollama
from fusion_hat.stt import Vosk
from fusion_hat.tts import Piper

ここでは、前のレッスンで構築した 3 つのサブシステムを読み込んでいます： 音声認識用の Vosk、LLM 用の Ollama、音声合成用の Piper です。

STT（Vosk）の初期化

stt = Vosk(language="en-us")

米国英語用の Vosk モデルを読み込みます。 認識精度を高めるため、言語コード（例： zh-cn 、 es ）は使用する音声パックに合わせて変更してください。

TTS（Piper）の初期化

tts = Piper()
tts.set_model("en_US-amy-low")

Piper エンジンを作成し、特定の音声モデルを選択します。 1. Piper のテスト で確認済みのモデルを使ってください。低品質の音声モデルは高速で、CPU 使用率も低くなります。

LLM への指示とウェルカムメッセージ

INSTRUCTIONS = (
    "You are a helpful assistant. Answer directly in plain English. "
    "Do NOT include any hidden thinking, analysis, or tags like <think>."
)
WELCOME = "Hello! I'm your voice chatbot. Speak when you're ready."

ここには UX 上の重要な 2 つの工夫があります：

  • 短く直接的な回答 にすること（TTS で聞き取りやすくなるため）。

  • 不要な出力を減らすため、内部的な “chain-of-thought” タグを明示的に禁止すること。

Ollama への接続と会話履歴の設定

llm = Ollama(ip="localhost", model="llama3.2:3b")
llm.set_max_messages(20)
llm.set_instructions(INSTRUCTIONS)

  • ip="localhost" は、Ollama サーバーが同じ Pi 上で動作している前提です。別の LAN 内マシンで動作している場合は、そのコンピュータの LAN IP を指定し、Ollama で Expose to network を有効にしてください。

  • set_max_messages(20) は短めの会話履歴を保持します。メモリや遅延が厳しい場合は、この値をさらに小さくしてください。

読み上げ前に内部思考 / タグを除去

def strip_thinking(text: str) -> str:
    if not text:
        return ""
    text = re.sub(r"<\s*think[^>]*>.*?<\s*/\s*think\s*>", "", text, flags=re.DOTALL|re.IGNORECASE)
    text = re.sub(r"<\s*thinking[^>]*>.*?<\s*/\s*thinking\s*>", "", text, flags=re.DOTALL|re.IGNORECASE)
    text = re.sub(r"```(?:\s*thinking)?\s*.*?```", "", text, flags=re.DOTALL|re.IGNORECASE)
    text = re.sub(r"\[/?thinking\]", "", text, flags=re.IGNORECASE)
    return re.sub(r"\s+\n", "\n", text).strip()

一部のモデルは <think>… のような内部用タグを出力することがあります。 この関数はそれらを削除し、TTS が 最終的な回答だけ を読み上げるようにします。

Tip: 画面上には生のトークンが表示される場合がありますが、この関数によって 音声出力 はクリーンに保たれます。

メインループ：最初に挨拶し、その後 listen → think → speak を繰り返す

print(WELCOME)
tts.say(WELCOME)

ターミナルとスピーカーの両方でユーザーに挨拶します。これは起動時に 1 回だけ実行されます。

聞く（ライブ partial 付きのストリーミング STT）

print("\n🎤 Listening... (Press Ctrl+C to stop)")

text = ""
for result in stt.listen(stream=True):
    if result["done"]:
        text = result["final"].strip()
        print(f"[YOU] {text}")
    else:
        print(f"[YOU] {result['partial']}", end="\r", flush=True)

  • stream=True により、すぐに確認できる partial 文字起こしと、発話終了時の final 文字起こしが得られます。

  • 最終的に認識されたテキストは text に保存され、一度だけ表示されます。

Guard: 何も認識されなかった場合は、LLM 呼び出しをスキップします：

if not text:
    print("[INFO] Nothing recognized. Try again.")
    time.sleep(0.1)
    continue

これにより、空のプロンプトをモデルへ送信することを防ぎ、時間とトークンを節約できます。

考える（LLM）＋ ストリーミング表示

reply_accum = ""
response = llm.prompt(text, stream=True)
for next_word in response:
    if next_word:
        print(next_word, end="", flush=True)
        reply_accum += next_word
print("")

  • 最終的な文字起こし結果をローカル LLM に送信し、 トークン到着ごとに表示 することで低遅延の応答を実現します。

  • 同時に、後処理のために reply_accum に完全な応答を蓄積します。

Note: 生のトークンを表示したくない場合は、 stream=False にして最終文字列だけを表示してください。

話す（先に整形し、その後 TTS を 1 回だけ実行）

clean = strip_thinking(reply_accum)
if clean:
    tts.say(clean)
else:
    tts.say("Sorry, I didn't catch that.")

  • 最終テキストから不要なタグを除去した後、 1 回だけ 読み上げます。

  • TTS を 1 回にまとめることで、「[LLM] / [SAY]」のような重複した出力を避けられます。

終了処理とクリーンアップ

except KeyboardInterrupt:
    print("\n[INFO] Stopping...")
finally:
    tts.say("Goodbye!")
    print("Bye.")

Ctrl+C で停止できます。終了時に短い別れの言葉を発声し、正常終了したことを示します。

トラブルシューティング

  • モデルが大きすぎる（メモリエラー）

    moondream:1.8b のような小さなモデルを使用するか、より高性能なコンピュータ上で Ollama を実行してください。

  • Ollama から応答が返らない

    Ollama が起動していることを確認してください（ ollama serve またはデスクトップアプリ）。リモート接続の場合は Expose to network を有効にし、IP アドレスを確認してください。

  • Vosk が音声を認識しない

    マイクが正常に動作していることを確認してください。必要に応じて別の言語パック（ zh-cn 、 es など）も試してください。

  • Piper が無音、またはエラーになる

    選択した音声モデルがダウンロード済みであり、1. Piper のテスト で正常にテストされていることを確認してください。

  • 回答が長すぎる、または話がそれる

    INSTRUCTIONS に “Keep answers short and to the point.” を追加してください。