【ミーア】ChatGPTとCloud Functions, Firestoreを用いて会話記憶するLINE Bot開発②：コード実装

前回こちらの記事で、LINE DevelopersアカウントとCloud Functionsの設定までを記載した。

【ミーア】ChatGPTとCloud Functions, Firestoreを用いて会話記憶するLINE Bot開発①：環境設定まで

今回は、続きを記載。主に実装するコードに関して。

CloudFunctions内に実装するコード

main.py

import os
from datetime import datetime

from flask import Request, abort
from google.cloud import firestore
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage
from openai import OpenAI

line_bot_api = LineBotApi(os.environ["LINE_CHANNEL_ACCESS_TOKEN"])
handler = WebhookHandler(os.environ["LINE_CHANNEL_SECRET"])
db = firestore.Client()
client = OpenAI()

USE_HISTORY = True

def get_previous_messages(user_id: str) -> list:
    query = (
        db.collection("users")
        .document(user_id)
        .collection("messages")
        .order_by("timestamp", direction=firestore.Query.DESCENDING)
        .limit(3)
    )
    return [
        {
            "chatgpt_response": doc.to_dict()["chatgpt_response"],
            "user_message": doc.to_dict()["user_message"],
        }
        for doc in query.stream()
    ]


def format_history(previous_messages: list) -> str:
    return "".join(
        f"ユーザー: {d['user_message']}\nアシスタント: {d['chatgpt_response']}\n"
        for d in previous_messages[::-1]
    )


def generate_response(user_message: str, history: str) -> str:
    if USE_HISTORY and history:
        history_section = f"これまでの会話履歴:\n{history}\n"
    else:
        history_section = ""
    prompt = f"""
    【ミーアじゃっどのしゃべり方について指示を入力する】
    {history_section}
    ユーザーからのメッセージ: {user_message}
		"""

    # OpenAI APIを使用して応答を生成
    chatgpt_response = client.chat.completions.create(
        model="gpt-3.5-turbo-1106",
        messages=[{"role": "system", "content": prompt}, {"role": "user", "content": user_message}],
        temperature=0.2,
    )

    # 応答のテキスト部分を取得
    return chatgpt_response.choices[0].message.content


def reply_to_user(reply_token: str, chatgpt_response: str) -> None:
    line_bot_api.reply_message(reply_token, TextSendMessage(text=chatgpt_response))


def save_message_to_db(user_id: str, user_message: str, chatgpt_response: str) -> None:
    doc_ref = db.collection("users").document(user_id).collection("messages").document()
    doc_ref.set(
        {
            "user_message": user_message,
            "chatgpt_response": chatgpt_response,
            "timestamp": datetime.now(),
        }
    )


def main(request: Request) -> str:
    signature = request.headers["X-Line-Signature"]
    body = request.get_data(as_text=True)

    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        abort(400)

    return "OK"


@handler.add(MessageEvent, message=TextMessage)
def handle_message(event: MessageEvent) -> None:
    user_id = event.source.user_id
    user_message = event.message.text

    if USE_HISTORY:
        previous_messages = get_previous_messages(user_id)
        history = format_history(previous_messages)
    else:
        history = None
    chatgpt_response = generate_response(user_message, history)

    reply_to_user(event.reply_token, chatgpt_response)
    if USE_HISTORY:
        save_message_to_db(user_id, user_message, chatgpt_response)

requirements.txt

line-bot-sdk
openai
google-cloud-firestore

コード全体の流れ

1. main関数:

   • Cloud Functionsのエントリポイント。LINEプラットフォームからのリクエストを受け取る。

   • リクエストヘッダからLINEプラットフォームが送信したX-Line-Signature署名を取得する。

   • 署名とリクエストボディをhandler.handleメソッドに渡して、署名の検証とイベント処理を行う。署名を検証する際にChannel secretを使用している

続きは、こちらで記載しています。

【ミーア】ChatGPTとCloud Functions, Firestoreを用いて会話記憶するLINE Bot開発②：コード実装