周回遅れも甚だしいネタですが、OpenAI APIを使ってチャットbotを実装しようとする時、会話の記憶を持たせたい（何もしないとステートレスな会話になる）というのはよくあるユースケースだと思います。
AWS上でLambdaとDynamoDBを使い、会話の記憶を持たせたLINEチャットボットを実装したので、自分への備忘録的に記事を残しておきます。
仕様
- 会話を記憶する数は5往復（10メッセージ）とするが、適宜変更もできるようにする
- 会話履歴を記憶させる媒体としてAWS DynamoDBとする
- DynamoDBへは1ユーザー1レコードとし、保存対象の会話をJSONで1フィールドにまとめて保存する
会話履歴の保存数を5往復とした理由
会話履歴の保存数を多く（長く）すればするほど、理論的には文脈の理解度は上がるはずですが、パラメータを弄りながらテストしていると、あまり古い会話を覚えておいても文脈理解にはかえって邪魔になることが分かりました。
また、当然のことながら記憶しておく履歴が長いほどTokenの消費量も多くなるため、コスト的にも不利になります。
今回は、記憶数を当初10往復で検証スタートし、少しずつ記憶数を減らしていって会話の整合性を確認していきました。
その結果、GPT-4であれば5往復程度でも十分に文脈を読み取った会話になることが分かりました。体感では理解度90%程度といった感じです。これを10往復まで増やすと+5%程度は文脈理解度が向上したようにも感じましたが、正直チャットボットにそこまでは求めていないというユーザーの声もあり、コストとのバランスを取った結果、この数値になりました。
1ユーザー1レコードとした理由
DynamoDBにはあまり詳しくないのですが、料金設定を見ていると、1会話ごとに複数レコードの読み出しが発生するのはコスト的に不利なのかな、と。
特に会話履歴のランダムリードが必要な訳でもなく、古い会話履歴を保存しておく必要性もないのであれば、必要な分だけを1フィールドにJSONで固めておけば、コスト的にもいいのではないかと思った次第。（もし誤りあればフォローくださると助かります）
【余談】長期記憶の実装実験
ユーザーテストの段階で、会話の整合性とは別に、長期記憶を持っているといいのではないかという意見があり、長期記憶の実装も実験してみました。
ざっくり言えば、ChatGPTへのプロンプトに、ユーザーの発言内容から長期に記憶すべきものを選別するような指示を含め、その結果をJsonないしMarkDownで出力させるという試みです。
長期記憶の選別と取り出しはうまく行ったのですが、プロンプトに余計なものが混ざるせいか、本文に関する指示は変えていないにもかかわらず、bot発言の自然さがやや失われる傾向が見られました。
ユーザー体験的には長期記憶を持たせることによるメリットより、botの発言に違和感があるデメリットの方が大きかったため、今回は実装を見送りました。
機会があれば再チャレンジしてみたいと思っています。
ざっくりとデータの流れ
| ユーザー | DynamoDB | OpenAI API | LINE Messaging API |
|---|---|---|---|
| メッセージ送信 | |||
| 会話履歴取得 | |||
| Chat Completion | |||
| Botメッセージ送信 |
DB設計
メインテーブル
| Field | Type | Description | |
|---|---|---|---|
| PK | userId | str | LINEのユーザーID |
| registDate | str | 友だち追加日時 | |
| messages | str | 会話履歴(JSON形式) | |
| timeStamp | str | 最終会話日時 |
私の普段の業務としてはゴリゴリのRDBMS使いのため、DynamoDBのようなキーバリューストア構造のDBMSにはあまり慣れていません。
そのため、このテーブル設計が正しいのかは分かりませんが、まあ動いているのでヨシという感じです。
userId以外での問い合わせは発生しない想定のため、外部キーなどの設定はしていません。
LINE側設定
（略）
会話履歴管理クラス
会話履歴を管理するために簡単なクラスを作成しました。
正直やっていることは、DBへのストアと読み出しのみです。
コンストラクタ
class DB:
def __init__(self, table_name: str):
self.dynamodb = boto3.resource('dynamodb')
self.table = self.dynamodb.Table(table_name)
会話履歴の問い合わせ
def query_record(self, userId: str) -> Dict[str, Any]:
"""
会話履歴を取得する関数。
Args:
userId (str): 取得するユーザーのID。
Returns:
Dict[str, Any]: ユーザー情報と会話履歴
"""
response = self.table.get_item( Key={'userId': userId} )
log.logger.debug(f"Found user record: {response}")
messages = None
timeStamp = None
if 'Item' in response:
if 'messages' in response['Item']:
messages = response['Item']['messages']
if 'timeStamp' in response['Item']:
timeStamp = response['Item']['timeStamp']
return {
"timeStamp": timeStamp,
"messages": messages
}
会話履歴の保存
def save_record(self, userId: str, messages: List[Dict[str, str]]) -> Dict[str, Any]:
"""会話履歴を保存する関数
Args:
userId (str): ユーザーID
messages (List[Dict[str, str]): 会話履歴
Returns:
Dict[str, Any]: DynamoDBに保存されたレスポンス
"""
response = self.table.put_item(
Item={
"userId": userId,
"timeStamp": f"{datetime.now().timestamp()}",
"messages": json.dumps(messages, ensure_ascii=False),
}
)
return response
メッセージ受信～Botメッセージ送信までの流れ
実際のbot実装のためのコードのイメージは以下の通りです。
読みやすさのため、初期化やエラー処理、API回りのラッパー関数などは省いたコードです。
# 会話履歴を取得
history = []
rec = db.query_record(user_id)
if "messages" in rec and rec["messages"]:
history = rec["messages"]
# 新しいユーザー発言を追加
history.append({"role": "user", "content": original_prompt})
# 古い履歴を削除
if len(history) >= 10:
history = history[(len(history) - 10):]
# 会話履歴を保存
db.save_record(user_id, history)
# 会話履歴にユーザーの質問とシステムプロンプトを追加する
history.append({"role": "user", "content": original_prompt})
history.append({"role": "system", "content": SYSYTEM_PROMPT})
# プロンプトをOpenAI APIに送信
response: str = query_completion(openai_secret_key, history, model_name)
# 受信した回答をLINEに送信
LINE.send_reply(headers, reply_token, response, is_test)
# システムプロンプトを削除してから会話履歴を保存
history.pop()
history.append({"role": "assistant", "content": response})
if len(history) >= 10:
history = history[(len(history) - 10):]
db.save_record(user_id, history)
まとめ
LINE Botとしてユーザーとの会話が成り立つよう、会話の記憶機能を実装してみました。
やってみると大したことはなく、意外とあっさりと実装ができました。
この機能を実装したLINE botについて→