ChatGPT APIのSlackボットに1言ずつ喋らせる。

はじめに

見ていただくのが早いです。

快適です。無限に遊んでしまう。。。

前回、ChatGPT APIの返答をウェブサイト版ChatGPTのように逐次受け取る方法を紹介しました。

これをそのままSlackボットにしたら非常に良い感じだったので、実装なんかをメモとして残しておきます。

早速実装

slack_boltで実装を行いました。
ボットの前準備などは、公式でも詳しく手順が載っていますのでその通りに。

Slash Commandsの登録

これは必須ではないですが、今回の実装例では記憶の削除とキャラクター設定にスラッシュコマンドを利用しています。

• Slack APIのページにて、左側メニューからSlash Commandsを開く。
• Create New Commandから「/reset」「/character」の2種類を登録する。説明などは適当で大丈夫です。

ChatGPTのAPIを叩く部分（前回の記事とほぼ同じ）

import openai

API_KEY = "sk-************************************************"


class ChatGPT:
    def __init__(self):
        openai.api_key = API_KEY

    def chat(
        self,
        text,
        messages=None,
        settings="",
        max_tokens=2000,
        temperature=1.0,
        top_p=0.1,
        presence_penalty=0.0,
        frequency_penalty=0.0,
    ):

        # やり取りの管理
        character = [{"role": "system", "content": settings}] if settings else []
        messages = messages if messages is not None else []
        messages.append({"role": "user", "content": text})

        # APIを叩く、streamをTrueに
        resp = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=character+messages,
            max_tokens=max_tokens,
            temperature=temperature,
            top_p=top_p,
            presence_penalty=presence_penalty,
            frequency_penalty=frequency_penalty,
            stream=True,
        )

        # 返答を受け取り、逐次yield
        response_text = ""
        for chunk in resp:
            if chunk:
                content = chunk["choices"][0]["delta"].get("content")
                if content:
                    response_text += content
                    yield content
        else:  #
            messages += [{"role": "assistant", "content": response_text}]

Slackボットの部分

from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
import time

BOT_TOKEN = "xoxb-*************-*************-************************"
APP_TOKEN = "xapp-*-***********-*************-****************************************************************"

app = App(token=BOT_TOKEN)
gpt = ChatGPT()
messages = {}
settings = {}

# ボットが追加されているチャンネルのすべての投稿に反応
@app.message(".")
def respond(message, say):
    global messages
    
    mes = say(text="...") # とりあえず一言
    ch = mes["channel"] # チャンネル
    
    # 履歴にないチャンネルの場合、履歴の初期化
    if not messages.get(ch):
        messages[ch] = []
    if not settings.get(ch):
        settings[ch] = ""

    last_update = -1 # 最後の更新時間
    text = last_post_text = "" # 返答と更新済みの内容を保存する変数。

    # 逐次返答を受け取る
    for sentence in gpt.chat(message['text'], messages[ch], settings[ch]):
        text += sentence
        if (time.time() - last_update) > .3: # あまり高頻度に更新すると逆に遅くなる（Slack側の問題）ため、0.3秒以上毎に更新
            last_update = time.time()
            last_post_text = text
            app.client.chat_update(channel=ch,
                                   ts=mes["ts"],
                                   text=text)
    if last_post_text != text: # 残った未更新部分を投稿
        app.client.chat_update(channel=mes["channel"], ts=mes["ts"], text=text)

    # 履歴は過去10回まで保存（tokenの肥大化防止）
    messages[ch] =  messages[ch][-10:]
        

# /reset コマンド（履歴とキャラ設定の削除）
@app.command('/reset')
def forget(ack, respond, command):
    ack()
    ch = command['channel_id']
    messages.pop(ch, -1)
    settings.pop(ch, -1)
    app.client.chat_postMessage(channel=ch, text='(ᐛ)＜ばなな')

# /character コマンド（キャラ設定の追加/更新）
@app.command('/character')
def forget(ack, respond, command):
    ack()
    ch = command['channel_id']
    settings[ch] = command['text']
    app.client.chat_postMessage(channel=ch, text=f"Character settings: {command['text']}")
        
    
if __name__ == "__main__":
    handler = SocketModeHandler(app, APP_TOKEN)
    handler.start()

実装例の使い方

• 上記プログラムをラズパイなりAWSなりで走らせる。テストでは普通のPCで問題ありません。
• SlackのボットとのDMやボットを追加したチャンネルで何か発言する。