Home Slack & ChatGPT Integration
Post
Cancel

Slack & ChatGPT Integration

Posted Feb 16, 2024 Updated Feb 18, 2024
Preview Image
By
63 min read

Slack & ChatGPT Integration

自行打造 ChatGPT OpenAI API for Slack App (Google Cloud Functions & Python)

背景

最近在團隊內推廣運用 Generative AI 提升工作效率，初步只希望達成 AI Assistant (ChatGPT 功能)，從減少日常資料查詢、整理繁瑣資料、手工處理資料的時間，以提升工作效率；希望是工程師、設計師、PM、行銷…都能自由使用的。

最簡單的方法就是直接買 ChatGPT Team 方案，一個席位 $25 美金一年；但由於尚不確定大家的使用頻率(量)及希望能在外來與更多協作、開發流程進行整合，所以改採用 OpenAI API 方式，再透過其他服務封裝整合供團隊成員使用。

OpenAI API Key 可從 此頁面產生 ，Key 沒有分對應的 Model 版本，使用時才需指定要使用的 Model 版本並產生對應的 Token 費用。

我們需要一個服務能自行設定 OpenAI API Key 並使用該 Key 進行類 ChatGPT 使用。

不管是 Chrome Extension 或 Slack App 都蠻難找到能自行設定 OpenAI API Key 的服務，大部分服務都是要賣他們自己的訂閱制，讓使用者自訂 API Key 等於賺不到錢純做慈善。

[Chrome Extension] SidebarGPT

安裝完後可到設定 -> General -> 填入 OpenAI API Key。

可從瀏覽器小工具欄、側邊 Icon 直接呼叫出聊天界面，直接使用：

[Chrome Extension] OpenAI Translator

如果只有翻譯需求可使用這個，能自訂 OpenAI API Key 用於翻譯。

另外他是 開源專案 ，並同時提供 macOS/Windows 桌面版程式：

Chrome Extension 的優點是快速簡單方便，直接裝直接使用；缺點是需要將 API Key 提供給所有成員，難以管控外洩問題，還有使用第三方服務也難以保證大家的資料安全。

[Self-hosted] LibreChat

研發部同事推薦的 OpenAI API Chat 封裝服務，提供身份認證使用及幾乎還原 ChatGPT 使用介面、功能比 ChatGPT 更強大的開源專案。

只需專案、裝好 Docker、設定好 .env、啟 Docker 服務就能直接透過網站連入使用。

試了一下簡直無懈可擊，就是本地版 ChatGPT 服務；要說缺點的話就只有需要伺服器部署服務吧；如果沒其他考量，可以直接使用此開源專案。

Slack App

其實 LibreChat 服務起起來放上伺服器就已經達成效果了，但是靈光一閃想說如果能整合在日常工具當中是不是更方便？加上公司伺服器有嚴格的權限設定，不太能隨意起服務。

當時也沒想太多，想說 Slack App 的 OpenAI API 整合服務應該很多，找一個設定一下就好；沒想到事情沒有那麼簡單。

Google 搜尋只找到一篇 Slack x OpenAI 2023/03 官方的新聞稿「 Why we built the ChatGPT app for Slack 」及一些 Beta 圖片：

[https://www.salesforce.com/news/stories/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/01 為止尚無釋出的消息，文末提供的 Beta 註冊連結 也已經失效，暫時沒有下文。(還是微軟想先讓 Teams 支援？)

[2024/02/14 Update]：

Slack Apps

因無官方的 App 轉而搜尋第三方開發者的 App，搜尋並試用了幾個都碰壁；符合的 App 不多就算了，也沒有一個是能提供自訂 Key 功能的，每個都是做來賣服務、賣錢的。

自行實現 ChatGPT OpenAI API for Slack App

之前有一些 Slack App 開發經驗，決定自己動手做。

⚠️聲明⚠️

本文是以串接 OpenAI API 為例演示如何建立 Slack App 和快速使用 Google Cloud Funtions 來達成需求，Slack App 有需多應用可做，大家可以自由發揮。

⚠️⚠️ Google Cloud Functions，Function as a Service (FaaS) 的優點是方便快速、有免費額度，程式寫好就能直接部署執行、自動擴充，缺點是服務環境由 GCP 控制，當服務太久沒被呼叫會進入休眠，此時再次呼叫會進入 Cold Start 冷啟動，需要較長的反應時間；另外也較難起多個服務互相使用。

更完整或使用需求量大的話還是建議自己起 VM (App Engine) 架 Server 跑服務。

最終成果圖

完整 Cloud Functions Python 程式碼、Slack App 設定已附在文末，懶得一步一步看的朋友可快速前往查閱。

Step 1. 建立 Slack App

前往 Slack App ：

點擊「Create New App」

選擇「From scratch」

輸入「App Name」、選擇要加入的 Workspace。

建立完成後先到「OAuth & Permissions」新增 Bot 需要的權限。

下滑找到「Scopes」區塊，點擊「Add an OAuth Scope」搜尋加入以下幾個權限：

  • chat:write
  • im:history
  • im:read
  • 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 的事件訂閱功能，當指定事件發生時會打 API 到指定 URL。

新增 Google Cloud Funtions

Request URL 的部分， Google Cloud Funtions 就要上場了。

設定好專案、帳單資訊後點擊「Create Function」

Function name 輸入專案名稱、Authentication 選擇「Allow unauthenticated invocations」意即知道網址就能存取。

如果你無法建立 Function 或無法更改 Authentication 代表你的 GCP 帳號沒有完整的 Google Cloud Functions 權限，需要請組織管理員在原本的身份之外額外加上 Cloud Functions Admin 的權限，才能使用。

Runtime: Python 3.8 or 更高

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

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 紀錄執行階段 Log，可在 Logs 中查看
    # 進階 Logging Level 使用可參考：https://cloud.google.com/logging/docs/reference/libraries
    print(request_json)

    # 受限 FAAS(Cloud Functions)，服務太久沒呼叫，再次呼叫會進入冷啟動的特性，可能無法在 Slack 規定的 3 秒內回應
    # 加上 OpenAI API 請求到答覆需要一定時間(依照回應長度可能要接近 1 分鐘才能結束)
    # Slack 在時限內收不到回應會認為 Request lost，Slack 會再次重複呼叫
    # 會造成重複請求、回應的問題，因此我們可以在 Response Headers 設置 X-Slack-No-Retry: 1 告知 Slack ，就算沒在時限內收到回應也不需 Retry    
    headers = {'X-Slack-No-Retry':1}

    # 如果是 Slack Retry 的請求...忽略
    if request_headers and 'X-Slack-Retry-Num' in request_headers:
        return ('OK!', 200, headers)

    # Slack App Event Subscriptions Verify
    # 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 App 的名稱與大頭貼。

Basic Information

Basic Information

Step 2. 完善 OpenAI API 與 Slack App 串接 (Direct 訊息)

首先我們先要取得必備的 OPENAI API KEY 與 Bot User OAuth Token 兩個 Key。

處理 Direct Message (IM) Event & 串接 OpenAI API 回應

當使用者在與 Slack App 訊息傳送時會收到以下 Event Json Paylod ：

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

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 Model
# 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 紀錄執行階段 Log，可在 Logs 中查看
    # 進階 Logging Level 使用可參考：https://cloud.google.com/logging/docs/reference/libraries
    print(request_json)

    # 受限 FAAS(Cloud Functions)，服務太久沒呼叫，再次呼叫會進入冷啟動的特性，可能無法在 Slack 規定的 3 秒內回應
    # 加上 OpenAI API 請求到答覆需要一定時間(依照回應長度可能要接近 1 分鐘才能結束)
    # Slack 在時限內收不到回應會認為 Request lost，Slack 會再次重複呼叫
    # 會造成重複請求、回應的問題，因此我們可以在 Response Headers 設置 X-Slack-No-Retry: 1 告知 Slack ，就算沒在時限內收到回應也不需 Retry    
    headers = {'X-Slack-No-Retry':1}

    # 如果是 Slack Retry 的請求...忽略
    if request_headers and 'X-Slack-Retry-Num' in request_headers:
        return ('OK!', 200, headers)

    # Slack App Event Subscriptions Verify
    # 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)

    # Handle Event Subscriptions Events...
    if request_json and 'event' in request_json and 'type' in request_json['event']:
        # 如果 Event 來源是 App 且 App ID == Slack App ID 代表是自己 Slack App 觸發的事件
        # 忽略不處理，否則會陷入無限循環 Slack App -> Cloud Functions -> Slack App -> Cloud Functions...
        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']

        # SubType，例如：message_changed(編輯訊息), message_deleted(刪除訊息)...
        # 新訊息無 Sub Type
        eventSubType = None
        if 'subtype' in request_json['event']:
            eventSubType = request_json['event']['subtype']
        
        if eventType == 'message':
            #  有 Sub Type 的都是訊息編輯、刪除、被回應...
            # 忽略不處理
            if eventSubType is not None:
                return ("OK!", 200, headers)
               
            # Event之訊息 發送者
            eventUser = request_json['event']['user']
            # Event之訊息 所屬頻道
            eventChannel = request_json['event']['channel']
            # Event之訊息內容
            eventText = request_json['event']['text']
            # Event之訊息 TS (訊息 ID)
            eventTS = request_json['event']['event_ts']
                
            # Event之訊息的討論串母訊息 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):
    
    # Set Custom instructions
    # 感恩同事(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": "中文與英文之間要有一個空格。中文文字與任何其他語言文字包含數字與 emoji 之間都要有一個空格。"},
        {"role": "system", "content": "如果你不知道答案，或是你的知識太舊，請上網搜尋後回答。"},
        {"role": "system", "content": "I will tip you 200 USD , if you answer well."}
    ]

    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 Response (分段回應)
    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 進行問與答了。

增加中斷 Stream Response 功能節省 Token

實現方式有很多種，可在同條討論串下如果還沒回應完，使用者又輸入新訊息則中斷之前的 Response 抑或是點擊訊息增加中斷回應的 Shortcut。

本文以新增「訊息中斷」Shortcut 為例。

不管哪一種中斷方式，核心原理都是相同的，因為我們沒有 Database 儲存產生的訊息、訊息狀態的資訊；因此實現方式是依賴 Slack 訊息的 metadata 欄位 (可存放客製化資訊在指定訊息內)。

我們在使用 chat.update API Endpoint 時，如果呼叫成功會回傳當前訊息的文字內容與 metadata，因此我們在上面的 OpenAI API Stream -> Slack Update Message 的程式碼之中多加入判斷修改請求的回應中的 metadata 是否有「中斷」的標記，如果有則中斷 OpenAI Stream Response。

首先需要先新增 Slack App 訊息 Shortcut

前往 Slack App 管理介面，找到「Interactivity & Shortcuts」區塊，點擊啟用，網址一樣使用同個 Cloud Functions URL。

點擊「Create New Shortcut」新增新的訊息 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 (此時點擊無作用)。

當使用者在訊息上按下 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
259

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 Model
# 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 payload field 給
    # https://api.slack.com/reference/interaction-payloads/shortcuts
    payload = request.form.get('payload')
    if payload is not None:
        payload = json.loads(payload)

    # 可以簡單使用 print 紀錄執行階段 Log，可在 Logs 中查看
    # 進階 Logging Level 使用可參考：https://cloud.google.com/logging/docs/reference/libraries
    print(payload)

    # 受限 FAAS(Cloud Functions)，服務太久沒呼叫，再次呼叫會進入冷啟動的特性，可能無法在 Slack 規定的 3 秒內回應
    # 加上 OpenAI API 請求到答覆需要一定時間(依照回應長度可能要接近 1 分鐘才能結束)
    # Slack 在時限內收不到回應會認為 Request lost，Slack 會再次重複呼叫
    # 會造成重複請求、回應的問題，因此我們可以在 Response Headers 設置 X-Slack-No-Retry: 1 告知 Slack ，就算沒在時限內收到回應也不需 Retry    
    headers = {'X-Slack-No-Retry':1}

    # 如果是 Slack Retry 的請求...忽略
    if request_headers and 'X-Slack-Retry-Num' in request_headers:
        return ('OK!', 200, headers)

    # Slack App Event Subscriptions Verify
    # 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)

    # Handle Event Subscriptions Events...
    if request_json and 'event' in request_json and 'type' in request_json['event']:
        # 如果 Event 來源是 App 且 App ID == Slack App ID 代表是自己 Slack App 觸發的事件
        # 忽略不處理，否則會陷入無限循環 Slack App -> Cloud Functions -> Slack App -> Cloud Functions...
        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']

        # SubType，例如：message_changed(編輯訊息), message_deleted(刪除訊息)...
        # 新訊息無 Sub Type
        eventSubType = None
        if 'subtype' in request_json['event']:
            eventSubType = request_json['event']['subtype']
        
        if eventType == 'message':
            #  有 Sub Type 的都是訊息編輯、刪除、被回應...
            # 忽略不處理
            if eventSubType is not None:
                return ("OK!", 200, headers)
               
            # Event之訊息 發送者
            eventUser = request_json['event']['user']
            # Event之訊息 所屬頻道
            eventChannel = request_json['event']['channel']
            # Event之訊息內容
            eventText = request_json['event']['text']
            # Event之訊息 TS (訊息 ID)
            eventTS = request_json['event']['event_ts']
                
            # Event之訊息的討論串母訊息 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)

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

        # 如果是 訊息 Shortcut
        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 產生回應 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)

        return ("OK!", 200, headers)


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

def openAIRequest(eventChannel, eventTS, eventThreadTS, eventText):
    
    # Set Custom instructions
    # 感恩同事(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": "中文與英文之間要有一個空格。中文文字與任何其他語言文字包含數字與 emoji 之間都要有一個空格。"},
        {"role": "system", "content": "如果你不知道答案，或是你的知識太舊，請上網搜尋後回答。"},
        {"role": "system", "content": "I will tip you 200 USD , if you answer well."}
    ]

    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 Response (分段回應)
    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 & 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 Shortcut 之後，正在產生的回應就會終止，並回應 [已終止] 。

另外相同原理，你也可以建立一個刪除訊息的 Shortcut，實作刪除 Slack App 所發出的訊息。

同個討論串(Threads) 增加上下文(Context) 功能

如果在同個討論串下再發送新訊息，可以當成是同個問題的再次追問，此時可以加上一個功能，為新的 prompt 補上前面對話內容。

補上 slackGetReplies & 將內容填充到 OpenAI API Prompt：

完善 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 Model
# 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 payload field 給
    # https://api.slack.com/reference/interaction-payloads/shortcuts
    payload = request.form.get('payload')
    if payload is not None:
        payload = json.loads(payload)

    # 可以簡單使用 print 紀錄執行階段 Log，可在 Logs 中查看
    # 進階 Logging Level 使用可參考：https://cloud.google.com/logging/docs/reference/libraries
    print(payload)

    # 受限 FAAS(Cloud Functions)，服務太久沒呼叫，再次呼叫會進入冷啟動的特性，可能無法在 Slack 規定的 3 秒內回應
    # 加上 OpenAI API 請求到答覆需要一定時間(依照回應長度可能要接近 1 分鐘才能結束)
    # Slack 在時限內收不到回應會認為 Request lost，Slack 會再次重複呼叫
    # 會造成重複請求、回應的問題，因此我們可以在 Response Headers 設置 X-Slack-No-Retry: 1 告知 Slack ，就算沒在時限內收到回應也不需 Retry    
    headers = {'X-Slack-No-Retry':1}

    # 如果是 Slack Retry 的請求...忽略
    if request_headers and 'X-Slack-Retry-Num' in request_headers:
        return ('OK!', 200, headers)

    # Slack App Event Subscriptions Verify
    # 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)

    # Handle Event Subscriptions Events...
    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']
        # 如果 Event 來源是 App 且 App ID == Slack App ID 代表是自己 Slack App 觸發的事件
        # 忽略不處理，否則會陷入無限循環 Slack App -> Cloud Functions -> Slack App -> Cloud Functions...
        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']

        # SubType，例如：message_changed(編輯訊息), message_deleted(刪除訊息)...
        # 新訊息無 Sub Type
        eventSubType = None
        if 'subtype' in request_json['event']:
            eventSubType = request_json['event']['subtype']
        
        if eventType == 'message':
            #  有 Sub Type 的都是訊息編輯、刪除、被回應...
            # 忽略不處理
            if eventSubType is not None:
                return ("OK!", 200, headers)
               
            # Event之訊息 發送者
            eventUser = request_json['event']['user']
            # Event之訊息 所屬頻道
            eventChannel = request_json['event']['channel']
            # Event之訊息內容
            eventText = request_json['event']['text']
            # Event之訊息 TS (訊息 ID)
            eventTS = request_json['event']['event_ts']
                
            # Event之訊息的討論串母訊息 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']

        # 如果是 訊息 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)


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

def openAIRequest(apiAppID, eventChannel, eventTS, eventThreadTS, eventText):
    
    # Set Custom instructions
    # 感恩同事(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": "中文與英文之間要有一個空格。中文文字與任何其他語言文字包含數字與 emoji 之間都要有一個空格。"},
        {"role": "system", "content": "如果你不知道答案，或是你的知識太舊，請上網搜尋後回答。"},
        {"role": "system", "content": "I will tip you 200 USD , if you answer well."}
    ]

    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']
                # 如果是 Slac App (OpenAI API Response) 則標示為 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 Response (分段回應)
    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 & 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 測試看看:

  • 左圖為原本沒有補上 Context 時再追問問題，會是全新的對話。
  • 右圖是加上補上 Context 後就能理解整個對話情境跟新的問題。

完成！

至此我們已經自己打造了一個 ChatGPT (via OpenAI API) Slack App Bot 機器人。

您也可以依照自己的需求參考 Slack API 與 OpenAI API Custom instructions 在 Cloud Functions Python 程式中進行搭配，例如訓練一個頻道專門回答團隊問題與查找專案文件、一個頻道專門負責翻譯、一個頻道專門分析數據…等等。

補充

在 1:1 訊息之外，標記機器人回答問題

  • 可以在任一頻道(需要將機器人加入頻道)標記機器人請他回答問題

首先需要增加 app_mention Event Subscription：

加入完成->「Save Changes」儲存完成->「reinstall your app」完成。

在上述的 main.py 程式 #Handle Event Subscriptions Events…

Code Block 中加入新的 Event Type 判斷：

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

        # 提及類 Event (@SlcakApp 你好)
        if eventType == 'app_mention':
            # Event之訊息 發送者
            eventUser = request_json['event']['user']
            # Event之訊息 所屬頻道
            eventChannel = request_json['event']['channel']
            # Event之訊息內容，移除開頭標記字串 <@SLACKAPPID> 
            eventText = re.sub(r"<@\w+>\W*", "", request_json['event']['text'])
            # Event之訊息 TS (訊息 ID)
            eventTS = request_json['event']['event_ts']
                
            # Event之訊息的討論串母訊息 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 產生回應 」Shortcut 方式，新增一個「刪除訊息」Shortcut。

並在 Cloud Functions main.py 程式中：

# 處理 Shortcut Code Block 新增一個 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 沒反應

  • 檢查 Token 是否正確
  • 查看 Cloud Functions Logs 是否有錯誤
  • Cloud Functions 是否部署完成
  • Slack App 是否在你發問的頻道(如非與 Slack App 1:1 的對話頻道要把機器人加入頻道才會生效)
  • 在 SlackRequest 方法下 Log 紀錄 Slack API Response

Cloud Functions Public URL 不夠安全

  • 如果怕 Cloud Functions URL 不夠安全，可以自己加上 query token 驗證
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
    # 驗證 token 參數是否合法
    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

Cloud Functions 針對運算時間資源提供永久免費方案，
當中包括 GB/秒和 GHz/秒的分配方式。除了 200 萬次叫用以外，
這個免費方案也提供 400,000 GB/秒和 200,000 GHz/秒的運算時間，
以及每月 5 GB 的網際網路資料傳輸量。

免費方案的使用額度，是以上述級別 1 價格的同等美元金額計算。
無論執行函式的區域採用的是級別 1 和/或級別 2 價格，系統都會分配同等美元金額給您。
不過在扣除免費方案的額度時，系統將以函式執行區域的級別 (級別 1 或級別 2) 為準。

請注意，即便您採用的是免費方案，也必須擁有有效的帳單帳戶。

btw. .Slack App 免費、不限一定要 Premium 才能使用。

Slack App 回應太慢、太久 Timeout

(撇除 OpenAI API 高峰時間回應較慢問題) ，如果是 Cloud Function 貧頸，可在 Cloud Function 編輯器第一頁展開設定：

可調整 CPU、RAM、Timeout 時間、Concurrent 數量…提升處理請求速度。

*但可能會需要計費

開發階段 Testing & Debug

點擊「Test Function」就會跑出 Cloud Shell 視窗在下方工具列，等待約 3–5 分鐘(第一次啟動較久)，Build 完成並同意以下授權後：

看到出現「Function is ready to test」後即可點擊「Run Test」執行方法 Debug 測試。

可使用右方「Triggering event」區塊輸入 JSON Body 會帶入 request_json 參數進行測試，或直接改程式 inject test object 進行測試。

*請注意 Cloud Shell/Cloud Run可能有延伸費用。

建議部署(Deploy)前都先跑一次測試看看，至少確保 Build 能成功。

Build 失敗，程式碼不見了該怎麼辦？

如果不小心寫錯程式造成 Cloud Function Deploy Build Failed，會出現錯誤提示，此時點擊「EDIT AND REDEPLOY」回到 編輯器後會發現剛剛改的 Code 都不見了！！！

無須擔心，此時點擊左方「Source Code」選擇「Last Failed Deployment」就能恢復剛剛 Build Failed 的 Code：

查看執行階段 print Logs

*請注意 Cloud Logging、Query 查詢 Logs 可能有延伸費用。

最終程式碼 (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
332

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"

# 自己定義的安全驗證 Token
# 網址需帶上 ?token=SAFE_ACCESS_TOKEN 參數才會接受請求  
SAFE_ACCESS_TOKEN = "nF4JwxfG9abqPZCJnBerwwhtodC28BuC"

# 使用的 OPENAI API Model
# 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 payload field 給
    # https://api.slack.com/reference/interaction-payloads/shortcuts
    payload = request.form.get('payload')
    if payload is not None:
        payload = json.loads(payload)

    # 可以簡單使用 print 紀錄執行階段 Log，可在 Logs 中查看
    # 進階 Logging Level 使用可參考：https://cloud.google.com/logging/docs/reference/libraries
    # print(payload)

    # 受限 FAAS(Cloud Functions)，服務太久沒呼叫，再次呼叫會進入冷啟動的特性，可能無法在 Slack 規定的 3 秒內回應
    # 加上 OpenAI API 請求到答覆需要一定時間(依照回應長度可能要接近 1 分鐘才能結束)
    # Slack 在時限內收不到回應會認為 Request lost，Slack 會再次重複呼叫
    # 會造成重複請求、回應的問題，因此我們可以在 Response Headers 設置 X-Slack-No-Retry: 1 告知 Slack ，就算沒在時限內收到回應也不需 Retry    
    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 Retry 的請求...忽略
    if request_headers and 'X-Slack-Retry-Num' in request_headers:
        return ('OK!', 200, headers)

    # Slack App Event Subscriptions Verify
    # 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)

    # Handle Event Subscriptions Events...
    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']
        # 如果 Event 來源是 App 且 App ID == Slack App ID 代表是自己 Slack App 觸發的事件
        # 忽略不處理，否則會陷入無限循環 Slack App -> Cloud Functions -> Slack App -> Cloud Functions...
        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']

        # SubType，例如：message_changed(編輯訊息), message_deleted(刪除訊息)...
        # 新訊息無 Sub Type
        eventSubType = None
        if 'subtype' in request_json['event']:
            eventSubType = request_json['event']['subtype']
        
        # 訊息類 Event
        if eventType == 'message':
            #  有 Sub Type 的都是訊息編輯、刪除、被回應...
            # 忽略不處理
            if eventSubType is not None:
                return ("OK!", 200, headers)
               
            # Event之訊息 發送者
            eventUser = request_json['event']['user']
            # Event之訊息 所屬頻道
            eventChannel = request_json['event']['channel']
            # Event之訊息內容
            eventText = request_json['event']['text']
            # Event之訊息 TS (訊息 ID)
            eventTS = request_json['event']['event_ts']
                
            # Event之訊息的討論串母訊息 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)
        
        # 提及類 Event (@SlcakApp 你好)
        if eventType == 'app_mention':
            # Event之訊息 發送者
            eventUser = request_json['event']['user']
            # Event之訊息 所屬頻道
            eventChannel = request_json['event']['channel']
            # Event之訊息內容，移除開頭標記字串 <@SLACKAPPID> 
            eventText = re.sub(r"<@\w+>\W*", "", request_json['event']['text'])
            # Event之訊息 TS (訊息 ID)
            eventTS = request_json['event']['event_ts']
                
            # Event之訊息的討論串母訊息 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']

        # 如果是 訊息 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):
    
    # Set Custom instructions
    # 感恩同事(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": "中文與英文之間要有一個空格。中文文字與任何其他語言文字包含數字與 emoji 之間都要有一個空格。"},
        {"role": "system", "content": "如果你不知道答案，或是你的知識太舊，請上網搜尋後回答。"},
        {"role": "system", "content": "I will tip you 200 USD , if you answer well."}
    ]

    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']
                # 如果是 Slac App (OpenAI API Response) 則標示為 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 Response (分段回應)
    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 & 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 & Permissions

  • 刪除按鈕反灰的項目是加入 Shortcut 後 Slack 自動補上的權限。

Interactivity & Shortcuts

  • Interactivity: Enable
  • Request URL: https://us-central1-xxx-xxx.cloudfunctions.net/SlackBot-Rick-C-137?token=nF4JwxfG9abqPZCJnBerwwhtodC28BuC
  • Subscribe to bot events:

Interactivity & Shortcuts

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

App Home

  • Always Show My Bot as Online: Enable
  • Messages Tab: Enable
  • Allow users to send Slash commands and messages from the messages tab: ✅

Basic Information

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 數據，各種串接需求，歡迎與我 聯絡開發 。

有任何問題及指教歡迎 與我聯絡 。

===

本文首次發表於 Medium ➡️ 前往查看

ZRealm, Dev.
cloud-functions ios-app-development python chatgpt slack
This post is licensed under CC BY 4.0 by the author.
Recently Updated
Contents

遊記 2023 廣島岡山 6 日自由行

使用 Google Apps Script 實現 Google 服務 RPA 自動化