AITuber開発日誌 7　コメントを読み取る「耳」の実装

import pytchat
from config.settings import settings
import logging

# ログを見やすくするための設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class YouTubeCommentFetcher:
    def __init__(self):
        # .env (config/settings.py経由) から動画IDを読み込む
        self.video_id = settings.YOUTUBE_VIDEO_ID
        self.chat = None
        
        # インスタンス作成時に接続を開始する
        self.connect()

    def connect(self):
        """YouTubeチャットに接続する"""
        if not self.video_id:
            print("⚠️ YouTube Video IDが設定されていません。.envを確認してください。")
            return

        try:
            print(f"📡 YouTubeチャットに接続中... (ID: {self.video_id})")
            # pytchatを使用して指定された動画IDのチャットに接続
            self.chat = pytchat.create(video_id=self.video_id)
            print("✅ 接続成功！コメントを監視します。")
        except Exception as e:
            print(f"❌ 接続エラー: {e}")

    def fetch_comments(self):
        """
        前回の取得以降に来た新しいコメントを取得してリストで返す
        戻り値: [{"user": "名前", "text": "コメント内容"}, ...]
        """
        # チャット接続が生きていない場合は空リストを返す
        if not self.chat or not self.chat.is_alive():
            # ※ここで再接続処理を入れることも可能ですが、簡易化のためスキップします
            return []

        comments = []
        try:
            # sync_items() を使うことで、前回の取得分以降の「新しいコメント」だけを取得できます
            for c in self.chat.get().sync_items():
                # コメントデータから必要な「投稿者名」と「メッセージ」を抽出
                comments.append({
                    "user": c.author.name,
                    "text": c.message
                })
        except Exception as e:
            print(f"⚠️ コメント取得中にエラーが発生しました: {e}")
        
        return comments

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
from modules.perception.youtube import YouTubeCommentFetcher  # 【追加】耳モジュール

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

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

    # 1. 各モジュールの初期化
    brain = LLMClient(system_prompt=SYSTEM_PROMPT)
    voice = VoicevoxClient()
    body = VTubeStudioClient()
    youtube = YouTubeCommentFetcher()  # 【追加】YouTube接続インスタンス

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

    # 会話履歴
    history = []

    print(f"✅ 準備完了！YouTube配信（ID: {settings.YOUTUBE_VIDEO_ID}）のコメントを待機中...")

    try:
        while True:
            # --- 【変更点】YouTubeコメントの取得 ---
            # 新しいコメントを取得（リスト形式）
            comments = youtube.fetch_comments()

            # コメントが来ていない場合は、少し待ってから再確認（CPU負荷軽減のため）
            if not comments:
                await asyncio.sleep(3)  # 3秒待機
                continue

            # コメントが来た場合、一番新しいものをピックアップして反応する
            # （※一度に大量に来た場合、すべてに反応すると遅延するため最新のみ採用）
            last_comment = comments[-1]
            user_name = last_comment["user"]
            user_text = last_comment["text"]

            print(f"\n💬 {user_name}: {user_text}")

            # "終了" コマンドでシステム停止（管理者用）
            if user_text == "終了":
                print("システムを終了します。")
                break

            # --- 思考 (Brain) ---
            print("🤖 (思考中...)")
            # AIへの入力プロンプトを作成（誰からのコメントかを明記）
            prompt_text = f"リスナーの「{user_name}」さんが「{user_text}」と言っています。返事をしてあげて。"
            
            response_text = await brain.generate_response(prompt_text, history)
            print(f"アイ: {response_text}")

            # 履歴更新（AIの記憶用）
            history.append({"role": "user", "content": f"{user_name}: {user_text}"})
            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())
            
            # 声を再生（再生完了まで待機）
            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())