やりたいこと

SlackでLLMと会話できるBotを作ってみたので、仕組みと作成手順を備忘する

アーキテクチャ

今回は、OpenAIのAPIをLLMとして利用します。

シーケンス図は下記のようなイメージになります。

Slackは、メッセージを受け取ったことを3秒以内に伝えないとリトライして何度もリクエストしてしまうので、一度空メッセージを送るようにしています。

手順

アプリケーション構築の手順です。
大きく3つのステップがありました。

Slack Appの作成
Serverless Frameworkのプロジェクトの作成
動かす

1. Slack Appの作成

いつの間にか、App Manifestという機能ができていて、YAML・JSONファイルを使ってアプリケーションを構成できるようになっていました。

こちらのURLから、Create New App -> From an app manifest -> workspaceを選択 -> manifestを入力 -> 内容を確認しCreateボタンを押す -> Install to workspaceボタンを押す

で、同じ構成のアプリケーションを作れてしまいます。

今回作ったアプリのmanifestは下記のようになりました。

display_information:
  name: llm-bot-app
features:
  bot_user:
    display_name: llm-bot
    always_online: true
oauth_config:
  scopes:
    bot:
      - app_mentions:read # Botをメンションをトリガーに起動するので必要
      - chat:write # Botがチャネルに書き込むために必要
settings:
  event_subscriptions:
    # 後で作成するAPIGatewayのエンドポイント
    # 初回作成時はhttps://example.com/slack/eventsなどで良い
    request_url: https://XXXXXX.ap-northeast-1.amazonaws.com/slack/events
    bot_events:
      - app_mention
  org_deploy_enabled: false
  socket_mode_enabled: false
  token_rotation_enabled: false

2. Serverlessアプリケーションの作成

今回は、SlackとLLMを中継する機能をAPIGateway + Lambdaを使って実現します。

ディレクトリ構成は、下記のようになりました。

.
├── app.py
├── requirements.txt
└── serverless.yml

serverless.yml

frameworkVersion: '3'
service: llm-app-slack
provider:
  name: aws
  runtime: python3.9
  region: ap-northeast-1

  environment:
    SERVERLESS_STAGE: ${opt:stage, 'prod'}
    SLACK_SIGNING_SECRET: ${env:SLACK_SIGNING_SECRET}
    SLACK_BOT_TOKEN: ${env:SLACK_BOT_TOKEN}
    OPENAI_API_KEY: ${env:OPENAI_API_KEY}

functions:
  app:
    handler: app.handler
    timeout: 29 # デフォルトのtimeoutだとopenaiの回答に間に合わない
    reservedConcurrency: 1 # 同時実行を1に制限することで、会話の記憶がブレないようにする
    events:
      - httpApi:
          path: /slack/events
          method: post

package:
  patterns:
    - "!node_modules/**"

plugins:
  - serverless-python-requirements
  # ~/.aws/configのプロファイルを指定してデプロイできるようにするため。状況によってはなくても良い
  - serverless-better-credentials
custom:
  pythonRequirements:
    # boltがpydanticなどのsoファイルに依存しており、lambdaと同じ環境にあったsoを用意する必要があるため
    dockerizePip: true
    dockerImage: public.ecr.aws/sam/build-python3.9:latest-x86_64

今回デプロイするLambdaでは、pythonのLLMと会話するためのフレームワークであるlangchainやSlackとの通信のライブラリであるslack-boltを利用するため、requirements.txtを使って依存ライブラリをLambdaパッケージに組み込みます

requirements.txt

slack-bolt==1.18.0
langchain==0.0.279
openai==0.28.0
urllib3<2

urllib3については、2.Xだとエラーになってしまったためバージョンを1系に固定しています

app.py

# serverless-python-requirements を使って zip 圧縮しておいた依存ライブラリの読み込み
try:
    import unzip_requirements
except ImportError:
    pass

import logging
from collections import defaultdict
import os
from slack_bolt import App
from langchain.chat_models import ChatOpenAI
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
from langchain.prompts import (
    ChatPromptTemplate,
    MessagesPlaceholder,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
)

# --------------------------------
# LangChain周りの設定
# --------------------------------
chat = ChatOpenAI(
    temperature=0,
    openai_api_key=os.environ["OPENAI_API_KEY"],
    model_name="gpt-3.5-turbo",
)

# テンプレートの準備
template = """あなたは誠実に日本語で回答するAIです。"""
prompt = ChatPromptTemplate.from_messages(
    [
        SystemMessagePromptTemplate.from_template(template),
        MessagesPlaceholder(variable_name="history"),
        HumanMessagePromptTemplate.from_template("{input}"),
    ]
)

# ユーザーごとに会話をメモリ上に覚える
user_chat = defaultdict(
    lambda: ConversationChain(
        memory=ConversationBufferMemory(return_messages=True), prompt=prompt, llm=chat
    )
)
# --------------------------------
# Slack BotのBolt周りの設定
# --------------------------------
app = App(
    # リクエストの検証に必要な値
    # Settings > Basic Information > App Credentials > Signing Secret で取得可能な値
    signing_secret=os.environ["SLACK_SIGNING_SECRET"],
    # 上でインストールしたときに発行されたアクセストークン
    # Settings > Install App で取得可能な値
    token=os.environ["SLACK_BOT_TOKEN"],
    # AWS Lamdba では、必ずこの設定を true にしておく必要があります
    process_before_response=True,
)

# --------------------------------
# Slack BotのBolt周りの設定
# --------------------------------


@app.event("app_mention")
def answer_by_llm(body, say):
    user = body["event"]["user"]
    conversation = user_chat[user]
    text = body["event"].get("text", "")
    text = text.replace("<@BotのUserID>", "").strip()
    try:
        if not text:
            say("なんでしょうか？")
        elif "reset" in text:
            # 会話の履歴をリセット
            conversation.memory.clear()
            say(f"{user}さんの会話の履歴をリセットしました")
        else:
            res = conversation.predict(input=text)
            say(res)
    except Exception:
        import traceback
        # CloudWatch Logsでセクションをまとめるために改行をキャリッジリターンに変換している
        print(traceback.format_exc().replace("\n", "\r"))


# SlackはAPIへのPOSTリクエストに対してレスポンスがないとリトライを行う（それによって無駄にPOSTリクエストが増えてしまう）
# commandなどのイベントハンドラの場合は、遅延処理の仕組みがある https://github.com/slackapi/bolt-js/issues/1299
# しかし、eventの場合遅延処理仕組みが対応していないため、同じイベントに対して即時に200レスポンスを返すハンドラも用意してあります
# -> 記事へのコメントで、eventの場合でも対応していることがわかった。要調査
@app.event("app_mention")
def fast_response(say):
    say("")


if __name__ == "__main__":
    # LocalPCで動作確認するときはpython app.py のように実行します
    app.start()
    import sys

    # 開発時にはこれ以降を実行しない
    sys.exit(0)

# これより以降は AWS Lambda 環境で実行したときのみ実行されます

from slack_bolt.adapter.aws_lambda import SlackRequestHandler

SlackRequestHandler.clear_all_log_handlers()
logging.basicConfig(format="%(asctime)s %(message)s", level=logging.DEBUG)

def handler(event, context):
    slack_handler = SlackRequestHandler(app=app)
    # 万が一リトライ処理が行われたとしても、slack handlerでの処理を行わないようにする対応
    if event["headers"].get("x-slack-retry-num"):
        print("retry called")
        return {"statucCode": 200}
    return slack_handler.handle(event, context)

3. 動かす

Serverless のプロジェクトのディレクトリ下で、環境変数を設定した後

# Settings > Basic Information > App Credentials > Signing Secret で取得可能な値
export SLACK_SIGNING_SECRET=XXXXXX
# Settings > Install App で取得可能な値
export SLACK_BOT_TOKEN=xoxb-XXXXXXXX
# https://platform.openai.com/account/api-keys で発行
export OPENAI_API_KEY=XXXXXXXX
# serverlessでデプロイするときに、~/.aws/configからプロファイルを読み込む設定
export AWS_SDK_LOAD_CONFIG=1

docker デーモンを起動し、下記コマンドを実行しデプロイします

serverless plugin install -n serverless-python-requirements
serverless plugin install -n serverless-better-credentials
serverless deploy --aws-profile <~/.aws/configのプロファイル名>

すると、結果は下記のようになり、

...中略
✔ Service deployed to stack llm-app-slack-dev (142s)

endpoint: POST - https://XXXXXXXX.ap-northeast-1.amazonaws.com/slack/events
functions:
  app: llm-app-slack-dev-app

APIGatewayが作成され、APIのエンドポイントができるので、SlackのApplicationのEvent Subscriptions -> Request URLに入力します -> Verifiedの表示が出たらSave Changesボタンを押します

https://XXXXXXXX.ap-northeast-1.amazonaws.com/slack/events

次にSlackを開き、任意のチャネルに、SlackのApplicationを追加します。
Slackのチャネルを開いて -> 右上のメンバーのところをクリック -> インテグレーション -> Appを追加

Appを追加したチャネルで、botに対してメンションして投稿すると、しばらくすると、Botからの返信が来るので、これで完成です！

詰まったところ・工夫したところ

Serverlessデプロイの時に、依存ライブラリのエラーに苦戦
- 依存ライブラリ込みのzipファイルをdockerを使って作る必要があった
- urllib3のバージョンを下げる必要があった
Slackのリトライを止めるのに苦戦
- Slackは、登録したエンドポイントからのレスポンスがないと最大4回リトライ処理をします
- 最初、そのリトライによって何度も同じ内容をlambdaに送ってしまい、何度も同じ回答が返ってくる状況になってしまいました
- slackコマンドなどの場合は、回避策があったのですが、Botへのメンションのイベントをトリガーにして回答させる方法を見つけるのに試行錯誤しました
  - 結局、イベントハンドラを二つに分岐させ、先に空文字をsayするというアプローチに一旦落ち着きました（ベストプラクティスは引き続き要調査）
ユーザーごとに会話を覚えるようにした
- langchainのConversationBufferMemoryを使って、メモリ上に会話の履歴を残すようにしました
  - 今は、Lambdaが落ちると会話がリセットされてしまうので、今後永続化なども検討してみたい

参考にしたサイト