はじめに

以前OpenAI APIを利用したSlackのChatGPTボットアプリを作成しました。

今回はGCP Vertex AIのPython SDKを使って似たようなアプリをつくってみようと思います。

9/20時点でVertex AIの言語生成AIが突然日本語に対応してたので記事を修正してます

処理仕様

おおまかな処理仕様はOpenAI版とほぼ同じとなります。

今回はポイントとなるところのみを説明していきます。
動作環境は同じGCPのCloud Runを利用します。

Vertex AIを利用するためにはGCPのアカウントが必要です。
APIと連携するためにアプリ用のサービスアカウントを作成し「Vertex AI ユーザー」ロールを付与します。
サービスアカウントをCloud Runで作成したアプリに割り当ててください。
GCP以外の環境で動作させるにはサービスアカウントの鍵ファイルを作成してGOOGLE_APPLICATION_CREDENTIALSという環境変数に鍵ファイルのパスを指定することでAPIを動作させることができました。

処理内容

今回必要なPythonライブラリは以下となります。

pip install google-cloud-aiplatform
# pip install googletrans==3.1.0a0

~~googletransはなぜかバージョン指定をする必要があったので・・・ということで~~

import vertexai

from vertexai.preview.language_models import ChatModel, ChatMessage
# from googletrans import Translator

今回はテキストチャットのAPIを利用します。
会話履歴の取得と送信が必要になるのでSlackのスレッドから履歴を取得します。
~~プロンプトが日本語に対応していないので翻訳ライブラリを利用して「日本語→英語」と「英語→日本語」の変換を行います。~~

スレッド情報取得

    message_history = []
    input = ''
    # スレッドモードの場合スレッド内の発言をAIに送る
    if thread_ts != '':
        payload = {'token':APP_TOKEN_PALM, 'channel':channel_id, 'ts': thread_ts}
        res = requests.post("https://slack.com/api/conversations.replies", data=payload)
        res_json = res.json()
        print(res_json)
        messages = res_json['messages']
        for message in messages:

            # アプリへのメンションの場合
            if APP_USER_ID_PALM in message['text']:
                # 制御文字を抜く
                input = message['text'].replace(APP_USER_ID_PALM, '')
                # 何もない場合は無視する
                if input == '':
                    continue
                # 全て無条件に英語に訳す
                # input = traslate_ja_to_en(input)
                q_message = ChatMessage(author="user", content=input)
                message_history.append(q_message)
            # AIの回答の場合
            elif APP_USER_ID_PALM_RAW in message['user']:
                input = message['text']
                # 全て無条件に英語に訳す
                # input = traslate_ja_to_en(input)
                a_message = ChatMessage(author="bot", content=input)
                message_history.append(a_message)

会話履歴の作成処理です。
Slackからスレッドの内容を全て取得しユーザの質問とAIの回答のみを抽出していきます。
ChatMessageというオブジェクトを作成ユーザーの質問とAIボットの回答を交互に配列へ格納していきます。
~~スレッドから取得したメッセージをgoogletransライブラリにて全て英語に翻訳します。~~

#def traslate_ja_to_en(text):
#    translator = Translator()
#    translated = translator.translate(text, src='ja', dest='en')
#    print(translated)
#    return translated.text

#def traslate_en_to_ja(text):
#    translator = Translator()
#    translated = translator.translate(text, src='en', dest='ja')
#    print(translated)
#    return translated.text

プロンプト送信＆回答をSlackに投稿

    text = text.replace(APP_USER_ID_PALM, '')

    # 全て無条件に英語に訳す
    # text = traslate_ja_to_en(text)

    # 一番最後のメッセージは質問としてAIに送るため削除する
    message_history.pop()

    print(message_history)
    print(text)

    vertexai.init(project="GCPのプロジェクト名", location="us-central1")
    chat_model = ChatModel.from_pretrained("chat-bison@001")
    parameters = {
        "temperature": 0.2,
        "max_output_tokens": 1024,
        "top_p": 0.8,
        "top_k": 40
    }

    answer = ''

    try:
        chat = chat_model.start_chat(message_history=message_history)
        response = chat.send_message(text, **parameters)
        print(f"Response from Model: {response.text}")
        answer = response.text
    except Exception as e:
        answer = f"An exception occurred: {e}"
        print(answer)

    # 全て無条件に日本語に訳す
    # answer = traslate_en_to_ja(answer)

先ほど作成した会話履歴はmessage_historyへセットします。
データ構造は「ユーザ、ボット、ユーザ、ボット、、、」の組み合わせの順番になる必要があるのですが
スレッド内の全てのメッセージを取得しているため最後のユーザの発言をpop()関数で外します。

会話履歴データの異常エラーとしてデータ数は偶数になってないといけないとか交互になっていないなど例外が飛んできます

There should be odd number of messages for correct alternating turn.
Messages must alternate between authors.

Slackから送信されたユーザの発言自体はtextに格納していますので~~それを英訳して~~APIに送信します。
~~AIから戻ってきたanswerを和訳してからSlackに送信する感じになります。~~

おわりに

GCPのVertex AIコンソールプレイグラウンド上でサンプルソースも簡単に出力できるのでこれは楽勝かなと思いきや
履歴の積み上げサンプルが公式リファレンス上全く情報として見つけられなかったで最後は結局SDKのコードを読んで発見した感じですね。
https://github.com/googleapis/python-aiplatform