📖

Style-Bert-VITS2を使ったDiscord音声ボットの作り方

2025/03/04に公開

Style-Bert-VITS2を使ったDiscord音声ボットの作り方

こんにちは!今回は、最新の音声合成技術「Style-Bert-VITS2」を使って、テキストを自然な音声に変換し、Discordのボイスチャンネルで再生できるボットの作り方を紹介します。

はじめに

AIによる音声合成技術は急速に進化しており、Style-Bert-VITS2はその中でも高品質な日本語音声を生成できるモデルとして注目されています。このモデルを使ったDiscordボットを作れば、テキストチャットだけでなく、ボイスチャンネルでもコミュニケーションを取ることができるようになります。

本記事では、以下の内容を解説します:

  • Style-Bert-VITS2 APIサーバーの構築
  • Discordボットの実装と接続
  • 複数の音声モデルの管理方法
  • ユーザーごとの音声設定の保存方法

システム構成

このプロジェクトは主に2つのコンポーネントから構成されています:

  1. Style-Bert-VITS2 APIサーバー:テキストを音声に変換するAPIを提供
  2. Discordボット:ユーザーからのメッセージを受け取り、APIを呼び出して音声を再生

システム構成図

必要条件

  • Python 3.9以上
  • FFmpeg(音声処理用)
  • Style-Bert-VITS2の依存ライブラリ
  • Discordアカウントとボットトークン

1. Style-Bert-VITS2 APIサーバーの実装

まず、Style-Bert-VITS2を使ったAPIサーバーを実装します。このサーバーは、テキストを受け取り、音声データを返すエンドポイントを提供します。

import os
import numpy as np
from pathlib import Path
from style_bert_vits2.nlp import bert_models
from style_bert_vits2.constants import Languages
from style_bert_vits2.tts_model import TTSModel
from style_bert_vits2.logging import logger
from style_bert_vits2.nlp.japanese.user_dict import update_dict
import torch
from pydantic import BaseModel

from fastapi import FastAPI, Depends, Header
from typing import List, Dict, Any
from fastapi.security.api_key import APIKeyHeader
import uvicorn
import json
import time
import glob

# デバイスの設定(GPUがあれば使用)
device = "cuda" if torch.cuda.is_available() else "cpu"

# SBV2クラスの実装
class SBV2:
    def __init__(self, model_path):
        logger.remove()

        # デバイスの設定
        if device == "auto":
            self.DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
        else:
            self.DEVICE = device

        # BERTモデルの読み込み
        bert_models.load_model(Languages.JP, "ku-nlp/deberta-v2-large-japanese-char-wwm")
        bert_models.load_tokenizer(Languages.JP, "ku-nlp/deberta-v2-large-japanese-char-wwm")

        # モデルファイルの検索
        style_file = glob.glob(f'{model_path}/*.npy',recursive=True)[0]
        config_file = glob.glob(f'{model_path}/*.json',recursive=True)[0]
        model_file = glob.glob(f'{model_path}/*.safetensors',recursive=True)[0]

        print(style_file)
        print(config_file)
        print(model_file)

        # TTSモデルの初期化
        self.model_TTS = TTSModel(
            model_path=model_file,
            config_path=config_file,
            style_vec_path=style_file,
            device=self.DEVICE
        )

    def call_TTS(self, message):
        sr, audio = self.model_TTS.infer(text=message)
        return sr, audio

# FastAPIアプリケーションの設定
app = FastAPI()

# 入力データのモデル
class SBV2_inputs(BaseModel):
    text: str

class SBV2_init(BaseModel):
    modelname: str

# ユーザごとのインスタンスを管理する辞書
user_instances: Dict[str, Dict] = {}

# 依存関係の管理
class Dependencies:
    def __init__(self, api_key, model):
        model_path = f"model_assets/{model}"
        self.sbv2 = SBV2(model_path=model_path)

    def get_sbv2(self):   
        return self.sbv2

# APIエンドポイントの実装
@app.post("/initialize/")
async def initialize(
    inputs: SBV2_init,
    api_key: str = Header(None, alias="api_key")
):
    # ユーザーごとのインスタンスを初期化
    if api_key not in user_instances:
        user_instances[api_key] = Dependencies(api_key, inputs.modelname)
    
    # 初回実行のウォームアップ
    _, _ = user_instances[api_key].get_sbv2().call_TTS("初期化")
    return {"message": "Initialized"}

@app.post("/process/")
async def process_data(
    inputs: SBV2_inputs,
    api_key: str = Header(None, alias="api_key")
):
    # APIキーに対応するインスタンスを取得
    if api_key not in user_instances:
        return {"error": "Not initialized"}
    
    # テキストから音声を生成
    start_tts = time.time()
    sr, audio = user_instances[api_key].get_sbv2().call_TTS(inputs.text)
    print(f"Time taken for TTS: {time.time() - start_tts}")
    
    # 音声データをJSON形式で返す
    return {"audio": audio.tolist(), "sr": sr}

# サーバー起動
if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

このAPIサーバーの特徴:

  • 複数モデル対応:異なるAPIキーごとに異なるモデルを使用可能
  • 効率的なリソース管理:ユーザーごとにモデルインスタンスを管理
  • シンプルなAPI:初期化とテキスト処理の2つのエンドポイントのみ

2. Discordボットの実装

次に、Discordボットを実装します。このボットは、ユーザーからのメッセージを受け取り、APIサーバーを呼び出して音声を生成し、Discordのボイスチャンネルで再生します。

import discord
from discord.ext import commands
import requests
import numpy as np
import io
import asyncio
import json
import logging
import os
import glob
from pathlib import Path
from scipy.io.wavfile import write as write_wav
import aiohttp

# 設定
DISCORD_TOKEN = os.getenv('DISCORD_TOKEN')
SBV2_API_URL = os.getenv('SBV2_API_URL', 'http://127.0.0.1:8001')
DEFAULT_API_KEY = os.getenv('SBV2_API_KEY', 'sbv2_amitaro')
COMMAND_PREFIX = os.getenv('COMMAND_PREFIX', '!')

# ボットの初期化
intents = discord.Intents.default()
intents.message_content = True
bot = commands.Bot(command_prefix=COMMAND_PREFIX, intents=intents)

# ユーザー設定の保存用辞書
user_voice_settings = {}
available_models = {}

# モデル読み込み関数
def load_available_models():
    """model_assetsディレクトリから利用可能なモデルを読み込む"""
    global available_models
    available_models = {}
    
    model_assets_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "model_assets")
    
    # ディレクトリ内のモデルを検索
    model_dirs = [d for d in os.listdir(model_assets_dir)
                 if os.path.isdir(os.path.join(model_assets_dir, d)) and not d.startswith('.')]
    
    for model_dir in model_dirs:
        model_path = os.path.join(model_assets_dir, model_dir)
        
        # 必要なファイルが揃っているか確認
        has_safetensors = len(glob.glob(os.path.join(model_path, "*.safetensors"))) > 0
        has_config = len(glob.glob(os.path.join(model_path, "*.json"))) > 0
        has_style_vectors = len(glob.glob(os.path.join(model_path, "*.npy"))) > 0
        
        if has_safetensors and has_config and has_style_vectors:
            available_models[model_dir] = {
                "name": model_dir,
                "api_key": DEFAULT_API_KEY,
                "description": f"{model_dir}音声モデル"
            }
    
    return available_models

# API呼び出し関数
async def call_tts_api(text, api_key=DEFAULT_API_KEY, model_name=None):
    """テキストを音声に変換するAPIを呼び出す"""
    try:
        # モデル初期化(必要な場合)
        if model_name:
            init_url = f"{SBV2_API_URL}/initialize/"
            headers = {"api_key": api_key}
            init_inputs = {"modelname": model_name}
            
            async with bot.session.post(init_url, json=init_inputs, headers=headers) as response:
                if response.status != 200:
                    return None, None

        # テキスト処理API呼び出し
        url = f"{SBV2_API_URL}/process/"
        headers = {"api_key": api_key}
        inputs = {"text": text}
        
        async with bot.session.post(url, json=inputs, headers=headers) as response:
            if response.status == 200:
                data = await response.json()
                audio = np.array(data['audio'], dtype=np.float32)
                sr = data['sr']
                return audio, sr
            else:
                return None, None
    except Exception as e:
        print(f"Error calling TTS API: {e}")
        return None, None

# 音声ファイル作成関数
async def create_audio_file(audio, sr):
    """音声データをファイルに変換"""
    if audio is None or sr is None:
        return None
    
    try:
        # 音声の正規化
        audio = audio / np.max(np.abs(audio)) * 0.9
        audio_int16 = (audio * 32767).astype(np.int16)
        
        # メモリ上にファイルを作成
        buffer = io.BytesIO()
        write_wav(buffer, sr, audio_int16)
        buffer.seek(0)
        
        return buffer
    except Exception as e:
        print(f"Error creating audio file: {e}")
        return None

# ボットイベント
@bot.event
async def on_ready():
    """ボットが起動したときの処理"""
    print(f'Bot connected as {bot.user}')
    bot.session = aiohttp.ClientSession()
    
    # 利用可能なモデルを読み込み
    load_available_models()
    
    # すべてのモデルを初期化
    for model_name, model_info in available_models.items():
        await initialize_sbv2_model(model_info["api_key"], model_name)

# メッセージ受信イベント
@bot.event
async def on_message(message):
    """メッセージを受信したときの処理"""
    # ボット自身のメッセージは無視
    if message.author == bot.user:
        return
    
    # DMの場合の処理
    if isinstance(message.channel, discord.DMChannel):
        # ユーザー設定を取得
        user_id = str(message.author.id)
        user_settings = user_voice_settings.get(user_id, {})
        model_name = user_settings.get('model_name', None)
        
        # モデルが設定されていない場合は最初のモデルを使用
        if model_name is None and available_models:
            model_name = next(iter(available_models.keys()))
        
        # ユーザーがボイスチャンネルに接続しているか確認
        for guild in bot.guilds:
            member = guild.get_member(message.author.id)
            if member and member.voice and member.voice.channel:
                voice_channel = member.voice.channel
                voice_client = guild.voice_client
                
                # ボイスチャンネルに接続
                if voice_client is None:
                    voice_client = await voice_channel.connect()
                elif voice_client.channel != voice_channel:
                    await voice_client.move_to(voice_channel)
                
                # 音声生成と再生
                async with message.channel.typing():
                    await message.channel.send(f"「{message.content}」の音声を生成中...")
                    
                    audio, sr = await call_tts_api(message.content, DEFAULT_API_KEY, model_name)
                    if audio is None:
                        await message.channel.send("音声の生成に失敗しました。")
                        return
                    
                    audio_file = await create_audio_file(audio, sr)
                    if audio_file is None:
                        await message.channel.send("音声の処理に失敗しました。")
                        return
                    
                    await message.channel.send("音声を再生します...")
                    await play_audio(voice_client, audio_file)
                break
    
    # コマンド処理
    await bot.process_commands(message)

# コマンド定義
@bot.command(name='join')
async def join(ctx):
    """ボイスチャンネルに参加"""
    voice_client = await join_voice_channel(ctx)
    if voice_client:
        await ctx.author.send(f"ボイスチャンネルに参加しました")

@bot.command(name='leave')
async def leave(ctx):
    """ボイスチャンネルから退出"""
    if ctx.guild and ctx.voice_client:
        await ctx.voice_client.disconnect()
        await ctx.author.send("ボイスチャンネルから退出しました")

@bot.command(name='speak')
async def speak(ctx, *, text: str):
    """テキストを音声に変換して再生"""
    # ボイスチャンネルに接続
    voice_client = await join_voice_channel(ctx)
    if not voice_client:
        return
    
    # ユーザー設定を取得
    user_id = str(ctx.author.id)
    user_settings = user_voice_settings.get(user_id, {})
    model_name = user_settings.get('model_name', None)
    
    # 音声生成と再生
    async with ctx.author.typing():
        await ctx.author.send(f"「{text}」の音声を生成中...")
        
        audio, sr = await call_tts_api(text, DEFAULT_API_KEY, model_name)
        if audio is None:
            await ctx.author.send("音声の生成に失敗しました。")
            return
        
        audio_file = await create_audio_file(audio, sr)
        if audio_file is None:
            await ctx.author.send("音声の処理に失敗しました。")
            return
        
        await ctx.author.send("音声を再生します...")
        await play_audio(voice_client, audio_file)

@bot.command(name='voice')
async def set_voice(ctx, voice_name: str):
    """使用する音声モデルを設定"""
    voice_name_lower = voice_name.lower()
    
    # モデル名のマッピングを作成
    model_mapping = {}
    for model_name, model_info in available_models.items():
        short_name = model_name.lower().replace("_", "")
        model_mapping[short_name] = {
            "name": model_name,
            "api_key": model_info["api_key"]
        }
        model_mapping[model_name.lower()] = {
            "name": model_name,
            "api_key": model_info["api_key"]
        }
    
    # 指定された音声が見つからない場合
    if voice_name_lower not in model_mapping:
        available_voices = list(set([k for k in model_mapping.keys() if "_" not in k]))
        await ctx.author.send(f"音声が見つかりません。利用可能な音声: {', '.join(available_voices)}")
        return
    
    # ユーザー設定を保存
    user_id = str(ctx.author.id)
    selected_model = model_mapping[voice_name_lower]
    
    if user_id not in user_voice_settings:
        user_voice_settings[user_id] = {}
    user_voice_settings[user_id]['api_key'] = selected_model["api_key"]
    user_voice_settings[user_id]['model_name'] = selected_model["name"]
    
    await ctx.author.send(f"音声が「{selected_model['name']}」に設定されました。")

@bot.command(name='list_voices')
async def list_voices(ctx):
    """利用可能な音声モデルを表示"""
    if not available_models:
        load_available_models()
    
    if not available_models:
        await ctx.author.send("利用可能な音声モデルが見つかりません。")
        return
    
    # ユーザーの現在の設定を取得
    user_id = str(ctx.author.id)
    current_model = user_voice_settings.get(user_id, {}).get('model_name', "未設定")
    
    message = "利用可能な音声モデル:\n"
    
    # 各モデルの情報を表示
    for model_name, model_info in available_models.items():
        short_name = model_name.lower().replace("_", "")
        current_mark = "✓ " if model_name == current_model else ""
        message += f"• {current_mark}**{short_name}**: {model_info['description']}\n"
    
    message += "\n音声を設定するには `!voice 音声名` と入力してください。"
    
    await ctx.author.send(message)

# ボット起動
if __name__ == "__main__":
    from dotenv import load_dotenv
    load_dotenv()
    bot.run(DISCORD_TOKEN)

3. 設定と実行方法

環境変数の設定

.envファイルをプロジェクトのルートディレクトリに作成し、以下の内容を設定します:

DISCORD_TOKEN=あなたのDiscordボットトークン
SBV2_API_URL=http://127.0.0.1:8001
SBV2_API_KEY=sbv2_amitaro
COMMAND_PREFIX=!

モデルファイルの配置

音声モデルファイルをmodel_assets/モデル名/ディレクトリに配置します。必要なファイルは以下の通りです:

  • *.safetensors(モデルファイル)
  • *.json(設定ファイル)
  • *.npy(スタイルベクトル)

実行手順

  1. APIサーバーを起動します:
python sbv2_api.py
  1. Discordボットを起動します:
python discord_sbv2_bot.py

4. ボットの使い方

基本的な使い方

  • DMでの使用:ボットにDMでメッセージを送信すると、そのテキストが自動的に音声に変換され、あなたが参加しているボイスチャンネルで再生されます。

コマンド一覧

  • !join - ボットをボイスチャンネルに参加させる
  • !leave - ボットをボイスチャンネルから退出させる
  • !speak テキスト - テキストを音声に変換して再生する
  • !voice 音声名 - 使用する音声モデルを設定する(例:!voice notanimejpmanyspeaker
  • !list_voices - 利用可能な音声モデルを表示する

5. 拡張と応用例

このDiscordボットは、以下のような用途に応用できます:

  1. オンラインゲームでのボイスチャット:キーボード入力だけで音声チャットに参加できる
  2. 言語学習ツール:テキストの発音を確認するための補助ツール
  3. アクセシビリティ向上:声を出せない状況でもボイスチャットに参加可能
  4. ストリーミング配信:配信者の声の代わりにAI音声を使用

まとめ

Style-Bert-VITS2を使ったDiscord音声ボットを作ることで、テキストを自然な音声に変換し、Discordのボイスチャンネルでコミュニケーションを取ることができるようになりました。このプロジェクトは、音声合成技術とDiscord APIを組み合わせた実用的な例であり、さまざまな用途に応用できます。

ぜひ自分だけの音声ボットを作って、新しいコミュニケーション体験を楽しんでみてください!

参考リンク

Discussion