LoginSignup
4
1
@mamoru-ngyin株式会社タニタ

筋トレを継続するためのLINE Bot（AWS Lambda × ChatGPT でLINE Bot作成）

Last updated at Posted at 2024-02-19

筋トレを継続したい

ダイエットや筋肉をつけたいと思ったとき、多くの人が筋トレを始めます。しかし、途中で挫折してしまい、筋トレを継続できない人もいるでしょう。そこで、筋トレを継続するためのLINE Botを作成しました。

開発環境

AWS Lambda
Python 3.12
OpenAI API

トレーニングメニュー

トレーニングメニューを自分で考えるのは面倒ですよね。
そこで、ChatGPTにお願いして、トレーニングメニューを考えてもらいましょう。

いい感じのメニューを提案してくれましたね。

筋トレ後は褒めてもらおう

ただトレーニングをこなしていくだけだと、モチベーションが保てませんよね。
トレーニングが終わった後は、トレーニングしたことを報告して、褒めてもらいましょう。

全く何言ってるかわかりませんが、これはテンション上がりますね。
プロテイン摂取を忘れないように言ってくれます。

サボったときは叱ってもらおう

自分の心の弱さでトレーニングをサボってしまうこともありますよね。
サボりが続いてしまうといけないので、サボったときは叱ってもらいましょう。

はじめの2回は軽めに指摘されるだけですが、3回連続でサボってしまうと、厳しめに叱られます。

実装

ライブラリのインポートと設定

まず、必要なライブラリをインポートし、AWS Lambda関数の設定を行います。
OpenAIのAPIキーを環境変数から取得しています。これは、後でChatGPTを利用するためのものです。

lambda.py
# ライブラリのインポート
import os
import json
import boto3
import requests
import csv
import io

# GPTのAPIKEY
OPENAI_API_KEY = os.environ['OPENAI_API_KEY']

# AWS Lambdaハンドラー関数
def lambda_handler(event, context):

LINE Botの設定とS3からのデータ取得

LINE Botの設定を行い、S3から必要なデータを取得します。
CHANNEL_SECRET、CHANNEL_ACCESS_TOKENは環境変数から取得しています。
また、BUCKET_NAME、FILE_KEYには、S3のバケット名とファイル名を記入してください。

lambda.py
# LINE Botの設定
line_config = {
    'channelSecret': os.environ['CHANNEL_SECRET'],
    'channelAccessToken': os.environ['CHANNEL_ACCESS_TOKEN'],
}

# Lambdaクライアントの作成
lambda_client = boto3.client('lambda')

# S3のバケット名とファイルキーを取得
bucket_name = 'BUCKET_NAME'
file_key = 'FILE_KEY'

# S3オブジェクトを取得
s3 = boto3.client('s3')

ユーザーメッセージの処理とデータの更新

LINE Botから受け取ったメッセージイベントから、テキストメッセージの内容を抽出します。

S3から取得したCSVファイルのデータを読み込み、新しいユーザーメッセージを追加します。更新されたデータは、S3に再度アップロードされます。

CSVファイルの中身で応答が変わるようにするため、追加するデータを含めた行数や元々の最後尾データを取得しています。

lambda.py
# S3からファイルをダウンロードしてcsvデータを読み取る
response = s3.get_object(Bucket=bucket_name, Key=file_key)
data = response['Body'].read().decode('utf-8')

# イベントからメッセージイベントを取得して処理
body = json.loads(event['body'])
for message_event in body['events']:
    # テキストメッセージの場合のみ処理
    if message_event['type'] == 'message' and message_event['message']['type'] == 'text':
        # ユーザーメッセージを抽出
        user_message = {
            'role': 'user',
            'content': message_event['message']['text']
        }

        # 更新されたcsvデータを作成
        updated_contents = io.StringIO()
        writer = csv.writer(updated_contents)
        
        # 元データの書き込み、データ数カウント
        csv_data = csv.reader(io.StringIO(data))
        num_lines = 1
        for row in csv_data:
            merged_row = ["".join(row)]
            writer.writerow(merged_row)
            num_lines += 1
        
        # 新しいデータ
        new_data = str(message_event['message']['text'])
        
        # 新しいデータを追加
        merged_new_data = ["".join(new_data)]
        writer.writerow(merged_new_data)
        
        # 最後の行とその直前の行のデータを取得
        last_row = None
        prev_row = None
        csv_data = csv.reader(io.StringIO(data))
        for row in csv_data:
            prev_row = last_row
            last_row = row
        
        # S3に更新されたデータをアップロード
        s3.put_object(Bucket=bucket_name, Key=file_key, Body=updated_contents.getvalue())

ChatGPTへのメッセージ送信と応答の取得

2つのプロンプト設定を行い、ユーザーが前回送信したメッセージやデータの状態によって、どちらかが選択されるようにしています。

ChatGPTのAPIを呼び出し、ChatGPTに対してメッセージを送信します。その後、ChatGPTからの応答を取得しています。

lambda.py
# promptの設定
prompt1 = {
    "role":"system",
    "content":"「メニュー」と来たら、20分程度でできる自重トレーニングのメニューとその説明を教えてください"
              "「筋トレしたよ」と来たら、ぶっ飛んだように変なくらいユーモアのある感じで相手を褒めてください。"
              "「筋トレサボった」と来たら、軽く指摘してください"
}
prompt2 = {
    "role":"system",
    "content":"「筋トレサボった」と来たら、とても怖い感じでめっちゃ怒ってください"
}

# 行数で分岐
if num_lines < 3:
    prompt = prompt1
elif num_lines >= 3:
    # 3行のデータで分岐
    if (((merged_new_data == ['筋トレサボった']) | (merged_new_data == ['筋トレさぼった']) | (merged_new_data == ['サボった']) | (merged_new_data == ['さぼった'])) &
        ((prev_row == ['筋トレサボった']) | (prev_row == ['筋トレさぼった']) | (prev_row == ['サボった']) | (prev_row == ['さぼった'])) &
        ((last_row == ['筋トレサボった']) | (last_row == ['筋トレさぼった']) | (last_row == ['サボった']) | (last_row == ['さぼった']))):
        prompt = prompt2
    else:
        prompt = prompt1

# ChatGPTに送るメッセージを作成
send_message = [prompt, user_message]

# ChatGPT APIにリクエストを送信して応答を取得
response = requests.post(
    'https://api.openai.com/v1/chat/completions',
    headers={
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {OPENAI_API_KEY}',
    },
    json={
        'messages': send_message,
        'model': 'gpt-3.5-turbo-1106',
    }
)

# APIからのレスポンスをJSON形式で返す
completion = response.json()

# ChatGPTの応答を取得
reply = completion['choices'][0]['message']['content']

LINEに応答を送信

ChatGPTから得られた応答をLINE Botを介してLINEに送信します。

Lambda.py
# LINEに応答を送信
url = 'https://api.line.me/v2/bot/message/reply'
headers = {
    'Content-Type': 'application/json',
    'Authorization': f'Bearer {line_config["channelAccessToken"]}'
}
data = {
    'replyToken': message_event['replyToken'],
    'messages': [{'type': 'text', 'text': reply}]
}
response = requests.post(url, headers=headers, json=data)

正常終了

Lambda関数の正常終了を示すレスポンスを返します。

Lambda.py
# 正常終了を示すレスポンス
return {
    'statusCode': 200,
    'body': "successfully"
}

まとめ

上記のスクリプトをまとめたものです。

lambda.py
import os
import json
import boto3
import requests
import csv
import io

# GPTのAPIKEY
OPENAI_API_KEY = os.environ['OPENAI_API_KEY']

# AWS Lambdaハンドラー関数
def lambda_handler(event, context):
    
    # LINE Botの設定
    line_config = {
        'channelSecret': os.environ['CHANNEL_SECRET'],
        'channelAccessToken': os.environ['CHANNEL_ACCESS_TOKEN'],
    }
    
    # Lambdaクライアントの作成
    lambda_client = boto3.client('lambda')
    
    # S3のバケット名とファイルキーを取得
    bucket_name = 'BUCKET_NAME'
    file_key = 'FILE_KEY'

    # S3オブジェクトを取得
    s3 = boto3.client('s3')
    
    # S3からファイルをダウンロードしてcsvデータを読み取る
    response = s3.get_object(Bucket=bucket_name, Key=file_key)
    data = response['Body'].read().decode('utf-8')
    
    # イベントからメッセージイベントを取得して処理
    body = json.loads(event['body'])
    for message_event in body['events']:
        # テキストメッセージの場合のみ処理
        if message_event['type'] == 'message' and message_event['message']['type'] == 'text':
            # ユーザーメッセージを抽出
            user_message = {
                'role': 'user',
                'content': message_event['message']['text']
            }
            
            #更新されたcsvデータを作成
            updated_contents = io.StringIO()
            writer = csv.writer(updated_contents)
            
            #元データの書き込み、データ数カウント
            csv_data = csv.reader(io.StringIO(data))
            num_lines = 1
            for row in csv_data:
                merged_row = ["".join(row)]
                writer.writerow(merged_row)
                num_lines += 1
            
            #新しいデータ
            new_data = str(message_event['message']['text'])
            
            #新しいデータを追加
            merged_new_data = ["".join(new_data)]
            writer.writerow(merged_new_data)
            
            #最後の行とその直前の行のデータを取得
            last_row = None
            prev_row = None
            csv_data = csv.reader(io.StringIO(data))
            for row in csv_data:
                prev_row = last_row
                last_row = row
            
            #S3に更新されたデータをアップロード
            s3.put_object(Bucket=bucket_name, Key=file_key, Body=updated_contents.getvalue())
            
            #promptの設定
            prompt1 = {
                "role":"system",
                "content":"「メニュー」と来たら、20分程度でできる自重トレーニングのメニューとその説明を教えてください"
                          "「筋トレしたよ」と来たら、ぶっ飛んだように変なくらいユーモアのある感じで相手を褒めて、プロテイン摂取も促してください。"
                          "「筋トレサボった」と来たら、軽く指摘してください"
            }
            prompt2 = {
                "role":"system",
                "content":"「筋トレサボった」と来たら、とても怖い感じでめっちゃ怒ってください"
            }
            #行数で分岐
            if num_lines < 3:
                prompt = prompt1
            elif num_lines >= 3:
                #3行のデータで分岐
                if (((merged_new_data == ['筋トレサボった']) | (merged_new_data == ['筋トレさぼった']) | (merged_new_data == ['サボった']) | (merged_new_data == ['さぼった'])) &
                    ((prev_row == ['筋トレサボった']) | (prev_row == ['筋トレさぼった']) | (prev_row == ['サボった']) | (prev_row == ['さぼった'])) &
                    ((last_row == ['筋トレサボった']) | (last_row == ['筋トレさぼった']) | (last_row == ['サボった']) | (last_row == ['さぼった']))):
                    prompt = prompt2
                else:
                    prompt = prompt1                    
            
            #ChatGPTに送るメッセージを作成
            send_message = [prompt, user_message]
            
            #ChatGPT APIにリクエストを送信して応答を取得
            response = requests.post(
                'https://api.openai.com/v1/chat/completions',
                headers={
                    'Content-Type': 'application/json',
                    'Authorization': f'Bearer {OPENAI_API_KEY}',
                },
                json={
                    'messages': send_message,
                    'model': 'gpt-3.5-turbo-1106',
                }
            )
            
            # APIからのレスポンスをJSON形式で返す
            completion = response.json()
            
            # ChatGPTの応答を取得
            reply = completion['choices'][0]['message']['content']
            
            # LINEに応答を送信
            url = 'https://api.line.me/v2/bot/message/reply'
            headers = {
                'Content-Type': 'application/json',
                'Authorization': f'Bearer {line_config["channelAccessToken"]}'
            }
            data = {
                'replyToken': message_event['replyToken'],
                'messages': [{'type': 'text', 'text': reply}]
            }
            response = requests.post(url, headers=headers, json=data)
    
    # 正常終了を示すレスポンス
    return {
        'statusCode': 200,
        'body': "successfully"
    }

詰まったところ

今回、初めてAWS Lambdaを使用したのですが、初期設定のタイムアウトが3秒ということを知らず、ずっと気づかないまま1日くらい無駄にしてしまいました。初めてAWS Lambdaを使う方は気をつけてください。

4586D74E-C01E-418B-AE9A-79206469AE5B.jpeg

4
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
4
1