【初心者向け】AWS LambdaとOpenAI API(と、API gateway)を使い、スタンプを押すと翻訳してくれるSlack botを作る
こんにちは、浅井です。
大学院では自然言語処理を研究しており、インターン先でもLLMに関するプロジェクトに参加しています。
今回はインターン業務の傍らで作成した「Slack内で簡単に翻訳を行うbot」について紹介します。
言語モデルを利用した業務も研究も個人プロジェクトも中々うまく進まない中、しばしば現実逃避のためにこういうものを作って色々遊んでいます。
できる範囲でそれも発信していこう！と思い立ったので、こうしてnoteを書いています。
実際に業務に欲しい人や、そうでなくともOpenAI APIを使って遊んでみたい人は読んでもらえるとうれしいです。
「初心者向け」と書いてある通り、非エンジニアの人でも何とか作れると思います(僕もほとんどコード書けないので……)。非エンジニアの人、普段コードを書かない人もぜひ試してみてください。
背景
僕のインターン先には外国人のエンジニアさんがたくさんいます。
そのため仕事に関する場面、Slackやミーティングにおける公用語は基本的に英語です。
ほとんどの場合、英語のみの利用で問題ないのですが、以下のような場合に自動翻訳してほしい、と思うことがあります。
ブログなどのために用意する文章は日本語のため、外国人エンジニアが理解できない
仕事に関係のない所感チャンネルに落書き程度に書いている日本語の内容を知りたい
英語が苦手でたまに理解できてるか怪しいインターン生がいる🙃
etc…
そのため、サクッとSlack上の日本語↔️英語を「その場で」翻訳できるような簡易なSlack botを実装してみました。
使ったもの
OpenAI API(gpt-4)
今回の翻訳はgpt-4に行ってもらいました。
OpenAI APIを使用した理由は、クオリティとプライバシー保護の観点からです。クオリティは言わずもがな、プライバシーポリシーについてもある程度真っ当であるように思えます。
※とはいえ、送信するデータは慎重に選択する必要があります。業務に関係のないおしゃべりチャンネル等から試していくことをお勧めします。
AWS Lambda
関数をここで作成しました。サーバーレスで従量課金制なので、小さな社内ツールの作成などに便利。
Amazon API Gateway
SlackのWebhookをAPI Gatewayに接続し、API GatewayのイベントをトリガーにLambda関数を起動させるようにしました。
Slack API
Slackbotを作れます。
実装内容
⓪AWSアカウント、OpenAI API用のアカウントを作成
AWSのアカウントはすでに作成されているものとします、すみません。
初めての人は下記ページに従ってアカウントの作成を行なってください。
次にOpenAI APIの取得についてです。
下記のページからサインアップし、APIキーを作成してください。
アカウント作成後、こちらのAPI keysから新しいシークレットキーを作成します。作成したシークレットキーは初回にちゃんとコピーしてください。
①Lambda関数を作成する
AWS Lambdaを開き、「関数」ページより、「関数の作成」をクリックします。
関数名は好きにして大丈夫です。
ランタイムは新しいpythonのバージョンにすれば良いと思います。
それ以外は一旦そのままで「関数の作成」をクリックします。
問題なければ、無事Lambda関数が作成できるはずです。
作成が完了したら、いくつかの設定を行いましょう。
まず、Lambda関数の「設定」→「環境変数」ページより、先ほどOpenAI APIのページで取得したシークレットキーを格納してください。
また、lambdaの「一般設定」でのタイムアウト時間を伸ばしておきましょう。
OpenAI APIからのレスポンスはそれなりに時間を要すため、デフォルトの3秒ではタイムアウトしてしまいます。
②Lambdaレイヤーを作成し、OpenAIのpythonパッケージを使用する
次はOpenAIのpythonパッケージを導入するためのレイヤー作成です。
(お使いの環境によってはここで躓くことが多いので、はまりどころを末尾に記載しておきます。)
openaiにpythonからリクエストを送るには、OpenAIのpythonパッケージをあらかじめインストールしておく必要があります。
しかしLambdaの中で直接 pip install openai することはできません。
そのため、一旦ローカルでpythonパッケージをダウンロードしておき、それをLambdaレイヤーとしてアップロードする必要があります。
まずはローカルで以下のコマンドを実行し、Lambdaレイヤーのzipファイルをダウンロードします。
mkdir python
pip install -t ./python openai
zip -r openai.zip ./python最近はfastapiのバージョンのせいでエラーが起きるらしい？ので、こちらにしておいた方が安心かもしれません。(2024/01現在)
mkdir python
pip install -t ./python openai fastapi==0.99.0
zip -r openai_fastapi.zip ./pythonその後、上記のコマンドで作成したopenai.zipをLambda上にアップロードし、Lambdaレイヤーを作成します。
「アップロード」より、先ほど作成したzipファイルをアップロードしましょう。
レイヤーの名前は好きにつけて大丈夫です。
アーキテクチャやランタイムは作成したLambda関数と同じものを選択してください。
レイヤーの作成が完了したら、先ほど作ったLambda関数を再度開き、「レイヤーを追加」から作成したレイヤーを関数に追加します。
これでopenaiのライブラリが入ったLambdaレイヤーの作成・追加は完了です。レイヤーが適切に動作しているか確認しましょう。
Slackとの接続等は後回しにし、LambdaでOpenAIに対するAPIリクエストが成功するかだけを確かめます。そのためのコードを以下に記しておきます。
import os, sys, json
from openai import OpenAI
client = OpenAI(
api_key=os.environ['OPENAI_API_KEY']
)
content = '''
Please translate the text from user.
Do not output any words other than what is translated.
If the given text is in Japanese, translate it into English.
If the given text is in English, translate it into Japanese.
'''
completion = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": content},
{"role": "user", "content": "今日はいい天気ですね！"}
]
)
ans = completion.choices[0].message.content
print(ans)
def lambda_handler(event, context):
return ansコードを書き終わったら、「Deploy」→「Test」で動作をテストします。
test eventを設定する必要がありますが、まだ使わないのでデフォルトのままで保存して大丈夫です。
無事に翻訳ができていたら成功です。
上手くいかない場合、レイヤーからopenaiのライブラリを適切に読み込めていないかもしれません。
③(とりあえず)チャンネルに手動でポストしてくれるSlack botを作成する
無事にLambdaでOpenAI APIが動作することを確認できました。
次はLambda→Slackに情報を送り、マニュアルでbotを動かすことを目指します。
まず、Slackで新しいappを作成します。
Create New Appをクリックし、今回のためのSlack botを作成します。
作成したら、次はpermissionの設定を行います。
Slack APIの左サイドバーから、OAuth & Permissionsページを開きます。
Scopesを確認し、「Add an OAuth Scope」よりchat:writeを追加します。
次にInstall to Workspaceを押してワークスペースとの接続を行います。
接続が成功したら、「xoxb-…….」のようなBot User OAuth Tokenが取得できます。
早速、このTokenを用いてSlackに投稿を行えるか試してみましょう。
Lambdaの環境変数ページを開き、手に入ったBot User OAuth TokenをSLACK_TOKENとして追加します。
その後、投稿したいSlackのチャンネルにあらかじめSlackbotを追加しておきます。
適当に該当するbotをメンションして送信したらチャンネルに追加できます。
絶対にこれ以外のやり方がある気がします。
また、該当チャンネルのチャンネルIDが必要なため、こちらも取得します。
チャンネルIDはSlackでチャンネルを右クリックし、「コピー」→「リンクをコピー」を押せば取得できます。URLの最後の9~12文字くらいの文字列がチャンネルIDです。(https://xxxxxxx.slack.com/archives/…….. の archives/ 以降の部分)
これらの情報を用いて、lambda_function.pyを更新し、チャンネルに投稿してみましょう。
※下記の関数を実行するために、requestsというpythonライブラリが必要なので、openaiのライブラリと同様にレイヤーを作成し追加してください。
以下がテスト用コードになります。
import os, sys, json
from openai import OpenAI
import requests
slack_token = os.environ["SLACK_TOKEN"]
def send_message_to_slack(channel, message,token):
url = "https://slack.com/api/chat.postMessage"
headers = {
"Authorization": "Bearer " + token,
"Content-Type": "application/json"
}
payload = {
"channel": channel,
"text": message,
}
response = requests.post(url, data=json.dumps(payload), headers=headers)
return response.json()
client = OpenAI(
api_key=os.environ['OPENAI_API_KEY']
)
content = '''
Please translate the text from user.
Do not output any words other than what is translated.
If the given text is in Japanese, translate it into English.
If the given text is in English, translate it into Japanese.
'''
completion = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": content},
{"role": "user", "content": "今日はいい天気ですね！"}
]
)
ans = completion.choices[0].message.content
def lambda_handler(event, context):
channel = "ここに取得したチャンネルのIDを入れてください"
send_message_to_slack(channel, ans, slack_token)
return ans先ほどと同様にtestを実行します。
Slackのチャンネルに意図通りにbotの投稿が行われていたら成功です！
これでLambda→Slackの送信に関しては完了です。
ここまででひと段落なので、お茶を飲みます。🍵
④API Gatewayを構築し、Slackのユーザーのアクションを取得できるようにする
一つ前の項で、Lambda→Slackへの送信は行えることが確認できました。
次は、SlackでのユーザーのアクションをLambdaに通知させる仕組みを作ります。(スタンプを押したタイミングで実行させる必要があるためです)
SlackのWebhookをAPI Gatewayに接続し、API GatewayのイベントをトリガーにLambda関数を起動させるようにしました。
今回の場合のやり方を説明します。
まず、API Gatewayを開き、「APIを作成」から新しいAPIを作成します。
その後、リソースを作成します。
次にメソッドを作成します。
作成したリソースを開き、「メソッドを作成」を選択してください。
メソッドタイプは「POST」を選択し、統合タイプは「Lambda関数」を選択してください。Lambda関数の項では、作成したLambda関数を選択します。
それらの準備が終わったら、APIをデプロイします。
Slackからはapplication/x-www-form-urlencoded形式でデータが送られてくるので、これをjson形式に変換しておきます。
該当するPOSTについて、統合リクエストの設定を開き、マッピングテンプレートを編集します。
(こちらのページをそのまま使いました。ありがとうございます……)
application/x-www-form-urlencoded
## convert HTML POST data or HTTP GET query string to JSON
## get the raw post data from the AWS built-in variable and give it a nicer name
#if ($context.httpMethod == "POST")
#set($rawAPIData = $input.path('$'))
#elseif ($context.httpMethod == "GET")
#set($rawAPIData = $input.params().querystring)
#set($rawAPIData = $rawAPIData.toString())
#set($rawAPIDataLength = $rawAPIData.length() - 1)
#set($rawAPIData = $rawAPIData.substring(1, $rawAPIDataLength))
#set($rawAPIData = $rawAPIData.replace(", ", "&"))
#else
#set($rawAPIData = "")
#end
## first we get the number of "&" in the string, this tells us if there is more than one key value pair
#set($countAmpersands = $rawAPIData.length() - $rawAPIData.replace("&", "").length())
## if there are no "&" at all then we have only one key value pair.
## we append an ampersand to the string so that we can tokenise it the same way as multiple kv pairs.
## the "empty" kv pair to the right of the ampersand will be ignored anyway.
#if ($countAmpersands == 0)
#set($rawPostData = $rawAPIData + "&")
#end
## now we tokenise using the ampersand(s)
#set($tokenisedAmpersand = $rawAPIData.split("&"))
## we set up a variable to hold the valid key value pairs
#set($tokenisedEquals = [])
## now we set up a loop to find the valid key value pairs, which must contain only one "="
#foreach( $kvPair in $tokenisedAmpersand )
#set($countEquals = $kvPair.length() - $kvPair.replace("=", "").length())
#if ($countEquals == 1)
#set($kvTokenised = $kvPair.split("="))
#if ($kvTokenised[0].length() > 0)
## we found a valid key value pair. add it to the list.
#set($devNull = $tokenisedEquals.add($kvPair))
#end
#end
#end
## next we set up our loop inside the output structure "{" and "}"
{
#foreach( $kvPair in $tokenisedEquals )
## finally we output the JSON for this pair and append a comma if this isn't the last pair
#set($kvTokenised = $kvPair.split("="))
"$util.urlDecode($kvTokenised[0])" : #if($kvTokenised[1].length() > 0)"$util.urlDecode($kvTokenised[1])"#{else}""#end#if( $foreach.hasNext ),#end
#end
}
以下のようにマッピングテンプレートを設定しておけば、Slackからのリクエストが以下のようなjson形式で取得できるようになるはずです。
{'token': 'QHIqUgKBA3VDT0mkvc3IgaKX', 'challenge': 'lPwXCkXaPyri14Q3sIaWUcIjZXTPioYUTrvis7ivF0TZ606SxZ0U', 'type': 'url_verification'}Slack APIのページを開き、ステージのURLを登録しましょう。
前項で作成したステージではURLが生成されているので、こちらをコピーします。
その後、Slack APIからEvent Subscriptionsを開きます。
Request URLの欄に先ほど取得したステージのURLを入れると、Slack側からテストのリクエストが送信されます。
このテストでは、受け取ったevent内のchallengeの項目をそのまま返すことが求められます。
そのため、作成していたLambda関数のlambda_handler部分を一時的に以下のようにしておきます。
def lambda_handler(event, context):
send_message_to_slack(channel, ans, slack_token)
print(event)
return event['challenge']
上記のようにVerifiedの✅がついたら、接続は成功です。
⑤スタンプに反応し、自動で返信するSlack botを作成する
満を持してSlack botを完成させます。
まずは、botがスタンプを検知するよう、Slack側で許可を付与します。
Subscribe to bot eventsより、reaction_addedを追加して、Save Changesします。
reinstallを求められると思うので、eventは追加するたびにワークスペースにbotを再インストールしましょう。
上記の操作で、ユーザーの誰かがスタンプを押すと、Slack APIからAPI Gatewayを通してLambdaに情報が送られるようになりました。
また、スタンプに対するリアクションの情報だけでは、翻訳する本文を閲覧できないので、OAuth&PermissionsのページからScopesを新たに追加します。
以下の画像のようなスコープの付与を行なってください。
import os, sys, json
from openai import OpenAI
import requests
slack_token = os.environ["SLACK_TOKEN"]
def send_message_to_slack(channel, message,token,thread):
url = "https://slack.com/api/chat.postMessage"
headers = {
"Authorization": "Bearer " + token,
"Content-Type": "application/json"
}
payload = {
"channel": channel,
"text": message,
"thread_ts": thread
}
response = requests.post(url, data=json.dumps(payload), headers=headers)
return response.json()
client = OpenAI(
api_key=os.environ['OPENAI_API_KEY']
)
def get_message_text(event):
# Slack APIトークン
token = slack_token
# メッセージの情報を取得するためのリクエストパラメータ
channel = event["event"]["item"]["channel"]
message_ts = event["event"]["item"]["ts"]
# Slack APIを使ってメッセージの内容を取得
url = "https://slack.com/api/conversations.replies"
headers = {"Authorization": "Bearer " + token}
params = {"channel": channel, "ts": message_ts}
response = requests.get(url, headers=headers, params=params)
if response.status_code == 200:
message_data = response.json()
print("message_data: \n\n\n", message_data)
if message_data["ok"] and message_data["messages"]:
message_text = message_data["messages"][0]["text"]
print("message_text: \n\n\n",message_text)
return message_text
def run_gpt(text):
content = '''
Please translate the text from user.
Do not output any words other than what is translated.
If the given text is in Japanese, translate it into English.
If the given text is in English, translate it into Japanese.
'''
completion = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": content},
{"role": "user", "content": text}
]
)
ans = completion.choices[0].message.content
return ans
def lambda_handler(event, context):
if event["event"]["reaction"] != "white_check_mark":
return
channel = event["event"]["item"]["channel"]
text = get_message_text(event)
ans = run_gpt(text)
thread = event["event"]["item"]["ts"]
send_message_to_slack(channel, ans, slack_token,thread)
return上記のコードをdeployし、botが入っているSlackチャンネルでスタンプを押したら、その翻訳が投稿されるはずです。(✅スタンプの時のみ反応するようにしています)
これでほとんど完成です。
しかし、もう一点問題があります。
Slack botは3秒以内にリクエストが返ってこなかった場合、リクエストが何度も行われてしまうようです。
そのため、Slack API側からの複数回のリクエストを防止しなければ、以下のようになってしまいます。
Lambdaの設定から同時実行周りをいじれば解決するかな、と思ったのですが、上手くいきませんでした(知っている人いたら教えてください……)
これを予防する方法としては、
①S3やDBなどに受け取った文章を保存し、重複がないかチェックする。
②Lambda関数を二つに分ける。(イベントを受け取るものと、処理をするもの)
③ヘッダーの X-Slack-Retry-Numを確認し、再送のものだけを無視する
などを考えましたが、今回は①を利用してみたいと思います。
翻訳ではあまり有り難みが無いですが、ユーザーの質問と応答については今後のモデル改善のために保存しておきたいと考えたからです。
今回はDynamoDBを使い、その中にリクエストを保存し、重複を確認することにします。
DynamoDBを開き、「テーブルの作成」を押してください。
timestampをパーティションキーにし、テーブルを作成します。
また、LambdaからDynamoDBへのアクセス権限も追加しておいてください。
「設定」→「アクセス権限」より、青文字のロール名をクリックします。
現在のロールの権限について、「許可を追加」→「ポリシーをアタッチ」より、"AmazonDynamoDBFullAccess"を追加してください。
これでDynamoDBにアクセスする許可を与えることができました。
Lambda関数が実行されるたびにDynamoDB内に存在するかを確認した上で、
- 初めてのリクエストの場合、dynamodbに追加してから実行
- すでに存在する場合、そのまま実行終了
するような実装を行います。
(短時間に大量にDynamoDBを操作すると料金が大変なことになります、大人数のSlackワークスペースで使用する際にはご注意ください)
今回、最終的なコードは以下のようになりました。
import os, sys, json
from openai import OpenAI
import requests
import boto3
from boto3.dynamodb.conditions import Key
dynamodb = boto3.resource('dynamodb')
slack_token = os.environ["SLACK_TOKEN"]
def send_message_to_slack(channel, message,token,thread):
url = "https://slack.com/api/chat.postMessage"
headers = {
"Authorization": "Bearer " + token,
"Content-Type": "application/json"
}
payload = {
"channel": channel,
"text": message,
"thread_ts": thread
}
response = requests.post(url, data=json.dumps(payload), headers=headers)
return response.json()
client = OpenAI(
api_key=os.environ['OPENAI_API_KEY']
)
def get_message_text(event):
# Slack APIトークン
token = slack_token
# メッセージの情報を取得するためのリクエストパラメータ
channel = event["event"]["item"]["channel"]
message_ts = event["event"]["item"]["ts"]
# Slack APIを使ってメッセージの内容を取得
url = "https://slack.com/api/conversations.replies"
headers = {"Authorization": "Bearer " + token}
params = {"channel": channel, "ts": message_ts}
response = requests.get(url, headers=headers, params=params)
if response.status_code == 200:
message_data = response.json()
print("message_data: \n\n\n", message_data)
if message_data["ok"] and message_data["messages"]:
message_text = message_data["messages"][0]["text"]
print("message_text: \n\n\n",message_text)
return message_text
def check_translate_dict(event,thread,text):
table = dynamodb.Table('ここに作成したDynamoDBのテーブル名を入れてください')
response = table.get_item(
Key={
'timestamp': thread
}
)
print("responce: ",response)
if 'Item' in response:
print("already answered")
return False
else:
table.put_item(
Item={
"timestamp": thread,
"contents" : str(event),
"text" : text
}
)
return True
def run_gpt(thread,text):
content = '''
Please translate the text from user.
Do not output any words other than what is translated.
If the given text is in Japanese, translate it into English.
If the given text is in English, translate it into Japanese.
'''
completion = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": content},
{"role": "user", "content": text}
]
)
ans = completion.choices[0].message.content
table = dynamodb.Table('ここに作成したDynamoDBのテーブル名を入れてください')
table.update_item(
Key={
'timestamp': thread # 既存アイテムを特定するためのキー
},
UpdateExpression='SET answer = :val', # 追加する属性とその値
ExpressionAttributeValues={
':val': ans
})
return ans
def lambda_handler(event, context):
if event["event"]["reaction"] != "white_check_mark":
return
channel = event["event"]["item"]["channel"]
text = get_message_text(event)
thread = event["event"]["item"]["ts"]
check = check_translate_dict(event,thread,text)
if check == False:
return
ans = run_gpt(thread,text)
send_message_to_slack(channel, ans, slack_token,thread)
print(event)
returnこのコードをデプロイ後、Slack上でスタンプを押し、数秒後にそれに対してbotが返信を送っていたならば成功です！
完成！
周り道だったらごめんなさい！
プロンプトを調整すれば翻訳以外の用途にも使えるので、色々と試してみてください。
起こりがちなエラー
以下にまとめています、他に何かあれば連絡ください。
今後の展望
スタンプに反応するようにしましたが、メンションへのリプライやDMへの送信等も行えます。用途に合わせて色々と調整するといいと思います。
今回はスレッドにリプライを送るという形にしましたが、スレッドでの会話が遮断されることが不満だという意見もありました。
本人しか見ないのであれば、代わりに翻訳内容をDM送信する、等のオプションも検討すると、体験がより良くなるように感じています。
その他、何か不都合な部分やご意見等ありましたらご連絡ください。
We’re hiring!
株式会社Yoiiでは一緒に働く仲間を募集しています。
データエンジニアやソフトウェアエンジニアをはじめ、様々な職種での募集を随時行っていますので、興味がある方は上記の求人一覧をご覧ください。
感想
社内で英語わかんないの俺だけだから9割俺しか使ってない。
この記事が気に入ったらサポートをしてみませんか？