OpenAIで返答するSlackボットで「会話履歴」を加味する方法

 

＜目次＞

OpenAIで返答するSlackボットで「会話履歴」を加味する方法
　やりたいこと
　前提条件
　概要
　サンプルプログラム
　動かし方

OpenAIで返答するSlackボットで「会話履歴」を加味する方法

やりたいこと

前提条件

概要

サンプルプログラム

●メイン処理：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

動かし方

> python -m venv .venv

> .venv/Scripts/activate

> pip install -r requirements.txt