はじめに
大流行中のChatGPT。
このビックウェーブに乗り遅れないために、先日公開されたAPIを使ってちょっとしたLineBotを作ってみました。
つくったもの(SenryuSalaryman)
お題を与えると悲哀のこもったサラリーマン川柳を返すLineBotです。(5個作成してChatGPTが一番面白いと思った1つを理由とともに返答するようにしています。よほど伝えたいのか、時々5個全部返します。と思ったら選択した理由を返さなかったりします。受け手の想像にお任せするということだと思います。)
以下のリンク、QRコードより友達登録できます。
※API、有料なので1人1日5回までの制限をかけています。。API自体にも月間利用料の制限かけてます。すみません。
コード
※openaiを含んだLambdaLayerを作成する必要があります。
import json
import logging
import os
from decimal import Decimal
import boto3
import openai
import requests
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# -------------------------------------------------------------------------------
# OpenAI API用の初期設定
# -------------------------------------------------------------------------------
openai.api_key = os.environ["API_Key"]
API_ENDPOINT = "https://api.openai.com/v1/engine/davinci-codex/completions"
# -------------------------------------------------------------------------------
# MessagingAPI用の初期設定
# -------------------------------------------------------------------------------
## チャンネルアクセストークンの設定(環境変数)
LINE_CHANNEL_ACCESS_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"]
## URL設定
REQUEST_URL = "https://api.line.me/v2/bot/message/reply"
## ヘッダー設定
REQUEST_HEADERS = {
"Authorization": "Bearer " + LINE_CHANNEL_ACCESS_TOKEN,
"Content-Type": "application/json",
}
# -------------------------------------------------------------------------------
# dynamodbの初期設定
# -------------------------------------------------------------------------------
DYNAMODB = boto3.resource("dynamodb", region_name="ap-northeast-1")
DYNAMO_TABLE = DYNAMODB.Table("senryu_user_info")
def lambda_handler(event, context):
main(json.loads(event["body"]))
return 0
def main(event: dict) -> None:
"""メイン処理
Args:
event (dict): リクエスト内容
"""
## ユーザIDを取得
user_id = event["events"][0]["source"]["userId"]
## イベントタイプを取得
event_type = event["events"][0]["type"]
## リプライトークンを取得
reply_token = event["events"][0]["replyToken"]
# 友達登録時のリクエスト
if event_type == "follow":
create_user(user_id)
return 0
## ユーザ情報の取得
user_info = get_senryu_user_info(user_id)
## ユーザ登録されていない場合、登録
if user_info is None:
create_user(user_id)
## リクエスト回数のチェック(1日5回まで)
if (user_info["Item"]["request_count"]) <= 5: # 別のLambda関数で毎日0時に全ユーザーのリクエストカウントをリセットしている
# メッセージイベント
if event_type == "message":
## リクエストメッセージの種類を取得
case = event["events"][0]["message"]["type"]
if case == "text":
keyword = event["events"][0]["message"]["text"]
if len(keyword) > 15:
logger.error(f"長すぎるメッセージ/{keyword}")
message_post("お題が長すぎます。15文字以内でお願いします。", reply_token)
answer = create_answer(keyword)
message_post(answer, reply_token)
countup_user(user_id)
# 想定外のメッセージケースの場合エラーで返す
else:
logger.error(f"想定外のメッセージケース/{case}")
message_post("すみません、よくわかりませんでした。", reply_token)
# 想定外のイベントタイプの場合エラーで返す
else:
logger.error(f"想定外のイベントタイプ/{event_type}")
message_post("すみません、よくわかりませんでした。", reply_token)
else:
logger.error(f"一日のリクエスト回数上限超過/{user_info}")
message_post("すみません、1日のリクエスト上限(5回)をこえているようです。\nまた明日お試しください。", reply_token)
return 0
def decimal_to_int(obj):
if isinstance(obj, Decimal):
return int(obj)
def create_answer(message):
instructions = """
あなたはベテランのサラリーマンです。
与えられたのお題に対して悲哀のこもったクスッと笑える川柳を5つ作成してください。
返答として返すのは、作成した中で最も面白い1つのみとし、選択した理由もあわせて返答してください。
"""
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": instructions},
{"role": "user", "content": message},
],
)
print(
"Received response:"
+ json.dumps(response, default=decimal_to_int, ensure_ascii=False)
)
return response["choices"][0]["message"]["content"]
def create_user(user_id):
DYNAMO_TABLE.put_item(Item={"user_id": user_id, "request_count": 0})
def countup_user(user_id):
DYNAMO_TABLE.update_item(
Key={"user_id": user_id},
UpdateExpression="set request_count = request_count + :val",
ExpressionAttributeValues={":val": 1},
)
def get_senryu_user_info(user_id):
response = DYNAMO_TABLE.get_item(
Key={
"user_id": user_id,
}
)
return response
def message_post(message, reply_token) -> None:
"""リプライメッセージを送信する。(メッセージ指定)
Args:
reply_token (_type_): リプライトークン
"""
params = {
"replyToken": reply_token,
"messages": [{"type": "text", "altText": "SenryuSalaryman", "text": message}],
}
requests.post(REQUEST_URL, json=params, headers=REQUEST_HEADERS)
定期的にリクエスト回数をクリアする処理
こちらはシンプルな処理なのでChatGPTに作成を依頼してみました。
出力されたスクリプトのテーブル名を変更するだけで使えました。(毎日0時とは使えていないのに、そのようになるCron式まで教えてくれました。)
dynamodbの内容を定期的にクリアするpythonコードを作成してください。
対象とするレコードのキーはuser_idで、クリアする項目はrequest_countです。
以下は、DynamoDBの内容を定期的にクリアするPythonスクリプトです。このスクリプトは、AWS Lambdaで実行することを想定しています。レコードのキーはuser_idで、クリアする項目はrequest_countです。
このスクリプトをAWS Lambdaにデプロイし、AWS EventBridgeを使用して定期的に実行するよう設定することができます。たとえば、毎日0時にリセットする場合、EventBridgeで新しいルールを作成し、Cron式 0 0 * * ? * を使用してスケジュールを設定します。そして、作成したLambda関数をターゲットとして選択します。
import boto3
# DynamoDBリソースの取得
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("your_dynamodb_table_name")
def lambda_handler(event, context):
# DynamoDBから全レコードを取得する
response = table.scan()
# 取得したレコードを反復処理し、request_countをクリアする
for item in response["Items"]:
user_id = item["user_id"]
# request_countを0に設定する
table.update_item(
Key={"user_id": user_id},
UpdateExpression="SET request_count = :val",
ExpressionAttributeValues={":val": 0}
)
return {
"statusCode": 200,
"body": "Request counts have been cleared."
}
Botのネーミング
どうせなのでBotのネーミングもChatGPTに考えてもらいました。そのうちの一つを採用しています。
与えられたお題をもとにChatgpt でサラリーマン川柳を生成して返すLinebotのネーミングを5つ考えてください。
以下に、与えられたお題に基づいてChatGPTでサラリーマン川柳を生成して返すLinebotのネーミング案を5つ提案します。
HaikuMasterBot(ハイクマスターボット)
SenryuSalaryman(せんりゅうサラリーマン)
KansaiHaikuGenius(関西俳句ジーニアス)
HaikuForYouBot(ハイクフォーユーボット)
OfficePoetLinebot(オフィスポエットラインボット)
これらのネーミング案は、サラリーマン川柳の生成に焦点を当てたLinebotを表現しています。お好みの名称を選択し、Linebotの開発に取り組んでください。
おわりに
ChatGPT周りの実装だけなら30分もかからず行うことができました。(API呼び出すだけ)
AWS LambdaとDynamoDBを用いたLineBotの作成さえ慣れてしまえばいろんなBotをさくっと作れそうです。
※ちなみに、返される川柳はクオリティ低めなので実際のサラリーマン川柳では戦えそうにないです。。指示プロンプトの内容をチューニングすればもうちょっとクスッと笑えるようなものができるのだろうか。