🎧

ChainlitとVOICEVOXを使って、AIチャットのTTSをやってみる

2024/09/17に公開

はじめに

以前から作成しているChainlitとLangGraphを組み合わせた、AI AgentのWebアプリですが、今回はそこにVOICEVOXを組み込んで、AIチャットのTTSを実現してみました。

かなり前から試みようと思いつつ時間が取れなかったのですが、この前の夏休みにようやく試すことができたので、今回記事にしました。
(記事にするのに1カ月かかってしまいましたが・・・)

ChainlitやLangGraphについて

ChainlitやLangGraphについては、以前の記事で紹介していますので、気になる方は末尾にリンクを貼っておきますので、そちらをご覧ください。

VOICEVOXについて

VOICEVOXは無料で使える中品質なテキスト読み上げ・歌声合成ソフトです。

https://voicevox.hiroshiba.jp

無料で使えることからYoutubeなどでよく使われているようですので、
皆さんも一度くらいはこの中の声を聞いたことがあるかもしれません。

VOICEVOXはデスクトップアプリとして提供されていますが、
アプリを起動するとAPIサーバーが起動し、UIからだけでなくAPI経由でも音声合成ができるようになっています。

Dockerでイメージも提供されているので、今回はそちらを使用させていただきました。

今回の実装

VOICEVOXをdocker composeで起動

VOICEVOXをdocker composeで起動できるように、compose.yml, .devcontainer/compose-dev.ymlに以下のように記述しました。
音声合成にCPUのみを使う場合と、nvidiaのGPUを使う場合でイメージの指定と若干の設定が異なります。
ご自身の環境に合わせてコメントアウトを変更してください。

compose.yml
  # VOICEVOX Engineは環境に合わせて以下のいずれかを起動する(可能ならGPUを使用することを推奨)
  # CPUのみを使用する場合
  voicevox_engine-dev:
    image: voicevox/voicevox_engine:cpu-ubuntu20.04-latest
    ports:
      - "50021:50021"
    tty: true

  # Nvidia GPUを使用する場合
  # voicevox_engine-dev:
  #   image: voicevox/voicevox_engine:nvidia-ubuntu20.04-latest
  #   ports:
  #     - "50021:50021"
  #   tty: true
  #   deploy:
  #     resources:
  #       reservations:
  #         devices:
  #           - driver: nvidia
  #             count: 1
  #             capabilities: [gpu]

docker composeもしくは開発コンテナを起動すると、http://localhost:50021/でAPIサーバーが立ち上がります。
http://localhost:50021/docsでAPIの仕様を確認できます。

VOICEVOX用のサービスを作成

VOICEVOXのAPIを使うためのサービスを、src/service/voicevox.pyに作成しました。

src/service/voicevox.py
import requests
import base64
import json
import re
import os

from openai import OpenAI


class SpeakerData:
    """
    SpeakerDataクラス
    VOICEVOXのspeaker情報を取得するためのクラス
    """

    def __init__(self, domain: str = None):
        self.domain = domain or os.getenv(
            "VOICEVOX_API_DOMAIN", "http://voicevox_engine:50021/"
        )
        self.data = self._load_data()

    def _load_data(self) -> dict:
        """speakerの一覧をAPIから取得し、辞書形式で返す"""
        speakers_json = requests.get(url=f"{self.domain}speakers").json()
        return {
            item["name"]: {style["name"]: style["id"] for style in item["styles"]}
            for item in speakers_json
        }

    def get_all_speaker_and_style_list(self) -> list:
        """speakerとstyleの組み合わせに対するspeaker_idのリストを取得"""
        return [
            {
                f"{speaker}-{style}": self.data[speaker][style]
                for style in self.data[speaker]
            }
            for speaker in self.data
        ]

    def get_all_speaker_and_style_dict(self) -> dict:
        """speakerとstyleの組み合わせに対するspeaker_idの辞書を取得"""
        return {
            f"{speaker} - {style}": str(self.data[speaker][style])
            for speaker in self.data
            for style in self.data[speaker]
        }


class Voicevox:
    """
    Voicevoxクラス
    VOICEVOX APIを利用するためのクラス
    """

    def __init__(
        self,
        speaker_name: str = None,
        style_name: str = None,
        speaker_id: str = None,
        file_path: str = "./",
    ):
        self.domain = os.getenv("VOICEVOX_API_DOMAIN", "http://voicevox_engine:50021/")
        self.speaker_id = self._get_speaker_id(speaker_name, style_name, speaker_id)
        self.file_path = file_path

    def _get_speaker_id(
        self, speaker_name: str, style_name: str, speaker_id: str
    ) -> str:
        """speaker_idを取得するためのヘルパーメソッド"""
        if speaker_id:
            return speaker_id
        elif speaker_name and style_name:
            speakers = SpeakerData().data
            return speakers[speaker_name][style_name]
        else:
            raise ValueError(
                "speaker_id or speaker_name and style_name must be provided"
            )

    def _post_audio_query(self, text: str) -> json:
        """音声クエリをPOSTし、結果を返す"""
        response = requests.post(
            url=f"{self.domain}audio_query",
            params={"text": text, "speaker": self.speaker_id},
        )
        return response.json()

    def _post_synthesis(self, text: str) -> bytes:
        """音声合成を行い、結果のバイナリデータを返す"""
        query = self._post_audio_query(text)
        response = requests.post(
            url=f"{self.domain}synthesis",
            params={"speaker": self.speaker_id},
            json=query,
        )
        return response.content

    def post_synthesis_returned_in_base64(
        self, text: str, use_manuscript: bool = False
    ) -> str:
        """テキストを入力し、音声ファイルを生成してbase64形式で返す"""
        if use_manuscript:
            text = self._create_manuscript(text)
        audio_data = self._post_synthesis(text)
        return base64.b64encode(audio_data).decode()

    def post_synthesis_returned_in_file(
        self, text: str, use_manuscript: bool = False, file_name: str = "output"
    ) -> str:
        """テキストを入力し、音声ファイルを生成してファイルパスを返す"""
        if use_manuscript:
            text = self._create_manuscript(text)
        audio_data = self._post_synthesis(text)
        file_path = os.path.join(self.file_path, f"{file_name}.wav")
        with open(file_path, "wb") as file:
            file.write(audio_data)
        return file_path

    def _create_manuscript(self, text: str) -> str:
        """テキストを入力し、読み上げ原稿を生成して返す"""
        client = OpenAI()
        system_prompt = re.sub(
            r"\n\s*",
            "\n",
            """あなたは読み上げテキスト生成器です。ユーザーから提供された文章を、読み上げ原稿を返答してください。その際以下のルールに従ってください。
                - アルファベットの固有名詞はカタカナにし、漢字はひらがなにするなど、読み間違いをしないようにする
                - URLやコードブロックについては、そのまま読み上げるのではなく、画面を確認するように促すなど言い換えを行う
                入力文章:""",
        )
        completion = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": f"{text}\n\n読み上げ原稿\n:"},
            ],
        )
        return completion.choices[0].message.content


if __name__ == "__main__":
    voicevox = Voicevox(speaker_name="ずんだもん", style_name="ノーマル")
    text = "こんにちは"
    audio_file = voicevox.post_synthesis_returned_in_file(text, use_manuscript=True)
    print(audio_file)

SpeakerDataクラス

VOICEVOXのspeaker情報を取得するためのクラスです。
VOICEVOXでは、音声合成する際に、speaker_idを指定する必要があります。
speaker_idは、speaker名とstyle名の組み合わせに対応する値です。
例えば、ずんだもんノーマルスタイルのspeaker_idは3です。

このクラスは、APIからspeaker情報を取得し、speaker名とstyle名の組み合わせに対応するspeaker_idを、リストもしくは辞書形式で返します。

Voicevoxクラス

実際に音声合成を行うためのクラスです。

音声合成は2つのAPIを使います。
_post_audio_queryメソッドで音声クエリをPOSTし、結果を返し、
_post_synthesisメソッドで音声合成を行い、結果のバイナリデータを返します。

変換した音声データは、base64形式で返すpost_synthesis_returned_in_base64メソッドと、
ファイルに保存してファイルパスを返すpost_synthesis_returned_in_fileメソッドを用意して
それぞれの使い方に合わせて利用できるようにしました。

また、_create_manuscriptメソッドでは、入力されたテキストを読み上げ原稿に変換する処理を行っています。
方法としては、system_promptにルールを記述し、それに従ってgpt-4o-miniを使って言い換えを行っています。
これを用意した理由は、AIの生成した出力結果をそのまま読み上げると、URLやコードブロックなどがそのまま読み上げられてしまうため、
読み上げる際に適切な言い換えを行うことで、より自然な読み上げを実現するためです。

動作確認

if __name__ == "__main__":以下に、音声合成の動作確認用のコードを記述しています。
このコードを実行すると、./output.wavに音声ファイルが保存され、「こんにちは」というテキストが読み上げられます。

ChainlitにVOICEVOXを組み込む

実装したい機能

ChainlitにVOICEVOXを組み込む上で、実現したい機能は以下の通りです。

  • チャットを送信すると、AIが返答を生成し、VOICEVOXによる読み上げ音声が添付される
  • 音声は自動で再生される
  • 設定を開くと、キャラクターとスタイルが選択できる
  • 音声生成の有効無効を切り替えられる

実装

ChainlitにVOICEVOXを組み込むために、以下のように実装しました。
差分を知りたい方は、こちらのPRをご覧ください。

https://github.com/0msys/langgraph-chainlit-agent/pull/12

src/main.py
import chainlit as cl
import os

from services.chainlit_agent import ChainlitAgent


@cl.on_chat_start
async def on_chat_start():
    # フォルダを用意
    dir_path = f"./.files/{cl.user_session.get('id')}/"
    os.makedirs(dir_path, exist_ok=True)
    system_prompt = "あなたは最高のアシスタントチャットボットです。どんな依頼にも丁寧に最高のサービスを提供します。"
    chainlit_agent = ChainlitAgent(
        system_prompt=system_prompt,
        file_path=dir_path,
    )
    await chainlit_agent.on_chat_start()
    cl.user_session.set("chainlit_agent", chainlit_agent)


@cl.on_settings_update
async def on_settings_update(settings: dict):
    chainlit_agent = cl.user_session.get("chainlit_agent")
    await chainlit_agent.on_settings_update(settings)
    cl.user_session.set("chainlit_agent", chainlit_agent)


@cl.on_message
async def on_message(msg: cl.Message):

    inputs = cl.user_session.get("inputs", [])
    chainlit_agent = cl.user_session.get("chainlit_agent")

    output = await chainlit_agent.on_message(msg, inputs)
    inputs.append(output)
    cl.user_session.set("inputs", inputs)
src/services/chainlit_agent.py
import chainlit as cl
import pytz

from langchain_core.messages import HumanMessage, AIMessage
from datetime import datetime
from chainlit.input_widget import Select, Switch

from services.agent import SingleAgent
from services.voicevox import Voicevox
from services.voicevox import SpeakerData


class ChainlitAgent(SingleAgent):

    def __init__(
        self,
        system_prompt: str,
        speak: bool = False,
        speaker_name: str = "四国めたん",
        style_name: str = "ノーマル",
        file_path: str = "./",
    ):
        super().__init__(system_prompt=system_prompt)
        self.speak = speak
        self.file_path = file_path
        if speak:
            self.voicevox_service = Voicevox(
                speaker_name=speaker_name, style_name=style_name, file_path=file_path
            )

    async def on_chat_start(self):
        """
        チャットが開始されたときに呼び出される関数
        """
        # Settingsの初期値を設定
        settings = await cl.ChatSettings(
            [
                Switch(
                    id="Speak",
                    label="読み上げ",
                    initial=False,
                    description="読み上げを行うか選択してください。",
                ),
                Select(
                    id="Speaker_ID",
                    label="VOICEVOX - Speaker Name and Style",
                    items=SpeakerData().get_all_speaker_and_style_dict(),
                    initial_value="2",
                    description="読み上げに使用するキャラクターとスタイルを選択してください。",
                ),
            ]
        ).send()
        # Settingsの初期値を元に、設定を更新
        await self.on_settings_update(settings)

    async def on_settings_update(self, settings: dict):
        """
        Settingsが更新されたときに呼び出される関数
        """
        # Settingsの値を取得し、VOICEVOXの設定を更新
        self.speak = settings["Speak"]
        self.voicevox_service = Voicevox(
            speaker_id=settings["Speaker_ID"], file_path=self.file_path
        )

    async def on_message(self, msg: cl.Message, inputs: list):
        """
        メッセージが送信されたときに呼び出される関数
        """

        content = msg.content

        # 添付ファイルの情報を取得
        attachment_file_text = ""

        for element in msg.elements:
            attachment_file_text += f'- {element.name} (path: {element.path.replace("/workspace", ".")})\n'  # agentが参照するときは./files/***/***.pngのようになるので、それに合わせる

        if attachment_file_text:
            content += f"\n\n添付ファイル\n{attachment_file_text}"

        #  現在の日時を取得(JST)
        now = datetime.now(pytz.timezone("Asia/Tokyo")).strftime("%Y-%m-%d %H:%M:%S %Z")

        content += f"\n\n(入力日時: {now})"

        # ユーザーのメッセージを履歴に追加
        inputs.append(HumanMessage(content=content))

        res = cl.Message(content="")
        steps = {}
        async for output in self.astream_events(inputs):
            if output["kind"] == "on_chat_model_stream":
                await res.stream_token(output["content"])
            elif output["kind"] == "on_tool_start":
                async with cl.Step(name=output["tool_name"], type="tool") as step:
                    step.input = output["tool_input"]
                    steps[output["run_id"]] = step
            elif output["kind"] == "on_tool_end":
                step = steps[output["run_id"]]
                step.output = output["tool_output"]
                await step.update()

        await res.send()

        # 読み上げが有効な場合、読み上げファイルを生成し、メッセージに追加
        if self.speak:
            file_path = self.voicevox_service.post_synthesis_returned_in_file(
                text=res.content, use_manuscript=True, file_name="読み上げ"
            )
            elements = [
                cl.Audio(
                    name="読み上げ", path=file_path, display="inline", auto_play=True
                ),
            ]
            res.elements = elements
            await res.update()

        return AIMessage(content=res.content)
main.py

変更箇所は主に2か所です。

  • on_chat_startで読み上げ音声を保存するためのフォルダを作成
  • on_settings_updateでスピーカーやスタイル、読み上げの有効無効を設定
chainlit_agent.py

ChainlitAgentクラスにVOICEVOXを組み込むための処理を追加しました。
簡単に説明すると、以下のような変更を行っています。

  • __init__でVOICEVOXの初期化を行う
  • on_chat_startで読み上げ設定の初期値を設定
  • on_settings_updateで設定の更新処理を行う
  • on_messageで、読み上げが有効な場合、読み上げ音声を生成し、メッセージに追加

動作確認

Chainlitを起動し、チャットを送信すると、AIが返答を生成し、VOICEVOXによる読み上げ音声が添付されます。
音声は自動で再生され、設定を開くと、キャラクターとスタイルが選択でき、音声生成の有効無効を切り替えられます。
※gifなので音声は再生されませんが、雰囲気だけ。。。


動作イメージ

まとめ

今回は、ChainlitとVOICEVOXを組み合わせて、AIチャットのTTSを実現してみました。
TTSによりずっと画面を見なくても、AIの返答を聞くことができるようになり、より利便性が向上すると思います。
また、声がつくことで、AIに対してキャラ付けができるようになり、より愛着を持って使うことができるように思いました。
ただ、読み上げ原稿を用意していても、漢字の読み間違いなどがあるため、自然な読み上げを実現するためには、さらなる工夫が必要そうです。
さらに、GPUが使えないと長めの回答ではかなり時間がかかるため、そのあたりも課題になってくるかもしれません。

AIが喋ってくれるということで、それなりにインパクトが有り面白いと思いますので、興味のある方はぜひ試してみてください。

過去の関連記事

  • Chainlitについて
    • LangChainのAgentを、StreamlitとChainlitで作り比べてみた記事です。

https://zenn.dev/0msys/articles/d5a97c8670d5fb

  • LangGraphについて
    • LangGraphのAgentを、Chainlitで作ってみた記事です。

https://zenn.dev/0msys/articles/9873e25a610c5e

  • ↑のチャットアプリのVision対応版
    • AgentにVision toolを組み込んで、画像認識をやってみた記事です。

https://zenn.dev/0msys/articles/3d38729aa7f75b

  • ↑のリファクタリング
    • 今回の記事のためにリファクタリングを行った記事です。

https://zenn.dev/0msys/articles/49ebb76cea1af6

Discussion