サンプルコードで理解する Agents for Amazon Bedrock の Return of Control #Python

はじめに

Agents for Amazon Bedrock の Return of Control は 2024/4/23 のアップデートで新たに利用できるようになった機能です。

Return of Control は AWS Lambda 関数の使用をスキップし、エージェントを呼び出すアプリケーションに制御を返すことができる機能です。これにより Lambda 関数を使用しなくとも外部との連携ができるエージェントを作成できます。

非同期に並列処理を行いたい場合や Lambda 関数の 15 分制限を回避したい場合、プライベートネットワーク上のリソースに対してアクションを起こしたい場合などが主なユースケースのようです。

と書くと何となくわかった気になってしまいすが、冒頭のブログ等を見てもイマイチ使い方が理解できなかったので試してみました。

やりたいこと

東京の現在の天気を答えてくれる Agent を作成する
Lambda 関数を使用せずに Retrun of Control を使用し、クライアントアプリケーション側で API を叩く
結果を Agent に戻して LLM から最終的な応答を得る

天気情報の API は Open-Meteo が提供する Weather Forecast API を使用させていただきます。

当該 API は非営利目的であれば無料で使用できますが、API コール数などの利用規約が定められています。使用する場合は必ずご確認ください。

https://open-meteo.com/en/terms/

詳細は割愛しますが、東京の現在の天気情報を取得するには以下のような URL にリクエストを行います。

https://api.open-meteo.com/v1/forecast?latitude=35.6895&longitude=139.6917&current=weather_code&timezone=Asia%2FTokyo&forecast_days=1

動作確認用 Agent の作成

Agent を新規作成します。名前も適当に付けます。

モデルを選択して、エージェント向けの指示を入力します。今回は Claude 3 Sonnet を選択し、以下のような指示を設定しました。

あなたは東京の現在の天気情報を取得するAgentです。
あなたの目的はユーザーに東京の現在の天気情報を提供することです。 

タスク: 
・信頼できる天気情報APIから関連するデータを取得する 
・取得したデータから時刻およびWeather Codeを抽出し、ユーザーフレンドリーな形式で天気情報を提示する 
・東京以外の天気は取得できないため、ユーザーに聞かれても回答しないでください 

ユーザーへの回答例: 
東京の {time} 時点の天気は {Weather} です。

一旦エージェントの設定を保存した後、アクショングループを追加します。

アクショングループ名 getTokyoLiveWeather を入力します
ここでは Action group type は Define with function details を使用します
Action group invocation で Return control を選択します

Action group function 1 の名前と説明を入力します。ここでは以下のように設定し、アクショングループの作成を完了します。

Name: getTokyoLiveWeather
説明: Get the current weather condition in Tokyo.

テスト画面で「東京の現在の天気を教えて」と入力し、以下のような画面が返ってきます。クライアントアプリケーション側に制御が返ってきた状態 (Return Control)です。

冒頭の API で取得できたレスポンスをそのまま張り付けて送信します。

API のレスポンスには天気情報の更新時刻と Weather Code が含まれているため、LLM が内容を解釈した上で、回答を行ってくれました。

コンソールで動作確認できたら、エージェントのバージョンとエイリアスを新規発行しておきます。

サンプルコード

コンソール上で動作確認した内容を boto3 から実行してみます。コード的にとりあえず動いたレベルなのはご了承ください。

agent.py

import urllib3
from uuid import uuid4
import boto3

http = urllib3.PoolManager()

def fetch_weather_data(http):
    # Open-Metro の API で東京の天気を取得するための URL
    url = 'https://api.open-meteo.com/v1/forecast?latitude=35.6895&longitude=139.6917&current=weather_code&timezone=Asia%2FTokyo&forecast_days=1'
    try:
        response = http.request('GET', url)
        if response.status == 200:
            return response.data.decode('utf-8')
        else:
            print(f"HTTP request failed with status code: {response.status}")
            return None
    except urllib3.exceptions.HTTPError as e:
        print(f"HTTP request failed: {e}")
        return None
    
# 応答チャンクの処理を関数化
def process_chunk_event(event):
    if 'chunk' in event:
        data = event['chunk']['bytes'].decode("utf-8")
        print(data)

def main():
    client = boto3.client("bedrock-agent-runtime")
    agent_id = "YOURAGNTID" # エージェント I Dを設定
    agent_alias_id = "AGENTALIAS" # エージェントエイリアス ID を設定
    session_id = str(uuid4()) # セッション ID を生成
    
    initial_response = client.invoke_agent(
        agentId=agent_id,
        agentAliasId=agent_alias_id,
        sessionId=session_id,
        inputText="東京の現在の天気を教えて",
    )
    
    initial_event_stream = initial_response['completion']
    for initial_event in initial_event_stream:
        
        if 'returnControl' in initial_event:
            # Return Control から返されたアクショングループの情報を取得
            # 呼び出し ID は Agent に結果を戻す際に必要
            invocation_id = initial_event["returnControl"]["invocationId"] 
            action_group = initial_event["returnControl"]["invocationInputs"][0]["functionInvocationInput"]["actionGroup"]
            function = initial_event["returnControl"]["invocationInputs"][0]["functionInvocationInput"]["function"]
            
            if action_group == "getTokyoLiveWeather":
                # 東京の天気情報を取得する
                weather_data = fetch_weather_data(http)
                
                if weather_data:
                    # 天気情報を Agent に返し、最終的な応答を得る。
                    follow_up_response = client.invoke_agent(
                        agentId=agent_id,
                        agentAliasId=agent_alias_id,
                        sessionId=session_id, # initial_response と同一セッション内での処理
                        sessionState={
                            "invocationId": invocation_id,
                            "returnControlInvocationResults": [{
                                # 関数スキーマと OpenAPI スキーマではレスポンス型式が異なる
                                "functionResult": {
                                    "actionGroup": action_group,
                                    "function": function,
                                    "responseBody": {
                                        "TEXT": {
                                            "body": weather_data #取得した天気の情報
                                        }
                                    }
                                }       
                            }]
                        }
                    )
                    follow_up_event_stream = follow_up_response['completion']
                    for follow_up_event in follow_up_event_stream:
                        process_chunk_event(follow_up_event)
            break
            
        process_chunk_event(initial_event)
 
    
if __name__ == "__main__":
    main()

結果: 東京の現在の天気を教えてください

$ python3 agent.py
東京の 2024年4月30日 17時30分 時点の天気は晴れです

結果: 北海道の現在の天気を教えてください

$ python3 agent.py
すみません、私は東京の天気情報しか提供できません。
北海道の天気情報を取得する機能がないため、ご質問にお答えすることができません。

とりあえず boto3 経由で動作確認ができ、何となく仕組みも理解できた気がします。ただ使い方があっているかどうかが少し不安です。コードももうちょっと他に書きようがありそう。

以上です。
参考になれば幸いです。