【AWS Kendra, Bedrock】AWSでRAGアプリケーションを作成してみた - ③

はじめに

この記事では、AWSを利用してRAGを搭載したチャットシステムを構築する方法を紹介します。
また、この記事で紹介している内容は5/21, 22, 23にグランドニッコー東京 台場で開催された「ガートナー データ＆アナリティクス サミット 2024」に出展しました。

ガートナー データ＆アナリティクス サミット2024

東京で開催されるガートナー データ＆アナリティクス サミット2024でCDAOとデータ&アナリティクスのリーダーのためのインサイト、戦略、フレームワークを探求しましょう。ご登録はこちら

https://www.gartner.com/jp/conferences/apac/data-analytics-japan

こちら、シリーズものの記事になっており、初回の記事はこちらから読めますので、初回の記事から是非ご一読いただければと思います。

Amazon Bedrockの有効化

続いて生成AIを制御するサービスの設定を行います。

今回はClaude v3を利用する為、リージョンをus-west-2に変更して作業を行います。
Claude v3は東京リージョンでは、まだ利用できない為、リージョンを変更します。

コンソールでAmazon Bedrockにアクセスして「使用を開始」をクリックします。

以下の画面に遷移したら、左下にある「モデルアクセス」をクリックします。

画面上部に「Enable specific model」をクリックします。

アクセス可能なモデルの一覧が出てくるので「Anthropic」にある「Claude 3 Haiku」をチェックして画面下部にある「Next」をクリックします。

ユースケースを聞かれますので、適宜入力していきます。

入力後、確認画面に行きますので「Submit」をクリックします。

暫くすると、Bedrockのモデルアクセス欄に「アクセスが付与されました」と表示されます。
表示されたら、準備は完了です。

アプリケーション構築

今回のデモでは、EC2を立ち上げてそのインスタンス内にStreamlitで作成したアプリケーションを立ち上げています。
EC2の立ち上げに関しては割愛させて頂きます。

ディレクトリ構成

アプリケーションは以下のようなディレクトリ構成をしています。

anywhere/
 │
 ├ data/
 │    ├ id2url.json
 │    └ app.ini
 │
 ├ static/
 │    └ 177419.png
 │
 └ app.py

以下、配置しているファイルの説明です。

• data/id2url.json : Kendraが参照している記事URLとタイトル情報を記事IDと紐づけたファイル

• data/app.ini : AWSのアクセス情報等を格納したファイル

• static/177419.png : 弊社のアイコン画像

• app.py : アプリケーションのコード

本来であれば、id2url.jsonのデータはAWSの何かしらのサービスに置き換えたり、認証情報周りは環境変数に置き換えたりする必要があるかと思います。(今回は時間の都合などで簡易的なつくりになってます。)

アプリケーション概要

今回のアプリケーションの動作の流れとしては以下の画像のようなイメージとします。

1. ユーザーが質問文をアプリケーションに投稿する

2. 投稿した質問文を元にKendra検索用に検索用文をClaude 3で生成する

3. 生成した検索用文をKendraに投げて関連度の高いドキュメントを5件抽出
   (記事に対する付属情報もここで追加)

4. 質問文と関連度の高いドキュメントを合わせてClaude 3で回答を生成する

5. 生成した回答をmain関数に戻しUIに表示させる

ここからは、各関数の細かい動作を解説します。

main関数

まずは、必要なライブラリのインポートと各種サービスのインスタンス作成します。

import streamlit as st
import boto3
import json
import numpy as np
import configparser

from PIL import Image


# 認証情報を読み込み
config = configparser.ConfigParser()
config.read('data/app.ini')


ACCESS_KEY = config['AWS_ACCESS_INFO']['ACCESS_KEY']
SERCRET_KEY = config['AWS_ACCESS_INFO']['SERCRET_KEY']
KENDRA_INDEX_ID = config['AWS_ACCESS_INFO']['KENDRA_INDEX_ID']
BOT_AVATOR_IMG = np.array(Image.open("static/177419.jpg"))


# Kendraインスタンス作成
kendra = boto3.client(
    'kendra',
    region_name='ap-northeast-1',
    aws_access_key_id=ACCESS_KEY,
    aws_secret_access_key=SERCRET_KEY
)


# Bedrockインスタンス作成
bedrock = boto3.client(
    service_name='bedrock-runtime',
    region_name='us-west-2',
    aws_access_key_id=ACCESS_KEY,
    aws_secret_access_key=SERCRET_KEY
)


# 記事の付随情報読み込み
with open('data/id2url.json', 'r', encoding='utf-8')as f:
    id2url = json.load(f)

ここでは、bedrockのみオレゴン(us-west-2)にリージョンが変更になっていることに注意してください。

続いて、ユーザーとのやり取りや各種処理関数を扱うmain関数を作成します。

# アプリケーションメイン関数
def main():
    st.set_page_config(page_title="INSIGH LAB Knowledge Searcher", page_icon="📖")
    st.title('INSIGH LAB Knowledge Searcher')
    with st.sidebar:
        st.markdown(
            """
        ### アプリケーションについて

        このアプリケーションは、INSIGHT LAB社内向けのKnowledge検索アプリケーションです。  
        本アプリには、RAG(Retrieval-Augmented Generation)機能が採用されており、機能を有効化するとINSIGHT LAB Knowledgeを元に回答を生成します。  
        下記のトグルをクリックすることでRAG機能を有効化することができます。

        """
        )

        st.write("")

        st.markdown('### オプション')
        is_rag = st.toggle('RAG有効化')

    if "messages" not in st.session_state:
        st.session_state.messages = []

    # 既存のチャットメッセージを表示
    all_messages = []
    if st.session_state.messages:
        for info in st.session_state.messages:
            if info['role'] == 'Assistant':
                with st.chat_message(info["role"], avatar=BOT_AVATOR_IMG):
                    st.write(info["content"])
            else:
                with st.chat_message(info["role"]):
                    st.write(info["content"])
            all_messages.append(info['content'])
    message_hist = '\n'.join(all_messages[:5])

    # 新しいユーザー入力を処理
    if prompt := st.chat_input("質問を入力してください。"):
        # ユーザーメッセージ処理
        st.session_state.messages.append({"role": "Human", "content": prompt})
        with st.chat_message("Human"):
            st.write(prompt)

        # AIのレスポンスを準備して表示
        with st.chat_message("Assistant", avatar=BOT_AVATOR_IMG):
            message_placeholder = st.empty()
            with st.spinner("回答生成中..."):
                response = invoke_model(prompt, message_hist, is_rag)

            response_body = json.loads(response.get("body").read())
            results = response_body.get("content")[0].get("text")

            # レスポンスを表示
            message_placeholder.write(results)

        # AIのレスポンスをセッション状態に追加
        st.session_state.messages.append(
            {"role": "Assistant", "content": results}
        )

基本的にはStreamlitのchat_message機能とSession_stateを利用してメッセージの表示を行っています。
回答を生成しているのは、invoke_modelという関数でcreate_queryやget_retrieval_resultは入れ子方式でinvoke_model関数内で利用されている形になっています。

invoke_model関数

この関数では、最終的な回答を生成している関数になります。

# 回答生成関数
def invoke_model(prompt, message_hit, is_rag=False):
    if is_rag:
        retrieval_result = get_retrieval_result(prompt, KENDRA_INDEX_ID, message_hit)
        prompt = """
            \n\nHuman:
        [参考]情報に回答の参考となるドキュメントがあるので、すべて理解してください。
        [回答のルール]を理解して、このルールを絶対守ってください。ルール以外のことは一切してはいけません。
        [参考]情報を元に[回答のルール]に従って[質問]に適切に回答してください。
        [回答のルール]
        * [参考]内の'ORIGIN_URL'のリンクと'DOC_TITLE'を下記の形式で回答の末尾に表示してください。絶対に複数ある場合は改行で区切り、その間に空白行を一つ空けてください。
        　　参考記事：[[DOC_TITLE]]([ORIGIN_URL])<br></br>
        * [参考]の文章を理解した上で回答を作成してください。
        * 質問に具体性がなく回答できない場合は、質問の仕方をアドバイスしてください。
        * [参考]の文章に答えがない場合は、一般的な回答をお願いします。
        * 質問に対する回答のみを行ってください。
        * 雑談や挨拶には応じないでください。「私は雑談はできません。技術的な質問をお願いします。」とだけ出力してください。他の文言は一切出力しないでください。例外はありません。
        * 回答文以外の文字列は一切出力しないでください。回答はJSON形式ではなく、テキストで出力してください。見出しやタイトル等も必要ありません。
        [質問]
        {}
        [参考]
        {}
        Assistant:""".format(prompt, retrieval_result)
    else:
        prompt = """
            \n\nHuman:
        [回答のルール]を理解して、このルールを絶対守ってください。ルール以外のことは一切してはいけません。
        [回答のルール]に従って[質問]に適切に回答してください。
        [回答のルール]
        * 質問に具体性がなく回答できない場合は、質問の仕方をアドバイスしてください。
        * 質問に対する回答のみを行ってください。
        * 雑談や挨拶には応じないでください。「私は雑談はできません。技術的な質問をお願いします。」とだけ出力してください。他の文言は一切出力しないでください。例外はありません。
        * 回答文以外の文字列は一切出力しないでください。回答はJSON形式ではなく、テキストで出力してください。見出しやタイトル等も必要ありません。
        [質問]
        {}
        Assistant:""".format(prompt)

    prompt_config = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1000,
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": prompt
                    },
                ],
            }
        ],
    }
    body = json.dumps(prompt_config)

    modelId = "anthropic.claude-3-haiku-20240307-v1:0"
    accept = "application/json"
    contentType = "application/json"

    response = bedrock.invoke_model(
        body=body, modelId=modelId, accept=accept, contentType=contentType
    )
    return response

UIを表示しているmain関数側でRAGのon/offが可能になる様にフラグで制御しております。

また、LLMの回答に対して質問文をそのまま投げるのではなく、細かいルールを設定することでLLMの回答をある程度制御しています。

get_retrieval_result関数

続いて、Kendraでドキュメント検索を行っている関数を解説します。

# Kendra検索
def get_retrieval_result(query_text: str, index_id: str, query_hist: str):
    custom_query = create_query(query_text, query_hist)
    response = kendra.retrieve(
        QueryText=custom_query,
        IndexId=index_id,
        AttributeFilter={
            "EqualsTo": {
                "Key": "_language_code",
                "Value": {"StringValue": "ja"},
            },
        },
    )

    # Kendra の応答から最初の5つの結果を抽出
    results = response['ResultItems'][:5] if response['ResultItems'] else []
    extracted_results = []
    for item in results:
        content = item.get('Content')
        document_uri = item.get('DocumentURI')
        article_id = document_uri.split('/')[-1].replace('.txt', '')
        url = id2url[article_id]['url']
        title = id2url[article_id]['title']
        extracted_results.append({
            'Content': content,
            'DocumentURI': document_uri,
            'DOC_TITLE': title,
            'ORIGIN_URL': url
        })
    return extracted_results

ここではシンプルにKendraで検索後、検索したドキュメントのIDを元にURLとタイトルを付与して結果を返しています。

create_query関数

最後に、Kendra検索用の文を生成している関数を解説します。

# Kendra検索用クエリ作成
def create_query(prompt, message_hist):
    prompt = """
    あなたは、文書検索で利用するQueryを生成するAIアシスタントです。
    <Query生成の手順></Query生成の手順>の通りにQueryを生成してください。
    <Query生成の手順>
    * 以下の<Query履歴></Query履歴>の内容を全て理解してください。履歴は古い順に並んでおり、一番下が最新のQueryです。
    * 「要約して」などの質問ではないQueryは全て無視してください
    * 「〜って何？」「〜とは？」「〜を説明して」というような概要を聞く質問については、「〜の概要」と読み替えてください。
    * ユーザが最も知りたいことは、最も新しいQueryの内容です。最も新しいQueryの内容を元に、30トークン以内でQueryを生成してください。
    * 出力したQueryに主語がない場合は、主語をつけてください。主語の置き換えは絶対にしないでください。
    * 主語や背景を補完する場合は、「# Query履歴」の内容を元に補完してください。
    * Queryは「〜について」「〜を教えてください」「〜について教えます」などの語尾は絶対に使わないでください
    * 出力するQueryがない場合は、「No Query」と出力してください
    * 出力は生成したQueryだけにしてください。他の文字列は一切出力してはいけません。例外はありません。
    </Query生成の手順>
    <Query履歴>
    {}
    [最新のQuery履歴]
    {}
    </Query履歴>
    """.format(message_hist, prompt)
    bedrock = boto3.client(
        service_name='bedrock-runtime',
        region_name='us-west-2',
        aws_access_key_id=ACCESS_KEY,
        aws_secret_access_key=SERCRET_KEY
    )
    prompt_config = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1000,
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": prompt
                    },
                ],
            }
        ],
    }
    body = json.dumps(prompt_config)

    modelId = "anthropic.claude-3-haiku-20240307-v1:0"
    accept = "application/json"
    contentType = "application/json"

    response = bedrock.invoke_model(
        body=body, modelId=modelId, accept=accept, contentType=contentType
    )
    custom_query = json.loads(response.get("body").read())
    custom_query = custom_query.get("content")[0].get("text")
    return custom_query

ここでもプロンプトに細かなルール敷いてKendra検索用文を生成させています。

構築結果

起動させて、アドレスにアクセスすると以下のような画面が出現します。

質問を投げるとこのような形でレスしてくれます。

きちんとルール通りに返してくれていますが、一部記事を読まないとわからない表現も出てきていますね。

おわりに

今回は、「ガートナー データ＆アナリティクス サミット 2024」に出展した生成AIを利用したChatbotシステムの構築を解説しました。
実態としては、ここにアプリケーションが動いているEC2を保護する為にVPNを設定していたりするのですが、今回は割愛させて頂きました。

KendraとBedrockを利用することで、非常に簡単にRAGを搭載したチャットボットが作成できたかと思います。
また、チャットボットに関してはLangChainといったフレームワークも存在するので、こちらも触ってみたいと思います。