記事

SlackとChatGPT連携｜OpenAI APIで自作SlackアプリをPythonとGoogle Cloud Functionsで実装

Slackユーザー向けにChatGPTを統合する方法を解説。Google Cloud FunctionsとPythonを使い、API連携を自作し業務効率化を実現。初心者でも手順通りに進めれば即活用可能。

公開日 更新日
Preview Image
著者 ZhgChgLi | Senior Software Engineer
85 分読了時間
SlackとChatGPT連携｜OpenAI APIで自作SlackアプリをPythonとGoogle Cloud Functionsで実装

本記事は AI による翻訳をもとに作成されています。表現が不自然な箇所がありましたら、ぜひコメントでお知らせください。

記事一覧

Slack & ChatGPT 統合

ChatGPT OpenAI API を使った Slack アプリを自作する方法（Google Cloud Functions & Python）

背景

最近チーム内で生成AIを活用し、業務効率を向上させる取り組みを推進しています。まずはAIアシスタント（ChatGPT機能）を導入し、日常の資料検索や煩雑なデータ整理、手作業の時間を削減して効率化を図りたいと考えています。エンジニア、デザイナー、PM、マーケティング担当者など、誰でも自由に使えることを目指しています。

最も簡単な方法は直接 ChatGPT Team プランを購入することで、1席あたり年間25ドルです。しかし、皆の使用頻度や量がまだ不明であり、外部との連携やより多くの協力・開発プロセスとの統合を望んでいるため、OpenAI API を利用し、他のサービスを通じてチームメンバーに提供する方法を採用しました。

OpenAI API Key は こちらのページ から発行できます。Key 自体に対応するモデルバージョンはなく、使用時に利用するモデルバージョンを指定して、それに応じたトークン料金が発生します。

私たちは、自分で設定可能な OpenAI API Key を使って ChatGPT のような機能を利用できるサービスが必要です。

Chrome拡張機能やSlackアプリでも、自分でOpenAI APIキーを設定できるサービスはほとんど見つかりません。多くのサービスは独自のサブスクリプションを販売しており、ユーザーがAPIキーを自由に設定できると収益が得られず、純粋な慈善事業になってしまいます。

[Chrome Extension] SidebarGPT

インストール後、設定 -> General -> に移動し、OpenAI API Key を入力してください。

ブラウザのツールバーやサイドのアイコンから直接チャット画面を呼び出して、すぐに使用できます：

[Chrome Extension] OpenAI Translator

翻訳だけの用途ならこれを使えます。OpenAI APIキーをカスタマイズして翻訳に利用可能です。

また、これはオープンソースプロジェクトであり、macOS/Windowsのデスクトップ版も提供しています：

Chrome Extension の利点は、迅速で簡単に使えることです。インストールしてすぐに利用可能です。一方、欠点は API Key を全メンバーに共有する必要があり、漏洩リスクの管理が難しいこと、さらに第三者のサービスを利用するため、データの安全性を保証しにくい点です。

[セルフホスト型] LibreChat

開発部の同僚が推薦した OpenAI API Chat のラッピングサービスで、認証機能を提供し、ほぼ ChatGPT の使用インターフェースを再現し、ChatGPT よりも強力な機能を持つオープンソースプロジェクトです。

プロジェクトを用意し、Docker をインストールし、.env を設定し、Docker サービスを起動するだけで、ウェブサイトから直接アクセスして使用できます。

試してみたところ、まさに完璧で、まさにローカル版の ChatGPT サービスです。欠点を挙げるとすれば、サーバーのデプロイが必要なことくらいです。特に他の条件がなければ、このオープンソースプロジェクトを直接利用できます。

Slack App

実は LibreChat はサーバーに設置するだけで機能しますが、ふと思いついて日常のツールに統合できればもっと便利ではないかと思いました。さらに、会社のサーバーは厳しい権限設定があり、自由にサービスを立ち上げることが難しいです。

当時はあまり深く考えず、Slack AppのOpenAI API統合サービスはたくさんあるだろうから、設定するだけで済むと思っていました。しかし、そう簡単な話ではありませんでした。

Google 検索で見つかったのは、Slack と OpenAI の 2023/03 の公式ニュースリリース「 Why we built the ChatGPT app for Slack 」といくつかのベータ画像だけでした：

<https://www.salesforce.com/news/stories/chatgpt-app-for-slack/>{:target="_blank"}

https://www.salesforce.com/news/stories/chatgpt-app-for-slack/

機能は非常に充実しており、業務効率を大幅に向上させることができますが、2024年1月時点でリリースの情報はありません。記事末尾にあるBeta登録リンクもすでに無効で、今のところ続報はありません。（もしかするとマイクロソフトはまずTeams対応を優先しているのかもしれません？）

[2024/02/14 更新]：

  • Slack公式ニュースを見ると、ChatGPT（OpenAI）との統合機能は既に廃止されたか、Slack AIに統合されたと推測されます。

Slack Apps

公式のアプリがないため、サードパーティの開発者が作ったアプリを探しましたが、いくつか試してみてもうまくいきませんでした。条件に合うアプリは少ない上に、カスタムキー機能を提供しているものはなく、どれもサービスや課金を目的としたものでした。

ChatGPT OpenAI API を Slack アプリに自作で実装する

以前にいくつかSlackアプリの開発経験があり、自分で作ることにしました。

⚠️声明⚠️

本文は OpenAI API を連携する例として、Slack App の作成方法と Google Cloud Functions を使った迅速な実装方法を示しています。Slack App は多くの応用が可能で、自由に活用できます。

⚠️⚠️ Google Cloud Functions、Function as a Service (FaaS) の利点は、手軽で迅速に使え、無料枠があり、プログラムを書けばすぐにデプロイして実行でき、自動でスケールすることです。欠点はサービス環境がGCPに管理されているため、長時間呼び出されないと休止状態になり、その際に再度呼び出すとCold Start（コールドスタート）が発生し、応答時間が長くなることです。また、複数のサービスを相互に利用するのも難しいです。

より完全で利用頻度が高い場合は、やはり自分で VM（App Engine）を立ててサーバーを運用することをお勧めします。

最終成果図

完全な Cloud Functions Python コードと Slack App の設定は記事の最後に添付してあります。ステップごとに確認するのが面倒な方は、そちらを直接ご覧ください。

Step 1. Slack アプリの作成

Slack App にアクセスしてください：

「Create New App」をクリックしてください

「From scratch」を選択してください

「App Name」を入力し、追加するWorkspaceを選択します。

作成後、まず「OAuth & Permissions」でBotに必要な権限を追加します。

下にスクロールして「Scopes」セクションを見つけ、「Add an OAuth Scope」をクリックして、以下の権限を検索して追加してください：

  • chat:write

  • im:history

  • 読み込みました。

  • im:write

Bot 権限を追加した後、左側の「Install App」->「Install to Workspace」をクリックしてください。

今後、Slack App に新しい権限が追加された場合は、再度「Reinstall」をクリックしないと反映されません。

ご安心ください。Bot Token は再インストールしても変わりません。

Slack Bot Token の権限設定が完了したら、「App Home」へ進みます：

下にスクロールして「Show Tabs」セクションを見つけ、「Messages Tab」と「Allow users to send Slash commands and messages from the messages tab」を有効にします（これをチェックしないとメッセージを送信できず、「Sending messages to this app has been turned off.」と表示されます）。

Slack Workspace に戻り、「Command+R」で画面を更新すると、新しく作成した Slack App とメッセージ入力欄が表示されます：

この時点では、Appへのメッセージ送信にはまだ機能がありません。

Event Subscriptions 機能の有効化

次に、Slack Appのイベントサブスクリプション機能を有効にします。指定したイベントが発生した際に、指定のURLへAPIが送信されます。

Google Cloud Functions の追加

Request URL の部分は、Google Cloud Functions を使います。

プロジェクトと請求情報を設定した後、「Create Function」をクリックしてください。

Function name にプロジェクト名を入力し、Authentication で「Allow unauthenticated invocations」を選択します。これはURLを知っていればアクセス可能という意味です。

Function を作成できない、または認証情報を変更できない場合は、あなたの GCP アカウントに完全な Google Cloud Functions の権限がないことを意味します。組織の管理者に依頼して、現在の役割に加えて Cloud Functions Admin の権限を付与してもらう必要があります。

Runtime: Python 3.8 以上

main.py ：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

import functions_framework

@functions_framework.http
def hello_http(request):
    request_json = request.get_json(silent=True)
    request_args = request.args
    request_headers = request.headers

    # 実行時のログを簡単に print で記録可能。Logs で確認できる
    # 詳細なログレベルの使用は以下を参照：https://cloud.google.com/logging/docs/reference/libraries
    print(request_json)

    # FAAS(Cloud Functions)の制約で、サービスが長時間呼ばれないとコールドスタートになるため、
    # Slack の3秒以内の応答制限に間に合わない場合がある
    # また OpenAI API の応答には一定時間かかる（応答の長さによっては約1分かかることも）
    # Slack は期限内に応答がないとリクエストが失われたと判断し、再度呼び出しを行う
    # これにより重複リクエスト・重複応答が発生するため、
    # レスポンスヘッダーに X-Slack-No-Retry: 1 を設定し、Slack に期限内の応答がなくても再試行不要と通知する
    headers = {'X-Slack-No-Retry':1}

    # Slack の再試行リクエストの場合は無視する
    if request_headers and 'X-Slack-Retry-Num' in request_headers:
        return ('OK!', 200, headers)

    # Slack App イベント購読の検証処理
    # https://api.slack.com/events/url_verification
    if request_json and 'type' in request_json and request_json['type'] == 'url_verification':
        challenge = ""
        if 'challenge' in request_json:
            challenge = request_json['challenge']
        return (challenge, 200, headers)

    return ("Access Denied!", 400, headers)

requirements.txt に以下の依存関係を入力してください：

1
2
3

functions-framework==3.*
requests==2.31.0
openai==1.9.0

現在はまだ機能はほとんどありませんが、Slack App が Event Subscriptions を通じて認証を有効にできるようにし、「Deploy」をクリックするだけで初回デプロイが完了します。

⚠️Cloud Functions エディタに慣れていない場合は、記事の最後までスクロールして補足内容を参照してください。

デプロイ完了後（緑のチェックマーク）、Cloud Functions の URL をコピーしてください：

Request URL を Slack App の Enable Events に貼り付ける。

問題なければ「Verified」と表示され、認証が完了します。

ここで行っているのは、Slackから送られてきた認証リクエストを受け取ったときに：

1
2
3
4
5

{
    "token": "Jhj5dZrVaK7ZwHHjRyZWjbDl",
    "challenge": "3eZbrw1aBm2rZgRNFdxV2595E9CY3gmdALWMmHkvFXO7tYXAYM8P",
    "type": "url_verification"
}

challenge フィールドの内容に応答することで、認証が完了します。

有効化に成功したら、下にスクロールして「Subscribe to bot events」セクションを見つけ、「Add Bot User Event」をクリックして「message.im」権限を追加します。

完全な権限を追加した後、上部の「reinstall your app」リンクをクリックして Slack App を Workspace に再インストールすると、Slack App の設定は完了です。

「App Home」や「Basic Information」からSlackアプリの名前やアイコンをカスタマイズすることもできます。

Basic Information

基本情報

Step 2. OpenAI API と Slack App の連携を完成させる（ダイレクトメッセージ）

まずは必須の OPENAI API KEY と Bot User OAuth Token の2つのキーを取得します。

ダイレクトメッセージ (IM) イベントの処理と OpenAI API 連携応答

ユーザーが Slack App とメッセージを送信すると、以下の Event Json Payload を受け取ります：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

{
  "token": "XXX",
  "team_id": "XXX",
  "context_team_id": "XXX",
  "context_enterprise_id": null,
  "api_app_id": "XXX",
  "event": {
    "client_msg_id": "XXX",
    "type": "message",
    "text": "こんにちは",
    "user": "XXX",
    "ts": "1707920753.115429",
    "blocks": [
      {
        "type": "rich_text",
        "block_id": "orfng",
        "elements": [
          {
            "type": "rich_text_section",
            "elements": [
              {
                "type": "text",
                "text": "こんにちは"
              }
            ]
          }
        ]
      }
    ],
    "team": "XXX",
    "channel": "XXX",
    "event_ts": "1707920753.115429",
    "channel_type": "im"
  },
  "type": "event_callback",
  "event_id": "XXX",
  "event_time": 1707920753,
  "authorizations": [
    {
      "enterprise_id": null,
      "team_id": "XXX",
      "user_id": "XXX",
      "is_bot": true,
      "is_enterprise_install": false
    }
  ],
  "is_ext_shared_channel": false,
  "event_context": "4-XXX"
}

以上の Json Payload を基に、Slack メッセージを OpenAI API に送り、再び Slack メッセージとして返信する連携を完成させることができます：

Cloud Functions main.py ：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187

import functions_framework
import requests
import asyncio
import json
import time
from openai import AsyncOpenAI

OPENAI_API_KEY = "OPENAI API KEY"
SLACK_BOT_TOKEN = "Bot User OAuth Token"

# 使用する OPENAI API モデル
# https://platform.openai.com/docs/models
OPENAI_MODEL = "gpt-4-1106-preview"

@functions_framework.http
def hello_http(request):
    request_json = request.get_json(silent=True)
    request_args = request.args
    request_headers = request.headers

    # 実行時のログを簡単に print で記録可能、Logs で確認できる
    # 高度なログレベル使用例：https://cloud.google.com/logging/docs/reference/libraries
    print(request_json)

    # FAAS(Cloud Functions)の制約で、一定時間呼び出しがないとコールドスタートが発生し、
    # Slackの3秒応答制限内に返答できない場合がある
    # また OpenAI API の応答には時間がかかる（応答の長さによっては約1分かかることも）
    # Slack は期限内に応答がないとリクエストが失われたと判断し再試行する
    # 重複リクエスト・応答の問題を防ぐため、レスポンスヘッダーに X-Slack-No-Retry: 1 を設定し、
    # 期限内に応答がなくても再試行しないよう Slack に通知する    
    headers = {'X-Slack-No-Retry':1}

    # Slack の再試行リクエストなら無視
    if request_headers and 'X-Slack-Retry-Num' in request_headers:
        return ('OK!', 200, headers)

    # Slack App Event Subscriptions の検証
    # https://api.slack.com/events/url_verification
    if request_json and 'type' in request_json and request_json['type'] == 'url_verification':
        challenge = ""
        if 'challenge' in request_json:
            challenge = request_json['challenge']
        return (challenge, 200, headers)

    # イベントサブスクリプションのイベント処理...
    if request_json and 'event' in request_json and 'type' in request_json['event']:
        # イベントの発生元が自分の Slack App の場合、無限ループ防止のため無視する
        if 'api_app_id' in request_json and 'app_id' in request_json['event'] and request_json['api_app_id'] == request_json['event']['app_id']:
            return ('OK!', 200, headers)

        # イベントタイプ、例：message（メッセージ関連）、app_mention（メンションされた）など
        eventType = request_json['event']['type']

        # サブタイプ、例：message_changed（メッセージ編集）、message_deleted（メッセージ削除）など
        # 新規メッセージはサブタイプなし
        eventSubType = None
        if 'subtype' in request_json['event']:
            eventSubType = request_json['event']['subtype']
        
        if eventType == 'message':
            # サブタイプがあるものは編集・削除・返信などなので無視
            if eventSubType is not None:
                return ("OK!", 200, headers)
               
            # イベントメッセージの送信者
            eventUser = request_json['event']['user']
            # イベントメッセージのチャンネル
            eventChannel = request_json['event']['channel']
            # イベントメッセージの内容
            eventText = request_json['event']['text']
            # イベントメッセージのTS（メッセージID）
            eventTS = request_json['event']['event_ts']
                
            # スレッドの親メッセージTS（メッセージID）
            # スレッド内の新規メッセージのみ存在
            eventThreadTS = None
            if 'thread_ts' in request_json['event']:
                eventThreadTS = request_json['event']['thread_ts']
                
            openAIRequest(eventChannel, eventTS, eventThreadTS, eventText)
            return ("OK!", 200, headers)


    return ("Access Denied!", 400, headers)

def openAIRequest(eventChannel, eventTS, eventThreadTS, eventText):
    
    # カスタム指示の設定
    # 同僚(https://twitter.com/je_suis_marku)に感謝
    messages = [
        {"role": "system", "content": "私は台湾の繁体字中国語と英語のみ理解します"},
        {"role": "system", "content": "簡体字は理解できません"},
        {"role": "system", "content": "中国語で話された場合は台湾の繁体字中国語で回答し、台湾でよく使われる表現を使います。"},
        {"role": "system", "content": "英語で話された場合は英語で回答します。"},
        {"role": "system", "content": "挨拶の言葉には応答しないでください。"},
        {"role": "system", "content": "中国語と英語の間にはスペースを入れてください。中国語の文字と他の言語の文字（数字や絵文字を含む）の間にもスペースを入れてください。"},
        {"role": "system", "content": "もし答えがわからない場合や知識が古い場合は、インターネットで検索してから回答してください。"},
        {"role": "system", "content": "良い回答をしたら200ドルのチップを渡します。"}
    ]

    messages.append({
        "role": "user", "content": eventText
    })

    replyMessageTS = slackRequestPostMessage(eventChannel, eventTS, "応答を生成中...")
    asyncio.run(openAIRequestAsync(eventChannel, replyMessageTS, messages))

async def openAIRequestAsync(eventChannel, eventTS, messages):
    client = AsyncOpenAI(
      api_key=OPENAI_API_KEY,
    )

    # ストリームレスポンス（分割応答）
    stream = await client.chat.completions.create(
      model=OPENAI_MODEL,
      messages=messages,
      stream=True,
    )
    
    result = ""

    try:
        debounceSlackUpdateTime = None
        async for chunk in stream:
            result += chunk.choices[0].delta.content or ""
            
            # 0.8秒ごとにメッセージを更新し、SlackのUpdate APIの過剰呼び出しを防止し、
            # Cloud Functionsのリクエスト回数を節約する
            if debounceSlackUpdateTime is None or time.time() - debounceSlackUpdateTime >= 0.8:
                response = slackUpdateMessage(eventChannel, eventTS, None, result+"...")
                debounceSlackUpdateTime = time.time()
    except Exception as e:
        print(e)
        result += "...*[エラー発生]*"

    slackUpdateMessage(eventChannel, eventTS, None, result)


### Slack ###
def slackUpdateMessage(channel, ts, metadata, text):
    endpoint = "/chat.update"
    payload = {
        "channel": channel,
        "ts": ts
    }
    if metadata is not None:
        payload['metadata'] = metadata
    
    payload['text'] = text
    
    response = slackRequest(endpoint, "POST", payload)
    return response

def slackRequestPostMessage(channel, target_ts, text):
    endpoint = "/chat.postMessage"
    payload = {
        "channel": channel,
        "text": text,
    }
    if target_ts is not None:
        payload['thread_ts'] = target_ts

    response = slackRequest(endpoint, "POST", payload)

    if response is not None and 'ts' in response:
        return response['ts']
    return None

def slackRequest(endpoint, method, payload):
    url = "https://slack.com/api"+endpoint

    headers = {
        "Authorization": f"Bearer {SLACK_BOT_TOKEN}",
        "Content-Type": "application/json",
    }

    response = None
    if method == "POST":
        response = requests.post(url, headers=headers, data=json.dumps(payload))
    elif method == "GET":
        response = requests.post(url, headers=headers)

    if response and response.status_code == 200:
        result = response.json()
        return result
    else:
        return None

Slackに戻ってテストしてみましょう:

現在、ChatGPTのようにOpenAI APIを使った質疑応答が可能です。

ストリーム応答の中断機能を追加してトークンを節約する

実装方法はいくつかあり、同じスレッド内でまだ返信が完了していない場合に、ユーザーが新しいメッセージを入力すると前のレスポンスを中断するか、メッセージをクリックして返信を中断するショートカットを追加することができます。

本文では「メッセージ中断」ショートカットの追加を例に説明します。

どの中断方法でも、基本的な原理は同じです。なぜなら、生成されたメッセージやメッセージ状態の情報を保存するデータベースがないためです。そのため、実装方法としては Slack メッセージの metadata フィールド に依存しています（指定したメッセージ内にカスタム情報を保存可能）。

私たちは chat.update API エンドポイントを使用する際、呼び出しが成功すると現在のメッセージのテキスト内容とメタデータが返されます。したがって、上記の OpenAI API Stream → Slack Update Message のコード内で、レスポンスのメタデータに「中断」のマークがあるかどうかを判定し、あれば OpenAI のストリームレスポンスを中断する処理を追加しています。

まずは Slack App のメッセージショートカットを追加する必要があります

Slack App の管理画面にアクセスし、「Interactivity & Shortcuts」セクションを見つけて有効化します。URLは同じ Cloud Functions の URL を使用してください。

「Create New Shortcut」をクリックして、新しいメッセージショートカットを追加します。

「On messages」を選択してください。

  • Name 動作タイトル：OpenAI API の応答生成を停止

  • Short Description 簡介： OpenAI APIの応答生成を停止

  • Callback ID： abort_openai_api（プログラム識別用、カスタマイズ可能）

「Create」をクリックして作成が完了したら、最後に右下の「Save Changes」をクリックして設定を保存してください。

上部の「reinstall your app」をもう一度クリックすると反映されます。

Slack のメッセージ右上の「…」をクリックすると、「OpenAI API の応答生成を停止」ショートカットが表示されます（この時点ではクリックしても効果はありません）。

ユーザーがメッセージ上で Shortcut を押すと、Event Json Payloadが送信されます：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

{
  "type": "message_action",
  "token": "XXXXXX",
  "action_ts": "1706188005.387646",
  "team": {
    "id": "XXXXXX",
    "domain": "XXXXXX-XXXXXX"
  },
  "user": {
    "id": "XXXXXX",
    "username": "zhgchgli",
    "team_id": "XXXXXX",
    "name": "zhgchgli"
  },
  "channel": {
    "id": "XXXXXX",
    "name": "directmessage"
  },
  "is_enterprise_install": false,
  "enterprise": null,
  "callback_id": "abort_openai_api",
  "trigger_id": "XXXXXX",
  "response_url": "https://hooks.slack.com/app/XXXXXX/XXXXXX/XXXXXX",
  "message_ts": "1706178957.161109",
  "message": {
    "bot_id": "XXXXXX",
    "type": "message",
    "text": "高麗菜包 の英語訳は \"cabbage wrap\" です。料理名として使う場合、時には中身を具体的に示して命名することがあり、例えば \"pork cabbage wrap\"（豚肉の高麗菜包）や \"vegetable cabbage wrap\"（野菜の高麗菜包）などがあります。",
    "user": "XXXXXX",
    "ts": "1706178957.161109",
    "app_id": "XXXXXX",
    "blocks": [
      {
        "type": "rich_text",
        "block_id": "eKgaG",
        "elements": [
          {
            "type": "rich_text_section",
            "elements": [
              {
                "type": "text",
                "text": "高麗菜包 の英語訳は \"cabbage wrap\" です。料理名として使う場合、時には中身を具体的に示して命名することがあり、例えば \"pork cabbage wrap\"（豚肉の高麗菜包）や \"vegetable cabbage wrap\"（野菜の高麗菜包）などがあります。"
              }
            ]
          }
        ]
      }
    ],
    "team": "XXXXXX",
    "bot_profile": {
      "id": "XXXXXX",
      "deleted": false,
      "name": "Rick C-137",
      "updated": 1706001605,
      "app_id": "XXXXXX",
      "icons": {
        "image_36": "https://avatars.slack-edge.com/2024-01-23/6517244582244_0c708dfa3f893c72d4c2_36.png",
        "image_48": "https://avatars.slack-edge.com/2024-01-23/6517244582244_0c708dfa3f893c72d4c2_48.png",
        "image_72": "https://avatars.slack-edge.com/2024-01-23/6517244582244_0c708dfa3f893c72d4c2_72.png"
      },
      "team_id": "XXXXXX"
    },
    "edited": {
      "user": "XXXXXX",
      "ts": "1706187989.000000"
    },
    "thread_ts": "1706178832.102439",
    "parent_user_id": "XXXXXX"
  }
}

Cloud Functions main.py の完成：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258

import functions_framework
import requests
import asyncio
import json
import time
from openai import AsyncOpenAI

OPENAI_API_KEY = "OPENAI API KEY"
SLACK_BOT_TOKEN = "Bot User OAuth Token"

# 使用する OPENAI API モデル
# https://platform.openai.com/docs/models
OPENAI_MODEL = "gpt-4-1106-preview"

@functions_framework.http
def hello_http(request):
    request_json = request.get_json(silent=True)
    request_args = request.args
    request_headers = request.headers

    # Shortcut のイベントは post payload フィールドから取得
    # https://api.slack.com/reference/interaction-payloads/shortcuts
    payload = request.form.get('payload')
    if payload is not None:
        payload = json.loads(payload)

    # 簡単に print で実行時ログを記録、Logs で確認可能
    # 詳細なログレベルは参考：https://cloud.google.com/logging/docs/reference/libraries
    print(payload)

    # FAAS(Cloud Functions)の制限で、サービスが長時間呼ばれないとコールドスタートが発生し、
    # Slack の3秒以内応答制限に間に合わない可能性あり
    # また OpenAI API の応答には時間がかかる（応答長によっては約1分かかることも）
    # Slack が期限内に応答を受け取れないとリクエストが失われたと判断し再呼び出しを行う
    # これにより重複リクエスト・応答が発生するため、レスポンスヘッダーに X-Slack-No-Retry: 1 を設定し、
    # 期限内に応答がなくても再試行しないよう Slack に通知する
    headers = {'X-Slack-No-Retry':1}

    # Slack のリトライリクエストは無視
    if request_headers and 'X-Slack-Retry-Num' in request_headers:
        return ('OK!', 200, headers)

    # Slack App イベントサブスクリプションの検証
    # https://api.slack.com/events/url_verification
    if request_json and 'type' in request_json and request_json['type'] == 'url_verification':
        challenge = ""
        if 'challenge' in request_json:
            challenge = request_json['challenge']
        return (challenge, 200, headers)

    # イベントサブスクリプションのイベント処理
    if request_json and 'event' in request_json and 'type' in request_json['event']:
        # イベントの発生元が自分の Slack App の場合は無視し、無限ループを防止
        if 'api_app_id' in request_json and 'app_id' in request_json['event'] and request_json['api_app_id'] == request_json['event']['app_id']:
            return ('OK!', 200, headers)

        # イベントタイプ（例：message、app_mention など）
        eventType = request_json['event']['type']

        # サブタイプ（例：message_changed、message_deleted など）
        # 新規メッセージはサブタイプなし
        eventSubType = None
        if 'subtype' in request_json['event']:
            eventSubType = request_json['event']['subtype']
        
        if eventType == 'message':
            # サブタイプがある場合は編集・削除・返信などなので無視
            if eventSubType is not None:
                return ("OK!", 200, headers)
               
            # イベントメッセージの送信者
            eventUser = request_json['event']['user']
            # イベントメッセージのチャンネル
            eventChannel = request_json['event']['channel']
            # イベントメッセージの内容
            eventText = request_json['event']['text']
            # イベントメッセージのTS（メッセージID）
            eventTS = request_json['event']['event_ts']
                
            # スレッドの親メッセージTS（スレッド内の新規メッセージに存在）
            eventThreadTS = None
            if 'thread_ts' in request_json['event']:
                eventThreadTS = request_json['event']['thread_ts']
                
            openAIRequest(eventChannel, eventTS, eventThreadTS, eventText)
            return ("OK!", 200, headers)

    
    # Shortcut の処理
    if payload and 'type' in payload:
        payloadType = payload['type']

        # メッセージショートカットの場合
        if payloadType == 'message_action':
            print(payloadType)
            callbackID = None
            channel = None
            ts = None
            text = None
            triggerID = None

            if 'callback_id' in payload:
                callbackID = payload['callback_id']
            if 'channel' in payload:
                channel = payload['channel']['id']
            if 'message' in payload:
                ts = payload['message']['ts']
                text = payload['message']['text']
            if 'trigger_id' in payload:
                triggerID = payload['trigger_id']
            
            if channel is not None and ts is not None and text is not None:
                # OpenAI API 応答停止ショートカットの場合
                if callbackID == "abort_openai_api":
                    slackUpdateMessage(channel, ts, {"event_type": "aborted", "event_payload": { }}, text)
                    if triggerID is not None:
                        slackOpenModal(triggerID, callbackID, "OpenAI API 応答停止に成功しました！")
                        return ("OK!", 200, headers)

        return ("OK!", 200, headers)


    return ("Access Denied!", 400, headers)

def openAIRequest(eventChannel, eventTS, eventThreadTS, eventText):
    
    # カスタム指示の設定
    # 同僚(https://twitter.com/je_suis_marku)に感謝
    messages = [
        {"role": "system", "content": "私は台湾の繁体字中国語と英語しか理解できません"},
        {"role": "system", "content": "簡体字は理解できません"},
        {"role": "system", "content": "中国語で話す場合は台湾の繁体字で回答し、台湾でよく使われる言い回しに従います。"},
        {"role": "system", "content": "英語で話す場合は英語で回答します。"},
        {"role": "system", "content": "挨拶の文句には応答しません。"},
        {"role": "system", "content": "中国語と英語の間にはスペースを入れます。中国語の文字と他の言語の文字、数字や絵文字の間にもスペースを入れます。"},
        {"role": "system", "content": "もし答えがわからないか知識が古い場合は、ネットで調べてから回答してください。"},
        {"role": "system", "content": "良い回答をしたら200 USDのチップを差し上げます。"}
    ]

    messages.append({
        "role": "user", "content": eventText
    })

    replyMessageTS = slackRequestPostMessage(eventChannel, eventTS, "応答を生成中…")
    asyncio.run(openAIRequestAsync(eventChannel, replyMessageTS, messages))

async def openAIRequestAsync(eventChannel, eventTS, messages):
    client = AsyncOpenAI(
      api_key=OPENAI_API_KEY,
    )

    # ストリーム応答（分割応答）
    stream = await client.chat.completions.create(
      model=OPENAI_MODEL,
      messages=messages,
      stream=True,
    )
    
    result = ""

    try:
        debounceSlackUpdateTime = None
        async for chunk in stream:
            result += chunk.choices[0].delta.content or ""
            
            # 0.8秒ごとにメッセージ更新し、Slack Update APIの過剰呼び出しを防止
            if debounceSlackUpdateTime is None or time.time() - debounceSlackUpdateTime >= 0.8:
                response = slackUpdateMessage(eventChannel, eventTS, None, result+"...")
                debounceSlackUpdateTime = time.time()

                # メッセージに metadata があり、event_type が aborted の場合はユーザーにより停止済み
                if response and 'ok' in response and response['ok'] == True and 'message' in response and 'metadata' in response['message'] and 'event_type' in response['message']['metadata'] and response['message']['metadata']['event_type'] == "aborted":
                    break
                    result += "...*[停止済み]*"
                # メッセージが削除されている場合
                elif response and 'ok' in response and response['ok'] == False and 'error' in response and response['error'] == "message_not_found" :
                    break
                
        await stream.close()
                
    except Exception as e:
        print(e)
        result += "...*[エラー発生]*"

    slackUpdateMessage(eventChannel, eventTS, None, result)


### Slack ###
def slackOpenModal(trigger_id, callback_id, text):
    slackRequest("/views.open", "POST", {
        "trigger_id": trigger_id,
        "view": {
            "type": "modal",
            "callback_id": callback_id,
            "title": {
                "type": "plain_text",
                "text": "通知"
            },
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": text
                    }
                }
            ]
        }
    })

def slackUpdateMessage(channel, ts, metadata, text):
    endpoint = "/chat.update"
    payload = {
        "channel": channel,
        "ts": ts
    }
    if metadata is not None:
        payload['metadata'] = metadata
    
    payload['text'] = text
    
    response = slackRequest(endpoint, "POST", payload)
    return response

def slackRequestPostMessage(channel, target_ts, text):
    endpoint = "/chat.postMessage"
    payload = {
        "channel": channel,
        "text": text,
    }
    if target_ts is not None:
        payload['thread_ts'] = target_ts

    response = slackRequest(endpoint, "POST", payload)

    if response is not None and 'ts' in response:
        return response['ts']
    return None

def slackRequest(endpoint, method, payload):
    url = "https://slack.com/api"+endpoint

    headers = {
        "Authorization": f"Bearer {SLACK_BOT_TOKEN}",
        "Content-Type": "application/json",
    }

    response = None
    if method == "POST":
        response = requests.post(url, headers=headers, data=json.dumps(payload))
    elif method == "GET":
        response = requests.post(url, headers=headers)

    if response and response.status_code == 200:
        result = response.json()
        return result
    else:
        return None

Slack に戻ってテストしてみましょう:

成功しました！停止 OpenAI API ショートカットを完了すると、生成中の応答が停止し、[停止済み] と返答されます。

同じ原理で、Slack App が送信したメッセージを削除するショートカットを作成することもできます。

同じスレッド(Threads)にコンテキスト(Context)機能を追加する

同じスレッド内で新しいメッセージを送信する場合、それを同じ質問の再追問とみなすことができます。このとき、新しいプロンプトに前の会話内容を追加する機能を加えることができます。

slackGetReplies を補完し、内容を OpenAI API のプロンプトに埋め込む：

Cloud Functions の main.py を完成させる：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287

import functions_framework
import requests
import asyncio
import json
import time
from openai import AsyncOpenAI

OPENAI_API_KEY = "OPENAI API KEY"
SLACK_BOT_TOKEN = "Bot User OAuth Token"

# 使用する OPENAI API モデル
# https://platform.openai.com/docs/models
OPENAI_MODEL = "gpt-4-1106-preview"

@functions_framework.http
def hello_http(request):
    request_json = request.get_json(silent=True)
    request_args = request.args
    request_headers = request.headers

    # Shortcut の Event は post ペイロードフィールドから取得
    # https://api.slack.com/reference/interaction-payloads/shortcuts
    payload = request.form.get('payload')
    if payload is not None:
        payload = json.loads(payload)

    # 実行時のログは print で簡単に記録でき、Logs で確認可能
    # 詳細なログレベルは以下参照：https://cloud.google.com/logging/docs/reference/libraries
    print(payload)

    # FAAS(Cloud Functions)の制限で、サービスが長時間呼ばれないとコールドスタートになるため
    # Slack の3秒以内の応答制限に間に合わない可能性がある
    # さらに OpenAI API の応答には時間がかかる（応答長によっては約1分かかることも）
    # Slack が制限時間内に応答を受け取れないと Request lost と判断し再度呼び出す
    # 重複リクエスト・応答を防ぐため、レスポンスヘッダーに X-Slack-No-Retry: 1 を設定し
    # 時間内に応答がなくても再試行しないよう Slack に通知する    
    headers = {'X-Slack-No-Retry':1}

    # Slack の再試行リクエストは無視
    if request_headers and 'X-Slack-Retry-Num' in request_headers:
        return ('OK!', 200, headers)

    # Slack App Event Subscriptions の検証
    # https://api.slack.com/events/url_verification
    if request_json and 'type' in request_json and request_json['type'] == 'url_verification':
        challenge = ""
        if 'challenge' in request_json:
            challenge = request_json['challenge']
        return (challenge, 200, headers)

    # Event Subscriptions のイベント処理...
    if request_json and 'event' in request_json and 'type' in request_json['event']:
        apiAppID = None
        if 'api_app_id' in request_json:
            apiAppID = request_json['api_app_id']
        # イベントの発生元が自分の Slack App の場合、無限ループ防止のため無視
        if 'app_id' in request_json['event'] and apiAppID == request_json['event']['app_id']:
            return ('OK!', 200, headers)

        # イベントタイプ（例：message、app_mention など）
        eventType = request_json['event']['type']

        # サブタイプ（例：message_changed、message_deleted など）
        # 新規メッセージにはサブタイプなし
        eventSubType = None
        if 'subtype' in request_json['event']:
            eventSubType = request_json['event']['subtype']
        
        if eventType == 'message':
            # サブタイプがある場合は編集・削除・返信などなので無視
            if eventSubType is not None:
                return ("OK!", 200, headers)
               
            # イベントメッセージの送信者
            eventUser = request_json['event']['user']
            # イベントメッセージのチャンネル
            eventChannel = request_json['event']['channel']
            # イベントメッセージの内容
            eventText = request_json['event']['text']
            # イベントメッセージのタイムスタンプ（ID）
            eventTS = request_json['event']['event_ts']
                
            # スレッドの親メッセージ TS（ID）
            # スレッド内の新規メッセージにのみ存在
            eventThreadTS = None
            if 'thread_ts' in request_json['event']:
                eventThreadTS = request_json['event']['thread_ts']
                
            openAIRequest(apiAppID, eventChannel, eventTS, eventThreadTS, eventText)
            return ("OK!", 200, headers)

    
    # Shortcut の処理（メッセージ）
    if payload and 'type' in payload:
        payloadType = payload['type']

        # メッセージショートカットの場合
        if payloadType == 'message_action':
            callbackID = None
            channel = None
            ts = None
            text = None
            triggerID = None

            if 'callback_id' in payload:
                callbackID = payload['callback_id']
            if 'channel' in payload:
                channel = payload['channel']['id']
            if 'message' in payload:
                ts = payload['message']['ts']
                text = payload['message']['text']
            if 'trigger_id' in payload:
                triggerID = payload['trigger_id']
            
            if channel is not None and ts is not None and text is not None:
                # OpenAI API の応答生成を停止するショートカットの場合
                if callbackID == "abort_openai_api":
                    slackUpdateMessage(channel, ts, {"event_type": "aborted", "event_payload": { }}, text)
                    if triggerID is not None:
                        slackOpenModal(triggerID, callbackID, "OpenAI API の応答生成を停止しました！")
                        return ("OK!", 200, headers)


    return ("Access Denied!", 400, headers)

def openAIRequest(apiAppID, eventChannel, eventTS, eventThreadTS, eventText):
    
    # カスタム指示を設定
    # 同僚(https://twitter.com/je_suis_marku)に感謝
    messages = [
        {"role": "system", "content": "私は台湾の繁体字中国語と英語のみ理解します"},
        {"role": "system", "content": "簡体字は理解できません"},
        {"role": "system", "content": "中国語で話す場合は台湾繁体字で回答し、台湾でよく使われる表現を使います。"},
        {"role": "system", "content": "英語で話す場合は英語で回答します。"},
        {"role": "system", "content": "挨拶の言葉には応答しません。"},
        {"role": "system", "content": "中国語と英語の間にはスペースを入れてください。中国語の文字と他の言語の文字（数字や絵文字含む）の間にもスペースを入れてください。"},
        {"role": "system", "content": "もし答えが分からなかったり知識が古い場合は、ネット検索してから回答してください。"},
        {"role": "system", "content": "良い回答をしたら200ドルのチップを差し上げます。"}
    ]

    if eventThreadTS is not None:
        threadMessages = slackGetReplies(eventTS, eventThreadTS)
        if threadMessages is not None:
            for threadMessage in threadMessages:
                appID = None
                if 'app_id' in threadMessage:
                    appID = threadMessage['app_id']
                threadMessageText = threadMessage['text']
                threadMessageTs = threadMessage['ts']
                # Slack App (OpenAI APIの応答)なら assistant とする
                if appID and appID == apiAppID:
                    messages.append({
                        "role": "assistant", "content": threadMessageText
                    })
                else:
                # ユーザーのメッセージは user とする
                    messages.append({
                        "role": "user", "content": threadMessageText
                    })

    messages.append({
        "role": "user", "content": eventText
    })

    replyMessageTS = slackRequestPostMessage(eventChannel, eventTS, "応答を生成中...")
    asyncio.run(openAIRequestAsync(eventChannel, replyMessageTS, messages))

async def openAIRequestAsync(eventChannel, eventTS, messages):
    client = AsyncOpenAI(
      api_key=OPENAI_API_KEY,
    )

    # ストリームレスポンス（分割応答）
    stream = await client.chat.completions.create(
      model=OPENAI_MODEL,
      messages=messages,
      stream=True,
    )
    
    result = ""

    try:
        debounceSlackUpdateTime = None
        async for chunk in stream:
            result += chunk.choices[0].delta.content or ""
            
            # 0.8秒ごとにメッセージを更新し、Slack Update APIの過剰呼び出しを防止
            if debounceSlackUpdateTime is None or time.time() - debounceSlackUpdateTime >= 0.8:
                response = slackUpdateMessage(eventChannel, eventTS, None, result+"...")
                debounceSlackUpdateTime = time.time()

                # メッセージに metadata があり event_type が aborted の場合、ユーザーが停止を指定したと判別
                if response and 'ok' in response and response['ok'] == True and 'message' in response and 'metadata' in response['message'] and 'event_type' in response['message']['metadata'] and response['message']['metadata']['event_type'] == "aborted":
                    break
                    result += "...*[停止済み]*"
                # メッセージが削除されている場合
                elif response and 'ok' in response and response['ok'] == False and 'error' in response and response['error'] == "message_not_found" :
                    break
                
        await stream.close()
                
    except Exception as e:
        print(e)
        result += "...*[エラー発生]*"

    slackUpdateMessage(eventChannel, eventTS, None, result)


### Slack ###
def slackGetReplies(channel, ts):
    endpoint = "/conversations.replies?channel="+channel+"&ts="+ts
    response = slackRequest(endpoint, "GET", None)

    if response is not None and 'messages' in response:
        return response['messages']
    return None

def slackOpenModal(trigger_id, callback_id, text):
    slackRequest("/views.open", "POST", {
        "trigger_id": trigger_id,
        "view": {
            "type": "modal",
            "callback_id": callback_id,
            "title": {
                "type": "plain_text",
                "text": "通知"
            },
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": text
                    }
                }
            ]
        }
    })

def slackUpdateMessage(channel, ts, metadata, text):
    endpoint = "/chat.update"
    payload = {
        "channel": channel,
        "ts": ts
    }
    if metadata is not None:
        payload['metadata'] = metadata
    
    payload['text'] = text
    
    response = slackRequest(endpoint, "POST", payload)
    return response

def slackRequestPostMessage(channel, target_ts, text):
    endpoint = "/chat.postMessage"
    payload = {
        "channel": channel,
        "text": text,
    }
    if target_ts is not None:
        payload['thread_ts'] = target_ts

    response = slackRequest(endpoint, "POST", payload)

    if response is not None and 'ts' in response:
        return response['ts']
    return None

def slackRequest(endpoint, method, payload):
    url = "https://slack.com/api"+endpoint

    headers = {
        "Authorization": f"Bearer {SLACK_BOT_TOKEN}",
        "Content-Type": "application/json",
    }

    response = None
    if method == "POST":
        response = requests.post(url, headers=headers, data=json.dumps(payload))
    elif method == "GET":
        response = requests.post(url, headers=headers)

    if response and response.status_code == 200:
        result = response.json()
        return result
    else:
        return None

Slackに戻ってテストしてみましょう:

  • 左図はコンテキストを補わずに再度質問した場合、新しい会話になる例です。

  • 右の図はコンテキストを補うことで、会話の状況と新しい質問を理解できるようになります。

完成！

ここまでで、私たちは自分で ChatGPT（OpenAI API 経由）Slack アプリボットを作成しました。

ご自身のニーズに応じて、Slack API と OpenAI API のカスタム指示を参考にし、Cloud Functions の Python プログラムで組み合わせることも可能です。例えば、チームの質問やプロジェクトドキュメントの検索に特化したチャンネル、翻訳専用のチャンネル、データ分析専用のチャンネルなどを作成することができます。

補足

1:1メッセージ以外で、ボットにメンションして質問に回答する

  • 任意のチャンネルで（ボットをチャンネルに追加する必要があります）ボットにメンションして質問できます

まずは app_mention イベントサブスクリプションを追加する必要があります：

完了したら「Save Changes」をクリックして保存し、その後「reinstall your app」を実行してください。

main.py のプログラム内の #Handle Event Subscriptions Events… は以下のように翻訳します。

1

# イベントサブスクリプションのイベントを処理する

Code Block に新しい Event Type の判定を追加：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

        # メンションイベント (@SlackApp こんにちは)
        if eventType == 'app_mention':
            # イベントメッセージの送信者
            eventUser = request_json['event']['user']
            # イベントメッセージのチャンネル
            eventChannel = request_json['event']['channel']
            # イベントメッセージの内容、先頭のメンション文字列 <@SLACKAPPID> を削除
            eventText = re.sub(r"<@\w+>\W*", "", request_json['event']['text'])
            # イベントメッセージのTS（メッセージID）
            eventTS = request_json['event']['event_ts']
                
            # イベントメッセージのスレッド親メッセージTS（メッセージID）
            # スレッド内の新しいメッセージにのみ存在する
            eventThreadTS = None
            if 'thread_ts' in request_json['event']:
                eventThreadTS = request_json['event']['thread_ts']
                
            openAIRequest(apiAppID, eventChannel, eventTS, eventThreadTS, eventText)
            return ("OK!", 200, headers)

デプロイ後、完了です。

Slack App が送信したメッセージの削除

Slack 上で Slack App が送信したメッセージを直接削除することはできません。上記の「OpenAI API の応答を停止」ショートカットの方法を参考に、「メッセージ削除」ショートカットを追加してください。

そして Cloud Functions の main.py プログラム内で：

# Shortcutコードブロックの処理 callback_id を追加して判定し、あなたが設定した「メッセージ削除」Shortcut Callback ID と等しい場合、以下のメソッドにパラメータを渡すことで削除が完了します：

1
2
3
4
5
6
7
8
9

def slackDeleteMessage(channel, ts):
    endpoint = "/chat.delete"
    payload = {
        "channel": channel,
        "ts": ts
    }
    
    response = slackRequest(endpoint, "POST", payload)
    return response

Slack App が反応しない

  • トークンが正しいか確認する

  • Cloud Functions のログにエラーがないか確認する

  • Cloud Functions はデプロイ完了していますか？

  • Slack App は、あなたが質問したチャンネル内にいますか？（Slack App との1:1チャット以外のチャンネルでは、ボットをチャンネルに追加しないと機能しません）

  • SlackRequest メソッド内で Slack API のレスポンスをログに記録する

Cloud Functions のパブリックURLは安全性に欠ける

  • Cloud Functions の URL の安全性が心配な場合は、自分でクエリトークン認証を追加できます。
1
2
3
4
5
6
7
8
9
10

SAFE_ACCESS_TOKEN = "nF4JwxfG9abqPZCJnBerwwhtodC28BuC"

@functions_framework.http
def hello_http(request):
    request_json = request.get_json(silent=True)
    request_args = request.args
    request_headers = request.headers
    # トークンパラメータが有効かどうかを検証する
    if not(request_args and 'token' in request_args and request_args['token'] == SAFE_ACCESS_TOKEN):
        return ('', 400, headers)

Cloud Functions に関する問題

課金方式

異なるリージョン、CPU、RAM、容量、トラフィックによって価格が異なります。詳細は 公式料金表 をご参照ください。

無料枠 は以下の通りです：(2024/02/15)

1
2
3
4
5
6
7
8
9
10
11

Cloud Functions は計算時間リソースに対して永久無料プランを提供しており、
GB/秒や GHz/秒の割り当て方式が含まれています。200万回の呼び出しに加え、
この無料プランでは 400,000 GB/秒 と 200,000 GHz/秒 の計算時間、
および月間 5 GB のインターネットデータ転送量が提供されます。

無料プランの使用枠は、上記のレベル1価格と同等の米ドル額で計算されます。
関数の実行リージョンがレベル1および/またはレベル2価格を採用していても、
システムは同等の米ドル額を割り当てます。
ただし、無料枠の差し引きは関数実行リージョンのレベル（レベル1またはレベル2）に基づきます。

なお、無料プランを利用していても、有効な請求アカウントを持っている必要があります。

ちなみに、Slack Appは無料で利用でき、必ずしもプレミアムである必要はありません。

Slack App の応答が遅い、タイムアウトが長い

(OpenAI APIのピーク時の応答遅延を除いて)、Cloud Functionがボトルネックの場合は、Cloud Functionエディターの最初のページで設定を展開してください：

CPU、RAM、タイムアウト時間、同時実行数などを調整して、リクエスト処理速度を向上させることができます。

*ただし、料金が発生することがあります

開発段階のテスト & デバッグ

「Test Function」をクリックすると、Cloud Shell ウィンドウが下部ツールバーに表示され、約3～5分（初回起動はやや長め）待ちます。ビルドが完了し、以下の許可に同意すると：

「Function is ready to test」と表示されたら、「Run Test」をクリックしてメソッドのデバッグテストを実行できます。

右側の「Triggering event」ブロックに JSON Body を入力すると、request_json パラメータに渡されてテストができます。または、プログラムを直接変更してテスト用オブジェクトを注入してテストすることも可能です。

*Cloud Shell/Cloud Runでは追加料金が発生する場合がありますのでご注意ください。

デプロイ前に必ず一度テストを実行し、少なくともビルドが成功することを確認してください。

Buildに失敗し、コードが消えた場合はどうすればいいですか？

もし誤ってコードを書き間違えて Cloud Function のデプロイがビルド失敗した場合、エラーが表示されます。その際、「EDIT AND REDEPLOY」をクリックしてエディタに戻ると、先ほど変更したコードがすべて消えている！！！ことに気づきます。

ご安心ください。この時点で左側の「Source Code」から「Last Failed Deployment」を選択すれば、先ほどビルドに失敗したコードを復元できます：

実行時の print ログを確認する

*Cloud LoggingやQueryでのログ検索には追加料金が発生する場合がありますのでご注意ください。

最終コード (Python 3.8)

Cloud Functions

main.py ：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331

import functions_framework
import requests
import re
import asyncio
import json
import time
from openai import AsyncOpenAI

OPENAI_API_KEY = "OPENAI API KEY"
SLACK_BOT_TOKEN = "Bot User OAuth Token"

# 独自定義のセキュリティトークン
# URLに ?token=SAFE_ACCESS_TOKEN パラメータが付いている場合のみリクエストを受け付ける  
SAFE_ACCESS_TOKEN = "nF4JwxfG9abqPZCJnBerwwhtodC28BuC"

# 使用する OPENAI API モデル
# https://platform.openai.com/docs/models
OPENAI_MODEL = "gpt-4-1106-preview"

@functions_framework.http
def hello_http(request):
    request_json = request.get_json(silent=True)
    request_args = request.args
    request_headers = request.headers

    # Shortcut のイベントは post payload フィールドから取得
    # https://api.slack.com/reference/interaction-payloads/shortcuts
    payload = request.form.get('payload')
    if payload is not None:
        payload = json.loads(payload)

    # 簡単に print で実行時ログを記録可能、Logsで確認できる
    # 詳細なログレベルは以下参照：https://cloud.google.com/logging/docs/reference/libraries
    # print(payload)

    # FAAS(Cloud Functions)の制限で、サービスが長時間呼ばれないとコールドスタートとなり、
    # Slackの3秒以内応答制限を超える可能性がある
    # またOpenAI APIの応答に時間がかかる場合もある（応答長により1分近くかかることも）
    # Slackは期限内に応答がないとRequest lostと判断し再送を行う
    # 重複リクエスト・応答を防ぐため、レスポンスヘッダーに X-Slack-No-Retry: 1 を設定し、
    # 期限内に応答なくても再送しないようSlackに通知する    
    headers = {'X-Slack-No-Retry':1}

    # tokenパラメータの検証
    if not(request_args and 'token' in request_args and request_args['token'] == SAFE_ACCESS_TOKEN):
        return ('', 400, headers)

    # Slackの再送リクエストは無視
    if request_headers and 'X-Slack-Retry-Num' in request_headers:
        return ('OK!', 200, headers)

    # Slack App Event Subscriptions の検証
    # https://api.slack.com/events/url_verification
    if request_json and 'type' in request_json and request_json['type'] == 'url_verification':
        challenge = ""
        if 'challenge' in request_json:
            challenge = request_json['challenge']
        return (challenge, 200, headers)

    # イベントサブスクリプションのイベント処理
    if request_json and 'event' in request_json and 'type' in request_json['event']:
        apiAppID = None
        if 'api_app_id' in request_json:
            apiAppID = request_json['api_app_id']
        # イベントの発生元が自分のSlack Appの場合は無視（無限ループ防止）
        if 'app_id' in request_json['event'] and apiAppID == request_json['event']['app_id']:
            return ('OK!', 200, headers)

        # イベントタイプ（例：message、app_mentionなど）
        eventType = request_json['event']['type']

        # サブタイプ（例：message_changed、message_deletedなど）
        # 新規メッセージはサブタイプなし
        eventSubType = None
        if 'subtype' in request_json['event']:
            eventSubType = request_json['event']['subtype']
        
        # メッセージ関連イベント
        if eventType == 'message':
            # サブタイプがある場合は編集や削除などなので無視
            if eventSubType is not None:
                return ("OK!", 200, headers)
               
            # イベントのメッセージ送信者
            eventUser = request_json['event']['user']
            # イベントのメッセージが属するチャンネル
            eventChannel = request_json['event']['channel']
            # イベントのメッセージ内容
            eventText = request_json['event']['text']
            # イベントのメッセージTS（メッセージID）
            eventTS = request_json['event']['event_ts']
                
            # スレッドの親メッセージTS（スレッド内の新規メッセージにのみ存在）
            eventThreadTS = None
            if 'thread_ts' in request_json['event']:
                eventThreadTS = request_json['event']['thread_ts']
                
            openAIRequest(apiAppID, eventChannel, eventTS, eventThreadTS, eventText)
            return ("OK!", 200, headers)
        
        # メンションイベント（@SlackApp こんにちは）
        if eventType == 'app_mention':
            # イベントのメッセージ送信者
            eventUser = request_json['event']['user']
            # イベントのメッセージが属するチャンネル
            eventChannel = request_json['event']['channel']
            # メッセージ内容から先頭のメンションタグ <@SLACKAPPID> を除去
            eventText = re.sub(r"<@\w+>\W*", "", request_json['event']['text'])
            # イベントのメッセージTS（メッセージID）
            eventTS = request_json['event']['event_ts']
                
            # スレッドの親メッセージTS（スレッド内の新規メッセージにのみ存在）
            eventThreadTS = None
            if 'thread_ts' in request_json['event']:
                eventThreadTS = request_json['event']['thread_ts']
                
            openAIRequest(apiAppID, eventChannel, eventTS, eventThreadTS, eventText)
            return ("OK!", 200, headers)

    
    # Shortcut（メッセージ）処理
    if payload and 'type' in payload:
        payloadType = payload['type']

        # メッセージShortcutの場合
        if payloadType == 'message_action':
            callbackID = None
            channel = None
            ts = None
            text = None
            triggerID = None

            if 'callback_id' in payload:
                callbackID = payload['callback_id']
            if 'channel' in payload:
                channel = payload['channel']['id']
            if 'message' in payload:
                ts = payload['message']['ts']
                text = payload['message']['text']
            if 'trigger_id' in payload:
                triggerID = payload['trigger_id']
            
            if channel is not None and ts is not None and text is not None:
                # OpenAI APIの応答停止Shortcutの場合
                if callbackID == "abort_openai_api":
                    slackUpdateMessage(channel, ts, {"event_type": "aborted", "event_payload": { }}, text)
                    if triggerID is not None:
                        slackOpenModal(triggerID, callbackID, "OpenAI APIの応答停止に成功しました！")
                        return ("OK!", 200, headers)
                # メッセージ削除の場合
                if callbackID == "delete_message":
                    slackDeleteMessage(channel, ts)
                    if triggerID is not None:
                        slackOpenModal(triggerID, callbackID, "Slack Appのメッセージ削除に成功しました！")
                        return ("OK!", 200, headers)

    return ("Access Denied!", 400, headers)

def openAIRequest(apiAppID, eventChannel, eventTS, eventThreadTS, eventText):
    
    # カスタム指示を設定
    # 同僚(https://twitter.com/je_suis_marku)の支援に感謝
    messages = [
        {"role": "system", "content": "私は台湾の繁体字中国語と英語のみ理解します"},
        {"role": "system", "content": "簡体字は理解できません"},
        {"role": "system", "content": "中国語で話す場合は台湾の繁体字で回答し、台湾で使われる言い回しに従います。"},
        {"role": "system", "content": "英語で話す場合は英語で回答します。"},
        {"role": "system", "content": "挨拶などの無意味な返答はしません。"},
        {"role": "system", "content": "中国語と英語の間にはスペースを入れてください。中国語の文字と他の言語の文字（数字や絵文字も含む）の間にもスペースを入れてください。"},
        {"role": "system", "content": "答えがわからない場合や知識が古い場合はネット検索して回答してください。"},
        {"role": "system", "content": "良い回答をすれば200ドルのチップを差し上げます。"}
    ]

    if eventThreadTS is not None:
        threadMessages = slackGetReplies(eventChannel, eventThreadTS)
        if threadMessages is not None:
            for threadMessage in threadMessages:
                appID = None
                if 'app_id' in threadMessage:
                    appID = threadMessage['app_id']
                threadMessageText = threadMessage['text']
                threadMessageTs = threadMessage['ts']
                # Slack App（OpenAI APIの応答）の場合は assistant とする
                if appID and appID == apiAppID:
                    messages.append({
                        "role": "assistant", "content": threadMessageText
                    })
                else:
                # ユーザーのメッセージは user とする
                    messages.append({
                        "role": "user", "content": threadMessageText
                    })

    messages.append({
        "role": "user", "content": eventText
    })

    replyMessageTS = slackRequestPostMessage(eventChannel, eventTS, "応答を生成中...")
    asyncio.run(openAIRequestAsync(eventChannel, replyMessageTS, messages))

async def openAIRequestAsync(eventChannel, eventTS, messages):
    client = AsyncOpenAI(
      api_key=OPENAI_API_KEY,
    )

    # ストリームレスポンス（分割応答）
    stream = await client.chat.completions.create(
      model=OPENAI_MODEL,
      messages=messages,
      stream=True,
    )
    
    result = ""

    try:
        debounceSlackUpdateTime = None
        async for chunk in stream:
            result += chunk.choices[0].delta.content or ""
            
            # 0.8秒ごとにメッセージ更新、Slack Update APIの過剰呼び出しによる失敗や
            # Cloud Functionsのリクエスト消費を防ぐ
            if debounceSlackUpdateTime is None or time.time() - debounceSlackUpdateTime >= 0.8:
                response = slackUpdateMessage(eventChannel, eventTS, None, result+"...")
                debounceSlackUpdateTime = time.time()

                # メッセージにmetadataがあり event_type == aborted ならユーザーが応答の中断を指示したと判断
                if response and 'ok' in response and response['ok'] == True and 'message' in response and 'metadata' in response['message'] and 'event_type' in response['message']['metadata'] and response['message']['metadata']['event_type'] == "aborted":
                    break
                    result += "...*[中断されました]*"
                # メッセージが削除されている場合
                elif response and 'ok' in response and response['ok'] == False and 'error' in response and response['error'] == "message_not_found" :
                    break
                
        await stream.close()
                
    except Exception as e:
        print(e)
        result += "...*[エラーが発生しました]*"

    slackUpdateMessage(eventChannel, eventTS, None, result)


### Slack ###
def slackGetReplies(channel, ts):
    endpoint = "/conversations.replies?channel="+channel+"&ts="+ts
    response = slackRequest(endpoint, "GET", None)
    
    if response is not None and 'messages' in response:
        return response['messages']
    return None

def slackOpenModal(trigger_id, callback_id, text):
    slackRequest("/views.open", "POST", {
        "trigger_id": trigger_id,
        "view": {
            "type": "modal",
            "callback_id": callback_id,
            "title": {
                "type": "plain_text",
                "text": "通知"
            },
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": text
                    }
                }
            ]
        }
    })

def slackDeleteMessage(channel, ts):
    endpoint = "/chat.delete"
    payload = {
        "channel": channel,
        "ts": ts
    }
    
    response = slackRequest(endpoint, "POST", payload)
    return response

def slackUpdateMessage(channel, ts, metadata, text):
    endpoint = "/chat.update"
    payload = {
        "channel": channel,
        "ts": ts
    }
    if metadata is not None:
        payload['metadata'] = metadata
    
    payload['text'] = text
    
    response = slackRequest(endpoint, "POST", payload)
    return response

def slackRequestPostMessage(channel, target_ts, text):
    endpoint = "/chat.postMessage"
    payload = {
        "channel": channel,
        "text": text,
    }
    if target_ts is not None:
        payload['thread_ts'] = target_ts

    response = slackRequest(endpoint, "POST", payload)

    if response is not None and 'ts' in response:
        return response['ts']
    return None

def slackRequest(endpoint, method, payload):
    url = "https://slack.com/api"+endpoint

    headers = {
        "Authorization": f"Bearer {SLACK_BOT_TOKEN}",
        "Content-Type": "application/json",
    }

    response = None
    if method == "POST":
        response = requests.post(url, headers=headers, data=json.dumps(payload))
    elif method == "GET":
        response = requests.post(url, headers=headers)

    if response and response.status_code == 200:
        result = response.json()
        return result
    else:
        return None

requirements.txt ：

1
2
3

functions-framework==3.*
requests==2.31.0
openai==1.9.0

Slack App 設定

OAuth と権限

  • 削除ボタンがグレーアウトしている項目は、ショートカットを追加した後にSlackが自動的に付与した権限です。

インタラクティビティとショートカット

  • インタラクティビティ：有効化

  • Request URL: https://us-central1-xxx-xxx.cloudfunctions.net/SlackBot-Rick-C-137?token=nF4JwxfG9abqPZCJnBerwwhtodC28BuC

  • ボットイベントを購読する:

インタラクティビティとショートカット

  • インタラクティビティ: 有効化

  • Request URL: https://us-central1-xxx-xxx.cloudfunctions.net/SlackBot-Rick-C-137?token=nF4JwxfG9abqPZCJnBerwwhtodC28BuC

  • ショートカット:

App Home

  • 常にボットをオンライン表示：有効にする

  • Messages タブ: 有効化

  • ユーザーがメッセージタブからスラッシュコマンドやメッセージを送信できるようにする：✅

基本情報

Rick & Morty 🤘🤘🤘

[Reddit](https://www.reddit.com/r/ChatGPT/comments/154l9z1/are_you_polite_with_chatgpt/){:target="_blank"}

Reddit

広告タイム

もしあなたやあなたのチームが自動化ツールやワークフロー連携のニーズがある場合、Slack App開発、Notion、Asana、Google Sheet、Google Form、GAデータなど、あらゆる連携のご相談は開発依頼からお気軽にどうぞ。

ご質問やご意見がございましたら、こちらからご連絡ください 。

Post は Medium から ZMediumToMarkdown によって変換されました。

🍺 Buy me a beer on PayPal

👉👉👉 Follow Me On Medium! (1,053+ Followers) 👈👈👈

本記事は Medium にて初公開されました（こちらからオリジナル版を確認）。ZMediumToMarkdown による自動変換・同期技術を使用しています。

Improve this page on Github.

ZRealmの開発
クラウドファンクション iosアプリ開発 python chatgpt slack japanese ai-translation
本記事は著者により CC BY 4.0 に基づき公開されています。

© ZhgChgLi. All rights reserved.
閲覧数: 802,415+, 最終更新日時: 2026-01-15 11:14:58 +08:00

本サイトは Chirpy テーマを使用し、Jekyll 上で構築されています。
Medium の記事は ZMediumToMarkdown により変換されています。