Amazon Bedrock を利用してスマートホームスキルを起動するパーソナルアシスタントAIっぽいコードを書いた

何の記事？

• 最近GAした Amazon Bedrock をpythonのSDK(boto3)から実行してみる
• 前回記事の続き
• この中で、プロンプトエンジニアリングっぽくClaude-v2にユーザの意図を判定してもらい、後続の処理を選択してもうらう
• 後続の処理の中で再度ユーザの意図をCluade-v2に判定してもらい処理のパラメータを生成する
• 後続処理としては下記を行う

1. 単純なチャットの対話
2. 画像生成AIでの画像生成
3. ALexa SmartHomeSkillの内部処理であるLambdaの呼び出し

Code

personal_assistant.py

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

sample_prompt = "A photograph of an dog on the top of a mountain covered in snow."
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 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') 

def image_response(prompt):
  promt_str =  "次の文章から画像生成AIに渡すpromtを英語で生成してください\n「{0}」生成したpromptは[[[ と ]]]　でくくってください".format(prompt)
  prompt_data = chat_response(promt_str)
  print("{}というpromtで画像を出力します".format(prompt_data))
  body = json.dumps({
    "text_prompts": [
      { 
        "text": prompt_data 
      }
    ],
    "cfg_scale":10,
    "seed":20,
    "steps":50
  })
  modelId = "stability.stable-diffusion-xl-v0" 
  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())
  print(response_body['result'])

  image_base64 = response_body.get("artifacts")[0].get("base64")

  image_binary = base64.b64decode(image_base64)

  file_name = "image{0}.png".format(datetime.now().strftime("%Y%m%d%H%M%S"))
  print("saving file: {0}".format(file_name))
  with open(file_name, "wb") as image_file:
      image_file.write(image_binary)
  print("image is saved.")

def invoke_lambda(prompot):
  prompt_base  =  """次の問いかけに対して、ユーザの意図を解析して、対応するdevice_idとstatusをレスポンスとして返してください
  device_id は次のリストのいずれかです [ desklight, kitchin, livingroom ]
  status は on または off です
  レスポンスは必ず次のJson文字列して、返却してください"""
  prompt_json_format = """
  {
      "device_id": "device_id" ,
      "status": " on or off "
  }
  """
  lambda_payload = chat_response(prompt_base + prompt_json_format + "問いかけ：「{0}」".format(prompt))
  #print(prompt_data)

  session_lambda = boto3.Session(profile_name='bedrock', region_name="us-west-2")
  lambda_client = session_lambda.client(service_name="lambda")
  response = lambda_client.invoke(
    FunctionName='MyhomeSkill',
    InvocationType='Event',
    Payload=lambda_payload, # 本当はSmartHomeSkillのRequest/Contextは非常に階層が深いが今回はLambda側でバラメータをバイパスしている
    Qualifier='1',
  )



if __name__ == "__main__":
  func_list = [
                 {"function": invoke_lambda, "discription": "デバイスの操作の依頼。例：リビングの電気をつけて、など"},
                 {"function": image_response, "discription": "画像の生成の依頼。例：〜というイメージの画像を生成して、〜をしている画像をください、など"}
              ]
# ↑に処理定義を足す + 関数を足すことで機能拡張可能
  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からの回答をここに表示 "
  }
  """
  prompt_string = prompt_base + prompt_json_format + "問いかけ：「{0}」".format(prompt)
  # print(prompt_string)
  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"])

処理の内容

main

func_list = [
                 {"function": invoke_lambda, "discription": "デバイスの操作の依頼。例：リビングの電気をつけて、など"},
                 {"function": image_response, "discription": "画像の生成の依頼。例：〜というイメージの画像を生成して、〜をしている画像をください、など"}
              ]
func_list.append({"function": print, "discription": "それ以外。通常の対話"})

• 処理定義のlistに記載された内容を後続の処理で呼び出す。
• この定義を最初のClaude-v2に渡すPromptに使う
• Claude-v2に渡すpromtは下記のイメージになる

root@e61f141a4140:~# python personal_assistant.py 'デスクライトのスイッチつけて'

  次の問いかけに対して、ユーザ意図を解析して、以下の３パターンのいずれかを選択して、レスポンスを返してください
  0. デバイスの操作の依頼。例：リビングの電気をつけて、など
1. 画像の生成の依頼。例：〜というイメージの画像を生成して、〜をしている画像をください、など
2. それ以外。通常の対話
  レスポンスは必ず次のJson文字列して、返却してください
  
  {
      "response_type":  解析したパターンの番号をここに表示 ,
      "response_text": " Calude2からの回答をここに表示 "
  }
  問いかけ：「デスクライトのスイッチつけて」

• Claude-v2からの回答は下記のJson形式になる

{'response_type': 0, 'response_text': '「デスクライトのスイッチつけて」という依頼は、デバイスの操作を求めるものと解析しました。デバイス操作の依頼と判断したため、response_typeを0に設定しました。'}

• Claude-v2からの回答をJsonパースして後続の処理を呼びだす

• 後続処理では、Claude-v2にさらにユーザの意図を解析してもらい処理に必要なパラメータを生成してもらう

一応、生成AIから任意のAWSリソースを実行する下地ができた
このあとは、モデルのカスタマイズとRAGパターンの実装にチャレンジする