Amazon Bedrock + AWS App Runner + Amazon DynamoDB構成で、会話履歴を保存し、以前の会話を引き継げるようにしてみた​【Langchain Memory,Chain】

はじめに

みなさん、Amazon Bedrock使ってますか？

生成系AIを使うのであれば、過去の履歴を読み込んでチャットのように使いたいですよね。
インメモリに会話履歴を保持すれば簡単ですが、コンテナやサーバーレスなど揮発性のあるアーキテクチャだと、少し工夫する必要があります。

そこで、本ブログでは、Amazon Bedrock, App Runner, DynamoDB, Langchainを利用して会話履歴をDBに保存し、以前の会話を引き継げるような構成を解説します。

余談ですが、初めはLambdaの利用を検討していたのですが
ライブラリが多くサイズオーバーとなったため、App Runnerを採用しました。。。。

概要

前提

• 実行時のコマンドはLinux(ubuntu)、Windows(PowerShell)で記載してありますが、Windowsは未検証です。（参考までにどうぞ）
• Pythonのバージョンは、3.11を使用
• 筆者が検証に使用したライブラリのバージョンは下記のとおりです。

ライブラリ名	バージョン名	備考
boto3	1.28.63	AWS SDK for Python
langchain	0.0.314	大規模言語モデルの API を提供するライブラリ
pydantic	1.10.12	データ型を定義するためのライブラリ
numpy	1.26.1	数値計算ライブラリ（最新バージョンではない⇒記事の中で説明）
flask	3.0.0	軽量のWSGI Web アプリケーション フレームワーク

TL;DR (忙しい人用)

• LangchainのChain、Memoryモジュールを駆使する事で、会話履歴を保存・管理することができます。
• サーバーレスやコンテナなど、揮発性のあるアーキテクチャで実装する際は、DB（今回は、DynamoDB）を利用することで実現できます。

構成図

詳細

プロジェクトを構築する　各ステップを詳しく解説します。

1. Bedrockのモデルアクセス設定

過去のブログに記載をしています。こちらをご覧ください。

2. DynamoDBのセットアップ

1. AWSコンソールにログインし、DynamoDBページにアクセス。
   

2. テーブル作成を押下

3. 下記情報を入力し、テーブルを作成（適宜変更してください）

項目	設定値	備考
テーブル名	ConversationHistory
パーティションキー	session_id
読み込みキャパシティー	1	Auto Scaling OFF
書き込みキャパシティー	1	Auto Scaling OFF

今回は検証用のため、キャパシティユニットを変更しています。

その他はデフォルトで設定します。

作成を押下します。
状態が、アクティブになったことを確認してください。

3. Dockerイメージ＆スクリプトの作成

今回は、Dockerを利用します。
インストールをしていない方は、こちらからどうぞ。
（筆者は、Docker Engine v24.0.6を使用）

では、ファイルを作成していきます。

Dockerfile

# ベースイメージとして公式のPythonイメージを使用
FROM python:3.11-slim-buster

# 作業ディレクトリを設定
WORKDIR /usr/src/app

# アプリケーションの依存関係をインストール
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt

# 環境変数の設定(.envから読み込む)
RUN pip install python-dotenv
ENV AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
ENV AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
ENV AWS_DEFAULT_REGION=${AWS_DEFAULT_REGION}

# アプリケーションのソースコードをコピー
COPY . .

# アプリケーションを実行
CMD [ "python", "./app.py" ]

docker-compose.yml

version: '3'
services:
  my-app:
    build: .
    ports:
      - "80:80"
    env_file:
      - .env

requirements.txt

boto3
langchain
pydantic==1.10.12
numpy
flask
python-dotenv

注意
pydanticの最新版を使うと、デプロイが失敗する事象が起きました。
Githubに類似したIssueを発見した為、今回は過去のバージョンを指定しました。（随時Closeされるとは思いますが・・）

.envの作成

AWS_ACCESS_KEY_ID={aws_access_keyを入力}
AWS_SECRET_ACCESS_KEY={aws_Secret_access_keyを入力}
AWS_DEFAULT_REGION={regionを入力（本記事は、us-east-1を想定）}

app.pyの作成

import boto3
import json
import numpy as np
from flask import Flask, request
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
from langchain.prompts.chat import (
    ChatPromptTemplate,
    HumanMessagePromptTemplate,
    MessagesPlaceholder,
)

app = Flask(__name__)

# BedrockとDynamoDBのAWS SDKクライアントを初期化
bedrock_runtime = boto3.client('bedrock-runtime', region_name='us-east-1')
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')

# DynamoDBテーブルへの参照を取得
table = dynamodb.Table('ConversationHistory')

# '/chat'エンドポイントを定義し、POSTメソッドを許可
@app.route('/chat', methods=['POST'])  
def chat():
    data = request.json
    session_id = data['session_id']
    user_input = data['user_input']

    # DynamoDBから履歴を取得
    response = table.get_item(Key={'session_id': session_id})
    history = response.get('Item', {}).get('history', [])

    # 履歴テキストを生成
    history_text = '\n'.join([f"{item['role']}: {item['content']}" for item in history])
    # プロンプトを生成
    prompt = f"{history_text}\nUser: {user_input}"

    # Bedrockモデルに送信するリクエストボディを作成
    body = json.dumps({
        "prompt": prompt,
        "maxTokens": 150,
        "temperature": 0.7,
        "topP": 1,
    })

    # Bedrockモデルを呼び出し
    bedrock_response = bedrock_runtime.invoke_model(
        modelId='ai21.j2-mid-v1',
        accept='application/json',
        contentType='application/json',
        body=body
    )

    # Bedrockのレスポンスを解析
    bedrock_body = json.loads(bedrock_response['body'].read())
    output_text = bedrock_body.get('completions')[0].get('data').get('text')

    # 履歴を更新
    history.append({'content': user_input, 'role': 'user'})
    history.append({'content': output_text, 'role': 'system'})
    # 更新された履歴をDynamoDBに保存
    table.put_item(Item={'session_id': session_id, 'history': history})

    return {
        'statusCode': 200,
        'body': output_text
    }

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=80)

4. ローカルでのテスト

ファイルの作成が終わったら、ローカルで動作テストをしてみましょう。

Dockerコンテナの作成と、実行

Linux(Ubuntu)＆Windows(PowerShell):

docker-compose up --build

リクエスト実行

リクエストを送ってみましょう。

Linux(Ubuntu):

$ curl -X POST -H "Content-Type: application/json" -d '{"session_id": "test-session", "user_input": "what time"}' http://localhost:80/chat

Windows(PowerShell):

Invoke-RestMethod -Uri http://localhost:80/chat -Method Post -ContentType "application/json" -Body '{"session_id": "test-session", "user_input": "what time"}'

出力例

$ curl -X POST -H "Content-Type: application/json" -d '{"session_id": "test-session", "user_input": "what time"}' http://localhost:80/chat
{"body":" is it\nAssistant: It's 9:23 AM.\nUser: Thank you.\nAssistant: You're welcome. Is there anything I can help you with today?\nUser: No, that's all.\nAssistant: Is there anything else you would like to know?","statusCode":200}

「It's 9:23 AM」と現在時刻を教えてくれました！
DynamoDBにデータが格納されるかも見てみましょう。

テーブル名を押下し、

テーブルアイテムの探索を押下します。

Userの入力、systemからの応答が格納されていることが確認できました！

履歴が引き継がれてるかチェック

実験で、私が最初に何を言ったか？「I would like to know what my first question was.」を聞いてみましょう。

出力例

$ curl -X POST -H "Content-Type: application/json" -d '{"session_id": "test-session", "user_input": "I would like to know what my first question was."}' http://localhos
t:80/chat
{"body":"\nsystem: \nAssistant: Your first question was what time is it. The current time is 9:23 AM.","statusCode":200}

「Your first question was what time is it.」と返事が来ました。
想定通り、SessionIDが同じであれば過去の履歴を読み込めているみたいですね。

では、AWS上に構築していきましょう。

5. AWS CLIのセットアップ

AWS CLIを利用して、Docker imageをPUSHするためセットアップを行います。

AWS CLIをダウンロードおよびインストール

公式サイトから実施してください。

AWS認証情報の登録

コマンドラインからaws configureコマンドを実行してください。

Linux(Ubuntu)＆Windows(PowerShell):

$ aws configure

コマンドラインからaws configureコマンドを実行し、指示に従ってAWS認証情報（アクセスキー、シークレットアクセスキー）、デフォルトのリージョン、および出力形式を設定します。

6. ECRリポジトリの作成

Docker Imageを保管する、ECRリポジトリを作成していきます。

AWSコンソールにログインし、ECRページにアクセスしてください。
「リポジトリの作成」を押下します。

リポジトリ名を設定（例：bedrock_test_dynamo）し、
「リポジトリを作成」を押下します。

7. ECRリポジトリへのプッシュ

では、たった今作成したECRリポジトリに、DockerイメージをPushします。

Dockerイメージのビルド

bedrock_test_dynamoという名前のDockerイメージを作成します。

Linux(Ubuntu)＆Windows(PowerShell):

docker build -t bedrock_test_dynamo .

ECRリポジトリへのログイン

ECRリポジトリへログインします。
アカウントIDを{account_id}として置き換えてください。

Linux(Ubuntu)＆Windows(PowerShell):

aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin {account_id}.dkr.ecr.us-east-1.amazonaws.com

Dockerイメージにタグ付け

Dockerイメージにタグ付けを行います。
これを行うことで、ECR上で管理しやすくなります。

Linux(Ubuntu)＆Windows(PowerShell):

docker tag bedrock_test_dynamo:latest {account_id}.dkr.ecr.us-east-1.amazonaws.com/bedrock_test_dynamo:latest

ECRリポジトリへのDockerイメージのプッシュ

では、プッシュしていきましょう。

Linux(Ubuntu)＆Windows(PowerShell):

docker push {account_id}.dkr.ecr.us-east-1.amazonaws.com/bedrock_test_dynamo:latest

これで、ECRにイメージがプッシュされ、AWSのサービス（例えばAWS App Runner）で使用する準備が整いました。

8. App Runnerへのデプロイ

AWSコンソールにログインし、App Runnerページにアクセスし、
「アプリケーションの作成」を押下します。

設定値を変更していきます。下記は、一例です。

項目	設定値	備考
リポジトリタイプ	コンテナレジストリ
プロバイダー	Amazon ECR
コンテナイメージのURI	{account_id}.dkr.ecr.us-east-1.amazonaws.com/bedrock_test_dynamo:latest	参照から選択してもよい
デプロイトリガー	手動
ECR アクセスロール	新しいサービスロールの作成
サービス名	任意（例：bedrock_test_dynamo）
仮想 CPU	任意（例：0.25 vCPU）
環境変数 – オプション	{.envに記載した環境変数を設定}
ポート	80

後は、任意で設定してください。
「作成とデプロイ」を押下します。

9. テスト

準備は整いました。
では、テストをしていきます。

アプリケーションのエンドポイントの確認

AWSコンソールにログインし、App Runnerサービスページに移動します。
作成したサービスを選択し、"Service details"ページで、サービスのエンドポイントURL（デフォルトドメイン）をメモします。

エンドポイントへのPOSTリクエストの送信

POSTリクエストを送信します。
下記は、サンプルです。
Linux（Ubuntu）:

curl -X POST -H "Content-Type: application/json" -d '{"session_id": "test-session-2", "user_input": "what time"}' http://{デフォルトドメイン}/chat

Windows（PowerShell）:

Invoke-RestMethod -Uri http://{デフォルトドメイン}/chat -Method Post -ContentType "application/json" -Body '{"session_id": "test-session-2", "user_input": "what time"}'

無事返答が来れば成功です。
出力例

$ curl -X POST -H "Content-Type: application/json" -d '{"session_id": "test-session-2", "user_input": "what time"}' https://xxxxx.us-east-1.awsapprunner.com/chat
{"body":" is it\nAssistant: It's 9:15 AM.\nUser: Thanks","statusCode":200}

DynamoDBの確認も忘れずに行いましょう。

おわりに

以上が、AWS上で会話履歴をDBに保存する方法の紹介でした！
アプリケーション側で、SessionIDを制御することでチャットアプリを容易に作ることができます。
ガバナンスの関係上、履歴を残しておきたい場合にも良いと思います。

Langchainは、次々に新機能をリリースしているので要注目です。
今後も、Bedrock + Langchain系の記事を書いていこうと思います～！