LambdとAPI Gatewayを用いて、S3にファイルをアップロードし、Amazon BedRockのナレッジベースの同期を行ったので、備忘録としてまとめます。クライアントがファイルをアップロードして、それを基にRAGできるようになります。

ナレッジベースの作成

S3 > バケットを作成
AmazonBedRock > ナレッジベース > ナレッジベースを作成
Choose data sourceでAmazon S3を選択
「s3を参照」から先ほど作成したS3をデータソースとして選択
適当な埋め込みモデルとベクトル次元を選択
ナレッジベースを作成
ナレッジベースに戻って、作成できたことをチェック

Lambdaを作成

Lambda > 関数作成
- 適当な関数名
- ランタイム：Python 3.8
- 実行ロール：新しいロール作成
ロールが自動で作成されるの、権限を付与します

ロールに権限を付与

IAM > ロール > 先ほど作成したロールを選択
許可を追加 > ポリシーをアタッチ
AmazonBedrockFullAccessとAmazonS3FullAccessを付与したロールを作成

API Gatewayを作成

Lambda > 関数 > 作成した関数
トリガーを追加
トリガー設定
- APIタイプ：REST API
- セキュリティ：開く
- バイナリメディアタイプ：multipart/formd-data

API Gatewayを作成

API Gateway > API > APIを作成
REST APIを選択
任意のAPI名を入力
APIを作成
メソッドを作成
- メソッドタイプ：POST
- Lmabda関数：作成したLambda関数
メソッドを作成

コードを編集

Lambda > 関数 > 作成した関数
コードは以下のものを使用します

import json
import boto3
import base64
import io
from datetime import datetime
import cgi

BUCKET_NAME = 'バケット名'
DIRECTORY = 'uploadedfiles/'

s3 = boto3.resource('s3')

def lambda_handler(event, context):
    bucket = s3.Bucket(BUCKET_NAME)

    body = base64.b64decode(event['body-json'])
    fp = io.BytesIO(body)

    environ = {'REQUEST_METHOD': 'POST'}
    headers = {
        'content-type': event['params']['header']['content-type'],
        'content-length': str(len(body))
    }

    fs = cgi.FieldStorage(fp=fp, environ=environ, headers=headers)
    for f in fs.list:
        print("filename=" + f.filename)
        bucket.put_object(Body=f.value, Key=DIRECTORY + f.filename)

    client = boto3.client('bedrock-agent')

    response = client.start_ingestion_job(
        dataSourceId='データソースID',
        knowledgeBaseId='ナレッジベースID'
    )

    return {
        "Access-Control-Allow-Headers": "Content-Type",
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "POST",
        "Content-Type": "multipart/form-data",
        'statusCode': 200,
        'body': json.dumps('アップロード完了')
    }

バケット名とナレッジベース

コードにS3のバケット名とナレッジベースのデータソースID, ナレッジベースIDを入力します

APIの設定

あとがき

CORSのエラーに手を焼きました。

参考サイト