AWS Lambda Web Adapterを使ってAgents for Amazon Bedrockを高速開発

2024年1月27日 · 約8分

moritalous

Maintainer of this blog

お知らせ

過去にQiitaに投稿した内容のアーカイブです。

Amazon Bedrockでエージェントを構築できるAgents for Amazon Bedrockという機能があります。この機能はLambdaで構築するのですが、構築に際して次のものが必要です。

エージェントが実行するアクションのビジネスロジックを含んだLambda関数
APIの説明、構造、パラメーターを含むOpenAPIスキーマ

また、イベントのJSONが専用フォーマットなので、どういったイベントが来るのか気にしながら開発する必要があります。

AWS謹製のAWS Lambda Web AdapterがAgents for Amazon Bedrockに対応しましたので、使用方法を紹介します。

AWS Lambda Web Adapterとは

AWS Lambda 上で Web アプリケーションを実行するツールです。

AWS Lambda Web Adaptor を使用すると、開発者は使い慣れたフレームワーク (Express.js、Next.js、Flask、SpringBoot、ASP.NET、Laravel など、HTTP 1.1/1.0 を使用するものはすべて) を使用して Web アプリ (http API) を構築し、AWS Lambda上で実行できます。また、同じ Docker イメージを AWS Lambda、Amazon EC2、AWS Fargate、およびローカルコンピューター上で実行できます。

https://github.com/awslabs/aws-lambda-web-adapter

構成のイメージ図です。（公式サイトより引用）

これまでは、API GatewayやALBなどのHTTPイベントのみのサポートでしたが、この度、Non-HTTP Eventがサポートされました🎉🎉🎉

Non-HTTPの場合のイメージ図です。

SNSやSQSなどのHTTPではないイベントも受けられます。イベントの種類は自動で判別されて、HTTPイベントの場合はそのパスに、Non-HTTPイベントの場合は/eventsにリクエストが転送される形となります。（/eventsは変更も可能です）

Agents for Amazon Bedrockの場合は以下JSONが/eventsに渡されます。/eventsに届いたJSONを解析し、本来受け取りたいパスにルーティングすることができれば、良いこととなります。

参考：Lambda input event from Amazon Bedrock

{
    "messageVersion": "1.0",
    "agent": {
        "name": "string",
        "id": "string",
        "alias": "string",
        "version": "string"
    },
    "inputText": "string",
    "sessionId": "string",
    "actionGroup": "string",
    "apiPath": "string",
    "httpMethod": "string",
    "parameters": [
        {
            "name": "string",
            "type": "string",
            "value": "string"
        },
    ...
    ],
    "requestBody": {
        "content": {
            "<content_type>": {
                "properties": [
                   {
                       "name": "string",
                       "type": "string",
                       "value": "string"
                    },
                            ...
                ]
            }
        }
    },
    "sessionAttributes": {
        "string": "string",
    },
    "promptSessionAttributes": {
        "string": "string"
    }
}

ルーティングする処理は共通になるので、FastAPIのMiddlewareという仕組みを利用してライブラリーにして公開しました。

このライブラリーを使えば、/eventsのことを意識せず、FastAPIでのAPI開発のみを行えばOKです。

https://github.com/moritalous/lwa-fastapi-middleware-bedrock-agent

注記

Lambda Web AdapterでのAgents for Amazon Bedrockのサポートを機能リクエストとして送り、実現するに至りました。サンプルの作成や汎用ライブラリー化についても持ちかけていただきました。

I hope support Agents for Amazon Bedrock #317

とても貴重な経験となりました。

サンプルアプリの解説

この構成を使ったサンプルアプリをこちらに用意しています。

https://github.com/moritalous/lwa-fastapi-middleware-bedrock-agent/tree/main/example/bedrock-agent-fastapi

ディレクトリ構成

bedrock-agent-fastapi
├── README.md
├── app
│   ├── Dockerfile
│   ├── __init__.py
│   ├── main.py
│   └── requirements.txt
├── events
│   ├── s3_bucket_count.json
│   ├── s3_object.json
│   └── s3_object_count.json
└── template.yaml

使用するライブラリー
Pythonのアプリケーションとして使用する主なライブラリー
- FastAPI
- lwa-fastapi-middleware-bedrock-agent（FastAPI Middleware for "Agents for Amazon Bedrock"）

Lambda Web Adapterの組み込み

Lambdaのデプロイタイプはコンテナとしています。Dockerfileの2行目でLambda Web Adapterの内容をコピーしています。Lambda Web Adapterを使用するための最低限の指定はこの1行のみです。

Dockerfile
FROM public.ecr.aws/docker/library/python:3.12.0-slim
COPY --from=public.ecr.aws/awsguru/aws-lambda-adapter:0.8.1 /lambda-adapter /opt/extensions/lambda-adapter
ENV PORT=8000 AWS_LWA_READINESS_CHECK_PROTOCOL=tcp 
WORKDIR /var/task
COPY requirements.txt ./
RUN python -m pip install -r requirements.txt
COPY *.py ./
CMD exec uvicorn --port=$PORT main:app

FastAPIのAPI開発

プログラムのコードはmain.pyのみです。Middlewareを指定している箇所を除き、一般的なFastAPIと同じです。

サンプルでは以下のAPIが含まれています。

パス	メソッド	内容
/s3_bucket_count	GET	S3バケットの数を返却
/s3_object_count	GET	指定したS3バケットの中のオブジェクトの数を返却
/s3_object	GET	指定したバケットとオブジェクトキーの最終更新日を返却

main.py
import datetime
import logging

import boto3
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field

from bedrock_agent.middleware import BedrockAgentMiddleware

app = FastAPI(
    description="This agent allows you to query the S3 information in your AWS account.",
)
app.openapi_version = "3.0.2"
app.add_middleware(BedrockAgentMiddleware)

middleware_logger = logging.getLogger("bedrockagent-middleware")
middleware_logger.setLevel(level=logging.DEBUG)

s3 = boto3.resource("s3")


class S3BucketCountResponse(BaseModel):
    count: int = Field(description="the number of S3 buckets")


@app.get("/s3_bucket_count")
async def get_s3_bucket_count() -> S3BucketCountResponse:
    """
    This method returns the number of S3 buckets in your AWS account.

    Return:
        S3BucketCountResponse: A json object containing the number of S3 buckets in your AWS account.
    """

    count = len(list(s3.buckets.all()))

    return S3BucketCountResponse(count=count)


class S3ObjectCountResponse(BaseModel):
    count: int = Field(description="the number of S3 objects")


@app.get("/s3_object_count")
async def get_s3_object_count(
    bucket_name: str = Query(description="Bucket name"),
) -> S3ObjectCountResponse:
    """
    This method returns the number of S3 objects in your specified bucket.

    Return:
        S3ObjectCountResponse: A json object containing the number of S3 objects in your specified bucket.
    """

    count = len(list(s3.Bucket(bucket_name).objects.all()))
    return S3ObjectCountResponse(count=count)


class S3GetObjectRequest(BaseModel):
    bucket_name: str = Field(description="Bucket name")
    object_key: str = Field(description="Object key")


class S3GetObjectResponse(BaseModel):
    last_modified: datetime.datetime = Field(description="the last modified date")


@app.post("/s3_object")
async def get_s3_object(request: S3GetObjectRequest):
    """
    This method returns the last modified date of S3 object.

    Return:
        S3GetObjectResponse: A json object containing the last modified date of S3 objects.
    """

    object = s3.Object(request.bucket_name, request.object_key)
    last_modified = object.get()["LastModified"]
    return S3GetObjectResponse(last_modified=last_modified)

Middlewareの指定は以下の部分です。この指定だけで/eventsに来たイベントを適切にルーティングしてくれます。

from bedrock_agent.middleware import BedrockAgentMiddleware

app.add_middleware(BedrockAgentMiddleware)

ローカルでのFastAPIの起動

FastAPIアプリですので、そのまま起動することもできます。

pip install fastapi uvicorn pydantic==1.10.13 lwa-fastapi-middleware-bedrock-agent boto3

uvicorn main:app --reload

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [4999] using StatReload
INFO:     Started server process [5001]
INFO:     Waiting for application startup.
INFO:     pass_through_path: /events
INFO:     Application startup complete.

http://127.0.0.1:8000/docsにアクセスすると、Swagger UIでOpenAPIドキュメントが表示されます。

エージェント機能をFastAPIのAPIとして開発を進めれば良いことになります。

OpenAPIスキーマの生成
FastAPIの機能でOpenAPIのスキーマを出力できます。Agents for Amazon Bedrockでこの出力をそのまま利用することができます。
(appディレクトリで実行）
```
python -c "import main;import json; print(json.dumps(main.app.openapi()))" > openapi.json
```
ビルド、デプロイ
SAMプロジェクトのため、標準的なコマンドでビルドとデプロイが可能です。
```
sam build

sam deploy --guided
```
ローカルテスト
sam local invokeコマンドでローカル環境でテストもできます。
```
sam local invoke --event events/s3_bucket_count.json
```
Agents for Amazon Bedrockの作成
LambdaとOpenAPIスキーマの準備ができましたので、マネジメントコンソールでエージェントを作成します。

まとめ

最後まで読んでいただいてありがとうございます。API定義だけに集中してエージェント開発ができますので、ぜひご活用ください

AWS Lambda Web Adapterとは​

サンプルアプリの解説​

まとめ​

AWS Lambda Web Adapterとは

サンプルアプリの解説

まとめ