AITuber開発日誌 6 声・頭脳・身体をひとつなぎに！

前回「声」のテストはしましたが、使いやすくするための「クラス化（部品化）」がまだでした。 Cursorの Composer (Ctrl + I) を開き、以下の指示でファイルを作成してくだい。
Composerへの指示（以下のプロンプトをコピペ）:

modules/expression/voicevox.py に、VOICEVOXを操作する VoicevoxClient クラスを作成してください。
【要件】
1. requests ライブラリを使用する（非同期ではなく同期でOK）。
2. config/settings.py から VOICEVOX_URL と VOICEVOX_SPEAKER_ID を読み込む。
3. speak(text) メソッドを実装する:
   - 音声合成を行い、wavデータを取得する。
   - simpleaudio ライブラリを使ってその場で再生する。
   - 再生が完了するまで待機する (play_obj.wait_done())。

いよいよ全ての指揮を執る「本体」を作ります。 ルートディレクトリに main.py を作成し、以下のコードを貼り付けてください。
このコードは、「入力を受け取る → AIが考える → 声を出す＆口を動かす」 という無限ループを行います。

コピペするコード

import asyncio
import random
from config.settings import settings
from modules.cognition.llm_client import LLMClient
from modules.expression.vtube_studio import VTubeStudioClient
from modules.expression.voicevox import VoicevoxClient

# キャラクター設定
SYSTEM_PROMPT = """
あなたは「AITuber」のアイです。
元気で明るい、少しおっちょこちょいな性格です。
短くテンポよく返事をしてください。
"""

async def main():
    print("🚀 AITuberシステムを起動します...")

    # 1. 各モジュールの初期化
    brain = LLMClient(system_prompt=SYSTEM_PROMPT)
    voice = VoicevoxClient()
    body = VTubeStudioClient()

    # 2. VTube Studioに接続
    if not await body.connect():
        print("❌ VTube Studioへの接続に失敗しました。")
        return

    # 会話履歴
    history = []

    print("✅ 準備完了！何か話しかけてね（'exit'で終了）")

    try:
        while True:
            # --- ユーザー入力 ---
            user_input = input("\nあなた: ")
            if user_input.lower() in ["exit", "quit", "終了"]:
                break

            # --- 思考 (Brain) ---
            print("🤖 (思考中...)")
            response_text = await brain.generate_response(user_input, history)
            print(f"アイ: {response_text}")

            # 履歴更新
            history.append({"role": "user", "content": user_input})
            history.append({"role": "assistant", "content": response_text})

            # --- 並列処理: 声を出しながら口を動かす ---
            # ※本来は音声波形に合わせて口を動かしますが、
            #   簡易的に「喋っている間、ランダムに口をパクパクさせる」タスクを走らせます
            
            # 口パクのタスクを作成
            async def lip_sync_simulation():
                while True:
                    await body.set_parameter("MouthOpen", random.uniform(0.0, 1.0))
                    await asyncio.sleep(0.1)
            
            # 音声再生と口パクを同時に開始
            lip_task = asyncio.create_task(lip_sync_simulation())
            
            # 声を再生（再生が終わるまでここで待機します）
            # ※VoicevoxClientは同期処理で作ったため、簡易的に executor で実行
            await asyncio.to_thread(voice.speak, response_text)
            
            # 喋り終わったら口パク停止
            lip_task.cancel()
            await body.set_parameter("MouthOpen", 0.0) # 口を閉じる

    except KeyboardInterrupt:
        print("\nシステムを終了します")
    finally:
        await body.close()

if __name__ == "__main__":
    asyncio.run(main())

main.py は、あなたのPC内で待機している「VOICEVOX」と「VTube Studio」という別のアプリに、Pythonから命令（通信）を送ることで動作します。

そのため、これらが起動していないと、Pythonは「命令を送る相手がいない（接続エラー）」と判断して止まってしまいます。

実行前に、以下の「起動チェックリスト」を確認してください。

以下のコマンドで実行してください。

python main.py

これが動けば…

1. あなたがターミナルで話しかける。

2. アイちゃん（AI）が返事を考える。

3. VTube Studioのキャラクターが、喋りながら（声が出て）口をパクパク動かす！

まさにAITuberの誕生です。 動きましたか？感動の瞬間を教えてください！

import asyncio
import random
# 設定ファイルの読み込み
from config.settings import settings
# 作成した各モジュールをインポート
from modules.cognition.llm_client import LLMClient
from modules.expression.vtube_studio import VTubeStudioClient
from modules.expression.voicevox import VoicevoxClient

# キャラクター設定
SYSTEM_PROMPT = """
あなたは「AITuber」のアイです。
元気で明るい、少しおっちょこちょいな性格です。
短くテンポよく返事をしてください。
"""

async def main():
    print("🚀 AITuberシステムを起動します...")

    # 1. 各モジュールの初期化
    # 脳（AI）、声（VOICEVOX）、体（VTube Studio）のインスタンスを作成
    brain = LLMClient(system_prompt=SYSTEM_PROMPT)
    voice = VoicevoxClient()
    body = VTubeStudioClient()

    # 2. VTube Studioに接続
    if not await body.connect():
        print("❌ VTube Studioへの接続に失敗しました。")
        return

    # 会話履歴
    history = []

    print("✅ 準備完了！何か話しかけてね（'exit'で終了）")

    try:
        while True:
            # --- ユーザー入力 ---
            user_input = input("\nあなた: ")
            if user_input.lower() in ["exit", "quit", "終了"]:
                break

            # --- 思考 (Brain) ---
            print("🤖 (思考中...)")
            # AIに返答を生成させる
            response_text = await brain.generate_response(user_input, history)
            print(f"アイ: {response_text}")

            # 履歴更新
            history.append({"role": "user", "content": user_input})
            history.append({"role": "assistant", "content": response_text})

            # --- 並列処理: 声を出しながら口を動かす ---
            
            # 口パクのタスクを作成（0.1秒ごとにランダムに開閉）
            async def lip_sync_simulation():
                while True:
                    await body.set_parameter("MouthOpen", random.uniform(0.0, 1.0))
                    await asyncio.sleep(0.1)
            
            # 音声再生と口パクを同時に開始する準備
            lip_task = asyncio.create_task(lip_sync_simulation())
            
            # 声を再生（再生が終わるまでここで待機）
            # voice.speakは同期処理なので、asyncio.to_threadで実行してメインループを止めないようにする
            await asyncio.to_thread(voice.speak, response_text)
            
            # 喋り終わったら口パク停止
            lip_task.cancel()
            await body.set_parameter("MouthOpen", 0.0) # 口を閉じる

    except KeyboardInterrupt:
        print("\nシステムを終了します")
    finally:
        # 終了時に接続を切る
        await body.close()

if __name__ == "__main__":
    asyncio.run(main())