Agents for Amazon Bedrock から複数のLambda関数を呼び分ける

Agents for Amazon BedrockからLambda関数を呼び出します。
自然言語の問い合わせを元に、複数のLambda関数の中から適したものを選択し引数を考えて実行、実行結果をLLMで整形して応答します。

そもそもAgents for Amazon Bedrockとは何なのか

• 自然言語で呼び出せるAPI（Lambda）のオーケストレーターです
• API（Lambda）の呼び分けルール、順序性や関連性も自然言語（とOpenAPIスキーマ）で定義できます
• ベクトルデータストアとしてKnowledge baseもオーケストレーションできます
• オーケストレーターなので、Agentだけで何か処理が出来るわけではありません（これ重要）

今回作ったもの

今回作ったものは、以下の3サイトに対応したLambda関数を作成しておき、それをAgentから呼び出しています。

• Yahoo天気（手抜きで東京限定）
• Yahoo乗換案内（手抜きで今すぐ出発）
• Wikipedia

今回はWeb検索している例のみですが、Lambda関数の呼出しが出来るので、つまりその気になれば更新処理だろうが何だろうが何でも出来ます。

実行イメージ

Lambda関数の作成

Lambda関数その1 Yahoo天気から天気予報を取得

引数無しで実行する例としてYahoo天気から今日明日の天気予報を取得します。手抜きで東京限定です。
requestsとBeautifulSoupを使ってスクレイピングするので、Lambdaレイヤーで追加しておきます。
このLambdaはYahoo天気の東京のページから「今日明日の天気予報」のあたりを取得します。Agentの最後にLLMの整形処理が入っているので、それを頼って引っこ抜いたHTMLをそのままリターンしています。

tokyoWeather

import requests
from bs4 import BeautifulSoup
import json

def lambda_handler(event, context):

	# 東京のYahoo天気
	url = 'https://weather.yahoo.co.jp/weather/jp/13/4410.html'

	# 今日明日の天気のみ取得
	page = requests.get(url).text
	contents = str(BeautifulSoup(page, 'html.parser').select('#main > div.forecastCity'))

	# agentの形式でreturn
	response_body = {"application/json": {"body": contents}}
	action_response = {
		"actionGroup": event["actionGroup"],
		"apiPath": event["apiPath"],
		"httpMethod": event["httpMethod"],
		"httpStatusCode": 200,
		"responseBody": response_body,
	}
	api_response = {"messageVersion": "1.0", "response": action_response}
	return api_response

後はアクセス権限の設定から以下の感じのリソースベースのポリシーを追加しておきます。

{
  "Version": "2012-10-17",
  "Id": "default",
  "Statement": [
    {
      "Sid": "tokyoWeather",
      "Effect": "Allow",
      "Principal": {
        "Service": "bedrock.amazonaws.com"
      },
      "Action": "lambda:InvokeFunction",
      "Resource": "arn:aws:lambda:us-east-1:999999999999:function:tokyoWeather"
    }
  ]
}

Lambda関数その2 Yahoo乗換案内から乗換情報を取得

この関数は引数を2つ取ります（dep_stationとdest_station。後で使います）。手抜きで今すぐ出発のみです。
こちらもrequestsとBeautifulSoupを使ってスクレイピングするので、Lambdaレイヤーで追加しておきます。
天気と同様にHTMLのままアバウトに引っこ抜いてagent最後のLLMにどうにかしてもらいます。リソースベースのポリシーは同様に設定しておきます。

import requests
from bs4 import BeautifulSoup
import json

def lambda_handler(event, context):

	print(event) # パラメータの出力（確認用）
	
	# 出発駅と到着駅の取得
	parameters = event.get('parameters')
	dep_station = next(p for p in parameters if p['name'] == 'dep_station')['value']
	dest_station = next(p for p in parameters if p['name'] == 'dest_station')['value']
	url = "https://transit.yahoo.co.jp/search/print?from="+dep_station+"&flatlon=&to="+ dest_station
	
    # 乗換案内のページからメイン部分のみ取得
	page = requests.get(url).text
	contents = str(BeautifulSoup(page, 'html.parser').select('#srline'))

	# agentの形式でreturn
	response_body = {"application/json": {"body": contents}}
	action_response = {
		"actionGroup": event["actionGroup"],
		"apiPath": event["apiPath"],
		"httpMethod": event["httpMethod"],
		"httpStatusCode": 200,
		"responseBody": response_body,
	}
	api_response = {"messageVersion": "1.0", "response": action_response}
	return api_response

Lambda関数その3 日本語Wikipediaから情報を取得

この関数は引数を1つ取ります（search_text。後で使います）。
Lambdaレイヤーとしてwikipediaを追加しておきます。リソースベースのポリシーは同様。

import wikipedia
import json

def lambda_handler(event, context):

	print(event) # パラメータの出力（確認用）
	
	# 検索キーワード（search_textのvalue）を取得
	parameters = event.get('parameters')
	search_text = next(p for p in parameters if p['name'] == 'search_text')['value']
	print("search_text: " + search_text) # 検索キーワードの出力（確認用）

	# 日本語Wikipediaを検索
	wikipedia.set_lang("ja")
	search_response = wikipedia.search(search_text)

	if search_response:
		# 候補の1つ目のページを取得
		wiki_page = wikipedia.page(search_response[0])
		
		# wikipediaに記載されている文字列を4000文字まで取得（ここを制限しないと何かエラーが起きる時がある）
		wiki_content = wiki_page.content[:4000]
		response_string = wiki_content
	else:
		response_string = "その単語は登録されていません。"

	# agentの形式でreturn
	response_body = {"application/json": {"body": response_string}}
	action_response = {
		"actionGroup": event["actionGroup"],
		"apiPath": event["apiPath"],
		"httpMethod": event["httpMethod"],
		"httpStatusCode": 200,
		"responseBody": response_body,
	}
	api_response = {"messageVersion": "1.0", "response": action_response}
	return api_response

Agentの作成

AgentのInstructionは以下のように、敢えていい加減に、Functionの細かい情報を記載しないようにしています（このInstructionの影響を確認したかったので）。
後は名前を付けたぐらいで作成を完了します。関数の実行結果に対して何か（整形の仕方の指定や、複数の関数の組み合わせなど）したい場合はこのInstructionに色々書けば良いような気がします。

この段階ではAction groupsもKnowledge baseも何も指定せずに作成を完了します。
（APIスキーマは後からならコンソールから編集できるのですが、この段階ではS3に格納しないといけない…？）

Action groupの追加

Action groupその1 天気予報を追加

名前を適当につけて、ここのdescriptionもどうやら使われなさそうなので適当にいきます（省略しても良いと思います）。

先ほど作成したLambda関数を選択し、OpenAPIスキーマを手打ちします。

OpenAPIスキーマは以下にしました。このdescriptionがどうやら重要で、プロンプト内で関数選択の条件として使用されています。
これでAddします。

openapi: 3.0.0
info:
  title: Lambda
  version: 1.0.0
paths:
  /search:
    get:
      summary: Weather Search
      description: 今日明日の天気を検索します
      operationId: search
      responses:
        "200":
          description: 検索成功
          content:
            application/json:
              schema:
                type: object
                properties:
                  body:
                    type: string

Action groupその2 乗換案内を追加

同様に作成したLambda関数を選択し、OpenAPIスキーマは以下としました。パラメータとして、出発駅dep_stationと到着駅dest_stationを要求しています（これをLambda関数で取得しています）。

openapi: 3.0.0
info:
  title: Lambda
  version: 1.0.0
paths:
  /search:
    get:
      summary: Transit Search
      description: Yahoo乗換案内を検索します
      operationId: search
      parameters:
        - name: dep_station
          in: path
          description: 出発駅
          required: true
          schema:
            type: string
        - name: dest_station
          in: path
          description: 到着駅
          required: true
          schema:
            type: string
      responses:
        "200":
          description: 検索成功
          content:
            application/json:
              schema:
                type: object
                properties:
                  body:
                    type: string

Action groupその3 Wikipedia検索を追加

同様に作成したLambda関数を選択し、OpenAPIスキーマは以下としました。パラメータとして、検索キーワードsearch_textを要求しています（これをLambda関数で取得しています）。

openapi: 3.0.0
info:
  title: Lambda
  version: 1.0.0
paths:
  /search:
    get:
      summary: Wikipedia Search
      description: Wikipediaを検索します
      operationId: search
      parameters:
        - name: search_text
          in: path
          description: 検索キーワード
          required: true
          schema:
            type: string
      responses:
        "200":
          description: 検索成功
          content:
            application/json:
              schema:
                type: object
                properties:
                  body:
                    type: string

実行結果の確認

Test欄で試してみます。

トレースを見てみると、各Lambda関数を呼び分け、その実行結果をLLMで整形してくれているのが分かります。LLMはHTMLを適当に渡しても解釈して整形してくれるのでスクレイピングに便利ですね。
という事で、（LambdaとOpenAPIスキーマとAgent自体の呼出機能構築に目を瞑れば）ノーコードでAgent機能を構築できる事が分かりました。

PythonからのAgent呼出し

マネコンを使い続けるわけにもいかないと思うので、PythonからAgentを呼び出すガワのサンプルを作っておきます。
作成したAgentのAliasを作成しておいて、以下のように呼び出します。これはStreamlitからの呼び出し例です。

agent.py

import uuid
import boto3

# Agentの定義
agent_id:str = 'XXXXXXXXXX' # ご自身のAgentのID
agent_alias_id:str = 'XXXXXXXXXX' # ご自身のAgentのAlias ID
session_id:str = str(uuid.uuid1()) # 乱数

# Clientの定義
client = boto3.client("bedrock-agent-runtime")

# Streamlitでガワを被せる
import streamlit as st

st.title("Agentから複数のLambda関数の呼び分け")
input_text = st.text_input("このテキストをAgentに送信します")
send_button = st.button("送信")

if send_button:
    result_area = st.empty()
    text = ''

    # Agentの実行
    response = client.invoke_agent(
        inputText=input_text,
        agentId=agent_id,
        agentAliasId=agent_alias_id,
        sessionId=session_id,
        enableTrace=False
    )

    # Agent実行結果の取得
    event_stream = response['completion']
    for event in event_stream:        
        if 'chunk' in event:
            text += event['chunk']['bytes'].decode("utf-8")
            result_area.write(text)

python -m streamlit run agent.py

これで冒頭のように実行できます。

使用感はLangChainのAgentとほぼ同じですが、Agentの処理に使われているプロンプトがAWS謹製でClaude用にチューニングされたものなので、LangChainのプロンプトやパーサーとにらめっこするよりは安心感も安定感もありそうです。その分OpenAPIスキーマやLambdaに渡されているパラメータとにらめっこする時間は必要でしたが、こちらは慣れればどうにでもなりそうです。
後はLangChainを使わなくて済むのも利点のひとつですかね。LangChainは便利なんですが開発が活発すぎたり、プロンプトがOpenAIに最適化されていたり、むしろ使いにくいところがあるので。

1回の問い合わせで複数のLambda関数を実行する

ここまでは1回の問い合わせでLambda関数を1回実行して終了していましたが、それを組み合わせるようなInstructionに変更してみます。

以下を追加し、天気予報と乗換案内を両方実行して結果を纏めてもらえないか見てみます。

今日の予定を聞かれた場合は、今日の天気と、横浜から東京の乗換案内を返します。

実行結果はこちら。

うまくいきました。おそらく前段のLambda関数の結果を後段のLambda関数で使用する事も出来ると思います。

今度は問い合わせ内容に複数のタスクを要求してみます。

こちらもうまくいきました…が、乗換検索する駅名が英語になる（検索が出来ない）場合がありました。InstructionやOpenAPIスキーマで日本語である旨も追加した方が処理が安定しそうです。やはりAgentのプロンプトテンプレートが英語で記述されている影響はありそうです。

自然言語でのLambda関数の呼び分けや、組み合わせた連続実行が出来る事が分かりましたので、色んなパーツを組み合わせた複雑な処理も実行できそうです（どこまでやるかは別として）。
簡単なところでは、LLMを使うとスクレイピングがかなり楽なので、よく巡回するサイト（安直な例だと、ニュース、天気、乗換案内、相場情報など）を纏めて引っ張ってきたり、それをアレクサやSlackやLINEからWebhookで連動させたり？、なんていうライトな使い方も出来そうです。
複雑なところでは、API（Lambda）のオーケストレーターとして順序性などもInstructionに定義して、自然言語の対話を通じて更新系も含めた各種のAPI（Lambda）を実行するような処理でしょうか。

今回試していないこと

• 対話型のインターフェイス（セッションID内で履歴保持してるのかな？）
• Knowledge baseの併用
  まあKnowledge baseは以下を組み合わせるだけかなと思います（多分まとめてFunctionとしてリストされる）。