＜目次＞
OpenAIで返答するSlackボットで「会話履歴」を加味する方法
やりたいこと
前提条件
概要
サンプルプログラム
動かし方
OpenAIで返答するSlackボットで「会話履歴」を加味する方法
やりたいこと
・SlackボットでOpenAIのAPIから回答を行い、尚且つその会話履歴を保持する
（図１００）イメージ
前提条件
概要
基本的なSlackボットの考え方は「（手順）Slackでボットを開発する手順をご紹介（Python）」で紹介しているので割愛しますが、今回のポイントとしては以下です。
●【ポイント】「会話履歴」を加味した応答を実現するために、以下の実装を行った
・①スレッド履歴を保持する「メッセージ情報保持クラス」を用意
（図１１２）
・②それらをLangchainを使って、「人からのメッセージ」と「ボットからの応答メッセージ」を識別して配列に格納
（図１１３）
サンプルプログラム
（図１１１）３つのPythonモジュールがあります。
●メイン処理：0774_app.py
from slack_bolt import App
from slack_sdk import WebClient
# Flaskクラスのインポート
from flask import Flask, request
from slack_bolt.adapter.flask import SlackRequestHandler
#ソケットモード用
from slack_bolt.adapter.socket_mode import SocketModeHandler
# 環境変数読み込み
import env
# OpenAI
import openai
import pkg_resources
import respond_to_message
# モードに応じて書き換え
BOT_USER_ID = env.get_env_variable("BOT_USER_ID")
# Botトークン（Flask）
WEBAPPS_SLACK_TOKEN = env.get_env_variable("WEBAPPS_SLACK_TOKEN")
WEBAPPS_SIGNING_SECRET = env.get_env_variable("WEBAPPS_SIGNING_SECRET")
# Botトークン（ソケットモード）
SOCK_SLACK_BOT_TOKEN = env.get_env_variable("SOCK_SLACK_BOT_TOKEN")
SOCK_SLACK_APP_TOKEN = env.get_env_variable("SOCK_SLACK_APP_TOKEN")
# OpenAIのAPI設定
openai.api_key = env.get_env_variable('OPEN_AI_KEY')
openai_version = pkg_resources.get_distribution("openai").version
# モード入れ替え（WebAPサーバ実行＝Flask／ローカル実行＝ソケットモード)
def app_mode_change(i_name):
if i_name == "__main__":
return App(token=SOCK_SLACK_BOT_TOKEN)
else:
return App(token=WEBAPPS_SLACK_TOKEN, signing_secret=WEBAPPS_SIGNING_SECRET)
# グローバルオブジェクト
s_app = app_mode_change(__name__)
# Flaskクラスのインスタンス生成
app = Flask(__name__)
#app.config['JSON_AS_ASCII'] = False
handler_flask, handler_socket = None,None
#ソケットーモードの場合のハンドラ設定
if __name__ == "__main__":
handler_socket = SocketModeHandler(app=s_app, app_token=SOCK_SLACK_APP_TOKEN, trace_enabled=True)
#Flaskでのハンドラー設定
else:
handler_flask = SlackRequestHandler(s_app)
# Flask httpエンドポイント
# 疎通確認用1
@app.route('/', methods=['GET', 'POST'])
def home():
return "Hello World Rainbow 2!!"
# 疎通確認用2
@app.route("/test", methods=['GET', 'POST'])
def hello_test():
return "Hello, This is test.2!!"
#イベント登録されたリクエストを受け付けるエンドポイント
@app.route("/slack/events", methods=["POST"])
def slack_events():
# ------------------------------------
# Challenge用
# ------------------------------------
# # Slackから送られてくるPOSTリクエストのBodyの内容を取得
# json = request.json
# print(json)
# # レスポンス用のJSONデータを作成
# # 受け取ったchallengeのKey/Valueをそのまま返却する
# d = {'challenge' : json["challenge"]}
# # レスポンスとしてJSON化して返却
# return jsonify(d)
# ------------------------------------
# 本番用
# ------------------------------------
return handler_flask.handle(request)
@s_app.event("message")
@s_app.event("app_mention")
def respondToRequestMsg(body, client:WebClient, ack):
ack()
type = body["event"].get("type", None)
print("-------------"+str(type))
# 二重で応答するのを防ぐため、メンションの時のイベントのみ応答対象とする
if type == 'app_mention':
respond_to_message.respond_to_message(body=body,client=client)
# __name__はPythonにおいて特別な意味を持つ変数です。
# 具体的にはスクリプトの名前を値として保持します。
# この記述により、Flaskがmainモジュールとして実行された時のみ起動する事を保証します。
# （それ以外の、例えば他モジュールから呼ばれた時などは起動しない）
if __name__ == '__main__':
EXEC_MODE = "SLACK_SOCKET_MODE"
# Slack ソケットモード実行
if EXEC_MODE == "SLACK_SOCKET_MODE":
handler_socket.start()
# Flask Web/APサーバ 実行
elif EXEC_MODE == "FLASK_WEB_API":
# Flaskアプリの起動
# →Webサーバが起動して、所定のURLからアクセス可能になります。
# →hostはFlaskが起動するサーバを指定しています（今回はローカル端末）
# →portは起動するポートを指定しています（デフォルト5000）
app.run(port=8000, debug=True)
●Slackイベント解析＆応答処理：respond_to_message.py
from slack_sdk import WebClient
import conversation_util
import env
import re
import time
# OpenAI
import openai
openai.api_key = env.get_env_variable('OPEN_AI_KEY')
model=env.get_env_variable('MODEL')
import pkg_resources
openai_version = pkg_resources.get_distribution("openai").version
# langchain
from langchain import schema
from langchain.chat_models import ChatOpenAI
# from langchain.chat_models import AzureChatOpenAI
from langchain.schema import HumanMessage
# ロギング
import traceback
def respond_to_message(body, client: WebClient):
try:
#-----------------------------------
# Slackのイベント情報から各種パラメータを取得
bot_user_id = env.get_env_variable('BOT_USER_ID')
ts = body["event"]["ts"]
thread_ts = body["event"].get("thread_ts", None)
channel = body["event"]["channel"]
user = body["event"]["user"]
input_text = schema.HumanMessage(content=body["event"].get("text", None))
attachment_files = body["event"].get("files", None)
system_message = schema.SystemMessage(content=env.get_env_variable('SYSTEM_MESSAGE'))
# やり取りに関するインスタンス生成
conversation_info = conversation_util.ConversationInfoSlack(
client=client,
bot_user_id=bot_user_id,
ts=ts,
thread_ts=thread_ts,
channel=channel,
user=user,
human_message_latest=input_text,
messages=None,
system_message=system_message
)
# メッセージ情報の構築
conversation_info.build_messages()
print("==============="+str(conversation_info._messages))
# OpenAIからの返答を生成
output_text = generate_response_v2(str(conversation_info._messages))
time.sleep(1) # n秒待機 (実施しないと「The server responded with: {'ok': False, 'error': 'no_text'}」になる)
# # Slackに返答
client.chat_postMessage(channel=channel, text=output_text ,thread_ts=ts)
except Exception as e:
print("-------------------------------------------------")
print("======== react_to_msg 例外発生："+str(e))
traceback.print_exc()
def generate_response_v2(prompt) ->str:
print("============ generate_response : TOTAL_PROMPT（過去分含む）："+prompt)
# 言語モデル（OpenAIのチャットモデル）のラッパークラスをインスタンス化
llm = ChatOpenAI(
model = "gpt-3.5-turbo",
openai_api_key=env.get_env_variable('OPEN_AI_KEY'),
max_tokens=500,
temperature=0.5
)
# APIを使用して、応答を生成します
# モデルにPrompt（入力）を与えCompletion（出力）を取得する
# SystemMessage: OpenAIに事前に連携したい情報。キャラ設定や前提知識など。
# HumanMessage: OpenAIに聞きたい質問
response = llm(messages=[HumanMessage(content=prompt)])
# 文字列から必要なSystemMessageのみを抽出（以下例で言うaaaaaの部分のみ）
# [SystemMessage(content='aaaaa', additional_kwargs={}), HumanMessage(content='bbbbb', additional_kwargs={}, example=False)]
pattern = r"SystemMessage\(content='([^']*)'"
matches = re.findall(pattern, response.content)
content = ""
if matches:
content = matches[0]
else:
content = response.content
print("============ generate_response : COMPLETION："+str(response.content))
return content
●やり取りに関する情報を保持するクラス：conversation_util.py
import re
from langchain import schema
from slack_sdk import WebClient
class ConversationInfo:
"""
親クラス
入力データ（会話履歴＋System Prompt+Human Prompt）をクラス
開発者は当クラスを継承し、状況に応じたInput情報生成を行う。
"""
def __init__(
self,
messages:list[schema.BaseMessage]=None,
**kwargs) -> None:
# 「_変数」は非公開orプライベート変数
self._messages = messages
# 以降、@property(Decorator)で、ゲッター（getter）メソッドを定義。
# 以降、@messages.setter(Decorator)で、セッター（setter）メソッドを定義。
@property
def messages(self) -> list[schema.BaseMessage]:
"""
メッセージ履歴のgetter
Returns:
list: メッセージ履歴のリスト
"""
return self._messages
@messages.setter
def messages(self, value:list[schema.BaseMessage]):
"""
メッセージ履歴のsetter
Args:
value (list[schema.BaseMessage]): メッセージ履歴
"""
self._messages = value
def build_messages(self, **kwargs):
"""
messagesを構築するための処理を実装するメソッド。
必須でなく、メンバ変数「messages」を直接セットする事も可能
"""
pass
def remove_mention_str(remove_user_id:str, target_text:str) -> str:
"""
文頭/文末のボットへのメンション文字列"<@Uxxxx>"を除外する
Args:
remove_user_id (str): 除外したいボットのユーザーID
target_text (str): 除外を行いたいテキスト文字列
Returns:
str: メンション文字列が除外された新しいテキスト文字列
"""
# 例：^<@UAAAAAAAAAA.*?>|<@UAAAAAAAAAA.*?>$
pattern = f'^<@{remove_user_id}.*?>|<@{remove_user_id}.*?>$'
# target_textのうち、パターンにマッチした部分を除去
ret = re.sub(pattern, '', target_text.strip())
return ret
def get_thread_history(
client:WebClient,
bot_user_id:str,
ts: str,
thread_ts:str,
channel:str) ->list:
"""
スレッドメッセージの履歴も含めた、一連のやり取りの履歴を取得する。
（親メッセージ＋conversation_repliesのlimitの件数まで取得）
Args:
bot_user_id (str): ボットのユーザーID。メッセージが"ai" or "human"を判定するために必要。
client (WebClient) :Slackとインタラクトする為のWebClientクラスのインスタンス
bot_user_id (str) :ボットのユーザーID
ts (str) :ユーザーメッセージのタイムスタンプ
thread_ts (str) :ユーザーメッセージがスレッドの一部である場合の親メッセージのtimestamp
channel (str) :チャンネルID
Returns:
list: schema.HumanMessage または schema.SystemMessageが格納されたlist
"""
history = []
# 会話履歴を取得
resp = client.conversations_replies(
channel=channel, inclusive=False, latest=ts, ts=thread_ts)
messages = resp.get("messages")
# ボットの応答はAIMessageに、人からの回答はHumanMessageに格納
for message in messages:
if message.get("user", "-") == bot_user_id:
content=message.get("text", "**** メッセージの取得に失敗しました ****")
history.append(schema.AIMessage(content=content))
else:
content=message.get("text", "**** メッセージの取得に失敗しました ****")
history.append(schema.HumanMessage(content=content))
return history
class ConversationInfoSlack(ConversationInfo):
"""
TaskInputChatを継承した子クラス
Slackからのメッセージ(処理依頼)に対して、メッセージ情報を保持するためのクラス
"""
def __init__(self,
client:WebClient,
bot_user_id:str,
ts:str,
thread_ts:str,
channel:str,
user:str,
human_message_latest:schema.HumanMessage=None,
messages:list[schema.BaseMessage]=None,
system_message:schema.SystemMessage=None,
**kwargs) -> None:
"""
コンストラクタ。
Args:
client (WebClient) :Slackとインタラクトする為のWebClientクラスのインスタンス
bot_user_id (str) :ボットのユーザーID
ts (str) :ユーザーメッセージのタイムスタンプ
thread_ts (str) :ユーザーメッセージがスレッドの一部である場合の親メッセージのtimestamp
channel (str) :チャンネルID
user (str) :投稿したユーザーのSlackユーザーID
human_message_latest (schema.HumanMessage) : 最新のユーザーメッセージ (省略可能)
messages (list[schema.BaseMessage])
:メッセージ履歴 (省略された場合は、後からsetterから格納するか、build_messages()から構築)
schema.BaseMessageクラスのインスタンスのlistを受け取ることができます(Noneのため省略可能)
system_message (schema.SystemMessage) : システムメッセージ (省略可能)
Notes:
- system_message、human_message_latestを設定する事で、build_messages()を呼び、過去スレッド履歴も含めたmessagesが構築できる。
- build_messages()を使用せずに、直接メンバ変数messagesに値を格納する事も可能。
"""
# 親クラスのコンストラクタを呼ぶ
super().__init__(messages=messages,**kwargs)
self.client = client
self.bot_user_id = bot_user_id
self.ts = ts
self.thread_ts = thread_ts
self.channel = channel
self.user = user
self.human_message_latest = human_message_latest
self.system_message = system_message
def build_messages(self, **kwargs):
"""
messagesを構築するための処理を実装するメソッド。
必須でなく、メンバ変数「messages」を直接セットする事も可能
"""
ret = []
# システムメッセージを追加
if not self.system_message is None:
ret.append(self.system_message)
# スレッド履歴を追加
if self.thread_ts is not None:
thread_history = get_thread_history(
client=self.client,
bot_user_id=self.bot_user_id,
ts=self.ts,
thread_ts=self.thread_ts,
channel=self.channel
)
ret.extend(thread_history)
# 最新のメッセージを追加
ret.append(self.human_message_latest)
# ボットへの文頭/文末のメンションを除去する
# Before例: <@U03U3JE514N>\n私は赤色の携帯が欲しいです。
# After例: '私は赤色の携帯が欲しいです。'
for message in ret:
message.content = remove_mention_str(self.bot_user_id, message.content)
self._messages = ret
●環境変数：env.py
def get_env_variable(key):
env_variable_dict = {
# ------------------------------------
# Azure Form Recognizer(RDL)
# ------------------------------------
"FR_ENDPOINT" : "",
"FR_KEY" : "",
# ------------------------------------
# OpenAI
# ------------------------------------
"OPEN_AI_KEY" : "s",
"MODEL" : "text-davinci-003",
"SYSTEM_MESSAGE" : "あなたはxxxトです。xxxxしてください。",
# ------------------------------------
# App名：Slack_Python_Flask
# ------------------------------------
# BotユーザーID
"BOT_USER_ID" : "",
# Botトークン（Flask）
"WEBAPPS_SLACK_TOKEN" : "",
"WEBAPPS_SIGNING_SECRET" : "",
# Botトークン（ソケットモード）
"SOCK_SLACK_BOT_TOKEN" : "",
"SOCK_SLACK_APP_TOKEN" : ""
}
ret_val = env_variable_dict.get(key, None)
return ret_val
動かし方
・①venvを作る
> python -m venv .venv
（図２１１）
↓
↓
・②venvのアクティベート
> .venv/Scripts/activate
（図２１２）
↓
↓
・③パッケージのインストール（requirements.txt）
> pip install -r requirements.txt
（図２１３）
↓
↓
・④メインモジュール実行（0774_app.py）
（図２１４）
↓
・⑤Slackに投稿
（図１００）
