SlackへのChatGPTのBotの導入について

はじめに

ユーザーがメンション付きメッセージを送信
Slack Event APIがメッセージをサーバーに送信
OpenAI APIにメッセージを送信
chat.postMessageでBotがメンション付きメッセージのスレッドに返信

といった流れでユーザーのメッセージに対して、ChatGPTが返信を行うBotの作成を行います。アーキテクチャは以下の通りです。

APIサーバーの実装

import os

from fastapi import FastAPI, Request, Response
from openai import OpenAI, OpenAIError, RateLimitError
from slack_sdk import WebClient

from logs import get_logger
from schema import SlackOutcomingRequest
from settings import DEFAULT_MODEL

# FastAPIアプリケーションの初期化
app = FastAPI()

# ロガーのセットアップ
logger = get_logger(__name__)

# Slackクライアントの初期化（OAuthトークンを環境変数から読み込む）
slack_client = WebClient(token=os.getenv("BOT_OAUTH_TOKEN"))

@app.post("/slack-bot", status_code=200)
async def slack_bot(request: Request) -> Response:
    # Slackからのリクエストヘッダーとボディを取得
    slack_headers = request.headers
    request_body = await request.json()

    # SlackのURL検証リクエストに応答
    if request_body.get("type") == "url_verification":
        return Response(content=request_body.get("challenge"), media_type="text/plain")

    # 再送リクエストの場合は処理をスキップ
    if is_retry_request(slack_headers):
        return Response(status_code=200, content="is_retry_request")

    # Slackリクエストデータを解析
    slack_request = SlackOutcomingRequest(**request_body)

    # スレッド内のメッセージを取得
    thread_messages = get_thread_messages(
        slack_request.event.channel,
        slack_request.event.thread_ts,
    )

    # GPTを使って回答を生成
    answer = generate_answer_with_gpt(slack_request.event.text, thread_messages)

    # ボットによるメッセージの送信
    _ = send_message_by_bot(
        answer, slack_request.event.channel, slack_request.event.event_ts
    )
    return Response(status_code=200)

def generate_answer_with_gpt(text, thread_messages) -> str:
    """
    GPTを使用して回答を生成する関数。

    Args:
        text (str): ユーザーからのメッセージ。
        thread_messages (list): スレッド内の過去のメッセージリスト。

    Returns:
        str: GPTによって生成された回答。
    """
    try:
        # OpenAIクライアントの初期化
        client = OpenAI(api_key=os.getenv("OPENAI_KEY"))

        # 過去のメッセージをプロンプトとして整形
        past_prompt = ", ".join(thread_messages[-11:-1])

        # GPTに送るメッセージの構成
        messages = [
            {
                "role": "system",
                "content": f"""
                You are a chatbot designed to respond in simple, direct Japanese.
                Based on the conversation history, provide concise and relevant answers.

                Conversation History:
                [{past_prompt}]

                Considering this history, generate an appropriate response to the latest message.
                """,
            },
            {"role": "user", "content": text},
        ]

        # GPTによる回答の生成
        response = client.chat.completions.create(
            model=DEFAULT_MODEL,
            messages=messages,
        )
        return response.choices[0].message.content.strip()

    except RateLimitError as e:
        logger.exception(e)
        return "短時間でGPTのAPIのリクエスト数制限を超えました。しばらく待って、再度実行してください。"
    except OpenAIError as e:
        logger.exception(e)

        # メッセージ長の制限チェック
        max_lengths = 7000
        content_lengths = sum(len(message["content"]) for message in messages)
        if content_lengths > max_lengths:
            return "入力文字数が大きすぎます。スレッドを変更、文字数を小さくして再度実行してください。"
        return "OpenAIのサービスでエラーが発生しました。"

def send_message_by_bot(answer, channel_id, thread_ts) -> bool:
    """
    ボットを使用してSlackチャンネルにメッセージを送信する関数。

    Args:
        answer (str): 送信するメッセージ。
        channel_id (str): メッセージを送信するチャンネルのID。
        thread_ts (str): スレッドのタイムスタンプ。

    Returns:
        bool: メッセージの送信が成功したかどうか。
    """
    slack_client.chat_postMessage(channel=channel_id, text=answer, thread_ts=thread_ts)
    return True

def get_thread_messages(channel_id, thread_ts) -> list:
    """
    指定されたスレッドからメッセージを取得する関数。

    Args:
        channel_id (str): チャンネルのID。
        thread_ts (str): スレッドのタイムスタンプ。

    Returns:
        list: スレッド内のメッセージリスト。
    """
    if thread_ts:
        response = slack_client.conversations_replies(channel=channel_id, ts=thread_ts)
        thread_messages = [message["text"] for message in response["messages"]]
        logger.info(f"thread_messages: {thread_messages}")
        return thread_messages
    return []

def is_retry_request(slack_headers) -> bool:
    """
    リクエストがSlackからの再送リクエストかどうかを判断する関数。

    Args:
        slack_headers (dict): リクエストのヘッダー。

    Returns:
        bool: 再送リクエストかどうか。
    """
    return "X-Slack-Retry-Num" in slack_headers

OpenAIのAPI KEYの発行方法 BOT_OAUTH_TOKENはSlack Appの設定時に取得します。

各関数の機能は以下の通りです。

@app.post("/slack-bot", status_code=200): FastAPIのエンドポイントになります。

  # SlackのURL検証リクエストに応答
  if request_body.get("type") == "url_verification":
      return Response(content=request_body.get("challenge"), media_type="text/plain")

Event APIのの設定をSlack Appから行う際にURL認証のために、特定の文字列をresponseとして返す必要があります。

generate_answer_with_gpt: リクエストをもとにOpenAI APIを用いて回答を生成します。プロンプトは使用用途に応じて作成してください。

send_message_by_bot: Slackにメッセージを返します。ここでは自分のメッセージのに対してBotが返信を行うようにしたいため、chat.postMessageを使います。Incoming Webhooksを代わりに用いることもできますが、その場合はユーザーのメッセージの下にBotが新しくメッセージをチャンネルに送信する形になります。

get_thread_messages: スレッドのメッセージ履歴を取得します。ここでメッセージの履歴を取得することで、Botがスレッド内の過去の会話を踏まえて回答を出力することができるようになります。

is_retry_request: EventAPIはrequestを何度も送信してしまうことがあります。多重のresponseを返すのを防ぐためにヘッダーからrequestが再送されたものであるかを判断します。

FROM python:3.10.0

EXPOSE 8080

ENV POETRY_HOME="/opt/poetry" \
    POETRY_VERSION=1.4.2
ENV PATH="$POETRY_HOME/bin:$PATH"

RUN apt-get update &amp;&amp; \
    apt-get install -y curl &amp;&amp; \
    pip install -U pip &amp;&amp; \
    curl -sSL https://install.python-poetry.org | python3 - &amp;&amp; \
    poetry config virtualenvs.create false

WORKDIR /api
COPY pyproject.toml poetry.lock ./
RUN poetry install --no-dev

COPY . /api

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Dockerfileは上のようになります。これをCloud Runにデプロイします。

Slackの設定

実装を行う前にWorkplaceにBotを導入するためにSlack Appを作成をします。

SlackAppの作成

画面上部の[Create New App]→[From Scratch]と進みます。アプリ名、アプリを使用するワークスペースを選択します。この際に、ブラウザからワークスペースにログインを行う必要があります。アプリ名とワークスペースの設定が終わり、[Create App]で次のページに遷移します。

Event APIの設定

画面左のサイドバーの[Settings]→[Basic Information]から[Event Subscription]を選択します。 [Enable Event]をOnにしてRequest URLにCloudRunにデプロイしたエンドポイントのURLを入力します。画面下にある[Subscribe to bot events]からapp_mentionの権限を追加します。完了したら、画面の最下部で変更を保存します。

Botの権限の設定

サイドバーから[OAuth & Permissions]を選択します。画面下部の[Bot Token Scopes]からchannels

、chat

、app_mentions

の権限を選択します。

アプリのインストール

再び[Basic Information]のページに戻り、作成したアプリをワークスペースにインストールします。管理者権限がない場合はワークスペースの管理者へのリクエストが再び必要になリます。

Oauth Tokenを取得

次に[Basic Information]→[Add features and functionality]→[OAuth & Permissions]を選択します。アプリのインストール後に[Bot User OAuth Token]が表示されます。取得完了後、取得したBot User OAuth Tokenを環境変数に設定します。

Botのユーザーネームやプロフィール画像について

[Basic information]の画面下の[Display Information]から設定が可能です。

APIサーバーの実装とSlackの設定が完了後、Slackからメンション付きメッセージを送ることでBotから返信を得ることができます。

備考

ChatGPT を業務で活用したい場合、機密情報を入力することがセキュリティの観点から NG だったり、自社のシステムに組み込まないと効率が悪かったりする場合があります。Hakky では、ChatGPT を用いた企業独自のシステムを構築するご支援を行っています。ソリューションサイト：https://www.about.st-hakky.com/chatgpt-solution 「どんなことが出来るのか」「いま考えているこのようなことは本当に実現可能か」など、ご検討段階でも構いませんので、ぜひお気軽にフォームよりお問い合わせくださいませ。