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

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

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

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/41574/bf4b387f-978a-61af-cfe9-3613dccc993e.png)

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

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

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/41574/7d04b466-45ef-58a3-479b-9d4882b15d86.png)

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

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

参考：[Lambda input event from Amazon Bedrock](https://docs.aws.amazon.com/bedrock/latest/userguide/agents-lambda.html#agents-lambda-input)

```json
{
    "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

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

[I hope support Agents for Amazon Bedrock #317](https://github.com/awslabs/aws-lambda-web-adapter/issues/317)

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


## サンプルアプリの解説

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

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

* ディレクトリ構成

    ```shell
    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](https://pypi.org/project/fastapi/)
    * [lwa-fastapi-middleware-bedrock-agent](https://pypi.org/project/lwa-fastapi-middleware-bedrock-agent/)（FastAPI Middleware for "Agents for Amazon Bedrock"）

1. Lambda Web Adapterの組み込み

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

1. FastAPIのAPI開発

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

    サンプルでは以下のAPIが含まれています。
    
    | パス | メソッド | 内容 |
    | --- | --- | --- |
    | /s3_bucket_count | GET | S3バケットの数を返却 |
    | /s3_object_count | GET | 指定したS3バケットの中のオブジェクトの数を返却 |
    | /s3_object | GET | 指定したバケットとオブジェクトキーの最終更新日を返却 | 
    
    ```python: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に来たイベントを適切にルーティングしてくれます。
    
    ```python
    from bedrock_agent.middleware import BedrockAgentMiddleware
    
    app.add_middleware(BedrockAgentMiddleware)
    ```

1. ローカルでのFastAPIの起動

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

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

    ```shell:
    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ドキュメントが表示されます。

    ![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/41574/e73c2ece-69a4-94f6-3329-b76ebeb30c36.png)

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

1. OpenAPIスキーマの生成

    FastAPIの機能でOpenAPIのスキーマを出力できます。Agents for Amazon Bedrockでこの出力をそのまま利用することができます。
    
    (appディレクトリで実行）

    ```shell
    python -c "import main;import json; print(json.dumps(main.app.openapi()))" > openapi.json
    ```

1. ビルド、デプロイ

    SAMプロジェクトのため、標準的なコマンドでビルドとデプロイが可能です。

    ```shell
    sam build
    
    sam deploy --guided
    ```

1. ローカルテスト

    `sam local invoke`コマンドでローカル環境でテストもできます。

    ```shell
    sam local invoke --event events/s3_bucket_count.json
    ```

1. Agents for Amazon Bedrockの作成

    LambdaとOpenAPIスキーマの準備ができましたので、マネジメントコンソールでエージェントを作成します。


## まとめ

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