こんにちは、こんばんは、主に課金システム担当のrouganと言います。普段はJavaで課金周りの開発や運用、メンテナンスを中心に作業をしています。その為、今回のシステムはほぼ手探りの状態で作ったので自分に取ってはなかなか大変でした。

今回はSlackでChatGPT APIを使えるようにした話をしたいと思います。既に多くの現場で利用されているとは思いますが、Slack公式でChatGPTが使えるようになるまでのつなぎのシステムとして稼働しています。が、まだ弊社では（2023/12/12現在)は使っておりません。
ChatGPT app for Slack を発表

概要

今回のシステムは、AWS Lambda上で動作するSlack Botを構築するためのPythonスクリプトです。Slack Botは、OpenAIのGPT-4を利用してユーザーからのメッセージに対応し、対話形式で応答します。ソースコードは2つの主要部分で構成されています: Slack Botの実装と AWSサービスの構成を記述するSAMテンプレートです。

Slack Botの実装

AWS SDKboto3の使用: AWSのリソース（DynamoDB, SSM）にアクセスするために使用されます。
Slack Bolt Framework: Slackとの通信を容易にするためのフレームワークです。
LangChain: GPT-4を活用して対話を生成するライブラリです。
DynamoDBへのデータ格納: ユーザーからの入力とAIの応答を記録します。
Lambdaハンドラー: AWS Lambda上でのSlackイベントの処理を行います。

main handler

def lambda_handler(event, context):
    # Slack challenge key handling
    if "challenge" in event.get("body-json", {}):
        return {
            "statusCode": 200,
            "body": event["body-json"]["challenge"],
            "headers": {"Content-Type": "text/plain"},
        }
    else:
        handler = SlackRequestHandler(app)
        return handler.handle(event, context)

Slackのチャレンジ応答

if "challenge" in event.get("body-json", {}):
この行は、Slackからのイベントがチャレンジ要求かどうかを確認します。Slackは、新しいURLがイベントを受け取るために設定されたとき、そのURLが有効かどうかを確認するためにチャレンジ要求を送信します。3秒以内でお返事しないといけないです。ここが結構苦労しました。
return {...}: チャレンジ要求が検出された場合、LambdaはSlackにチャレンジコードを含む応答を返します。これにより、Slackは送信元のURLが正しいかどうかを確認できます。

Slackイベントの処理:

else: この部分は、受信したイベントがチャレンジ要求ではない場合に実行されます。
handler = SlackRequestHandler(app): Slackイベントを処理するために、Slack BoltフレームワークのSlackRequestHandlerオブジェクトを初期化します。このハンドラーは、app（Slack Boltアプリケーション）を引数として受け取ります。
return handler.handle(event, context): Slackハンドラーにイベントを渡し、処理を委ねます。この行は、Slackからのイベント（メッセージ、メンション等）に対して定義されたロジック（例えば、メッセージに応答する等）に基づいて、適切なアクションを実行します。

Slack Botユーザーアクション

app.event("app_mention")(
    ack=respond_to_slack_within_3_seconds,
    lazy=[run_chat_gpt]
)

app.event("message")(
    ack=respond_to_slack_within_3_seconds,
    lazy=[run_chat_gpt]
)

イベントリスナーの設定:

app.event("app_mention")とapp.event("message"): これらの行は、Slackのapp_mentionイベント（ユーザーがBotをメンションした場合）とmessageイベント（ユーザーがチャネルにメッセージを送信した場合）に対するリスナーを設定しています。

応答確認（Acknowledgement）:

ack=respond_to_slack_within_3_seconds: この引数は、イベントを受け取ったことをSlackに伝えるための確認応答を指定します。respond_to_slack_within_3_seconds関数は、イベントを受け取ってから3秒以内にSlackに応答を返すように設計されています。これは、Slackの要件に従って、Botがイベントを受け取ったことを迅速に通知するために重要です。

非同期処理

lazy=[run_chat_gpt]: この部分は、実際の処理（この場合はrun_chat_gpt関数）を非同期処理するための設定です。lazyパラメータを使用することで、応答確認を送った後に、実際の処理を非同期的に実行することができます。run_chat_gpt関数は、イベントの内容に基づいて適切なアクション（例えば、ユーザーのメッセージに対するGPT-4を使用した応答生成）を行います。

主要コード

ChatGPT APIを利用して返答を受け取る主要部分です。

def run_chat_gpt(body, say):
    event = body['event']

    user, message_text = process_event(event)

    if user not in user_chains:
        user_chains[user] = ConversationChain(
            llm=openai_instance,
            prompt=prompt_template,
            memory=ConversationBufferWindowMemory(k=3),
            verbose=True
        )
    llm_chain = user_chains[user]

    try:
        thread_ts = event["ts"]
        channel_id = event["channel"]
        # OpenAI APIで返答を取得
        openai_response = llm_chain.predict(input=message_text)

        insert_dynamo_db(user, message_text, openai_response, thread_ts, channel_id)

        say(f'{openai_response}', thread_ts=thread_ts)
    except Exception as e:
        logger.error(f'Error sending message to user: {user}', exc_info=True)

イベントデータの処理:

event = body['event']: Slackからのイベントデータを取得します。
user, message_text = process_event(event): イベントからユーザーIDとメッセージテキストを抽出します。

ユーザーごとの会話チェーンの管理

if user not in user_chains: ...: ユーザーIDがuser_chainsに存在しない場合、新しいConversationChainを作成して登録します。これにより、ユーザーごとに独立した会話履歴を管理できます。

OpenAI GPT-4を使用した応答の生成

openai_response = llm_chain.predict(input=message_text): 抽出されたメッセージテキストを使用して、GPT-4モデルによる応答を生成します。

DynamoDBへのデータ記録

insert_dynamo_db(user, message_text, openai_response, thread_ts, channel_id): ユーザーのメッセージとAIの応答をDynamoDBに記録します。これにより、対話の履歴を追跡できます。

Slackチャネルへの応答の投稿

say(f'{openai_response}', thread_ts=thread_ts): 生成された応答をSlackチャネルに送信します。thread_tsを使用することで、応答が適切なスレッド（会話）に紐づけられます。

エラーハンドリング

except Exception as e: ...: 応答の生成または送信中にエラーが発生した場合、ログにエラー情報を記録します。

ChatGPTとの会話の継続

ConversationBufferWindowMemoryを使って3つ前までの会話をpromptに含めて送信しています。この数を大きくすれば返答の精度はあがるのですが、その分tokenの数も大きくなるので課金が進みます。また、一度に送信できるtokenの上限にひっかかる可能性もでてきます。今回は3つとしてみました。GPT-4 Turboを使えばtoken数はもっと大きくなります。

ConversationBufferWindowMemory以外にも以下のMemoryがあります。

Type	内容
ConversationBufferMemory	このメモリはメッセージを保存し、変数にメッセージを取り出します
ConversationEntityMemory	会話の中の特定のエンティティに関する与えられた事実を記憶する
ConversationKGMemory	このタイプの記憶は、ナレッジグラフを使って記憶を再現する
ConversationSummaryMemory	このタイプのメモリは、時間の経過とともに会話の要約を作成する。これは、時間の経過とともに会話の情報を凝縮するのに便利です
ConversationSummaryBufferMemory	最近のやりとりのバッファをメモリに保持しますが、古いやりとりを完全にフラッシュするのではなく、それらをサマリーにまとめ、両方を使用します
ConversationTokenBufferMemory	最近のインタラクションのバッファをメモリに保持し、インタラクションをフラッシュするタイミングを決定するために、インタラクションの数ではなくトークンの長さを使用します
VectorStoreRetrieverMemory	ベクトルストアに記憶を保存し、呼び出されるたびに上位K個の最も "顕著な "ドキュメントを問い合わせる。これは他のほとんどのMemoryクラスと異なり、相互作用の順序を明示的に追跡しない

それぞれのMemoryの特徴を理解して試行錯誤すればもっと適切なBotが作れるとは思いますが、まずは理解が簡単なConversationBufferWindowMemoryを使いました。
詳細な情報はLangChainこちらを参照ください。

AWSサービスの構成を記述するSAMテンプレート

AWSリソース（DynamoDBテーブル、Lambda関数、API Gateway）の定義。
Lambda関数のメモリサイズやタイムアウト設定を含みます

Lambdaのウォーム状態

SAM templateの詳細は割愛しますが、1つだけ。LambdaがコールドスタートするためにSlackに3秒以内で返答できずにChatGPT APIから同じ応答を複数回受信してしまう現象に悩まされました。これを無理やり解決したのが、Lambdaをウォーム状態にしておくことです。

ProvisionedConcurrencyConfig:
  ProvisionedConcurrentExecutions: 1

ProvisionedConcurrentExecutions: 1: この設定は、常に少なくとも1つのプロビジョニングされたLambda関数のインスタンスをアクティブに保持することを意味します。
つまり、このLambda関数は常に「ウォーム」状態であり、新しいイベントに対してより迅速に応答できます。これにより、コールドスタートによる遅延を低減できます。
利点としては特に、パフォーマンスが重要な場合や、一定のトラフィックが予想される場合に有効です。これにより、Lambda関数の応答時間が短縮され、より一貫したパフォーマンスが実現されます。考慮すべき点はコストがかかることです。（えー、関係者様ごめんなさい）その為、最小限の数だけを設定してあります。

まとめ

ChatGPT Slack Botを紹介しました。LangChainを使いこなせばもっと複雑な色々な物が作れそうです。
ただし、これはSlackが提供するChatGPTの機能を使うまでのつなぎBotですので、自分の業務との折り合いをつけて（言い訳）ここまでしか実装できていません。Slack公式のBotであればSlack独自の色々な機能と連携してもっと便利なものになるのではないかと、期待しています。早く使ってみたいものですね。
最後にこのBotを作るために色んなサイトを参考にさせていただきました。改めて感謝を申し上げます。

明日は@c_hammerheadさんのGitHub Copilot Chatの記事です。