Amazon Bedrcok + Amazon KendraのRAGパターンをboto3で自前実装してみる

何の記事

• Amazon Bedrock上のClaude-v2に対してRAGパターンぽくKendaraの検索結果をpromtに混ぜて回答を生成してもらう
• 正攻法はSageMaker Canvas または　LangChainのRetrieverのはず
• 投稿者のキャッチアップが追いついていないで、boto3から叩くことで無理やり実装してみた
• こちらの記事とこちらの記事の続き

環境

Kendra Indexの前提

下記のようなテキストファイルをS3バケットに配置し、KendraのIndexにSyncさせている

office_info.txt

■オフィス間取りについて
一般受付は2Fの東側にあります。
オフィスフロアは19Fから35Fです。
来客向け会議室は36Fの南側にあります。
展望室は37Fです。

■ワークフロー申請方法について
交通費申請はグループ会社共有Webシステムからワークフローの申請をしてください。
在宅勤務申請は自社Webシステムから申請してください。
部内備品の持ち出しはExcelにて申請してください。

IAM ポリシー

下記のポリシーをアッタッチしたIAMのプロファイルで実行

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "bedrock:InvokeModelWithResponseStream",
                "bedrock:InvokeModel"
            ],
            "Resource": [
                "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-v2",
                "arn:aws:bedrock:us-east-1::foundation-model/stability.stable-diffusion-xl-v0"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "kendra:Query"
            ],
            "Resource": [
                "arn:aws:kendra:ap-northeast-1::index/${kendraIndxeId}""
            ]
        }
    ]
}

コード

invoke_rag.py

import boto3
import json
import base64
import sys
from datetime import datetime

prompt_data = sys.argv[1] if len(sys.argv) > 0 else sample_prompt

session = boto3.Session(profile_name='bedrock', region_name="us-east-1")
bedrock = session.client(service_name='bedrock-runtime')

prompt = sys.argv[1] if len(sys.argv) > 0 else 'hello!!'

def search_kendra(prompt):
  session_kendra = boto3.Session(profile_name='bedrock', region_name="ap-northeast-1")
  kendra = session_kendra.client(service_name='kendra')
  response = kendra.query(IndexId="this is your index id", QueryText=prompt,AttributeFilter={
            "EqualsTo": {
                "Key": "_language_code",
                "Value": {"StringValue": "ja"}, # 日本語を設定
            },
        })
  excerpt = ''
  for item in response['ResultItems']:
      excerpt = item['DocumentExcerpt']['Text']
  return excerpt

def chat_response(prompt):
  body = json.dumps({
      'prompt': '\n\nHuman:{0}\n\nAssistant:'.format(prompt),
      'max_tokens_to_sample': 500,
      'temperature': 0.1,
      'top_p': 0.9
  })
                    
  modelId = 'anthropic.claude-v2'
  accept = 'application/json'
  contentType = 'application/json'

  response = bedrock.invoke_model(body=body, modelId=modelId, accept=accept, contentType=contentType)
  
  response_body = json.loads(response.get('body').read())
  return response_body.get('completion') 

if __name__ == "__main__":
  func_list = []
  func_list.append({"function": print, "discription": "それ以外。通常の対話"})

  func_discription = "\n".join([ "{0}. {1}".format(i, dict["discription"]) for i, dict in enumerate(func_list)])
  func_list = [x["function"] for x in func_list]

  prompt_base = """
  次の問いかけに対して、ユーザ意図を解析して、以下の３パターンのいずれかを選択して、レスポンスを返してください
  {0}
  レスポンスは必ず次のJson文字列して、返却してください
  """.format(func_discription)
  prompt_json_format = """
  {
      "response_type":  解析したパターンの番号をここに表示 ,
      "response_text": " Calude2からの回答をここに表示 "
  }
  """
  resp_kendra = search_kendra(prompt)
    
  prompt_string = prompt_base + prompt_json_format + "問いかけ：「{0}」".format(prompt)
  if resp_kendra:
    prompt_string = prompt_string + """
    レスポンス文章に改行を含まないでください。
    なお、社内ドキュメントの検索結果は下記です。
    通常の対話の場合は下記から必要な方法を集約して回答を生成してください
    「{0}」""".format(resp_kendra)

  resp_json = chat_response(prompt_string)
  try:
    resp_dict = json.loads(resp_json)
  except :
    print(chat_response(prompt))

  func_list[resp_dict["response_type"]](resp_dict["response_text"])

Claude-v2とRAGパターンの実行結果比較

下記のようになったが、これだけでも質問によっては回答精度が低いものもあった。

質問１: 一般受付は何階ですか

こちらは意図に沿った回答が生成された。

Claude-v2のみ

root@e61f141a4140:~# python invoke_claude.py '一般受付は何階ですか'
ご質問の内容から、一般的な問い合わせと判断しました。ご要望の階数については、お店の詳細をご確認いただくか、直接お店にお問い合わせください。

Kendraの検索結果を追加

root@e61f141a4140:~# python invoke_rag.py '一般受付は何階ですか'
一般受付は2Fの東側にあります。

質問2: 交通費申請はどのようにすればいいですか

こちらは意図に沿った回答ではあったが、ドキュメントの記載にない手順をClaude-v2が追加していた

Claude-v2のみ

root@e61f141a4140:~# python invoke_claude.py '交通費申請はどのようにすればいいですか'
交通費の申請についてですが、通常は出張報告書に領収書を添付して上長に提出することになります。報告書の書式や申請手順の詳細については、所属部署の上長や担当者に確認することをおすすめします。ご不明な点があれば遠慮なく聞いてください。

Kendraの検索結果を追加

root@e61f141a4140:~# python invoke_rag.py '一般受付は何階ですか'
交通費の申請はグループ会社共有Webシステムからワークフローの申請を行う必要があります。具体的な手順は以下の通りです。
1. グループ会社共有Webシステムにログインします。
2. メニューから「ワークフロー」を選択します。 
3. 「新規申請」をクリックして、申請種別を「交通費申請」と選択します。
4. 必要な情報を入力して申請を行います。日付、金額、目的、領収書画像等を入力します。
5. 申請内容を確認して「申請」ボタンをクリックして完了です。
申請後は承認フローに沿って上長の承認を得る必要があります。不明な点があればご質問ください。"

所感

• おそらくだが、回答精度の向上にはpromptエンジニアリングの改善 + Kendraに読み込ませるドキュメント記載方法の改善に継続的かつ相当な工数投入が必要
• 複数のドキュメント候補が見つかった場合や、新旧のドキュメントがあった場合にどちらを正にするかは人間が困るのと同様にシステムでも困ると思う
• 正直、例外系の実装が結構大変だと思う
• モデルに余計な情報を足すなと指示を出さないといけないと思う
• 場合によっては、不正確な情報に正しい用語を混ぜてくる分だけ「嘘をつかれた」場合に人間側が判断しずらくなるかもしれない
• やはりLangChainなりSageMaker Canvasなりを利用した方が絶対に良い
• LLM周りの基礎的素養が足りなすぎるので、勉強が必要そう