LlamaIndexを使ってコンフルエンスのデータを元に回答するSlack botを作った

import os
from typing import List

from llama_index.core.readers import Document
from llama_index.readers.confluence import ConfluenceReader

from ..base.Base import Base


class LoadConfluenceDocument(Base):
    """Confluenceからドキュメントを読み込む

    Attributes:
        label(str): 収集したいドキュメントに付与されているlabel
        reader(ConfluenceReader): ConfluenceReaderのインスタンス
    """

    def __init__(self, label: str) -> None:
        """コンストラクタ

        Args:
            label(str): 収集したいドキュメントに付与されているlabel
        """
        super().__init__(__file__)
        self.label = label
        self.create_reader()

    def create_reader(self) -> None:
        """ConfluenceReaderのインスタンスを生成"""
        self.reader = ConfluenceReader(
            base_url=os.environ["CONFLUENCE_BASE_URL"],
        )

    def load_data(self) -> List[Document]:
        """Confluenceから読み込んだドキュメント

        Returns:
            List[Document]: 条件にマッチしたConfluenceのドキュメントをListで返す
        """
        cql = f'type="page" AND label="{self.label}"'
        return self.reader.load_data(cql=cql)


if __name__ == "__main__":
    load = LoadConfluenceDocument("android")
    docs = load.load_data()
    print(docs)

import os
from os.path import dirname, join

import yaml
from dotenv import load_dotenv


class Base(object):
    """各クラスで継承するための元クラス

    RailsみたいにRails.env.production的なことをしたいので親クラスで読み込んで使い回す
    yamlやenvファイルで設定した項目を各クラスで使うので実装している
    それ以外は基本的に実装しない（肥大化すると管理が大変だから）

    Attributes:
        current_dir(str): 各コードのディレクトリ
        dataset_yaml(list[str]): データセットの区分設定
        datasource_yaml(dict): どこのデータソースから取得するかの設定
    """

    def __init__(self, file_name: str) -> None:
        """コンストラクタ。環境変数を読み込む

        Args:
            file_name(str): ファイル名
        """
        self.current_dirname = dirname(file_name)
        self.setup_env()
        self.setup_yaml()

    def setup_env(self) -> None:
        """環境変数をセットアップする"""
        dotenv_path = join(self.current_dirname, "../../.env")
        load_dotenv(dotenv_path)

    def setup_yaml(self) -> None:
        """データセットの設定ファイルを読み込む"""
        datasource_yaml_path = join(self.current_dirname, "../../config/datasource.yml")
        with open(datasource_yaml_path, "r") as file:
            self.datasource_yaml = yaml.safe_load(file)

        dataset_yaml_path = join(self.current_dirname, "../../config/dataset.yml")
        with open(dataset_yaml_path, "r") as file:
            self.dataset_yaml = yaml.safe_load(file)


def is_development() -> bool:
    """開発モードかどうか。ローカルでの開発中に使う想定"""
    return os.environ["ENV"] == "development"


def is_production() -> bool:
    """本番かどうか。本番実行中でのみ使う想定"""
    return os.environ["ENV"] == "production"


def is_test() -> bool:
    """testかどうか。testで使う想定"""
    return os.environ["ENV"] == "test"

from os.path import join

from llama_index.core import GPTVectorStoreIndex, Settings
from llama_index.embeddings.google_genai import GoogleGenAIEmbedding
from llama_index.llms.google_genai import GoogleGenAI

from ..base.Base import is_development, is_production
from ..document_reader.LoadConfluenceDocument import LoadConfluenceDocument
from .GeminiBase import GeminiBase


class CreateGeminiIndex(GeminiBase):
    """文章をGeminiのLLMで使うためにベクトル化するクラス。

    Attributes:
        llm_model(str): LLMの使用モデル
        embed_model(str): 文章のベクトル化(embed)に使うモデル名
    """

    def __init__(
        self,
    ):
        """コンストラクタ"""
        super().__init__(__file__)
        # self.setup_gemini()

    def create_and_save_index(self) -> None:
        """LLMに渡すために文章のindex化&保存。

        indexの再生成はコストがかかるので生成後は保存する。
        """
        for dataset_name in self.dataset_yaml:
            # 各データソースから読み込んだドキュメントを保存する一時変数
            documents = []
            for datasource_name, datasource_value in self.datasource_yaml[
                dataset_name
            ].items():
                # コンフルエンスのデータソースが存在したら取り込む
                if datasource_name == "confluence":
                    for label in datasource_value:
                        load_confluence_document = LoadConfluenceDocument(label)
                        # 一括でindex化したいので一時変数に追加
                        documents.extend(load_confluence_document.load_data())
                # if datasource_name == "hogehoge" ここで他のデータソースが増えたら分岐を追加する
            # データセットに対するデータソースを全部探索したらindex化
            index = GPTVectorStoreIndex.from_documents(documents)
            if is_development:
                save_dir = self.save_index_file_dir_path(dataset_name)
                index.storage_context.persist(save_dir)
            elif is_production:
                pass


if __name__ == "__main__":
    print("start!!!")
    create_gemini_index = CreateGeminiIndex()
    create_gemini_index.create_and_save_index()

from os.path import join

from llama_index.core import Settings
from llama_index.embeddings.google_genai import GoogleGenAIEmbedding
from llama_index.llms.google_genai import GoogleGenAI

from ..base.Base import Base


class GeminiBase(Base):
    """Geminiに問い合わせする時に使う便利クラス

    Parameters:
        file(str): ファイル名
        llm_model(str): 使用するLLMのモデル名
        embed_model(str): index化する時に使うモデル名
    """

    def __init__(
        self,
        file: str,
        llm_model: str = "models/gemini-2.5-flash",
        embed_model: str = "text-embedding-004",
    ) -> None:
        """コンストラクタ

        Args:
            file(str): ファイル名
        """
        super().__init__(file)
        self.llm_model = llm_model
        self.embed_model = embed_model
        self.setup_gemini()

    def setup_gemini(self) -> None:
        """使用するGeminiのモデルを設定する"""
        Settings.llm = GoogleGenAI(model_name=self.llm_model)
        Settings.embed_model = GoogleGenAIEmbedding(model_name=self.embed_model)

    def save_index_file_dir_path(self, dataset_name: str) -> str:
        """index化したデータを保存するディレクトリパスを生成。

        ${dataset_name}_index_data というディレクトリ名で保存する。

        Args:
            dataset_name(str): データセット名

        Returns:
            パス形式で返す
        """
        dir_name = self.save_index_file_dir_name(dataset_name)
        return f"{join(self.current_dirname, dir_name)}"

    def save_index_file_dir_name(self, dataset_name: str) -> str:
        """index化したデータを保存するディレクトリ名を生成。

        ${dataset_name}_index_data というディレクトリ名にする。

        Args:
            dataset_name(str): データセット名

        Returns:
            ${dataset_name}_index_data という形式で返す
        """
        return f"{dataset_name}_index_data"

from llama_index.core import StorageContext, load_index_from_storage
from llama_index.core.base.response.schema import RESPONSE_TYPE

from .GeminiBase import GeminiBase


class AskGemini(GeminiBase):
    """Geminiに質問するクラス"""

    def __init__(self) -> None:
        super().__init__(__file__)

    def query(self, dataset_name: str, question: str) -> RESPONSE_TYPE:
        """質問をする

        indexデータは再生成するのではなくすでに保存しているものを流用する。
        先に保存していないとエラーになる。検証に使う。

        Args:
            dataset_name(str): 質問したいデータセット名
            question(str): 質問内容

        Returns:
            質問した内容に対する回答。
            返り値の中のsource_nodes.metadataから参考にしたドキュメントのタイトル(title)とURL(url)を取得できる。
            単純に返り値を出力した場合は回答内容が取れる。

        Examples:
            >>> ask_gemini = AskGemini()
            >>> question = "作っているアプリは何ですか？"
            >>> answer = ask_gemini.query("android", question)
            >>> print(answer) # LLMからの回答結果
            >>> for ans in answer.source_nodes:
            >>>     print(f'{ans.metadata["title"]}: {ans.metadata["url"]}')

        """
        storage_context = StorageContext.from_defaults(
            persist_dir=self.save_index_file_dir_path(dataset_name)
        )
        index = load_index_from_storage(storage_context)
        query_engine = index.as_query_engine()
        return query_engine.query(question)

    def prompt_tuning(self):
        """Geminiに問い合わせるためのプロンプトをカスタマイズ

        TODO: 後でいい感じにプロンプトをチューニングしてあげる
        """
        # a = ChatPromptTemplate()
        pass


if __name__ == "__main__":
    print("質問します！！！！！！")
    question = "誰がアプリを作っていますか?"
    ask_gemini = AskGemini()
    answer = ask_gemini.query("android", question)
    print(answer)
    for ans in answer.source_nodes:
        print("---------")
        print(f'{ans.metadata["title"]}: {ans.metadata["url"]}')
    print("終了")

import os

from slack_bolt import App

from ..gemini.AskGemini import AskGemini
from ..slack.SlackAppMention import SlackAppMention

"""Slackから返ってきたイベントに対して色々反応する関数たち
"""

slack_app_mention = SlackAppMention(__file__)
app = App(
    token=os.environ.get("SLACK_BOT_TOKEN"),
    signing_secret=os.environ.get("SLACK_SIGNING_SECRET"),
)


@app.event("app_mention")
def handle_message_events(event, say):
    """メンションされたらこの関数が呼ばれる

    ユーザが指示したデータセットと質問内容を判断し、
    適切なindexを使ってGeminiに問い合わせた後、Slackの質問してきたスレに返答
    """
    slack_app_mention.initialize(event)
    # 改行コードで分割する。最初にほしいデータセットが入っていることを期待
    lines = slack_app_mention.text_without_mentions.splitlines()

    # サイズが1だと@t-gpt hoge みたいな形式でメッセージが返っている
    # この場合は質問ではないのでそのコマンドに応じた返答をする
    if len(lines) == 1:
        command = lines[0].strip().lower()
        if command == "ping":
            say(
                text=slack_app_mention.ping_message(),
                thread_ts=slack_app_mention.thread_ts,
            )
            return
        elif command == "help":
            help_message = slack_app_mention.help_message()
            say(
                text=help_message[0],
                blocks=[
                    help_message[1],
                    help_message[2],
                    slack_app_mention.footer_message(),
                ],
                thread_ts=slack_app_mention.thread_ts,
            )
            return
        else:
            say(
                text="そのコマンドには対応していません。",
                thread_ts=slack_app_mention.thread_ts,
            )
            return
    elif len(lines) >= 2:
        say(
            text="回答を生成中です。しばらくお待ちください。",
            thread_ts=slack_app_mention.thread_ts,
        )
        dataset_name = lines[0].strip()
        # データセットが対応してない場合
        if not slack_app_mention.contains_dataset_name(dataset_name):
            message = f"データセット「{dataset_name}」は対応していません。"
            say(
                text=message,
                blocks=[
                    {
                        "type": "section",
                        "text": {"type": "mrkdwn", "text": message},
                    },
                    {
                        "type": "section",
                        "text": {
                            "type": "mrkdwn",
                            "text": "もしよければデータの作成にご協力ください！",
                        },
                    },
                    slack_app_mention.footer_message()[0],
                    slack_app_mention.footer_message()[1],
                ],
                thread_ts=slack_app_mention.thread_ts,
            )
            return
        gemini = AskGemini()
        # 質問を改行してくるかもしれないので結合してあげる。
        question = "".join(lines[1:])
        answer = gemini.query(dataset_name, question)
        say(
            text=str(answer),
            blocks=[
                slack_app_mention.answer_message(answer),
                slack_app_mention.reference_anchors(answer.source_nodes),
                slack_app_mention.footer_message()[0],
                slack_app_mention.footer_message()[1],
            ],
            thread_ts=slack_app_mention.thread_ts,
        )
    if not slack_app_mention.text_without_mentions:
        print("メンションのみで空文字です。")
    else:
        print("メッセージが詰まってます。")


if __name__ == "__main__":
    app.start(port=int(8080))

import re
from typing import List

from llama_index.core.base.response.schema import RESPONSE_TYPE
from llama_index.core.schema import NodeWithScore

from ..base.Base import Base


class SlackAppMention(Base):
    """Slackで返信する時に使うクラス
    各メソッドやメンバ変数を使う前にコンストラクタ->イニシャライザで初期化して使うことを想定している

    Params:
        user(str): メンションしてきたユーザ
        text(str): ユーザが送ってきたメッセージ
        thread_ts(str): スレ中でのメンションならその中に、新規だったらそのメンションの中にスレを作る
    """

    def __init__(self, file) -> None:
        """コンストラクタ

        Args:
            file(str): 実行するファイルのパス
        """
        super().__init__(file)

    def initialize(self, event) -> None:
        """イニシャライザ
        コンストラクタはslack_boltを使うために一度環境変数を読み込ませる。
        イニシャライザでは、Slackからメンションされた時に呼び出される変数をセットする。

        Args:
            event: Slackから返ってきたEvent。中に色々情報が詰まってる。
        """
        self.event = event
        self.user = event.get("user")
        self.text = event.get("text")
        self.thread_ts = event.get("thread_ts") or event.get("ts")
        self.text_without_mentions = self._text_without_mentions()

    def _text_without_mentions(self) -> str:
        """ユーザが送ってきたメッセージからメンション部分を削ぎ落として返す。"""
        return re.sub(r"<@([A-Z0-9]+)>", "", self.text)

    def answer_message(self, answer: RESPONSE_TYPE) -> dict:
        """LLMから返ってきた回答をblock形式で返す。

        Args:
            answer(RESPONSE_TYPE): LLMから返ってきた回答

        Returns:
            Slackのblock形式に合わせた回答
        """
        return {"type": "section", "text": {"type": "mrkdwn", "text": str(answer)}}

    def reference_anchors(self, refs: List[NodeWithScore]) -> dict:
        """LLMから返ってきたレスポンスがどのドキュメントを参考にしたかをSlack形式のアンカーのListで返す。

        Args:
            refs(List[NodeWithScore]): LLMから返ってきた参考にしたmetadata一覧

        Returns:
            Slack形式のアンカーを箇条書きのblockで返す。

            Slack形式のアンカー = <url|title>
        """
        anchors = []
        for ref in refs:
            title = ref.metadata["title"]
            url = ref.metadata["url"]
            anchors.append(f"• <{url}|{title}>")
        anchor_text = "\n".join(anchors)
        return {"type": "section", "text": {"type": "mrkdwn", "text": anchor_text}}

    def help_message(self) -> tuple[str, dict, dict]:
        """helpコマンドが返ってきた時にメッセージを返す。

        Returns:
            タプル形式で返す。
            messageはそのままblock kitのtextにセットすることを想定。
            それ以降のdictはblock kitにいれる想定。

        Examples:
            >>> say(
            >>>     text=help_message()[0],
            >>>         blocks=[
            >>>             help_message()[1],
            >>>             help_message()[2]
            >>>         ]
            >>> )
        """

        message0 = f"対応しているデータセットは{",".join(self.dataset_yaml)}になります。質問は以下の形式で質問してください。\n"
        message1 = "@g-gpt データセット名\n質問内容"
        return (
            message0 + message1,
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": message0},
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": message1,
                },
            },
        )

    def contains_dataset_name(self, dataset_name: str) -> bool:
        """Slackから指定されたデータセット名が対応しているかどうかを返す。

        Args:
            dataset_name(str): Slackから受け取ったデータセット名

        Returns:
            dataset.ymlにSlackから指定されたデータセット名が定義されているか。対応していたらTrue
        """
        return dataset_name in self.dataset_yaml

    def ping_message(self) -> str:
        """pingコマンド時に返すメッセージ。

        Returns:
            メンション付きで返す
        """
        return f"<@{self.user}> ping pong!🏓"

    def footer_message(self) -> List[dict]:
        """メッセージの最後にいれるフッターを返す。

        Returns:
            Listで返す。最初に区切り線で最後にメッセージ
        """
        url = "https://github.com/Tatsumi0000/t-gpt/"
        return [
            {"type": "divider"},
            {
                "type": "context",
                "elements": [
                    {
                        "type": "mrkdwn",
                        "text": f"powered by <{url}|t-gpt>",
                    }
                ],
            },
        ]


if __name__ == "__main__":
    print("start")