前回こちらの記事で、LINE DevelopersアカウントとCloud Functionsの設定までを記載した。
今回は、続きを記載。主に実装するコードに関して。
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コード全体の流れ
main関数:- Cloud Functionsのエントリポイント。LINEプラットフォームからのリクエストを受け取る。
- リクエストヘッダからLINEプラットフォームが送信した
X-Line-Signature署名を取得する。 - 署名とリクエストボディを
handler.handleメソッドに渡して、署名の検証とイベント処理を行う。署名を検証する際にChannel secretを使用している
handle_message関数:- LINEからのメッセージイベントを処理するために
@handler.addデコレータによってMessageEventに紐づけられている。 - ユーザーIDを取得し、ユーザーからのメッセージ（
user_message）を抽出。 USE_HISTORYがTrueの場合、get_previous_messages関数を呼び出して過去のメッセージを取得し、format_history関数でフォーマットされた会話履歴をhistory変数に格納する。履歴がない場合はhistoryはNoneになる。generate_response関数を使用して、ユーザーメッセージと会話履歴を元にOpenAI APIを通じて応答を生成する。- 生成された応答を
reply_to_user関数を通じてLINEユーザーに送信する。 - 応答をFirestoreに保存するために、
save_message_to_db関数が呼び出されます。
- LINEからのメッセージイベントを処理するために
get_previous_messages関数:- Firestoreデータベースから特定のユーザーIDに関連する最新の3件のメッセージを取得。
format_history関数:- get_previous_messages関数から取得したメッセージのリストを、適切な形式の文字列に変換。
generate_response関数:- 応答を生成するために必要なプロンプトを作成し、OpenAI APIにリクエストを送信する。リクエストには、会話履歴とユーザーからの最新のメッセージが含まれる。
reply_to_user関数:- LINE APIを使用して、生成された応答をユーザーに送信する。
save_message_to_db関数:- ユーザーのメッセージと生成された応答をFirestoreデータベースに保存。
応答を生成しているgenerate_response関数に関して詳しく見ていく。
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.contentOpenAI APIを使用して応答を生成
下記のコード部分が、OpenAI APIを使用して応答を生成している部分。
OpenAIのAPIは定期的に更新され、変更されるので、最新のAPI呼び出し方法を公式サイトで確認する必要あり。bot作成して動かないときの原因の1つは、OpenAIのAPIの呼び出し関数のことがある。直近だと、2023年11月にAPIのアップデートがあった。
OpenAIはAPIとして、ChatCompletions, Embeddings, Imagesなど様々なAPIを提供しているが、今回使うのはChatCompletions 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,
)OpenAI modelについて
今回は、modelの部分で、gpt-3.5-turbo-1106を指定している。
gpt-3.5-turbo-1106に関しては、下記記事がわかりやすかった。簡単に言えば、gpt-3.5-turboと比較して、安くなり、かつ、沢山の情報量を扱うことができるようになった＋回答スピードが速くなった。
吉野家の牛丼の「うまい、安い、早い」みたいな感じか（違。
ちなみに、finetuningした場合は、そのモデル名を記載する。finetuningに関しては、こちらの記事で記載。
応答のテキスト部分を取得
Chat Completionsで返ってくるオブジェクトのうち、テキスト部分を取得する。
返ってくる型はChatCompletion型のオブジェクト。
# 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,
)
print(chatgpt_response)
# 戻り値の型を出力
print(type(chatgpt_response))上記のように戻り値の型を出力して、ログを見ると
<class 'openai.types.chat.chat_completion.ChatCompletion'>と記載されている。
chatgpt_responseは
ChatCompletion(id='chatcmpl-8T6yWaDfV7t4o5YdpfMm73QRcO8wc', choices=[Choice(finish_reason='stop', index=0, message=ChatCompletionMessage(content='おはようよ。頭痛いんかい？それは大変やね。お医者さんに行った方がええかもしれんが。お大事にな。', role='assistant', function_call=None, tool_calls=None))], created=1701949016, model='gpt-3.5-turbo-1106', object='chat.completion', system_fingerprint='fp_eeff13170a', usage=CompletionUsage(completion_tokens=51, prompt_tokens=2601, total_tokens=2652))ちなみに、lineでメッセージした内容はこちら。返信はChatCompletionオブジェクトの属性の中のcontentに入ってきている。
公式サイトで分かりやすく記載されている、戻り値はこちら
{
"id": "chatcmpl-123",
"object": "chat.completion",
"created": 1677652288,
"model": "gpt-3.5-turbo-0613",
"system_fingerprint": "fp_44709d6fcb",
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": "nnHello there, how may I assist you today?",
},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": 9,
"completion_tokens": 12,
"total_tokens": 21
}
}pythonでオブジェクトの属性にアクセスするときは、「.」を使うので、下記のように繋げていく。choicesに関しては、リスト型[]なので、リストの最初の要素を参照する（indexに0番が記載されている）ために、choices[0]を使う。choicesのリストが複数入ってくることはなさそうだが。
return chatgpt_response.choices[0].message.content個人的にハマったところ：戻り値が辞書型だと勘違い
公式サイト見ると、下記のように記載があり、しっかりドキュメントを読むと、returnsのところにReturns a chat completion objectとオブジェクトであることが記載されているのだが、右側のresponseの部分を見て、辞書型で返ってきていると早合点してしまい、
return chatgpt_response["choices"][0]["message"]["content"]と応答のテキストへのアクセスを書いてしまい、デプロイしてもなんで取得されないんだろうと悩んだ。
Firestoreを用いて、会話履歴を保存し呼び出す
Firestoreとは？
FirestoreはGoogle Cloud Platformが提供するNoSQL（非関係型）データベースサービスである。その主な特徴は以下の通り。
高速な書き込みと読み出し：
- Firestoreは読み書きの応答時間が短く、リアルタイムでのデータ更新に適している。
非関係型（NoSQL）データベース：
- Firestoreは非関係型データベースに分類され、データはJSON形式で格納される。これによりデータを階層的に管理し、柔軟なデータ構造を持たせることが可能だ。
データの格納形式：
- Firestoreでのデータ格納はコレクションとドキュメントという形で行われる。ドキュメントはJSON形式で、複雑なデータ構造を持つことができる。コレクションはドキュメントのグループ。
SQLとの比較：
- 従来のリレーショナルデータベース（例：Cloud SQL、BigQuery）はデータを行と列を持つテーブルに格納し、SQL（Structured Query Language）を用いてデータを分析・操作する。
- Firestoreでは、SQLの使用は難しく、データの構造もテーブル形式ではないが、データの書き込みと読み出しは非常に高速。
使用シナリオ：
- Firestoreは、会話の履歴のような構造化されていないデータを迅速に保存し、取得する用途に適している。データ分析が主目的でない場合、データの迅速な入出力を求めるシナリオにはFirestoreが適している。
というわけで、LINE Botのようなアプリケーションで会話の履歴をリアルタイムに記録し、取得するためにFirestoreを使用するのは相性が良さそう。特に今回、会話履歴を分析するわけではない。
会話履歴を保存
こちらで、ユーザーとの会話履歴（ユーザーID、ユーザーからのメッセージ、およびChatGPTからの応答）をFirestoreに保存している。
from google.cloud import firestore
db = firestore.Client()
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(),
}
)PythonでFirestoreを使用するには、google-cloud-firestoreライブラリが必要なので、requirements.txtにgoogle-cloud-firestore を追加する。そして、google-cloud-firestoreライブラリからfirestoreモジュールをインポートし、db = firestore.Client() で、Firestoreサービスへの接続を表すクライアントオブジェクトを作成する。
ドキュメントリファレンスの作成:
- **
db.collection("users").document(user_id).collection("messages").document()を呼び出すことで、Firestoreデータベース内の特定のユーザーIDに紐づけられた"messages"コレクション内に新しいドキュメントリファレンスを作成。document()**メソッドに引数がないため、Firestoreは自動的に一意のドキュメントIDを生成する。
データの設定:
- **
doc_ref.set()**メソッドを使用して、作成したフィールドにデータを保存。ここで設定されるデータは辞書形式であり、以下のキーと値を含んでいる："user_message": ユーザーが送信したメッセージの内容。"chatgpt_response": ChatGPTが生成した応答の内容。"timestamp": このデータが保存される時点の日時。**datetime.now()**を使用して現在の日時を取得。
データの書き込み:
- **
set**メソッドは、与えられた辞書の内容でドキュメントを上書き（もしくは新規作成）する。
CloudFunctionsをデプロイして、うまく実行できると、下記のようにusersコレクションの中に、LINEユーザー別（user_id）にデータがDocumentとして保存される。
その中に、user_idに紐づいてmessagesコレクションが作成され、その中にDocumentとして、ユーザーの送信メッセージ、ChatGPTの応答内容、データ保存日時が保存される。
Collections→document→collection→documentという流れで、データをjson形式で幾つでも階層的に管理できる（jsonの中にjsonデータを入れられる）。データはフィールド（ドキュメント内の個々のデータ項目）に格納する。
会話履歴の呼び出し
get_previous_messages関数で、指定されたユーザーIDに関連する最新のメッセージをFirestoreデータベースから取得する。
取得したメッセージを、タイムスタンプ順に降順に並べ替える。これにより、最新のメッセージが先に来る。そして、.limit(3)で、取得するメッセージの数を最新の3件に限定する。
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()
]
ドキュメントデータの辞書型への変換:
query.stream()によって返されるストリームをイテレートすると、各イテレーションでFirestoreドキュメントのスナップショットが得られる。- 各ドキュメントスナップショットに対して
.to_dict()メソッドを呼び出すことで、ドキュメントのデータをPythonの辞書型に変換できる。この辞書には、ドキュメントのフィールド名と値がキーと値として格納されている。
queryは、Firestoreクエリオブジェクトの内部表現を示す。特定のユーザーのmessagesサブコレクションに対するクエリを表している。
Query(reference=CollectionReference(Database(…), 'users/user_id_1/messages'), orders=[…], limit=3, …)query.stream()は、Pythonのジェネレータオブジェクト。クエリの結果を反復処理するためのイテレータ。
<generator object Query.stream at 0x7f4d8c3d4150>docは、Firestoreのドキュメントスナップショットを表している。各ドキュメントスナップショットには、そのドキュメントのデータが含まれており、データは辞書型で表現される。
<DocumentSnapshot (…), data: {'user_message': 'こんにちは', 'chatgpt_response': 'こんにちは！お手伝いできることはありますか？', 'timestamp': datetime.datetime(2023, 3, 1, 12, 0, 0)}>
<DocumentSnapshot (…), data: {'user_message': '最新のニュースは？', 'chatgpt_response': '今日のトップニュースは...', 'timestamp': datetime.datetime(2023, 3, 1, 12, 5, 0)}>この関数を実行した戻り値は、下記のように、辞書型（リスト）になる。最新の3件のメッセージがそのタイムスタンプの降順（最新から最も古い順）に含まれる。
[
{
"chatgpt_response": "今日の天気は晴れの予報です。",
"user_message": "天気はどうですか？",
},
{
"chatgpt_response": "今日のトップニュースは...",
"user_message": "最新のニュースは？",
},
{
"chatgpt_response": "こんにちは！お手伝いできることはありますか？",
"user_message": "こんにちは",
}
]会話履歴をプロンプトに組み込んで記憶する
下記のformat_history関数で、Firestoreから取得した過去のメッセージのリストを文字列にフォーマットしている。previous_messages[::-1]により、リストを逆順にし、メッセージを時系列順（古い順）に並べかえている。
def format_history(previous_messages: list) -> str:
return "".join(
f"ユーザー: {d['user_message']}nアシスタント: {d['chatgpt_response']}n"
for d in previous_messages[::-1]
)戻り値は下記のようになる。これで、ユーザーとアシスタント間の会話の流れが自然な時系列順で表現され、過去の会話を文脈として理解しやすくなる。
ユーザー: こんにちは
アシスタント: こんにちは！お手伝いできることはありますか？
ユーザー: 最新のニュースは？
アシスタント: 今日のトップニュースは...
ユーザー: 天気はどうですか？
アシスタント: 今日の天気は晴れの予報です。generate_response関数では、上述のformat_history関数から得られた履歴のテキスト（history）を使用して、応答を生成する。
履歴セクションの構築:
- **
USE_HISTORYがTrueで、かつhistoryが空でない場合、history_sectionに"これまでの会話履歴:\\n{history}\\n"**という文字列が格納される。 - これにより、履歴データがプロンプトに含まれ、会話の文脈が考慮される。
プロンプトの構築:
- 最終的なプロンプトには、履歴セクションと現在のユーザーメッセージ（
user_message）が含まれる。 - このプロンプトはOpenAI APIに渡され、応答の生成に使用される。
このように、history（履歴）は、ユーザーとボット間の過去の会話を考慮して、より関連性の高い応答を生成するために重要な役割を果たす。履歴を利用することで、ボットは以前のやり取りを「記憶」し、続きの会話や関連する内容に適切に応答できるようになる。
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コードの全体の解説は以上。
エントリポイントで「main」を選択して（実行する関数はmainなので）、「保存して再デプロイ」ボタンをクリックする。
2-3分くらいでデプロイが完了する。
「保存して再デプロイ」ボタンの上部に表示されている、エンドポイントURLが、LINE Botで設定するWebhookに記載するURLになるので、コピーする。
LINE BotでWebhook設定
LINE Developersコンソールの「Messaging API設定」タブのWebhook設定欄に、先ほどのCloud FunctionsのエンドポイントURLを貼り付ける。
その下の「Webhookの利用」の設定をONにする
「検証」ボタンを押して、成功と表示されると、OK。
指定されたWebhook URLが有効であり、LINEプラットフォームからそのURLへのリクエストが正常に送信された＋署名の検証も正常に行われたことを意味する。
上記画像の②のみが成功したことを指しており、下記は検証されていない。
Cloud Functions内部の処理:
- LINEからのリクエストを受け取った後のCloud Functions内での具体的な処理内容。例えば、データベースへの書き込み、応答メッセージの生成などの処理は検証されない
Cloud FunctionsからLINEへの返信:
- Cloud FunctionsからLINEのAPI（LINE Channel Access Tokenを使用）への応答（ユーザーへのメッセージ送信など）も、この検証プロセスでは確認されない。
実行し検証
無事うまくコードを実行できると、下記のようにLINEbotで会話できる。
「私は一つ前の質問で、なんて聞きましたか？」の質問に対して、「焼酎の銘柄、いくつか教えて」と返信きており、会話履歴を無事保存してくれていることがわかる。
というわけで、無事、会話履歴を記憶して文脈に応じた会話をするLINE BOTの構築に成功。次は、プロンプト設定で鹿児島弁を話せるように試行錯誤してみる。
コメント