注釈

こんにちは。SunFounder Raspberry Pi & Arduino & ESP32 Facebook 愛好家コミュニティへようこそ！ Raspberry Pi、Arduino、ESP32 について、仲間の愛好家と一緒にさらに深く探求しましょう。

参加する理由

  • 専門的なサポート ：コミュニティメンバーや公式チームの支援を受けて、購入後の問題や技術的な課題を解決できます。

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

  • 限定先行情報 ：新製品の発表やプレビュー情報をいち早く入手できます。

  • 特別割引 ：最新製品を対象とした限定割引をお楽しみいただけます。

  • 季節限定プロモーションとプレゼント企画 ：プレゼントキャンペーンや祝日限定のプロモーションに参加できます。

👉 私たちと一緒に探求し、創造する準備はできましたか？ [here] をクリックして、今すぐ参加しましょう！

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

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

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

  1. 聞く — マイクが音声を収音し、 Vosk で文字起こしします。

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

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

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

始める前に

以下が準備できていることを確認してください：

コードを実行する

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

    cd ~/fusion-hat/examples/
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 を使用します。Ollama が LAN 内の別のコンピュータで動作している場合は、Ollama で Expose to network を有効にし、ip をそのコンピュータの LAN IP に設定します。

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

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

    cd ~/fusion-hat/examples/
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 つのサブシステムを取り込みます：音声→テキスト（ STT ）の Vosk 、LLM の Ollama 、テキスト→音声（ TTS ）の 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 をテストする でテスト済みのモデルを選んでください。低品質（ low ）の音声は高速で、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 が最終回答 のみ を読み上げるようにします。

ヒント： 生トークンをストリーミング表示しているため画面上に他の断片が見えても、この関数により 音声 出力はクリーンな状態に保たれます。

メインループ：一度挨拶し、その後は 聞く → 考える → 話す

print(WELCOME)
tts.say(WELCOME)

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

聞く（ライブ部分結果付きの 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 にすると、即時フィードバック用の 部分 文字起こしと、発話終了時の 最終 文字起こしが得られます。

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

ガード： 何も認識されなかった場合は、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 に蓄積します。

注： 生トークンを表示したくない場合は 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 で停止できます。ボットは短い別れのメッセージを話し、正常終了したことを知らせます。

トラブルシューティングと FAQ

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

    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.」